d29030f · streamly · git.iko.soy

git.iko.soy/streamly

Use Pinned arrays only for IO and Unpinned arrays everywhere else

commitd29030f3bf25c088bc73ed958fdcefc32fe76bc5 authorAdithya Kumar <adithya@composewell.com> dateTue 18 Jul 2023, 20:35 (3 years ago) parent4727060

12 files changed+184−108

compare with master

unifiedsplit

core/docs/Changelog.md+2

@@ -2,6 +2,8 @@

33## Unreleased

5+* Arrays can now be pinned or unpinned. They were always pinned earlier.

57## 0.1.0 (March 2023)

79Also see [streamly-core-0.1.0 API Changelog](/core/docs/ApiChangelogs/0.1.0.txt) or

core/src/Streamly/Data/Array.hs+3−6

@@ -116,12 +116,9 @@ import Prelude hiding (read)

116116--

117117-- == Pinned and Unpinned Arrays

118118--

119−-- The array type can use both pinned and unpinned memory under the hood.

120−-- Currently the array creation APIs create arrays in pinned memory but it will

121−-- change to unpinned in future releases. The change should not affect users

122−-- functionally unless they are directly accessing the internal memory of the

123−-- array via internal APIs. As of now unpinned arrays can be created using

124−-- unreleased APIs.

119+-- The array type can use both pinned and unpinned memory under the hood. The

120+-- arrays are unpinned by default and are only pinned while doing any I/O

121+-- operations.

125122--

126123-- Unpinned arrays have the advantage of allowing automatic defragmentation of

127124-- the memory by GC. Whereas pinned arrays have the advantage of not requiring

core/src/Streamly/Internal/Data/Array.hs+6−4

@@ -31,7 +31,7 @@ module Streamly.Internal.Data.Array

3131

3232 -- Monadic Folds

3333 , A.writeN -- drop new

34− , A.writeNAligned

34+ , A.pinnedWriteNAligned

3535 , A.write -- full buffer

3636 , writeLastN

3737

@@ -297,7 +297,7 @@ writeLastN n

297297 in fmap f $ liftIO $ RB.new n

298298

299299 done (Tuple3Fused' rb rh i) = do

300− arr <- liftIO $ MA.newPinned n

300+ arr <- liftIO $ MA.new n

301301 foldFunc i rh snoc' arr rb

302302

303303 -- XXX We should write a read unfold for ring.

@@ -534,7 +534,9 @@ cast arr =

534534 then Nothing

535535 else Just $ castUnsafe arr

536536

537−-- | Convert an array of any type into a null terminated CString Ptr.

537+-- | Convert an array of any type into a null terminated CString Ptr. If the

538+-- array is unpinned it is first converted to a pinned array which requires a

539+-- copy.

538540--

539541-- /Unsafe/

540542--

@@ -544,8 +546,8 @@ cast arr =

544546--

545547asCStringUnsafe :: Array a -> (CString -> IO b) -> IO b

546548asCStringUnsafe arr act = do

547− -- XXX Ensure a pinned allocation here.

548549 let arr1 = asBytes arr <> A.fromList [0]

550+ -- asPtrUnsafe makes sure the array is pinned

549551 asPtrUnsafe arr1 $ \ptr -> act (castPtr ptr)

550552

551553-------------------------------------------------------------------------------

core/src/Streamly/Internal/Data/Array/Mut/Type.hs+97−60

@@ -35,6 +35,7 @@ module Streamly.Internal.Data.Array.Mut.Type

3535 , touch

3636 , pin

3737 , unpin

38+ , isPinned

3839

3940 -- * Constructing and Writing

4041 -- ** Construction

@@ -48,15 +49,17 @@ module Streamly.Internal.Data.Array.Mut.Type

4849 , newArrayWith

4950

5051 -- *** Initialized Arrays

51− , withNewArrayUnsafe

52+ , withNewArrayUnsafePinned

5253

5354 -- *** From streams

5455 , ArrayUnsafe (..)

5556 , writeNWithUnsafe

5657 , writeNWith

5758 , writeNUnsafe

59+ , writeNUnsafeAs

5860 , writeN

59− , writeNAligned

61+ , writeNAs

62+ , pinnedWriteNAligned

6063

6164 , writeWith

6265 , write

@@ -185,6 +188,7 @@ module Streamly.Internal.Data.Array.Mut.Type

185188

186189 -- ** Construct from streams

187190 , chunksOf

191+ , chunksOfAs

188192 , arrayStreamKFromStreamD

189193 , writeChunks

190194

@@ -236,6 +240,7 @@ import Foreign.Ptr (plusPtr, minusPtr, nullPtr)

236240import Streamly.Internal.Data.Unbox

237241 ( MutableByteArray(..)

238242 , Unbox(..)

243+ , PinnedState(..)

239244 , getMutableByteArray#

240245 , touch

241246 )

@@ -366,6 +371,10 @@ unpin arr@MutArray{..} = do

366371 contents <- Unboxed.unpin arrContents

367372 return $ arr {arrContents = contents}

368373

374+{-# INLINE isPinned #-}

375+isPinned :: MutArray a -> Bool

376+isPinned MutArray{..} = Unboxed.isPinned arrContents

377+

369378-------------------------------------------------------------------------------

370379-- Construction

371380-------------------------------------------------------------------------------

@@ -412,6 +421,20 @@ nil ::

412421 MutArray a

413422nil = MutArray Unboxed.nil 0 0 0

414423

424+{-# INLINE newBytesAs #-}

425+newBytesAs :: MonadIO m =>

426+#ifdef DEVBUILD

427+ Unbox a =>

428+#endif

429+ PinnedState -> Int -> m (MutArray a)

430+newBytesAs ps bytes = do

431+ contents <- liftIO $ Unboxed.newBytes ps bytes

432+ return $ MutArray

433+ { arrContents = contents

434+ , arrStart = 0

435+ , arrEnd = 0

436+ , arrBound = bytes

437+ }

415438

416439-- | Allocates a pinned empty array that can hold 'count' items. The memory of

417440-- the array is uninitialized and the allocation is aligned as per the

@@ -424,14 +447,7 @@ newPinnedBytes :: MonadIO m =>

424447 Unbox a =>

425448#endif

426449 Int -> m (MutArray a)

427−newPinnedBytes bytes = do

428− contents <- liftIO $ Unboxed.newPinnedBytes bytes

429− return $ MutArray

430− { arrContents = contents

431− , arrStart = 0

432− , arrEnd = 0

433− , arrBound = bytes

434− }

450+newPinnedBytes = newBytesAs Pinned

435451

436452-- | Like 'newArrayWith' but using an allocator is a pinned memory allocator and

437453-- the alignment is dictated by the 'Unboxed' instance of the type.

@@ -442,6 +458,13 @@ newAlignedPinned :: (MonadIO m, Unbox a) => Int -> Int -> m (MutArray a)

442458newAlignedPinned =

443459 newArrayWith (\s a -> liftIO $ Unboxed.newAlignedPinnedBytes s a)

444460

461+{-# INLINE newAs #-}

462+newAs :: (MonadIO m, Unbox a) => PinnedState -> Int -> m (MutArray a)

463+newAs ps =

464+ newArrayWith

465+ (\s _ -> liftIO $ Unboxed.newBytes ps s)

466+ (error "new: alignment is not used in unpinned arrays.")

467+

445468-- XXX can unaligned allocation be more efficient when alignment is not needed?

446469--

447470-- | Allocates an empty pinned array that can hold 'count' items. The memory of

@@ -450,20 +473,14 @@ newAlignedPinned =

450473--

451474{-# INLINE newPinned #-}

452475newPinned :: forall m a. (MonadIO m, Unbox a) => Int -> m (MutArray a)

453−newPinned =

454− newArrayWith

455− (\s _ -> liftIO $ Unboxed.newPinnedBytes s)

456− (error "newPinned: alignSize is not used")

476+newPinned = newAs Pinned

457477

458478-- | Allocates an empty unpinned array that can hold 'count' items. The memory

459479-- of the array is uninitialized.

460480--

461481{-# INLINE new #-}

462482new :: (MonadIO m, Unbox a) => Int -> m (MutArray a)

463−new =

464− newArrayWith

465− (\s _ -> liftIO $ Unboxed.newUnpinnedBytes s)

466− (error "new: alignment is not used in unpinned arrays.")

483+new = newAs Unpinned

467484

468485-- XXX This should create a full length uninitialzed array so that the pointer

469486-- can be used.

@@ -472,10 +489,10 @@ new =

472489-- the array start pointer.

473490--

474491-- /Internal/

475−{-# INLINE withNewArrayUnsafe #-}

476−withNewArrayUnsafe ::

492+{-# INLINE withNewArrayUnsafePinned #-}

493+withNewArrayUnsafePinned ::

477494 (MonadIO m, Unbox a) => Int -> (Ptr a -> m ()) -> m (MutArray a)

478−withNewArrayUnsafe count f = do

495+withNewArrayUnsafePinned count f = do

479496 arr <- newPinned count

480497 asPtrUnsafe arr

481498 $ \p -> f p >> return arr

@@ -736,7 +753,10 @@ reallocExplicit elemSize newCapacityInBytes MutArray{..} = do

736753

737754 -- Allocate new array

738755 let newCapMaxInBytes = roundUpLargeArray newCapacityInBytes

739− contents <- Unboxed.newPinnedBytes newCapMaxInBytes

756+ contents <-

757+ if Unboxed.isPinned arrContents

758+ then Unboxed.newPinnedBytes newCapMaxInBytes

759+ else Unboxed.newUnpinnedBytes newCapMaxInBytes

740760 let !(MutableByteArray mbarrFrom#) = arrContents

741761 !(MutableByteArray mbarrTo#) = contents

742762

@@ -1299,20 +1319,10 @@ data GroupState s contents start end bound

12991319 contents start end bound (GroupState s contents start end bound)

13001320 | GroupFinish

13011321

1302−-- | @chunksOf n stream@ groups the input stream into a stream of

1303−-- arrays of size n.

1304−--

1305−-- @chunksOf n = StreamD.foldMany (MutArray.writeN n)@

1306−--

1307−-- /Pre-release/

1308−{-# INLINE_NORMAL chunksOf #-}

1309−chunksOf :: forall m a. (MonadIO m, Unbox a)

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

1311−-- XXX the idiomatic implementation leads to large regression in the D.reverse'

1312−-- benchmark. It seems it has difficulty producing optimized code when

1313−-- converting to StreamK. Investigate GHC optimizations.

1314−-- chunksOf n = D.foldMany (writeN n)

1315−chunksOf n (D.Stream step state) =

1322+{-# INLINE_NORMAL chunksOfAs #-}

1323+chunksOfAs :: forall m a. (MonadIO m, Unbox a)

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

1325+chunksOfAs ps n (D.Stream step state) =

13161326 D.Stream step' (GroupStart state)

13171327

13181328 where

@@ -1324,7 +1334,7 @@ chunksOf n (D.Stream step state) =

13241334 error $ "Streamly.Internal.Data.MutArray.Mut.Type.chunksOf: "

13251335 ++ "the size of arrays [" ++ show n

13261336 ++ "] must be a natural number"

1327− (MutArray contents start end bound :: MutArray a) <- liftIO $ newPinned n

1337+ (MutArray contents start end bound :: MutArray a) <- newAs ps n

13281338 return $ D.Skip (GroupBuffer st contents start end bound)

13291339

13301340 step' gst (GroupBuffer st contents start end bound) = do

@@ -1350,6 +1360,22 @@ chunksOf n (D.Stream step state) =

13501360

13511361 step' _ GroupFinish = return D.Stop

13521362

1363+-- | @chunksOf n stream@ groups the input stream into a stream of

1364+-- arrays of size n.

1365+--

1366+-- @chunksOf n = StreamD.foldMany (MutArray.writeN n)@

1367+--

1368+-- /Pre-release/

1369+{-# INLINE_NORMAL chunksOf #-}

1370+chunksOf :: forall m a. (MonadIO m, Unbox a)

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

1372+-- XXX the idiomatic implementation leads to large regression in the D.reverse'

1373+-- benchmark. It seems it has difficulty producing optimized code when

1374+-- converting to StreamK. Investigate GHC optimizations.

1375+-- chunksOf n = D.foldMany (writeN n)

1376+chunksOf = chunksOfAs Unpinned

1377+

1378+-- XXX This should take a PinnedState

13531379-- XXX buffer to a list instead?

13541380-- | Buffer the stream into arrays in memory.

13551381{-# INLINE arrayStreamKFromStreamD #-}

@@ -1751,18 +1777,23 @@ writeNWithUnsafe alloc n = fromArrayUnsafe <$> FL.foldlM' step initial

17511777 return

17521778 $ ArrayUnsafe contents start (INDEX_NEXT(end,a))

17531779

1780+{-# INLINE_NORMAL writeNUnsafeAs #-}

1781+writeNUnsafeAs :: forall m a. (MonadIO m, Unbox a)

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

1783+writeNUnsafeAs ps = writeNWithUnsafe (newAs ps)

1784+

17541785-- | Like 'writeN' but does not check the array bounds when writing. The fold

17551786-- driver must not call the step function more than 'n' times otherwise it will

17561787-- corrupt the memory and crash. This function exists mainly because any

17571788-- conditional in the step function blocks fusion causing 10x performance

17581789-- slowdown.

17591790--

1760−-- >>> writeNUnsafe = MutArray.writeNWithUnsafe MutArray.newPinned

1791+-- >>> writeNUnsafe = MutArray.writeNWithUnsafe MutArray.new

17611792--

17621793{-# INLINE_NORMAL writeNUnsafe #-}

17631794writeNUnsafe :: forall m a. (MonadIO m, Unbox a)

17641795 => Int -> Fold m a (MutArray a)

1765−writeNUnsafe = writeNWithUnsafe newPinned

1796+writeNUnsafe = writeNUnsafeAs Unpinned

17661797

17671798-- | @writeNWith alloc n@ folds a maximum of @n@ elements into an array

17681799-- allocated using the @alloc@ function.

@@ -1775,16 +1806,24 @@ writeNWith :: forall m a. (MonadIO m, Unbox a)

17751806 => (Int -> m (MutArray a)) -> Int -> Fold m a (MutArray a)

17761807writeNWith alloc n = FL.take n (writeNWithUnsafe alloc n)

17771808

1809+{-# INLINE_NORMAL writeNAs #-}

1810+writeNAs ::

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

1812+ => PinnedState

1813+ -> Int

1814+ -> Fold m a (MutArray a)

1815+writeNAs ps = writeNWith (newAs ps)

1816+

17781817-- | @writeN n@ folds a maximum of @n@ elements from the input stream to an

17791818-- 'MutArray'.

17801819--

1781−-- >>> writeN = MutArray.writeNWith MutArray.newPinned

1820+-- >>> writeN = MutArray.writeNWith MutArray.new

17821821-- >>> writeN n = Fold.take n (MutArray.writeNUnsafe n)

1783−-- >>> writeN n = MutArray.writeAppendN n (MutArray.newPinned n)

1822+-- >>> writeN n = MutArray.writeAppendN n (MutArray.new n)

17841823--

17851824{-# INLINE_NORMAL writeN #-}

17861825writeN :: forall m a. (MonadIO m, Unbox a) => Int -> Fold m a (MutArray a)

1787−writeN = writeNWith newPinned

1826+writeN = writeNAs Unpinned

17881827

17891828-- | Like writeNWithUnsafe but writes the array in reverse order.

17901829--

@@ -1820,20 +1859,20 @@ writeRevNWith alloc n = FL.take n (writeRevNWithUnsafe alloc n)

18201859-- /Pre-release/

18211860{-# INLINE_NORMAL writeRevN #-}

18221861writeRevN :: forall m a. (MonadIO m, Unbox a) => Int -> Fold m a (MutArray a)

1823−writeRevN = writeRevNWith newPinned

1862+writeRevN = writeRevNWith new

18241863

1825−-- | @writeNAligned align n@ folds a maximum of @n@ elements from the input

1826−-- stream to a 'MutArray' aligned to the given size.

1864+-- | @pinnedWriteNAligned align n@ folds a maximum of @n@ elements from the

1865+-- input stream to a 'MutArray' aligned to the given size.

18271866--

1828−-- >>> writeNAligned align = MutArray.writeNWith (MutArray.newAlignedPinned align)

1829−-- >>> writeNAligned align n = MutArray.writeAppendN n (MutArray.newAlignedPinned align n)

1867+-- >>> pinnedWriteNAligned align = MutArray.writeNWith (MutArray.newAlignedPinned align)

1868+-- >>> pinnedWriteNAligned align n = MutArray.writeAppendN n (MutArray.newAlignedPinned align n)

18301869--

18311870-- /Pre-release/

18321871--

1833−{-# INLINE_NORMAL writeNAligned #-}

1834−writeNAligned :: forall m a. (MonadIO m, Unbox a)

1872+{-# INLINE_NORMAL pinnedWriteNAligned #-}

1873+pinnedWriteNAligned :: forall m a. (MonadIO m, Unbox a)

18351874 => Int -> Int -> Fold m a (MutArray a)

1836−writeNAligned align = writeNWith (newAlignedPinned align)

1875+pinnedWriteNAligned align = writeNWith (newAlignedPinned align)

18371876

18381877-- XXX Buffer to a list instead?

18391878--

@@ -1876,7 +1915,7 @@ writeChunks n = FL.many (writeN n) FL.toStreamK

18761915--

18771916-- /Caution! Do not use this on infinite streams./

18781917--

1879−-- >>> f n = MutArray.writeAppendWith (* 2) (MutArray.newPinned n)

1918+-- >>> f n = MutArray.writeAppendWith (* 2) (MutArray.new n)

18801919-- >>> writeWith n = Fold.rmapM MutArray.rightSize (f n)

18811920-- >>> writeWith n = Fold.rmapM MutArray.fromArrayStreamK (MutArray.writeChunks n)

18821921--

@@ -1884,7 +1923,7 @@ writeChunks n = FL.many (writeN n) FL.toStreamK

18841923{-# INLINE_NORMAL writeWith #-}

18851924writeWith :: forall m a. (MonadIO m, Unbox a)

18861925 => Int -> Fold m a (MutArray a)

1887−-- writeWith n = FL.rmapM rightSize $ writeAppendWith (* 2) (newPinned n)

1926+-- writeWith n = FL.rmapM rightSize $ writeAppendWith (* 2) (new n)

18881927writeWith elemCount =

18891928 FL.rmapM extract $ FL.foldlM' step initial

18901929

@@ -1892,7 +1931,7 @@ writeWith elemCount =

18921931

18931932 initial = do

18941933 when (elemCount < 0) $ error "writeWith: elemCount is negative"

1895− liftIO $ newPinned elemCount

1934+ liftIO $ new elemCount

18961935

18971936 step arr@(MutArray _ start end bound) x

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

@@ -1928,7 +1967,7 @@ fromStreamDN :: forall m a. (MonadIO m, Unbox a)

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

19291968-- fromStreamDN n = D.fold (writeN n)

19301969fromStreamDN limit str = do

1931− (arr :: MutArray a) <- liftIO $ newPinned limit

1970+ (arr :: MutArray a) <- liftIO $ new limit

19321971 end <- D.foldlM' (fwrite (arrContents arr)) (return $ arrEnd arr) $ D.take limit str

19331972 return $ arr {arrEnd = end}

19341973

@@ -2040,7 +2079,7 @@ spliceCopy arr1 arr2 = liftIO $ do

20402079 start2 = arrStart arr2

20412080 len1 = arrEnd arr1 - start1

20422081 len2 = arrEnd arr2 - start2

2043− newArrContents <- liftIO $ Unboxed.newPinnedBytes (len1 + len2)

2082+ newArrContents <- liftIO $ Unboxed.newUnpinnedBytes (len1 + len2)

20442083 let len = len1 + len2

20452084 newArr = MutArray newArrContents 0 len len

20462085 putSliceUnsafe arr1 start1 newArr 0 len1

@@ -2251,12 +2290,10 @@ cast arr =

22512290--

22522291asPtrUnsafe :: MonadIO m => MutArray a -> (Ptr a -> m b) -> m b

22532292asPtrUnsafe arr f = do

2254− let contents = arrContents arr

2255− !ptr = Ptr (byteArrayContents#

2293+ contents <- liftIO $ Unboxed.pin $ arrContents arr

2294+ let !ptr = Ptr (byteArrayContents#

22562295 (unsafeCoerce# (getMutableByteArray# contents)))

2257− -- XXX Check if the array is pinned, if not, copy it to a pinned array

2258− -- XXX We should probably pass to the IO action the byte length of the array

2259− -- as well so that bounds can be checked.

2296+

22602297 r <- f (ptr `plusPtr` arrStart arr)

22612298 liftIO $ touch contents

22622299 return r

core/src/Streamly/Internal/Data/Array/Type.hs+32−6

@@ -24,6 +24,7 @@ module Streamly.Internal.Data.Array.Type

2424 -- * Pinning and Unpinning

2525 , pin

2626 , unpin

27+ , isPinned

2728

2829 -- * Construction

2930 , splice

@@ -64,14 +65,17 @@ module Streamly.Internal.Data.Array.Type

6465 -- * Folds

6566 , writeWith

6667 , writeN

68+ , writeNAs

6769 , writeNUnsafe

70+ , writeNUnsafeAs

6871 , MA.ArrayUnsafe (..)

69− , writeNAligned

72+ , pinnedWriteNAligned

7073 , write

7174 , unsafeMakePure

7275

7376 -- * Streams of arrays

7477 , chunksOf

78+ , chunksOfAs

7579 , bufferChunks

7680 , flattenArrays

7781 , flattenArraysRev

@@ -95,7 +99,7 @@ import GHC.Ptr (Ptr(..))

9599import Streamly.Internal.Data.Array.Mut.Type (MutArray(..), MutableByteArray)

96100import Streamly.Internal.Data.Fold.Type (Fold(..))

97101import Streamly.Internal.Data.Stream.StreamD.Type (Stream)

98−import Streamly.Internal.Data.Unbox (Unbox(..))

102+import Streamly.Internal.Data.Unbox (PinnedState, Unbox(..))

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

100104import Text.Read (readPrec)

101105

@@ -226,6 +230,10 @@ pin = fmap unsafeFreeze . MA.pin . unsafeThaw

226230unpin :: Array a -> IO (Array a)

227231unpin = fmap unsafeFreeze . MA.unpin . unsafeThaw

228232

233+{-# INLINE isPinned #-}

234+isPinned :: Array a -> Bool

235+isPinned = MA.isPinned . unsafeThaw

236+

229237-------------------------------------------------------------------------------

230238-- Construction

231239-------------------------------------------------------------------------------

@@ -286,6 +294,11 @@ bufferChunks :: (MonadIO m, Unbox a) =>

286294 D.Stream m a -> m (K.StreamK m (Array a))

287295bufferChunks m = D.foldr K.cons K.nil $ chunksOf defaultChunkSize m

288296

297+{-# INLINE_NORMAL chunksOfAs #-}

298+chunksOfAs :: forall m a. (MonadIO m, Unbox a)

299+ => PinnedState -> Int -> D.Stream m a -> D.Stream m (Array a)

300+chunksOfAs ps n str = D.map unsafeFreeze $ MA.chunksOfAs ps n str

301+

289302-- | @chunksOf n stream@ groups the elements in the input stream into arrays of

290303-- @n@ elements each.

291304--

@@ -461,6 +474,14 @@ toList s = build (\c n -> toListFB c n s)

461474-- Folds

462475-------------------------------------------------------------------------------

463476

477+{-# INLINE_NORMAL writeNAs #-}

478+writeNAs ::

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

480+ => PinnedState

481+ -> Int

482+ -> Fold m a (Array a)

483+writeNAs ps = fmap unsafeFreeze . MA.writeNAs ps

484+

464485-- | @writeN n@ folds a maximum of @n@ elements from the input stream to an

465486-- 'Array'.

466487--

@@ -468,15 +489,20 @@ toList s = build (\c n -> toListFB c n s)

468489writeN :: forall m a. (MonadIO m, Unbox a) => Int -> Fold m a (Array a)

469490writeN = fmap unsafeFreeze . MA.writeN

470491

471−-- | @writeNAligned alignment n@ folds a maximum of @n@ elements from the input

492+-- | @pinnedWriteNAligned alignment n@ folds a maximum of @n@ elements from the input

472493-- stream to an 'Array' aligned to the given size.

473494--

474495-- /Pre-release/

475496--

476−{-# INLINE_NORMAL writeNAligned #-}

477−writeNAligned :: forall m a. (MonadIO m, Unbox a)

497+{-# INLINE_NORMAL pinnedWriteNAligned #-}

498+pinnedWriteNAligned :: forall m a. (MonadIO m, Unbox a)

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

479−writeNAligned alignSize = fmap unsafeFreeze . MA.writeNAligned alignSize

500+pinnedWriteNAligned alignSize = fmap unsafeFreeze . MA.pinnedWriteNAligned alignSize

501+

502+{-# INLINE_NORMAL writeNUnsafeAs #-}

503+writeNUnsafeAs :: forall m a. (MonadIO m, Unbox a)

504+ => PinnedState -> Int -> Fold m a (Array a)

505+writeNUnsafeAs ps n = unsafeFreeze <$> MA.writeNUnsafeAs ps n

480506

481507-- | Like 'writeN' but does not check the array bounds when writing. The fold

482508-- driver must not call the step function more than 'n' times otherwise it will

core/src/Streamly/Internal/Data/Ring/Unboxed.hs+2−1

@@ -574,7 +574,8 @@ slidingWindowWith n (Fold step1 initial1 extract1) = Fold step initial extract

574574 Done b -> Done b

575575

576576 toArray foldRing rb rh = do

577− arr <- liftIO $ MA.newPinned n

577+ -- Using unpinned array here instead of pinned

578+ arr <- liftIO $ MA.new n

578579 let snoc' b a = liftIO $ MA.snocUnsafe b a

579580 foldRing rh snoc' arr rb

580581

core/src/Streamly/Internal/Data/Unbox.hs+12

@@ -7,11 +7,14 @@

88module Streamly.Internal.Data.Unbox

99 ( Unbox(..)

10+ , PinnedState(..)

1011 , MutableByteArray(..)

1112 , touch

1213 , getMutableByteArray#

14+ , isPinned

1315 , pin

1416 , unpin

17+ , newBytes

1518 , newUnpinnedBytes

1619 , newPinnedBytes

1720 , newAlignedPinnedBytes

@@ -72,6 +75,10 @@ import Prelude hiding (read)

7275-- The ArrayContents type

7376--------------------------------------------------------------------------------

7477

78+data PinnedState

79+ = Pinned

80+ | Unpinned

81+

7582-- XXX can use UnliftedNewtypes

7683data MutableByteArray = MutableByteArray (MutableByteArray# RealWorld)

7784

@@ -130,6 +137,11 @@ newAlignedPinnedBytes (I# nbytes) (I# align) = IO $ \s ->

130137 let c = MutableByteArray mbarr#

131138 in (# s', c #)

132139

140+{-# INLINE newBytes #-}

141+newBytes :: PinnedState -> Int -> IO MutableByteArray

142+newBytes Unpinned = newUnpinnedBytes

143+newBytes Pinned = newPinnedBytes

144+

133145-------------------------------------------------------------------------------

134146-- Pinning & Unpinning

135147-------------------------------------------------------------------------------

core/src/Streamly/Internal/FileSystem/File.hs+7−5

@@ -103,10 +103,10 @@ import qualified Control.Monad.Catch as MC

103103import qualified System.IO as SIO

104104

105105import Streamly.Data.Fold (groupsOf, drain)

106−import Streamly.Internal.Data.Array.Type (Array(..), writeNUnsafe)

106+import Streamly.Internal.Data.Array.Type (Array(..), writeNUnsafeAs)

107107import Streamly.Internal.Data.Fold.Type (Fold(..))

108108import Streamly.Data.Stream (Stream)

109−import Streamly.Internal.Data.Unbox (Unbox)

109+import Streamly.Internal.Data.Unbox (PinnedState(..), Unbox)

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

111111-- import Streamly.String (encodeUtf8, decodeUtf8, foldLines)

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

@@ -114,6 +114,7 @@ import Streamly.Internal.System.IO (defaultChunkSize)

114114import qualified Streamly.Data.Array as A

115115import qualified Streamly.Data.Stream as S

116116import qualified Streamly.Data.Unfold as UF

117+import qualified Streamly.Internal.Data.Array.Type as IA (chunksOfAs)

117118import qualified Streamly.Internal.Data.Unfold as UF (bracketIO)

118119import qualified Streamly.Internal.Data.Fold.Type as FL

119120 (Step(..), snoc, reduce)

@@ -404,7 +405,7 @@ fromChunks = fromChunksMode WriteMode

404405{-# INLINE fromBytesWith #-}

405406fromBytesWith :: (MonadIO m, MonadCatch m)

406407 => Int -> FilePath -> Stream m Word8 -> m ()

407−fromBytesWith n file xs = fromChunks file $ S.chunksOf n xs

408+fromBytesWith n file xs = fromChunks file $ IA.chunksOfAs Pinned n xs

408409

409410{-# DEPRECATED fromBytesWithBufferOf "Please use 'fromBytesWith' instead" #-}

410411{-# INLINE fromBytesWithBufferOf #-}

@@ -463,7 +464,7 @@ writeChunks path = Fold step initial extract

463464writeWith :: (MonadIO m, MonadCatch m)

464465 => Int -> FilePath -> Fold m Word8 ()

465466writeWith n path =

466− groupsOf n (writeNUnsafe n) (writeChunks path)

467+ groupsOf n (writeNUnsafeAs Pinned n) (writeChunks path)

467468

468469{-# DEPRECATED writeWithBufferOf "Please use 'writeWith' instead" #-}

469470{-# INLINE writeWithBufferOf #-}

@@ -501,7 +502,8 @@ writeAppendChunks = fromChunksMode AppendMode

501502{-# INLINE writeAppendWith #-}

502503writeAppendWith :: (MonadIO m, MonadCatch m)

503504 => Int -> FilePath -> Stream m Word8 -> m ()

504−writeAppendWith n file xs = writeAppendChunks file $ S.chunksOf n xs

505+writeAppendWith n file xs =

506+ writeAppendChunks file $ IA.chunksOfAs Pinned n xs

505507

506508-- | Append a byte stream to a file. Combines the bytes in chunks of size up to

507509-- 'A.defaultChunkSize' before writing. If the file exists then the new data

core/src/Streamly/Internal/FileSystem/Handle.hs+6−6

@@ -115,7 +115,7 @@ import Control.Monad.IO.Class (MonadIO(..))

115115import Data.Function ((&))

116116import Data.Maybe (isNothing, fromJust)

117117import Data.Word (Word8)

118−import Streamly.Internal.Data.Unbox (Unbox)

118+import Streamly.Internal.Data.Unbox (PinnedState(..), Unbox)

119119import System.IO (Handle, SeekMode(..), hGetBufSome, hPutBuf, hSeek)

120120import Prelude hiding (read)

121121

@@ -123,7 +123,7 @@ import Streamly.Internal.Data.Fold (Fold)

123123import Streamly.Internal.Data.Refold.Type (Refold(..))

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

125125import Streamly.Internal.Data.Array.Type

126− (Array(..), writeNUnsafe, unsafeFreezeWithShrink, byteLength)

126+ (Array(..), writeNUnsafeAs, unsafeFreezeWithShrink, byteLength)

127127import Streamly.Internal.Data.Stream.StreamD.Type (Stream)

128128import Streamly.Internal.Data.Stream.Chunked (lpackArraysChunksOf)

129129-- import Streamly.String (encodeUtf8, decodeUtf8, foldLines)

@@ -433,7 +433,7 @@ putChunksWith n h xs = putChunks h $ AS.compact n xs

433433--

434434{-# INLINE putBytesWith #-}

435435putBytesWith :: MonadIO m => Int -> Handle -> Stream m Word8 -> m ()

436−putBytesWith n h m = putChunks h $ A.chunksOf n m

436+putBytesWith n h m = putChunks h $ A.chunksOfAs Pinned n m

437437

438438-- | Write a byte stream to a file handle. Accumulates the input in chunks of

439439-- up to 'Streamly.Internal.Data.Array.Type.defaultChunkSize' before writing.

@@ -502,7 +502,7 @@ writeChunksWithBufferOf = writeChunksWith

502502--

503503{-# INLINE writeWith #-}

504504writeWith :: MonadIO m => Int -> Handle -> Fold m Word8 ()

505−writeWith n h = FL.groupsOf n (writeNUnsafe n) (writeChunks h)

505+writeWith n h = FL.groupsOf n (writeNUnsafeAs Pinned n) (writeChunks h)

506506

507507-- | Same as 'writeWith'

508508--

@@ -520,7 +520,7 @@ writeWithBufferOf = writeWith

520520writeMaybesWith :: (MonadIO m )

521521 => Int -> Handle -> Fold m (Maybe Word8) ()

522522writeMaybesWith n h =

523− let writeNJusts = FL.lmap fromJust $ A.writeN n

523+ let writeNJusts = FL.lmap fromJust $ A.writeNAs Pinned n

524524 writeOnNothing = FL.takeEndBy_ isNothing writeNJusts

525525 in FL.many writeOnNothing (writeChunks h)

526526

@@ -530,7 +530,7 @@ writeMaybesWith n h =

530530{-# INLINE writerWith #-}

531531writerWith :: MonadIO m => Int -> Refold m Handle Word8 ()

532532writerWith n =

533− FL.refoldMany (FL.take n $ writeNUnsafe n) chunkWriter

533+ FL.refoldMany (FL.take n $ writeNUnsafeAs Pinned n) chunkWriter

534534

535535-- | Write a byte stream to a file handle. Accumulates the input in chunks of

536536-- up to 'Streamly.Internal.Data.Array.Type.defaultChunkSize' before writing

core/src/Streamly/Unicode/Stream.hs+4−10

@@ -24,14 +24,11 @@

2424-- routines in this module and then written to IO devices or to arrays in

2525-- memory.

2626--

27−-- If you have to store a 'Char' stream in memory you can convert it into a

28−-- 'String' using 'Streamly.Data.Fold.toList' fold. The 'String' type can be

29−-- more efficient than pinned arrays for short and short lived strings.

30−--

31−-- For longer or long lived streams you can 'Streamly.Data.Stream.fold' the

27+-- If you have to store a 'Char' stream in memory you can

28+-- 'Streamly.Data.Stream.fold' the

3229-- 'Char' stream as @Array Char@ using the array 'Streamly.Data.Array.write'

33−-- fold. The 'Array' type provides a more compact representation and pinned

34−-- memory reducing GC overhead. If space efficiency is a concern you can use

30+-- fold. The 'Array' type provides a more compact representation

31+-- reducing GC overhead. If space efficiency is a concern you can use

3532-- 'encodeUtf8'' on the 'Char' stream before writing it to an 'Array' providing

3633-- an even more compact representation.

3734--

@@ -69,9 +66,6 @@

6966-- Some experimental APIs to conveniently process text using the

7067-- @Array Char@ represenation directly can be found in

7168-- "Streamly.Internal.Unicode.Array".

72−

73−-- XXX an unpinned array representation can be useful to store short and short

74−-- lived strings in memory.

7569--

7670module Streamly.Unicode.Stream

7771 (

src/Streamly/Internal/Network/Inet/TCP.hs+5−4

@@ -106,11 +106,11 @@ import Prelude hiding (read)

106106

107107import Streamly.Internal.Control.Concurrent (MonadAsync)

108108import Streamly.Internal.Control.ForkLifted (fork)

109−import Streamly.Internal.Data.Array.Type (Array(..), writeNUnsafe)

109+import Streamly.Internal.Data.Array.Type (Array(..), writeNUnsafeAs)

110110import Streamly.Internal.Data.Fold.Type (Fold(..))

111111import Streamly.Data.Stream (Stream)

112112import Streamly.Internal.Data.Tuple.Strict (Tuple'(..))

113−import Streamly.Internal.Data.Unbox (Unbox)

113+import Streamly.Internal.Data.Unbox (PinnedState(..), Unbox)

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

115115import Streamly.Internal.Network.Socket (SockSpec(..), accept, acceptor)

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

@@ -385,7 +385,8 @@ putBytesWithBufferOf

385385 -> PortNumber

386386 -> Stream m Word8

387387 -> m ()

388−putBytesWithBufferOf n addr port m = putChunks addr port $ A.chunksOf n m

388+putBytesWithBufferOf n addr port m =

389+ putChunks addr port $ A.chunksOfAs Pinned n m

389390

390391-- | Like 'write' but provides control over the write buffer. Output will

391392-- be written to the IO device as soon as we collect the specified number of

@@ -399,7 +400,7 @@ writeWithBufferOf

399400 -> PortNumber

400401 -> Fold m Word8 ()

401402writeWithBufferOf n addr port =

402− FL.groupsOf n (writeNUnsafe n) (writeChunks addr port)

403+ FL.groupsOf n (writeNUnsafeAs Pinned n) (writeChunks addr port)

403404

404405-- | Write a stream to the supplied IPv4 host address and port number.

405406--

src/Streamly/Internal/Network/Socket.hs+8−6

@@ -72,7 +72,7 @@ import Control.Monad.IO.Class (MonadIO(..))

7272import Data.Maybe (isNothing, fromJust)

7373import Data.Word (Word8)

7474import Foreign.Ptr (plusPtr, Ptr, castPtr)

75−import Streamly.Internal.Data.Unbox (Unbox)

75+import Streamly.Internal.Data.Unbox (PinnedState(..), Unbox)

7676import Network.Socket

7777 (Socket, SocketOption(..), Family(..), SockAddr(..),

7878 ProtocolNumber, withSocketsDo, SocketType(..), socket, bind,

@@ -94,10 +94,12 @@ import Streamly.Internal.Data.Unfold.Type (Unfold(..))

9494-- import Streamly.String (encodeUtf8, decodeUtf8, foldLines)

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

9696

97−import qualified Streamly.Data.Array as A (reader, length, writeN)

97+import qualified Streamly.Data.Array as A (reader, length)

9898import qualified Streamly.Data.Fold as FL

9999import qualified Streamly.Internal.Data.Array.Type as A

100− (unsafeFreeze, asPtrUnsafe, byteLength, writeNUnsafe, chunksOf)

100+ ( unsafeFreeze, asPtrUnsafe, byteLength, writeNUnsafeAs, chunksOfAs

101+ , writeNAs

102+ )

101103import qualified Streamly.Internal.Data.Array.Mut as MArray

102104 (MutArray(..), newPinnedBytes, asPtrUnsafe)

103105import qualified Streamly.Internal.Data.Stream as S

@@ -495,14 +497,14 @@ writeChunksWithBufferOf = writeChunksWith

495497--

496498{-# INLINE putBytesWith #-}

497499putBytesWith :: MonadIO m => Int -> Socket -> Stream m Word8 -> m ()

498−putBytesWith n h m = putChunks h $ A.chunksOf n m

500+putBytesWith n h m = putChunks h $ A.chunksOfAs Pinned n m

499501

500502-- | Write a byte stream to a socket. Accumulates the input in chunks of

501503-- specified number of bytes before writing.

502504--

503505{-# INLINE writeWith #-}

504506writeWith :: MonadIO m => Int -> Socket -> Fold m Word8 ()

505−writeWith n h = FL.groupsOf n (A.writeNUnsafe n) (writeChunks h)

507+writeWith n h = FL.groupsOf n (A.writeNUnsafeAs Pinned n) (writeChunks h)

506508

507509-- | Same as 'writeWith'

508510--

@@ -520,7 +522,7 @@ writeWithBufferOf = writeWith

520522writeMaybesWith :: (MonadIO m )

521523 => Int -> Socket -> Fold m (Maybe Word8) ()

522524writeMaybesWith n h =

523− let writeNJusts = FL.lmap fromJust $ A.writeN n

525+ let writeNJusts = FL.lmap fromJust $ A.writeNAs Pinned n

524526 writeOnNothing = FL.takeEndBy_ isNothing writeNJusts

525527 in FL.many writeOnNothing (writeChunks h)

526528