Эффективные потоки битов в Haskell

Ответ 1

Промежуточные распределения и их соответствующие накладные расходы могут быть удалены, когда операции с потоками сливаются вместе. Прелюдия GHC обеспечивает фальсификацию fold/build для ленивых потоков в виде правил перезаписи. Общая идея состоит в том, что если одна функция создает результат, который выглядит как foldr (у него есть тип (a → b → b) → b → b примененный к (:) и []), а другая функция расходует список который выглядит как foldr, можно создать промежуточный список.

Для вашей проблемы я собираюсь построить что-то подобное, но используя строгие левые складки (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

Чтобы найти самый длинный прогон, мы будем отслеживать предыдущее значение, длину текущего прогона и длину самого длинного пробега. Мы создаем поля строгими, чтобы строгость складки препятствовала накоплению накопленных в памяти цепей. Создание строгого типа данных для состояния - это простой способ получить контроль над представлением своей памяти и при оценке его полей.

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 Мб для чтения 1000000 байт с ввода.

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

Обновлен код github

Ответ 2

Я нашел другое решение, которое находится на одном уровне с C. Data.Vector.Fusion.Stream.Monadic имеет реализацию потока, основанную на этой статье Coutts, Leshchinskiy, Stewart 2007. Идея этого заключается в использовании слияния с потоком destroy/unfoldr.

Напомним, что список unfoldr :: (b → Maybe (a, b)) → b → [a] создает список, повторно применяя (разворачивая) функцию шага вперед, начиная с начального значения. Stream - это просто функция unfoldr с начальным состоянием. (Библиотека Data.Vector.Fusion.Stream.Monadic использует GADT для создания конструкторов для Step которые могут быть удобно сопоставлены с образцами. Этого можно было бы сделать и без GADT, я думаю.)

Центральным элементом решения является mkBitstream :: BSL.ByteString → Stream Bool которая превращает BytesString в поток Bool. В основном, мы отслеживаем текущий ByteString, текущий байт и сколько текущего байта все еще не загружено. Всякий раз, когда байт используется, другой байт прерывается ByteString. Когда Nothing не осталось, поток будет Done.

Функция longestRun берется прямо из решения @Cirdec.

Здесь этюд:

{-# LANGUAGE CPP #-}
#define PHASE_FUSED [1]
#define PHASE_INNER [0]
#define INLINE_FUSED INLINE PHASE_FUSED
#define INLINE_INNER INLINE PHASE_INNER
module Main where

import           Control.Monad.Identity            (Identity)
import           Data.Bits                         (shiftR, (.&.))
import qualified Data.ByteString.Lazy              as BSL
import           Data.Functor.Identity             (runIdentity)
import qualified Data.Vector.Fusion.Stream.Monadic as S
import           Data.Word8                        (Word8)

type Stream a = S.Stream Identity a   -- no need for any monad, really

data Step = Step BSL.ByteString !Word8 !Word8   -- could use tuples, but this is faster

mkBitstream :: BSL.ByteString -> Stream Bool
mkBitstream bs' = S.Stream step (Step bs' 0 0) where
    {-# INLINE_INNER step #-}
    step (Step bs w n) | n==0 = case (BSL.uncons bs) of
                            Nothing        -> return S.Done
                            Just (w', bs') -> return $ 
                                S.Yield (w' .&. 1 == 1) (Step bs' (w' 'shiftR' 1) 7)
                       | otherwise = return $ 
                                S.Yield (w .&. 1 == 1) (Step bs (w 'shiftR' 1) (n-1))


data LongestRun = LongestRun !Bool !Int !Int

{-# INLINE extendRun #-}
extendRun :: LongestRun -> Bool -> LongestRun
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 :: Stream Bool -> Int
longestRun s = runIdentity $ do
    (LongestRun _ _ longest) <- S.foldl' extendRun (LongestRun False 0 0) s
    return longest

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