Fix Fold.takeEndBySeq for negative and large elements

core/docs/Changelog.md

@@ -37,6 +37,10 @@
   for `Array`, `Proxy` and derived instances of types with one constructor
   and no fields. `()` and `Proxy` now serialize to a 0 byte, the byte was
   left unwritten.
+* Fix `Fold.takeEndBySeq` and `Fold.takeEndBySeq_` for elements whose
+  `fromEnum` is larger than `maxBound :: Int`, e.g. `Word64`, and
+  `Fold.takeEndBySeq` for elements whose `fromEnum` is negative, e.g.
+  `Int8`.
 
 ### Breaking changes
 

core/src/Streamly/Internal/Data/Fold/Combinators.hs

@@ -1395,6 +1395,8 @@ takeEndBySeq patArr (Fold fstep finitial fextract ffinal) =
     -- Word pattern related
     elemBits = SIZE_OF(a) * 8
 
+    proxy = Proxy :: Proxy a
+
     wordMask :: Word
     wordMask = (1 `shiftL` (elemBits * patLen)) - 1
 
@@ -1403,20 +1405,18 @@ takeEndBySeq patArr (Fold fstep finitial fextract ffinal) =
 
     addToWord = SeqMatch.addToWord elemBits
 
-    -- For Rabin-Karp search
+    -- For Rabin-Karp search, hashes of the element bytes
     coeff = SeqMatch.hashPower patLen
 
-    addCksum cksum a = SeqMatch.addHash cksum (fromIntegral (fromEnum a))
-
-    deltaCksum cksum old new =
-        addCksum cksum new - coeff * fromIntegral (fromEnum old)
+    deltaCksum cksum oldHash newHash =
+        SeqMatch.addHash cksum newHash - coeff * oldHash
 
     -- XXX shall we use a random starting hash or 1 instead of 0?
     -- XXX Need to keep this cached across fold calls in foldmany
     -- XXX We may need refold to inject the cached state instead of
     -- initializing the state every time.
     -- XXX Allocation of ring buffer should also be done once
-    patHash = Array.foldl' addCksum 0 patArr
+    patHash = SeqMatch.arrayHash patArr
 
     step (SplitOnSeqEmpty s) x = do
         res <- fstep s x
@@ -1464,7 +1464,7 @@ takeEndBySeq patArr (Fold fstep finitial fextract ffinal) =
                                 , arrStart = 0
                                 , arrEnd = patBytes
                                 }
-                    let ringHash = Array.foldl' addCksum 0 arr
+                    let ringHash = SeqMatch.arrayHash arr
                     if ringHash == patHash && Array.byteEq arr patArr
                     then Done <$> ffinal s1
                     else return $ Partial $ SplitOnSeqKRLoop s1 ringHash mba 0
@@ -1480,8 +1480,10 @@ takeEndBySeq patArr (Fold fstep finitial fextract ffinal) =
                         , ringSize = patBytes
                         , ringHead = offset
                         }
-                (rb1, old :: a) <- liftIO (RingArray.replace rb x)
-                let ringHash = deltaCksum cksum old x
+                oldHash <- liftIO $ SeqMatch.elemHashAt proxy mba offset
+                rb1 <- liftIO (RingArray.replace_ rb x)
+                newHash <- liftIO $ SeqMatch.elemHashAt proxy mba offset
+                let ringHash = deltaCksum cksum oldHash newHash
                 let rh1 = ringHead rb1
                 matches <-
                     if ringHash == patHash
@@ -1520,6 +1522,10 @@ takeEndBySeq patArr (Fold fstep finitial fextract ffinal) =
 -- See also: 'Streamly.Data.Stream.splitOnSeq' and
 -- 'Streamly.Data.Stream.splitEndBySeq_'.
 --
+-- Known Issues: For a signed element type, e.g. 'Int8', a pattern of two or
+-- more elements whose total size is at most one machine word fails with an
+-- error if the input has a negative element.
+--
 -- /Pre-release/
 --
 {-# INLINE takeEndBySeq_ #-}
@@ -1561,6 +1567,8 @@ takeEndBySeq_ patArr (Fold fstep finitial fextract ffinal) =
     -- Word pattern related
     elemBits = SIZE_OF(a) * 8
 
+    proxy = Proxy :: Proxy a
+
     wordMask :: Word
     wordMask = (1 `shiftL` (elemBits * patLen)) - 1
 
@@ -1571,20 +1579,18 @@ takeEndBySeq_ patArr (Fold fstep finitial fextract ffinal) =
 
     addToWord = SeqMatch.addToWord elemBits
 
-    -- For Rabin-Karp search
+    -- For Rabin-Karp search, hashes of the element bytes
     coeff = SeqMatch.hashPower patLen
 
-    addCksum cksum a = SeqMatch.addHash cksum (fromIntegral (fromEnum a))
-
-    deltaCksum cksum old new =
-        addCksum cksum new - coeff * fromIntegral (fromEnum old)
+    deltaCksum cksum oldHash newHash =
+        SeqMatch.addHash cksum newHash - coeff * oldHash
 
     -- XXX shall we use a random starting hash or 1 instead of 0?
     -- XXX Need to keep this cached across fold calls in foldMany
     -- XXX We may need refold to inject the cached state instead of
     -- initializing the state every time.
     -- XXX Allocation of ring buffer should also be done once
-    patHash = Array.foldl' addCksum 0 patArr
+    patHash = SeqMatch.arrayHash patArr
 
     step (SplitOnSeqEmpty s) x = do
         res <- fstep s x
@@ -1627,7 +1633,7 @@ takeEndBySeq_ patArr (Fold fstep finitial fextract ffinal) =
                         , arrStart = 0
                         , arrEnd = patBytes
                         }
-            let ringHash = Array.foldl' addCksum 0 arr
+            let ringHash = SeqMatch.arrayHash arr
             if ringHash == patHash && Array.byteEq arr patArr
             then Done <$> ffinal s
             else return $ Partial $ SplitOnSeqKRLoop s ringHash mba 0
@@ -1638,11 +1644,13 @@ takeEndBySeq_ patArr (Fold fstep finitial fextract ffinal) =
                 , ringSize = patBytes
                 , ringHead = offset
                 }
+        oldHash <- liftIO $ SeqMatch.elemHashAt proxy mba offset
         (rb1, old :: a) <- liftIO (RingArray.replace rb x)
+        newHash <- liftIO $ SeqMatch.elemHashAt proxy mba offset
         res <- fstep s old
         case res of
             Partial s1 -> do
-                let ringHash = deltaCksum cksum old x
+                let ringHash = deltaCksum cksum oldHash newHash
                 let rh1 = ringHead rb1
                 matches <-
                     if ringHash == patHash

core/src/Streamly/Internal/Data/SeqMatch.hs

@@ -46,11 +46,12 @@ import Streamly.Internal.System.IO (unsafeInlineIO)
 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.
+-- | @addToWord bits word a@ shifts @word@ left by @bits@ and puts the low
+-- @bits@ bits of @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)
+addToWord bits wd a =
+    (wd `shiftL` bits) .|. (fromIntegral (fromEnum a) .&. elemMask bits)
 
 -- | @wordToElem bits word@ decodes the element packed by 'addToWord' in the
 -- low @bits@ bits of @word@, @bits@ must be less than the word size.

core/src/Streamly/Internal/Data/Stream/Parse.hs

@@ -1093,6 +1093,10 @@ takeEndBySeq = takeEndBySeqWith True
 -- >>> f "" "abcdefghijk"
 -- ""
 --
+-- Known Issues: For a signed element type, e.g. 'Int8', a pattern of two or
+-- more elements whose total size is at most one machine word fails with an
+-- error if the input has a negative element.
+--
 {-# INLINE takeEndBySeq_ #-}
 takeEndBySeq_
     :: forall m a. (MonadIO m, Unbox a, Enum a, Eq a)
@@ -1175,6 +1179,10 @@ data SplitOnSeqState mba rb rh ck w fs s b x =
 --
 -- Uses Rabin-Karp algorithm for substring search.
 --
+-- Known Issues: For a signed element type, e.g. 'Int8', a pattern of two or
+-- more elements whose total size is at most one machine word fails with an
+-- error if the input has a negative element.
+--
 {-# INLINE_NORMAL splitSepBySeq_ #-}
 splitSepBySeq_, splitOnSeq
     :: forall m a b. (MonadIO m, Unbox a, Enum a, Eq a)
@@ -2013,6 +2021,10 @@ splitEndBySeq = splitOnSuffixSeq True
 --
 -- Uses Rabin-Karp algorithm for substring search.
 --
+-- Known Issues: For a signed element type, e.g. 'Int8', a pattern of two or
+-- more elements whose total size is at most one machine word fails with an
+-- error if the input has a negative element.
+--
 {-# INLINE_NORMAL splitEndBySeq_ #-}
 splitEndBySeq_
     :: forall m a b. (MonadIO m, Unbox a, Enum a, Eq a)
@@ -2755,6 +2767,10 @@ data DropInfixState mba rb rh ck w s a =
 --
 -- Space: @O(n)@ where n is the length of the infix.
 --
+-- Known Issues: For a signed element type, e.g. 'Int8', a pattern of two or
+-- more elements whose total size is at most one machine word fails with an
+-- error if the input has a negative element.
+--
 -- See also 'Streamly.Internal.Data.Stream.Eliminate.stripInfix'.
 {-# INLINE_NORMAL dropMatches #-}
 dropMatches
@@ -3008,6 +3024,10 @@ data ReplaceInfixState mba rb rh ck w s a =
 --
 -- Space: @O(n)@ where n is the length of the infix pattern.
 --
+-- Known Issues: For a signed element type, e.g. 'Int8', a pattern of two or
+-- more elements whose total size is at most one machine word fails with an
+-- error if the input has a negative element.
+--
 {-# INLINE_NORMAL replaceMatches #-}
 replaceMatches
     :: forall m a. (MonadIO m, Unbox a, Enum a, Eq a)
@@ -3272,6 +3292,10 @@ data DropSuffixState mba rb rh w s a =
 --
 -- Space: @O(n)@ where n is the length of the suffix.
 --
+-- Known Issues: For a signed element type, e.g. 'Int8', a pattern of two or
+-- more elements whose total size is at most one machine word fails with an
+-- error if the input has a negative element.
+--
 -- See also 'Streamly.Internal.Data.Stream.Eliminate.stripSuffix'.
 {-# INLINE_NORMAL dropSuffix #-}
 dropSuffix

test/Streamly/Test/Data/Fold/Combinators.hs

@@ -590,11 +590,11 @@ main = hspec $ do
         it "slide2" slide2
         it "takeEndBySeq" takeEndBySeq
         it "takeEndBySeq_" takeEndBySeq_
-        xit "takeEndBySeq signed elements" takeEndBySeqSigned
+        it "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 "takeEndBySeq Word64" takeEndBySeqWord64
+        it "takeEndBySeq_ Word64" takeEndBySeq_Word64
         it "distributeScan" distributeScan
         it "distributeScan last input" distributeScanLast
         it "sampleFromthen offset" sampleFromthenOffset