具有下游状态且没有损失的惯用双向管道

Question

假设我有简单的生产者/消费者模型,消费者想要将一些状态传递给生产者.例如,让下游流动的对象成为我们想要写入文件的对象,上游对象是表示在文件中写入对象的位置的一些标记(例如,偏移).

这两个过程可能看起来像这样(带pipes-4.0),

{-# LANGUAGE GeneralizedNewtypeDeriving #-}

import Pipes
import Pipes.Core
import Control.Monad.Trans.State       
import Control.Monad

newtype Object = Obj Int
               deriving (Show)

newtype ObjectId = ObjId Int
                 deriving (Show, Num)

writeObjects :: Proxy ObjectId Object () X IO r
writeObjects = evalStateT (forever go) (ObjId 0)
  where go = do i <- get
                obj <- lift $ request i
                lift $ lift $ putStrLn $ "Wrote "++show obj
                modify (+1)

produceObjects :: [Object] -> Proxy X () ObjectId Object IO ()
produceObjects = go
  where go [] = return ()
        go (obj:rest) = do
            lift $ putStrLn $ "Producing "++show obj
            objId <- respond obj
            lift $ putStrLn $ "Object "++show obj++" has ID "++show objId
            go rest

objects = [ Obj i | i <- [0..10] ]

虽然这可能很简单,但我在如何撰写它们时遇到了相当大的困难.理想情况下,我们需要基于推送的控制流程,如下所示,

1. writeObjects通过阻止开始request,已经发送了初始ObjId 0上游.
2. produceObjects发送第一个对象Obj 0,下游
3. writeObjects写入对象并递增其状态,并等待request,这次发送ObjId 1上游
4. respond在produceObjects回报中ObjId 0
5. produceObjects 继续步骤(2)与第二个对象, Obj 1

我最初的尝试是基于推送的组合,如下所示,

main = void $ run $ produceObjects objects >>~ const writeObjects

注意使用const解决其他不兼容的类型(这可能是问题所在).然而,在这种情况下,我们发现ObjId 0吃了,

Producing Obj 0
Wrote Obj 0
Object Obj 0 has ID ObjId 1
Producing Obj 1
...

基于拉的方法,

main = void $ run $ const (produceObjects objects) +>> writeObjects

遭遇类似的问题,这一次下降Obj 0.

怎么可能以理想的方式组成这些作品呢？

Answer 1

选择使用哪种成分取决于哪个成分应该启动整个过程.如果您希望下游管道启动该过程,那么您希望使用基于拉的合成(即(>+>)/ (+>>)),但如果您希望上游管道启动该过程,那么您应该使用基于推的合成(即(>>~)/ (>~>)).您获得的类型错误实际上警告您代码中存在逻辑错误:您尚未明确确定哪个组件首先启动该进程.

从您的描述中,很明显您希望控制流开始,produceObjects因此您希望使用基于推送的合成.使用基于推送的合成后,合成运算符的类型将告诉您需要了解的有关如何修复代码的所有信息.我将采用它的类型并将其专门化为你的构图链:

-- Here I'm using the `Server` and `Client` type synonyms to simplify the types
(>>~) :: Server ObjectId Object IO ()
      -> (Object -> Client ObjectId Object IO ())
      -> Effect IO ()

正如你已经注意到了,当你尝试你得到了类型错误使用(>>~)告诉你,你失踪类型的参数Object,以你的writeObjects功能.这静态地强制您writeObjects在收到第一个代码之前不能运行任何代码Object(通过初始参数).

解决方案是重写您的writeObjects函数,如下所示:

writeObjects :: Object -> Proxy ObjectId Object () X IO r
writeObjects obj0 = evalStateT (go obj0) (ObjId 0)
  where go obj = do i <- get
                    lift $ lift $ putStrLn $ "Wrote "++ show obj
                    modify (+1)
                    obj' <- lift $ request i
                    go obj'

然后,这给出了正确的行为:

>>> run $ produceObjects objects >>~ writeObjects
Producing Obj 0
Wrote Obj 0
Object Obj 0 has ID ObjId 0
Producing Obj 1
Wrote Obj 1
Object Obj 1 has ID ObjId 1
Producing Obj 2
Wrote Obj 2
Object Obj 2 has ID ObjId 2
Producing Obj 3
Wrote Obj 3
Object Obj 3 has ID ObjId 3
Producing Obj 4
Wrote Obj 4
Object Obj 4 has ID ObjId 4
Producing Obj 5
Wrote Obj 5
Object Obj 5 has ID ObjId 5
Producing Obj 6
Wrote Obj 6
Object Obj 6 has ID ObjId 6
Producing Obj 7
Wrote Obj 7
Object Obj 7 has ID ObjId 7
Producing Obj 8
Wrote Obj 8
Object Obj 8 has ID ObjId 8
Producing Obj 9
Wrote Obj 9
Object Obj 9 has ID ObjId 9
Producing Obj 10
Wrote Obj 10
Object Obj 10 has ID ObjId 10

您可能想知道为什么这两个管道中的一个管道采用初始参数的要求是有道理的,除了抽象的理由,这是类别法律所要求的.简单的英语解释是,替代方案是,Object在writeObjects到达第一个request语句之前,您需要缓冲区首先在两个管道之间" 传输" .这种方法产生了许多有问题的行为和错误的角落情况,但可能最重要的问题是管道组合将不再是关联的,并且效果的顺序将根据您组合事物的顺序而改变.

双向管道组合操作员的好处是,这些类型可以解决,因此您可以通过研究类型来总是推断出组件是"活动"(即启动控制)还是"被动"(即等待输入) .如果组合说某个管道(如writeObjects)必须采取参数,那么它是被动的.如果它不需要参数(例如produceObjects),则它处于活动状态并启动控制.因此,组合强制您在管道中最多有一个活动管道(不接受初始参数的管道),这是开始控制的管道.