无限量的有效行动

Question

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

更具体地说，我有一个无限的type流[IO(ByteString)]。另一方面，我有一个纯解析函数parse :: [ByteString] -> [Object]（ObjectHaskell数据类型在哪里）

有没有办法将我的无限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

只需要一部分结果就没关系。在sequence完成整个操作之前，您不会从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机制非常简单明了，但由于资源管理问题，空间泄漏的脆弱性（对代码进行小的更改会消耗内存空间），惰性IO仍不适合生产代码。 ，以及异常处理问题。

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

对于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]

我敢肯定，您不能将其直接与管道集成在一起（或者至少不能以不会放弃使用管道的所有优势的任何方式）。您需要对其进行重写以使其在使用字节字符串和产生对象方面更加友好。