06b0cb1 · streamly · git.iko.soy

git.iko.soy/streamly

Implement IO write operations as folds.

commit06b0cb1328d06204dc2422ef181b397ab34fe7a7 authorHarendra Kumar <harendra.kumar@gmail.com> dateThu 8 Aug 2019, 11:02 (7 years ago) parent3d32be5

16 files changed+269−78

compare with master

unifiedsplit

benchmark/ArrayOps.hs+5−5 benchmark/FileIO.hs+10−10 benchmark/Linear.hs+1−1 examples/EchoServer.hs+2−1 examples/FileSinkServer.hs+2−2 examples/FromFileClient.hs+2−1 examples/HandleIO.hs+2−2 src/Streamly/FileSystem/Handle.hs+54−12 src/Streamly/Fold/Types.hs+22−7 src/Streamly/Internal.hs+2 src/Streamly/Memory/Array.hs+12−12 src/Streamly/Memory/Array/Types.hs+61−8 src/Streamly/Network/Client.hs+50−1 src/Streamly/Network/Socket.hs+38−10 src/Streamly/String.hs+2−2 test/Arrays.hs+4−4

benchmark/ArrayOps.hs+5−5

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

4949

5050{-# INLINE sourceUnfoldr #-}

5151sourceUnfoldr :: 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

5353 where

5454 step cnt =

5555 if cnt > n + value

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

5858

5959{-# INLINE sourceIntFromTo #-}

6060sourceIntFromTo :: 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)

6262

6363{-# INLINE sourceIntFromToFromStream #-}

6464sourceIntFromToFromStream :: 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)

6666

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

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

6969

7070{-# INLINE sourceFromList #-}

7171sourceFromList :: 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]

7373

7474{-# INLINE sourceIsList #-}

7575sourceIsList :: Int -> Stream Int

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

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

281281 -> Stream Int

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

284284

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

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

benchmark/FileIO.hs+10−10

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

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

119119 , mkBench "cat" href $ do

120120 Handles inh _ <- readIORef href

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

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

122122 ]

123123 , bgroup "readStream"

124124 [ mkBench "last" href $ do

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

138138 S.sum $ FH.read inh

139139 , mkBench "cat" href $ do

140140 Handles inh _ <- readIORef href

141− FH.write devNull $ FH.read inh

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

142142 ]

143143 , bgroup "copyArray"

144144 [ mkBench "copy" href $ do

145145 Handles inh outh <- readIORef href

146146 let s = FH.readArrays inh

147− FH.writeArrays outh s

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

148148 ]

149149#ifdef DEVBUILD

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

151151 , bgroup "copyStream"

152152 [ mkBench "fromToHandle" href $ do

153153 Handles inh outh <- readIORef href

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

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

155155 ]

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

157157 , bgroup "decode-encode"

158158 [ mkBench "char8" href $ do

159159 Handles inh outh <- readIORef href

160− FH.write outh

160+ S.runFold (FH.write outh)

161161 $ SS.encodeChar8

162162 $ SS.decodeChar8

163163 $ FH.read inh

164164 , mkBench "utf8" href $ do

165165 Handles inh outh <- readIORef href

166− FH.write outh

166+ S.runFold (FH.write outh)

167167 $ SS.encodeUtf8

168168 $ SS.decodeUtf8

169169 $ FH.read inh

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

171171 , bgroup "grouping"

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

173173 Handles inh _ <- readIORef href

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

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

175175 (FH.read inh)

176176

177177 , mkBench "chunksOf 1" href $ do

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

187187 , bgroup "group-ungroup"

188188 [ mkBench "lines-unlines" href $ do

189189 Handles inh outh <- readIORef href

190− FH.write outh

190+ S.runFold (FH.write outh)

191191 $ SS.encodeChar8

192192 $ SS.unlines

193193 $ SS.lines

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

195195 $ FH.read inh

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

197197 Handles inh outh <- readIORef href

198− FH.writeArraysPackedUpto (1024*1024) outh

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

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

200200 $ AS.splitOn 10

201201 $ FH.readArraysOf (1024*1024) inh

202202 , mkBench "words-unwords" href $ do

203203 Handles inh outh <- readIORef href

204− FH.write outh

204+ S.runFold (FH.write outh)

205205 $ SS.encodeChar8

206206 $ SS.unwords

207207 $ SS.words

benchmark/Linear.hs+1−1

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

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

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

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

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

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

228228

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

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

examples/EchoServer.hs+2−1

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

88main :: IO ()

99main = S.drain

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

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

1111 $ serially $ connectionsOnAllAddrs 8090

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

examples/FileSinkServer.hs+2−2

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

2121main = do

2222 file <- fmap head getArgs

2323 withFile file AppendMode

24− (\src -> FH.write src

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

2525 $ encodeChar8Unchecked

2626 $ S.concatMap A.read

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

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

3030 where

3131

3232 recv =

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

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

3434 . decodeChar8

3535 . NS.read

examples/FromFileClient.hs+2−1

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

1616main =

1717 let sendFile file =

1818 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

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

examples/HandleIO.hs+2−2

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

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

1313

1414cat :: 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

1616-- byte stream version

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

1818

1919cp :: 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

2121-- byte stream version

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

2323

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

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

8686 -- -- * Array Write

8787 , writeArray

8888 , writeArrays

89− , writeArraysPackedUpto

89+ -- , writeArraysPackedUpto

90+ , writeArraysInChunksOf

9091

9192 -- -- * Random Access (Seek)

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

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

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

123124import Streamly.Streams.Serial (SerialT)

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

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

126−-- import Streamly.Fold (Fold)

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

127+import Streamly.Fold (Fold)

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

128129

130+import qualified Streamly.Fold as FL

131+import qualified Streamly.Fold.Types as FL

129132import qualified Streamly.Memory.Array as A

130133import qualified Streamly.Memory.ArrayStream as AS

131134import qualified Streamly.Prelude as S

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

265268-- Writing

266269-------------------------------------------------------------------------------

267270

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+

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

269280--

270281-- @since 0.7.0

271282{-# INLINE writeArrays #-}

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

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

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

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

274285

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

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

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

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

279290--

280291-- @since 0.7.0

281−{-# INLINE writeArraysPackedUpto #-}

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

292+{-# INLINE _writeArraysPackedUpto #-}

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

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

285308

286309-- GHC buffer size dEFAULT_FD_BUFFER_SIZE=8192 bytes.

287310--

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

296319-- collect the required input size.

297320--

298321-- @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

299331{-# INLINE writeInChunksOf #-}

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

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

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

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

302344

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

304346--

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

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

307349--

308350-- @since 0.7.0

309351{-# INLINE write #-}

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

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

311353write = writeInChunksOf defaultChunkSize

312354

313355{-

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

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

2323 , ltakeWhile

2424 , lsessionsOf

2525 , lchunksOf

26+

2627 , duplicate

28+ , initialize

29+ , runStep

2730 )

2831where

2932

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

318321duplicate (Fold step begin done) =

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

320323

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

322−-- group using the provided fold function.

323−--

324−-- @

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

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

325326--

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

327−--

328−-- @

327+{-# INLINABLE initialize #-}

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

329+initialize (Fold step initial extract) = do

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.

329344--

330345{-# INLINE lchunksOf #-}

331346lchunksOf :: 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

6464 , lsessionsOf

6565 , lchunksOf

6666 , duplicate

67+ , initialize

68+ , runStep

6769 )

6870where

6971

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

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

7979

8080 -- Monadic APIs

8181 -- , newArray

82− , writeN

82+ , A.writeN

8383 , write

8484

8585 -- Stream Folds

86− , A.writeNF

87− , writeF

86+ -- , writeNS

87+ -- , writeS

8888

8989 -- * Elimination

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

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

157157-- array may hold less than N elements.

158158--

159159-- @since 0.7.0

160−{-# INLINE writeN #-}

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

162−writeN n m = do

160+{-# INLINE _writeNS #-}

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

162+_writeNS n m = do

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

164164 A.fromStreamDN n $ D.toStreamD m

165165

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

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

205205--

206206-- @since 0.7.0

207−{-# INLINE writeF #-}

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

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

207+{-# INLINE write #-}

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

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

210210

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

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

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

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

218218--

219−{-# INLINE write #-}

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

221−write = P.runFold writeF

219+{-# INLINE _writeS #-}

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

221+_writeS = P.runFold write

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

223223

224224-------------------------------------------------------------------------------

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

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

3939 , flattenArrays

4040 , flattenArraysRev

4141 , packArraysChunksOf

42+ , lpackArraysChunksOf

4243#if !defined(mingw32_HOST_OS)

4344 , groupIOVecsOf

4445#endif

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

5960 , toStreamK

6061 , toStreamKRev

6162 , toList

62− , writeNF

63+ , writeN

6364 , read

6465

6566 -- * Utilities

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

99100import GHC.Ptr (Ptr(..))

100101

101102import Streamly.Fold.Types (Fold(..))

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

102104import Streamly.SVar (adaptState)

103105

104106#if !defined(mingw32_HOST_OS)

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

472474-- Instances

473475-------------------------------------------------------------------------------

474476

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

476478-- 'Array'.

477479--

478480-- @since 0.7.0

479−{-# INLINE_NORMAL writeNF #-}

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

481−writeNF n = Fold step initial extract

481+{-# INLINE_NORMAL writeN #-}

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

483+writeN n = Fold step initial extract

482484

483485 where

484486

485487 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 ()

487489 liftIO $ newArray n

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

489491 step (Array start end bound) x = do

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

516518{-# INLINE_NORMAL fromStreamDArraysOf #-}

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

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

520522fromStreamDArraysOf n (D.Stream step state) =

521523 D.Stream step' (GroupStart state)

522524

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

887889

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

889891-- | 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.

891896--

892897-- @since 0.7.0

893898{-# INLINE_NORMAL packArraysChunksOf #-}

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

935940

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

937942

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+

938991#if !defined(mingw32_HOST_OS)

939992data GatherState s arr

940993 = GatherInitial s

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

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

6666import qualified Network.Socket as Net

6767

6868import Streamly (MonadAsync)

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

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

70−import Streamly.Streams.Serial (SerialT)

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

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

7273

74+import qualified Streamly.Fold.Types as FL

75+import qualified Streamly.Memory.Array as A

7376import qualified Streamly.Memory.ArrayStream as AS

7477import qualified Streamly.Prelude as S

7578import qualified Streamly.Network.Socket as SK

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

114117-- Writing

115118-------------------------------------------------------------------------------

116119

120+{-

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

118122-- number.

119123--

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

127131 -> m ()

128132writeArrays addr port xs =

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

134+-}

130135

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

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

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

133157-- input elements.

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

142166 -> SerialT m Word8

143167 -> m ()

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

169+-}

145170

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

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

147187--

148188-- @since 0.7.0

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

150190write :: (MonadCatch m, MonadAsync m)

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

152192write = 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

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

9090import Streamly.Streams.Serial (SerialT)

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

92−-- import Streamly.Fold (Fold)

92+import Streamly.Fold (Fold)

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

9494

95+import qualified Streamly.Fold as FL

96+import qualified Streamly.Fold.Types as FL

9597import qualified Streamly.Memory.Array as A

9698import qualified Streamly.Memory.ArrayStream as AS

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

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

228230

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

230232--

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

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

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

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

234−-- on the 'Handle'.

235235--

236236-- @since 0.7.0

237237{-# INLINE readArrays #-}

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

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

261261--

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

264264-- stream ends when EOF is encountered.

265265--

266266-- @since 0.7.0

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

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

276276--

277277-- @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

278286{-# INLINE writeArrays #-}

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

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

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

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

281289

282290-- GHC buffer size dEFAULT_FD_BUFFER_SIZE=8192 bytes.

283291--

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

292300-- input elements.

293301--

294302-- @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

295311{-# INLINE writeInChunksOf #-}

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

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

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

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

298314

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

300316--

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

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

304320--

305321-- @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

306334{-# INLINE write #-}

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

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

308336write = writeInChunksOf A.defaultChunkSize

309337

310338{-

src/Streamly/String.hs+2−2

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

204204-- 'foldLines' instead.

205205{-# INLINE lines #-}

206206lines :: (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

208208

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

210210-- by characters representing white space.

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

216216-- 'foldWords' instead.

217217{-# INLINE words #-}

218218words :: (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

220220

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

222222-- newline to each string.

test/Arrays.hs+4−4

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

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

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

4444 monadicIO $ do

45− arr <- A.writeN len

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

4646 $ S.fromList list

4747 assert (A.length arr == len)

4848

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

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

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

5353 monadicIO $ do

54− arr <- A.writeN len

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

5555 $ S.fromList list

5656 xs <- S.toList

5757 $ A.read arr

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

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

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

6464 monadicIO $ do

65− arr <- A.writeN len

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

6666 $ S.fromList list

6767 xs <- S.toList

6868 $ A.readRev arr

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

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

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

9797 monadicIO $ do

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

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

9999 xs <- S.toList

100100 $ A.read arr

101101 assert (xs == list)