【问题标题】:Infinite stream of effectful actions无限的有效行动流
【发布时间】:2019-08-27 18:11:15
【问题描述】:

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

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

有没有办法将我的无限单子流插入我的解析函数?

例如,是否可以编写[IO(ByteString)] -> IO [ByteString] 类型的函数以便我在monad 中使用我的函数parse

【问题讨论】:

  • 如前所述,这需要“惰性 IO”——你可以用谷歌搜索,但请注意它带有一些微妙的问题,例如IO 将在程序中基本上不可预测的点执行。我宁愿尝试重写parse 以使用管道或导管,或其他一些用于“IO 流”的库。
  • 好的,谢谢。例如[IO ByteString] 和 Haskell Stream 有什么区别?
  • @abitbol 你指的是哪个Streampipesconduit 是两个不同的库,它们提供自己的无限流概念实现。
  • 其实我对管道和导管流只有一个模糊的直觉,恐怕我会误导你。我认为您应该能够将 [IO a] 转换为管道 StreamSource IO a 或类似的东西。
  • @radrow 他们不是,因为sequenceIO 和无限列表不会终止。

标签: haskell stream effects infinite


【解决方案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 机制非常简单明了,但由于资源管理问题、空间泄漏方面的脆弱性(对代码稍作改动会导致内存占用),以及异常处理问题。

一个解决方案是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]

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

【讨论】:

  • 注意,如果我们有parse1这样的函数,那么问题就简单多了。 fmap (fmap parse1) inputs :: [IO Object] 已经向前迈出了一大步。相反,我预计这里的一个主要问题是我们不知道解析器必须消耗多少字节串才能生成一个对象,并且最后一个字节串也可能只被部分消耗。当然,一个合适的管道解析器应该以一种很好的方式忽略这些问题,因为它本质上是在生成一个对象之前等待足够的字节。
  • 感谢您非常详细的回复!它有助于我理解 Haskell 中这种 IO 处理的习惯用法。
猜你喜欢
  • 1970-01-01
  • 1970-01-01
  • 2014-05-25
  • 1970-01-01
  • 2012-08-31
  • 2015-06-02
  • 1970-01-01
  • 2012-07-13
相关资源
最近更新 更多