eff742a · streamly · git.iko.soy

git.iko.soy/streamly

Use native dir listing functions for listdir example

We are not using the monad instance any more in the example. With these changes the performance of list dir improved by 2x.

commiteff742aa3ae795d26bb79c533f640b9bc2eef62b authorHarendra Kumar <harendra.kumar@gmail.com> dateWed 13 Nov 2019, 22:02 (7 years ago) parent5b64ae8

5 files changed+121−50

compare with master

unifiedsplit

examples/ListDir.hs+23−11 src/Streamly/Internal/Data/SVar.hs+2−2 src/Streamly/Internal/Data/Unfold.hs+17 src/Streamly/Internal/FileSystem/Dir.hs+78−36 streamly.cabal+1−1

examples/ListDir.hs+23−11

@@ -1,19 +1,31 @@

1−import Control.Monad.IO.Class (liftIO)

2−import Path.IO (listDir, getCurrentDir)

31import System.IO (stdout, hSetBuffering, BufferMode(LineBuffering))

4−import Streamly (aheadly, (<>))

2+import Streamly (aheadly, ahead, AheadT)

3+import Data.Function ((&))

55import qualified Streamly.Prelude as S

6+import qualified Streamly.Internal.FileSystem.Dir as Dir

78-- | List the current directory recursively using concurrent processing

89--

9−-- This example demonstrates that there is little difference between regular

10−-- IO code and concurrent streamly code. You can just remove

11−-- 'runStream . aheadly' and this becomes your regular IO code.

1210main :: IO ()

1311main = do

1412 hSetBuffering stdout LineBuffering

15− S.drain . aheadly $ getCurrentDir >>= readdir

16− where readdir d = do

17− (ds, fs) <- listDir d

18− liftIO $ mapM_ putStrLn $ fmap show fs <> fmap show ds

19− foldMap readdir ds

13+ S.mapM_ print $ aheadly $ recursePath (Left ".")

14+

15+ where

16+

17+ -- XXX Fix bug in enqueueAhead when mixing serial with ahead:

18+ -- (\x -> S.yield dir <> S.concatMapWith ahead recursePath x)

19+ -- :: SerialT IO String

20+ -- or S.cons dir . S.concatMapWith ahead recursePath

21+ recursePath :: Either String String -> AheadT IO String

22+ recursePath (Left dir) =

23+ Dir.toStreamEither dir -- SerialT IO (Either String String)

24+ & S.map (prefixDir dir) -- SerialT IO (Either String String)

25+ & S.consM (return dir)

26+ . S.concatMapWith ahead recursePath -- SerialT IO String

27+ recursePath (Right file) = S.yield file -- SerialT IO String

28+

29+ prefixDir :: String -> Either String String -> Either String String

30+ prefixDir dir (Right x) = Right $ dir ++ "/" ++ x

31+ prefixDir dir (Left x) = Left $ dir ++ "/" ++ x

src/Streamly/Internal/Data/SVar.hs+2−2

@@ -1303,7 +1303,7 @@ enqueueAhead :: SVar t m a -> IORef ([t m a], Int) -> t m a -> IO ()

13031303enqueueAhead sv q m = do

13041304 atomicModifyIORefCAS_ q $ \ case

13051305 ([], n) -> ([m], n + 1) -- increment sequence

1306− _ -> error "not empty"

1306+ _ -> error "enqueueAhead: queue is not empty"

13071307 ringDoorBell sv

13081308

13091309-- enqueue without incrementing the sequence number

@@ -1312,7 +1312,7 @@ reEnqueueAhead :: SVar t m a -> IORef ([t m a], Int) -> t m a -> IO ()

13121312reEnqueueAhead sv q m = do

13131313 atomicModifyIORefCAS_ q $ \ case

13141314 ([], n) -> ([m], n) -- DO NOT increment sequence

1315− _ -> error "not empty"

1315+ _ -> error "reEnqueueAhead: queue is not empty"

13161316 ringDoorBell sv

13171317

13181318-- Normally the thread that has the token should never go away. The token gets

src/Streamly/Internal/Data/Unfold.hs+17

@@ -96,6 +96,7 @@ module Streamly.Internal.Data.Unfold

9696 -- * Transformations

9797 , map

9898 , mapM

99+ , mapMWithInput

99100

100101 -- * Filtering

101102 , takeWhileM

@@ -265,6 +266,22 @@ mapM f (Unfold ustep uinject) = Unfold step uinject

265266 Skip s -> return $ Skip s

266267 Stop -> return $ Stop

267268

269+{-# INLINE_NORMAL mapMWithInput #-}

270+mapMWithInput :: Monad m => (a -> b -> m c) -> Unfold m a b -> Unfold m a c

271+mapMWithInput f (Unfold ustep uinject) = Unfold step inject

272+ where

273+ inject a = do

274+ r <- uinject a

275+ return (a, r)

276+

277+ {-# INLINE_LATE step #-}

278+ step (inp, st) = do

279+ r <- ustep st

280+ case r of

281+ Yield x s -> f inp x >>= \a -> return $ Yield a (inp, s)

282+ Skip s -> return $ Skip (inp, s)

283+ Stop -> return $ Stop

284+

268285-------------------------------------------------------------------------------

269286-- Convert streams into unfolds

270287-------------------------------------------------------------------------------

src/Streamly/Internal/FileSystem/Dir.hs+78−36

@@ -21,10 +21,13 @@ module Streamly.Internal.FileSystem.Dir

2121 -- ** Read from Directory

2222 read

2323 , readFiles

24− {-

25− , readWithBufferOf

24+ , readDirs

25+ , readEither

26+ -- , readWithBufferOf

2627

2728 , toStream

29+ , toStreamEither

30+ {-

2831 , toStreamWithBufferOf

2932

3033 , readChunks

@@ -53,31 +56,32 @@ module Streamly.Internal.FileSystem.Dir

5356where

5457

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

56−import Data.Word (Word8)

57−import Foreign.ForeignPtr (withForeignPtr)

58−import Foreign.ForeignPtr.Unsafe (unsafeForeignPtrToPtr)

59−import Foreign.Ptr (minusPtr, plusPtr)

60−import Foreign.Storable (Storable(..))

61−import GHC.ForeignPtr (mallocPlainForeignPtrBytes)

62−import System.IO (Handle, hGetBufSome, hPutBuf)

59+import Data.Either (isRight, fromRight, isLeft, fromLeft)

60+-- import Data.Word (Word8)

61+-- import Foreign.ForeignPtr (withForeignPtr)

62+-- import Foreign.ForeignPtr.Unsafe (unsafeForeignPtrToPtr)

63+-- import Foreign.Ptr (minusPtr, plusPtr)

64+-- import Foreign.Storable (Storable(..))

65+-- import GHC.ForeignPtr (mallocPlainForeignPtrBytes)

66+-- import System.IO (Handle, hGetBufSome, hPutBuf)

6367import Prelude hiding (read)

6468

65−import Streamly.Data.Fold (Fold)

69+-- import Streamly.Data.Fold (Fold)

6670import Streamly.Internal.Data.Unfold.Types (Unfold(..))

67−import Streamly.Internal.Memory.Array.Types

68− (Array(..), writeNUnsafe, defaultChunkSize, shrinkToFit,

69− lpackArraysChunksOf)

70−import Streamly.Streams.Serial (SerialT)

71−import Streamly.Streams.StreamK.Type (IsStream, mkStream)

71+-- import Streamly.Internal.Memory.Array.Types

72+-- (Array(..), writeNUnsafe, defaultChunkSize, shrinkToFit,

73+-- lpackArraysChunksOf)

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

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

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

7377

74−import qualified Streamly.Data.Fold as FL

75−import qualified Streamly.Internal.Data.Fold.Types as FL

78+-- import qualified Streamly.Data.Fold as FL

79+-- import qualified Streamly.Internal.Data.Fold.Types as FL

7680import qualified Streamly.Internal.Data.Unfold as UF

77−import qualified Streamly.Internal.Memory.ArrayStream as AS

81+-- import qualified Streamly.Internal.Memory.ArrayStream as AS

7882import qualified Streamly.Internal.Prelude as S

79−import qualified Streamly.Memory.Array as A

80−import qualified Streamly.Streams.StreamD.Type as D

83+-- import qualified Streamly.Memory.Array as A

84+-- import qualified Streamly.Internal.Data.Stream.StreamD.Type as D

8185import qualified System.Directory as Dir

8286

8387{-

@@ -200,33 +204,71 @@ toStreamWithBufferOf :: (IsStream t, MonadIO m) => Int -> Handle -> t m Word8

200204toStreamWithBufferOf chunkSize h = AS.concat $ toChunksWithBufferOf chunkSize h

201205-}

202206

203−-- TODO

204−-- 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

207+-- XXX exception handling

208+-- | Raw read of a directory

206209--

207−-- | Unfolds a file handle into a byte stream. IO requests to the device are

208−-- performed in sizes of

209−-- 'Streamly.Internal.Memory.Array.Types.defaultChunkSize'.

210+-- /Internal/

210211--

211−-- @since 0.7.0

212−-- XXX exception handling

213−-- XXX this is not stream read

214212{-# INLINE read #-}

215213read :: MonadIO m => Unfold m String String

216−read = UF.lmapM (liftIO . Dir.getDirectoryContents) UF.fromList

214+read =

215+ -- XXX use proper streaming read of the dir

216+ UF.lmapM (liftIO . Dir.getDirectoryContents) UF.fromList

217+

218+-- XXX We can use a more general mechanism to filter the contents of a

219+-- directory. We can just stat each child and pass on the stat information. We

220+-- can then use that info to do a general filtering. "find" like filters can be

221+-- created.

222+

223+-- | Read directories as Left and files as Right. Filter out "." and ".."

224+-- entries.

225+--

226+-- /Internal/

227+--

228+{-# INLINE readEither #-}

229+readEither :: MonadIO m => Unfold m String (Either String String)

230+readEither =

231+ UF.mapMWithInput classify

232+ $ UF.filter (\x -> x /= "." && x /= "..")

233+ -- XXX use proper streaming read of the dir

234+ $ UF.lmapM (liftIO . Dir.getDirectoryContents) UF.fromList

235+ where

236+ classify dir x = do

237+ r <- liftIO $ Dir.doesDirectoryExist (dir ++ "/" ++ x)

238+ return $ if r then Left x else Right x

217239

240+--

241+-- | Read files only.

242+--

243+-- /Internal/

244+--

218245{-# INLINE readFiles #-}

219246readFiles :: MonadIO m => Unfold m String String

220−readFiles = UF.filterM (fmap not . liftIO . Dir.doesDirectoryExist) read

247+readFiles = UF.map (fromRight undefined) $ UF.filter isRight readEither

221248

222−{-

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

249+-- | Read directories only. Filter out "." and ".." entries.

250+--

251+-- /Internal/

252+--

253+{-# INLINE readDirs #-}

254+readDirs :: MonadIO m => Unfold m String String

255+readDirs = UF.map (fromLeft undefined) $ UF.filter isLeft readEither

256+

257+-- | Raw read of a directory.

224258--

225259-- /Internal/

226260{-# INLINE toStream #-}

227−toStream :: (IsStream t, MonadIO m) => Handle -> t m Word8

228−toStream = AS.concat . toChunks

229−-}

261+toStream :: (IsStream t, MonadIO m) => String -> t m String

262+toStream = S.unfold read

263+

264+-- | Read directories as Left and files as Right. Filter out "." and ".."

265+-- entries.

266+--

267+-- /Internal/

268+{-# INLINE toStreamEither #-}

269+toStreamEither :: (IsStream t, MonadIO m)

270+ => String -> t m (Either String String)

271+toStreamEither = S.unfold readEither

230272

231273{-

232274-------------------------------------------------------------------------------

streamly.cabal+1−1

@@ -814,7 +814,7 @@ executable ListDir

814814 build-Depends:

815815 streamly

816816 , base >= 4.8 && < 5

817− , path-io >= 0.1.0 && < 1.5

817+ , directory >= 1.3 && < 1.4

818818 if impl(ghc < 8.0)

819819 build-depends:

820820 transformers >= 0.4 && < 0.6