06b0cb1 · streamly · git.iko.soy

benchmark/ArrayOps.hs+5−5

compare with master

unifiedsplit

@@ -49,7 +49,7 @@ type Stream = A.Array

49

50{-# INLINE sourceUnfoldr #-}

51sourceUnfoldr :: MonadIO m => Int -> m (Stream Int)

52−sourceUnfoldr n = A.writeN value $ S.unfoldr step n

52+sourceUnfoldr n = S.runFold (A.writeN value) $ S.unfoldr step n

53 where

54 step cnt =

55 if cnt > n + value

@@ -58,18 +58,18 @@ sourceUnfoldr n = A.writeN value $ S.unfoldr step n

58

59{-# INLINE sourceIntFromTo #-}

60sourceIntFromTo :: MonadIO m => Int -> m (Stream Int)

61−sourceIntFromTo n = A.writeN value $ S.enumerateFromTo n (n + value)

61+sourceIntFromTo n = S.runFold (A.writeN value) $ S.enumerateFromTo n (n + value)

62

63{-# INLINE sourceIntFromToFromStream #-}

64sourceIntFromToFromStream :: MonadIO m => Int -> m (Stream Int)

65−sourceIntFromToFromStream n = A.write $ S.enumerateFromTo n (n + value)

65+sourceIntFromToFromStream n = S.runFold A.write $ S.enumerateFromTo n (n + value)

66

67sourceIntFromToFromList :: MonadIO m => Int -> m (Stream Int)

68sourceIntFromToFromList n = P.return $ A.fromList $ [n..n + value]

69

70{-# INLINE sourceFromList #-}

71sourceFromList :: MonadIO m => Int -> m (Stream Int)

72−sourceFromList n = A.writeN value $ S.fromList [n..n+value]

72+sourceFromList n = S.runFold (A.writeN value) $ S.fromList [n..n+value]

73

74{-# INLINE sourceIsList #-}

75sourceIsList :: Int -> Stream Int

@@ -280,7 +280,7 @@ onArray

280 :: MonadIO m => (S.SerialT m Int -> S.SerialT m Int)

281 -> Stream Int

282 -> m (Stream Int)

283−onArray f arr = A.writeN value $ f $ A.read arr

283+onArray f arr = S.runFold (A.writeN value) $ f $ A.read arr

284

285scanl' n = composeN n $ onArray $ S.scanl' (+) 0

286scanl1' n = composeN n $ onArray $ S.scanl1' (+)

benchmark/FileIO.hs+10−10

@@ -118,7 +118,7 @@ main = do

118 S.foldl' (\acc arr -> acc + foldlArr' (+) 0 arr) 0 s

119 , mkBench "cat" href $ do

120 Handles inh _ <- readIORef href

121− FH.writeArrays devNull $ FH.readArraysOf (256*1024) inh

121+ S.runFold (FH.writeArrays devNull) $ FH.readArraysOf (256*1024) inh

122 ]

123 , bgroup "readStream"

124 [ mkBench "last" href $ do

@@ -138,32 +138,32 @@ main = do

138 S.sum $ FH.read inh

139 , mkBench "cat" href $ do

140 Handles inh _ <- readIORef href

141− FH.write devNull $ FH.read inh

141+ S.runFold (FH.write devNull) $ FH.read inh

142 ]

143 , bgroup "copyArray"

144 [ mkBench "copy" href $ do

145 Handles inh outh <- readIORef href

146 let s = FH.readArrays inh

147− FH.writeArrays outh s

147+ S.runFold (FH.writeArrays outh) s

148 ]

149#ifdef DEVBUILD

150 -- This takes a little longer therefore put under the dev conditional

151 , bgroup "copyStream"

152 [ mkBench "fromToHandle" href $ do

153 Handles inh outh <- readIORef href

154− FH.write outh (FH.read inh)

154+ S.runFold (FH.write outh) (FH.read inh)

155 ]

156 -- This needs an ascii file, as decode just errors out.

157 , bgroup "decode-encode"

158 [ mkBench "char8" href $ do

159 Handles inh outh <- readIORef href

160− FH.write outh

160+ S.runFold (FH.write outh)

161 $ SS.encodeChar8

162 $ SS.decodeChar8

163 $ FH.read inh

164 , mkBench "utf8" href $ do

165 Handles inh outh <- readIORef href

166− FH.write outh

166+ S.runFold (FH.write outh)

167 $ SS.encodeUtf8

168 $ SS.decodeUtf8

169 $ FH.read inh

@@ -171,7 +171,7 @@ main = do

171 , bgroup "grouping"

172 [ mkBench "chunksOf 1 (A.writeNF)" href $ do

173 Handles inh _ <- readIORef href

174− S.length $ S.chunksOf fileSize (A.writeNF fileSize)

174+ S.length $ S.chunksOf fileSize (A.writeN fileSize)

175 (FH.read inh)

176

177 , mkBench "chunksOf 1" href $ do

@@ -187,7 +187,7 @@ main = do

187 , bgroup "group-ungroup"

188 [ mkBench "lines-unlines" href $ do

189 Handles inh outh <- readIORef href

190− FH.write outh

190+ S.runFold (FH.write outh)

191 $ SS.encodeChar8

192 $ SS.unlines

193 $ SS.lines

@@ -195,13 +195,13 @@ main = do

195 $ FH.read inh

196 , mkBench "lines-unlines-arrays" href $ do

197 Handles inh outh <- readIORef href

198− FH.writeArraysPackedUpto (1024*1024) outh

198+ S.runFold (FH.writeArraysInChunksOf (1024*1024) outh)

199 $ Internal.insertAfterEach (return $ A.fromList [10])

200 $ AS.splitOn 10

201 $ FH.readArraysOf (1024*1024) inh

202 , mkBench "words-unwords" href $ do

203 Handles inh outh <- readIORef href

204− FH.write outh

204+ S.runFold (FH.write outh)

205 $ SS.encodeChar8

206 $ SS.unwords

207 $ SS.words

benchmark/Linear.hs+1−1

@@ -224,7 +224,7 @@ main =

224 , benchIOSink "toListRevF" (S.runFold Internal.toListRevF)

225 , benchIOSink "toStream" (S.runFold Internal.toStream)

226 , benchIOSink "toStreamRev" (S.runFold Internal.toStreamRev)

227− , benchIOSink "writeNF" (S.runFold (A.writeNF Ops.value))

227+ , benchIOSink "writeN" (S.runFold (A.writeN Ops.value))

228

229 , benchIOSink "index" (S.runFold (FL.index Ops.maxValue))

230 , benchIOSink "head" (S.runFold FL.head)

examples/EchoServer.hs+2−1

@@ -7,5 +7,6 @@ import qualified Streamly.Prelude as S

7

8main :: IO ()

9main = S.drain

10− $ parallely $ S.mapM (flip withSocket (\sk -> writeArrays sk $ readArrays sk))

10+ $ parallely $ S.mapM (flip withSocket echo)

11 $ serially $ connectionsOnAllAddrs 8090

12+ where echo sk = S.runFold (writeArrays sk) $ readArrays sk

examples/FileSinkServer.hs+2−2

@@ -21,7 +21,7 @@ main :: IO ()

21main = do

22 file <- fmap head getArgs

23 withFile file AppendMode

24− (\src -> FH.write src

24+ (\src -> S.runFold (FH.write src)

25 $ encodeChar8Unchecked

26 $ S.concatMap A.read

27 $ S.concatMapBy parallel (flip NS.withSocketS recv)

@@ -30,6 +30,6 @@ main = do

30 where

31

32 recv =

33− S.splitBySuffix (== '\n') A.writeF

33+ S.splitBySuffix (== '\n') A.write

34 . decodeChar8

35 . NS.read

examples/FromFileClient.hs+2−1

@@ -16,5 +16,6 @@ main :: IO ()

16main =

17 let sendFile file =

18 withFile file ReadMode $ \src ->

19− Client.writeArrays (127, 0, 0, 1) 8090 $ FH.readArrays src

19+ S.runFold (Client.writeArrays (127, 0, 0, 1) 8090)

20+ $ FH.readArrays src

20 in getArgs >>= S.drain . parallely . S.mapM sendFile . S.fromList

21 in getArgs >>= S.drain . parallely . S.mapM sendFile . S.fromList

examples/HandleIO.hs+2−2

@@ -12,12 +12,12 @@ import System.Environment (getArgs)

12import System.IO (IOMode(..), hSeek, SeekMode(..))

13

14cat :: FH.Handle -> IO ()

15−cat src = FH.writeArrays FH.stdout $ FH.readArraysOf (256*1024) src

15+cat src = S.runFold (FH.writeArrays FH.stdout) $ FH.readArraysOf (256*1024) src

16-- byte stream version

17-- cat src = FH.write FH.stdout $ FH.read src

18

19cp :: FH.Handle -> FH.Handle -> IO ()

20−cp src dst = FH.writeArrays dst $ FH.readArraysOf (256*1024) src

20+cp src dst = S.runFold (FH.writeArrays dst) $ FH.readArraysOf (256*1024) src

21-- byte stream version

22-- cp src dst = FH.write dst $ FH.read src

23

src/Streamly/FileSystem/Handle.hs+54−12

@@ -86,7 +86,8 @@ module Streamly.FileSystem.Handle

86 -- -- * Array Write

87 , writeArray

88 , writeArrays

89− , writeArraysPackedUpto

89+ -- , writeArraysPackedUpto

90+ , writeArraysInChunksOf

90

91

91 -- -- * Random Access (Seek)

92 -- -- * Random Access (Seek)

92 -- -- | Unlike the streaming APIs listed above, these APIs apply to devices or

93 -- -- | Unlike the streaming APIs listed above, these APIs apply to devices or

@@ -122,10 +123,12 @@ import Prelude hiding (read)

122import Streamly.Memory.Array.Types (Array(..))

123import Streamly.Memory.Array.Types (Array(..))

123import Streamly.Streams.Serial (SerialT)

124import Streamly.Streams.Serial (SerialT)

124import Streamly.Streams.StreamK.Type (IsStream, mkStream)

125import Streamly.Streams.StreamK.Type (IsStream, mkStream)

125−import Streamly.Memory.Array.Types (defaultChunkSize, shrinkToFit)

126+import Streamly.Memory.Array.Types (defaultChunkSize, shrinkToFit, lpackArraysChunksOf)

126−-- import Streamly.Fold (Fold)

127+import Streamly.Fold (Fold)

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

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

128

129

130+import qualified Streamly.Fold as FL

131+import qualified Streamly.Fold.Types as FL

129import qualified Streamly.Memory.Array as A

132import qualified Streamly.Memory.Array as A

130import qualified Streamly.Memory.ArrayStream as AS

133import qualified Streamly.Memory.ArrayStream as AS

131import qualified Streamly.Prelude as S

134import qualified Streamly.Prelude as S

@@ -265,12 +268,20 @@ read = AS.flatten . readArrays

265-- Writing

268-- Writing

266-------------------------------------------------------------------------------

269-------------------------------------------------------------------------------

267

270

271+-- | Write a stream of arrays to a handle.

272+--

273+-- @since 0.7.0

274+{-# INLINE writeArraysS #-}

275+writeArraysS :: (MonadIO m, Storable a)

276+ => Handle -> SerialT m (Array a) -> m ()

277+writeArraysS h m = S.mapM_ (liftIO . writeArray h) m

278+

268-- | Write a stream of arrays to a handle.

279-- | Write a stream of arrays to a handle.

269--

280--

270-- @since 0.7.0

281-- @since 0.7.0

271{-# INLINE writeArrays #-}

282{-# INLINE writeArrays #-}

272−writeArrays :: (MonadIO m, Storable a) => Handle -> SerialT m (Array a) -> m ()

283+writeArrays :: (MonadIO m, Storable a) => Handle -> Fold m (Array a) ()

273−writeArrays h m = S.mapM_ (liftIO . writeArray h) m

284+writeArrays h = FL.drainBy (liftIO . writeArray h)

274

285

275-- | @writeArraysPackedUpto chunkSize handle stream@ writes a stream of arrays

286-- | @writeArraysPackedUpto chunkSize handle stream@ writes a stream of arrays

276-- to @handle@ after coalescing the adjacent arrays in chunks of @chunkSize@.

287-- to @handle@ after coalescing the adjacent arrays in chunks of @chunkSize@.

@@ -278,10 +289,22 @@ writeArrays h m = S.mapM_ (liftIO . writeArray h) m

278-- we do not split the arrays to fit exactly to the specified size.

289-- we do not split the arrays to fit exactly to the specified size.

279--

290--

280-- @since 0.7.0

291-- @since 0.7.0

281−{-# INLINE writeArraysPackedUpto #-}

292+{-# INLINE _writeArraysPackedUpto #-}

282−writeArraysPackedUpto :: (MonadIO m, Storable a)

293+_writeArraysPackedUpto :: (MonadIO m, Storable a)

283 => Int -> Handle -> SerialT m (Array a) -> m ()

294 => Int -> Handle -> SerialT m (Array a) -> m ()

284−writeArraysPackedUpto n h xs = writeArrays h $ AS.compact n xs

295+_writeArraysPackedUpto n h xs = writeArraysS h $ AS.compact n xs

296+

297+-- | @writeArraysInChunksOf chunkSize handle@ writes a stream of arrays

298+-- to @handle@ after coalescing the adjacent arrays in chunks of @chunkSize@.

299+-- We never split an array, if a single array is bigger than the specified size

300+-- it emitted as it is. Multiple arrays are coalesed as long as the total size

301+-- remains below the specified size.

302+--

303+-- @since 0.7.0

304+{-# INLINE writeArraysInChunksOf #-}

305+writeArraysInChunksOf :: (MonadIO m, Storable a)

306+ => Int -> Handle -> Fold m (Array a) ()

307+writeArraysInChunksOf n h = lpackArraysChunksOf n (writeArrays h)

285

308

286-- GHC buffer size dEFAULT_FD_BUFFER_SIZE=8192 bytes.

309-- GHC buffer size dEFAULT_FD_BUFFER_SIZE=8192 bytes.

287--

310--

@@ -296,18 +319,37 @@ writeArraysPackedUpto n h xs = writeArrays h $ AS.compact n xs

296-- collect the required input size.

319-- collect the required input size.

297--

320--

298-- @since 0.7.0

321-- @since 0.7.0

322+{-# INLINE writeInChunksOfS #-}

323+writeInChunksOfS :: MonadIO m => Int -> Handle -> SerialT m Word8 -> m ()

324+writeInChunksOfS n h m = writeArraysS h $ AS.arraysOf n m

325+

326+-- | @writeInChunksOf chunkSize handle@ writes the input stream to @handle@ in

327+-- chunks of @chunkSize@. A write is performed to the IO device as soon as we

328+-- collect the required input size.

329+--

330+-- @since 0.7.0

299{-# INLINE writeInChunksOf #-}

331{-# INLINE writeInChunksOf #-}

300−writeInChunksOf :: MonadIO m => Int -> Handle -> SerialT m Word8 -> m ()

332+writeInChunksOf :: MonadIO m => Int -> Handle -> Fold m Word8 ()

301−writeInChunksOf n h m = writeArrays h $ AS.arraysOf n m

333+writeInChunksOf n h = FL.lchunksOf n A.write (writeArrays h)

334+

335+-- > write = 'writeInChunksOf' A.defaultChunkSize

336+--

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

338+-- up to 'A.defaultChunkSize' before writing.

339+--

340+-- @since 0.7.0

341+{-# INLINE _writeS #-}

342+_writeS :: MonadIO m => Handle -> SerialT m Word8 -> m ()

343+_writeS = writeInChunksOfS defaultChunkSize

302

344

303-- > write = 'writeInChunksOf' A.defaultChunkSize

345-- > write = 'writeInChunksOf' A.defaultChunkSize

304--

346--

305−-- | Write a byte stream to a file handle. Combines the bytes in chunks of

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

306-- up to 'A.defaultChunkSize' before writing.

348-- up to 'A.defaultChunkSize' before writing.

307--

349--

308-- @since 0.7.0

350-- @since 0.7.0

309{-# INLINE write #-}

351{-# INLINE write #-}

310−write :: MonadIO m => Handle -> SerialT m Word8 -> m ()

352+write :: MonadIO m => Handle -> Fold m Word8 ()

311write = writeInChunksOf defaultChunkSize

353write = writeInChunksOf defaultChunkSize

312

354

313{-

355{-

src/Streamly/Fold/Types.hs+22−7

@@ -23,7 +23,10 @@ module Streamly.Fold.Types

23 , ltakeWhile

24 , lsessionsOf

25 , lchunksOf

26+

26 , duplicate

27 , duplicate

28+ , initialize

29+ , runStep

27 )

30 )

28where

31where

29

32

@@ -318,14 +321,26 @@ duplicate :: Applicative m => Fold m a b -> Fold m a (Fold m a b)

318duplicate (Fold step begin done) =

321duplicate (Fold step begin done) =

319 Fold step begin (\x -> pure (Fold step (pure x) done))

322 Fold step begin (\x -> pure (Fold step (pure x) done))

320

323

321−-- | Group the input stream into groups of @n@ elements each and then fold each

324+-- | Run the initialization effect of a fold. The returned fold would use the

322−-- group using the provided fold function.

325+-- value returned by this effect as its initial value.

323−--

324−-- @

325--

326--

327+{-# INLINABLE initialize #-}

326−-- -----Fold m a b----|-Fold n a c-|-Fold n a c-|-...-|----Fold m a c

328+initialize :: Monad m => Fold m a b -> m (Fold m a b)

327−--

329+initialize (Fold step initial extract) = do

328−-- @

330+ i <- initial

331+ return $ Fold step (return i) extract

332+

333+-- | Run one step of a fold and store the accumulator as an initial value in

334+-- the returned fold.

335+{-# INLINABLE runStep #-}

336+runStep :: Monad m => Fold m a b -> a -> m (Fold m a b)

337+runStep (Fold step initial extract) a = do

338+ i <- initial

339+ r <- step i a

340+ return $ (Fold step (return r) extract)

341+

342+-- | For every n input items, apply the first fold and supply the result to the

343+-- next fold.

329--

344--

330{-# INLINE lchunksOf #-}

345{-# INLINE lchunksOf #-}

331lchunksOf :: Monad m => Int -> Fold m a b -> Fold m b c -> Fold m a c

346lchunksOf :: Monad m => Int -> Fold m a b -> Fold m b c -> Fold m a c

src/Streamly/Internal.hs+2

@@ -64,6 +64,8 @@ module Streamly.Internal

64 , lsessionsOf

65 , lchunksOf

66 , duplicate

67+ , initialize

68+ , runStep

67 )

69 )

68where

70where

69

71

src/Streamly/Memory/Array.hs+12−12

@@ -79,12 +79,12 @@ module Streamly.Memory.Array

79

80 -- Monadic APIs

81 -- , newArray

82− , writeN

82+ , A.writeN

83 , write

84

85 -- Stream Folds

86− , A.writeNF

86+ -- , writeNS

87− , writeF

87+ -- , writeS

88

89 -- * Elimination

90 -- 'GHC.Exts.toList' from "GHC.Exts" can be used to convert an array to a

@@ -157,9 +157,9 @@ newArray len = undefined

157-- array may hold less than N elements.

158--

159-- @since 0.7.0

160−{-# INLINE writeN #-}

160+{-# INLINE _writeNS #-}

161−writeN :: (MonadIO m, Storable a) => Int -> SerialT m a -> m (Array a)

161+_writeNS :: (MonadIO m, Storable a) => Int -> SerialT m a -> m (Array a)

162−writeN n m = do

162+_writeNS n m = do

163 if n < 0 then error "writeN: negative write count specified" else return ()

164 A.fromStreamDN n $ D.toStreamD m

165

@@ -204,9 +204,9 @@ toArrayMinChunk elemCount = Fold step initial extract

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

205--

206-- @since 0.7.0

207−{-# INLINE writeF #-}

207+{-# INLINE write #-}

208−writeF :: forall m a. (MonadIO m, Storable a) => Fold m a (Array a)

208+write :: forall m a. (MonadIO m, Storable a) => Fold m a (Array a)

209−writeF = toArrayMinChunk (A.bytesToCount (undefined :: a) (A.mkChunkSize 1024))

209+write = toArrayMinChunk (A.bytesToCount (undefined :: a) (A.mkChunkSize 1024))

210

211-- | Create an 'Array' from a stream. This is useful when we want to create a

212-- single array from a stream of unknown size. 'writeN' is at least twice

@@ -216,9 +216,9 @@ writeF = toArrayMinChunk (A.bytesToCount (undefined :: a) (A.mkChunkSize 1024))

216-- may fail. When the stream size is not known, `arraysOf` followed by

217-- processing of indvidual arrays in the resulting stream should be preferred.

218--

219−{-# INLINE write #-}

219+{-# INLINE _writeS #-}

220−write :: (MonadIO m, Storable a) => SerialT m a -> m (Array a)

220+_writeS :: (MonadIO m, Storable a) => SerialT m a -> m (Array a)

221−write = P.runFold writeF

221+_writeS = P.runFold write

222-- write m = A.fromStreamD $ D.toStreamD m

223

224-------------------------------------------------------------------------------

src/Streamly/Memory/Array/Types.hs+61−8

@@ -39,6 +39,7 @@ module Streamly.Memory.Array.Types

39 , flattenArrays

40 , flattenArraysRev

41 , packArraysChunksOf

42+ , lpackArraysChunksOf

42#if !defined(mingw32_HOST_OS)

43#if !defined(mingw32_HOST_OS)

43 , groupIOVecsOf

44 , groupIOVecsOf

44#endif

45#endif

@@ -59,7 +60,7 @@ module Streamly.Memory.Array.Types

59 , toStreamK

60 , toStreamK

60 , toStreamKRev

61 , toStreamKRev

61 , toList

62 , toList

62− , writeNF

63+ , writeN

63 , read

64 , read

64

65

65 -- * Utilities

66 -- * Utilities

@@ -99,6 +100,7 @@ import GHC.IO (IO(IO), unsafePerformIO)

99import GHC.Ptr (Ptr(..))

100import GHC.Ptr (Ptr(..))

100

101

101import Streamly.Fold.Types (Fold(..))

102import Streamly.Fold.Types (Fold(..))

103+import Streamly.Strict (Tuple'(..))

102import Streamly.SVar (adaptState)

104import Streamly.SVar (adaptState)

103

105

104#if !defined(mingw32_HOST_OS)

106#if !defined(mingw32_HOST_OS)

@@ -472,18 +474,18 @@ foldr f z arr = runIdentity $ D.foldr f z $ toStreamD arr

472-- Instances

474-- Instances

473-------------------------------------------------------------------------------

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

474

476

475−-- | @writeNF n@ folds a maximum of @n@ elements from the input stream to an

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

476-- 'Array'.

478-- 'Array'.

477--

479--

478-- @since 0.7.0

480-- @since 0.7.0

479−{-# INLINE_NORMAL writeNF #-}

481+{-# INLINE_NORMAL writeN #-}

480−writeNF :: forall m a. (MonadIO m, Storable a) => Int -> Fold m a (Array a)

482+writeN :: forall m a. (MonadIO m, Storable a) => Int -> Fold m a (Array a)

481−writeNF n = Fold step initial extract

483+writeN n = Fold step initial extract

482

484

483 where

485 where

484

486

485 initial = do

487 initial = do

486− if n < 0 then error "writeNF: negative count specified" else return ()

488+ if n < 0 then error "writeN: negative count specified" else return ()

487 liftIO $ newArray n

489 liftIO $ newArray n

488 step arr@(Array _ end bound) _ | end == bound = return arr

490 step arr@(Array _ end bound) _ | end == bound = return arr

489 step (Array start end bound) x = do

491 step (Array start end bound) x = do

@@ -516,7 +518,7 @@ data GroupState s start end bound

516{-# INLINE_NORMAL fromStreamDArraysOf #-}

518{-# INLINE_NORMAL fromStreamDArraysOf #-}

517fromStreamDArraysOf :: forall m a. (MonadIO m, Storable a)

519fromStreamDArraysOf :: forall m a. (MonadIO m, Storable a)

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

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

519−-- fromStreamDArraysOf n str = D.groupsOf n (writeNF n) str

521+-- fromStreamDArraysOf n str = D.groupsOf n (writeN n) str

520fromStreamDArraysOf n (D.Stream step state) =

522fromStreamDArraysOf n (D.Stream step state) =

521 D.Stream step' (GroupStart state)

523 D.Stream step' (GroupStart state)

522

524

@@ -887,7 +889,10 @@ data SpliceState s arr

887

889

888-- XXX can use general grouping combinators to achieve this?

890-- XXX can use general grouping combinators to achieve this?

889-- | Coalesce adjacent arrays in incoming stream to form bigger arrays of a

891-- | Coalesce adjacent arrays in incoming stream to form bigger arrays of a

890−-- maximum specified size.

892+-- maximum specified size. Note that if a single array is bigger than the

893+-- specified size we do not split it to fit. When we coalesce multiple arrays

894+-- if the size would exceed the specified size we do not coalesce therefore the

895+-- actual array size may be less than the specified chunk size.

891--

896--

892-- @since 0.7.0

897-- @since 0.7.0

893{-# INLINE_NORMAL packArraysChunksOf #-}

898{-# INLINE_NORMAL packArraysChunksOf #-}

@@ -935,6 +940,54 @@ packArraysChunksOf n (D.Stream step state) =

935

940

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

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

937

942

943+-- XXX instead of writing two different versions of this operation, we should

944+-- write it as a pipe.

945+{-# INLINE_NORMAL lpackArraysChunksOf #-}

946+lpackArraysChunksOf :: (MonadIO m, Storable a)

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

948+lpackArraysChunksOf n (Fold step1 initial1 extract1) =

949+ Fold step initial extract

950+

951+ where

952+

953+ initial = do

954+ when (n <= 0) $

955+ -- XXX we can pass the module string from the higher level API

956+ error $ "Streamly.Memory.Array.Types.packArraysChunksOf: the size of "

957+ ++ "arrays [" ++ show n ++ "] must be a natural number"

958+ r1 <- initial1

959+ return (Tuple' Nothing r1)

960+

961+ extract (Tuple' Nothing r1) = extract1 r1

962+ extract (Tuple' (Just buf) r1) = do

963+ r <- step1 r1 buf

964+ extract1 r

965+

966+ step (Tuple' Nothing r1) arr = do

967+ let len = byteLength arr

968+ in if len >= n

969+ then do

970+ r <- step1 r1 arr

971+ extract1 r

972+ r1' <- initial1

973+ return (Tuple' Nothing r1')

974+ else return (Tuple' (Just arr) r1)

975+

976+ step (Tuple' (Just buf) r1) arr = do

977+ let len = byteLength buf + byteLength arr

978+ buf' <- if byteCapacity buf < len

979+ then liftIO $ realloc len buf

980+ else return buf

981+ buf'' <- spliceWith buf' arr

982+

983+ if len >= n

984+ then do

985+ r <- step1 r1 buf''

986+ extract1 r

987+ r1' <- initial1

988+ return (Tuple' Nothing r1')

989+ else return (Tuple' (Just buf'') r1)

990+

938#if !defined(mingw32_HOST_OS)

991#if !defined(mingw32_HOST_OS)

939data GatherState s arr

992data GatherState s arr

940 = GatherInitial s

993 = GatherInitial s

src/Streamly/Network/Client.hs+50−1

@@ -66,10 +66,13 @@ import Prelude hiding (read)

66import qualified Network.Socket as Net

67

68import Streamly (MonadAsync)

69+import Streamly.Fold.Types (Fold(..))

69import Streamly.Memory.Array.Types (Array(..), defaultChunkSize)

70import Streamly.Memory.Array.Types (Array(..), defaultChunkSize)

70−import Streamly.Streams.Serial (SerialT)

71+-- import Streamly.Streams.Serial (SerialT)

71import Streamly.Streams.StreamK.Type (IsStream)

72import Streamly.Streams.StreamK.Type (IsStream)

72

73

74+import qualified Streamly.Fold.Types as FL

75+import qualified Streamly.Memory.Array as A

73import qualified Streamly.Memory.ArrayStream as AS

76import qualified Streamly.Memory.ArrayStream as AS

74import qualified Streamly.Prelude as S

77import qualified Streamly.Prelude as S

75import qualified Streamly.Network.Socket as SK

78import qualified Streamly.Network.Socket as SK

@@ -114,6 +117,7 @@ read addr port = AS.flatten $ withConnection addr port SK.readArrays

114-- Writing

117-- Writing

115-------------------------------------------------------------------------------

118-------------------------------------------------------------------------------

116

119

120+{-

117-- | Write a stream of arrays to the supplied IPv4 host address and port

121-- | Write a stream of arrays to the supplied IPv4 host address and port

118-- number.

122-- number.

119--

123--

@@ -127,7 +131,27 @@ writeArrays

127 -> m ()

131 -> m ()

128writeArrays addr port xs =

132writeArrays addr port xs =

129 S.drain $ withConnection addr port (\sk -> S.yieldM $ SK.writeArrays sk xs)

133 S.drain $ withConnection addr port (\sk -> S.yieldM $ SK.writeArrays sk xs)

134+-}

130

135

136+-- | Write a stream of arrays to the supplied IPv4 host address and port

137+-- number.

138+--

139+-- @since 0.7.0

140+{-# INLINE writeArrays #-}

141+writeArrays

142+ :: (MonadAsync m)

143+ => (Word8, Word8, Word8, Word8)

144+ -> PortNumber

145+ -> Fold m (Array Word8) ()

146+writeArrays addr port = Fold step initial extract

147+ where

148+ initial = do

149+ skt <- liftIO (openConnection addr port)

150+ FL.initialize (SK.writeArrays skt)

151+ step = FL.runStep

152+ extract (Fold _ initial1 extract1) = initial1 >>= extract1

153+

154+{-

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

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

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

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

133-- input elements.

157-- input elements.

@@ -142,7 +166,23 @@ writeInChunksOf

142 -> SerialT m Word8

166 -> SerialT m Word8

143 -> m ()

167 -> m ()

144writeInChunksOf n addr port m = writeArrays addr port $ AS.arraysOf n m

168writeInChunksOf n addr port m = writeArrays addr port $ AS.arraysOf n m

169+-}

145

170

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

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

173+-- input elements.

174+--

175+-- @since 0.7.0

176+{-# INLINE writeInChunksOf #-}

177+writeInChunksOf

178+ :: (MonadAsync m)

179+ => Int

180+ -> (Word8, Word8, Word8, Word8)

181+ -> PortNumber

182+ -> Fold m Word8 ()

183+writeInChunksOf n addr port = FL.lchunksOf n A.write (writeArrays addr port)

184+

185+{-

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

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

147--

187--

148-- @since 0.7.0

188-- @since 0.7.0

@@ -150,3 +190,12 @@ writeInChunksOf n addr port m = writeArrays addr port $ AS.arraysOf n m

150write :: (MonadCatch m, MonadAsync m)

190write :: (MonadCatch m, MonadAsync m)

151 => (Word8, Word8, Word8, Word8) -> PortNumber -> SerialT m Word8 -> m ()

191 => (Word8, Word8, Word8, Word8) -> PortNumber -> SerialT m Word8 -> m ()

152write = writeInChunksOf defaultChunkSize

192write = writeInChunksOf defaultChunkSize

193+-}

194+

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

196+--

197+-- @since 0.7.0

198+{-# INLINE write #-}

199+write :: MonadAsync m

200+ => (Word8, Word8, Word8, Word8) -> PortNumber -> Fold m Word8 ()

201+write = writeInChunksOf defaultChunkSize

src/Streamly/Network/Socket.hs+38−10

@@ -89,9 +89,11 @@ import qualified Network.Socket as Net

89import Streamly.Memory.Array.Types (Array(..))

90import Streamly.Streams.Serial (SerialT)

91import Streamly.Streams.StreamK.Type (IsStream, mkStream)

92−-- import Streamly.Fold (Fold)

92+import Streamly.Fold (Fold)

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

94

95+import qualified Streamly.Fold as FL

96+import qualified Streamly.Fold.Types as FL

95import qualified Streamly.Memory.Array as A

97import qualified Streamly.Memory.Array as A

96import qualified Streamly.Memory.ArrayStream as AS

98import qualified Streamly.Memory.ArrayStream as AS

97import qualified Streamly.Memory.Array.Types as A hiding (flattenArrays)

99import qualified Streamly.Memory.Array.Types as A hiding (flattenArrays)

@@ -228,10 +230,8 @@ readArraysOf = readArraysUptoWith readArrayOf

228

230

229-- XXX read 'Array a' instead of Word8

231-- XXX read 'Array a' instead of Word8

230--

232--

231−-- | @readArrays h@ reads a stream of arrays from file handle @h@.

233+-- | @readArrays h@ reads a stream of arrays from socket handle @h@.

232-- The maximum size of a single array is limited to @defaultChunkSize@.

234-- The maximum size of a single array is limited to @defaultChunkSize@.

233−-- 'readArrays' ignores the prevailing 'TextEncoding' and 'NewlineMode'

234−-- on the 'Handle'.

235--

236-- @since 0.7.0

237{-# INLINE readArrays #-}

@@ -260,7 +260,7 @@ readInChunksOf chunkSize h = A.flattenArrays $ readArraysUpto chunkSize h

260-- read :: (IsStream t, MonadIO m, Storable a) => Handle -> t m a

261--

262-- > read = 'readByChunks' A.defaultChunkSize

263−-- | Generate a stream of elements of the given type from a file 'Handle'. The

263+-- | Generate a stream of elements of the given type from a socket. The

264-- stream ends when EOF is encountered.

265--

266-- @since 0.7.0

@@ -275,9 +275,17 @@ read = AS.flatten . readArrays

275-- | Write a stream of arrays to a handle.

276--

277-- @since 0.7.0

278+{-# INLINE writeArraysS #-}

279+writeArraysS :: (MonadIO m, Storable a)

280+ => Socket -> SerialT m (Array a) -> m ()

281+writeArraysS h m = S.mapM_ (liftIO . writeArray h) m

282+

283+-- | Write a stream of arrays to a socket.

284+--

285+-- @since 0.7.0

278{-# INLINE writeArrays #-}

286{-# INLINE writeArrays #-}

279−writeArrays :: (MonadIO m, Storable a) => Socket -> SerialT m (Array a) -> m ()

287+writeArrays :: (MonadIO m, Storable a) => Socket -> Fold m (Array a) ()

280−writeArrays h m = S.mapM_ (liftIO . writeArray h) m

288+writeArrays h = FL.drainBy (liftIO . writeArray h)

281

289

282-- GHC buffer size dEFAULT_FD_BUFFER_SIZE=8192 bytes.

290-- GHC buffer size dEFAULT_FD_BUFFER_SIZE=8192 bytes.

283--

291--

@@ -292,9 +300,17 @@ writeArrays h m = S.mapM_ (liftIO . writeArray h) m

292-- input elements.

300-- input elements.

293--

301--

294-- @since 0.7.0

302-- @since 0.7.0

303+{-# INLINE writeInChunksOfS #-}

304+writeInChunksOfS :: MonadIO m => Int -> Socket -> SerialT m Word8 -> m ()

305+writeInChunksOfS n h m = writeArraysS h $ AS.arraysOf n m

306+

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

308+-- specified number of bytes before writing.

309+--

310+-- @since 0.7.0

295{-# INLINE writeInChunksOf #-}

311{-# INLINE writeInChunksOf #-}

296−writeInChunksOf :: MonadIO m => Int -> Socket -> SerialT m Word8 -> m ()

312+writeInChunksOf :: MonadIO m => Int -> Socket -> Fold m Word8 ()

297−writeInChunksOf n h m = writeArrays h $ AS.arraysOf n m

313+writeInChunksOf n h = FL.lchunksOf n A.write (writeArrays h)

298

314

299-- > write = 'writeInChunksOf' A.defaultChunkSize

315-- > write = 'writeInChunksOf' A.defaultChunkSize

300--

316--

@@ -303,8 +319,20 @@ writeInChunksOf n h m = writeArrays h $ AS.arraysOf n m

303-- depends on the 'IOMode' and the current seek position of the handle.

319-- depends on the 'IOMode' and the current seek position of the handle.

304--

320--

305-- @since 0.7.0

321-- @since 0.7.0

322+{-# INLINE _writeS #-}

323+_writeS :: MonadIO m => Socket -> SerialT m Word8 -> m ()

324+_writeS = writeInChunksOfS A.defaultChunkSize

325+

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

327+-- up to 'A.defaultChunkSize' bytes before writing.

328+--

329+-- @

330+-- write = 'writeInChunksOf' 'A.defaultChunkSize'

331+-- @

332+--

333+-- @since 0.7.0

306{-# INLINE write #-}

334{-# INLINE write #-}

307−write :: MonadIO m => Socket -> SerialT m Word8 -> m ()

335+write :: MonadIO m => Socket -> Fold m Word8 ()

308write = writeInChunksOf A.defaultChunkSize

336write = writeInChunksOf A.defaultChunkSize

309

337

310{-

338{-

src/Streamly/String.hs+2−2

@@ -204,7 +204,7 @@ isSpace c

204-- 'foldLines' instead.

205{-# INLINE lines #-}

206lines :: (MonadIO m, IsStream t) => t m Char -> t m (Array Char)

207−lines = S.splitOnSuffix (== '\n') A.writeF

207+lines = S.splitOnSuffix (== '\n') A.write

208

209-- | Break a string up into a list of strings, which were delimited

210-- by characters representing white space.

@@ -216,7 +216,7 @@ lines = S.splitOnSuffix (== '\n') A.writeF

216-- 'foldWords' instead.

217{-# INLINE words #-}

218words :: (MonadIO m, IsStream t) => t m Char -> t m (Array Char)

219−words = S.wordsBy isSpace A.writeF

219+words = S.wordsBy isSpace A.write

220

221-- | Flattens the stream of @Array Char@, after appending a terminating

222-- newline to each string.

test/Arrays.hs+4−4

@@ -42,7 +42,7 @@ testLength =

42 forAll (choose (0, maxArrLen)) $ \len ->

43 forAll (vectorOf len (arbitrary :: Gen Int)) $ \list ->

44 monadicIO $ do

45− arr <- A.writeN len

45+ arr <- S.runFold (A.writeN len)

46 $ S.fromList list

47 assert (A.length arr == len)

48

@@ -51,7 +51,7 @@ testFromToStreamN =

51 forAll (choose (0, maxArrLen)) $ \len ->

52 forAll (vectorOf len (arbitrary :: Gen Int)) $ \list ->

53 monadicIO $ do

54− arr <- A.writeN len

54+ arr <- S.runFold (A.writeN len)

55 $ S.fromList list

56 xs <- S.toList

57 $ A.read arr

@@ -62,7 +62,7 @@ testToStreamRev =

62 forAll (choose (0, maxArrLen)) $ \len ->

63 forAll (vectorOf len (arbitrary :: Gen Int)) $ \list ->

64 monadicIO $ do

65− arr <- A.writeN len

65+ arr <- S.runFold (A.writeN len)

66 $ S.fromList list

67 xs <- S.toList

68 $ A.readRev arr

@@ -95,7 +95,7 @@ testFromToStream =

95 forAll (choose (0, maxArrLen)) $ \len ->

96 forAll (vectorOf len (arbitrary :: Gen Int)) $ \list ->

97 monadicIO $ do

98− arr <- A.write $ S.fromList list

98+ arr <- S.runFold A.write $ S.fromList list

99 xs <- S.toList

100 $ A.read arr

101 assert (xs == list)