3cdecba · streamly · git.iko.soy

Changelog.md+2

@@ -83,6 +83,8 @@ example, concat streams concurrently using this.

8383 character streams and other character stream operations.

8484 * _Arrays_: `Streamly.Mem.Array` module provides arrays for efficient

8585 in-memory buffering and efficient interfacing with IO.

86+ * `Streamly.Mem.Array.Stream` module provide combinators to work on streams

87+ of arrays.

8688

8789* Add the following to `Streamly.Prelude`:

8890 * `foldrS` to fold a stream to a stream

README.md+6−8

@@ -409,6 +409,7 @@ module Main where

409409import qualified Streamly.Prelude as S

410410import qualified Streamly.Fold as FL

411411import qualified Streamly.Mem.Array as A

412+import qualified Streamly.Mem.Array.Stream as AS

412413import qualified Streamly.FileSystem.Handle as FH

413414import qualified System.IO as FH

414415

@@ -431,24 +432,21 @@ withArg2 f = do

431432### cat

432433

433434``` haskell

434−cat = FH.writeArrays stdout . FH.readArraysOfUpto (256*1024)

435+cat = FH.writeArrays stdout . FH.readArrays

435436main = withArg cat

436437```

437438

438439### cp

439440

440441``` haskell

441−cp src dst = FH.writeArrays dst $ FH.readArraysOfUpto (256*1024) src

442+cp src dst = FH.writeArrays dst $ FH.readArrays src

442443main = withArg2 cp

443444```

444445

445446### wc -l

446447

447448``` haskell

448−wcl =

449− S.length

450− . FL.splitBySuffix (== fromIntegral (ord '\n')) FL.drain

451− . FH.read

449+wcl = S.length . AS.splitOn 10 . FH.readArrays

452450main = withArg wcl >>= print

453451```

454452

@@ -457,7 +455,7 @@ main = withArg wcl >>= print

457455``` haskell

458456avgll =

459457 FL.foldl' avg

460− . FL.splitBySuffix (== fromIntegral (ord '\n')) FL.length

458+ . FL.splitBySuffix (== 10) FL.length

461459 . FH.read

462460

463461 where avg = (/) <$> toDouble FL.sum <*> toDouble FL.length

@@ -472,7 +470,7 @@ main = withArg avgll >>= print

472470llhisto =

473471 FL.foldl' (FL.classify FL.length)

474472 . S.map bucket

475− . FL.splitBySuffix (== fromIntegral (ord '\n')) FL.length

473+ . FL.splitBySuffix (== 10) FL.length

476474 . FH.read

477475

478476 where

benchmark/FileIO.hs+38−33

@@ -17,6 +17,7 @@ import Gauge

1717

1818import qualified Streamly.FileSystem.Handle as FH

1919import qualified Streamly.Mem.Array as A

20+import qualified Streamly.Mem.Array.Stream as AS

2021import qualified Streamly.Prelude as S

2122import qualified Streamly.Fold as FL

2223import qualified Streamly.String as SS

@@ -108,7 +109,7 @@ main = do

108109 S.sum (S.map A.length s)

109110 , mkBench "linecount" href $ do

110111 Handles inh _ <- readIORef href

111− S.length $ A.splitArraysOn 10 $ FH.readArrays inh

112+ S.length $ AS.splitOn 10 $ FH.readArrays inh

112113 , mkBench "sum" href $ do

113114 let foldlArr' f z = runIdentity . S.foldl' f z . A.read

114115 Handles inh _ <- readIORef href

@@ -191,12 +192,14 @@ main = do

191192 $ SS.lines

192193 $ SS.decodeChar8

193194 $ FH.read inh

195+ {-

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

195197 Handles inh outh <- readIORef href

196198 FH.writeArraysPackedUpto (1024*1024) outh

197199 $ S.insertAfterEach (return $ A.fromList [10])

198− $ A.splitArraysOn 10

200+ $ AS.splitOn 10

199201 $ FH.readArraysOfUpto (1024*1024) inh

202+ -}

200203 , mkBench "words-unwords" href $ do

201204 Handles inh outh <- readIORef href

202205 FH.write outh

@@ -208,18 +211,18 @@ main = do

208211 ]

209212

210213 , let lf = fromIntegral (ord '\n')

211− lfarr = A.fromList [lf]

214+ -- lfarr = A.fromList [lf]

212215 isSp = isSpace . chr . fromIntegral

213− toarr = A.fromList . map (fromIntegral . ord)

216+ -- toarr = A.fromList . map (fromIntegral . ord)

214217 in bgroup "splitting"

215218 [ bgroup "predicate"

216− [ mkBench "splitBy \\n (line count)" href $ do

219+ [ mkBench "splitOn \\n (line count)" href $ do

217220 Handles inh _ <- readIORef href

218− (S.length $ FL.splitBy (== lf) FL.drain

221+ (S.length $ FL.splitOn (== lf) FL.drain

219222 $ FH.read inh) -- >>= print

220− , mkBench "splitSuffixBy \\n (line count)" href $ do

223+ , mkBench "splitOnSuffix \\n (line count)" href $ do

221224 Handles inh _ <- readIORef href

222− (S.length $ FL.splitSuffixBy (== lf) FL.drain

225+ (S.length $ FL.splitOnSuffix (== lf) FL.drain

223226 $ FH.read inh) -- >>= print

224227 , mkBench "wordsBy isSpace (word count)" href $ do

225228 Handles inh _ <- readIORef href

@@ -227,72 +230,74 @@ main = do

227230 $ FH.read inh) -- >>= print

228231 ]

229232

233+ {-

230234 , bgroup "empty-pattern"

231− [ mkBench "splitOn \"\"" href $ do

235+ [ mkBench "splitOnSeq \"\"" href $ do

232236 Handles inh _ <- readIORef href

233− (S.length $ FL.splitOn (A.fromList []) FL.drain

237+ (S.length $ FL.splitOnSeq (A.fromList []) FL.drain

234238 $ FH.read inh) -- >>= print

235− , mkBench "splitSuffixOn \"\"" href $ do

239+ , mkBench "splitOnSuffixSeq \"\"" href $ do

236240 Handles inh _ <- readIORef href

237− (S.length $ FL.splitSuffixOn (A.fromList []) FL.drain

241+ (S.length $ FL.splitOnSuffixSeq (A.fromList []) FL.drain

238242 $ FH.read inh) -- >>= print

239243 ]

240244 , bgroup "short-pattern"

241− [ mkBench "splitOn \\n (line count)" href $ do

245+ [ mkBench "splitOnSeq \\n (line count)" href $ do

242246 Handles inh _ <- readIORef href

243247 (S.length $ FL.splitOn lfarr FL.drain

244248 $ FH.read inh) -- >>= print

245− , mkBench "splitSuffixOn \\n (line count)" href $ do

249+ , mkBench "splitOnSuffixSeq \\n (line count)" href $ do

246250 Handles inh _ <- readIORef href

247− (S.length $ FL.splitSuffixOn lfarr FL.drain

251+ (S.length $ FL.splitOnSuffixSeq lfarr FL.drain

248252 $ FH.read inh) -- >>= print

249− , mkBench "splitOn a" href $ do

253+ , mkBench "splitOnSeq a" href $ do

250254 Handles inh _ <- readIORef href

251255 (S.length $ FL.splitOn (toarr "a") FL.drain

252256 $ FH.read inh) -- >>= print

253− , mkBench "splitOn \\r\\n" href $ do

257+ , mkBench "splitOnSeq \\r\\n" href $ do

254258 Handles inh _ <- readIORef href

255259 (S.length $ FL.splitOn (toarr "\r\n") FL.drain

256260 $ FH.read inh) -- >>= print

257− , mkBench "splitSuffixOn \\r\\n)" href $ do

261+ , mkBench "splitOnSuffixSeq \\r\\n)" href $ do

258262 Handles inh _ <- readIORef href

259− (S.length $ FL.splitSuffixOn (toarr "\r\n") FL.drain

263+ (S.length $ FL.splitOnSuffixSeq (toarr "\r\n") FL.drain

260264 $ FH.read inh) -- >>= print

261− , mkBench "splitOn aa" href $ do

265+ , mkBench "splitOnSeq aa" href $ do

262266 Handles inh _ <- readIORef href

263− (S.length $ FL.splitOn (toarr "aa") FL.drain

267+ (S.length $ FL.splitOnSeq (toarr "aa") FL.drain

264268 $ FH.read inh) -- >>= print

265− , mkBench "splitOn aaaa" href $ do

269+ , mkBench "splitOnSeq aaaa" href $ do

266270 Handles inh _ <- readIORef href

267− (S.length $ FL.splitOn (toarr "aaaa") FL.drain

271+ (S.length $ FL.splitOnSeq (toarr "aaaa") FL.drain

268272 $ FH.read inh) -- >>= print

269− , mkBench "splitOn abcdefgh" href $ do

273+ , mkBench "splitOnSeq abcdefgh" href $ do

270274 Handles inh _ <- readIORef href

271− (S.length $ FL.splitOn (toarr "abcdefgh") FL.drain

275+ (S.length $ FL.splitOnSeq (toarr "abcdefgh") FL.drain

272276 $ FH.read inh) -- >>= print

273277 ]

274278 , bgroup "long-pattern"

275− [ mkBench "splitOn abcdefghi" href $ do

279+ [ mkBench "splitOnSeq abcdefghi" href $ do

276280 Handles inh _ <- readIORef href

277− (S.length $ FL.splitOn (toarr "abcdefghi") FL.drain

281+ (S.length $ FL.splitOnSeq (toarr "abcdefghi") FL.drain

278282 $ FH.read inh) -- >>= print

279− , mkBench "splitOn catcatcatcatcat" href $ do

283+ , mkBench "splitOnSeq catcatcatcatcat" href $ do

280284 Handles inh _ <- readIORef href

281− (S.length $ FL.splitOn (toarr "catcatcatcatcat") FL.drain

285+ (S.length $ FL.splitOnSeq (toarr "catcatcatcatcat") FL.drain

282286 $ FH.read inh) -- >>= print

283− , mkBench "splitOn abc...xyz" href $ do

287+ , mkBench "splitOnSeq abc...xyz" href $ do

284288 Handles inh _ <- readIORef href

285− (S.length $ FL.splitOn

289+ (S.length $ FL.splitOnSeq

286290 (toarr "abcdefghijklmnopqrstuvwxyz")

287291 FL.drain

288292 $ FH.read inh) -- >>= print

289− , mkBench "splitSuffixOn abc...xyz" href $ do

293+ , mkBench "splitOnSuffixSeq abc...xyz" href $ do

290294 Handles inh _ <- readIORef href

291− (S.length $ FL.splitSuffixOn

295+ (S.length $ FL.splitOnSuffixSeq

292296 (toarr "abcdefghijklmnopqrstuvwxyz")

293297 FL.drain

294298 $ FH.read inh) -- >>= print

295299 ]

300+ -}

296301 ]

297302#endif

298303 ]

examples/HandleIO.hs+3−2

@@ -1,6 +1,7 @@

11import qualified Streamly.Prelude as S

22import qualified Streamly.Fold as FL

3−import qualified Streamly.Mem.Array as A

3+-- import qualified Streamly.Mem.Array as A

4+import qualified Streamly.Mem.Array.Stream as AS

45import qualified Streamly.FileSystem.Handle as FH

56import qualified System.IO as FH

67-- import qualified Streamly.FileSystem.FD as FH

@@ -25,7 +26,7 @@ ord' = (fromIntegral . ord)

2526

2627wcl :: FH.Handle -> IO ()

2728wcl src = print =<< (S.length

28− $ A.splitArraysOn 10

29+ $ AS.splitOn 10

2930 $ FH.readArrays src)

3031{-

3132-- Char stream version

src/Streamly/FileSystem/FD.hs+5−4

@@ -151,6 +151,7 @@ import qualified Streamly.FileSystem.FDIO as RawIO hiding (write)

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

152152

153153import qualified Streamly.Mem.Array as A

154+import qualified Streamly.Mem.Array.Stream as AS

154155import qualified Streamly.Prelude as S

155156import qualified Streamly.Streams.StreamD.Type as D

156157

@@ -326,7 +327,7 @@ readArrays = readArraysOfUpto defaultChunkSize

326327--

327328{-# INLINE readByChunksUpto #-}

328329readByChunksUpto :: (IsStream t, MonadIO m) => Int -> Handle -> t m Word8

329−readByChunksUpto chunkSize h = A.flattenArrays $ readArraysOfUpto chunkSize h

330+readByChunksUpto chunkSize h = AS.flatten $ readArraysOfUpto chunkSize h

330331

331332-- TODO

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

@@ -338,7 +339,7 @@ readByChunksUpto chunkSize h = A.flattenArrays $ readArraysOfUpto chunkSize h

338339-- @since 0.7.0

339340{-# INLINE read #-}

340341read :: (IsStream t, MonadIO m) => Handle -> t m Word8

341−read = A.flattenArrays . readArrays

342+read = AS.flatten . readArrays

342343

343344-------------------------------------------------------------------------------

344345-- Writing

@@ -360,7 +361,7 @@ writeArrays h m = S.mapM_ (liftIO . writeArray h) m

360361{-# INLINE writeArraysPackedUpto #-}

361362writeArraysPackedUpto :: (MonadIO m, Storable a)

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

363−writeArraysPackedUpto n h xs = writeArrays h $ A.packArraysChunksOf n xs

364+writeArraysPackedUpto n h xs = writeArrays h $ AS.compact n xs

364365

365366#if !defined(mingw32_HOST_OS)

366367-- XXX this is incomplete

@@ -400,7 +401,7 @@ _writevArraysPackedUpto n h xs =

400401-- @since 0.7.0

401402{-# INLINE writeByChunksOf #-}

402403writeByChunksOf :: MonadIO m => Int -> Handle -> SerialT m Word8 -> m ()

403−writeByChunksOf n h m = writeArrays h $ A.arraysOf n m

404+writeByChunksOf n h m = writeArrays h $ AS.arraysOf n m

404405

405406-- > write = 'writeByChunks' A.defaultChunkSize

406407--

src/Streamly/FileSystem/File.hs+4−4

@@ -123,7 +123,7 @@ import Streamly.SVar (MonadAsync)

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

124124

125125import qualified Streamly.FileSystem.Handle as FH

126−import qualified Streamly.Mem.Array as A

126+import qualified Streamly.Mem.Array.Stream as AS

127127import qualified Streamly.Mem.Array.Types as A hiding (flattenArrays)

128128import qualified Streamly.Prelude as S

129129

@@ -233,7 +233,7 @@ readByChunksUpto chunkSize h = A.flattenArrays $ readArraysOfUpto chunkSize h

233233-- @since 0.7.0

234234{-# INLINE read #-}

235235read :: (IsStream t, MonadCatch m, MonadIO m) => FilePath -> t m Word8

236−read file = A.flattenArrays $ withFile file ReadMode FH.readArrays

236+read file = AS.flatten $ withFile file ReadMode FH.readArrays

237237

238238{-

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

@@ -289,7 +289,7 @@ writeArrays = writeArraysMode WriteMode

289289{-# INLINE writeByChunks #-}

290290writeByChunks :: (MonadAsync m, MonadCatch m)

291291 => Int -> FilePath -> SerialT m Word8 -> m ()

292−writeByChunks n file xs = writeArrays file $ A.arraysOf n xs

292+writeByChunks n file xs = writeArrays file $ AS.arraysOf n xs

293293

294294-- > write = 'writeByChunks' A.defaultChunkSize

295295--

@@ -325,7 +325,7 @@ appendArrays = writeArraysMode AppendMode

325325{-# INLINE appendByChunks #-}

326326appendByChunks :: (MonadAsync m, MonadCatch m)

327327 => Int -> FilePath -> SerialT m Word8 -> m ()

328−appendByChunks n file xs = appendArrays file $ A.arraysOf n xs

328+appendByChunks n file xs = appendArrays file $ AS.arraysOf n xs

329329

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

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

src/Streamly/FileSystem/Handle.hs+5−4

@@ -127,6 +127,7 @@ import Streamly.Mem.Array.Types (defaultChunkSize, shrinkToFit)

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

128128

129129import qualified Streamly.Mem.Array as A

130+import qualified Streamly.Mem.Array.Stream as AS

130131import qualified Streamly.Prelude as S

131132import qualified Streamly.Streams.StreamD.Type as D

132133

@@ -244,7 +245,7 @@ readArrays = readArraysOfUpto defaultChunkSize

244245-- @since 0.7.0

245246{-# INLINE readByChunksUpto #-}

246247readByChunksUpto :: (IsStream t, MonadIO m) => Int -> Handle -> t m Word8

247−readByChunksUpto chunkSize h = A.flattenArrays $ readArraysOfUpto chunkSize h

248+readByChunksUpto chunkSize h = AS.flatten $ readArraysOfUpto chunkSize h

248249

249250-- TODO

250251-- Generate a stream of elements of the given type from a file 'Handle'.

@@ -256,7 +257,7 @@ readByChunksUpto chunkSize h = A.flattenArrays $ readArraysOfUpto chunkSize h

256257-- @since 0.7.0

257258{-# INLINE read #-}

258259read :: (IsStream t, MonadIO m) => Handle -> t m Word8

259−read = A.flattenArrays . readArrays

260+read = AS.flatten . readArrays

260261

261262-------------------------------------------------------------------------------

262263-- Writing

@@ -278,7 +279,7 @@ writeArrays h m = S.mapM_ (liftIO . writeArray h) m

278279{-# INLINE writeArraysPackedUpto #-}

279280writeArraysPackedUpto :: (MonadIO m, Storable a)

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

281−writeArraysPackedUpto n h xs = writeArrays h $ A.packArraysChunksOf n xs

282+writeArraysPackedUpto n h xs = writeArrays h $ AS.compact n xs

282283

283284-- GHC buffer size dEFAULT_FD_BUFFER_SIZE=8192 bytes.

284285--

@@ -295,7 +296,7 @@ writeArraysPackedUpto n h xs = writeArrays h $ A.packArraysChunksOf n xs

295296-- @since 0.7.0

296297{-# INLINE writeByChunksOf #-}

297298writeByChunksOf :: MonadIO m => Int -> Handle -> SerialT m Word8 -> m ()

298−writeByChunksOf n h m = writeArrays h $ A.arraysOf n m

299+writeByChunksOf n h m = writeArrays h $ AS.arraysOf n m

299300

300301-- > write = 'writeByChunksOf' A.defaultChunkSize

301302--

src/Streamly/Mem/Array.hs+7−143

@@ -122,23 +122,12 @@ module Streamly.Mem.Array

122122 , readSliceRev

123123 -}

124124

125− , writeIndex

125+ -- , writeIndex

126126 {-

127127 , writeSlice

128128 , writeSliceRev

129129 -}

130130

131− -- * Streams of arrays

132− , arraysOf

133−

134− -- Streams of arrays

135− , flattenArrays

136− -- , flattenArraysRev

137− , packArrays

138− , packArraysChunksOf

139− , unlinesArraysBy

140− , splitArraysOn

141−

142131 -- * Immutable Transformations

143132 , transformWith

144133

@@ -151,10 +140,9 @@ where

151140import Control.Monad (when)

152141import Control.Monad.IO.Class (MonadIO(..))

153142-- import Data.Functor.Identity (Identity)

154−import Data.Word (Word8)

155143import Foreign.ForeignPtr (withForeignPtr)

156144import Foreign.ForeignPtr.Unsafe (unsafeForeignPtrToPtr)

157−import Foreign.Ptr (minusPtr, plusPtr, castPtr)

145+import Foreign.Ptr (minusPtr, plusPtr)

158146import Foreign.Storable (Storable(..))

159147import Prelude hiding (length, null, last, map, (!!), read)

160148

@@ -165,9 +153,7 @@ import Streamly.Fold.Types (Fold(..))

165153

166154import qualified Streamly.Fold as FL

167155import qualified Streamly.Mem.Array.Types as A

168−import qualified Streamly.Prelude as S

169156import qualified Streamly.Streams.StreamD as D

170−import qualified Streamly.Streams.Prelude as P

171157

172158-------------------------------------------------------------------------------

173159-- Construction

@@ -351,9 +337,9 @@ readSliceRev arr i len = undefined

351337-- Performs in-place mutation of the array.

352338--

353339-- @since 0.7.0

354−{-# INLINE writeIndex #-}

355−writeIndex :: (MonadIO m, Storable a) => Array a -> Int -> a -> m ()

356−writeIndex arr i a = do

340+{-# INLINE _writeIndex #-}

341+_writeIndex :: (MonadIO m, Storable a) => Array a -> Int -> a -> m ()

342+_writeIndex arr i a = do

357343 let maxIndex = length arr - 1

358344 if i < 0

359345 then error "writeIndex: negative array index"

@@ -422,12 +408,6 @@ _toArraysOf n = FL.lchunksOf n (A.toArrayN n) FL.toStream

422408-- XXX check if GHC's memory allocator is efficient enough. We can try the C

423409-- malloc to compare against.

424410

425−{-# INLINE bytesToCount #-}

426−bytesToCount :: Storable a => a -> Int -> Int

427−bytesToCount x n =

428− let elemSize = sizeOf x

429− in n + elemSize - 1 `div` elemSize

430−

431411{-# INLINE_NORMAL toArrayMinChunk #-}

432412toArrayMinChunk :: forall m a. (MonadIO m, Storable a)

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

@@ -459,124 +439,8 @@ toArrayMinChunk elemCount = Fold step initial extract

459439-- @since 0.7.0

460440{-# INLINE toArray #-}

461441toArray :: forall m a. (MonadIO m, Storable a) => Fold m a (Array a)

462−toArray = toArrayMinChunk (bytesToCount (undefined :: a) (A.mkChunkSize 1024))

463−

464−-- | Convert a stream of arrays into a stream of their elements.

465−--

466−-- Same as the following but more efficient:

467−--

468−-- > flattenArrays = S.concatMap A.read

469−--

470−-- @since 0.7.0

471−{-# INLINE flattenArrays #-}

472−flattenArrays :: (IsStream t, MonadIO m, Storable a) => t m (Array a) -> t m a

473−flattenArrays m = D.fromStreamD $ A.flattenArrays (D.toStreamD m)

474−

475−-- XXX should we have a reverseArrays API to reverse the stream of arrays

476−-- instead?

477−--

478−-- | Convert a stream of arrays into a stream of their elements reversing the

479−-- contents of each array before flattening.

480−--

481−-- @since 0.7.0

482−{-# INLINE _flattenArraysRev #-}

483−_flattenArraysRev :: (IsStream t, MonadIO m, Storable a) => t m (Array a) -> t m a

484−_flattenArraysRev m = D.fromStreamD $ A.flattenArraysRev (D.toStreamD m)

485−

486−-- XXX use an Array instead as separator? Or use a separate unlinesArraysBySeq

487−-- API for that?

488−--

489−-- | Flatten a stream of arrays appending the given element after each

490−-- array.

491−--

492−-- @since 0.7.0

493−{-# INLINE unlinesArraysBy #-}

494−unlinesArraysBy :: (MonadIO m, IsStream t, Storable a)

495− => a -> t m (Array a) -> t m a

496−unlinesArraysBy x = D.fromStreamD . A.unlines x . D.toStreamD

497−

498−-- | Split a stream of arrays on a given separator byte, dropping the separator

499−-- and coalescing all the arrays between two separators into a single array.

500−--

501−-- @since 0.7.0

502−{-# INLINE splitArraysOn #-}

503−splitArraysOn

504− :: (IsStream t, MonadIO m)

505− => Word8

506− -> t m (Array Word8)

507− -> t m (Array Word8)

508−splitArraysOn byte s = D.fromStreamD $ A.splitOn byte $ D.toStreamD s

509−

510−-- | Coalesce adajcent arrays in incoming stream to form bigger arrays of a

511−-- maximum specified size.

512−--

513−-- @since 0.7.0

514−{-# INLINE packArraysChunksOf #-}

515−packArraysChunksOf :: (MonadIO m, Storable a)

516− => Int -> SerialT m (Array a) -> SerialT m (Array a)

517−packArraysChunksOf n xs =

518− D.fromStreamD $ A.packArraysChunksOf n (D.toStreamD xs)

519−

520−-- | Groups the elements in an input stream into arrays of given size.

521−--

522−-- Same as the following but more efficient:

523−--

524−-- > arraysOf n = FL.groupsOf n (FL.toArrayN n)

525−--

526−-- @since 0.7.0

527−{-# INLINE arraysOf #-}

528−arraysOf :: (IsStream t, MonadIO m, Storable a)

529− => Int -> t m a -> t m (Array a)

530−arraysOf n str =

531− D.fromStreamD $ A.fromStreamDArraysOf n (D.toStreamD str)

532−

533−-- XXX Both of these implementations of splicing seem to perform equally well.

534−-- We need to perform benchmarks over a range of sizes though.

535−

536−-- CAUTION! length must more than equal to lengths of all the arrays in the

537−-- stream.

538−{-# INLINE spliceArraysLenUnsafe #-}

539−spliceArraysLenUnsafe :: (MonadIO m, Storable a)

540− => Int -> SerialT m (Array a) -> m (Array a)

541−spliceArraysLenUnsafe len buffered = do

542− arr <- liftIO $ A.newArray len

543− end <- S.foldlM' writeArr (aEnd arr) buffered

544− return $ arr {aEnd = end}

545−

546− where

547−

548− writeArr dst Array{..} =

549− liftIO $ withForeignPtr aStart $ \src -> do

550− let count = aEnd `minusPtr` src

551− A.memcpy (castPtr dst) (castPtr src) count

552− return $ dst `plusPtr` count

553−

554−{-# INLINE _spliceArraysBuffered #-}

555−_spliceArraysBuffered :: (MonadIO m, Storable a)

556− => SerialT m (Array a) -> m (Array a)

557−_spliceArraysBuffered s = do

558− buffered <- P.foldr S.cons S.nil s

559− len <- S.sum (S.map length buffered)

560− spliceArraysLenUnsafe len s

561−

562−{-# INLINE spliceArraysRealloced #-}

563−spliceArraysRealloced :: forall m a. (MonadIO m, Storable a)

564− => SerialT m (Array a) -> m (Array a)

565−spliceArraysRealloced s = do

566− idst <- liftIO $ A.newArray (bytesToCount (undefined :: a)

567− (A.mkChunkSizeKB 4))

568−

569− arr <- S.foldlM' A.spliceWithDoubling idst s

570− liftIO $ A.shrinkToFit arr

571−

572−-- | Given a stream of arrays, splice them all together to generate a single

573−-- array. The stream must be /finite/.

574−--

575−-- @since 0.7.0

576−{-# INLINE packArrays #-}

577−packArrays :: (MonadIO m, Storable a) => SerialT m (Array a) -> m (Array a)

578−packArrays = spliceArraysRealloced

579−-- spliceArrays = _spliceArraysBuffered

442+toArray =

443+ toArrayMinChunk (A.bytesToCount (undefined :: a) (A.mkChunkSize 1024))

580444

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

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

src/Streamly/Mem/Array/Stream.hsnew file+173

@@ -0,0 +1,173 @@

1+{-# LANGUAGE BangPatterns #-}

2+{-# LANGUAGE CPP #-}

3+{-# LANGUAGE MagicHash #-}

4+{-# LANGUAGE RecordWildCards #-}

5+{-# LANGUAGE UnboxedTuples #-}

6+{-# LANGUAGE ScopedTypeVariables #-}

7+

8+#include "inline.hs"

9+

10+-- |

11+-- Module : Streamly.Mem.Array.Stream

13+--

14+-- License : BSD3

15+-- Maintainer : harendra.kumar@gmail.com

16+-- Stability : experimental

17+-- Portability : GHC

18+--

19+-- Combinators to efficiently manipulate streams of arrays.

20+--

21+module Streamly.Mem.Array.Stream

22+ (

23+ -- * Creation

24+ arraysOf

25+

26+ -- * Flattening to elements

27+ , flatten

28+ -- , _flattenArraysRev

29+ , unlinesBy

30+

31+ -- * Transformation

32+ , splitOn

33+ , compact -- compact

34+

35+ -- * Elimination

36+ , toArray

37+ )

38+where

39+

40+import Control.Monad.IO.Class (MonadIO(..))

41+-- import Data.Functor.Identity (Identity)

42+import Data.Word (Word8)

43+import Foreign.ForeignPtr (withForeignPtr)

44+import Foreign.Ptr (minusPtr, plusPtr, castPtr)

45+import Foreign.Storable (Storable(..))

46+import Prelude hiding (length, null, last, map, (!!), read)

47+

48+import Streamly.Mem.Array.Types (Array(..), length)

49+import Streamly.Streams.Serial (SerialT)

50+import Streamly.Streams.StreamK.Type (IsStream)

51+

52+import qualified Streamly.Mem.Array.Types as A

53+import qualified Streamly.Prelude as S

54+import qualified Streamly.Streams.StreamD as D

55+import qualified Streamly.Streams.Prelude as P

56+

57+-- | Convert a stream of arrays into a stream of their elements.

58+--

59+-- Same as the following but more efficient:

60+--

61+-- > flatten = S.concatMap A.read

62+--

63+-- @since 0.7.0

64+{-# INLINE flatten #-}

65+flatten :: (IsStream t, MonadIO m, Storable a) => t m (Array a) -> t m a

66+flatten m = D.fromStreamD $ A.flattenArrays (D.toStreamD m)

67+

68+-- XXX should we have a reverseArrays API to reverse the stream of arrays

69+-- instead?

70+--

71+-- | Convert a stream of arrays into a stream of their elements reversing the

72+-- contents of each array before flattening.

73+--

74+-- @since 0.7.0

75+{-# INLINE _flattenArraysRev #-}

76+_flattenArraysRev :: (IsStream t, MonadIO m, Storable a) => t m (Array a) -> t m a

77+_flattenArraysRev m = D.fromStreamD $ A.flattenArraysRev (D.toStreamD m)

78+

79+-- XXX use an Array instead as separator? Or use a separate unlinesArraysBySeq

80+-- API for that?

81+--

82+-- | Flatten a stream of arrays appending the given element after each

83+-- array.

84+--

85+-- @since 0.7.0

86+{-# INLINE unlinesBy #-}

87+unlinesBy :: (MonadIO m, IsStream t, Storable a)

88+ => a -> t m (Array a) -> t m a

89+unlinesBy x = D.fromStreamD . A.unlines x . D.toStreamD

90+

91+-- | Split a stream of arrays on a given separator byte, dropping the separator

92+-- and coalescing all the arrays between two separators into a single array.

93+--

94+-- @since 0.7.0

95+{-# INLINE splitOn #-}

96+splitOn

97+ :: (IsStream t, MonadIO m)

98+ => Word8

99+ -> t m (Array Word8)

100+ -> t m (Array Word8)

101+splitOn byte s = D.fromStreamD $ A.splitOn byte $ D.toStreamD s

102+

103+-- | Coalesce adajcent arrays in incoming stream to form bigger arrays of a

104+-- maximum specified size.

105+--

106+-- @since 0.7.0

107+{-# INLINE compact #-}

108+compact :: (MonadIO m, Storable a)

109+ => Int -> SerialT m (Array a) -> SerialT m (Array a)

110+compact n xs =

111+ D.fromStreamD $ A.packArraysChunksOf n (D.toStreamD xs)

112+

113+-- | Groups the elements in an input stream into arrays of given size.

114+--

115+-- Same as the following but more efficient:

116+--

117+-- > arraysOf n = FL.groupsOf n (FL.toArrayN n)

118+--

119+-- @since 0.7.0

120+{-# INLINE arraysOf #-}

121+arraysOf :: (IsStream t, MonadIO m, Storable a)

122+ => Int -> t m a -> t m (Array a)

123+arraysOf n str =

124+ D.fromStreamD $ A.fromStreamDArraysOf n (D.toStreamD str)

125+

126+-- XXX Both of these implementations of splicing seem to perform equally well.

127+-- We need to perform benchmarks over a range of sizes though.

128+

129+-- CAUTION! length must more than equal to lengths of all the arrays in the

130+-- stream.

131+{-# INLINE spliceArraysLenUnsafe #-}

132+spliceArraysLenUnsafe :: (MonadIO m, Storable a)

133+ => Int -> SerialT m (Array a) -> m (Array a)

134+spliceArraysLenUnsafe len buffered = do

135+ arr <- liftIO $ A.newArray len

136+ end <- S.foldlM' writeArr (aEnd arr) buffered

137+ return $ arr {aEnd = end}

138+

139+ where

140+

141+ writeArr dst Array{..} =

142+ liftIO $ withForeignPtr aStart $ \src -> do

143+ let count = aEnd `minusPtr` src

144+ A.memcpy (castPtr dst) (castPtr src) count

145+ return $ dst `plusPtr` count

146+

147+{-# INLINE _spliceArraysBuffered #-}

148+_spliceArraysBuffered :: (MonadIO m, Storable a)

149+ => SerialT m (Array a) -> m (Array a)

150+_spliceArraysBuffered s = do

151+ buffered <- P.foldr S.cons S.nil s

152+ len <- S.sum (S.map length buffered)

153+ spliceArraysLenUnsafe len s

154+

155+{-# INLINE spliceArraysRealloced #-}

156+spliceArraysRealloced :: forall m a. (MonadIO m, Storable a)

157+ => SerialT m (Array a) -> m (Array a)

158+spliceArraysRealloced s = do

159+ idst <- liftIO $ A.newArray (A.bytesToCount (undefined :: a)

160+ (A.mkChunkSizeKB 4))

161+

162+ arr <- S.foldlM' A.spliceWithDoubling idst s

163+ liftIO $ A.shrinkToFit arr

164+

165+-- | Given a stream of arrays, splice them all together to generate a single

166+-- array. The stream must be /finite/.

167+--

168+-- @since 0.7.0

169+{-# INLINE toArray #-}

170+toArray :: (MonadIO m, Storable a) => SerialT m (Array a) -> m (Array a)

171+toArray = spliceArraysRealloced

172+-- spliceArrays = _spliceArraysBuffered

173+

src/Streamly/Mem/Array/Types.hs+7

@@ -71,6 +71,7 @@ module Streamly.Mem.Array.Types

7171 , shrinkToFit

7272 , memcpy

7373 , memcmp

74+ , bytesToCount

7475

7576 , unlines

7677 )

@@ -210,6 +211,12 @@ memcmp p1 p2 len = do

210211unsafeInlineIO :: IO a -> a

211212unsafeInlineIO (IO m) = case m realWorld# of (# _, r #) -> r

212213

214+{-# INLINE bytesToCount #-}

215+bytesToCount :: Storable a => a -> Int -> Int

216+bytesToCount x n =

217+ let elemSize = sizeOf x

218+ in n + elemSize - 1 `div` elemSize

219+

213220-------------------------------------------------------------------------------

214221-- Construction

215222-------------------------------------------------------------------------------

src/Streamly/Network/Client.hs+3−3

@@ -70,7 +70,7 @@ import Streamly.Mem.Array.Types (Array(..), defaultChunkSize)

7070import Streamly.Streams.Serial (SerialT)

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

7272

73−import qualified Streamly.Mem.Array as A

73+import qualified Streamly.Mem.Array.Stream as AS

7474import qualified Streamly.Prelude as S

7575import qualified Streamly.Network.Socket as SK

7676

@@ -108,7 +108,7 @@ withConnection addr port =

108108{-# INLINE read #-}

109109read :: (IsStream t, MonadCatch m, MonadIO m)

110110 => (Word8, Word8, Word8, Word8) -> PortNumber -> t m Word8

111−read addr port = A.flattenArrays $ withConnection addr port SK.readArrays

111+read addr port = AS.flatten $ withConnection addr port SK.readArrays

112112

113113-------------------------------------------------------------------------------

114114-- Writing

@@ -141,7 +141,7 @@ writeByChunks

141141 -> PortNumber

142142 -> SerialT m Word8

143143 -> m ()

144−writeByChunks n addr port m = writeArrays addr port $ A.arraysOf n m

144+writeByChunks n addr port m = writeArrays addr port $ AS.arraysOf n m

145145

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

147147--

src/Streamly/Network/Socket.hs+3−2

@@ -93,6 +93,7 @@ import Streamly.Streams.StreamK.Type (IsStream, mkStream)

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

9494

9595import qualified Streamly.Mem.Array as A

96+import qualified Streamly.Mem.Array.Stream as AS

9697import qualified Streamly.Mem.Array.Types as A hiding (flattenArrays)

9798import qualified Streamly.Prelude as S

9899

@@ -265,7 +266,7 @@ readByChunksUpto chunkSize h = A.flattenArrays $ readArraysUpto chunkSize h

265266-- @since 0.7.0

266267{-# INLINE read #-}

267268read :: (IsStream t, MonadIO m) => Socket -> t m Word8

268−read = A.flattenArrays . readArrays

269+read = AS.flatten . readArrays

269270

270271-------------------------------------------------------------------------------

271272-- Writing

@@ -293,7 +294,7 @@ writeArrays h m = S.mapM_ (liftIO . writeArray h) m

293294-- @since 0.7.0

294295{-# INLINE writeByChunks #-}

295296writeByChunks :: MonadIO m => Int -> Socket -> SerialT m Word8 -> m ()

296−writeByChunks n h m = writeArrays h $ A.arraysOf n m

297+writeByChunks n h m = writeArrays h $ AS.arraysOf n m

297298

298299-- > write = 'writeByChunks' A.defaultChunkSize

299300--

src/Streamly/String.hs+2−1

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

7676import qualified Streamly.Fold as FL

7777import qualified Streamly.Mem.Array.Types as A (unlines)

7878import qualified Streamly.Mem.Array as A

79+import qualified Streamly.Mem.Array.Stream as AS

7980import qualified Streamly.Streams.StreamD as D

8081

8182-- type String = List Char

@@ -246,4 +247,4 @@ unlines = D.fromStreamD . A.unlines '\n' . D.toStreamD

246247-- > unwords . words /= id

247248{-# INLINE unwords #-}

248249unwords :: (MonadAsync m, IsStream t) => t m (Array Char) -> t m Char

249−unwords = A.flattenArrays . (S.intersperse (A.fromList " "))

250+unwords = AS.flatten . (S.intersperse (A.fromList " "))

streamly.cabal+1

@@ -232,6 +232,7 @@ library

232232

233233 -- IO devices

234234 , Streamly.Mem.Array

235+ , Streamly.Mem.Array.Stream

235236 , Streamly.FileSystem.Handle

236237

237238 , Streamly.Tutorial

test/Arrays.hs+4−3

@@ -12,6 +12,7 @@ import Test.QuickCheck.Monadic (monadicIO, assert)

1212import Test.Hspec as H

1313

1414import qualified Streamly.Mem.Array as A

15+import qualified Streamly.Mem.Array.Stream as AS

1516import qualified Streamly.Prelude as S

1617

1718-- Coverage build takes too long with default number of tests

@@ -74,7 +75,7 @@ testArraysOf =

7475 monadicIO $ do

7576 xs <- S.toList

7677 $ S.concatMap A.read

77− $ A.arraysOf 240

78+ $ AS.arraysOf 240

7879 $ S.fromList list

7980 assert (xs == list)

8081

@@ -84,8 +85,8 @@ testFlattenArrays =

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

8586 monadicIO $ do

8687 xs <- S.toList

87− $ A.flattenArrays

88− $ A.arraysOf 240

88+ $ AS.flatten

89+ $ AS.arraysOf 240

8990 $ S.fromList list

9091 assert (xs == list)

9192

test/String.hs+2−1

@@ -13,6 +13,7 @@ import Test.QuickCheck.Monadic (run, monadicIO, assert)

1313import Test.Hspec as H

1414

1515import qualified Streamly.Mem.Array as A

16+import qualified Streamly.Mem.Array.Stream as AS

1617import qualified Streamly.Prelude as S

1718import qualified Streamly.String as SS

1819

@@ -57,7 +58,7 @@ testLinesArray =

5758 monadicIO $ do

5859 xs <- S.toList

5960 $ S.map A.toList

60− $ A.splitArraysOn 10

61+ $ AS.splitOn 10

6162 $ S.yield (A.fromList list)

6263 assert (xs == map (map (fromIntegral . ord))

6364 (lines (map (chr . fromIntegral) list)))