d29030f · streamly · git.iko.soy

git.iko.soy/streamly

on master help sign in

Use Pinned arrays only for IO and Unpinned arrays everywhere else

d29030fAdithya Kumar3 years ago

core/docs/Changelog.md+2

compare with master

unifiedsplit

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

3## Unreleased

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

5## 0.1.0 (March 2023)

7## 0.1.0 (March 2023)

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

9Also 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)

116--

117-- == Pinned and Unpinned Arrays

118--

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

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

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

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

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

121+-- operations.

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.

125--

122--

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

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

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

124-- 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

32 -- Monadic Folds

33 , A.writeN -- drop new

34− , A.writeNAligned

34+ , A.pinnedWriteNAligned

35 , A.write -- full buffer

36 , writeLastN

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

297 in fmap f $ liftIO $ RB.new n

298

299 done (Tuple3Fused' rb rh i) = do

300− arr <- liftIO $ MA.newPinned n

300+ arr <- liftIO $ MA.new n

301 foldFunc i rh snoc' arr rb

302

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

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

534 then Nothing

535 else Just $ castUnsafe arr

536

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.

538--

540--

539-- /Unsafe/

541-- /Unsafe/

540--

542--

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

544--

546--

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

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

546asCStringUnsafe arr act = do

548asCStringUnsafe arr act = do

547− -- XXX Ensure a pinned allocation here.

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

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

550+ -- asPtrUnsafe makes sure the array is pinned

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

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

550

552

551-------------------------------------------------------------------------------

553-------------------------------------------------------------------------------

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

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

35 , touch

36 , pin

37 , unpin

38+ , isPinned

39 -- * Constructing and Writing

40 -- * Constructing and Writing

40 -- ** Construction

41 -- ** Construction

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

48 , newArrayWith

49 , newArrayWith

50 -- *** Initialized Arrays

51 -- *** Initialized Arrays

51− , withNewArrayUnsafe

52+ , withNewArrayUnsafePinned

53 -- *** From streams

54 -- *** From streams

54 , ArrayUnsafe (..)

55 , ArrayUnsafe (..)

55 , writeNWithUnsafe

56 , writeNWithUnsafe

56 , writeNWith

57 , writeNWith

57 , writeNUnsafe

58 , writeNUnsafe

59+ , writeNUnsafeAs

58 , writeN

60 , writeN

59− , writeNAligned

61+ , writeNAs

62+ , pinnedWriteNAligned

61 , writeWith

64 , writeWith

62 , write

65 , write

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

185

188

186 -- ** Construct from streams

189 -- ** Construct from streams

187 , chunksOf

190 , chunksOf

191+ , chunksOfAs

188 , arrayStreamKFromStreamD

192 , arrayStreamKFromStreamD

189 , writeChunks

193 , writeChunks

190

194

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

236import Streamly.Internal.Data.Unbox

240import Streamly.Internal.Data.Unbox

237 ( MutableByteArray(..)

241 ( MutableByteArray(..)

238 , Unbox(..)

242 , Unbox(..)

243+ , PinnedState(..)

239 , getMutableByteArray#

244 , getMutableByteArray#

240 , touch

245 , touch

241 )

246 )

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

366 contents <- Unboxed.unpin arrContents

371 contents <- Unboxed.unpin arrContents

367 return $ arr {arrContents = contents}

372 return $ arr {arrContents = contents}

368

373

374+{-# INLINE isPinned #-}

375+isPinned :: MutArray a -> Bool

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

377+

369-------------------------------------------------------------------------------

378-------------------------------------------------------------------------------

370-- Construction

379-- Construction

371-------------------------------------------------------------------------------

380-------------------------------------------------------------------------------

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

412 MutArray a

421 MutArray a

413nil = MutArray Unboxed.nil 0 0 0

422nil = MutArray Unboxed.nil 0 0 0

414

423

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+ }

415

438

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

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

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

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

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

424 Unbox a =>

447 Unbox a =>

425#endif

448#endif

426 Int -> m (MutArray a)

449 Int -> m (MutArray a)

427−newPinnedBytes bytes = do

450+newPinnedBytes = newBytesAs Pinned

428− contents <- liftIO $ Unboxed.newPinnedBytes bytes

429− return $ MutArray

430− { arrContents = contents

431− , arrStart = 0

432− , arrEnd = 0

433− , arrBound = bytes

434− }

435

451

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

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

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

453-- 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)

442newAlignedPinned =

458newAlignedPinned =

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

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

444

460

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+

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

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

446--

469--

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

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

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

450--

473--

451{-# INLINE newPinned #-}

474{-# INLINE newPinned #-}

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

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

453−newPinned =

476+newPinned = newAs Pinned

454− newArrayWith

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

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

457

477

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

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

459-- of the array is uninitialized.

479-- of the array is uninitialized.

460--

480--

461{-# INLINE new #-}

481{-# INLINE new #-}

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

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

463−new =

483+new = newAs Unpinned

464− newArrayWith

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

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

467

484

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

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

469-- can be used.

486-- can be used.

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

472-- the array start pointer.

489-- the array start pointer.

473--

490--

474-- /Internal/

491-- /Internal/

475−{-# INLINE withNewArrayUnsafe #-}

492+{-# INLINE withNewArrayUnsafePinned #-}

476−withNewArrayUnsafe ::

493+withNewArrayUnsafePinned ::

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

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

478−withNewArrayUnsafe count f = do

495+withNewArrayUnsafePinned count f = do

479 arr <- newPinned count

496 arr <- newPinned count

480 asPtrUnsafe arr

497 asPtrUnsafe arr

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

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

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

736

753

737 -- Allocate new array

754 -- Allocate new array

738 let newCapMaxInBytes = roundUpLargeArray newCapacityInBytes

755 let newCapMaxInBytes = roundUpLargeArray newCapacityInBytes

756+ contents <-

757+ if Unboxed.isPinned arrContents

739− contents <- Unboxed.newPinnedBytes newCapMaxInBytes

758+ then Unboxed.newPinnedBytes newCapMaxInBytes

759+ else Unboxed.newUnpinnedBytes newCapMaxInBytes

740 let !(MutableByteArray mbarrFrom#) = arrContents

760 let !(MutableByteArray mbarrFrom#) = arrContents

741 !(MutableByteArray mbarrTo#) = contents

761 !(MutableByteArray mbarrTo#) = contents

742

762

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

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

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

1300 | GroupFinish

1320 | GroupFinish

1301

1321

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 #-}

1322+{-# INLINE_NORMAL chunksOfAs #-}

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

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

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

1324+ => PinnedState -> 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) =

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

1316 D.Stream step' (GroupStart state)

1326 D.Stream step' (GroupStart state)

1317

1327

1318 where

1328 where

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

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

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

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

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

1326 ++ "] must be a natural number"

1336 ++ "] 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

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

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

1329

1339

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

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

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

1350

1360

1351 step' _ GroupFinish = return D.Stop

1361 step' _ GroupFinish = return D.Stop

1352

1362

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

1353-- XXX buffer to a list instead?

1379-- XXX buffer to a list instead?

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

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

1355{-# INLINE arrayStreamKFromStreamD #-}

1381{-# INLINE arrayStreamKFromStreamD #-}

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

1751 return

1777 return

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

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

1753

1779

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+

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

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

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

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

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

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

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

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

1758-- slowdown.

1789-- slowdown.

1759--

1790--

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

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

1761--

1792--

1762{-# INLINE_NORMAL writeNUnsafe #-}

1793{-# INLINE_NORMAL writeNUnsafe #-}

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

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

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

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

1765−writeNUnsafe = writeNWithUnsafe newPinned

1796+writeNUnsafe = writeNUnsafeAs Unpinned

1766

1797

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

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

1768-- allocated using the @alloc@ function.

1799-- allocated using the @alloc@ function.

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

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

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

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

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

1777

1808

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+

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

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

1779-- 'MutArray'.

1818-- 'MutArray'.

1780--

1819--

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

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

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

1821-- >>> 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)

1784--

1823--

1785{-# INLINE_NORMAL writeN #-}

1824{-# INLINE_NORMAL writeN #-}

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

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

1787−writeN = writeNWith newPinned

1826+writeN = writeNAs Unpinned

1788

1827

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

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

1790--

1829--

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

1820-- /Pre-release/

1859-- /Pre-release/

1821{-# INLINE_NORMAL writeRevN #-}

1860{-# INLINE_NORMAL writeRevN #-}

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

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

1823−writeRevN = writeRevNWith newPinned

1862+writeRevN = writeRevNWith new

1824

1863

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

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

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

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

1827--

1866--

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

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

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

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

1830--

1869--

1831-- /Pre-release/

1870-- /Pre-release/

1832--

1871--

1833−{-# INLINE_NORMAL writeNAligned #-}

1872+{-# INLINE_NORMAL pinnedWriteNAligned #-}

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

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

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

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

1836−writeNAligned align = writeNWith (newAlignedPinned align)

1875+pinnedWriteNAligned align = writeNWith (newAlignedPinned align)

1837

1876

1838-- XXX Buffer to a list instead?

1877-- XXX Buffer to a list instead?

1839--

1878--

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

1876--

1915--

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

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

1878--

1917--

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

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

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

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

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

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

1882--

1921--

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

1884{-# INLINE_NORMAL writeWith #-}

1923{-# INLINE_NORMAL writeWith #-}

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

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

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

1925 => 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)

1888writeWith elemCount =

1927writeWith elemCount =

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

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

1890

1929

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

1892

1931

1893 initial = do

1932 initial = do

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

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

1895− liftIO $ newPinned elemCount

1934+ liftIO $ new elemCount

1896

1935

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

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

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

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

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

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

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

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

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

1930fromStreamDN limit str = do

1969fromStreamDN limit str = do

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

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

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

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

1933 return $ arr {arrEnd = end}

1972 return $ arr {arrEnd = end}

1934

1973

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

2040 start2 = arrStart arr2

2079 start2 = arrStart arr2

2041 len1 = arrEnd arr1 - start1

2080 len1 = arrEnd arr1 - start1

2042 len2 = arrEnd arr2 - start2

2081 len2 = arrEnd arr2 - start2

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

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

2044 let len = len1 + len2

2083 let len = len1 + len2

2045 newArr = MutArray newArrContents 0 len len

2084 newArr = MutArray newArrContents 0 len len

2046 putSliceUnsafe arr1 start1 newArr 0 len1

2085 putSliceUnsafe arr1 start1 newArr 0 len1

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

2251--

2290--

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

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

2253asPtrUnsafe arr f = do

2292asPtrUnsafe arr f = do

2254− let contents = arrContents arr

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

2255− !ptr = Ptr (byteArrayContents#

2294+ let !ptr = Ptr (byteArrayContents#

2256 (unsafeCoerce# (getMutableByteArray# contents)))

2295 (unsafeCoerce# (getMutableByteArray# contents)))

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

2296+

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.

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

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

2261 liftIO $ touch contents

2298 liftIO $ touch contents

2262 return r

2299 return r

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

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

24 -- * Pinning and Unpinning

25 , pin

26 , unpin

27+ , isPinned

28 -- * Construction

29 -- * Construction

29 , splice

30 , splice

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

64 -- * Folds

65 -- * Folds

65 , writeWith

66 , writeWith

66 , writeN

67 , writeN

68+ , writeNAs

67 , writeNUnsafe

69 , writeNUnsafe

70+ , writeNUnsafeAs

68 , MA.ArrayUnsafe (..)

71 , MA.ArrayUnsafe (..)

69− , writeNAligned

72+ , pinnedWriteNAligned

70 , write

73 , write

71 , unsafeMakePure

74 , unsafeMakePure

73 -- * Streams of arrays

76 -- * Streams of arrays

74 , chunksOf

77 , chunksOf

78+ , chunksOfAs

75 , bufferChunks

79 , bufferChunks

76 , flattenArrays

80 , flattenArrays

77 , flattenArraysRev

81 , flattenArraysRev

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

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

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

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

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

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

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

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

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

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

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

100import Text.Read (readPrec)

104import Text.Read (readPrec)

101

105

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

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

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

227unpin = fmap unsafeFreeze . MA.unpin . unsafeThaw

231unpin = fmap unsafeFreeze . MA.unpin . unsafeThaw

228

232

233+{-# INLINE isPinned #-}

234+isPinned :: Array a -> Bool

235+isPinned = MA.isPinned . unsafeThaw

236+

229-------------------------------------------------------------------------------

237-------------------------------------------------------------------------------

230-- Construction

238-- Construction

231-------------------------------------------------------------------------------

239-------------------------------------------------------------------------------

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

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

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

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

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

288

296

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+

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

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

290-- @n@ elements each.

303-- @n@ elements each.

291--

304--

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

461-- Folds

474-- Folds

462-------------------------------------------------------------------------------

475-------------------------------------------------------------------------------

463

476

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+

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

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

465-- 'Array'.

486-- 'Array'.

466--

487--

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

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

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

469writeN = fmap unsafeFreeze . MA.writeN

490writeN = fmap unsafeFreeze . MA.writeN

470

491

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

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

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

473--

494--

474-- /Pre-release/

495-- /Pre-release/

475--

496--

476−{-# INLINE_NORMAL writeNAligned #-}

497+{-# INLINE_NORMAL pinnedWriteNAligned #-}

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

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

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

499 => 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

480

506

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

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

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

508-- 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

574 Done b -> Done b

575

576 toArray foldRing rb rh = do

577+ -- Using unpinned array here instead of pinned

577− arr <- liftIO $ MA.newPinned n

578+ arr <- liftIO $ MA.new n

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

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

579 foldRing rh snoc' arr rb

580 foldRing rh snoc' arr rb

580

581

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

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

8module Streamly.Internal.Data.Unbox

9 ( Unbox(..)

10+ , PinnedState(..)

10 , MutableByteArray(..)

11 , MutableByteArray(..)

11 , touch

12 , touch

12 , getMutableByteArray#

13 , getMutableByteArray#

14+ , isPinned

13 , pin

15 , pin

14 , unpin

16 , unpin

17+ , newBytes

15 , newUnpinnedBytes

18 , newUnpinnedBytes

16 , newPinnedBytes

19 , newPinnedBytes

17 , newAlignedPinnedBytes

20 , newAlignedPinnedBytes

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

72-- The ArrayContents type

75-- The ArrayContents type

73--------------------------------------------------------------------------------

76--------------------------------------------------------------------------------

78+data PinnedState

79+ = Pinned

80+ | Unpinned

81+

75-- XXX can use UnliftedNewtypes

82-- XXX can use UnliftedNewtypes

76data MutableByteArray = MutableByteArray (MutableByteArray# RealWorld)

83data MutableByteArray = MutableByteArray (MutableByteArray# RealWorld)

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

130 let c = MutableByteArray mbarr#

137 let c = MutableByteArray mbarr#

131 in (# s', c #)

138 in (# s', c #)

132

139

140+{-# INLINE newBytes #-}

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

142+newBytes Unpinned = newUnpinnedBytes

143+newBytes Pinned = newPinnedBytes

144+

133-------------------------------------------------------------------------------

145-------------------------------------------------------------------------------

134-- Pinning & Unpinning

146-- Pinning & Unpinning

135-------------------------------------------------------------------------------

147-------------------------------------------------------------------------------

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

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

103import qualified System.IO as SIO

104

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

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

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

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

108import Streamly.Data.Stream (Stream)

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

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

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

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

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

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

114import qualified Streamly.Data.Array as A

115import qualified Streamly.Data.Stream as S

116import qualified Streamly.Data.Unfold as UF

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

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

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

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

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

119 (Step(..), snoc, reduce)

120 (Step(..), snoc, reduce)

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

404{-# INLINE fromBytesWith #-}

405{-# INLINE fromBytesWith #-}

405fromBytesWith :: (MonadIO m, MonadCatch m)

406fromBytesWith :: (MonadIO m, MonadCatch m)

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

407 => 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

408

409

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

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

410{-# INLINE fromBytesWithBufferOf #-}

411{-# INLINE fromBytesWithBufferOf #-}

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

463writeWith :: (MonadIO m, MonadCatch m)

464writeWith :: (MonadIO m, MonadCatch m)

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

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

465writeWith n path =

466writeWith n path =

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

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

467

468

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

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

469{-# INLINE writeWithBufferOf #-}

470{-# INLINE writeWithBufferOf #-}

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

501{-# INLINE writeAppendWith #-}

502{-# INLINE writeAppendWith #-}

502writeAppendWith :: (MonadIO m, MonadCatch m)

503writeAppendWith :: (MonadIO m, MonadCatch m)

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

504 => 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

505

507

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

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

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

509-- '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(..))

115import Data.Function ((&))

116import Data.Maybe (isNothing, fromJust)

117import Data.Word (Word8)

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

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

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

120import Prelude hiding (read)

121

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

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

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

125import Streamly.Internal.Data.Array.Type

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

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

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

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

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

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

433--

434{-# INLINE putBytesWith #-}

435putBytesWith :: 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

437

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

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

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

502--

503{-# INLINE writeWith #-}

504writeWith :: 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)

506

507-- | Same as 'writeWith'

508--

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

520writeMaybesWith :: (MonadIO m )

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

522writeMaybesWith n h =

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

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

524 writeOnNothing = FL.takeEndBy_ isNothing writeNJusts

525 in FL.many writeOnNothing (writeChunks h)

526

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

530{-# INLINE writerWith #-}

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

532writerWith n =

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

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

534

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

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

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

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

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

25-- memory.

26--

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

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

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

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

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

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

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

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

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

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

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

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

36-- an even more compact representation.

33-- an even more compact representation.

37--

34--

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

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

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

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

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

71-- "Streamly.Internal.Unicode.Array".

68-- "Streamly.Internal.Unicode.Array".

72−

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

74−-- lived strings in memory.

75--

69--

76module Streamly.Unicode.Stream

70module Streamly.Unicode.Stream

77 (

71 (

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

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

106

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

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

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

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

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

111import Streamly.Data.Stream (Stream)

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

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

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

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

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

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

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

385 -> PortNumber

386 -> Stream m Word8

387 -> 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

389

390

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

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

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

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

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

399 -> PortNumber

400 -> PortNumber

400 -> Fold m Word8 ()

401 -> Fold m Word8 ()

401writeWithBufferOf n addr port =

402writeWithBufferOf n addr port =

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

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

403

404

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

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

405--

406--

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

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

72import Data.Maybe (isNothing, fromJust)

73import Data.Word (Word8)

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

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

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

76import Network.Socket

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

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

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

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

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

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

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

98import qualified Streamly.Data.Fold as FL

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

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

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

101+ , writeNAs

102+ )

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

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

102 (MutArray(..), newPinnedBytes, asPtrUnsafe)

104 (MutArray(..), newPinnedBytes, asPtrUnsafe)

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

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

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

495--

497--

496{-# INLINE putBytesWith #-}

498{-# INLINE putBytesWith #-}

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

499putBytesWith :: 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

499

501

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

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

501-- specified number of bytes before writing.

503-- specified number of bytes before writing.

502--

504--

503{-# INLINE writeWith #-}

505{-# INLINE writeWith #-}

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

506writeWith :: 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)

506

508

507-- | Same as 'writeWith'

509-- | Same as 'writeWith'

508--

510--

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

520writeMaybesWith :: (MonadIO m )

522writeMaybesWith :: (MonadIO m )

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

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

522writeMaybesWith n h =

524writeMaybesWith n h =

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

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

524 writeOnNothing = FL.takeEndBy_ isNothing writeNJusts

526 writeOnNothing = FL.takeEndBy_ isNothing writeNJusts

525 in FL.many writeOnNothing (writeChunks h)

527 in FL.many writeOnNothing (writeChunks h)

526

528