haskell - 无穷无尽的有效行动

Question

我想将无限的字节流解析为无限的 Haskell 数据流。每个字节都是从网络读取的，因此它们被包装到 IO monad 中。

更具体地说，我有一个类型为[IO(ByteString)]的无限流。另一方面，我有一个纯解析函数 parse::[ByteString] -> [Object] (其中 Object 是 Haskell 数据类型)

有没有办法将我的无限 monad 流插入到我的解析函数中？

例如，是否可以编写一个类型为 [IO(ByteString)] -> IO [ByteString] 的函数，以便我使用我的函数 parse在一个单子(monad)中？

Answer 1

问题

一般来说，为了使 IO 操作正确排序并可预测地运行，每个操作都需要在下一个操作运行之前完全完成。在 do block 中，这意味着它可以工作:

main = do
    sequence (map putStrLn ["This","action","will","complete"])
    putStrLn "before we get here"

但不幸的是，如果最终的 IO 操作很重要，这将不起作用:

dontRunMe = do
    putStrLn "This is a problem when an action is"
    sequence (repeat (putStrLn "infinite"))
    putStrLn "<not printed>"

因此，即使 sequence 可以专门用于正确的类型签名:

sequence :: [IO a] -> IO [a]

它无法在无限的 IO 操作列表上按预期工作。定义这样的序列不会有任何问题:

badSeq :: IO [Char]
badSeq = sequence (repeat (return '+'))

但是任何执行 IO 操作的尝试(例如，通过尝试打印结果列表的头部)都会挂起:

main = (head <$> badSeq) >>= print

如果您只需要结果的一部分，也没关系。在整个序列完成之前，您不会从 IO monad 中获取任何内容(因此，如果列表是无限的，则“永远不会”)。

“惰性 IO”解决方案

如果您想从部分完成的 IO 操作中获取数据，您需要明确说明并使用听起来吓人的 Haskell 逃生舱口，unsafeInterleaveIO。该函数接受 IO 操作并“推迟”它，以便在需要该值之前它不会实际执行。

这通常是不安全的原因是，现在有意义的 IO 操作如果在以后的时间点实际执行，可能意味着不同的东西。举一个简单的例子，截断/删除文件的 IO 操作如果在写入更新的文件内容之前和之后执行，则会产生非常不同的效果!

无论如何，您想要在这里做的是编写 sequence 的惰性版本:

import System.IO.Unsafe (unsafeInterleaveIO)

lazySequence :: [IO a] -> IO [a]
lazySequence [] = return []  -- oops, not infinite after all
lazySequence (m:ms) = do
  x <- m
  xs <- unsafeInterleaveIO (lazySequence ms)
  return (x:xs)

这里的关键点是，当执行lazySequence infstream操作时，它实际上只会执行第一个操作；其余操作将包含在延迟 IO 操作中，该操作在需要返回列表的第二个及后续元素之前不会真正执行。

这适用于虚假 IO 操作:

> take 5 <$> lazySequence (repeat (return ('+'))
"+++++"
>

(如果将 lazySequence 替换为 sequence，它将挂起)。它也适用于真实的 IO 操作:

> lns <- lazySequence (repeat getLine)
<waits for first line of input, then returns to prompt>
> print (head lns)
<prints whatever you entered>
> length (head (tail lns))  -- force next element
<waits for second line of input>
<then shows length of your second line before prompt>
>

无论如何，通过 lazySequence 和类型的定义:

parse :: [ByteString] -> [Object]
input :: [IO ByteString]

你应该没有问题写:

outputs :: IO [Object]
outputs = parse <$> lazySequence inputs

然后随心所欲地使用它:

main = do
    objs <- outputs
    mapM_ doSomethingWithObj objs

使用管道

尽管上述延迟 IO 机制非常简单明了，但由于资源管理问题、空间泄漏方面的脆弱性(代码的一个小改动就会导致内存占用)以及异常处理问题。

一个解决方案是conduit 库。另一个是管道。两者都是精心设计的流媒体库，可以支持无限流。

对于conduit，如果您有一个为每个字节字符串创建一个对象的解析函数，例如:

parse1 :: ByteString -> Object
parse1 = ...

然后给出:

inputs :: [IO ByteString]
inputs = ...

useObject :: Object -> IO ()
useObject = ...

管道看起来像:

import Conduit

main :: IO ()
main = runConduit $  mapM_ yieldM inputs
                  .| mapC parse1
                  .| mapM_C useObject

鉴于您的解析函数具有签名:

parse :: [ByteString] -> [Object]

我很确定您不能直接将其与管道集成(或者至少不能以任何不会放弃使用管道的所有好处的方式)。您需要将其重写为管道友好的方式来使用字节字符串和生成对象。