Haskell：在不使用 spawn 的情况下拆分管道（广播）答案

【问题标题】：Haskell: Splitting pipes (broadcast) without using spawnHaskell：在不使用 spawn 的情况下拆分管道（广播）
【发布时间】：2013-11-14 13:00:02
【问题描述】：

这个问题有点代码高尔夫，而且很新。

我在 Haskell 中使用了很棒的 pipes 库，我想拆分管道以沿多个通道发送相同的数据（进行广播）。 Pipes.Concurrent tutorial 建议使用spawn 创建邮箱，利用Output 的monoid 状态。例如，我们可能会这样做：

main = do
 (output1, input1) <- spawn Unbounded
 (output2, input2) <- spawn Unbounded
 let effect1 = fromInput input1 >-> pipe1
 let effect2 = fromInput input2 >-> pipe2
 let effect3 = P.stdinLn >-> toOutput (output1 <> output2)
 ...

这种间接通过邮箱真的有必要吗？我们可以写这样的东西吗？

main = do
 let effect3 = P.stdinLn >-> (pipe1 <> pipe2)
 ...

上面没有编译，因为Pipe 没有Monoid 实例。这有充分的理由吗？第一种方法真的是最干净的管道分割方法吗？

【问题讨论】：

标签： haskell haskell-pipes

【解决方案1】：

有两种方法可以在不使用并发的情况下做到这一点，但都需要注意。

第一种方法是，如果 pipe1 和 pipe2 只是简单的 Consumers，则永远循环如下：

p1 = for cat f  -- i.e. p1 = forever $ await >>= f
p2 = for cat g  -- i.e. p2 = forever $ await >>= g

...那么解决这个问题的简单方法就是写：

for P.stdinLn $ \str -> do
    f str
    g str

例如，如果p1 只是printing 每个值：

p1 = for cat (lift . print)

...而p2 只是将该值写入句柄：

p2 = for cat (lift . hPutStrLn h)

...然后你可以像这样组合它们：

for P.stdinLn $ \str -> do
    lift $ print str
    lift $ hPutStrLn h str

但是，这种简化仅适用于简单循环的Consumers。还有另一种更通用的解决方案，即为管道定义一个ArrowChoice 实例。我相信基于拉的Pipes 不允许正确的守法实例，但基于推的Pipes 允许：

newtype Edge m r a b = Edge { unEdge :: a -> Pipe a b m r }

instance (Monad m) => Category (Edge m r) where
    id = Edge push
    (Edge p2) . (Edge p1) = Edge (p1 >~> p2)

instance (Monad m) => Arrow (Edge m r) where
    arr f = Edge (push />/ respond . f)
    first (Edge p) = Edge $ \(b, d) ->
        evalStateP d $ (up \>\ unsafeHoist lift . p />/ dn) b
      where
        up () = do
            (b, d) <- request ()
            lift $ put d
            return b
        dn c = do
            d <- lift get
            respond (c, d)

instance (Monad m) => ArrowChoice (Edge m r) where
    left (Edge k) = Edge (bef >=> (up \>\ (k />/ dn)))
      where
          bef x = case x of
              Left b -> return b
              Right d -> do
                  _ <- respond (Right d)
                  x2 <- request ()
                  bef x2
          up () = do
              x <- request ()
              bef x
          dn c = respond (Left c)

这需要一个新类型，以便类型参数符合ArrowChoice 所期望的顺序。

如果您对基于推送的Pipe 一词不熟悉，它基本上是一个Pipe，它从最上游的管道而不是最下游的管道开始，它们都具有以下形状：

a -> Pipe a b m r

将其视为Pipe，直到从上游接收到至少一个值才能“运行”。

这些基于推送的Pipes 是传统基于拉取的Pipes 的“双重”，具有自己的组合运算符和身份：

(>~>) :: (Monad m)
      => (a -> Pipe a b m r)
      -> (b -> Pipe b c m r)
      -> (a -> Pipe a c m r)

push  :: (Monad m)
      ->  a -> Pipe a a m r

...但默认情况下，单向Pipes API 不会导出它。您只能从 Pipes.Core 获得这些运算符（并且您可能希望更仔细地研究该模块以建立对它们如何工作的直觉）。该模块显示基于推送的Pipes 和基于拉取的Pipes 都是更通用的双向版本的特例，了解双向情况是您了解它们为何彼此对偶的方式。

一旦你有了一个用于推送管道的 Arrow 实例，你就可以编写如下内容：

p >>> bifurcate >>> (p1 +++ p2)
  where
    bifurcate = Edge $ pull ~> \a -> do
        yield (Left  a)  -- First give `p1` the value
        yield (Right a)  -- Then give `p2` the value

然后您将在完成后使用runEdge 将其转换为基于拉取的管道。

这种方法有一个主要缺点，即您不能自动将基于拉的管道升级为基于推送的管道（但通常不难弄清楚如何手动完成）。例如，要将Pipes.Prelude.map 升级为基于推送的Pipe，您可以编写：

mapPush :: (Monad m) => (a -> b) -> (a -> Pipe a b m r)
mapPush f a = do
    yield (f a)
    Pipes.Prelude.map f

那么有正确的类型被包裹在Arrow:

mapEdge :: (Monad m) => (a -> b) -> Edge m r a b
mapEdge f = Edge (mapPush f)

当然，更简单的方法是从头开始编写：

mapEdge f = Edge $ push ~> yield . f

使用最适合您的方法。

事实上，我提出Arrow 和ArrowChoice 实例正是因为我试图回答与您完全相同的问题：如何在不使用并发的情况下解决这些问题？我在另一个 Stack Overflow 答案 here 中写了一个关于这个更普遍主题的长答案，我在其中描述了如何使用这些 Arrow 和 ArrowChoice 实例将并发系统提炼成等效的纯系统。

【讨论】：