eff742a · streamly · git.iko.soy

git.iko.soy/streamly

on master help sign in

Use native dir listing functions for listdir example

eff742aHarendra Kumar7 years ago

examples/ListDir.hs+23−11

compare with master

unifiedsplit

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

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

2−import Path.IO (listDir, getCurrentDir)

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

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

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

2+import Streamly (aheadly, ahead, AheadT)

3+import Data.Function ((&))

5import qualified Streamly.Prelude as S

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

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

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

8--

9--

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.

12main :: IO ()

10main :: IO ()

13main = do

11main = do

14 hSetBuffering stdout LineBuffering

12 hSetBuffering stdout LineBuffering

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

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

14+

16− where readdir d = do

15+ where

17− (ds, fs) <- listDir d

16+

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

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

19− foldMap readdir ds

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

1303enqueueAhead sv q m = do

1304 atomicModifyIORefCAS_ q $ \ case

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

1306− _ -> error "not empty"

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

1307 ringDoorBell sv

1308

1309-- enqueue without incrementing the sequence number

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

1312reEnqueueAhead sv q m = do

1313 atomicModifyIORefCAS_ q $ \ case

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

1315− _ -> error "not empty"

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

1316 ringDoorBell sv

1317

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

96 -- * Transformations

97 , map

98 , mapM

99+ , mapMWithInput

100

100 -- * Filtering

101 -- * Filtering

101 , takeWhileM

102 , takeWhileM

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

265 Skip s -> return $ Skip s

266 Skip s -> return $ Skip s

266 Stop -> return $ Stop

267 Stop -> return $ Stop

267

268

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+

268-------------------------------------------------------------------------------

285-------------------------------------------------------------------------------

269-- Convert streams into unfolds

286-- Convert streams into unfolds

270-------------------------------------------------------------------------------

287-------------------------------------------------------------------------------

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

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

21 -- ** Read from Directory

22 read

23 , readFiles

24− {-

24+ , readDirs

25+ , readEither

25− , readWithBufferOf

26+ -- , readWithBufferOf

27 , toStream

28 , toStream

29+ , toStreamEither

30+ {-

28 , toStreamWithBufferOf

31 , toStreamWithBufferOf

30 , readChunks

33 , readChunks

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

53where

56where

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

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

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

56−import Data.Word (Word8)

60+-- import Data.Word (Word8)

57−import Foreign.ForeignPtr (withForeignPtr)

61+-- import Foreign.ForeignPtr (withForeignPtr)

58−import Foreign.ForeignPtr.Unsafe (unsafeForeignPtrToPtr)

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

59−import Foreign.Ptr (minusPtr, plusPtr)

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

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

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

61−import GHC.ForeignPtr (mallocPlainForeignPtrBytes)

65+-- import GHC.ForeignPtr (mallocPlainForeignPtrBytes)

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

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

63import Prelude hiding (read)

67import Prelude hiding (read)

65−import Streamly.Data.Fold (Fold)

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

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

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

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

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

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

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

69− lpackArraysChunksOf)

73+-- lpackArraysChunksOf)

70−import Streamly.Streams.Serial (SerialT)

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

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

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

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

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

74−import qualified Streamly.Data.Fold as FL

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

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

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

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

80import 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

78import qualified Streamly.Internal.Prelude as S

82import qualified Streamly.Internal.Prelude as S

79−import qualified Streamly.Memory.Array as A

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

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

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

81import qualified System.Directory as Dir

85import qualified System.Directory as Dir

83{-

87{-

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

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

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

201-}

205-}

202

206

203−-- TODO

207+-- XXX exception handling

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

208+-- | Raw read of a directory

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

206--

209--

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

210+-- /Internal/

208−-- performed in sizes of

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

210--

211--

211−-- @since 0.7.0

212−-- XXX exception handling

213−-- XXX this is not stream read

214{-# INLINE read #-}

212{-# INLINE read #-}

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

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

214+read =

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

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

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

217

239

240+--

241+-- | Read files only.

242+--

243+-- /Internal/

244+--

218{-# INLINE readFiles #-}

245{-# INLINE readFiles #-}

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

246readFiles :: 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

221

248

222−{-

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

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

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.

224--

258--

225-- /Internal/

259-- /Internal/

226{-# INLINE toStream #-}

260{-# INLINE toStream #-}

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

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

228−toStream = AS.concat . toChunks

262+toStream = S.unfold read

229−-}

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

230

272

231{-

273{-

232-------------------------------------------------------------------------------

274-------------------------------------------------------------------------------

streamly.cabal+1−1

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

814 build-Depends:

815 streamly

816 , base >= 4.8 && < 5

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

817+ , directory >= 1.3 && < 1.4

818 if impl(ghc < 8.0)

819 build-depends:

820 transformers >= 0.4 && < 0.6