abdc6e1 · streamly · git.iko.soy

git.iko.soy/streamly

Update Array docs and fix spliceCopy

commitabdc6e1a4b3faf4d914b2dae6be9e67545a27b99 authorHarendra Kumar <harendra@composewell.com> dateWed 26 Jul 2023, 01:21 (3 years ago) parent0bbfb4a

6 files changed+47−24

compare with master

unifiedsplit

core/docs/Changelog.md+5−2 core/src/Streamly/Data/Array.hs+10−14 core/src/Streamly/Internal/Data/Array/Mut/Type.hs+10−2 core/src/Streamly/Internal/Data/Unbox.hs+18−2 core/src/Streamly/Internal/FileSystem/Dir.hs+2−2 core/src/Streamly/Internal/FileSystem/Handle.hs+2−2

core/docs/Changelog.md+5−2

@@ -2,8 +2,11 @@

33## Unreleased

5−* Array creation APIs now return unpinned arrays. To create pinned arrays, use

6− the internal APIs with the `pinned*` prefix.

5+* Arrays are now created unpinned by default, they were created pinned

6+ earlier. During IO operations unpinned arrays are automatically copied

7+ to pinned memory. When arrays are directly passed to IO operations

8+ programmers can choose to create them pinned to avoid a copy. To

9+ create pinned arrays, use the internal APIs with the `pinned*` prefix.

710

811### Deprecations

912

core/src/Streamly/Data/Array.hs+10−14

@@ -117,9 +117,11 @@ import Prelude hiding (read)

117117--

118118-- == Pinned and Unpinned Arrays

119119--

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

121−-- arrays are unpinned by default and are only pinned while doing any I/O

122−-- operations.

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

121+-- default array creation operations create unpinned arrays. IO operations

122+-- automatically copy an array to pinned memory if the array passed to it is

123+-- unpinned. Programmers can use appropriate pinned array generation APIs to

124+-- reduce the copying if it happens.

123125--

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

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

@@ -130,20 +132,14 @@ import Prelude hiding (read)

130132--

131133-- == Creating Arrays from Non-IO Streams

132134--

133−-- Array creation folds require 'MonadIO' because they need to sequence effects

134−-- in IO streams. To operate on streams in pure Monads like 'Identity' you can

135−-- morph it to IO monad as follows:

135+-- Array creation folds require 'MonadIO' otherwise the compiler may

136+-- incorrectly share the array memory thinking it is pure.

136137--

137−-- The 'MonadIO' based folds can be morphed to 'Identity' stream folds:

138+-- See the 'fromPureStream' unreleased API to generate an array from an

139+-- Identity stream safely without using MonadIO constraint.

138140--

139−-- >>> purely = Fold.morphInner (Identity . unsafePerformIO)

140−-- >>> Stream.fold (purely Array.write) $ Stream.fromList [1,2,3::Int]

141−-- Identity fromList [1,2,3]

142141--

143−-- Since it is a pure stream we can use 'unsafePerformIO' to extract the result

144−-- of fold from IO.

145−--

146−-- Alternatively, 'Identity' streams can be generalized to IO streams:

142+-- Note that 'Identity' streams can be generalized to IO streams:

147143--

148144-- >>> pure = Stream.fromList [1,2,3] :: Stream Identity Int

149145-- >>> generally = Stream.morphInner (return . runIdentity)

core/src/Streamly/Internal/Data/Array/Mut/Type.hs+10−2

@@ -773,6 +773,7 @@ reallocExplicit elemSize newCapacityInBytes MutArray{..} = do

773773-- down to multiples of array size. If the new size is more than

774774-- 'largeObjectThreshold' then it is rounded up to the block size (4K).

775775--

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

776777{-# INLINABLE realloc #-}

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

778779realloc bytes arr = liftIO $ reallocExplicit (SIZE_OF(a)) bytes arr

@@ -2094,7 +2095,8 @@ putSliceUnsafe src srcStartBytes dst dstStartBytes lenBytes = liftIO $ do

20942095 arrS# srcStartBytes# arrD# dstStartBytes# lenBytes# s#

20952096 , () #)

20962097

2097−-- | Copy two arrays into a newly allocated array.

2098+-- | Copy two arrays into a newly allocated array. If the first array is pinned

2099+-- the spliced array is also pinned.

20982100{-# INLINE spliceCopy #-}

20992101spliceCopy :: forall m a. MonadIO m =>

21002102#ifdef DEVBUILD

@@ -2106,7 +2108,11 @@ spliceCopy arr1 arr2 = liftIO $ do

21062108 start2 = arrStart arr2

21072109 len1 = arrEnd arr1 - start1

21082110 len2 = arrEnd arr2 - start2

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

2111+ let newLen = len1 + len2

2112+ newArrContents <-

2113+ if Unboxed.isPinned (arrContents arr1)

2114+ then Unboxed.newPinnedBytes newLen

2115+ else Unboxed.newUnpinnedBytes newLen

21102116 let len = len1 + len2

21112117 newArr = MutArray newArrContents 0 len len

21122118 putSliceUnsafe arr1 start1 newArr 0 len1

@@ -2169,6 +2175,8 @@ spliceWith sizer dst@(MutArray _ start end bound) src = do

21692175--

21702176-- >>> splice = MutArray.spliceWith (+)

21712177--

2178+-- If the original array is pinned the spliced array is also pinned.

2179+--

21722180-- /Pre-release/

21732181{-# INLINE splice #-}

21742182splice :: (MonadIO m, Unbox a) => MutArray a -> MutArray a -> m (MutArray a)

core/src/Streamly/Internal/Data/Unbox.hs+18−2

@@ -53,6 +53,9 @@ import Data.Functor.Const (Const(..))

5353import Data.Functor.Identity (Identity(..))

5454import Data.Kind (Type)

5555import Data.Proxy (Proxy (..))

56+#ifdef DEBUG

57+import Debug.Trace (trace)

58+#endif

5659import Foreign.Ptr (IntPtr(..), WordPtr(..))

5760import GHC.Base (IO(..))

5861import GHC.Fingerprint.Type (Fingerprint(..))

@@ -150,6 +153,7 @@ newBytes Pinned = pinnedNewBytes

150153isPinned :: MutableByteArray -> Bool

151154isPinned (MutableByteArray arr#) =

152155 let pinnedInt = I# (isMutableByteArrayPinned# arr#)

156+ -- XXX /= 0 would be safer

153157 in pinnedInt == 1

154158

155159

@@ -173,7 +177,13 @@ pin :: MutableByteArray -> IO MutableByteArray

173177pin arr@(MutableByteArray marr#) =

174178 if isPinned arr

175179 then return arr

176− else IO

180+ else

181+#ifdef DEBUG

182+ do

183+ -- XXX dump stack trace

184+ trace ("pin: Copying array") (return ())

185+#endif

186+ IO

177187 $ \s# ->

178188 case cloneMutableArrayWith# newPinnedByteArray# marr# s# of

179189 (# s1#, marr1# #) -> (# s1#, MutableByteArray marr1# #)

@@ -183,7 +193,13 @@ unpin :: MutableByteArray -> IO MutableByteArray

183193unpin arr@(MutableByteArray marr#) =

184194 if not (isPinned arr)

185195 then return arr

186− else IO

196+ else

197+#ifdef DEBUG

198+ do

199+ -- XXX dump stack trace

200+ trace ("unpin: Copying array") (return ())

201+#endif

202+ IO

187203 $ \s# ->

188204 case cloneMutableArrayWith# newByteArray# marr# s# of

189205 (# s1#, marr1# #) -> (# s1#, MutableByteArray marr1# #)

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.chunksOf n m

447+fromStreamWithBufferOf n h m = fromChunks h $ S.chunksOfWith Pinned 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 (writeNUnsafe n) (writeChunks h)

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

499499

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

501501--

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

@@ -425,12 +425,12 @@ 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

429+

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

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

430432-- collect the required input size.

431433--

432−-- >>> putBytesWith n h m = Handle.putChunks h $ Stream.chunksOf n m

433−--

434434{-# INLINE putBytesWith #-}

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

436436putBytesWith n h m = putChunks h $ A.chunksOfAs Pinned n m