haskell - Haskell 中的高效比特流

Question

在不断努力有效地摆弄位(例如，参见此 SO question)中，最新的挑战是位的有效流传输和消耗。

作为第一个简单的任务，我选择在 /dev/urandom 生成的比特流中找到最长的相同比特序列。 .典型的咒语是 head -c 1000000 </dev/urandom | my-exe .实际目标是流式传输比特并解码 Elias gamma code ，例如，即不是字节块或其倍数的代码。

对于这种可变长度的代码，最好使用 take , takeWhile , group等用于列表操作的语言。自 BitStream.take实际上会消耗一些 monad 可能会发挥作用的双流的一部分。

明显的起点是来自 Data.ByteString.Lazy 的惰性字节串。 .

A. 计数字节

正如预期的那样，这个非常简单的 Haskell 程序的性能与 C 程序相当。

import qualified Data.ByteString.Lazy as BSL

main :: IO ()
main = do
    bs <- BSL.getContents
    print $ BSL.length bs

B. 添加字节

一旦我开始使用 unpack事情应该变得更糟。

main = do
    bs <- BSL.getContents
    print $ sum $ BSL.unpack bs

令人惊讶的是，Haskell 和 C 表现出几乎相同的性能。

C. 相同位的最长序列

作为第一个重要任务，可以像这样找到最长的相同位序列:

module Main where

import           Data.Bits            (shiftR, (.&.))
import qualified Data.ByteString.Lazy as BSL
import           Data.List            (group)
import           Data.Word8           (Word8)

splitByte :: Word8 -> [Bool]
splitByte w = Prelude.map (\i-> (w `shiftR` i) .&. 1 == 1) [0..7]

bitStream :: BSL.ByteString -> [Bool]
bitStream bs = concat $ map splitByte (BSL.unpack bs)

main :: IO ()
main = do
    bs <- BSL.getContents
    print $ maximum $ length <$> (group $ bitStream bs)

惰性字节串转换为列表 [Word8]然后，使用类次，每个 Word被拆分成位，产生一个列表 [Bool] .这个列表列表然后用 concat 扁平化。 .已获得 Bool 的(懒惰)列表, 使用 group将列表拆分为相同位的序列，然后映射 length超过它。最后maximum给出了想要的结果。很简单，但不是很快:

# C
real    0m0.606s

# Haskell
real    0m6.062s

这种幼稚的实现恰好慢了一个数量级。

分析显示分配了相当多的内存(大约 3GB 用于解析 1MB 的输入)。不过，没有观察到大量的空间泄漏。

从这里我开始四处寻找:

有一个 bitstream package promise “具有半自动流融合的快速、打包、严格的位流(即 bool 列表)。”。不幸的是，它不是最新的 vector包，见 here详情。

接下来我调查 streaming .我不太明白为什么我需要“有效”的流媒体来让一些 monad 发挥作用——至少在我开始与所提出的任务相反的时候，即将比特流编码和写入文件。

只是fold怎么样？ -ing 在 ByteString ?我必须引入状态来跟踪消耗的位。那不是很好take , takeWhile , group等语言是可取的。

现在我不太确定该去哪里。

更新 :

我想出了如何用 streaming 做到这一点和 streaming-bytestring .我可能做得不对，因为结果是灾难性的。

import           Data.Bits                 (shiftR, (.&.))
import qualified Data.ByteString.Streaming as BSS
import           Data.Word8                (Word8)
import qualified Streaming                 as S
import           Streaming.Prelude         (Of, Stream)
import qualified Streaming.Prelude         as S

splitByte :: Word8 -> [Bool]
splitByte w = (\i-> (w `shiftR` i) .&. 1 == 1) <$> [0..7]

bitStream :: Monad m => Stream (Of Word8) m () -> Stream (Of Bool) m ()
bitStream s = S.concat $ S.map splitByte s

main :: IO ()
main = do
    let bs = BSS.unpack BSS.getContents :: Stream (Of Word8) IO ()
        gs = S.group $ bitStream bs ::  Stream (Stream (Of Bool) IO) IO ()
    maxLen <- S.maximum $ S.mapped S.length gs
    print $ S.fst' maxLen

这将考验您对来自标准输入的数千字节输入以外的任何内容的耐心。分析器说它在 Streaming.Internal.>>=.loop 中花费了疯狂的时间(输入大小的二次方)和 Data.Functor.Of.fmap .我不太确定第一个是什么，但 fmap表示 (?) 这些 Of a b 的杂耍对我们没有任何好处，因为我们处于 IO monad 中，因此无法对其进行优化。

我还有字节加法器的流等效项 here: SumBytesStream.hs ，这比简单的懒惰 ByteString 稍慢实现，但仍然体面。自 streaming-bytestring是 proclaimed要成为“正确完成字节串 io”，我期望更好。那我可能做得不对。

在任何情况下，所有这些位计算都不应该发生在 IO monad 中。但是BSS.getContents强制我进入 IO monad 因为 getContents :: MonadIO m => ByteString m ()而且没有出路。

更新 2

按照@dfeuer 的建议，我使用了 streaming 包在 master@HEAD。结果如下。

longest-seq-c       0m0.747s    (C)
longest-seq         0m8.190s    (Haskell ByteString)
longest-seq-stream  0m13.946s   (Haskell streaming-bytestring)

Streaming.concat 的 O(n^2) 问题已经解决了，但我们仍然没有接近 C 基准。

更新 3

Cirdec 的解决方案产生了与 C 相当的性能。使用的构造称为“教堂编码列表”，请参阅此 SO answer或 Haskell Wiki 上的 rank-N types .

源文件:

所有源文件都可以在 github 上找到. Makefile具有运行实验和分析的所有各种目标。默认make将构建所有内容(首先创建一个 bin/ 目录!)然后是 make time将在 longest-seq 上做计时可执行文件。 C 可执行文件得到一个 -c附加以区分它们。

Answer 1

当对流的操作融合在一起时，可以删除中间分配及其相应的开销。 GHC prelude 以 rewrite rules 的形式为惰性流提供了折叠/构建融合。 .一般的想法是，如果一个函数产生一个看起来像文件夹的结果(它的类型 (a -> b -> b) -> b -> b 应用于 (:) 和 [] )，而另一个函数使用一个看起来像文件夹的列表，构建中间列表可以被移除。

对于您的问题，我将构建类似的东西，但使用严格的左折叠( foldl' )而不是 foldr。而不是使用重写规则来尝试检测何时看起来像 foldl ，我将使用强制列表看起来像左折叠的数据类型。

-- A list encoded as a strict left fold.
newtype ListS a = ListS {build :: forall b. (b -> a -> b) -> b -> b}

由于我已经放弃了列表，因此我们将重新实现列表前奏的一部分。

可以从 foldl' 创建严格的左折叠列表和字节串的函数。

{-# INLINE fromList #-}
fromList :: [a] -> ListS a
fromList l = ListS (\c z -> foldl' c z l)

{-# INLINE fromBS #-}
fromBS :: BSL.ByteString -> ListS Word8
fromBS l = ListS (\c z -> BSL.foldl' c z l)

使用一个的最简单的例子是找到一个列表的长度。

{-# INLINE length' #-}
length' :: ListS a -> Int
length' l = build l (\z a -> z+1) 0

我们还可以映射和连接左折叠。

{-# INLINE map' #-}
-- fmap renamed so it can be inlined
map' f l = ListS (\c z -> build l (\z a -> c z (f a)) z)

{-# INLINE concat' #-}
concat' :: ListS (ListS a) -> ListS a
concat' ll = ListS (\c z -> build ll (\z l -> build l c z) z)

对于您的问题，我们需要能够将一个字分成几位。

{-# INLINE splitByte #-}
splitByte :: Word8 -> [Bool]
splitByte w = Prelude.map (\i-> (w `shiftR` i) .&. 1 == 1) [0..7]

{-# INLINE splitByte' #-}
splitByte' :: Word8 -> ListS Bool
splitByte' = fromList . splitByte

还有一个 ByteString成比特

{-# INLINE bitStream' #-}
bitStream' :: BSL.ByteString -> ListS Bool
bitStream' = concat' . map' splitByte' . fromBS

为了找到最长的运行，我们将跟踪前一个值、当前运行的长度和最长运行的长度。我们使字段变得严格，这样折叠的严格性可以防止 thunk 链在内存中积累。为状态创建严格的数据类型是一种控制其内存表示和何时评估其字段的简单方法。

data LongestRun = LongestRun !Bool !Int !Int

{-# INLINE extendRun #-}
extendRun (LongestRun previous run longest) x = LongestRun x current (max current longest)
  where
    current = if x == previous then run + 1 else 1

{-# INLINE longestRun #-}
longestRun :: ListS Bool -> Int
longestRun l = longest
 where
   (LongestRun _ _ longest) = build l extendRun (LongestRun False 0 0)

我们完成了

main :: IO ()
main = do
    bs <- BSL.getContents
    print $ longestRun $ bitStream' bs

这要快得多，但不完全是 c 的性能。

longest-seq-c       0m00.12s    (C)
longest-seq         0m08.65s    (Haskell ByteString)
longest-seq-fuse    0m00.81s    (Haskell ByteString fused)

程序分配了大约 1 Mb 从输入中读取 1000000 字节。

total alloc =   1,173,104 bytes  (excludes profiling overheads)

更新 github code