可以阻塞的流处理器的 ArrowCircuit 实例

Question

Control.Arrow.Operations.ArrowCircuit class 适用于：

An arrow type that can be used to interpret synchronous circuits.

我想知道这里的同步是什么意思。我在 Wikipedia 上查找了它，他们在那里谈论数字电子产品。我的电子设备生锈了，所以问题来了：所谓的异步流处理器的这种实例有什么问题（如果有的话）：

data StreamProcessor a b = Get (a -> StreamProcessor a b) | 
                           Put b    (StreamProcessor a b) |
                           Halt

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

instance Arrow StreamProcessor where
    ...

getThroughBlocks :: [a] -> StreamProcessor a b -> StreamProcessor a b
getThroughBlocks ~(a : input) (Get f)   = getThroughBlocks input (f a)
getThroughBlocks _input       putOrHalt = putOrHalt

getThroughSameArgBlocks :: a -> StreamProcessor a b -> StreamProcessor a b
getThroughSameArgBlocks = getThroughBlocks . repeat

instance ArrowLoop StreamProcessor where
    loop Halt               = Halt
    loop (Put (c, d) bdcd') = Put c (loop bdcd')
    loop (Get f)            = Get $ \ b -> 
         let 
            Put (c, d) bdcd' = getThroughSameArgBlocks (b, d) (f (b, d))
         in Put c (loop bdcd')

instance ArrowCircuit StreamProcessor where
    delay b = Put b id

我认为这个解决方案对我们有用：我们希望 someArrowCircuit >>> delay b 被 someArrowCircuit 延迟一个滴答声，并且 b 出现在它之前。很容易看出我们得到了我们想要的：

someArrowCircuit >>> delay b
= someArrowCircuit >>> Put b id 
= Put b id . someArrowCircuit
= Put b (id . someArrowCircuit)
= Put b someArrowCircuit

对于这样的 class 有法律规定吗？如果我没有记错 delay，那么 synchronous 如何与 asynchronous 并存？

Answer 1

我所知道的与 ArrowCircuit 相关的唯一定律实际上是针对来自 Causal Commutative Arrows 的类似 ArrowInit class，它表示 delay i *** delay j = delay (i,j)。我很确定你的版本满足这个（它看起来是一个完全合理的实现），但考虑到 StreamProcessor 不是同步的，它仍然感觉有点奇怪。

特别是，同步电路遵循单一输入产生单一输出的模式。例如，如果你有一个 Circuit a b 并为它提供一个 a 类型的值，那么你将得到一个且只有一个输出 b。 delay引入的“一滴延迟”因此是一个输出一步的延迟。

但是对于异步电路来说事情有点奇怪。让我们考虑一个例子：

runStreamProcessor :: StreamProcessor a b -> [a] -> [b]
runStreamProcessor (Put x s) xs = x : runStreamProcessor s xs
runStreamProcessor _ [] = []
runStreamProcessor Halt _ = []
runStreamProcessor (Get f) (x:xs) = runStreamProcessor (f x) xs

multiplyOneThroughFive :: StreamProcessor Int Int
multiplyOneThroughFive = Get $ \x -> 
  Put (x*1) $ Put (x*2) $ Put (x*3) $ Put (x*4) $ Put (x*5) multiplyOneThroughFive

此处，multiplyOneThroughFive 为其接收的每个输入生成 5 个输出。现在，考虑 multiplyOneThroughFive >>> delay 100 和 delay 100 >>> multiplyOneThroughFive 之间的区别：

> runStreamProcessor (multiplyOneThroughFive >>> delay 100) [1,2]
[100,1,2,3,4,5,2,4,6,8,10]
> runStreamProcessor (delay 100 >>> multiplyOneThroughFive) [1,2]
[100,200,300,400,500,1,2,3,4,5,2,4,6,8,10]

在电路的不同点插入 delay 实际上导致我们产生不同数量的结果。事实上，整个电路似乎经历了 5 个滴答的延迟，而不是仅仅 1 个滴答的延迟。这绝对是同步环境中的意外行为！

可以阻塞的流处理器的 ArrowCircuit 实例

An ArrowCircuit instance for stream processors which could block

haskell

functional-programming

circuit

stream-processing

typeclass