管道流中的并行处理

Question

我非常喜欢用于将操作应用于流式 IO 源的管道/管道概念。我对构建适用于非常大的日志文件的工具很感兴趣。从 Python/Ruby 迁移到 Haskell 的吸引力之一是编写并行代码的更简单方法，但我找不到任何相关文档。我如何设置一个从文件中读取行并并行处理它们的管道流（即，使用 8 个内核，它应该读取 8 行，并将它们交给八个不同的线程进行处理，然后再次收集等），理想情况下，尽可能少的“仪式”......

如果可能会影响流程的速度，可以选择是否需要按顺序重新连接线路？

我确信可以使用 Parallel Haskell 书中的想法自己拼凑一些东西，但在我看来，在 Conduit 工作流程中间并行运行纯函数（parmap 等）应该非常容易?

Answer 1

作为 Petr Pudlák 在他的评论中提到的“内部并行性”的一个例子，考虑这个函数（我正在使用 pipes，但可以很容易地用 conduit 实现）：

import Control.Monad
import Control.Lens (view)
import Control.Concurrent.Async (mapConcurrently)
import Pipes
import qualified Pipes.Group as G
import qualified Control.Foldl as L

concProd :: Int -> (a -> IO b) -> Producer a IO r -> Producer b IO r
concProd groupsize action producer = 
      L.purely G.folds L.list (view (G.chunksOf groupsize) producer)
      >->
      forever (await >>= liftIO . mapConcurrently action >>= mapM G.yield) 

此函数将组大小、我们要为a 类型的每个值运行的操作以及Producer 的a 值作为参数。

它返回一个新的Producer。在内部，生产者分批读取a值groupsize，并发处理，并一一产生结果。

代码使用Pipes.Group将原始生产者“划分”为大小为groupsize的子生产者，然后Control.Foldl将每个子生产者“折叠”成一个列表。

对于更复杂的任务，您可以使用pipes-concurrency 或stm-conduit 提供的异步通道。但这些让你在某种程度上摆脱了香草管道/导管的“单一管道”世界观。