如何使管道与 Haskell 的管道库并发？答案

【问题标题】：How can I make a Pipe concurrent with Haskell's Pipe library?如何使管道与 Haskell 的管道库并发？
【发布时间】：2017-01-01 01:43:49
【问题描述】：

我有一些使用管道的 Haskell 代码：

module Main(main) where
import Pipes

a :: Producer Int IO ()
a = each [1..10]

b :: Pipe Int Int IO ()
b = do
  x <- await
  yield (x*2)
  b

c :: Consumer Int IO ()
c = do
  x <- await
  lift $ print x
  c

main :: IO ()
main = runEffect $ a >-> b >-> c

Pipes.Concurrent tutorial 演示了使用多个工作人员以及窃取工作。我怎样才能在b 内部做类似的事情？我希望b 使用一定数量的工人同时执行它的工作。

显然，并发在这种情况下没有用，但这是我能想到的最简单的例子。在我的实际用例中，我想使用有限数量的工作人员同时发出一些 Web 请求。

【问题讨论】：

hackage.haskell.org/package/pipes-async-0.1.1/docs/src/… 看起来可以调整为将多个给定管道馈送到 TQueue，所以我们得到[Pipe a b] -> Pipe a b。我已经发布了feature request... 一小时前 >:D

标签： haskell concurrency haskell-pipes

【解决方案1】：

编辑：我误解了你的要求；您可能可以在管道内执行此操作，但我不确定动机是什么。我建议构建可重复使用的管道链，并使用工人分配给他们，而不是尝试在管道内构建工人。如果您将其构建到管道本身中，您将失去任何先入先出的排序保证。

Work Stealing 上的部分是您要查找的内容，此代码基本上是教程中的逐字记录，但让我们分解一下它是如何工作的。这是我们可以做你想做的一种方式：

module Main(main) where
import Pipes
import Pipes.Concurrent

import Control.Concurrent.Async (async, wait)
import Control.Concurrent (threadDelay)
import Control.Monad (forM)

a :: Producer Int IO ()
a = each [1..10]

b :: Pipe Int Int IO ()
b = do
  x <- await
  yield (x*2)
  b

c :: Consumer Int IO ()
c = do
  x <- await
  lift $ print x
  c

main :: IO ()
main = do
  (output, input) <- spawn unbounded
  feeder <- async $ do runEffect $ a >-> toOutput output
                       performGC

  workers <- forM [1..3] $ \i ->
    async $ do runEffect $ fromInput input  >-> b >-> c
               performGC

  mapM_ wait (feeder:workers)

第一行spawn unbounded 来自 Pipes.Concurrent，它初始化一个具有输入和输出句柄的“邮箱”。一开始我很困惑，但在这种情况下，我们将消息发送到输出并从输入中提取它们。这类似于 golang 等语言中的推拉消息通道。

我们指定一个Buffer 来说明我们可以存储多少条消息，在这种情况下，我们将无限制设置为无限制。

好的，邮箱已初始化，我们现在可以创建Effects 来向其发送消息。邮箱通道是使用STM实现的，所以它可以异步收集消息。

让我们创建一个为邮箱提供数据的异步作业；

feeder <- async $ do runEffect $ a >-> toOutput output
                     performGC

a >-> toOutput output 只是普通的管道组合，我们需要toOutput 将输出转换为管道。注意performGC 调用也是 IO 的一部分，它允许 Pipes.Concurrent 知道在作业完成后进行清理。如果我们愿意，我们可以使用forkIO 运行它，但在这种情况下我们使用async 以便我们可以等待结果稍后完成。好的，所以我们的邮箱应该异步接收消息，让我们把它们拉出来做一些工作。

workers <- forM [1..3] $ \i ->
  async $ do runEffect $ fromInput input  >-> b >-> c
             performGC

与之前的想法相同，但这次我们只是生成了其中的一些。我们像使用fromInput 的普通管道一样从输入中读取数据，然后在链的其余部分运行它，完成后进行清理。 input 将确保每次提取一个值时只有一个工作人员接收它。当所有输入到output 的作业完成时（它会跟踪所有打开的作业）然后它将关闭input 管道并且工作人员将完成。

如果您在 web-worker 场景中使用它，您将拥有一个主循环，该循环不断向 toOutput output 通道发送请求，然后生成任意数量的从 fromInput input 拉入其管道的工作人员.

【讨论】：