Nondeterministically interleaving ống dẫn của

Question

Tôi đã hy vọng để xem một hoạt động interleaving không xác định cho nguồn khác nhau, với một loại chữ ký như

interleave :: WhateverIOMonadClassItWouldWant m => [(k, Source m a)] -> Source m (k, a) 

Các trường hợp sử dụng là tôi có một ứng dụng p2p duy trì các kết nối mở rộng cho nhiều nút trên mạng, và nó chủ yếu chỉ là ngồi xung quanh chờ đợi tin nhắn từ bất kỳ người trong số họ. Khi một tin nhắn đến, nó không quan tâm nó đến từ đâu, nhưng cần xử lý tin nhắn càng sớm càng tốt. Về lý thuyết loại ứng dụng này (ít nhất là khi được sử dụng cho các nguồn giống như socket) có thể bỏ qua toàn bộ trình quản lý IO của GHC và chạy select/epoll/etc. gọi trực tiếp, nhưng tôi không đặc biệt quan tâm đến cách nó được triển khai, miễn là nó hoạt động.

Có phải thứ như thế này có thể có ống dẫn không? Một cách tiếp cận ít tổng quát hơn nhưng có lẽ khả thi hơn có thể là viết một hàm [(k, Socket)] -> Source m (k, ByteString) xử lý việc nhận trên tất cả các ổ cắm cho bạn.

Tôi nhận thấy các hoạt động ResumableSource trong ống dẫn, nhưng tất cả dường như muốn biết về một số Sink cụ thể, cảm giác như một chút rò rỉ trừu tượng, ít nhất là cho hoạt động này.

Answer 1

Gói stm-conduit cung cấp mergeSources thực hiện điều gì đó tương tự- mặc dù không giống với những gì bạn đang tìm kiếm. Đó có lẽ là một nơi tốt để bắt đầu.

Answer 2

Có, điều đó là có thể.

Bạn có thể thăm dò ý kiến ​​một loạt các Source s mà không ngăn chặn bởi forking đề để thăm dò ý kiến ​​ở đâu trong mỗi chủ đề bạn ghép Source lên với một Sink sẽ gửi kết quả đến một số kênh đồng thời:

concur :: (WhateverIOMonadClassItWouldWant m) => TChan a -> Sink a m r 

... và sau đó bạn xác định một Source mà đọc từ kênh đó:

synchronize :: (WhateverIOMonadClassItWouldWant m) => TChan a -> Source a m r 

Chú ý rằng điều này sẽ không khác gì hơn là chỉ forking các chủ đề để thăm dò ý kiến ​​các ổ cắm bản thân, nhưng nó sẽ hữu ích cho những người dùng khác của conduit có thể muốn thăm dò ý kiến ​​của những thứ khác so với các ổ cắm sử dụng Source vì chúng được xác định bởi vì nó tổng quát hơn.

Nếu bạn kết hợp những khả năng thành một chức năng, thì API tổng thể của cuộc gọi sẽ giống như thế:

poll :: (WhateverIOMonadClassItWouldWant m) => [Source a m r] -> m (Source a m r) 

... nhưng bạn vẫn có thể ném vào những k s nếu bạn muốn.