Skip to content

Use readEitherBytesChunksAt for dir travseral #76

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
wants to merge 1 commit into
base: master
Choose a base branch
from
Open

Use readEitherBytesChunksAt for dir travseral #76

Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
21 changes: 14 additions & 7 deletions examples/ListDir.hs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -28,7 +28,7 @@ import Streamly.Data.Array (Array)
import Streamly.Data.Stream (Stream)
import Streamly.Data.Unfold (Unfold)
import Streamly.FileSystem.Path (Path)
import System.IO (stdout, hSetBuffering, BufferMode(LineBuffering))
import System.IO (hFlush, Handle, stdout, IOMode(..), hSetBuffering, BufferMode(LineBuffering))

import qualified Streamly.Data.Stream.Prelude as Stream
import qualified Streamly.Data.Array as Array
Expand All @@ -48,13 +48,17 @@ import qualified Streamly.FileSystem.Path as Path
import qualified Streamly.Internal.FileSystem.Path as Path (toChunk)
#if !defined(mingw32_HOST_OS) && !defined(__MINGW32__)
import qualified Streamly.Internal.FileSystem.Posix.ReadDir as Dir
(readEitherByteChunks)
(readEitherByteChunks, readEitherByteChunksAt)
import qualified Streamly.Internal.FileSystem.Posix.File as File
import Streamly.Internal.FileSystem.Posix.File (Fd, OpenMode(ReadOnly))
#endif

#if !defined(mingw32_HOST_OS) && !defined(__MINGW32__)
-- Fastest implementation, only works for posix as of now.
listDirByteChunked :: IO ()
listDirByteChunked = do
ppath <- Path.fromString "."

Stream.fold (Handle.writeChunks stdout)
-- $ Array.compactMax' 32000
$ Stream.catRights
Expand All @@ -65,17 +69,18 @@ listDirByteChunked = do
-- $ Stream.concatIterateBfsRev streamDirMaybe -- 154 ms

-- Serial using stream append and interleave
-- $ concatIterateWith StreamK.append -- 154 ms
$ concatIterateWith StreamK.append -- 154 ms
-- $ mergeIterateWith StreamK.interleave -- 154 ms

-- Concurrent
-- XXX To reduce concurrency overhead, perform buffering in each worker
-- and post the buffer or return [Path] and then unfold it.
$ Stream.parConcatIterate id streamDir -- 94 ms
-- $ Stream.parConcatIterate id streamDir -- 94 ms
-- $ Stream.parConcatIterate (Stream.interleaved True) streamDir -- 94 ms
-- $ Stream.parConcatIterate (Stream.ordered True) streamDir -- 154 ms

$ Stream.fromPure (Left [fromJust $ Path.fromString "."])
-- $ Stream.fromPure (Left [fromJust $ Path.fromString "."])
$ Stream.fromPure (Left (ppath, [fromJust $ Path.fromString "."]))

where

Expand All @@ -90,8 +95,10 @@ listDirByteChunked = do
. StreamK.fromStream

-- cfg = Stream.eager False . Stream.maxBuffer 2000 . Stream.maxThreads 2
streamDir :: Either [Path] b -> Stream IO (Either [Path] (Array Word8))
streamDir = either Dir.readEitherByteChunks (const Stream.nil)
-- streamDir :: Either [Path] b -> Stream IO (Either [Path] (Array Word8))
-- streamDir = either Dir.readEitherByteChunks (const Stream.nil)
streamDir :: Either (Path, [Path]) b -> Stream IO (Either (Path, [Path]) (Array Word8))
streamDir = either Dir.readEitherByteChunksAt (const Stream.nil)

streamDirMaybe :: Either [Path] b -> Maybe (Stream IO (Either [Path] (Array Word8)))
streamDirMaybe = either (Just . Dir.readEitherByteChunks) (const Nothing)
Expand Down
Loading