在Haskell中将管道组成循环或循环 [英] Composing Pipes into a loop or cycle in haskell
问题描述
这个问题是关于Haskell库管道。
This question is about the Haskell library Pipes.
This question is related to 2019 Advent of Code Day 11 (possible spoiler warning)
我有两个 Pipe Int Int mr 大脑和机器人在一个连续的循环中。那是大脑的输出，需要转到 robot 的输入和的输出机器人需要输入大脑的输入。 大脑完成后，我需要计算结果。
I have two Pipe Int Int m r brain and robot that need to pass information too each other in a continuous loop. That is the output of brain need to go to the input of robot and the output of robot needs to go the the input of brain. When brain finished I need the result of the computation.
我如何组成大脑和 robot 进入循环？理想情况下，我可以传递给 runEffect
How do I compose brain and robot into a loop? Ideally a loop with the type Effect m r that I can pass to runEffect
编辑：结果应如下所示：
The result should look like this:
+-----------+ +-----------+
| | | |
| | | |
a ==> f ==> b ==> g ==> a=|
^ | | | | |
| | | | | | | |
| +-----|-----+ +-----|-----+ |
| v v |
| () r |
+=====================================+
推荐答案
答案
最简单的解决方案是使用 Client 和 Server ，如danidiaz在评论中建议的那样，因为 pipes 没有对循环管道的任何内置支持，而且即使不是不可能的话，这也将非常困难。这主要是因为我们需要处理 await s的数量与 yield s的数量不匹配的情况
The answer
The easiest solution would be to use Client and Server as danidiaz suggested in the comments, since pipes doesn't have any built in support for cyclic pipes and it would be incredibly difficult, if not impossible to do so correctly. This is mostly because we need to handle cases where the number of awaits doesn't match the number of yields.
编辑：：我添加了关于其他答案问题的章节。参见另一个有问题的替代方案部分。
I added a section about the problems with the other answer. See section "Another problematic alternative"
编辑2：我在下面添加了一个问题较少的解决方案。参见可能的解决方案部分。
Edit 2: I have added a less problematic possible solution below. See the section "A possible solution"
但是可以使用 Client 和 Server ）和整洁的功能 generalize ，它将单向 Pipe 变成双向 Proxy 。
It is however possible to simulate it with the help of the Proxy framework (with Client and Server) and the neat function generalize, which turns a unidirectional Pipe into a bidirectional Proxy.
generalize f x0
+-----------+ +---------------------+
| | | |
| | x <======================== x
a ==> f ==> b becomes | |
| | a ==> f ==> b
| | | | |
+-----|-----+ +----------|----------+
v v
r r
现在我们可以使用 //> 和 > \\ 插入两端并使其具有周期性：
Now we can use //> and >\\ to plug the ends and make the flow cyclic:
loop :: Monad m => Pipe a a m r -> a -> Effect m r
loop p x0 = pure >\\ generalize p x0 //> pure
loop f
a
+-----|-----+
| | |
/====<=======/===<========\
| | | |
\=> a ==> f ==> a ==/
| |
+-----|-----+
v
r
如您所见，我们需要输入 a 的初始值。这是因为无法保证管道在屈服之前不会 await ，这将迫使它永远等待。
As you can see, we are required to input an initial value for a. This is because there is no guarantee that the pipe won't await before it yields, which would force it to wait forever.
但是请注意，如果管道收益在等待将丢弃数据。 c $ c> ing，因为generalize在内部使用状态monad实现，该状态monad在屈服时保存最后一个值，在等待时检索最后一个值。
Note however that this will throw away data if the pipe yields multiple times before awaiting, since generalize is internally implemented with a state monad that saves the last value when yielding and retrieves the last value when awaiting.
要在管道中使用它，只需对其进行组合，然后将其放入 loop ：
To use it with your pipes, simply compose them and give them to loop:
runEffect $ loop (f >-> g)
但是请不要使用它，因为如果不小心，它将随机丢弃数据
But please don't use it, since it will randomly throw away data if you are not careful
您还可以像mingmingrr建议的那样制作一个无限懒散的管道链
You could also make a lazily infinite chain of pipes like mingmingrr suggested
infiniteChain :: Functor m => Pipe a a m r -> Producer a m r
infiniteChain f = infiniteChain >-> f
这解决了丢弃/重复值的问题，但还有其他一些问题。首先是在屈服之前先等待会导致无限循环使用无限的内存，但这已在mingmingrr的答案中得到解决。
This solves the problem of discarded/duplicated values, but has several other problems. First is that awaiting first before yielding will cause an infinite loop with infinite memory usage, but that is already addressed in mingmingrr's answer.
另一个更难解决的问题是在每次等待之前，将在相应收益率之前的每个动作重复一次。如果我们修改他们的示例以记录正在发生的情况，我们可以看到这一点：
Another, more difficult to solve, issue is that every action before the corresponding yield is duplicated once for each await. We can see this if we modify their example to log what is happening:
import Pipes
import qualified Pipes.Prelude as P
f :: Monad m => Pipe Int Int m r
f = P.map (* 2)
g :: Monad m => Int -> Pipe Int Int m ()
g 0 = return ()
g n = do
lift . putStrLn $ "Awaiting. n = " ++ show n
x <- await
lift . putStrLn $ "Got: x = " ++ show x ++ " and n = "++ show n ;
yield (x + 1)
g (n - 1)
cyclic' :: Monad m => Int -> Producer Int m Int
cyclic' input = let pipe = (yield input >> pipe) >-> f >-> g 6 in pipe
现在，运行 runEffect（循环'0>- > P.print）将打印以下内容：
Now, running runEffect (cyclic' 0 >-> P.print) will print the following:
Awaiting. n = 6
Got: x = 0 and n = 6
1
Awaiting. n = 5
Awaiting. n = 6
Got: x = 0 and n = 6
Got: x = 2 and n = 5
3
Awaiting. n = 4
Awaiting. n = 5
Awaiting. n = 6
Got: x = 0 and n = 6
Got: x = 2 and n = 5
Got: x = 6 and n = 4
7
Awaiting. n = 3
Awaiting. n = 4
Awaiting. n = 5
Awaiting. n = 6
Got: x = 0 and n = 6
Got: x = 2 and n = 5
Got: x = 6 and n = 4
Got: x = 14 and n = 3
15
Awaiting. n = 2
Awaiting. n = 3
Awaiting. n = 4
Awaiting. n = 5
Awaiting. n = 6
Got: x = 0 and n = 6
Got: x = 2 and n = 5
Got: x = 6 and n = 4
Got: x = 14 and n = 3
Got: x = 30 and n = 2
31
Awaiting. n = 1
Awaiting. n = 2
Awaiting. n = 3
Awaiting. n = 4
Awaiting. n = 5
Awaiting. n = 6
Got: x = 0 and n = 6
Got: x = 2 and n = 5
Got: x = 6 and n = 4
Got: x = 14 and n = 3
Got: x = 30 and n = 2
Got: x = 62 and n = 1
63
如您所见，对于每次 ，我们都会重新执行所有操作，直到对应的收益。更具体地说，等待将触发管道的新副本运行，直到达到产量为止。当我们再次等待时，该副本将一直运行到再次产生下一个收益，并且如果在此期间触发 await ，它将创建另一个副本并将其运行直到第一个收益，
As you can see, for each await, we re-executed everything until the corresponding yield. More specifically, an await triggers a new copy of the pipe to run until it reaches a yield. When we await again, the copy will run until the next yield again and if it triggers an await during that, it will create yet another copy and run it until the first yield, and so on.
这意味着，在最佳情况下，我们得到的是 O（n ^ 2）而不是线性性能（并使用 O（n）代替 O（1）内存），因为我们要重复每个动作。在最坏的情况下，例如如果我们正在读取或写入文件，则由于重复出现副作用，我们可能会得到完全错误的结果。
This means that in the best case, we get O(n^2) instead of linear performance (And using O(n) instead of O(1) memory), since we are repeating everything for each action. In the worst case, e.g. if we were reading from or writing to a file, we could get completely wrong results since we are repeating side-effects.
如果您确实必须使用 Pipe s并且不能使用 request / 响应，您可以确保代码的等待永远不会超过（或之前）其收益 （或者在这些情况下有一个很好的默认设置），我们可以在我之前的尝试基础上提出一个解决方案，至少可以处理 yield 的情况。 code>比您 await 多得多。
If you really must use Pipes and can't use request/respond instead and you are sure that your code will never await more than (or before) it yields (or have a good default to give it in those cases), we could build upon my previous attempt above to make a solution that at least handles the case when yielding more than you await.
诀窍是为 generalize ，因此多余的值将被存储而不是被丢弃。我们还可以将多余的参数保留为缓冲区为空时的默认值。
The trick is adding a buffer to the implementation of generalize, so the excess values are stored instead of being thrown away. We can also keep the extra argument as a default value for when the buffer is empty.
import Pipes.Lift (evalStateP)
import Control.Monad.Trans.State.Strict (state, modify)
import qualified Data.Sequence
generalize' :: Monad m => Pipe a b m r -> x -> Proxy x a x b m r
generalize' p x0 = evalStateP Seq.empty $ up >\\ hoist lift p //> dn
where
up () = do
x <- lift $ state (takeHeadDef x0)
request x
dn a = do
x <- respond a
lift $ modify (Seq.|> x)
takeHeadDef :: a -> Seq.Seq a -> (a, Seq.Seq a)
takeHeadDef x0 xs = (foldr const x0 xs, Seq.drop 1 xs)
如果现在将其插入到 loop 的定义中，我们将解决丢弃多余值的问题（以保持缓冲）。它还可以防止复制默认值以外的任何其他值，并且仅在缓冲区为空时才使用默认值。
If we now plug this into our definition of of loop, we will have solved the problem of discarding excess values (at the memory cost of keeping a buffer). It also prevents duplicating any values other than the default value and only uses the default value when the buffer is empty.
loop' :: Monad m => a -> Pipe a a m r -> Effect m r
loop' x0 p = pure >\\ generalize' p x0 //> pure
如果我们想在之前等待 yield 表示错误，我们可以简单地给出 error 作为我们的默认值： loop' （错误等待，直到无收益）。somePipe 。
If we want awaiting before yielding to be an error, we can simply give error as our default value: loop' (error "Await without yield") somePipe.
使用来自 Server 4.3.13 / docs / Pipes-Core.html rel = nofollow noreferrer> Pipes.Core 。它将解决您的问题，并且不会引起大量奇怪的错误。
Use Client and Server from Pipes.Core. It will solve your problem and not cause a ton of strange bugs.
如果这不可能，那么我的可能的解决方案部分的修改版本为概括应该可以胜任这项工作。
If that is not possible, my "Possible solution" section with a modified version of generalize should do the job in most cases.
这篇关于在Haskell中将管道组成循环或循环的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持IT屋！
