From 65c104f0f44f2b98c01f0e72d5379dc5f485280d Mon Sep 17 00:00:00 2001 From: Harendra Kumar Date: Tue, 20 Oct 2026 13:13:12 +0530 Subject: [PATCH 1/4] Hash the bytes of the elements in splitSepByOneOf The Karp-Rabin hash called fromEnum, which fails on Word64 values above maxBound :: Int. Hash the bytes of the elements instead, reading an element of size 1, 2, 4 or 8 as one unsigned word; a hash match is already confirmed by comparing bytes. splitSepByOneOf no longer needs an Enum constraint. The hash helpers are in the new Streamly.Internal.Data.SeqMatch module. In the loop phase, read the falling-out element of the longest pattern before overwriting it with the new element and the others after, in a single zipWithM pass. --- core/src/Streamly/Internal/Data/SeqMatch.hs | 101 ++++++++++++++++++ .../Streamly/Internal/Data/Stream/Parse.hs | 64 ++++++----- core/streamly-core.cabal | 3 + test/Streamly/Test/Data/Stream/Parse.hs | 12 ++- 4 files changed, 152 insertions(+), 28 deletions(-) create mode 100644 core/src/Streamly/Internal/Data/SeqMatch.hs diff --git a/core/src/Streamly/Internal/Data/SeqMatch.hs b/core/src/Streamly/Internal/Data/SeqMatch.hs new file mode 100644 index 0000000000..4a08f6bd34 --- /dev/null +++ b/core/src/Streamly/Internal/Data/SeqMatch.hs @@ -0,0 +1,101 @@ +-- | +-- Module : Streamly.Internal.Data.SeqMatch +-- Copyright : (c) 2026 Composewell Technologies +-- License : BSD-3-Clause +-- Maintainer : streamly@composewell.com +-- Stability : experimental +-- Portability : GHC +-- +-- Helpers for searching a sequence of 'Unbox' elements in a stream. +-- +-- A pattern is matched using a Karp-Rabin rolling hash of the elements, and a +-- hash match is confirmed by comparing bytes. + +module Streamly.Internal.Data.SeqMatch + ( + -- * Karp-Rabin hash + hashPower + , addHash + , arrayHash + , elemHashAt + ) +where + +import Data.Bits (shiftR, xor) +import Data.Proxy (Proxy(..)) +import Data.Word (Word8, Word16, Word32, Word64) +import Streamly.Internal.Data.Array.Type (Array(..)) +import Streamly.Internal.Data.MutByteArray.Type (MutByteArray) +import Streamly.Internal.Data.Unbox (Unbox(..)) +import Streamly.Internal.System.IO (unsafeInlineIO) + +------------------------------------------------------------------------------- +-- Karp-Rabin hash +------------------------------------------------------------------------------- + +-- The hash of a sequence of elements e0, e1, ..., e(n-1) is: +-- +-- h0 * k^(n-1) + h1 * k^(n-2) + ... + h(n-1) +-- +-- in 'Word32' arithmetic, where hi is the hash of the bytes of ei computed by +-- 'elemHashAt'. Hashing the bytes, instead of 'fromEnum' of the elements, +-- works for any 'Unbox' type, e.g. 'fromEnum' fails on Word64 values above +-- @maxBound :: Int@. +-- +-- If @h@ is the hash of a window of @n@ elements, sliding the window by one +-- element gives: +-- +-- h * k + elemHash new - k^n * elemHash old + +k :: Word32 +k = 2891336453 + +-- | @hashPower n@ is @k^n@, the multiplier to remove the oldest element from +-- the hash of a window of @n@ elements. +{-# INLINE hashPower #-} +hashPower :: Int -> Word32 +hashPower n = k ^ n + +-- | @addHash h e@ appends an element with hash @e@ to a sequence with hash +-- @h@. +{-# INLINE addHash #-} +addHash :: Word32 -> Word32 -> Word32 +addHash h e = h * k + e + +-- | The hash of the bytes of the element at the given byte offset. Elements +-- of size 1, 2, 4 or 8 are read as one unsigned word; a 64-bit word is folded +-- to 32 bits. +{-# INLINE elemHashAt #-} +elemHashAt :: forall a. Unbox a => Proxy a -> MutByteArray -> Int -> IO Word32 +elemHashAt _ mba off = + case sizeOf (Proxy :: Proxy a) of + 1 -> fromIntegral <$> (peekAt off mba :: IO Word8) + 2 -> fromIntegral <$> (peekAt off mba :: IO Word16) + 4 -> peekAt off mba + 8 -> do + w <- peekAt off mba :: IO Word64 + return $ fromIntegral (w `xor` (w `shiftR` 32)) + n -> go 0 off (off + n) + + where + + go h i end + | i >= end = return h + | otherwise = do + b <- peekAt i mba :: IO Word8 + go (addHash h (fromIntegral b)) (i + 1) end + +-- | The hash of the elements of an array. +{-# INLINE arrayHash #-} +arrayHash :: forall a. Unbox a => Array a -> Word32 +arrayHash (Array contents start end) = unsafeInlineIO $ go 0 start + + where + + size = sizeOf (Proxy :: Proxy a) + + go h i + | i >= end = return h + | otherwise = do + e <- elemHashAt (Proxy :: Proxy a) contents i + go (addHash h e) (i + size) diff --git a/core/src/Streamly/Internal/Data/Stream/Parse.hs b/core/src/Streamly/Internal/Data/Stream/Parse.hs index 899ab3ced9..2623ae3c6f 100644 --- a/core/src/Streamly/Internal/Data/Stream/Parse.hs +++ b/core/src/Streamly/Internal/Data/Stream/Parse.hs @@ -147,6 +147,7 @@ import qualified Streamly.Internal.Data.Parser as PR import qualified Streamly.Internal.Data.Parser as PRD import qualified Streamly.Internal.Data.ParserDrivers as Drivers import qualified Streamly.Internal.Data.RingArray as RB +import qualified Streamly.Internal.Data.SeqMatch as SeqMatch import Streamly.Internal.Data.Stream.Type @@ -2184,7 +2185,7 @@ data SplitOnSeqOneOfState fs s b = -- {-# INLINE_NORMAL splitSepByOneOf #-} splitSepByOneOf - :: forall m a b. (MonadIO m, Unbox a, Enum a) + :: forall m a b. (MonadIO m, Unbox a) => [Array a] -> Fold m a b -> Stream m a @@ -2197,11 +2198,7 @@ splitSepByOneOf patArrs0 (Fold fstep initial _ final) (Stream step state) = elemSize = SIZE_OF(a) patArrs = filter (\arr -> A.length arr > 0) patArrs0 - k :: Word32 - k = 2891336453 - - addCksum :: Word32 -> a -> Word32 - addCksum cksum a = cksum * k + fromIntegral (fromEnum a) + proxy = Proxy :: Proxy a -- (patLen elements, patBytes, k^patLen, [(hash, pattern)]) -- Sorted by patLen descending for longest-match priority. @@ -2217,8 +2214,8 @@ splitSepByOneOf patArrs0 (Fold fstep initial _ final) (Stream step state) = mkGroup ps@(p:_) = let pLen = A.length p pBytes = A.byteLength p - cf = k ^ pLen - hps = Prelude.map (\arr -> (A.foldl' addCksum 0 arr, arr)) ps + cf = SeqMatch.hashPower pLen + hps = Prelude.map (\arr -> (SeqMatch.arrayHash arr, arr)) ps in (pLen, pBytes, cf, hps) initialHashes :: [Word32] @@ -2240,32 +2237,46 @@ splitSepByOneOf patArrs0 (Fold fstep initial _ final) (Stream step state) = yieldReinit nextGen b = initial >>= skip . SOOYield b . nextAfterInit nextGen + -- TODO: the per-group hashes are kept in a list that is rebuilt for each + -- input element, allocating about 90 bytes per element. Keep the hashes in + -- a MutByteArray and the groups in unboxed arrays, and update them in + -- place. + -- Phase 1 (Fill) hash update. count is the new count after inserting x. -- Buffer is direct-indexed: position p at byte offset p * SIZE_OF(a). - updateHashesFill :: Int -> MutByteArray -> a -> [Word32] -> IO [Word32] - updateHashesFill count mba x = Control.Monad.zipWithM upd groups + updateHashesFill :: Int -> MutByteArray -> [Word32] -> IO [Word32] + updateHashesFill count mba hashes = do + new <- SeqMatch.elemHashAt proxy mba ((count - 1) * SIZE_OF(a)) + Control.Monad.zipWithM (upd new) groups hashes where - upd (pLen, _, coeff, _) hash - | count <= pLen = pure (addCksum hash x) + upd new (pLen, _, coeff, _) hash + | count <= pLen = pure $! SeqMatch.addHash hash new | otherwise = do let oldOffset = (count - pLen - 1) * SIZE_OF(a) - old :: a <- peekAt oldOffset mba - pure (addCksum hash x - - coeff * fromIntegral (fromEnum old)) + old <- SeqMatch.elemHashAt proxy mba oldOffset + pure $! SeqMatch.addHash hash new - coeff * old -- Phase 2 (Loop) hash update. ringHead is the byte offset of the oldest - -- element BEFORE insertion. Reads the falling-out element for each L - -- (which for L=maxLen is the element about to be evicted at ringHead). + -- element BEFORE insertion. The falling-out element for L=maxLen is the + -- one at ringHead, it is read before x overwrites it. For L < maxLen the + -- falling-out element is elsewhere in the ring and is read after. updateHashesLoop :: Int -> MutByteArray -> a -> [Word32] -> IO [Word32] - updateHashesLoop ringHead mba x = Control.Monad.zipWithM upd groups + updateHashesLoop ringHead mba x hashes = do + oldMax <- SeqMatch.elemHashAt proxy mba ringHead + pokeAt ringHead mba x + new <- SeqMatch.elemHashAt proxy mba ringHead + Control.Monad.zipWithM (upd new oldMax) groups hashes where - upd (pLen, _, coeff, _) hash = do - let oldOffset = - (ringHead + (maxLen - pLen) * SIZE_OF(a)) - `mod` maxBytes - old :: a <- peekAt oldOffset mba - pure (addCksum hash x - - coeff * fromIntegral (fromEnum old)) + upd new oldMax (pLen, _, coeff, _) hash = do + old <- + if pLen == maxLen + then pure oldMax + else + let oldOffset = + (ringHead + (maxLen - pLen) * SIZE_OF(a)) + `mod` maxBytes + in SeqMatch.elemHashAt proxy mba oldOffset + pure $! SeqMatch.addHash hash new - coeff * old -- Phase 1 match check. count is the new count after insertion. -- Returns the matched pattern length (the longest one), or Nothing. @@ -2384,7 +2395,7 @@ splitSepByOneOf patArrs0 (Fold fstep initial _ final) (Stream step state) = Yield x s -> do liftIO $ pokeAt (count * SIZE_OF(a)) mba x let count' = count + 1 - hashes' <- liftIO $ updateHashesFill count' mba x hashes + hashes' <- liftIO $ updateHashesFill count' mba hashes matchRes <- liftIO $ checkMatchesFill count' mba hashes' case matchRes of Just pLen -> @@ -2411,7 +2422,6 @@ splitSepByOneOf patArrs0 (Fold fstep initial _ final) (Stream step state) = Yield x s -> do evicted :: a <- liftIO $ peekAt ringHead mba hashes' <- liftIO $ updateHashesLoop ringHead mba x hashes - liftIO $ pokeAt ringHead mba x let ringHead' = (ringHead + elemSize) `mod` maxBytes r <- fstep fs evicted case r of diff --git a/core/streamly-core.cabal b/core/streamly-core.cabal index adfd1ac1aa..b48eec6cbf 100644 --- a/core/streamly-core.cabal +++ b/core/streamly-core.cabal @@ -443,6 +443,9 @@ library , Streamly.Internal.Data.RingArray , Streamly.Internal.Data.RingArray.Generic + -- Sequence search helpers + , Streamly.Internal.Data.SeqMatch + -- streamly-console , Streamly.Internal.Console.Stdio diff --git a/test/Streamly/Test/Data/Stream/Parse.hs b/test/Streamly/Test/Data/Stream/Parse.hs index 24c5bd7a6f..c2d59fcb71 100644 --- a/test/Streamly/Test/Data/Stream/Parse.hs +++ b/test/Streamly/Test/Data/Stream/Parse.hs @@ -19,7 +19,7 @@ import Control.Monad.IO.Class (MonadIO) import Data.Functor.Identity (Identity) import Data.List (group, intercalate) import Data.Semigroup (Sum(..), getSum) -import Data.Word (Word8) +import Data.Word (Word8, Word64) import Streamly.Internal.Data.Array (Array) import Streamly.Internal.Data.Fold (Fold) import Streamly.Internal.Data.MutByteArray (Unbox) @@ -807,6 +807,15 @@ testSplitSepByOneOfEmpty = do (Stream.fromList "") xs `shouldBe` [] +-- The hash must not use fromEnum, it fails on large Word64 values. +testSplitSepByOneOfWord64 :: Expectation +testSplitSepByOneOfWord64 = do + let pats = map Array.fromList [[maxBound, 2], [3 :: Word64]] + xs <- Stream.toList $ + Stream.splitSepByOneOf pats Fold.toList + (Stream.fromList [5, maxBound, 2, 7, 3, maxBound]) + xs `shouldBe` [[5], [7], [maxBound]] + ------------------------------------------------------------------------------- -- splitInnerBy / splitInnerBySuffix ------------------------------------------------------------------------------- @@ -1059,6 +1068,7 @@ main = hspec it "splits on any of the patterns" testSplitSepByOneOfSingle it "splits on multi-element pattern" testSplitSepByOneOfSeq it "empty stream" testSplitSepByOneOfEmpty + it "Word64 elements" testSplitSepByOneOfWord64 describe "splitInnerBy" $ do it "splits inner containers" testSplitInnerBy From db0666a100047631500dafa47c3a9c1787307e78 Mon Sep 17 00:00:00 2001 From: Harendra Kumar Date: Tue, 20 Oct 2026 13:46:26 +0530 Subject: [PATCH 2/4] Use SeqMatch helpers for word packing in sequence search Move the word packing and decoding and the Karp-Rabin hash arithmetic of Fold.takeEndBySeq, Fold.takeEndBySeq_ and, in Stream, takeEndBySeq, takeEndBySeq_, splitSepBySeq_, splitOnSuffixSeq, dropMatches, replaceMatches and dropSuffix to Streamly.Internal.Data.SeqMatch. The code is unchanged, elements are still packed and hashed using fromEnum. --- .../Internal/Data/Fold/Combinators.hs | 30 ++--- core/src/Streamly/Internal/Data/SeqMatch.hs | 37 +++++- .../Streamly/Internal/Data/Stream/Parse.hs | 118 ++++++++---------- 3 files changed, 95 insertions(+), 90 deletions(-) diff --git a/core/src/Streamly/Internal/Data/Fold/Combinators.hs b/core/src/Streamly/Internal/Data/Fold/Combinators.hs index a38c5fdd51..8ffc1dfb22 100644 --- a/core/src/Streamly/Internal/Data/Fold/Combinators.hs +++ b/core/src/Streamly/Internal/Data/Fold/Combinators.hs @@ -224,11 +224,10 @@ where import Control.Monad (void) import Control.Monad.IO.Class (MonadIO(..)) import Data.Bifunctor (first) -import Data.Bits (shiftL, shiftR, (.|.), (.&.)) +import Data.Bits (shiftL, shiftR, (.&.)) import Data.Either (isLeft, isRight, fromLeft, fromRight) import Data.Int (Int64) import Data.Proxy (Proxy(..)) -import Data.Word (Word32) import Streamly.Internal.Data.Array.Type (Array(..)) import Streamly.Internal.Data.Scanl.Type (Scanl(..)) import Streamly.Internal.Data.Unbox (Unbox(..)) @@ -248,6 +247,7 @@ import qualified Streamly.Internal.Data.Pipe.Type as Pipe import qualified Streamly.Internal.Data.RingArray as RingArray import qualified Streamly.Internal.Data.Scanl.Combinators as Scanl import qualified Streamly.Internal.Data.Scanl.Type as Scanl +import qualified Streamly.Internal.Data.SeqMatch as SeqMatch import qualified Streamly.Internal.Data.Stream.Type as StreamD import Prelude hiding @@ -1401,13 +1401,12 @@ takeEndBySeq patArr (Fold fstep finitial fextract ffinal) = wordPat :: Word wordPat = wordMask .&. Array.foldl' addToWord 0 patArr - addToWord wd a = (wd `shiftL` elemBits) .|. fromIntegral (fromEnum a) + addToWord = SeqMatch.addToWord elemBits -- For Rabin-Karp search - k = 2891336453 :: Word32 - coeff = k ^ patLen + coeff = SeqMatch.hashPower patLen - addCksum cksum a = cksum * k + fromIntegral (fromEnum a) + addCksum cksum a = SeqMatch.addHash cksum (fromIntegral (fromEnum a)) deltaCksum cksum old new = addCksum cksum new - coeff * fromIntegral (fromEnum old) @@ -1565,19 +1564,17 @@ takeEndBySeq_ patArr (Fold fstep finitial fextract ffinal) = wordMask :: Word wordMask = (1 `shiftL` (elemBits * patLen)) - 1 - elemMask :: Word - elemMask = (1 `shiftL` elemBits) - 1 + wordToElem = SeqMatch.wordToElem elemBits wordPat :: Word wordPat = wordMask .&. Array.foldl' addToWord 0 patArr - addToWord wd a = (wd `shiftL` elemBits) .|. fromIntegral (fromEnum a) + addToWord = SeqMatch.addToWord elemBits -- For Rabin-Karp search - k = 2891336453 :: Word32 - coeff = k ^ patLen + coeff = SeqMatch.hashPower patLen - addCksum cksum a = cksum * k + fromIntegral (fromEnum a) + addCksum cksum a = SeqMatch.addHash cksum (fromIntegral (fromEnum a)) deltaCksum cksum old new = addCksum cksum new - coeff * fromIntegral (fromEnum old) @@ -1612,9 +1609,8 @@ takeEndBySeq_ patArr (Fold fstep finitial fextract ffinal) = else return $ Partial $ SplitOnSeqWord s (idx + 1) wrd1 step (SplitOnSeqWordLoop s wrd) x = do let wrd1 = addToWord wrd x - old = (wordMask .&. wrd) - `shiftR` (elemBits * (patLen - 1)) - res <- fstep s (toEnum $ fromIntegral old) + old = wordToElem (wrd `shiftR` (elemBits * (patLen - 1))) + res <- fstep s old case res of Partial s1 | wrd1 .&. wordMask == wordPat -> @@ -1666,8 +1662,8 @@ takeEndBySeq_ patArr (Fold fstep finitial fextract ffinal) = if n == 0 then fex s else do - let old = elemMask .&. (wrd `shiftR` (elemBits * (n - 1))) - r <- fstep s (toEnum $ fromIntegral old) + let old = wordToElem (wrd `shiftR` (elemBits * (n - 1))) + r <- fstep s old case r of Partial s1 -> consumeWord s1 (n - 1) wrd Done b -> return b diff --git a/core/src/Streamly/Internal/Data/SeqMatch.hs b/core/src/Streamly/Internal/Data/SeqMatch.hs index 4a08f6bd34..5c58bf8cb5 100644 --- a/core/src/Streamly/Internal/Data/SeqMatch.hs +++ b/core/src/Streamly/Internal/Data/SeqMatch.hs @@ -8,20 +8,27 @@ -- -- Helpers for searching a sequence of 'Unbox' elements in a stream. -- --- A pattern is matched using a Karp-Rabin rolling hash of the elements, and a --- hash match is confirmed by comparing bytes. +-- A short pattern (at most one machine word in total size) is matched by +-- packing the elements into a 'Word', each element taking @8 * sizeOf a@ +-- bits. A longer pattern is matched using a Karp-Rabin rolling hash of the +-- elements, and a hash match is confirmed by comparing bytes. module Streamly.Internal.Data.SeqMatch ( + -- * Word packing + elemMask + , addToWord + , wordToElem + -- * Karp-Rabin hash - hashPower + , hashPower , addHash , arrayHash , elemHashAt ) where -import Data.Bits (shiftR, xor) +import Data.Bits (shiftL, shiftR, xor, (.|.), (.&.)) import Data.Proxy (Proxy(..)) import Data.Word (Word8, Word16, Word32, Word64) import Streamly.Internal.Data.Array.Type (Array(..)) @@ -29,6 +36,28 @@ import Streamly.Internal.Data.MutByteArray.Type (MutByteArray) import Streamly.Internal.Data.Unbox (Unbox(..)) import Streamly.Internal.System.IO (unsafeInlineIO) +------------------------------------------------------------------------------- +-- Word packing +------------------------------------------------------------------------------- + +-- | A mask of the low @bits@ bits of a 'Word', @bits@ must be less than the +-- word size. +{-# INLINE elemMask #-} +elemMask :: Int -> Word +elemMask bits = (1 `shiftL` bits) - 1 + +-- | @addToWord bits word a@ shifts @word@ left by @bits@ and puts +-- @fromEnum a@ in the vacated bits. +{-# INLINE addToWord #-} +addToWord :: Enum a => Int -> Word -> a -> Word +addToWord bits wd a = (wd `shiftL` bits) .|. fromIntegral (fromEnum a) + +-- | @wordToElem bits word@ decodes the element packed by 'addToWord' in the +-- low @bits@ bits of @word@, @bits@ must be less than the word size. +{-# INLINE wordToElem #-} +wordToElem :: Enum a => Int -> Word -> a +wordToElem bits wd = toEnum $ fromIntegral (wd .&. elemMask bits) + ------------------------------------------------------------------------------- -- Karp-Rabin hash ------------------------------------------------------------------------------- diff --git a/core/src/Streamly/Internal/Data/Stream/Parse.hs b/core/src/Streamly/Internal/Data/Stream/Parse.hs index 2623ae3c6f..a07af7e9ee 100644 --- a/core/src/Streamly/Internal/Data/Stream/Parse.hs +++ b/core/src/Streamly/Internal/Data/Stream/Parse.hs @@ -118,7 +118,7 @@ where import Control.Exception (assert) import Control.Monad (zipWithM) import Control.Monad.IO.Class (MonadIO(..)) -import Data.Bits (shiftR, shiftL, (.|.), (.&.)) +import Data.Bits (shiftR, shiftL, (.&.)) import Data.Functor.Identity (Identity(..), runIdentity) import Data.List (groupBy, sortBy) import Data.Ord (comparing, Down(..)) @@ -832,19 +832,17 @@ takeEndBySeqWith withSep patArr (Stream step state) = wordMask :: Word wordMask = (1 `shiftL` (elemBits * patLen)) - 1 - elemMask :: Word - elemMask = (1 `shiftL` elemBits) - 1 + wordToElem = SeqMatch.wordToElem elemBits wordPat :: Word wordPat = wordMask .&. A.foldl' addToWord 0 patArr - addToWord wd a = (wd `shiftL` elemBits) .|. fromIntegral (fromEnum a) + addToWord = SeqMatch.addToWord elemBits -- For Rabin-Karp search - k = 2891336453 :: Word32 - coeff = k ^ patLen + coeff = SeqMatch.hashPower patLen - addCksum cksum a = cksum * k + fromIntegral (fromEnum a) + addCksum cksum a = SeqMatch.addHash cksum (fromIntegral (fromEnum a)) deltaCksum cksum old new = addCksum cksum new - coeff * fromIntegral (fromEnum old) @@ -921,9 +919,9 @@ takeEndBySeqWith withSep patArr (Stream step state) = stepOuter _ (TakeEndBySeqWordDone 0 _) = do return Stop stepOuter _ (TakeEndBySeqWordDone n wrd) = do - let old = elemMask .&. (wrd `shiftR` (elemBits * (n - 1))) + let old = wordToElem (wrd `shiftR` (elemBits * (n - 1))) in yield - (toEnum $ fromIntegral old) + old (TakeEndBySeqWordDone (n - 1) wrd) -- XXX If we remove this init state for perf experiment the time taken @@ -956,12 +954,11 @@ takeEndBySeqWith withSep patArr (Stream step state) = -- in simplification because the state argument of Yield is -- lazy, maybe we can make that strict. let wrd1 = addToWord wrd x - old = (wordMask .&. wrd) - `shiftR` (elemBits * (patLen - 1)) + old = wordToElem (wrd `shiftR` (elemBits * (patLen - 1))) !y = if withSep then x - else toEnum $ fromIntegral old + else old -- Note: changing the nesting order of if and yield makes a -- difference in performance. if wrd1 .&. wordMask /= wordPat @@ -1202,19 +1199,17 @@ splitSepBySeq_ patArr (Fold fstep initial _ final) (Stream step state) = wordMask :: Word wordMask = (1 `shiftL` (elemBits * patLen)) - 1 - elemMask :: Word - elemMask = (1 `shiftL` elemBits) - 1 + wordToElem = SeqMatch.wordToElem elemBits wordPat :: Word wordPat = wordMask .&. A.foldl' addToWord 0 patArr - addToWord wd a = (wd `shiftL` elemBits) .|. fromIntegral (fromEnum a) + addToWord = SeqMatch.addToWord elemBits -- For Rabin-Karp search - k = 2891336453 :: Word32 - coeff = k ^ patLen + coeff = SeqMatch.hashPower patLen - addCksum cksum a = cksum * k + fromIntegral (fromEnum a) + addCksum cksum a = SeqMatch.addHash cksum (fromIntegral (fromEnum a)) deltaCksum cksum old new = addCksum cksum new - coeff * fromIntegral (fromEnum old) @@ -1343,8 +1338,8 @@ splitSepBySeq_ patArr (Fold fstep initial _ final) (Stream step state) = r <- final fs skip $ SplitOnSeqYield r SplitOnSeqDone stepOuter _ (SplitOnSeqWordDone n fs wrd) = do - let old = elemMask .&. (wrd `shiftR` (elemBits * (n - 1))) - r <- fstep fs (toEnum $ fromIntegral old) + let old = wordToElem (wrd `shiftR` (elemBits * (n - 1))) + r <- fstep fs old case r of FL.Partial fs1 -> skip $ SplitOnSeqWordDone (n - 1) fs1 wrd FL.Done b -> do @@ -1401,9 +1396,8 @@ splitSepBySeq_ patArr (Fold fstep initial _ final) (Stream step state) = Yield x s -> do let jump c = SplitOnSeqWordInit 0 0 c s wrd1 = addToWord wrd x - old = (wordMask .&. wrd) - `shiftR` (elemBits * (patLen - 1)) - r <- fstep fs (toEnum $ fromIntegral old) + old = wordToElem (wrd `shiftR` (elemBits * (patLen - 1))) + r <- fstep fs old case r of FL.Partial fs1 -> do if wrd1 .&. wordMask == wordPat @@ -1602,13 +1596,12 @@ splitOnSuffixSeq withSep patArr (Fold fstep initial _ final) (Stream step state) wordMask :: Word wordMask = (1 `shiftL` (elemBits * patLen)) - 1 - elemMask :: Word - elemMask = (1 `shiftL` elemBits) - 1 + wordToElem = SeqMatch.wordToElem elemBits wordPat :: Word wordPat = wordMask .&. A.foldl' addToWord 0 patArr - addToWord wd a = (wd `shiftL` elemBits) .|. fromIntegral (fromEnum a) + addToWord = SeqMatch.addToWord elemBits nextAfterInit nextGen stepRes = case stepRes of @@ -1639,10 +1632,9 @@ splitOnSuffixSeq withSep patArr (Fold fstep initial _ final) (Stream step state) FL.Done b -> yieldReinit jump b -- For Rabin-Karp search - k = 2891336453 :: Word32 - coeff = k ^ patLen + coeff = SeqMatch.hashPower patLen - addCksum cksum a = cksum * k + fromIntegral (fromEnum a) + addCksum cksum a = SeqMatch.addHash cksum (fromIntegral (fromEnum a)) deltaCksum cksum old new = addCksum cksum new - coeff * fromIntegral (fromEnum old) @@ -1731,8 +1723,8 @@ splitOnSuffixSeq withSep patArr (Fold fstep initial _ final) (Stream step state) r <- final fs skip $ SplitOnSuffixSeqYield r SplitOnSuffixSeqDone stepOuter _ (SplitOnSuffixSeqWordDone n fs wrd) = do - let old = elemMask .&. (wrd `shiftR` (elemBits * (n - 1))) - r <- fstep fs (toEnum $ fromIntegral old) + let old = wordToElem (wrd `shiftR` (elemBits * (n - 1))) + r <- fstep fs old case r of FL.Partial fs1 -> skip $ SplitOnSuffixSeqWordDone (n - 1) fs1 wrd FL.Done b -> do @@ -1792,12 +1784,11 @@ splitOnSuffixSeq withSep patArr (Fold fstep initial _ final) (Stream step state) Yield x s -> do let jump c = SplitOnSuffixSeqWordInit c s wrd1 = addToWord wrd x - old = (wordMask .&. wrd) - `shiftR` (elemBits * (patLen - 1)) + old = wordToElem (wrd `shiftR` (elemBits * (patLen - 1))) r <- if withSep then fstep fs x - else fstep fs (toEnum $ fromIntegral old) + else fstep fs old case r of FL.Partial fs1 -> if wrd1 .&. wordMask == wordPat @@ -2762,18 +2753,16 @@ dropMatches patArr (Stream step state) = wordMask :: Word wordMask = (1 `shiftL` (elemBits * patLen)) - 1 - elemMask :: Word - elemMask = (1 `shiftL` elemBits) - 1 + wordToElem = SeqMatch.wordToElem elemBits wordPat :: Word wordPat = wordMask .&. A.foldl' addToWord 0 patArr - addToWord wd a = (wd `shiftL` elemBits) .|. fromIntegral (fromEnum a) + addToWord = SeqMatch.addToWord elemBits - k = 2891336453 :: Word32 - coeff = k ^ patLen + coeff = SeqMatch.hashPower patLen - addCksum cksum a = cksum * k + fromIntegral (fromEnum a) + addCksum cksum a = SeqMatch.addHash cksum (fromIntegral (fromEnum a)) deltaCksum cksum old new = addCksum cksum new - coeff * fromIntegral (fromEnum old) @@ -2830,10 +2819,10 @@ dropMatches patArr (Stream step state) = stepOuter _ (DropInfixWordDone 0 _) = return Stop stepOuter _ (DropInfixWordDone n wrd) = do - let old = elemMask .&. (wrd `shiftR` (elemBits * (n - 1))) + let old = wordToElem (wrd `shiftR` (elemBits * (n - 1))) return $ Yield - (toEnum $ fromIntegral old) + old (DropInfixWordDone (n - 1) wrd) stepOuter gst (DropInfixWordInit idx0 wrd0 st0) = @@ -2870,12 +2859,10 @@ dropMatches patArr (Stream step state) = case res of Yield x s -> do let wrd1 = addToWord wrd x - old = (wordMask .&. wrd) - `shiftR` (elemBits * (patLen - 1)) - oldA = toEnum (fromIntegral old) + old = wordToElem (wrd `shiftR` (elemBits * (patLen - 1))) if wrd1 .&. wordMask == wordPat - then return $ Yield oldA (DropInfixWordInit 0 0 s) - else return $ Yield oldA (DropInfixWordLoop wrd1 s) + then return $ Yield old (DropInfixWordInit 0 0 s) + else return $ Yield old (DropInfixWordLoop wrd1 s) Skip s -> go SPEC wrd s Stop -> skip $ DropInfixWordDone patLen wrd @@ -3020,18 +3007,16 @@ replaceMatches patArr replArr (Stream step state) = wordMask :: Word wordMask = (1 `shiftL` (elemBits * patLen)) - 1 - elemMask :: Word - elemMask = (1 `shiftL` elemBits) - 1 + wordToElem = SeqMatch.wordToElem elemBits wordPat :: Word wordPat = wordMask .&. A.foldl' addToWord 0 patArr - addToWord wd a = (wd `shiftL` elemBits) .|. fromIntegral (fromEnum a) + addToWord = SeqMatch.addToWord elemBits - k = 2891336453 :: Word32 - coeff = k ^ patLen + coeff = SeqMatch.hashPower patLen - addCksum cksum a = cksum * k + fromIntegral (fromEnum a) + addCksum cksum a = SeqMatch.addHash cksum (fromIntegral (fromEnum a)) deltaCksum cksum old new = addCksum cksum new - coeff * fromIntegral (fromEnum old) @@ -3094,10 +3079,10 @@ replaceMatches patArr replArr (Stream step state) = stepOuter _ (ReplaceInfixWordDone 0 _) = return Stop stepOuter _ (ReplaceInfixWordDone n wrd) = do - let old = elemMask .&. (wrd `shiftR` (elemBits * (n - 1))) + let old = wordToElem (wrd `shiftR` (elemBits * (n - 1))) return $ Yield - (toEnum $ fromIntegral old) + old (ReplaceInfixWordDone (n - 1) wrd) stepOuter gst (ReplaceInfixWordInit idx0 wrd0 st0) = @@ -3134,12 +3119,10 @@ replaceMatches patArr replArr (Stream step state) = case res of Yield x s -> do let wrd1 = addToWord wrd x - old = (wordMask .&. wrd) - `shiftR` (elemBits * (patLen - 1)) - oldA = toEnum (fromIntegral old) + old = wordToElem (wrd `shiftR` (elemBits * (patLen - 1))) if wrd1 .&. wordMask == wordPat - then return $ Yield oldA (ReplaceInfixEmitWord 0 s) - else return $ Yield oldA (ReplaceInfixWordLoop wrd1 s) + then return $ Yield old (ReplaceInfixEmitWord 0 s) + else return $ Yield old (ReplaceInfixWordLoop wrd1 s) Skip s -> go SPEC wrd s Stop -> skip $ ReplaceInfixWordDone patLen wrd @@ -3287,13 +3270,12 @@ dropSuffix patArr (Stream step state) = wordMask :: Word wordMask = (1 `shiftL` (elemBits * patLen)) - 1 - elemMask :: Word - elemMask = (1 `shiftL` elemBits) - 1 + wordToElem = SeqMatch.wordToElem elemBits wordPat :: Word wordPat = wordMask .&. A.foldl' addToWord 0 patArr - addToWord wd a = (wd `shiftL` elemBits) .|. fromIntegral (fromEnum a) + addToWord = SeqMatch.addToWord elemBits skip = return . Skip @@ -3354,10 +3336,10 @@ dropSuffix patArr (Stream step state) = stepOuter _ (DropSuffixWordDone 0 _) = return Stop stepOuter _ (DropSuffixWordDone n wrd) = do - let old = elemMask .&. (wrd `shiftR` (elemBits * (n - 1))) + let old = wordToElem (wrd `shiftR` (elemBits * (n - 1))) return $ Yield - (toEnum $ fromIntegral old) + old (DropSuffixWordDone (n - 1) wrd) stepOuter gst (DropSuffixWordInit idx0 wrd0 st0) = @@ -3391,10 +3373,8 @@ dropSuffix patArr (Stream step state) = case res of Yield x s -> do let wrd1 = addToWord wrd x - old = (wordMask .&. wrd) - `shiftR` (elemBits * (patLen - 1)) - oldA = toEnum (fromIntegral old) - return $ Yield oldA (DropSuffixWordLoop wrd1 s) + old = wordToElem (wrd `shiftR` (elemBits * (patLen - 1))) + return $ Yield old (DropSuffixWordLoop wrd1 s) Skip s -> go SPEC wrd s Stop -> if wrd .&. wordMask == wordPat From f08bf02b109562a71b28e10b50628bd1a88a46e5 Mon Sep 17 00:00:00 2001 From: Harendra Kumar Date: Tue, 20 Oct 2026 14:25:08 +0530 Subject: [PATCH 3/4] Add sequence search tests for signed, unit and Word64 elements The tests are marked pending: packing a negative element into a word without masking it gives a false match, decoding a buffered element without sign extension fails for Int8, and the Karp-Rabin hash calls fromEnum, which fails on Word64 values above maxBound :: Int. --- test/Streamly/Test/Data/Fold/Combinators.hs | 65 +++++++++++++++- test/Streamly/Test/Data/Stream/Parse.hs | 85 +++++++++++++++++++++ 2 files changed, 149 insertions(+), 1 deletion(-) diff --git a/test/Streamly/Test/Data/Fold/Combinators.hs b/test/Streamly/Test/Data/Fold/Combinators.hs index 8a663e7165..945e6aea41 100644 --- a/test/Streamly/Test/Data/Fold/Combinators.hs +++ b/test/Streamly/Test/Data/Fold/Combinators.hs @@ -10,8 +10,9 @@ module Streamly.Test.Data.Fold.Combinators (main) where import Data.IORef (newIORef, atomicModifyIORef) -import Data.Int (Int64) +import Data.Int (Int8, Int64) import Data.Semigroup (Sum(..)) +import Data.Word (Word8, Word64) import qualified Prelude import qualified Streamly.Internal.Data.Array as Array @@ -414,6 +415,63 @@ sampleFromthenOffset = (Stream.fromList [0..9 :: Int]) `shouldReturn` [1,4,7] +-- The patterns of size at most one word are packed into a word, the elements +-- with the top bit set must be masked to the element size. +takeEndBySeqSigned :: Expectation +takeEndBySeqSigned = + Stream.fold + (Fold.takeEndBySeq (Array.fromList [1,-1 :: Int8]) Fold.toList) + (Stream.fromList [-1,-1,5]) + `shouldReturn` [-1,-1,5] + +-- The elements of a word pattern must be decoded with sign extension. +takeEndBySeq_Signed :: Expectation +takeEndBySeq_Signed = do + Stream.fold + (Fold.takeEndBySeq_ (Array.fromList [1,2 :: Int8]) Fold.toList) + (Stream.fromList [-1,3,4]) + `shouldReturn` [-1,3,4] + Stream.fold + (Fold.takeEndBySeq_ (Array.fromList [1,2 :: Int8]) Fold.toList) + (Stream.fromList [-1,-128,127,1,2,5]) + `shouldReturn` [-1,-128,127] + Stream.fold + (Fold.takeEndBySeq_ (Array.fromList [1,2 :: Word8]) Fold.toList) + (Stream.fromList [200,255,1,2,5]) + `shouldReturn` [200,255] + +-- Decoding the elements of a word pattern must not evaluate toEnum 1 for a +-- type with a single value. +takeEndBySeq_Unit :: Expectation +takeEndBySeq_Unit = do + Stream.fold + (Fold.takeEndBySeq_ (Array.fromList [(), ()]) Fold.toList) + (Stream.fromList [()]) + `shouldReturn` [()] + Stream.fold + (Fold.takeEndBySeq_ (Array.fromList [(), ()]) Fold.toList) + (Stream.fromList [(), (), ()]) + `shouldReturn` [] + +-- Karp-Rabin search must not use fromEnum, it fails on large Word64 values. +takeEndBySeqWord64 :: Expectation +takeEndBySeqWord64 = + Stream.fold + (Fold.takeEndBySeq (Array.fromList [1,2 :: Word64]) Fold.toList) + (Stream.fromList [maxBound,7,1,2,9]) + `shouldReturn` [maxBound,7,1,2] + +takeEndBySeq_Word64 :: Expectation +takeEndBySeq_Word64 = do + Stream.fold + (Fold.takeEndBySeq_ (Array.fromList [1,2 :: Word64]) Fold.toList) + (Stream.fromList [maxBound,7,1,2,9]) + `shouldReturn` [maxBound,7] + Stream.fold + (Fold.takeEndBySeq_ (Array.fromList [maxBound,2,3 :: Word64]) Fold.toList) + (Stream.fromList [5,maxBound,2,maxBound,2,3,9]) + `shouldReturn` [5,maxBound,2] + distributeScan :: Expectation distributeScan = do ref <- newIORef [Fold.take 2 Fold.sum, Fold.take 2 Fold.length :: Fold.Fold IO Int Int] @@ -532,6 +590,11 @@ main = hspec $ do it "slide2" slide2 it "takeEndBySeq" takeEndBySeq it "takeEndBySeq_" takeEndBySeq_ + xit "takeEndBySeq signed elements" takeEndBySeqSigned + xit "takeEndBySeq_ signed elements" takeEndBySeq_Signed + xit "takeEndBySeq_ unit" takeEndBySeq_Unit + xit "takeEndBySeq Word64" takeEndBySeqWord64 + xit "takeEndBySeq_ Word64" takeEndBySeq_Word64 it "distributeScan" distributeScan it "distributeScan last input" distributeScanLast it "sampleFromthen offset" sampleFromthenOffset diff --git a/test/Streamly/Test/Data/Stream/Parse.hs b/test/Streamly/Test/Data/Stream/Parse.hs index c2d59fcb71..f982ab21c1 100644 --- a/test/Streamly/Test/Data/Stream/Parse.hs +++ b/test/Streamly/Test/Data/Stream/Parse.hs @@ -19,6 +19,7 @@ import Control.Monad.IO.Class (MonadIO) import Data.Functor.Identity (Identity) import Data.List (group, intercalate) import Data.Semigroup (Sum(..), getSum) +import Data.Int (Int8) import Data.Word (Word8, Word64) import Streamly.Internal.Data.Array (Array) import Streamly.Internal.Data.Fold (Fold) @@ -988,6 +989,85 @@ testReplaceMatchesOverlap = (Stream.fromList "aaaa")) `shouldReturn` "bb" +------------------------------------------------------------------------------- +-- Sequence search on signed and wide elements +------------------------------------------------------------------------------- + +-- The patterns of size at most one word are packed into a word, the elements +-- with the top bit set must be masked and sign-extended correctly. +testSeqSearchSigned :: Expectation +testSeqSearchSigned = do + let pat = Array.fromList [1, 2 :: Int8] + input = [-1, -128, 1, 2, 127, -1] + Stream.toList + (Stream.takeEndBySeq (Array.fromList [1, -1 :: Int8]) + (Stream.fromList [-1, -1, 5])) + `shouldReturn` [-1, -1, 5] + Stream.toList (Stream.takeEndBySeq_ pat (Stream.fromList [-1, 3, 4])) + `shouldReturn` [-1, 3, 4] + Stream.toList + (Stream.splitSepBySeq_ (Array.fromList [1, -1 :: Int8]) Fold.toList + (Stream.fromList [-1, -1, 5])) + `shouldReturn` [[-1, -1, 5]] + Stream.toList + (Stream.splitSepBySeq_ pat Fold.toList (Stream.fromList input)) + `shouldReturn` [[-1, -128], [127, -1]] + Stream.toList + (Stream.splitOnSuffixSeq False pat Fold.toList (Stream.fromList input)) + `shouldReturn` [[-1, -128], [127, -1]] + Stream.toList + (Stream.splitOnSuffixSeq True pat Fold.toList (Stream.fromList input)) + `shouldReturn` [[-1, -128, 1, 2], [127, -1]] + Stream.toList (Stream.dropMatches pat (Stream.fromList input)) + `shouldReturn` [-1, -128, 127, -1] + Stream.toList + (Stream.replaceMatches pat (Array.fromList [-5]) + (Stream.fromList input)) + `shouldReturn` [-1, -128, -5, 127, -1] + Stream.toList (Stream.dropSuffix pat (Stream.fromList [-1, -128, 1, 2])) + `shouldReturn` [-1, -128] + Stream.toList (Stream.dropSuffix pat (Stream.fromList [-1, -128])) + `shouldReturn` [-1, -128] + +-- Decoding the elements of a word pattern must not evaluate toEnum 1 for a +-- type with a single value. +testSeqSearchUnit :: Expectation +testSeqSearchUnit = do + let pat = Array.fromList [(), ()] + Stream.toList (Stream.takeEndBySeq_ pat (Stream.fromList [()])) + `shouldReturn` [()] + Stream.toList (Stream.takeEndBySeq_ pat (Stream.fromList [(), (), ()])) + `shouldReturn` [] + Stream.toList + (Stream.splitSepBySeq_ pat Fold.toList (Stream.fromList [()])) + `shouldReturn` [[()]] + +-- Karp-Rabin search must not use fromEnum, it fails on large Word64 values. +testSeqSearchWord64 :: Expectation +testSeqSearchWord64 = do + let pat = Array.fromList [maxBound, 2 :: Word64] + input = [5, maxBound, 2, maxBound, 9] + Stream.toList + (Stream.takeEndBySeq (Array.fromList [1, 2 :: Word64]) + (Stream.fromList [maxBound, 7, 1, 2, 9])) + `shouldReturn` [maxBound, 7, 1, 2] + Stream.toList + (Stream.takeEndBySeq_ (Array.fromList [1, 2 :: Word64]) + (Stream.fromList [maxBound, 7, 1, 2, 9])) + `shouldReturn` [maxBound, 7] + Stream.toList + (Stream.splitSepBySeq_ pat Fold.toList (Stream.fromList input)) + `shouldReturn` [[5], [maxBound, 9]] + Stream.toList + (Stream.splitOnSuffixSeq False pat Fold.toList (Stream.fromList input)) + `shouldReturn` [[5], [maxBound, 9]] + Stream.toList (Stream.dropMatches pat (Stream.fromList input)) + `shouldReturn` [5, maxBound, 9] + Stream.toList + (Stream.replaceMatches pat (Array.fromList [0]) + (Stream.fromList input)) + `shouldReturn` [5, 0, maxBound, 9] + ------------------------------------------------------------------------------- moduleName :: String @@ -1103,3 +1183,8 @@ main = hspec it "replaces with longer replacement" testReplaceMatchesLonger it "replaces with empty (drops)" testReplaceMatchesEmptyRepl it "overlapping pattern" testReplaceMatchesOverlap + + describe "sequence search" $ do + xit "signed elements" testSeqSearchSigned + xit "unit elements" testSeqSearchUnit + xit "Word64 elements" testSeqSearchWord64 From 389e18834d2469d800514b11792d068ae6faa086 Mon Sep 17 00:00:00 2001 From: Harendra Kumar Date: Tue, 20 Oct 2026 15:54:30 +0530 Subject: [PATCH 4/4] Raise the GHC heap limit for streamly-core haddock in packdiff CI The haddock build of streamly-core on GHC 9.8.4 needs more than the 500M set by limit-build-mem (it peaked at 522M in use). Override it for packdiff only, via cabal.project.local which survives packdiff's git checkouts. --- .github/workflows/packdiff.yml | 11 +++++++++++ 1 file changed, 11 insertions(+) diff --git a/.github/workflows/packdiff.yml b/.github/workflows/packdiff.yml index 1f6aab5142..4f5861bf6a 100644 --- a/.github/workflows/packdiff.yml +++ b/.github/workflows/packdiff.yml @@ -79,6 +79,17 @@ jobs: cabal install packdiff --project-file=cabal.project.packdiff --installdir=$HOME/.local/bin --overwrite-policy=always fi + # packdiff runs "cabal haddock" with the default project. The haddock + # build of streamly-core needs more than the 500M that limit-build-mem + # sets. cabal.project.local is untracked, it survives packdiff's git + # checkouts. A later -M overrides the earlier one. + - name: Raise the GHC heap limit for haddock builds + run: | + cat > cabal.project.local <