管道 3.0：非线性拓扑

Question

我正在查看用于流处理的管道 3.0 包。 The tutorial 做得很好也很清楚，只是我无法绕开“压缩和合并”部分。

我的目标是像 ArrowChoice 允许的那样组合管道：

• 我有一个独特的 Either a a 制作人
• 我想将第一个管道应用于左值，另一个应用于右值
• 然后我想合并结果，并继续管道

+----------+                   +------+ - filterLeft ->  pipe1 -> +------------+ 
| producer | - (Either a a) -> | fork |                           | mergeD (?) |
+----------+                   +------+ - filterRight -> pipe2 -> +------------+

我在教程中定义fork：

fork () = 
    runIdentityP . hoist (runIdentityP . hoist runIdentityP) $ forever $ do
        a <- request ()
        lift $ respond a
        lift $ lift $ respond a

oddOrEven x = if odd x then Left x else Right x
producer = fromListS [1..0] >-> mapD oddOrEven
isLeft (Left _) = True
isLeft (Right _) = False
isRight = not . isLeft
filterLeft = filterD isLeft
filterRight = filterD isRight
pipe1 = mapD (\x -> ("seen on left", x))
pipe2 = mapD (\x -> ("seen on right", x))

p1 = producer >-> fork    

问题是我无法使类型正确。本教程似乎只展示了如何将内部（提升）管道链作为自包含会话运行，但我希望能够将其值重新注入管道，而不仅仅是对它们应用效果。我当然尝试过追随这些类型，但它们很快就会变得有点毛茸茸。

有人可以帮助我吗？提前致谢。

（PS：这种拓扑的示例将是本教程的一个很好的补充，或者更好的部分是关于如何使用管道模拟Control.Arrow 的东西）

Answer 1

pipe 抽象不支持菱形拓扑或任何形式的Arrow 类行为。这不是 API 问题，而是这种场景没有正确或明确定义的行为。

为了解释原因，请允许我将您的图表简化为以下图表：

          +----+
          | pL |
+----+ => +----+ => +----+
| p1 |              | p2 |
+----+ => +----+ => +----+
          | pR |
          +----+

想象一下我们在p1 管道，我们在respond 到pL。如果您还记得教程，代理法要求每个respond 阻塞直到上游。这意味着p1 无法重新获得控制权，直到再次pL requests。所以此时我们有：

• p1 阻塞等待来自 pL 的 request

但是，假设pL 还没有request，而是将responds 的值设为p2。所以现在我们有：

• p1 阻塞等待来自 pL 的 request
• pL 阻塞等待来自 p2 的 request

现在假设 p2 代替 requests 来自 pR。代理法规定p2 无法重新获得控制权，直到pR responds 再次出现。现在我们有：

• p1 阻塞等待来自 pL 的 request
• pL 阻塞等待来自p2 的request
• p2 阻塞等待来自 pR 的 respond

现在当pR requests 来自p1 的值时会发生什么？如果我们查看我们的块列表，p1 仍然被阻止等待来自pL 的request，因此无法接收来自pR 的request。可以说，没有正确的“打结”方式，即使pL 和pR 共享相同的request 签名。

更一般地说，代理法则确保以下两个不变量：

• 活动管道“上游”的每个管道都将在 respond 上阻塞
• 活动管道“下游”的每个管道都将在 request 上阻塞

循环或菱形打破了这些不变量。这就是为什么本教程非常简短地指出循环拓扑没有“意义”。

您可以在我刚刚给您的示例中看到为什么钻石会破坏这个不变量。当p1 拥有控制权时，它位于pR 的上游，这意味着pR 在request 上被阻止。然而，当p2 获得控制权时，它位于pR 的下游，这意味着pR 在respond 上被阻止。这会导致矛盾，因为pR 还不能改变，因为控制流经pL 而不是pR 到达p2。

机器

因此，您的问题有两种解决方案。一种解决方案是将您想要的拆分行为内联到单个管道中。您定义了一个 pE 管道，它将 pL 和 pR 的行为组合到一个管道中。

这个问题的更优雅的解决方案是 Edward 的 machines 风格。您定义了一个比支持ArrowChoice 的代理更受限制的抽象，您在该抽象的域内执行您的箭头式操作，然后在完成后将其升级为代理。

如果你眯着眼睛，你可以假装在 Haskell 中有一类当前可用的协程抽象是偏序的。协程抽象是对象，从协程抽象C1 到协程抽象C2 的箭头表示您可以将C1 类型的协程嵌入C2 类型的协程中（即C1 是@ 的不正确子集987654396@)。

在这个偏序中，代理可能是终端对象，这意味着您可以将代理视为协程的汇编语言。按照汇编语言的类比，代理提供的保证较少，但您可以在代理中嵌入更多限制性的协程抽象（即高级语言）。这些高级语言提供了更大的限制，从而实现了更强大的抽象（即Arrow 实例）。

如果您想要一个简单的例子，请考虑最简单的协程抽象之一：Kleisli 箭头：

newtype Kleisli m a b = Kleisli { runKleisli :: a -> m b }

instance Category (Kleisli m) where
    id = Kleisli return
    (Kleisli f) . (Kleisli g) = Kleisli (f <=< g)

Kleisli 箭头肯定比代理更具限制性，但由于这种限制，它们支持Arrow 实例。因此，当您需要 Arrow 实例时，您可以使用 Kleisli 箭头编写代码，并使用 Arrow 表示法将其组合，然后当您完成后，您可以使用将更高级别的 Kleisli 代码“编译”为代理汇编代码mapMD:

kleisliToProxy :: (Proxy p) => Kleisli m a b -> () -> Pipe p a b m r
kleisliToProxy (Kleisli f) = mapMD f

这个编译遵循函子定律：

kleisliToProxy id = idT

kleisliToProxy (f . g) = kleisliToProxy f <-< kleisliToProxy g

因此，如果您的分支代码可以用Kleisli 箭头编写，那么在该代码部分使用Kleisli 箭头，然后在完成后将其编译为代理。使用这个技巧，你可以将多个协程抽象编译成代理抽象来混合它们。