休斯(Fibonacci)数列

Question

我正在尝试理解约翰·休斯著名的《将箭头泛化为单子》 ("Generalising Arrows to Monads") 中的“Streams as arrows”部分。更确切地说，我对编写斐波那契流感兴趣。

我稍微调整了休斯的定义：

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，我们得到一个类似于无限列表的结构。

以下是我理解“斐波那契数列的流”两种方式：

• a non-blocked infinite stream (infinite-list-like):

  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
  

• A blocked stream waits for n, outputs the nth Fibonacci number and blocks again. Hughes' Arrow interface helps us express it in a quite concise way:

  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 !!)
  

然而，在我提供的论文中，约翰·休斯展示了另一种解决方案，通过箭头的方式进行：

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的模式 - 我看不出他们如何能搞砸这个过程）。

---

编辑： 正如评论中所说，在这篇论文的后期版本中，bypass稍作修改（我们使用这个版本）。它能够累积被扣留的b和（即有两个队列），而原始论文中的bypass只能累积（即一个队列）。

---

编辑 #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]

- Zhiltsoff Igor

---

这个链接论文中的bypass是否是修改过的版本？我认为问题在于(&&&)的定义，但我仍在努力确定核心问题。 - Li-yao Xia

@Li-yaoXia，这篇文章在另一版中发布，发表于2000年。我会寻找链接并将其添加到帖子中。 - Zhiltsoff Igor

@Li-yaoXia 我已经更改了链接。早期版本的链接仍然在编辑中。 - Zhiltsoff Igor

Answer 1

问题在于这篇论文。责任的具体归属大多数情况下是主观解释的问题。我认为这是一种被忽视的bug，因为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 库 的一个（称为 Proxy）。特别是，它与 SP 不同，因为它可以禁止使用 Get，而这提供了对真正的“生产者”（如斐波那契流）的忠实编码。

- Li-yao Xia

---

它确实包含Get，但它们是常量函数。很奇怪的是，当我模式匹配“Get f”并尝试将“f”应用于不同的参数时，它就会挂起。对我来说，它似乎真的像是某种底部。你是否让“f”生成了合法的输出，并且与“bypass”（使用两个队列）或“bypass'”（使用一个队列）一起工作？ - Zhiltsoff Igor

是的，我从您的帖子中获取了代码，并对其进行了微调以获得处理结果，并从该流中获得了斐波那契数列：https://gist.github.com/Lysxia/b864da64ad56f17314f8924232c8c17b - Li-yao Xia

你是对的，抱歉。我也遇到了这个“Get”-“Put”迭代：没有底部 :)。 - Zhiltsoff Igor

为什么你想要 feed :: SP () x -> SP Void x？对于不需要输入的流来说，自然类型确实只是 SP () x。你只需要将包含 Get 的 SP () x “规范化” 为不包含 Get 的 SP () x（即 [x]）。 - HTNW

@HTNW，正如“代理”类型所示，我们可能直观地尝试表达的“SP Void x”（即流无法等待）需要以不同的方式表达。在“代理”的情况下，通过将平淡无奇的“等待”替换为更丰富的“请求”来管理它。 - dfeuer

Answer 2

我认为你可能会有一个误解，即尽管您可以找到流处理器和可能阻塞的列表之间的相关性，但它们并不完全相同。道德上讲，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]

- DDub

---

谢谢你的回答。我认为你没有正确理解问题。能够运行fibsHughes流的函数在编辑＃2 - popToTheHaltThroughImproperBlocks中写下了（虽然我是另一种方式 - 感谢您展示第二种方式）。问题是为什么第一个位置会有这些块。您对将fibsHughes称为生成者的看法是正确的，但我认为没有理由允许（甚至放置）Gets到生产者中。 - Zhiltsoff Igor

我的观点是内部细节并不重要 - 正如我所说，鉴于无限流的领域，它们根本不是块。这就像在代码中有“no-ops” - 当然，它们不是必需的，但是对于它们的缺席，任何适当的视图都无法区分它们。 - DDub