df5d151 · streamly · git.iko.soy

git.iko.soy/streamly

Non-Functional renaming changes in the array APIs

- Add pinned version of chunksOf - Add documentation to pinned combinators - Use the combinators with "pinned*" prefix over "*As" suffix - Reduce the use of PinnedState

commitdf5d1518d3a2bdcc38c140d93616b8479c9f0284 authorAdithya Kumar <adithya@composewell.com> dateWed 26 Jul 2023, 18:34 (3 years ago) parent6780145

10 files changed+90−85

compare with master

unifiedsplit

core/src/Streamly/Internal/Data/Array/Mut/Type.hs+27−17 core/src/Streamly/Internal/Data/Array/Type.hs+19−24 core/src/Streamly/Internal/Data/IORef/Unboxed.hs+2−2 core/src/Streamly/Internal/Data/Unbox.hs+13−13 core/src/Streamly/Internal/FileSystem/Dir.hs+2−2 core/src/Streamly/Internal/FileSystem/File.hs+6−6 core/src/Streamly/Internal/FileSystem/Handle.hs+7−7 src/Streamly/Internal/Network/Inet/TCP.hs+4−4 src/Streamly/Internal/Network/Socket.hs+6−6 test/Streamly/Test/Data/Unbox.hs+4−4

core/src/Streamly/Internal/Data/Array/Mut/Type.hs+27−17

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

5353 , writeNWithUnsafe

5454 , writeNWith

5555 , writeNUnsafe

56− , writeNUnsafeAs

56+ , pinnedWriteNUnsafe

5757 , writeN

5858 , pinnedWriteN

59− , writeNAs

6059 , pinnedWriteNAligned

6160

6261 , writeWith

@@ -189,8 +188,7 @@ module Streamly.Internal.Data.Array.Mut.Type

189188

190189 -- ** Construct from streams

191190 , chunksOf

192− , chunksOfAs

193− , arrayStreamKFromStreamD

191+ , pinnedChunksOf

194192 , writeChunks

195193

196194 -- ** Eliminate to streams

@@ -429,7 +427,7 @@ newBytesAs :: MonadIO m =>

429427#endif

430428 PinnedState -> Int -> m (MutArray a)

431429newBytesAs ps bytes = do

432− contents <- liftIO $ Unboxed.newBytes ps bytes

430+ contents <- liftIO $ Unboxed.newBytesAs ps bytes

433431 return $ MutArray

434432 { arrContents = contents

435433 , arrStart = 0

@@ -463,7 +461,7 @@ pinnedNewAligned =

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

464462newAs ps =

465463 newArrayWith

466− (\s _ -> liftIO $ Unboxed.newBytes ps s)

464+ (\s _ -> liftIO $ Unboxed.newBytesAs ps s)

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

468466

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

@@ -742,7 +740,7 @@ reallocExplicit elemSize newCapacityInBytes MutArray{..} = do

742740 contents <-

743741 if Unboxed.isPinned arrContents

744742 then Unboxed.pinnedNewBytes newCapMaxInBytes

745− else Unboxed.newUnpinnedBytes newCapMaxInBytes

743+ else Unboxed.newBytes newCapMaxInBytes

746744 let !(MutableByteArray mbarrFrom#) = arrContents

747745 !(MutableByteArray mbarrTo#) = contents

748746

@@ -1362,6 +1360,16 @@ chunksOf :: forall m a. (MonadIO m, Unbox a)

13621360-- chunksOf n = D.foldMany (writeN n)

13631361chunksOf = chunksOfAs Unpinned

13641362

1363+-- | Like 'chunksOf' but creates pinned arrays.

1364+{-# INLINE_NORMAL pinnedChunksOf #-}

1365+pinnedChunksOf :: forall m a. (MonadIO m, Unbox a)

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

1367+-- pinnedChunksOf n = D.foldMany (pinnedWriteN n)

1368+pinnedChunksOf = chunksOfAs Pinned

1369+

1370+-- XXX This should take a PinnedState

1371+-- XXX buffer to a list instead?

1372+-- | Buffer the stream into arrays in memory.

13651373{-# INLINE arrayStreamKFromStreamDAs #-}

13661374arrayStreamKFromStreamDAs :: forall m a. (MonadIO m, Unbox a) =>

13671375 PinnedState -> D.Stream m a -> m (StreamK m (MutArray a))

@@ -1369,14 +1377,6 @@ arrayStreamKFromStreamDAs ps =

13691377 let n = allocBytesToElemCount (undefined :: a) defaultChunkSize

13701378 in D.foldr K.cons K.nil . chunksOfAs ps n

13711379

1372−-- XXX This should take a PinnedState

1373−-- XXX buffer to a list instead?

1374−-- | Buffer the stream into arrays in memory.

1375−{-# INLINE arrayStreamKFromStreamD #-}

1376−arrayStreamKFromStreamD :: forall m a. (MonadIO m, Unbox a) =>

1377− D.Stream m a -> m (StreamK m (MutArray a))

1378−arrayStreamKFromStreamD = arrayStreamKFromStreamDAs Pinned

1379−

13801380-------------------------------------------------------------------------------

13811381-- Streams of arrays - Flattening

13821382-------------------------------------------------------------------------------

@@ -1787,6 +1787,12 @@ writeNUnsafe :: forall m a. (MonadIO m, Unbox a)

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

17881788writeNUnsafe = writeNUnsafeAs Unpinned

17891789

1790+-- | Like 'writeNUnsafe' but creates a pinned array.

1791+{-# INLINE_NORMAL pinnedWriteNUnsafe #-}

1792+pinnedWriteNUnsafe :: forall m a. (MonadIO m, Unbox a)

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

1794+pinnedWriteNUnsafe = writeNUnsafeAs Pinned

1795+

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

17911797-- allocated using the @alloc@ function.

17921798--

@@ -1817,6 +1823,7 @@ writeNAs ps = writeNWith (newAs ps)

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

18181824writeN = writeNAs Unpinned

18191825

1826+-- | Like 'writeN' but creates a pinned array.

18201827{-# INLINE_NORMAL pinnedWriteN #-}

18211828pinnedWriteN ::

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

@@ -1959,6 +1966,7 @@ writeWith = writeWithAs Unpinned

19591966write :: forall m a. (MonadIO m, Unbox a) => Fold m a (MutArray a)

19601967write = writeWith (allocBytesToElemCount (undefined :: a) arrayChunkBytes)

19611968

1969+-- | Like 'write' but creates a pinned array.

19621970{-# INLINE pinnedWrite #-}

19631971pinnedWrite :: forall m a. (MonadIO m, Unbox a) => Fold m a (MutArray a)

19641972pinnedWrite =

@@ -2000,6 +2008,7 @@ fromStreamDN = fromStreamDNAs Unpinned

20002008fromListN :: (MonadIO m, Unbox a) => Int -> [a] -> m (MutArray a)

20012009fromListN n xs = fromStreamDN n $ D.fromList xs

20022010

2011+-- | Like 'fromListN' but creates a pinned array.

20032012{-# INLINABLE pinnedFromListN #-}

20042013pinnedFromListN :: (MonadIO m, Unbox a) => Int -> [a] -> m (MutArray a)

20052014pinnedFromListN n xs = fromStreamDNAs Pinned n $ D.fromList xs

@@ -2063,6 +2072,7 @@ fromStreamD = fromStreamDAs Unpinned

20632072fromList :: (MonadIO m, Unbox a) => [a] -> m (MutArray a)

20642073fromList xs = fromStreamD $ D.fromList xs

20652074

2075+-- | Like 'fromList' but creates a pinned array.

20662076{-# INLINE pinnedFromList #-}

20672077pinnedFromList :: (MonadIO m, Unbox a) => [a] -> m (MutArray a)

20682078pinnedFromList xs = fromStreamDAs Pinned $ D.fromList xs

@@ -2111,8 +2121,8 @@ spliceCopy arr1 arr2 = liftIO $ do

21112121 let newLen = len1 + len2

21122122 newArrContents <-

21132123 if Unboxed.isPinned (arrContents arr1)

2114− then Unboxed.newPinnedBytes newLen

2115− else Unboxed.newUnpinnedBytes newLen

2124+ then Unboxed.pinnedNewBytes newLen

2125+ else Unboxed.newBytes newLen

21162126 let len = len1 + len2

21172127 newArr = MutArray newArrContents 0 len len

21182128 putSliceUnsafe arr1 start1 newArr 0 len1

core/src/Streamly/Internal/Data/Array/Type.hs+19−24

@@ -68,9 +68,8 @@ module Streamly.Internal.Data.Array.Type

6868 , writeWith

6969 , writeN

7070 , pinnedWriteN

71− , writeNAs

7271 , writeNUnsafe

73− , writeNUnsafeAs

72+ , pinnedWriteNUnsafe

7473 , MA.ArrayUnsafe (..)

7574 , pinnedWriteNAligned

7675 , write

@@ -79,7 +78,7 @@ module Streamly.Internal.Data.Array.Type

7978

8079 -- * Streams of arrays

8180 , chunksOf

82− , chunksOfAs

81+ , pinnedChunksOf

8382 , bufferChunks

8483 , flattenArrays

8584 , flattenArraysRev

@@ -103,7 +102,7 @@ import GHC.Ptr (Ptr(..))

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

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

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

106−import Streamly.Internal.Data.Unbox (PinnedState, Unbox(..))

105+import Streamly.Internal.Data.Unbox (Unbox(..))

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

108107import Text.Read (readPrec)

109108

@@ -256,7 +255,7 @@ splice arr1 arr2 =

256255fromListN :: Unbox a => Int -> [a] -> Array a

257256fromListN n xs = unsafePerformIO $ unsafeFreeze <$> MA.fromListN n xs

258257

259−-- | Like 'fromListN' but creates a pinned 'Array'

258+-- | Like 'fromListN' but creates a pinned array.

260259{-# INLINABLE pinnedFromListN #-}

261260pinnedFromListN :: Unbox a => Int -> [a] -> Array a

262261pinnedFromListN n xs =

@@ -277,6 +276,7 @@ fromListRevN n xs = unsafePerformIO $ unsafeFreeze <$> MA.fromListRevN n xs

277276fromList :: Unbox a => [a] -> Array a

278277fromList xs = unsafePerformIO $ unsafeFreeze <$> MA.fromList xs

279278

279+-- | Like 'fromList' but creates a pinned array.

280280{-# INLINE pinnedFromList #-}

281281pinnedFromList :: Unbox a => [a] -> Array a

282282pinnedFromList xs = unsafePerformIO $ unsafeFreeze <$> MA.pinnedFromList xs

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

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

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

310310

311−{-# INLINE_NORMAL chunksOfAs #-}

312−chunksOfAs :: forall m a. (MonadIO m, Unbox a)

313− => PinnedState -> Int -> D.Stream m a -> D.Stream m (Array a)

314−chunksOfAs ps n str = D.map unsafeFreeze $ MA.chunksOfAs ps n str

315−

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

317312-- @n@ elements each.

318313--

@@ -326,6 +321,12 @@ chunksOf :: forall m a. (MonadIO m, Unbox a)

326321 => Int -> D.Stream m a -> D.Stream m (Array a)

327322chunksOf n str = D.map unsafeFreeze $ MA.chunksOf n str

328323

324+-- | Like 'chunksOf' but creates pinned arrays.

325+{-# INLINE_NORMAL pinnedChunksOf #-}

326+pinnedChunksOf :: forall m a. (MonadIO m, Unbox a)

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

328+pinnedChunksOf n str = D.map unsafeFreeze $ MA.pinnedChunksOf n str

329+

329330-- | Use the "read" unfold instead.

330331--

331332-- @flattenArrays = unfoldMany read@

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

488489-- Folds

489490-------------------------------------------------------------------------------

490491

491−{-# INLINE_NORMAL writeNAs #-}

492−writeNAs ::

493− forall m a. (MonadIO m, Unbox a)

494− => PinnedState

495− -> Int

496− -> Fold m a (Array a)

497−writeNAs ps = fmap unsafeFreeze . MA.writeNAs ps

498−

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

500493-- 'Array'.

501494--

@@ -503,6 +496,7 @@ writeNAs ps = fmap unsafeFreeze . MA.writeNAs ps

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

504497writeN = fmap unsafeFreeze . MA.writeN

505498

499+-- | Like 'fromListN' but creates a pinned array.

506500{-# INLINE_NORMAL pinnedWriteN #-}

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

508502pinnedWriteN = fmap unsafeFreeze . MA.pinnedWriteN

@@ -517,11 +511,6 @@ pinnedWriteNAligned :: forall m a. (MonadIO m, Unbox a)

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

518512pinnedWriteNAligned alignSize = fmap unsafeFreeze . MA.pinnedWriteNAligned alignSize

519513

520−{-# INLINE_NORMAL writeNUnsafeAs #-}

521−writeNUnsafeAs :: forall m a. (MonadIO m, Unbox a)

522− => PinnedState -> Int -> Fold m a (Array a)

523−writeNUnsafeAs ps n = unsafeFreeze <$> MA.writeNUnsafeAs ps n

524−

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

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

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

@@ -533,6 +522,11 @@ writeNUnsafe :: forall m a. (MonadIO m, Unbox a)

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

534523writeNUnsafe n = unsafeFreeze <$> MA.writeNUnsafe n

535524

525+{-# INLINE_NORMAL pinnedWriteNUnsafe #-}

526+pinnedWriteNUnsafe :: forall m a. (MonadIO m, Unbox a)

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

528+pinnedWriteNUnsafe n = unsafeFreeze <$> MA.pinnedWriteNUnsafe n

529+

536530{-# INLINE_NORMAL writeWith #-}

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

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

@@ -547,9 +541,10 @@ writeWith elemCount = unsafeFreeze <$> MA.writeWith elemCount

547541write :: forall m a. (MonadIO m, Unbox a) => Fold m a (Array a)

548542write = fmap unsafeFreeze MA.write

549543

544+-- | Like 'write' but creates a pinned array.

550545{-# INLINE pinnedWrite #-}

551546pinnedWrite :: forall m a. (MonadIO m, Unbox a) => Fold m a (Array a)

552−pinnedWrite = fmap unsafeFreeze MA.write

547+pinnedWrite = fmap unsafeFreeze MA.pinnedWrite

553548

554549-- | Fold "step" has a dependency on "initial", and each step is dependent on

555550-- the previous invocation of step due to state passing, finally extract

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

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

4343import Streamly.Internal.Data.Unbox

4444 ( MutableByteArray(..)

4545 , Unbox(..)

46− , newUnpinnedBytes

46+ , newBytes

4747 , sizeOf

4848 )

4949

@@ -58,7 +58,7 @@ newtype IORef a = IORef MutableByteArray

5858{-# INLINE newIORef #-}

5959newIORef :: forall a. Unbox a => a -> IO (IORef a)

6060newIORef x = do

61− var <- newUnpinnedBytes (sizeOf (Proxy :: Proxy a))

61+ var <- newBytes (sizeOf (Proxy :: Proxy a))

6262 pokeByteIndex 0 var x

6363 return $ IORef var

6464

core/src/Streamly/Internal/Data/Unbox.hs+13−13

@@ -14,8 +14,8 @@ module Streamly.Internal.Data.Unbox

1414 , isPinned

1515 , pin

1616 , unpin

17+ , newBytesAs

1718 , newBytes

18− , newUnpinnedBytes

1919 , pinnedNewBytes

2020 , pinnedNewAlignedBytes

2121 , nil

@@ -108,13 +108,13 @@ sizeOfMutableByteArray (MutableByteArray arr) =

108108

109109{-# NOINLINE nil #-}

110110nil :: MutableByteArray

111−nil = unsafePerformIO $ newUnpinnedBytes 0

111+nil = unsafePerformIO $ newBytes 0

112112

113−{-# INLINE newUnpinnedBytes #-}

114−newUnpinnedBytes :: Int -> IO MutableByteArray

115−newUnpinnedBytes nbytes | nbytes < 0 =

116− errorWithoutStackTrace "newUnpinnedBytes: size must be >= 0"

117−newUnpinnedBytes (I# nbytes) = IO $ \s ->

113+{-# INLINE newBytes #-}

114+newBytes :: Int -> IO MutableByteArray

115+newBytes nbytes | nbytes < 0 =

116+ errorWithoutStackTrace "newBytes: size must be >= 0"

117+newBytes (I# nbytes) = IO $ \s ->

118118 case newByteArray# nbytes s of

119119 (# s', mbarr# #) ->

120120 let c = MutableByteArray mbarr#

@@ -140,10 +140,10 @@ pinnedNewAlignedBytes (I# nbytes) (I# align) = IO $ \s ->

140140 let c = MutableByteArray mbarr#

141141 in (# s', c #)

142142

143−{-# INLINE newBytes #-}

144−newBytes :: PinnedState -> Int -> IO MutableByteArray

145−newBytes Unpinned = newUnpinnedBytes

146−newBytes Pinned = pinnedNewBytes

143+{-# INLINE newBytesAs #-}

144+newBytesAs :: PinnedState -> Int -> IO MutableByteArray

145+newBytesAs Unpinned = newBytes

146+newBytesAs Pinned = pinnedNewBytes

147147

148148-------------------------------------------------------------------------------

149149-- Pinning & Unpinning

@@ -178,7 +178,7 @@ pin arr@(MutableByteArray marr#) =

178178 then return arr

179179 else

180180#ifdef DEBUG

181− do

181+ do

182182 -- XXX dump stack trace

183183 trace ("pin: Copying array") (return ())

184184#endif

@@ -194,7 +194,7 @@ unpin arr@(MutableByteArray marr#) =

194194 then return arr

195195 else

196196#ifdef DEBUG

197− do

197+ do

198198 -- XXX dump stack trace

199199 trace ("unpin: Copying array") (return ())

200200#endif

core/src/Streamly/Internal/FileSystem/Dir.hs+2−2

@@ -444,7 +444,7 @@ fromChunksWithBufferOf n h xs = fromChunks h $ AS.compact n xs

444444-- @since 0.7.0

445445{-# INLINE fromStreamWithBufferOf #-}

446446fromStreamWithBufferOf :: MonadIO m => Int -> Handle -> Stream m Word8 -> m ()

447−fromStreamWithBufferOf n h m = fromChunks h $ S.chunksOfWith Pinned n m

447+fromStreamWithBufferOf n h m = fromChunks h $ S.pinnedChunksOf n m

448448-- fromStreamWithBufferOf n h m = fromChunks h $ AS.chunksOf n m

449449

450450-- > write = 'writeWithBufferOf' A.defaultChunkSize

@@ -495,7 +495,7 @@ writeChunksWithBufferOf n h = lpackArraysChunksOf n (writeChunks h)

495495-- @since 0.7.0

496496{-# INLINE writeWithBufferOf #-}

497497writeWithBufferOf :: MonadIO m => Int -> Handle -> Fold m Word8 ()

498−writeWithBufferOf n h = FL.groupsOf n (writeNUnsafeUsing Pinned n) (writeChunks h)

498+writeWithBufferOf n h = FL.groupsOf n (pinnedWriteNUnsafe n) (writeChunks h)

499499

500500-- > write = 'writeWithBufferOf' A.defaultChunkSize

501501--

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

@@ -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(..), writeNUnsafeAs)

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

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

108108import Streamly.Data.Stream (Stream)

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

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

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

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

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

@@ -114,7 +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)

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

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

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

120120 (Step(..), snoc, reduce)

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

405405{-# INLINE fromBytesWith #-}

406406fromBytesWith :: (MonadIO m, MonadCatch m)

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

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

408+fromBytesWith n file xs = fromChunks file $ IA.pinnedChunksOf n xs

409409

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

411411{-# INLINE fromBytesWithBufferOf #-}

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

464464writeWith :: (MonadIO m, MonadCatch m)

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

466466writeWith n path =

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

467+ groupsOf n (pinnedWriteNUnsafe n) (writeChunks path)

468468

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

470470{-# INLINE writeWithBufferOf #-}

@@ -503,7 +503,7 @@ writeAppendChunks = fromChunksMode AppendMode

503503writeAppendWith :: (MonadIO m, MonadCatch m)

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

505505writeAppendWith n file xs =

506− writeAppendChunks file $ IA.chunksOfAs Pinned n xs

506+ writeAppendChunks file $ IA.pinnedChunksOf n xs

507507

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

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

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

@@ -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 (PinnedState(..), Unbox)

118+import Streamly.Internal.Data.Unbox (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(..), writeNUnsafeAs, unsafeFreezeWithShrink, byteLength)

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

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

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

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

@@ -425,7 +425,7 @@ putChunksWith :: (MonadIO m, Unbox a)

425425 => Int -> Handle -> Stream m (Array a) -> m ()

426426putChunksWith n h xs = putChunks h $ AS.compact n xs

427427

428−-- > putBytesWith n h m = Handle.putChunks h $ A.chunksOfWith Pinned n m

428+-- > putBytesWith n h m = Handle.putChunks h $ A.pinnedChunksOf n m

429429

430430-- | @putBytesWith bufsize handle stream@ writes @stream@ to @handle@

431431-- in chunks of @bufsize@. A write is performed to the IO device as soon as we

@@ -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.chunksOfAs Pinned n m

436+putBytesWith n h m = putChunks h $ A.pinnedChunksOf 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 (writeNUnsafeAs Pinned n) (writeChunks h)

505+writeWith n h = FL.groupsOf n (pinnedWriteNUnsafe 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.writeNAs Pinned n

523+ let writeNJusts = FL.lmap fromJust $ A.pinnedWriteN 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 $ writeNUnsafeAs Pinned n) chunkWriter

533+ FL.refoldMany (FL.take n $ pinnedWriteNUnsafe 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

src/Streamly/Internal/Network/Inet/TCP.hs+4−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(..), writeNUnsafeAs)

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

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 (PinnedState(..), Unbox)

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

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

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

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

@@ -386,7 +386,7 @@ putBytesWithBufferOf

386386 -> Stream m Word8

387387 -> m ()

388388putBytesWithBufferOf n addr port m =

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

389+ putChunks addr port $ A.pinnedChunksOf n m

390390

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

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

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

400400 -> PortNumber

401401 -> Fold m Word8 ()

402402writeWithBufferOf n addr port =

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

403+ FL.groupsOf n (pinnedWriteNUnsafe n) (writeChunks addr port)

404404

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

406406--

src/Streamly/Internal/Network/Socket.hs+6−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 (PinnedState(..), Unbox)

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

7676import Network.Socket

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

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

@@ -97,8 +97,8 @@ import Streamly.Internal.System.IO (defaultChunkSize)

9797import 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, writeNUnsafeAs, chunksOfAs

101− , writeNAs

100+ ( unsafeFreeze, asPtrUnsafe, byteLength, pinnedWriteNUnsafe, pinnedChunksOf

101+ , pinnedWriteN

102102 )

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

104104 (MutArray(..), pinnedNewBytes, asPtrUnsafe)

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

497497--

498498{-# INLINE putBytesWith #-}

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

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

500+putBytesWith n h m = putChunks h $ A.pinnedChunksOf n m

501501

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

503503-- specified number of bytes before writing.

504504--

505505{-# INLINE writeWith #-}

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

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

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

508508

509509-- | Same as 'writeWith'

510510--

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

522522writeMaybesWith :: (MonadIO m )

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

524524writeMaybesWith n h =

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

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

526526 writeOnNothing = FL.takeEndBy_ isNothing writeNJusts

527527 in FL.many writeOnNothing (writeChunks h)

528528

test/Streamly/Test/Data/Unbox.hs+4−4

@@ -34,7 +34,7 @@ import Streamly.Internal.Data.Unbox

3434 , genericPeekByteIndex

3535 , genericPokeByteIndex

3636 , genericSizeOf

37− , newUnpinnedBytes

37+ , newBytes

3838 , pokeByteIndex

3939 )

4040

@@ -114,7 +114,7 @@ testSerialization ::

114114 => a

115115 -> IO ()

116116testSerialization val = do

117− arr <- newUnpinnedBytes (sizeOf (Proxy :: Proxy a))

117+ arr <- newBytes (sizeOf (Proxy :: Proxy a))

118118 pokeByteIndex 0 arr val

119119 peekByteIndex 0 arr `shouldReturn` val

120120

@@ -136,7 +136,7 @@ testGenericConsistency val = do

136136 sizeOf (Proxy :: Proxy a) `shouldBe` genericSizeOf (Proxy :: Proxy a)

137137

138138 -- Test the serialization and deserialization

139− arr <- newUnpinnedBytes (sizeOf (Proxy :: Proxy a))

139+ arr <- newBytes (sizeOf (Proxy :: Proxy a))

140140

141141 pokeByteIndex 0 arr val

142142 genericPeekByteIndex arr 0 `shouldReturn` val

@@ -162,7 +162,7 @@ testSerializeList sizeOfA val = do

162162

163163 sz `shouldBe` sizeOfA

164164

165− arr <- newUnpinnedBytes sz

165+ arr <- newBytes sz

166166

167167 off1 <- Serialize.serialize 0 arr val

168168 (off2, val2) <- Serialize.deserialize 0 arr