51f12c9 · streamly · git.iko.soy

git.iko.soy/streamly

Use unfolds to read from file handles

commit51f12c9d227cf95cbe71dd703f59903172242f9b authorHarendra Kumar <harendra.kumar@gmail.com> dateMon 30 Sep 2019, 03:02 (7 years ago) parentbf5e153

5 files changed+68−35

compare with master

unifiedsplit

benchmark/NanoBenchmarks.hs+2−1 examples/HandleIO.hs+2−2 examples/WordClassifier.hs+1−1 src/Streamly/Benchmark/FileIO/Stream.hs+24−24 src/Streamly/Internal/FileSystem/Handle.hs+39−7

benchmark/NanoBenchmarks.hs+2−1

@@ -11,6 +11,7 @@ import Streamly.Internal.Data.SVar (MonadAsync)

1111

1212import qualified Streamly.Memory.Array as A

1313import qualified Streamly.FileSystem.Handle as FH

14+import qualified Streamly.Internal.FileSystem.Handle as IFH

1415import qualified Streamly.Data.Fold as FL

1516import qualified Streamly.Internal.Prelude as Internal

1617import qualified Streamly.Prelude as S

@@ -113,7 +114,7 @@ main = do

113114 defaultMain [mkBenchText "splitOn abc...xyz" inText $ do

114115 (S.length $ Internal.splitOnSeq (A.fromList $ map (fromIntegral . ord)

115116 "abcdefghijklmnopqrstuvwxyz") FL.drain

116− $ FH.read inText) >>= print

117+ $ IFH.toStream inText) >>= print

117118 ]

118119 where

119120

examples/HandleIO.hs+2−2

@@ -47,7 +47,7 @@ grepc pat src = print . (subtract 1) =<< (S.length

4747avgll :: FH.Handle -> IO ()

4848avgll src = print =<< (S.fold avg

4949 $ S.splitWithSuffix (== ord' '\n') FL.length

50− $ FH.read src)

50+ $ S.unfold FH.read src)

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

5252 toDouble = fmap (fromIntegral :: Int -> Double)

5353

@@ -55,7 +55,7 @@ llhisto :: FH.Handle -> IO ()

5555llhisto src = print =<< (S.fold (FL.classify FL.length)

5656 $ S.map bucket

5757 $ S.splitWithSuffix (== ord' '\n') FL.length

58− $ FH.read src)

58+ $ S.unfold FH.read src)

5959 where

6060 bucket n = let i = n `div` 10 in if i > 9 then (9,n) else (i,n)

6161

examples/WordClassifier.hs+1−1

@@ -55,7 +55,7 @@ main =

5555 in do

5656 name <- fmap head getArgs

5757 src <- openFile name ReadMode

58− FH.read src -- SerialT IO Word8

58+ Streamly.unfold FH.read src -- SerialT IO Word8

5959 & Streamly.decodeChar8 -- SerialT IO Char

6060 & Streamly.map toLower -- SerialT IO Char

6161 & Streamly.foldWords FL.toList -- SerialT IO String

src/Streamly/Benchmark/FileIO/Stream.hs+24−24

@@ -75,7 +75,7 @@ import Test.Inspection

7575-- | Get the last byte from a file bytestream.

7676{-# INLINE last #-}

7777last :: Handle -> IO (Maybe Word8)

78−last = S.last . FH.read

78+last = S.last . S.unfold FH.read

7979

8080#ifdef INSPECTION

8181inspect $ hasNoTypeClasses 'last

@@ -88,7 +88,7 @@ inspect $ 'last `hasNoType` ''D.ConcatMapUState

8888-- | Count the number of bytes in a file.

8989{-# INLINE countBytes #-}

9090countBytes :: Handle -> IO Int

91−countBytes = S.length . FH.read

91+countBytes = S.length . S.unfold FH.read

9292

9393#ifdef INSPECTION

9494inspect $ hasNoTypeClasses 'countBytes

@@ -104,7 +104,7 @@ countLines =

104104 S.length

105105 . SS.foldLines FL.drain

106106 . SS.decodeChar8

107− . FH.read

107+ . S.unfold FH.read

108108

109109#ifdef INSPECTION

110110inspect $ hasNoTypeClasses 'countLines

@@ -120,7 +120,7 @@ countWords =

120120 S.length

121121 . SS.foldWords FL.drain

122122 . SS.decodeChar8

123− . FH.read

123+ . S.unfold FH.read

124124

125125#ifdef INSPECTION

126126inspect $ hasNoTypeClasses 'countWords

@@ -146,7 +146,7 @@ inspect $ 'countLinesU `hasNoType` ''D.ConcatMapUState

146146-- | Sum the bytes in a file.

147147{-# INLINE sumBytes #-}

148148sumBytes :: Handle -> IO Word8

149−sumBytes = S.sum . FH.read

149+sumBytes = S.sum . S.unfold FH.read

150150

151151#ifdef INSPECTION

152152inspect $ hasNoTypeClasses 'sumBytes

@@ -158,7 +158,7 @@ inspect $ 'sumBytes `hasNoType` ''D.ConcatMapUState

158158-- | Send the file contents to /dev/null

159159{-# INLINE cat #-}

160160cat :: Handle -> Handle -> IO ()

161−cat devNull inh = S.fold (FH.write devNull) $ FH.read inh

161+cat devNull inh = S.fold (FH.write devNull) $ S.unfold FH.read inh

162162

163163#ifdef INSPECTION

164164inspect $ hasNoTypeClasses 'cat

@@ -170,7 +170,7 @@ inspect $ 'cat `hasNoType` ''D.ConcatMapUState

170170-- | Send the file contents to /dev/null

171171{-# INLINE catStreamWrite #-}

172172catStreamWrite :: Handle -> Handle -> IO ()

173−catStreamWrite devNull inh = IFH.fromStream devNull $ FH.read inh

173+catStreamWrite devNull inh = IFH.fromStream devNull $ S.unfold FH.read inh

174174

175175#ifdef INSPECTION

176176inspect $ hasNoTypeClasses 'catStreamWrite

@@ -182,7 +182,7 @@ inspect $ 'catStreamWrite `hasNoType` ''D.ConcatMapUState

182182-- | Copy file

183183{-# INLINE copy #-}

184184copy :: Handle -> Handle -> IO ()

185−copy inh outh = S.fold (FH.write outh) (FH.read inh)

185+copy inh outh = S.fold (FH.write outh) (S.unfold FH.read inh)

186186

187187#ifdef INSPECTION

188188inspect $ hasNoTypeClasses 'copy

@@ -193,14 +193,14 @@ inspect $ 'copy `hasNoType` ''D.ConcatMapUState

193193

194194{-# INLINE readWord8 #-}

195195readWord8 :: Handle -> IO ()

196−readWord8 inh = S.drain $ FH.read inh

196+readWord8 inh = S.drain $ S.unfold FH.read inh

197197

198198{-# INLINE decodeChar8 #-}

199199decodeChar8 :: Handle -> IO ()

200200decodeChar8 inh =

201201 S.drain

202202 $ SS.decodeChar8

203− $ FH.read inh

203+ $ S.unfold FH.read inh

204204

205205-- | Copy file

206206{-# INLINE copyCodecChar8 #-}

@@ -209,7 +209,7 @@ copyCodecChar8 inh outh =

209209 S.fold (FH.write outh)

210210 $ SS.encodeChar8

211211 $ SS.decodeChar8

212− $ FH.read inh

212+ $ S.unfold FH.read inh

213213

214214#ifdef INSPECTION

215215inspect $ hasNoTypeClasses 'copyCodecChar8

@@ -223,7 +223,7 @@ decodeUtf8Lenient :: Handle -> IO ()

223223decodeUtf8Lenient inh =

224224 S.drain

225225 $ SS.decodeUtf8Lenient

226− $ FH.read inh

226+ $ S.unfold FH.read inh

227227

228228#ifdef INSPECTION

229229inspect $ hasNoTypeClasses 'decodeUtf8Lenient

@@ -239,7 +239,7 @@ copyCodecUtf8 inh outh =

239239 S.fold (FH.write outh)

240240 $ SS.encodeUtf8

241241 $ SS.decodeUtf8

242− $ FH.read inh

242+ $ S.unfold FH.read inh

243243

244244#ifdef INSPECTION

245245inspect $ hasNoTypeClasses 'copyCodecUtf8

@@ -255,7 +255,7 @@ copyCodecUtf8Lenient inh outh =

255255 S.fold (FH.write outh)

256256 $ SS.encodeUtf8

257257 $ SS.decodeUtf8Lenient

258− $ FH.read inh

258+ $ S.unfold FH.read inh

259259

260260#ifdef INSPECTION

261261inspect $ hasNoTypeClasses 'copyCodecUtf8Lenient

@@ -269,7 +269,7 @@ inspect $ hasNoTypeClasses 'copyCodecUtf8Lenient

269269chunksOf :: Int -> Handle -> IO Int

270270chunksOf n inh =

271271 -- writeNUnsafe gives 2.5x boost here over writeN.

272− S.length $ S.chunksOf n (AT.writeNUnsafe n) (FH.read inh)

272+ S.length $ S.chunksOf n (AT.writeNUnsafe n) (S.unfold FH.read inh)

273273

274274#ifdef INSPECTION

275275inspect $ hasNoTypeClasses 'chunksOf

@@ -288,7 +288,7 @@ chunksOfD :: Int -> Handle -> IO Int

288288chunksOfD n inh =

289289 D.foldlM' (\i _ -> return $ i + 1) 0

290290 $ D.groupsOf n (AT.writeNUnsafe n)

291− $ D.fromStreamK (FH.read inh)

291+ $ D.fromStreamK (S.unfold FH.read inh)

292292

293293#ifdef INSPECTION

294294inspect $ hasNoTypeClasses 'chunksOf

@@ -308,7 +308,7 @@ linesUnlinesCopy inh outh =

308308 $ SS.unlines

309309 $ SS.lines

310310 $ SS.decodeChar8

311− $ FH.read inh

311+ $ S.unfold FH.read inh

312312

313313#ifdef INSPECTION

314314inspect $ hasNoTypeClassesExcept 'linesUnlinesCopy [''Storable]

@@ -347,7 +347,7 @@ wordsUnwordsCopyWord8 inh outh =

347347 $ S.concatUnfold IUF.fromList

348348 $ S.intersperse [32]

349349 $ S.wordsBy isSp FL.toList

350− $ FH.read inh

350+ $ S.unfold FH.read inh

351351

352352#ifdef INSPECTION

353353inspect $ hasNoTypeClasses 'wordsUnwordsCopyWord8

@@ -379,7 +379,7 @@ wordsUnwordsCopy inh outh =

379379 $ S.wordsBy isSpace FL.toList

380380 -- -- $ S.splitOn isSpace FL.toList

381381 $ SS.decodeChar8

382− $ FH.read inh

382+ $ S.unfold FH.read inh

383383

384384#ifdef INSPECTION

385385-- inspect $ hasNoTypeClasses 'wordsUnwordsCopy

@@ -399,7 +399,7 @@ toarr = A.fromList . map (fromIntegral . ord)

399399splitOn :: Handle -> IO Int

400400splitOn inh =

401401 (S.length $ S.splitOn (== lf) FL.drain

402− $ FH.read inh) -- >>= print

402+ $ S.unfold FH.read inh) -- >>= print

403403

404404#ifdef INSPECTION

405405inspect $ hasNoTypeClasses 'splitOn

@@ -413,7 +413,7 @@ inspect $ 'splitOn `hasNoType` ''D.ConcatMapUState

413413splitOnSuffix :: Handle -> IO Int

414414splitOnSuffix inh =

415415 (S.length $ S.splitOnSuffix (== lf) FL.drain

416− $ FH.read inh) -- >>= print

416+ $ S.unfold FH.read inh) -- >>= print

417417

418418#ifdef INSPECTION

419419inspect $ hasNoTypeClasses 'splitOnSuffix

@@ -427,7 +427,7 @@ inspect $ 'splitOnSuffix `hasNoType` ''D.ConcatMapUState

427427wordsBy :: Handle -> IO Int

428428wordsBy inh =

429429 (S.length $ S.wordsBy isSp FL.drain

430− $ FH.read inh) -- >>= print

430+ $ S.unfold FH.read inh) -- >>= print

431431

432432#ifdef INSPECTION

433433inspect $ hasNoTypeClasses 'wordsBy

@@ -441,7 +441,7 @@ inspect $ 'wordsBy `hasNoType` ''D.ConcatMapUState

441441splitOnSeq :: String -> Handle -> IO Int

442442splitOnSeq str inh =

443443 (S.length $ IP.splitOnSeq (toarr str) FL.drain

444− $ FH.read inh) -- >>= print

444+ $ S.unfold FH.read inh) -- >>= print

445445

446446#ifdef INSPECTION

447447inspect $ hasNoTypeClasses 'splitOnSeq

@@ -463,7 +463,7 @@ splitOnSeqUtf8 str inh =

463463splitOnSuffixSeq :: String -> Handle -> IO Int

464464splitOnSuffixSeq str inh =

465465 (S.length $ IP.splitOnSuffixSeq (toarr str) FL.drain

466− $ FH.read inh) -- >>= print

466+ $ S.unfold FH.read inh) -- >>= print

467467

468468#ifdef INSPECTION

469469inspect $ hasNoTypeClasses 'splitOnSuffixSeq

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

@@ -24,6 +24,9 @@ module Streamly.Internal.FileSystem.Handle

2424 -- , readFrames

2525 , readInChunksOf

2626

27+ , toStream

28+ , toStreamInChunksOf

29+

2730 -- -- * Array Read

2831 -- , readArrayUpto

2932 -- , readArrayOf

@@ -84,6 +87,7 @@ import System.IO (Handle, hGetBufSome, hPutBuf)

8487import Prelude hiding (read)

8588

8689import Streamly.Data.Fold (Fold)

90+import Streamly.Internal.Data.Unfold.Types (Unfold(..))

8791import Streamly.Internal.Memory.Array.Types

8892 (Array(..), writeNUnsafe, defaultChunkSize, shrinkToFit,

8993 lpackArraysChunksOf)

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

9397

9498import qualified Streamly.Data.Fold as FL

9599import qualified Streamly.Internal.Data.Fold.Types as FL

100+import qualified Streamly.Internal.Data.Unfold as UF

96101import qualified Streamly.Internal.Memory.ArrayStream as AS

97102import qualified Streamly.Internal.Prelude as S

98103import qualified Streamly.Memory.Array as A

@@ -170,6 +175,18 @@ toStreamArraysOf size h = D.fromStreamD (D.Stream step ())

170175 0 -> D.Stop

171176 _ -> D.Yield arr ()

172177

178+{-# INLINE_NORMAL readArraysOf #-}

179+readArraysOf :: MonadIO m => Unfold m (Int, Handle) (Array Word8)

180+readArraysOf = Unfold step return

181+ where

182+ {-# INLINE_LATE step #-}

183+ step (size, h) = do

184+ arr <- liftIO $ readArrayUpto size h

185+ return $

186+ case A.length arr of

187+ 0 -> D.Stop

188+ _ -> D.Yield arr (size, h)

189+

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

174191--

175192-- | @toStreamArrays handle@ reads a stream of arrays from the specified file

@@ -192,25 +209,40 @@ toStreamArrays = toStreamArraysOf defaultChunkSize

192209-- read requests at the same time. For serial case we can use async IO. We can

193210-- also control the read throughput in mbps or IOPS.

194211

195−-- | @readInChunksOf chunkSize handle@ reads a byte stream from a file

196−-- handle, reads are performed in chunks of up to @chunkSize@.

212+-- | @readInChunksOf@ unfolds @(chunksize, handle)@ into a byte stream, reads

213+-- are performed in chunks of up to @chunkSize@.

197214--

198215-- @since 0.7.0

199216{-# INLINE readInChunksOf #-}

200−readInChunksOf :: (IsStream t, MonadIO m) => Int -> Handle -> t m Word8

201−readInChunksOf chunkSize h = AS.concat $ toStreamArraysOf chunkSize h

217+readInChunksOf :: MonadIO m => Unfold m (Int, Handle) Word8

218+readInChunksOf = UF.concat readArraysOf A.read

219+

220+-- | @toStreamInChunksOf chunkSize handle@ reads a byte stream from a file

221+-- handle, reads are performed in chunks of up to @chunkSize@.

222+--

223+-- /Internal/

224+{-# INLINE toStreamInChunksOf #-}

225+toStreamInChunksOf :: (IsStream t, MonadIO m) => Int -> Handle -> t m Word8

226+toStreamInChunksOf chunkSize h = AS.concat $ toStreamArraysOf chunkSize h

202227

203228-- TODO

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

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

206231--

207232-- > read = 'readInChunksOf' A.defaultChunkSize

208−-- | Generate a byte stream from a file 'Handle'.

233+-- | Unfolds a file handle into a byte stream.

209234--

210235-- @since 0.7.0

211236{-# INLINE read #-}

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

213−read = AS.concat . toStreamArrays

237+read :: MonadIO m => Unfold m Handle Word8

238+read = UF.first readInChunksOf defaultChunkSize

239+

240+-- | Generate a byte stream from a file 'Handle'.

241+--

242+-- /Internal/

243+{-# INLINE toStream #-}

244+toStream :: (IsStream t, MonadIO m) => Handle -> t m Word8

245+toStream = AS.concat . toStreamArrays

214246

215247-------------------------------------------------------------------------------

216248-- Writing