b7d8b96 · streamly · git.iko.soy

git.iko.soy/streamly

on master help sign in

Add pinned versions of compaction functions

b7d8b96Harendra Kumar3 years ago

core/src/Streamly/Internal/Data/Array.hs+20

compare with master

unifiedsplit

@@ -74,6 +74,8 @@ module Streamly.Internal.Data.Array

74 , interposeSuffix

75 , intercalateSuffix

77+ , compactLE

78+ , pinnedCompactLE

77 , compactOnByte

79 , compactOnByte

78 , compactOnByteSuffix

80 , compactOnByteSuffix

@@ -624,6 +626,24 @@ intercalateSuffix :: (Monad m, Unbox a)

624 => Array a -> Stream m (Array a) -> Stream m a

626 => Array a -> Stream m (Array a) -> Stream m a

625intercalateSuffix = D.intercalateSuffix reader

627intercalateSuffix = D.intercalateSuffix reader

626

628

629+-- | @compactLE n@ coalesces adjacent arrays in the input stream

630+-- only if the combined size would be less than or equal to n.

631+--

632+-- Generates unpinned arrays irrespective of the pinning status of input

633+-- arrays.

634+{-# INLINE_NORMAL compactLE #-}

635+compactLE :: (MonadIO m, Unbox a)

636+ => Int -> Stream m (Array a) -> Stream m (Array a)

637+compactLE n stream =

638+ D.map unsafeFreeze $ MA.compactLE n $ D.map unsafeThaw stream

639+

640+-- | Pinned version of 'compactLE'.

641+{-# INLINE_NORMAL pinnedCompactLE #-}

642+pinnedCompactLE :: (MonadIO m, Unbox a)

643+ => Int -> Stream m (Array a) -> Stream m (Array a)

644+pinnedCompactLE n stream =

645+ D.map unsafeFreeze $ MA.pinnedCompactLE n $ D.map unsafeThaw stream

646+

627-- | Split a stream of arrays on a given separator byte, dropping the separator

647-- | Split a stream of arrays on a given separator byte, dropping the separator

628-- and coalescing all the arrays between two separators into a single array.

648-- and coalescing all the arrays between two separators into a single array.

629--

649--

core/src/Streamly/Internal/Data/Array/Type.hs+21−9

@@ -123,9 +123,10 @@ module Streamly.Internal.Data.Array.Type

123

124 -- *** Compact

125 -- | Append the arrays in a stream to form a stream of larger arrays.

126− , compactLE

127 , fCompactGE

126 , fCompactGE

127+ , fPinnedCompactGE

128 , lCompactGE

129+ , lPinnedCompactGE

129 , compactGE

130 , compactGE

130

131

131 -- ** Deprecated

132 -- ** Deprecated

@@ -502,26 +503,28 @@ flattenArraysRev = concatRev

502-- arrays would have no capacity to append, therefore, a copy will be forced

503-- arrays would have no capacity to append, therefore, a copy will be forced

503-- anyway.

504-- anyway.

504

505

505−-- | Scan @compactLE n@ coalesces adjacent arrays in the input stream

506−-- only if the combined size would be less than or equal to n.

507−{-# INLINE_NORMAL compactLE #-}

508−compactLE :: (MonadIO m, Unbox a)

509− => Int -> Stream m (Array a) -> Stream m (Array a)

510−compactLE n stream =

511− D.map unsafeFreeze $ MA.rCompactLE n $ D.map unsafeThaw stream

512−

513-- | Fold @fCompactGE n@ coalesces adjacent arrays in the input stream

506-- | Fold @fCompactGE n@ coalesces adjacent arrays in the input stream

514-- until the size becomes greater than or equal to n.

507-- until the size becomes greater than or equal to n.

515--

508--

509+-- Generates unpinned arrays irrespective of the pinning status of input

510+-- arrays.

516{-# INLINE_NORMAL fCompactGE #-}

511{-# INLINE_NORMAL fCompactGE #-}

517fCompactGE :: (MonadIO m, Unbox a) => Int -> Fold m (Array a) (Array a)

512fCompactGE :: (MonadIO m, Unbox a) => Int -> Fold m (Array a) (Array a)

518fCompactGE n = fmap unsafeFreeze $ Fold.lmap unsafeThaw $ MA.fCompactGE n

513fCompactGE n = fmap unsafeFreeze $ Fold.lmap unsafeThaw $ MA.fCompactGE n

519

514

515+-- | PInned version of 'fCompactGE'.

516+{-# INLINE_NORMAL fPinnedCompactGE #-}

517+fPinnedCompactGE :: (MonadIO m, Unbox a) => Int -> Fold m (Array a) (Array a)

518+fPinnedCompactGE n =

519+ fmap unsafeFreeze $ Fold.lmap unsafeThaw $ MA.fPinnedCompactGE n

520+

520-- | @compactGE n stream@ coalesces adjacent arrays in the @stream@ until

521-- | @compactGE n stream@ coalesces adjacent arrays in the @stream@ until

521-- the size becomes greater than or equal to @n@.

522-- the size becomes greater than or equal to @n@.

522--

523--

523-- >>> compactGE n = Stream.foldMany (Array.fCompactGE n)

524-- >>> compactGE n = Stream.foldMany (Array.fCompactGE n)

524--

525--

526+-- Generates unpinned arrays irrespective of the pinning status of input

527+-- arrays.

525{-# INLINE compactGE #-}

528{-# INLINE compactGE #-}

526compactGE ::

529compactGE ::

527 (MonadIO m, Unbox a)

530 (MonadIO m, Unbox a)

@@ -533,12 +536,21 @@ compactGE n stream =

533--

536--

534-- >>> lCompactGE n = Fold.many (Array.fCompactGE n)

537-- >>> lCompactGE n = Fold.many (Array.fCompactGE n)

535--

538--

539+-- Generates unpinned arrays irrespective of the pinning status of input

540+-- arrays.

536{-# INLINE_NORMAL lCompactGE #-}

541{-# INLINE_NORMAL lCompactGE #-}

537lCompactGE :: (MonadIO m, Unbox a)

542lCompactGE :: (MonadIO m, Unbox a)

538 => Int -> Fold m (Array a) () -> Fold m (Array a) ()

543 => Int -> Fold m (Array a) () -> Fold m (Array a) ()

539lCompactGE n fld =

544lCompactGE n fld =

540 Fold.lmap unsafeThaw $ MA.lCompactGE n (Fold.lmap unsafeFreeze fld)

545 Fold.lmap unsafeThaw $ MA.lCompactGE n (Fold.lmap unsafeFreeze fld)

541

546

547+-- | Pinned version of 'lCompactGE'.

548+{-# INLINE_NORMAL lPinnedCompactGE #-}

549+lPinnedCompactGE :: (MonadIO m, Unbox a)

550+ => Int -> Fold m (Array a) () -> Fold m (Array a) ()

551+lPinnedCompactGE n fld =

552+ Fold.lmap unsafeThaw $ MA.lPinnedCompactGE n (Fold.lmap unsafeFreeze fld)

553+

542-------------------------------------------------------------------------------

554-------------------------------------------------------------------------------

543-- Splitting

555-- Splitting

544-------------------------------------------------------------------------------

556-------------------------------------------------------------------------------

core/src/Streamly/Internal/Data/MutArray.hs+20−3

@@ -14,6 +14,7 @@ module Streamly.Internal.Data.MutArray

14 , sliceIndexerFromLen

15 , slicerFromLen

16 , compactLE

17+ , pinnedCompactLE

17 , compactOnByte

18 , compactOnByte

18 , compactOnByteSuffix

19 , compactOnByteSuffix

19 -- * Unboxed IORef

20 -- * Unboxed IORef

@@ -29,6 +30,7 @@ where

30import Control.Monad.IO.Class (MonadIO(..))

31import Control.Monad.IO.Class (MonadIO(..))

31import Data.Word (Word8)

32import Data.Word (Word8)

33+import Streamly.Internal.Data.MutByteArray.Type (PinnedState(..))

32import Streamly.Internal.Data.Stream.Type (Stream)

34import Streamly.Internal.Data.Stream.Type (Stream)

33import Streamly.Internal.Data.Unbox (Unbox)

35import Streamly.Internal.Data.Unbox (Unbox)

34import Streamly.Internal.Data.Unfold.Type (Unfold(..))

36import Streamly.Internal.Data.Unfold.Type (Unfold(..))

@@ -91,15 +93,30 @@ getSlicesFromLen = slicerFromLen

91-- Compacting Streams of Arrays

93-- Compacting Streams of Arrays

92-------------------------------------------------------------------------------

94-------------------------------------------------------------------------------

94−-- | Scan @compactLE n@ coalesces adjacent arrays in the input stream

96+-- | @compactLE maxElems@ coalesces adjacent arrays in the input stream

95−-- only if the combined size would be less than or equal to n.

97+-- only if the combined size would be less than or equal to @maxElems@

98+-- elements. Note that it won't split an array if the original array is already

99+-- larger than maxElems.

100+--

101+-- @maxElems@ must be greater than 0.

102+--

103+-- Generates unpinned arrays irrespective of the pinning status of input

104+-- arrays.

96{-# INLINE compactLE #-}

105{-# INLINE compactLE #-}

97compactLE :: (MonadIO m, Unbox a) =>

106compactLE :: (MonadIO m, Unbox a) =>

98 Int -> Stream m (MutArray a) -> Stream m (MutArray a)

107 Int -> Stream m (MutArray a) -> Stream m (MutArray a)

99-- XXX compactLE can be moved to MutArray/Type if we are not using the parser

108-- XXX compactLE can be moved to MutArray/Type if we are not using the parser

100-- to implement it.

109-- to implement it.

110+compactLE = compactLeAs Unpinned

111+-- The parser version turns out to be a little bit slower.

101-- compactLE n = Stream.catRights . Stream.parseManyD (pCompactLE n)

112-- compactLE n = Stream.catRights . Stream.parseManyD (pCompactLE n)

113+

102−compactLE = rCompactLE

114+-- | Pinned version of 'compactLE'.

115+{-# INLINE pinnedCompactLE #-}

116+pinnedCompactLE :: forall m a. (MonadIO m, Unbox a)

117+ => Int -> Stream m (MutArray a) -> Stream m (MutArray a)

118+pinnedCompactLE = compactLeAs Pinned

119+-- pinnedCompactLE n = Stream.catRights . Stream.parseManyD (pPinnedCompactLE n)

103

120

104data SplitState s arr

121data SplitState s arr

105 = Initial s

122 = Initial s

core/src/Streamly/Internal/Data/MutArray/Type.hs+100−50

@@ -251,9 +251,12 @@ module Streamly.Internal.Data.MutArray.Type

251 -- | Append the arrays in a stream to form a stream of larger arrays.

252 , SpliceState (..)

253 , pCompactLE

254− , rCompactLE

254+ , pPinnedCompactLE

255+ , compactLeAs

255 , fCompactGE

256 , fCompactGE

257+ , fPinnedCompactGE

256 , lCompactGE

258 , lCompactGE

259+ , lPinnedCompactGE

257 , compactGE

260 , compactGE

258 , compactEQ

261 , compactEQ

259

262

@@ -839,15 +842,15 @@ roundDownTo elemSize size = size - (size `mod` elemSize)

839-- Since this is not inlined Unboxed consrraint leads to dictionary passing

842-- Since this is not inlined Unboxed consrraint leads to dictionary passing

840-- which complicates some inspection tests.

843-- which complicates some inspection tests.

841--

844--

842−{-# NOINLINE reallocExplicit #-}

845+{-# NOINLINE reallocExplicitAs #-}

843−reallocExplicit :: Int -> Int -> MutArray a -> IO (MutArray a)

846+reallocExplicitAs :: PinnedState -> Int -> Int -> MutArray a -> IO (MutArray a)

844−reallocExplicit elemSize newCapacityInBytes MutArray{..} = do

847+reallocExplicitAs ps elemSize newCapacityInBytes MutArray{..} = do

845 assertM(arrEnd <= arrBound)

848 assertM(arrEnd <= arrBound)

846

849

847 -- Allocate new array

850 -- Allocate new array

848 let newCapMaxInBytes = roundUpLargeArray newCapacityInBytes

851 let newCapMaxInBytes = roundUpLargeArray newCapacityInBytes

849 contents <-

852 contents <-

850− if Unboxed.isPinned arrContents

853+ if ps == Pinned

851 then Unboxed.pinnedNew newCapMaxInBytes

854 then Unboxed.pinnedNew newCapMaxInBytes

852 else Unboxed.new newCapMaxInBytes

855 else Unboxed.new newCapMaxInBytes

853 let !(MutByteArray mbarrFrom#) = arrContents

856 let !(MutByteArray mbarrFrom#) = arrContents

@@ -887,7 +890,12 @@ reallocExplicit elemSize newCapacityInBytes MutArray{..} = do

887-- If the original array is pinned, the newly allocated array is also pinned.

890-- If the original array is pinned, the newly allocated array is also pinned.

888{-# INLINABLE realloc #-}

891{-# INLINABLE realloc #-}

889realloc :: forall m a. (MonadIO m, Unbox a) => Int -> MutArray a -> m (MutArray a)

892realloc :: forall m a. (MonadIO m, Unbox a) => Int -> MutArray a -> m (MutArray a)

893+realloc bytes arr =

894+ let ps =

895+ if isPinned arr

896+ then Pinned

897+ else Unpinned

890−realloc bytes arr = liftIO $ reallocExplicit (SIZE_OF(a)) bytes arr

898+ in liftIO $ reallocExplicitAs ps (SIZE_OF(a)) bytes arr

891

899

892-- | @reallocWith label capSizer minIncrBytes array@. The label is used

900-- | @reallocWith label capSizer minIncrBytes array@. The label is used

893-- in error messages and the capSizer is used to determine the capacity of the

901-- in error messages and the capSizer is used to determine the capacity of the

@@ -2284,7 +2292,7 @@ writeWithAs ps elemCount =

2284 | INDEX_NEXT(end,a) > bound = do

2292 | INDEX_NEXT(end,a) > bound = do

2285 let oldSize = end - start

2293 let oldSize = end - start

2286 newSize = max (oldSize * 2) 1

2294 newSize = max (oldSize * 2) 1

2287− arr1 <- liftIO $ reallocExplicit (SIZE_OF(a)) newSize arr

2295+ arr1 <- liftIO $ reallocExplicitAs ps (SIZE_OF(a)) newSize arr

2288 snocUnsafe arr1 x

2296 snocUnsafe arr1 x

2289 step arr x = snocUnsafe arr x

2297 step arr x = snocUnsafe arr x

2290

2298

@@ -2943,19 +2951,11 @@ byteEq arr1 arr2 = fmap (EQ ==) $ byteCmp arr1 arr2

2943-- can leave some memory unused. They can split the last array to fit it

2951-- can leave some memory unused. They can split the last array to fit it

2944-- exactly in the space.

2952-- exactly in the space.

2945

2953

2946−-- | Parser @pCompactLE maxElems@ coalesces adjacent arrays in the input stream

2947−-- only if the combined size would be less than or equal to @maxElems@

2948−-- elements. Note that it won't split an array if the original array is already

2949−-- larger than maxElems.

2950−--

2951−-- @maxElems@ must be greater than 0.

2952−--

2953−-- /Internal/

2954−{-# INLINE_NORMAL pCompactLE #-}

2954+{-# INLINE_NORMAL pCompactLeAs #-}

2955−pCompactLE ::

2955+pCompactLeAs ::

2956 forall m a. (MonadIO m, Unbox a)

2957− => Int -> Parser (MutArray a) m (MutArray a)

2957+ => PinnedState -> Int -> Parser (MutArray a) m (MutArray a)

2958−pCompactLE maxElems = Parser step initial extract

2958+pCompactLeAs ps maxElems = Parser step initial extract

2959

2960 where

2961

@@ -2986,7 +2986,8 @@ pCompactLE maxElems = Parser step initial extract

2986 else do

2987 buf1 <-

2988 if byteCapacity buf < maxBytes

2989+ then liftIO $ reallocExplicitAs

2989− then realloc maxBytes buf

2990+ ps (SIZE_OF(a)) maxBytes buf

2990 else return buf

2991 else return buf

2991 buf2 <- spliceUnsafe buf1 arr

2992 buf2 <- spliceUnsafe buf1 arr

2992 return $ Parser.Partial 0 (Just buf2)

2993 return $ Parser.Partial 0 (Just buf2)

@@ -2994,6 +2995,30 @@ pCompactLE maxElems = Parser step initial extract

2994 extract Nothing = return $ Parser.Done 0 nil

2995 extract Nothing = return $ Parser.Done 0 nil

2995 extract (Just buf) = return $ Parser.Done 0 buf

2996 extract (Just buf) = return $ Parser.Done 0 buf

2996

2997

2998+-- | Parser @pCompactLE maxElems@ coalesces adjacent arrays in the input stream

2999+-- only if the combined size would be less than or equal to @maxElems@

3000+-- elements. Note that it won't split an array if the original array is already

3001+-- larger than maxElems.

3002+--

3003+-- @maxElems@ must be greater than 0.

3004+--

3005+-- Generates unpinned arrays irrespective of the pinning status of input

3006+-- arrays.

3007+--

3008+-- /Internal/

3009+{-# INLINE pCompactLE #-}

3010+pCompactLE ::

3011+ forall m a. (MonadIO m, Unbox a)

3012+ => Int -> Parser (MutArray a) m (MutArray a)

3013+pCompactLE = pCompactLeAs Unpinned

3014+

3015+-- | Pinned version of 'pCompactLE'.

3016+{-# INLINE pPinnedCompactLE #-}

3017+pPinnedCompactLE ::

3018+ forall m a. (MonadIO m, Unbox a)

3019+ => Int -> Parser (MutArray a) m (MutArray a)

3020+pPinnedCompactLE = pCompactLeAs Pinned

3021+

2997data SpliceState s arr

3022data SpliceState s arr

2998 = SpliceInitial s

3023 = SpliceInitial s

2999 | SpliceBuffering s arr

3024 | SpliceBuffering s arr

@@ -3005,18 +3030,10 @@ data SpliceState s arr

3005-- immutable array never has additional space so a new array is allocated

3030-- immutable array never has additional space so a new array is allocated

3006-- instead of mutating it.

3031-- instead of mutating it.

3007

3032

3008−-- | Scan @rCompactLE maxElems@ coalesces adjacent arrays in the input stream

3009−-- only if the combined size would be less than or equal to @maxElems@

3010−-- elements. Note that it won't split an array if the original array is already

3011−-- larger than maxElems.

3012−--

3013−-- @maxElems@ must be greater than 0.

3014−--

3015−-- /Internal/

3016−{-# INLINE_NORMAL rCompactLE #-}

3033+{-# INLINE_NORMAL compactLeAs #-}

3017−rCompactLE :: forall m a. (MonadIO m, Unbox a)

3034+compactLeAs :: forall m a. (MonadIO m, Unbox a)

3018− => Int -> D.Stream m (MutArray a) -> D.Stream m (MutArray a)

3035+ => PinnedState -> Int -> D.Stream m (MutArray a) -> D.Stream m (MutArray a)

3019−rCompactLE maxElems (D.Stream step state) =

3036+compactLeAs ps maxElems (D.Stream step state) =

3020 D.Stream step' (SpliceInitial state)

3037 D.Stream step' (SpliceInitial state)

3021

3038

3022 where

3039 where

@@ -3052,7 +3069,8 @@ rCompactLE maxElems (D.Stream step state) =

3052 D.Skip (SpliceYielding buf (SpliceBuffering s arr))

3069 D.Skip (SpliceYielding buf (SpliceBuffering s arr))

3053 else do

3070 else do

3054 buf1 <- if byteCapacity buf < maxBytes

3071 buf1 <- if byteCapacity buf < maxBytes

3072+ then liftIO $ reallocExplicitAs

3055− then realloc maxBytes buf

3073+ ps (SIZE_OF(a)) maxBytes buf

3056 else return buf

3074 else return buf

3057 buf2 <- spliceUnsafe buf1 arr

3075 buf2 <- spliceUnsafe buf1 arr

3058 return $ D.Skip (SpliceBuffering s buf2)

3076 return $ D.Skip (SpliceBuffering s buf2)

@@ -3063,14 +3081,12 @@ rCompactLE maxElems (D.Stream step state) =

3063

3081

3064 step' _ (SpliceYielding arr next) = return $ D.Yield arr next

3082 step' _ (SpliceYielding arr next) = return $ D.Yield arr next

3065

3083

3066−-- | Fold @fCompactGE minElems@ coalesces adjacent arrays in the input stream

3084+

3067−-- until the size becomes greater than or equal to @minElems@.

3068−--

3069−{-# INLINE_NORMAL fCompactGE #-}

3085+{-# INLINE_NORMAL fCompactGeAs #-}

3070−fCompactGE ::

3086+fCompactGeAs ::

3071 forall m a. (MonadIO m, Unbox a)

3087 forall m a. (MonadIO m, Unbox a)

3072− => Int -> FL.Fold m (MutArray a) (MutArray a)

3088+ => PinnedState -> Int -> FL.Fold m (MutArray a) (MutArray a)

3073−fCompactGE minElems = Fold step initial extract extract

3089+fCompactGeAs ps minElems = Fold step initial extract extract

3074

3090

3075 where

3091 where

3076

3092

@@ -3098,7 +3114,8 @@ fCompactGE minElems = Fold step initial extract extract

3098 let len = byteLength buf + byteLength arr

3114 let len = byteLength buf + byteLength arr

3099 buf1 <-

3115 buf1 <-

3100 if byteCapacity buf < len

3116 if byteCapacity buf < len

3117+ then liftIO $ reallocExplicitAs

3101− then realloc (max len minBytes) buf

3118+ ps (SIZE_OF(a)) (max minBytes len) buf

3102 else return buf

3119 else return buf

3103 buf2 <- spliceUnsafe buf1 arr

3120 buf2 <- spliceUnsafe buf1 arr

3104 if len >= minBytes

3121 if len >= minBytes

@@ -3108,15 +3125,30 @@ fCompactGE minElems = Fold step initial extract extract

3108 extract Nothing = return nil

3125 extract Nothing = return nil

3109 extract (Just buf) = return buf

3126 extract (Just buf) = return buf

3110

3127

3111−-- | Like 'compactGE' but for transforming folds instead of stream.

3128+-- | Fold @fCompactGE minElems@ coalesces adjacent arrays in the input stream

3112−--

3129+-- until the size becomes greater than or equal to @minElems@.

3113−-- >>> lCompactGE n = Fold.many (MutArray.fCompactGE n)

3114--

3130--

3131+-- Generates unpinned arrays irrespective of the pinning status of input

3132+-- arrays.

3133+{-# INLINE fCompactGE #-}

3134+fCompactGE ::

3135+ forall m a. (MonadIO m, Unbox a)

3136+ => Int -> FL.Fold m (MutArray a) (MutArray a)

3137+fCompactGE = fCompactGeAs Unpinned

3138+

3139+-- | Pinned version of 'fCompactGE'.

3140+{-# INLINE fPinnedCompactGE #-}

3141+fPinnedCompactGE ::

3142+ forall m a. (MonadIO m, Unbox a)

3143+ => Int -> FL.Fold m (MutArray a) (MutArray a)

3144+fPinnedCompactGE = fCompactGeAs Pinned

3145+

3115−{-# INLINE_NORMAL lCompactGE #-}

3146+{-# INLINE_NORMAL lCompactGeAs #-}

3116−lCompactGE :: forall m a. (MonadIO m, Unbox a)

3147+lCompactGeAs :: forall m a. (MonadIO m, Unbox a)

3117− => Int -> Fold m (MutArray a) () -> Fold m (MutArray a) ()

3148+ => PinnedState -> Int -> Fold m (MutArray a) () -> Fold m (MutArray a) ()

3118−-- lCompactGE n = Fold.many (fCompactGE n)

3149+-- The fold version turns out to be a little bit slower.

3150+-- lCompactGeAs ps n = FL.many (fCompactGeAs ps n)

3119−lCompactGE minElems (Fold step1 initial1 _ final1) =

3151+lCompactGeAs ps minElems (Fold step1 initial1 _ final1) =

3120 Fold step initial extract final

3152 Fold step initial extract final

3121

3153

3122 where

3154 where

@@ -3154,7 +3186,8 @@ lCompactGE minElems (Fold step1 initial1 _ final1) =

3154 step (Tuple' (Just buf) r1) arr = do

3186 step (Tuple' (Just buf) r1) arr = do

3155 let len = byteLength buf + byteLength arr

3187 let len = byteLength buf + byteLength arr

3156 buf1 <- if byteCapacity buf < len

3188 buf1 <- if byteCapacity buf < len

3189+ then liftIO $ reallocExplicitAs

3157− then realloc (max minBytes len) buf

3190+ ps (SIZE_OF(a)) (max minBytes len) buf

3158 else return buf

3191 else return buf

3159 buf2 <- spliceUnsafe buf1 arr

3192 buf2 <- spliceUnsafe buf1 arr

3160 runInner len r1 buf2

3193 runInner len r1 buf2

@@ -3174,6 +3207,23 @@ lCompactGE minElems (Fold step1 initial1 _ final1) =

3174 FL.Partial rr -> final1 rr

3207 FL.Partial rr -> final1 rr

3175 FL.Done _ -> return ()

3208 FL.Done _ -> return ()

3176

3209

3210+-- | Like 'compactGE' but for transforming folds instead of stream.

3211+--

3212+-- >>> lCompactGE n = Fold.many (MutArray.fCompactGE n)

3213+--

3214+-- Generates unpinned arrays irrespective of the pinning status of input

3215+-- arrays.

3216+{-# INLINE lCompactGE #-}

3217+lCompactGE :: forall m a. (MonadIO m, Unbox a)

3218+ => Int -> Fold m (MutArray a) () -> Fold m (MutArray a) ()

3219+lCompactGE = lCompactGeAs Unpinned

3220+

3221+-- | Pinned version of 'lCompactGE'.

3222+{-# INLINE lPinnedCompactGE #-}

3223+lPinnedCompactGE :: forall m a. (MonadIO m, Unbox a)

3224+ => Int -> Fold m (MutArray a) () -> Fold m (MutArray a) ()

3225+lPinnedCompactGE = lCompactGeAs Pinned

3226+

3177-- | @compactGE n stream@ coalesces adjacent arrays in the @stream@ until

3227-- | @compactGE n stream@ coalesces adjacent arrays in the @stream@ until

3178-- the size becomes greater than or equal to @n@.

3228-- the size becomes greater than or equal to @n@.

3179--

3229--

core/src/Streamly/Internal/Data/MutByteArray/Type.hs+1−1

@@ -58,7 +58,7 @@ import GHC.Exts

59data PinnedState

60 = Pinned

61− | Unpinned

61+ | Unpinned deriving (Show, Eq)

63-- XXX can use UnliftedNewtypes

core/src/Streamly/Internal/FileSystem/Handle.hs+1−2

@@ -137,8 +137,7 @@ import Streamly.Internal.Data.Stream.Type (Stream)

137import Streamly.Internal.System.IO (defaultChunkSize)

138

139import qualified Streamly.Data.Fold as FL

140−import qualified Streamly.Data.Array as A

140+import qualified Streamly.Internal.Data.Array as A

141−import qualified Streamly.Internal.Data.Array.Type as A

142import qualified Streamly.Internal.Data.MutArray.Type as MArray

141import qualified Streamly.Internal.Data.MutArray.Type as MArray

143import qualified Streamly.Internal.Data.Refold.Type as Refold

142import qualified Streamly.Internal.Data.Refold.Type as Refold

144import qualified Streamly.Internal.Data.Fold.Type as FL(refoldMany)

143import qualified Streamly.Internal.Data.Fold.Type as FL(refoldMany)