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 < @@ -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 new file mode 100644 index 0000000000..5c58bf8cb5 --- /dev/null +++ b/core/src/Streamly/Internal/Data/SeqMatch.hs @@ -0,0 +1,130 @@ +-- | +-- 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 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 + , addHash + , arrayHash + , elemHashAt + ) +where + +import Data.Bits (shiftL, 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) + +------------------------------------------------------------------------------- +-- 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 +------------------------------------------------------------------------------- + +-- 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..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(..)) @@ -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 @@ -831,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) @@ -920,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 @@ -955,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 @@ -1201,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) @@ -1342,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 @@ -1400,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 @@ -1601,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 @@ -1638,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) @@ -1730,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 @@ -1791,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 @@ -2184,7 +2176,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 +2189,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 +2205,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 +2228,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 +2386,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 +2413,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 @@ -2752,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) @@ -2820,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) = @@ -2860,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 @@ -3010,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) @@ -3084,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) = @@ -3124,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 @@ -3277,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 @@ -3344,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) = @@ -3381,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 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/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 24c5bd7a6f..f982ab21c1 100644 --- a/test/Streamly/Test/Data/Stream/Parse.hs +++ b/test/Streamly/Test/Data/Stream/Parse.hs @@ -19,7 +19,8 @@ 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.Int (Int8) +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 +808,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 ------------------------------------------------------------------------------- @@ -979,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 @@ -1059,6 +1148,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 @@ -1093,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