Haskell 中字节流的高效流式传输和操作

Question

在为大型(<bloblength><blob>)* 编码二进制文件编写反序列化器时，我被各种 Haskell 生产-转换-消费库困住了。到目前为止，我知道有四个流媒体库：

• Data.Conduit: 使用广泛，资源管理非常细心
• Pipes：类似于conduit（Haskell Cast #6很好地揭示了conduit和pipes之间的区别）
• Data.Binary.Get：提供了getWord32be等有用的函数，但是流式示例很尴尬
• System.IO.Streams: 好像是最容易用的一个

这是一个简化的示例，说明当我尝试使用conduit 进行Word32 流式传输时出现问题。一个稍微现实一点的例子是首先读取一个确定 blob 长度的Word32，然后产生一个该长度的惰性ByteString（然后进一步反序列化）。 但在这里我只是尝试从二进制文件中以流方式提取 Word32：

module Main where

-- build-depends: bytestring, conduit, conduit-extra, resourcet, binary

import           Control.Monad.Trans.Resource (MonadResource, runResourceT)
import qualified Data.Binary.Get              as G
import qualified Data.ByteString              as BS
import qualified Data.ByteString.Char8        as C
import qualified Data.ByteString.Lazy         as BL
import           Data.Conduit
import qualified Data.Conduit.Binary          as CB
import qualified Data.Conduit.List            as CL
import           Data.Word                    (Word32)
import           System.Environment           (getArgs)

-- gets a Word32 from a ByteString.
getWord32 :: C.ByteString -> Word32
getWord32 bs = do
    G.runGet G.getWord32be $ BL.fromStrict bs

-- should read BytesString and return Word32
transform :: (Monad m, MonadResource m) => Conduit BS.ByteString m Word32
transform = do
    mbs <- await
    case mbs of
        Just bs -> do
            case C.null bs of
                False -> do
                    yield $ getWord32 bs
                    leftover $ BS.drop 4 bs
                    transform
                True -> return ()
        Nothing -> return ()

main :: IO ()
main = do
    filename <- fmap (!!0) getArgs  -- should check length getArgs
    result <- runResourceT $ (CB.sourceFile filename) $$ transform =$ CL.consume
    print $ length result   -- is always 8188 for files larger than 32752 bytes

程序的输出只是读取的 Word32 的数量。事实证明，流在读取第一个块（大约 32KiB）后终止。出于某种原因，mbs 永远不是Nothing，所以我必须检查null bs，它会在块被消耗时停止流。显然，我的导管transform 有问题。我看到了两种解决方案：

1. await 不想去ByteStream 的第二个块，那么还有另一个函数可以拉下一个块吗？在我见过的例子中（例如Conduit 101），这不是它的做法
2. 这只是设置transform 的错误方式。

这是如何正确完成的？这是正确的方法吗？ （性能确实很重要。）

更新：这是使用Systems.IO.Streams的BAD方法：

module Main where

import           Data.Word                (Word32)
import           System.Environment       (getArgs)
import           System.IO                (IOMode (ReadMode), openFile)
import qualified System.IO.Streams        as S
import           System.IO.Streams.Binary (binaryInputStream)
import           System.IO.Streams.List   (outputToList)

main :: IO ()
main = do
    filename : _ <- getArgs
    h <- openFile filename ReadMode
    s <- S.handleToInputStream h
    i <- binaryInputStream s :: IO (S.InputStream Word32)
    r <- outputToList $ S.connect i
    print $ last r

'Bad'表示：对时间和空间要求很高，不处理Decode异常。

Answer 1

您的直接问题是由您使用leftover 的方式引起的。该函数用于“提供单个剩余输入以供当前单子绑定中的下一个组件使用”，因此当您在使用 transform 循环之前给它 bs 时，您实际上是在扔掉其余的字节串（即bs 之后的内容）。

基于您的代码的正确解决方案将使用 the incremental input interface 或 Data.Binary.Get 将您的 yield/leftover 组合替换为完全消耗每个块的东西。不过，更实用的方法是使用 binary-conduit 包，它以conduitGet 的形式提供（它的source 很好地了解了“手动”实现的外观喜欢）：

import           Data.Conduit.Serialization.Binary

-- etc.

transform :: (Monad m, MonadResource m) => Conduit BS.ByteString m Word32
transform = conduitGet G.getWord32be

需要注意的是，如果字节总数不是 4 的倍数（即最后一个 Word32 不完整），这将引发解析错误。在不太可能的情况下，这不是您想要的，一个懒惰的出路是简单地在输入字节串上使用\bs -> C.take (4 * truncate (C.length bs / 4)) bs。

Answer 2

使用pipes（以及pipes-group 和pipes-bytestring），演示问题简化为组合子。首先，我们将传入的未区分字节流解析为 4 字节的小块：

chunksOfStrict :: (Monad m) => Int -> Producer ByteString m r -> Producer ByteString m r
chunksOfStrict n = folds mappend mempty id . view (Bytes.chunksOf n) 

然后我们将这些映射到Word32s 并（在这里）计算它们。

main :: IO ()
main = do
   filename:_ <- getArgs
   IO.withFile filename IO.ReadMode $ \h -> do
     n <- P.length $ chunksOfStrict 4 (Bytes.fromHandle h) >-> P.map getWord32
     print n

如果我们的字节数少于 4 或无法解析，这将失败，但我们也可以使用

进行映射

getMaybeWord32 :: ByteString -> Maybe Word32
getMaybeWord32 bs = case  G.runGetOrFail G.getWord32be $ BL.fromStrict bs of
  Left r -> Nothing
  Right (_, off, w32) -> Just w32

下面的程序将打印出有效的 4 字节序列的解析

main :: IO ()
main = do
   filename:_ <- getArgs
   IO.withFile filename IO.ReadMode $ \h -> do
     runEffect $ chunksOfStrict 4 (Bytes.fromHandle h) 
                 >-> P.map getMaybeWord32
                 >-> P.concat  -- here `concat` eliminates maybes
                 >-> P.print 

当然，还有其他处理失败解析的方法。

不过，这里更接近您要求的程序。它从字节流 (Producer ByteString m r) 中获取一个四字节段，如果它足够长，则将其读取为 Word32；然后它需要 那么多 的传入字节并将它们累积成一个惰性字节串，从而产生它。它只是重复这个直到它用完字节。在下面的main 中，我打印了每个产生的惰性字节串：

module Main (main) where 
import Pipes 
import qualified Pipes.Prelude as P
import Pipes.Group (folds) 
import qualified Pipes.ByteString as Bytes ( splitAt, fromHandle, chunksOf )
import Control.Lens ( view ) -- or Lens.Simple (view) -- or Lens.Micro ((.^))
import qualified System.IO as IO ( IOMode(ReadMode), withFile )
import qualified Data.Binary.Get as G ( runGet, getWord32be )
import Data.ByteString ( ByteString )
import qualified Data.ByteString.Lazy.Char8 as BL 
import System.Environment ( getArgs )

splitLazy :: (Monad m, Integral n) =>
   n -> Producer ByteString m r -> m (BL.ByteString, Producer ByteString m r)
splitLazy n bs = do
  (bss, rest) <- P.toListM' $ view (Bytes.splitAt n) bs
  return (BL.fromChunks bss, rest)

measureChunks :: Monad m => Producer ByteString m r -> Producer BL.ByteString m r
measureChunks bs = do
 (lbs, rest) <- lift $ splitLazy 4 bs
 if BL.length lbs /= 4
   then rest >-> P.drain -- in fact it will be empty
   else do
     let w32 = G.runGet G.getWord32be lbs
     (lbs', rest') <- lift $ splitLazy w32 bs
     yield lbs
     measureChunks rest

main :: IO ()
main = do
  filename:_ <- getArgs
  IO.withFile filename IO.ReadMode $ \h -> do
     runEffect $ measureChunks (Bytes.fromHandle h) >-> P.print

这又是粗略的，因为它使用runGet 而不是runGetOrFail，但这很容易修复。管道标准过程是在解析失败时停止流转换并返回未解析的字节流。

如果您预计Word32s 用于大数字，因此您不想将相应的字节流累积为惰性字节串，而是说将它们写入不同的文件而不累积，我们可以更改程序很容易做到这一点。这需要复杂地使用管道，但这是pipes 和streaming 的首选方法。

Answer 3

这是一个相对简单的解决方案，我想把它扔进擂台。这是对splitAt 的重复使用，它被包装到State monad 中，它提供了与Data.Binary.Get （的子集）相同的接口。生成的[ByteString] 在main 中获得，whileJust 超过getBlob。

module Main (main) where

import           Control.Monad.Loops
import           Control.Monad.State
import qualified Data.Binary.Get      as G (getWord32be, runGet)
import qualified Data.ByteString.Lazy as BL
import           Data.Int             (Int64)
import           Data.Word            (Word32)
import           System.Environment   (getArgs)

-- this is going to mimic the Data.Binary.Get.Get Monad
type Get = State BL.ByteString

getWord32be :: Get (Maybe Word32)
getWord32be = state $ \bs -> do
    let (w, rest) = BL.splitAt 4 bs
    case BL.length w of
        4 -> (Just w', rest) where
            w' = G.runGet G.getWord32be w
        _ -> (Nothing, BL.empty)

getLazyByteString :: Int64 -> Get BL.ByteString
getLazyByteString n = state $ \bs -> BL.splitAt n bs

getBlob :: Get (Maybe BL.ByteString)
getBlob = do
    ml <- getWord32be
    case ml of
        Nothing -> return Nothing
        Just l -> do
            blob <- getLazyByteString (fromIntegral l :: Int64)
            return $ Just blob

runGet :: Get a -> BL.ByteString -> a
runGet g bs = fst $ runState g bs

main :: IO ()
main = do
    fname <- head <$> getArgs
    bs <- BL.readFile fname
    let ls = runGet loop bs where
        loop = whileJust getBlob return
    print $ length ls

getBlob 中没有错误处理，但很容易扩展。时间和空间复杂度相当不错，只要谨慎使用结果列表即可。 （上面创建一些随机数据供上面消费的python脚本是here）。