休斯的斐波那契流
Hughes' Fibonacci stream
我正在尝试理解 John Hughes 著名的 "Generalising Arrows to Monads" 中的“箭头流”部分。更准确地说，我有兴趣写下斐波那契流。
我稍微调整了 Hughes 的定义：
data StreamProcessor a b = Get (a -> StreamProcessor a b) |
Put b (StreamProcessor a b) |
Halt
put = Put
get = Get
首先，我将流处理器视为可能阻塞（等待输入）的列表。即：
Put :: b -> StreamProcessor a b -> StreamProcessor a b 匹配 (:) :: a -> [a] -> [a];Halt :: StreamProcessor a b 匹配 [] :: [a];Get :: (a -> StreamProcessor a b) -> StreamProcessor a b 帮助我们阻塞流并等待输入。
因此，如果我们删除 Get，我们会得到一个类似列表的结构。如果我们也删除 Halt，我们会得到一个类似无限列表的结构。
以下是我理解“斐波那契流”的两种方式：
一个非阻塞的无限流（类似无限列表）：
zipNonBlockedStreamsWith :: (a -> b -> c)
-> StreamProcessor () a
-> StreamProcessor () b
-> StreamProcessor () c
zipNonBlockedStreamsWith f (Put x sp) (Put y sp')
= Put (f x y) (zipNonBlockedStreamsWith f sp sp')
zipNonBlockedStreamsWith f Halt sp = Halt
zipNonBlockedStreamsWith f sp Halt = Halt
fibs :: StreamProcessor () Int
fibs =
put 0 (put 1 $ zipNonBlockedStreamsWith (+) fibs (tailNonBlockedStream fibs))
-- matches a well-known definition of an infinite Fibonacci list.
fibsList :: [Int]
fibsList = 0 : 1 : (zipWith (+) fibsList (tail fibsList))
-- with the 'fibsList', we can use folds to do the same thing.
putStream :: [b] -> StreamProcessor a b -> StreamProcessor a b
putStream bs sp = foldr Put sp bs
fibs' :: StreamProcessor () Int
fibs' = putStream fibsList Halt
一个阻塞流等待n，输出第n个斐波那契数并再次阻塞。 Hughes 的 Arrow 界面帮助我们以相当简洁的方式表达它：
instance Category StreamProcessor where
...
instance Arrow StreamProcessor where
arr f = Get $ \ a -> Put (f a) (arr f)
...
fibsList :: [Int]
fibsList = 0 : 1 : (zipWith (+) fibsList (tail fibsList))
blockedFibs :: StreamProcessor Int Int
blockedFibs = arr (fibsList !!)
然而，在 the paper I linked 中，John Hughes 展示了另一个解决方案，Arrow他通过：
instance Category StreamProcessor where
id = Get (\ x -> Put x id)
Put c bc . ab = Put c (bc . ab)
Get bbc . Put b ab = (bbc b) . ab
Get bbc . Get aab = Get $ \ a -> (Get bbc) . (aab a)
Get bbc . Halt = Halt
Halt . ab = Halt
bypass :: [b] -> [d] -> StreamProcessor b c -> StreamProcessor (b, d) (c, d)
bypass [] ds (Get f) = Get $ \ ~(b, d) -> bypass [] (ds ++ [d]) (f b)
bypass (b : bs) [] (Get f) = bypass bs [] (f b)
bypass [] (d : ds) (Put c sp) = Put (c, d) $ bypass [] ds sp
bypass bs [] (Put c sp) =
Get $ \ ~(b, d) -> Put (c, d) (bypass (bs ++ [b]) [] sp)
bypass bs ds Halt = Halt
instance Arrow StreamProcessor where
arr f = Get $ \ a -> Put (f a) (arr f)
first = bypass [] []
liftArr2 :: Arrow k => (a -> b -> c) -> k r a -> k r b -> k r c
liftArr2 f a b = a &&& b >>^ uncurry f
fibsHughes = let
fibsHughes' = put 1 (liftArr2 (+) fibsHughes fibsHughes')
in put 0 fibsHughes'
但它并不像我期望的那样工作。以下函数将帮助我们从流中获取值，直到它阻塞或停止（使用 Data.List.unfoldr）：
popToTheBlockOrHalt :: StreamProcessor a b -> [b]
popToTheBlockOrHalt = let
getOutput (Put x sp) = Just (x, sp)
getOutput getOrHalt = Nothing
in unfoldr getOutput
所以，这是我们得到的：
GHCi> popToTheBlockOrHalt fibsHughes
[0, 1]
GHCi> :t fibsHughes
fibsHughes :: StreamProcessor a Integer
如果我们检查模式，我们会发现它阻塞了（也就是说我们偶然发现 Get）。
我不知道是不是应该这样。如果这是我们想要的，为什么？如果不是，问题是什么？我检查并重新检查了我编写的代码，它与 Hughes 文章中的定义非常匹配（嗯，我不得不添加 id 和 Halt 的模式 - 我看不出他们怎么会搞砸了这个过程向上）。
编辑： 正如评论中所说，在 later edition of the paper bypass was slightly changed (we use that one). It is able to accumulate both withheld bs and ds (that is has two queues), whereas the bypass from the original paper 中仅累积 ds（即一个队列）。
编辑 #2: 我只是想写下一个函数，它会从 fibsHughes:
中弹出斐波那契数列
popToTheHaltThroughImproperBlocks :: StreamProcessor () b -> [b]
popToTheHaltThroughImproperBlocks = let
getOutput (Put x sp) = Just (x, sp)
getOutput (Get c) = getOutput $ c ()
getOutput Halt = Nothing
in unfoldr getOutput
我们开始：
GHCi> (take 10 . popToTheHaltThroughImproperBlocks) fibsHughes
[0,1,1,2,3,5,8,13,21,34]
问题出在纸张上。究竟应该归咎于何处，很大程度上取决于主观解释。我认为这是论文中一个被忽视的错误，因为类型 StreamProcessor 并不像看起来那么直观。
首先关注fibsHughes流的具体例子，它确实包含Get，但它们是常数函数。您必须提供一些参数才能访问流的其余部分。在某种程度上，fibsHughes 的“真实”类型是 SP () b，而您可能直觉上想要的是 SP Void b 来编码 Get 的缺失（这不太有效那样，这就是问题的根源），“喂养”它的输入是你如何从一个到另一个：
type SP = StreamProcessor
feed :: SP () b -> SP Void b
feed p = produceForever () >>> p
produceForever :: b -> SP Void b
produceForever b = Put b (produceForever b)
fibsHughes :: SP Void b
fibsHughes = feed (... {- rest of the definition -})
现在要了解我们是如何陷入这种情况的，我们必须回到 first 的定义。我的观点是，首先定义流上的操作是有问题的，因为它必须引入 Get 操作才能生成对的第二个组件作为输出：
first ::: SP a b -> SP (a, c) (b, c)
有问题的部分是 bypass 定义中的以下分支，它引入了 Get 然后能够 Put:
bypass bs [] (Put c sp) =
Get $ \ ~(b, d) -> Put (c, d) (bypass (bs ++ [b]) [] sp)
如果你想写一些预期类型的东西，这是你需要做的，但可以说这不是一件自然而然的事情。
定义 first 是导致定义和使用 (&&&) 运算符的原因，它具有不直观的语义。要了解它为什么不直观，请将 (&&&) 特化为 Void 作为流输入类型：
(&&&) :: SP Void b -> SP Void c -> SP Void (b, c)
任何人看到这里都会认为，当然，结果必须是一个生产者，这从来没有 Get 任何东西，那是荒谬的。除了 (&&&) 做了荒谬的事情；因此专用于 Void，它在道德上等同于以下内容（忽略 undefined 的存在，它在技术上可用于在 Haskell 中区分它们）：
_ &&& _ = Get (absurd :: Void -> SP a c)
有一个更自然的定义 (&&&) 的流递归避免了这个问题：如果两个参数从不做任何 Get，那么结果永远不会做任何 Get要么。
据我所知，这个“更好的”(&&&) 不能用 first、(>>>) 和 arr 来定义。
然而，这是有代价的：从箭头的图形解释的角度来看，它并不直观，因为它打破了这个等式（可以用图形表示为“滑动”f &&&):
f &&& g = (id &&& g) >>> first f
无论您选择 (&&&) 的哪个定义，都会让人感到困惑。
IMO 归结为 StreamProcessor 类型不能排除 Get 的使用。即使输入类型是Void，流仍然可以做一个空洞的Get.
没有此类定义问题的更好类型的流处理器是来自 pipes library (called Proxy) 的流处理器。特别是，它与 SP 不同，因为它可以禁止使用 Get，并且它提供了真实“生产者”的忠实编码，例如 Fibonacci 流。
我认为您可能有一个困惑是，虽然您可以找到流处理器和可能阻塞的列表之间的相关性，但它们并不完全相同。从道德上讲，StreamProcessor a b 消耗 a 流并产生 b 流。 （Hughes 论文的一个奇怪之处在于他没有明确定义流是什么。）这归结为您的 popToTheBlockOrHalt 函数实际上从未提供输入流，但它仍然期望给定流处理器产生输出流。
另一件要记住的事情是，在 Hughes 的论文中，没有 Halt — 流处理器是无限的，并且在无限流上工作。所以，对于像fibsHughes这样所谓的“生产者”（即输入流是任意的流处理器）来说，即使隐藏了Get，也确实不会发生“阻塞”在内部，因为输入流总是有更多的输入——它是无限的！
因此，您需要使用这些流处理器来 运行 它们，或者将 StreamProcessor a b 形式的东西变成从 a 流到 b 流的函数。因为您的版本允许流是有限的，所以使用常规旧列表作为您的“流”类型是有意义的。因此，您需要一个类型如下的函数：
runStreamProcessor :: StreamProcessor a b -> [a] -> [b]
对此有一个自然的定义：
runStreamProcessor Halt _ = []
runStreamProcessor (Put x s) xs = x : runStreamProcessor s xs
runStreamProcessor _ [] = []
runStreamProcessor (Get f) (x:xs) = runStreamProcessor (f x) xs
现在，您可以考虑 runStreamProcessor fibsHughes :: [a] -> [Integer] 的类型并意识到，您自然必须提供例如 repeat () 以保证无限输出流。这有效：
> take 10 $ runStreamProcessor fibsHughes (repeat ())
[0,1,1,2,3,5,8,13,21,34]
我正在尝试理解 John Hughes 著名的 "Generalising Arrows to Monads" 中的“箭头流”部分。更准确地说，我有兴趣写下斐波那契流。
我稍微调整了 Hughes 的定义：
data StreamProcessor a b = Get (a -> StreamProcessor a b) |
Put b (StreamProcessor a b) |
Halt
put = Put
get = Get
首先，我将流处理器视为可能阻塞（等待输入）的列表。即：
Put :: b -> StreamProcessor a b -> StreamProcessor a b匹配(:) :: a -> [a] -> [a];Halt :: StreamProcessor a b匹配[] :: [a];Get :: (a -> StreamProcessor a b) -> StreamProcessor a b帮助我们阻塞流并等待输入。
因此，如果我们删除 Get，我们会得到一个类似列表的结构。如果我们也删除 Halt，我们会得到一个类似无限列表的结构。
以下是我理解“斐波那契流”的两种方式：
一个非阻塞的无限流（类似无限列表）：
zipNonBlockedStreamsWith :: (a -> b -> c) -> StreamProcessor () a -> StreamProcessor () b -> StreamProcessor () c zipNonBlockedStreamsWith f (Put x sp) (Put y sp') = Put (f x y) (zipNonBlockedStreamsWith f sp sp') zipNonBlockedStreamsWith f Halt sp = Halt zipNonBlockedStreamsWith f sp Halt = Halt fibs :: StreamProcessor () Int fibs = put 0 (put 1 $ zipNonBlockedStreamsWith (+) fibs (tailNonBlockedStream fibs)) -- matches a well-known definition of an infinite Fibonacci list. fibsList :: [Int] fibsList = 0 : 1 : (zipWith (+) fibsList (tail fibsList)) -- with the 'fibsList', we can use folds to do the same thing. putStream :: [b] -> StreamProcessor a b -> StreamProcessor a b putStream bs sp = foldr Put sp bs fibs' :: StreamProcessor () Int fibs' = putStream fibsList Halt一个阻塞流等待
n，输出第n个斐波那契数并再次阻塞。 Hughes 的Arrow界面帮助我们以相当简洁的方式表达它：instance Category StreamProcessor where ... instance Arrow StreamProcessor where arr f = Get $ \ a -> Put (f a) (arr f) ... fibsList :: [Int] fibsList = 0 : 1 : (zipWith (+) fibsList (tail fibsList)) blockedFibs :: StreamProcessor Int Int blockedFibs = arr (fibsList !!)
然而，在 the paper I linked 中，John Hughes 展示了另一个解决方案，Arrow他通过：
instance Category StreamProcessor where
id = Get (\ x -> Put x id)
Put c bc . ab = Put c (bc . ab)
Get bbc . Put b ab = (bbc b) . ab
Get bbc . Get aab = Get $ \ a -> (Get bbc) . (aab a)
Get bbc . Halt = Halt
Halt . ab = Halt
bypass :: [b] -> [d] -> StreamProcessor b c -> StreamProcessor (b, d) (c, d)
bypass [] ds (Get f) = Get $ \ ~(b, d) -> bypass [] (ds ++ [d]) (f b)
bypass (b : bs) [] (Get f) = bypass bs [] (f b)
bypass [] (d : ds) (Put c sp) = Put (c, d) $ bypass [] ds sp
bypass bs [] (Put c sp) =
Get $ \ ~(b, d) -> Put (c, d) (bypass (bs ++ [b]) [] sp)
bypass bs ds Halt = Halt
instance Arrow StreamProcessor where
arr f = Get $ \ a -> Put (f a) (arr f)
first = bypass [] []
liftArr2 :: Arrow k => (a -> b -> c) -> k r a -> k r b -> k r c
liftArr2 f a b = a &&& b >>^ uncurry f
fibsHughes = let
fibsHughes' = put 1 (liftArr2 (+) fibsHughes fibsHughes')
in put 0 fibsHughes'
但它并不像我期望的那样工作。以下函数将帮助我们从流中获取值，直到它阻塞或停止（使用 Data.List.unfoldr）：
popToTheBlockOrHalt :: StreamProcessor a b -> [b]
popToTheBlockOrHalt = let
getOutput (Put x sp) = Just (x, sp)
getOutput getOrHalt = Nothing
in unfoldr getOutput
所以，这是我们得到的：
GHCi> popToTheBlockOrHalt fibsHughes
[0, 1]
GHCi> :t fibsHughes
fibsHughes :: StreamProcessor a Integer
如果我们检查模式，我们会发现它阻塞了（也就是说我们偶然发现 Get）。
我不知道是不是应该这样。如果这是我们想要的，为什么？如果不是，问题是什么？我检查并重新检查了我编写的代码，它与 Hughes 文章中的定义非常匹配（嗯，我不得不添加 id 和 Halt 的模式 - 我看不出他们怎么会搞砸了这个过程向上）。
编辑： 正如评论中所说，在 later edition of the paper bypass was slightly changed (we use that one). It is able to accumulate both withheld bs and ds (that is has two queues), whereas the bypass from the original paper 中仅累积 ds（即一个队列）。
编辑 #2: 我只是想写下一个函数，它会从 fibsHughes:
popToTheHaltThroughImproperBlocks :: StreamProcessor () b -> [b]
popToTheHaltThroughImproperBlocks = let
getOutput (Put x sp) = Just (x, sp)
getOutput (Get c) = getOutput $ c ()
getOutput Halt = Nothing
in unfoldr getOutput
我们开始：
GHCi> (take 10 . popToTheHaltThroughImproperBlocks) fibsHughes
[0,1,1,2,3,5,8,13,21,34]
问题出在纸张上。究竟应该归咎于何处，很大程度上取决于主观解释。我认为这是论文中一个被忽视的错误，因为类型 StreamProcessor 并不像看起来那么直观。
首先关注fibsHughes流的具体例子，它确实包含Get，但它们是常数函数。您必须提供一些参数才能访问流的其余部分。在某种程度上，fibsHughes 的“真实”类型是 SP () b，而您可能直觉上想要的是 SP Void b 来编码 Get 的缺失（这不太有效那样，这就是问题的根源），“喂养”它的输入是你如何从一个到另一个：
type SP = StreamProcessor
feed :: SP () b -> SP Void b
feed p = produceForever () >>> p
produceForever :: b -> SP Void b
produceForever b = Put b (produceForever b)
fibsHughes :: SP Void b
fibsHughes = feed (... {- rest of the definition -})
现在要了解我们是如何陷入这种情况的，我们必须回到 first 的定义。我的观点是，首先定义流上的操作是有问题的，因为它必须引入 Get 操作才能生成对的第二个组件作为输出：
first ::: SP a b -> SP (a, c) (b, c)
有问题的部分是 bypass 定义中的以下分支，它引入了 Get 然后能够 Put:
bypass bs [] (Put c sp) =
Get $ \ ~(b, d) -> Put (c, d) (bypass (bs ++ [b]) [] sp)
如果你想写一些预期类型的东西，这是你需要做的，但可以说这不是一件自然而然的事情。
定义 first 是导致定义和使用 (&&&) 运算符的原因，它具有不直观的语义。要了解它为什么不直观，请将 (&&&) 特化为 Void 作为流输入类型：
(&&&) :: SP Void b -> SP Void c -> SP Void (b, c)
任何人看到这里都会认为，当然，结果必须是一个生产者，这从来没有 Get 任何东西，那是荒谬的。除了 (&&&) 做了荒谬的事情；因此专用于 Void，它在道德上等同于以下内容（忽略 undefined 的存在，它在技术上可用于在 Haskell 中区分它们）：
_ &&& _ = Get (absurd :: Void -> SP a c)
有一个更自然的定义 (&&&) 的流递归避免了这个问题：如果两个参数从不做任何 Get，那么结果永远不会做任何 Get要么。
据我所知，这个“更好的”(&&&) 不能用 first、(>>>) 和 arr 来定义。
然而，这是有代价的：从箭头的图形解释的角度来看，它并不直观，因为它打破了这个等式（可以用图形表示为“滑动”f &&&):
f &&& g = (id &&& g) >>> first f
无论您选择 (&&&) 的哪个定义，都会让人感到困惑。
IMO 归结为 StreamProcessor 类型不能排除 Get 的使用。即使输入类型是Void，流仍然可以做一个空洞的Get.
没有此类定义问题的更好类型的流处理器是来自 pipes library (called Proxy) 的流处理器。特别是，它与 SP 不同，因为它可以禁止使用 Get，并且它提供了真实“生产者”的忠实编码，例如 Fibonacci 流。
我认为您可能有一个困惑是，虽然您可以找到流处理器和可能阻塞的列表之间的相关性，但它们并不完全相同。从道德上讲，StreamProcessor a b 消耗 a 流并产生 b 流。 （Hughes 论文的一个奇怪之处在于他没有明确定义流是什么。）这归结为您的 popToTheBlockOrHalt 函数实际上从未提供输入流，但它仍然期望给定流处理器产生输出流。
另一件要记住的事情是，在 Hughes 的论文中，没有 Halt — 流处理器是无限的，并且在无限流上工作。所以，对于像fibsHughes这样所谓的“生产者”（即输入流是任意的流处理器）来说，即使隐藏了Get，也确实不会发生“阻塞”在内部，因为输入流总是有更多的输入——它是无限的！
因此，您需要使用这些流处理器来 运行 它们，或者将 StreamProcessor a b 形式的东西变成从 a 流到 b 流的函数。因为您的版本允许流是有限的，所以使用常规旧列表作为您的“流”类型是有意义的。因此，您需要一个类型如下的函数：
runStreamProcessor :: StreamProcessor a b -> [a] -> [b]
对此有一个自然的定义：
runStreamProcessor Halt _ = []
runStreamProcessor (Put x s) xs = x : runStreamProcessor s xs
runStreamProcessor _ [] = []
runStreamProcessor (Get f) (x:xs) = runStreamProcessor (f x) xs
现在，您可以考虑 runStreamProcessor fibsHughes :: [a] -> [Integer] 的类型并意识到，您自然必须提供例如 repeat () 以保证无限输出流。这有效：
> take 10 $ runStreamProcessor fibsHughes (repeat ())
[0,1,1,2,3,5,8,13,21,34]