51f12c9 · streamly · git.iko.soy

git.iko.soy/streamly

on master help sign in

Use unfolds to read from file handles

51f12c9Harendra Kumar7 years ago

benchmark/NanoBenchmarks.hs+2−1

compare with master

unifiedsplit

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

12import qualified Streamly.Memory.Array as A

13import qualified Streamly.FileSystem.Handle as FH

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

14import qualified Streamly.Data.Fold as FL

15import qualified Streamly.Data.Fold as FL

15import qualified Streamly.Internal.Prelude as Internal

16import qualified Streamly.Internal.Prelude as Internal

16import qualified Streamly.Prelude as S

17import qualified Streamly.Prelude as S

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

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

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

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

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

115 "abcdefghijklmnopqrstuvwxyz") FL.drain

116 "abcdefghijklmnopqrstuvwxyz") FL.drain

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

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

117 ]

118 ]

118 where

119 where

119

120

examples/HandleIO.hs+2−2

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

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

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

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

50− $ FH.read src)

50+ $ S.unfold FH.read src)

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

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

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

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

56 $ S.map bucket

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

58− $ FH.read src)

58+ $ S.unfold FH.read src)

59 where

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

examples/WordClassifier.hs+1−1

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

55 in do

56 name <- fmap head getArgs

57 src <- openFile name ReadMode

58− FH.read src -- SerialT IO Word8

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

59 & Streamly.decodeChar8 -- SerialT IO Char

60 & Streamly.map toLower -- SerialT IO Char

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

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

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

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

76{-# INLINE last #-}

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

78−last = S.last . FH.read

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

80#ifdef INSPECTION

81inspect $ hasNoTypeClasses 'last

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

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

89{-# INLINE countBytes #-}

90countBytes :: Handle -> IO Int

91−countBytes = S.length . FH.read

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

93#ifdef INSPECTION

94inspect $ hasNoTypeClasses 'countBytes

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

104 S.length

105 . SS.foldLines FL.drain

106 . SS.decodeChar8

107− . FH.read

107+ . S.unfold FH.read

108

109#ifdef INSPECTION

110inspect $ hasNoTypeClasses 'countLines

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

120 S.length

121 . SS.foldWords FL.drain

122 . SS.decodeChar8

123− . FH.read

123+ . S.unfold FH.read

124

125#ifdef INSPECTION

126inspect $ hasNoTypeClasses 'countWords

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

146-- | Sum the bytes in a file.

147{-# INLINE sumBytes #-}

148sumBytes :: Handle -> IO Word8

149−sumBytes = S.sum . FH.read

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

150

151#ifdef INSPECTION

152inspect $ hasNoTypeClasses 'sumBytes

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

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

159{-# INLINE cat #-}

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

162

163#ifdef INSPECTION

164inspect $ hasNoTypeClasses 'cat

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

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

171{-# INLINE catStreamWrite #-}

172catStreamWrite :: 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

174

175#ifdef INSPECTION

176inspect $ hasNoTypeClasses 'catStreamWrite

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

182-- | Copy file

183{-# INLINE copy #-}

184copy :: 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)

186

187#ifdef INSPECTION

188inspect $ hasNoTypeClasses 'copy

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

193

194{-# INLINE readWord8 #-}

195readWord8 :: Handle -> IO ()

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

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

197

198{-# INLINE decodeChar8 #-}

199decodeChar8 :: Handle -> IO ()

200decodeChar8 inh =

201 S.drain

202 $ SS.decodeChar8

203− $ FH.read inh

203+ $ S.unfold FH.read inh

204

205-- | Copy file

206{-# INLINE copyCodecChar8 #-}

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

209 S.fold (FH.write outh)

210 $ SS.encodeChar8

211 $ SS.decodeChar8

212− $ FH.read inh

212+ $ S.unfold FH.read inh

213

214#ifdef INSPECTION

215inspect $ hasNoTypeClasses 'copyCodecChar8

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

223decodeUtf8Lenient inh =

224 S.drain

225 $ SS.decodeUtf8Lenient

226− $ FH.read inh

226+ $ S.unfold FH.read inh

227

228#ifdef INSPECTION

229inspect $ hasNoTypeClasses 'decodeUtf8Lenient

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

239 S.fold (FH.write outh)

240 $ SS.encodeUtf8

241 $ SS.decodeUtf8

242− $ FH.read inh

242+ $ S.unfold FH.read inh

243

244#ifdef INSPECTION

245inspect $ hasNoTypeClasses 'copyCodecUtf8

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

255 S.fold (FH.write outh)

256 $ SS.encodeUtf8

257 $ SS.decodeUtf8Lenient

258− $ FH.read inh

258+ $ S.unfold FH.read inh

259

260#ifdef INSPECTION

261inspect $ hasNoTypeClasses 'copyCodecUtf8Lenient

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

269chunksOf :: Int -> Handle -> IO Int

270chunksOf n inh =

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

273

274#ifdef INSPECTION

275inspect $ hasNoTypeClasses 'chunksOf

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

288chunksOfD n inh =

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

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

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

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

292

293#ifdef INSPECTION

294inspect $ hasNoTypeClasses 'chunksOf

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

308 $ SS.unlines

309 $ SS.lines

310 $ SS.decodeChar8

311− $ FH.read inh

311+ $ S.unfold FH.read inh

312

313#ifdef INSPECTION

314inspect $ hasNoTypeClassesExcept 'linesUnlinesCopy [''Storable]

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

347 $ S.concatUnfold IUF.fromList

348 $ S.intersperse [32]

349 $ S.wordsBy isSp FL.toList

350− $ FH.read inh

350+ $ S.unfold FH.read inh

351

352#ifdef INSPECTION

353inspect $ hasNoTypeClasses 'wordsUnwordsCopyWord8

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

379 $ S.wordsBy isSpace FL.toList

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

381 $ SS.decodeChar8

382− $ FH.read inh

382+ $ S.unfold FH.read inh

383

384#ifdef INSPECTION

385-- inspect $ hasNoTypeClasses 'wordsUnwordsCopy

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

399splitOn :: Handle -> IO Int

400splitOn inh =

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

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

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

403

404#ifdef INSPECTION

405inspect $ hasNoTypeClasses 'splitOn

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

413splitOnSuffix :: Handle -> IO Int

414splitOnSuffix inh =

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

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

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

417

418#ifdef INSPECTION

419inspect $ hasNoTypeClasses 'splitOnSuffix

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

427wordsBy :: Handle -> IO Int

428wordsBy inh =

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

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

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

431

432#ifdef INSPECTION

433inspect $ hasNoTypeClasses 'wordsBy

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

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

442splitOnSeq str inh =

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

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

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

445

446#ifdef INSPECTION

447inspect $ hasNoTypeClasses 'splitOnSeq

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

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

464splitOnSuffixSeq str inh =

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

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

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

467

468#ifdef INSPECTION

469inspect $ hasNoTypeClasses 'splitOnSuffixSeq

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

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

24 -- , readFrames

25 , readInChunksOf

27+ , toStream

28+ , toStreamInChunksOf

29+

27 -- -- * Array Read

30 -- -- * Array Read

28 -- , readArrayUpto

31 -- , readArrayUpto

29 -- , readArrayOf

32 -- , readArrayOf

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

84import Prelude hiding (read)

87import Prelude hiding (read)

86import Streamly.Data.Fold (Fold)

89import Streamly.Data.Fold (Fold)

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

87import Streamly.Internal.Memory.Array.Types

91import Streamly.Internal.Memory.Array.Types

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

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

89 lpackArraysChunksOf)

93 lpackArraysChunksOf)

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

94import qualified Streamly.Data.Fold as FL

98import qualified Streamly.Data.Fold as FL

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

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

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

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

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

97import qualified Streamly.Internal.Prelude as S

102import qualified Streamly.Internal.Prelude as S

98import qualified Streamly.Memory.Array as A

103import qualified Streamly.Memory.Array as A

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

170 0 -> D.Stop

175 0 -> D.Stop

171 _ -> D.Yield arr ()

176 _ -> D.Yield arr ()

172

177

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+

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

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

174--

191--

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

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

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

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

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

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

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

194

211

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

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

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

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

197--

214--

198-- @since 0.7.0

215-- @since 0.7.0

199{-# INLINE readInChunksOf #-}

216{-# INLINE readInChunksOf #-}

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 #-}

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

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

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

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

202

227

203-- TODO

228-- TODO

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

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

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

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

206--

231--

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

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

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

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

209--

234--

210-- @since 0.7.0

235-- @since 0.7.0

211{-# INLINE read #-}

236{-# INLINE read #-}

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 #-}

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

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

213−read = AS.concat . toStreamArrays

245+toStream = AS.concat . toStreamArrays

214

246

215-------------------------------------------------------------------------------

247-------------------------------------------------------------------------------

216-- Writing

248-- Writing