packages feed

streamly-examples 0.2.0 → 0.3.0

raw patch · 32 files changed

+795/−170 lines, 32 filesdep +filepathdep +streamly-fseventsdep ~basedep ~hashabledep ~streamlynew-component:exe:EchoClientnew-component:exe:ListDirBasicnew-component:exe:ListDirNaivenew-component:exe:ListDirOsPath

Dependencies added: filepath, streamly-fsevents

Dependency ranges changed: base, hashable, streamly, streamly-core, tasty-bench

Files

+ Cargo.toml view
@@ -0,0 +1,15 @@+[package]+name = "listdir"+version = "0.1.0"+edition = "2021"++[dependencies]+walkdir = "2.4"++[[example]]+name = "WalkDirBasic"+path = "examples/WalkDirBasic.rs"++[[example]]+name = "ListDirBasic"+path = "examples/ListDirBasic.rs"
Changelog.md view
@@ -1,5 +1,9 @@ # Changelog +## 0.3.0 (Sep 2025)++* Support streamly-0.11.0 and streamly-core-0.3.0+ ## 0.2.0 (Dec 2023)  * Support streamly-0.10.0 and streamly-core-0.2.0
Makefile view
@@ -1,10 +1,12 @@-CFLAGS += -std=gnu11 -Wall -Wextra -pedantic -march=native -Ofast+CFLAGS += -Wall -Wextra -pedantic -march=native -Ofast -all: examples/WordCount+EXAMPLES = examples/WordCount examples/ListDirBasic -examples/WordCount: examples/WordCount.c+all: $(EXAMPLES)++examples/%: examples/%.c 	$(CC) $(CFLAGS) $< -o $@ $(LDFLAGS)  .PHONY: clean clean:-	rm -f examples/WordCount+	rm -f $(EXAMPLES)
README.md view
@@ -59,7 +59,7 @@ * [FileSystemEvent](https://github.com/composewell/streamly-examples/blob/master/examples/FileSystemEvent.hs): File watching/fsnotify API   example. * [ListDir](https://github.com/composewell/streamly-examples/blob/master/examples/ListDir.hs): List a directory tree recursively and-  concurrently.+  concurrently, faster than rust fd.  ### Text Processing @@ -110,6 +110,42 @@   stream type to and from `conduit` stream type. * [Interop.Vector](https://github.com/composewell/streamly-examples/blob/master/examples/Interop/Vector.hs): Converting streamly   stream type to and from `vector` stream type.++## Comparing Haskell Performance with C and Rust++### Word Count++`examples/WordCount.c` is the C equivalent of `examples/WordCount.hs`:++```+$ make+$ examples/WordCount input.txt+```++### Directory Traversal++`examples/ListDirBasic.c` is the C equivalent of `examples/ListDirBasic.hs`:++```+$ make+$ examples/ListDirBasic+```++`examples/ListDirBasic.rs` is the Rust equivalent of+`examples/ListDirBasic.hs` using `std::fs` :++```+$ cargo build --example ListDirBasic --release+$ target/release/examples/ListDirBasic+```++`example/WalkDirBasic.rs` is the Rust equivalent of+``example/ListDirBasic.hs` using the `walkdir` crate:++```+$ cargo build --example WalkDirBasic --release+$ target/release/examples/WalkDirBasic+```  ## Licensing
+ examples/BasicDirStream.hsc view
@@ -0,0 +1,183 @@+{-# Language ScopedTypeVariables #-}+{-# Language CApiFFI #-}+-- {-# OPTIONS_GHC -ddump-simpl -ddump-to-file -dsuppress-all #-}++-- | Removing the print statements in both C program and Haskell program. C+-- takes 58ms CPU time, Haskell +RTS -s reports 60ms CPU time (time reports+-- 62ms which includes 2ms of RTS startup/teardown overhead). Actual Haskell+-- processing is approximately 3% slower.+--+-- Learnings and notes:+--+-- * Be frugal with allocations, our perf tool can help in precisely+-- pin-pointing the places where more allocations are happening.+--+-- * static argument transformation is very important in reducing unnecessary+-- overhead in recursive loops, it becomes especially important when other+-- inefficiencies are removed. It will be nice if compiler can automatically+-- detect and point out such opportunities.+--+-- * Using mutable memory in a local loop is beneficial due to cache+-- benefits and reduction in allocations/collections. Haskell lacks good+-- mutable memory abstractions which should be fixed.+--+-- * In C we use local storage on stack and the stack unwinds as the function+-- returns, allocations and deallocations are cheaper. Even if we allocate a+-- variable earlier than it is used, we do not pay much cost. In Haskell, there+-- is no such thing as stack, all allocations happen on heap, once we allocate+-- we pay the cost of collection as well. Therefore, if we simulate stack+-- like mutable allocations in Haskell, when allocating strictly we have to be+-- careful to allocate only when needed.+--+-- * In multithreaded programs however Haskell might work better because in+-- that case each C thread requires allocation of a stack, in Haskell it is+-- natural as it always uses the heap. Thus the memory requirement of Haskell+-- is lower and there is no limit for stack.+--+-- * XXX there should be a way to automatically suggest NOINLINE when using+-- unsafeperformIO.+--+-- * XXX A CString module to manipulate CStrings conveniently?+--+-- * XXX An efficient unsafe variable length mutable cell/array module?++module BasicDirStream (loopDir0)++where++import Control.Monad (when)+import Data.Char (ord)+import Foreign (Ptr, Word8, nullPtr, peek, peekByteOff, castPtr, plusPtr)+import Foreign.C+    (resetErrno, Errno(..), getErrno, eINTR, throwErrno, CChar, CSize(..))+import Streamly.Internal.FileSystem.Posix.ReadDir (DirStream)+import Streamly.Internal.Data.MutByteArray (MutByteArray(..))+import System.IO (hPutBuf, stdout)++import qualified Streamly.Internal.Data.CString as CString+import qualified Streamly.Internal.Data.MutByteArray as MutByteArray+import qualified Streamly.Internal.FileSystem.Posix.ReadDir as ReadDir++#include <dirent.h>++{-# INLINE isMetaDir #-}+isMetaDir :: Ptr CChar -> IO Bool+isMetaDir dname = do+    -- XXX Assuming an encoding that maps "." to ".", this is true for+    -- UTF8.+    -- Load as soon as possible to optimize memory accesses+    c1 <- peek dname+    c2 :: Word8 <- peekByteOff dname 1+    if (c1 /= fromIntegral (ord '.'))+    then return False+    else do+        if (c2 == 0)+        then return True+        else do+            if (c2 /= fromIntegral (ord '.'))+            then return False+            else do+                c3 :: Word8 <- peekByteOff dname 2+                if (c3 == 0)+                then return True+                else return False++foreign import capi unsafe "dirent.h readdir"+    c_readdir  :: DirStream -> IO (Ptr a)++foreign import ccall unsafe "string.h strlen" c_strlen_pinned+    :: Ptr a -> IO CSize++{-+-- Temporary mutable buffer+-- XXX In a multithreaded program, we will need a per thread mut var. So we+-- have to pass the mutvar in a function scope instead. It would be nice to+-- have first class mutable variable support in Haskell. We can use the current+-- thread-id as a token to create a variable on heap in the ST monad.+buffer :: MutByteArray+buffer = MutByteArray.emptyMutVar' 1024+-}++-- dirname should not be mutated, we can use an immutable ByteArray type for+-- that.+loopDir0 :: MutByteArray -> MutByteArray -> DirStream -> IO ()+loopDir0 buffer dirname0 dirp0 = loopDir dirname0 dirp0++    where++    -- XXX We can use a poke monad to do this more+    -- ergonomically. We can create a monad for composing+    -- CStrings, and equivalent of snprintf.+    allocBuf dirLen dname = do+        dentLen <- fmap fromIntegral $ c_strlen_pinned dname+        buf <- MutByteArray.new' (dirLen + dentLen + 2)+        return buf++    printName arr off = do+        -- If we use an Array then we can use a slice of the same+        -- buffer for printing and for passing to loopDir.+        MutByteArray.pokeAt off arr (10 :: Word8)+        -- hPutBuf is quite expensive compared to C printf.+        -- There should be an API that supplies the buffer directly to kernel.+        -- And we can build a buffer in the loop here and write it when it is+        -- ready.+        MutByteArray.unsafeAsPtr arr $ \p ->+            hPutBuf stdout p (off+1)++    -- XXX static argument transformations can create too many scoping levels,+    -- can there be a more convenient way to achieve this?+    loopDir dirname dirp = goDir++        where++        appendName dirLen buf dname = do+            -- XXX can use putCString variant instead+            MutByteArray.pokeAt 0 buf (0 :: Word8)+            _ <- CString.splice buf dirname+            MutByteArray.pokeAt dirLen buf (47 :: Word8)+            off <- CString.putCString buf (dirLen + 1) dname+            return off++        goDir = do+            resetErrno+            ptr <- c_readdir dirp -- streaming read, no buffering+            if (ptr /= nullPtr)+            then do+                let dname = #{ptr struct dirent, d_name} ptr+                dtype :: #{type unsigned char} <- #{peek struct dirent, d_type} ptr++                if dtype == (#const DT_DIR) -- dir/no-symlink check+                then do+                    isMeta <- isMetaDir dname+                    when (not isMeta) $ do+                        dirLen <- CString.length dirname+                        buf <- allocBuf dirLen dname+                        off <- appendName dirLen buf dname++                        -- Print the dir name+                        arr <- MutByteArray.unsafePinnedCloneSlice 0 off buf+                        printName arr off++                        MutByteArray.unsafeAsPtr buf $ \s ->+                            ReadDir.openDirStreamCString (castPtr s) >>= loopDir buf+                else do+                    dirLen <- CString.length dirname+                    -- Using a new buffer every time is more expensive+                    -- buf <- allocBuf dirLen dname+                    -- Use a mutable scratch buffer+                    -- XXX If the size exceeds the scratch buffer then allocate+                    -- a new buffer here.+                    off <- appendName dirLen buffer dname+                    printName buffer off+                    return ()+                goDir+            else do+                errno <- getErrno+                if (errno == eINTR)+                then goDir+                else do+                    let (Errno n) = errno+                    if (n == 0)+                    then do+                        ReadDir.closeDirStream dirp+                    else throwErrno "loopDir"
examples/CSVParser.hs view
@@ -13,7 +13,7 @@ import qualified Streamly.Data.Stream as Stream import qualified Streamly.FileSystem.Handle as Handle import qualified System.IO as IO-import qualified Streamly.Internal.Data.Array.Stream as ArrayStream (splitOn)+import qualified Streamly.Internal.Data.Array as Array (compactSepByByte_)  main :: IO () main = do@@ -21,7 +21,7 @@     src <- IO.openFile inFile ReadMode      Handle.readChunks src                        -- Stream IO (Array Word8)-        & ArrayStream.splitOn 10                 -- Stream IO (Array Word8)+        & Array.compactSepByByte_ 10             -- Stream IO (Array Word8)         & Stream.fold (Fold.drainMapM parseLine) -- IO ()      where@@ -30,7 +30,7 @@      parseLine arr =         (Array.read arr :: Stream IO Word8)-            & Stream.splitOn (== 44) Fold.toList     -- Stream IO [Word8]+            & Stream.splitSepBy_ (== 44) Fold.toList -- Stream IO [Word8]             & Stream.intersperse [32]                -- Stream IO [Word8]             & Stream.fold (Fold.drainMapM printList) -- IO ()         >> putStrLn ""
examples/CamelCase.hs view
@@ -6,7 +6,7 @@ import System.Environment (getArgs) import System.IO (Handle, IOMode(..), openFile, stdout) -import qualified Streamly.Data.Fold as Fold+import qualified Streamly.Data.Scanl as Scanl import qualified Streamly.Data.Stream as Stream import qualified Streamly.FileSystem.Handle as Handle @@ -14,7 +14,7 @@ camelCase :: Handle -> Handle -> IO () camelCase src dst =       Handle.read src                 -- Stream IO Word8-    & Stream.postscan fold            -- Stream IO (Bool, Maybe Word8)+    & Stream.postscanl fold           -- Stream IO (Bool, Maybe Word8)     & Stream.mapMaybe snd             -- Stream IO Word8     & Stream.fold (Handle.write dst)  -- IO () @@ -35,7 +35,7 @@         | isLower x = (False, Just $ if wasSpace then x - 32 else x)         | otherwise = (True, Nothing) -    fold = Fold.foldl' step (True, Nothing)+    fold = Scanl.mkScanl step (True, Nothing)  main :: IO () main = do
examples/CmdClient.hs view
@@ -7,6 +7,7 @@ import Streamly.Data.Stream.Prelude (Stream)  import qualified Streamly.Data.Fold as Fold+import qualified Streamly.Data.Scanl as Scanl import qualified Streamly.Data.Unfold as Unfold import qualified Streamly.Data.Stream.Prelude as Stream import qualified Streamly.Internal.Network.Inet.TCP as TCP (pipeBytes)@@ -34,7 +35,7 @@ sender :: Stream IO () sender =       Stream.repeat "time\nrandom\n"               -- Stream IO String-    & Stream.unfoldMany Unfold.fromList            -- Stream IO Char+    & Stream.unfoldEach Unfold.fromList            -- Stream IO Char     & Unicode.encodeLatin1                         -- Stream IO Word8     & TCP.pipeBytes remoteAddr remotePort          -- Stream IO Word8     & Unicode.decodeLatin1                         -- Stream IO Char@@ -50,9 +51,9 @@ main = do       Stream.replicate 4 sender                    -- Stream IO (Stream IO ())     & Stream.parConcat id                          -- Stream IO ()-    & Stream.postscan counts                       -- Stream IO Int+    & Stream.postscanl counts                      -- Stream IO Int     & Stream.fold Fold.drain                       -- IO ()      where -    counts = Fold.foldlM' (counter "rcvd") (return 0 :: IO Int)+    counts = Scanl.mkScanlM (counter "rcvd") (return 0 :: IO Int)
examples/CmdServer.hs view
@@ -38,7 +38,7 @@ sendValue sk x =       Stream.fromList (show x ++ "\n")     & Unicode.encodeLatin1-    & Stream.fold (Array.writeN 60)+    & Stream.fold (Array.createOf 60)     >>= Socket.putChunk sk  ------------------------------------------------------------------------------@@ -64,7 +64,7 @@ {-# INLINE demuxKvToMap #-} demuxKvToMap :: (Monad m, Ord k) =>     (k -> m (Fold m a b)) -> Fold m (k, a) (Map k b)-demuxKvToMap f = Fold.demuxToMap fst (\(k, _) -> fmap (Fold.lmap snd) (f k))+demuxKvToMap f = Fold.demuxerToMap fst (fmap (Just . Fold.lmap snd) . f)  demux :: Fold IO (String, Socket) () demux = void (demuxKvToMap commands)
examples/ControlFlow.hs view
@@ -27,7 +27,7 @@ import Control.Monad.Trans.Except (ExceptT, runExceptT, throwE, catchE) import Control.Monad.Trans.Cont (ContT(..), callCC) import Streamly.Data.Stream (Stream)-import Streamly.Internal.Data.Stream (CrossStream, mkCross, unCross)+import Streamly.Internal.Data.Stream (Nested(..))  import qualified Streamly.Data.Fold as Fold import qualified Streamly.Data.Stream as Stream@@ -40,10 +40,10 @@ -- non-determinism. -- getSequenceMaybeBelow :: MonadIO m => Stream (MaybeT m) ()-getSequenceMaybeBelow = unCross $ do+getSequenceMaybeBelow = unNested $ do     liftIO $ putStrLn "MaybeT below streamly: Enter one char per line: " -    i <- mkCross $ Stream.fromList [1..2 :: Int]+    i <- Nested $ Stream.fromList [1..2 :: Int]     liftIO $ putStrLn $ "iteration = " <> show i      r1 <- liftIO getLine@@ -69,11 +69,11 @@ -- Note that this is redundant configuration as the same behavior can be -- achieved with just streamly, using mzero. ---getSequenceMaybeAbove :: MonadIO m => MaybeT (CrossStream m) ()+getSequenceMaybeAbove :: MonadIO m => MaybeT (Nested m) () getSequenceMaybeAbove = do     liftIO $ putStrLn "MaybeT above streamly: Enter one char per line: " -    i <- lift $ mkCross $ Stream.fromList [1..2 :: Int]+    i <- lift $ Nested $ Stream.fromList [1..2 :: Int]     liftIO $ putStrLn $ "iteration = " <> show i      r1 <- liftIO getLine@@ -82,7 +82,7 @@     r2 <- liftIO getLine     when (r2 /= "y") mzero -mainMaybeAbove :: MonadIO m => MaybeT (CrossStream m) ()+mainMaybeAbove :: MonadIO m => MaybeT (Nested m) () mainMaybeAbove = do     getSequenceMaybeAbove     liftIO $ putStrLn "Bingo"@@ -96,10 +96,10 @@ -- Note that throwE would terminate all iterations of non-determinism -- altogether. getSequenceEitherBelow :: MonadIO m => Stream (ExceptT String m) ()-getSequenceEitherBelow = unCross $ do+getSequenceEitherBelow = unNested $ do     liftIO $ putStrLn "ExceptT below streamly: Enter one char per line: " -    i <- mkCross $ Stream.fromList [1..2 :: Int]+    i <- Nested $ Stream.fromList [1..2 :: Int]     liftIO $ putStrLn $ "iteration = " <> show i      r1 <- liftIO getLine@@ -124,10 +124,10 @@ --  getSequenceEitherAsyncBelow :: MonadIO m => Stream (ExceptT String m) ()-getSequenceEitherAsyncBelow = unCross $ do+getSequenceEitherAsyncBelow = unNested $ do     liftIO $ putStrLn "ExceptT below concurrent streamly: " -    i <- mkCross+    i <- Nested             $ Stream.consM                 (liftIO (threadDelay 1000)                     >> throwE "First task"@@ -158,11 +158,11 @@ -- -- Here we can use catchE directly but will have to use monad-control to lift -- stream operations with stream arguments.-getSequenceEitherAbove :: MonadIO m => ExceptT String (CrossStream m) ()+getSequenceEitherAbove :: MonadIO m => ExceptT String (Nested m) () getSequenceEitherAbove = do     liftIO $ putStrLn "ExceptT above streamly: Enter one char per line: " -    i <- lift $ mkCross $ Stream.fromList [1..2 :: Int]+    i <- lift $ Nested $ Stream.fromList [1..2 :: Int]     liftIO $ putStrLn $ "iteration = " <> show i      r1 <- liftIO getLine@@ -171,7 +171,7 @@     r2 <- liftIO getLine     when (r2 /= "y") $ throwE $ "Expecting y got: " <> r2 -mainEitherAbove :: MonadIO m => ExceptT String (CrossStream m) ()+mainEitherAbove :: MonadIO m => ExceptT String (Nested m) () mainEitherAbove =     catchE (getSequenceEitherAbove >> liftIO (putStrLn "Bingo"))            (liftIO . putStrLn)@@ -188,20 +188,20 @@ -- iterations of non-determinism rather then just the current iteration. -- getSequenceMonadThrow :: (MonadIO m, MonadThrow m) => Stream m ()-getSequenceMonadThrow = unCross $ do+getSequenceMonadThrow = unNested $ do     liftIO $ putStrLn "MonadThrow in streamly: Enter one char per line: " -    i <- mkCross $ Stream.fromList [1..2 :: Int]+    i <- Nested $ Stream.fromList [1..2 :: Int]     liftIO $ putStrLn $ "iteration = " <> show i      r1 <- liftIO getLine     when (r1 /= "x")-        $ mkCross+        $ Nested         $ Stream.fromEffect $ throwM $ Unexpected $ "Expecting x got: " <> r1      r2 <- liftIO getLine     when (r2 /= "y")-        $ mkCross+        $ Nested         $ Stream.fromEffect $ throwM $ Unexpected $ "Expecting y got: " <> r2  mainMonadThrow :: IO ()@@ -218,11 +218,11 @@ -- -- XXX need to have a specialized liftCallCC to actually lift callCC ---getSequenceContBelow :: MonadIO m => CrossStream (ContT r m) (Either String ())+getSequenceContBelow :: MonadIO m => Nested (ContT r m) (Either String ()) getSequenceContBelow = do     liftIO $ putStrLn "ContT below streamly: Enter one char per line: " -    i <- mkCross $ Stream.fromList [1..2 :: Int]+    i <- Nested $ Stream.fromList [1..2 :: Int]     liftIO $ putStrLn $ "iteration = " <> show i      r <- lift $ callCC $ \exit -> do@@ -239,7 +239,7 @@     return r  mainContBelow :: MonadIO m => Stream (ContT r m) ()-mainContBelow = unCross $ do+mainContBelow = unNested $ do     r <- getSequenceContBelow     case r of         Right _ -> liftIO $ putStrLn "Bingo"@@ -249,11 +249,11 @@ -- Using ContT above streamly ------------------------------------------------------------------------------- ---getSequenceContAbove :: MonadIO m => ContT r (CrossStream m) (Either String ())+getSequenceContAbove :: MonadIO m => ContT r (Nested m) (Either String ()) getSequenceContAbove = do     liftIO $ putStrLn "ContT above streamly: Enter one char per line: " -    i <- lift $ mkCross $ Stream.fromList [1..2 :: Int]+    i <- lift $ Nested $ Stream.fromList [1..2 :: Int]     liftIO $ putStrLn $ "iteration = " <> show i      callCC $ \exit -> do@@ -267,7 +267,7 @@         then exit $ Left $ "Expecting y got: " <> r2         else return $ Right () -mainContAbove :: MonadIO m => ContT r (CrossStream m) ()+mainContAbove :: MonadIO m => ContT r (Nested m) () mainContAbove = do     r <- getSequenceContAbove     case r of@@ -282,10 +282,10 @@ main :: IO () main = do     mainMaybeBelow-    Stream.fold Fold.drain $ unCross $ runMaybeT mainMaybeAbove+    Stream.fold Fold.drain $ unNested $ runMaybeT mainMaybeAbove     runContT (Stream.fold Fold.drain mainContBelow) return-    Stream.fold Fold.drain $ unCross $ runContT mainContAbove return+    Stream.fold Fold.drain $ unNested $ runContT mainContAbove return     mainEitherBelow-    Stream.fold Fold.drain $ unCross $ runExceptT mainEitherAbove+    Stream.fold Fold.drain $ unNested $ runExceptT mainEitherAbove     mainMonadThrow     mainEitherAsyncBelow
examples/CoreUtils.hs view
@@ -1,3 +1,5 @@+{-# LANGUAGE QuasiQuotes #-}+ -- Implement some simple coreutils commands  import Control.Monad (void)@@ -6,46 +8,49 @@ import System.Environment (getArgs) import Streamly.Data.Array (Array) import Streamly.Data.Fold (Fold)+import Streamly.FileSystem.Path (path)  import qualified Streamly.Console.Stdio as Stdio import qualified Streamly.Data.Fold as Fold import qualified Streamly.Data.Stream as Stream -import qualified Streamly.Internal.FileSystem.Dir as Dir (readFiles)-import qualified Streamly.Internal.FileSystem.File as File+import qualified Streamly.Internal.FileSystem.DirIO as Dir (readFiles)+import qualified Streamly.Internal.FileSystem.FileIO as File  -- | > cat input.txt cat :: IO () cat =-      File.readChunks "input.txt"     -- Stream IO (Array Word8)-    & Stream.fold Stdio.writeChunks -- IO ()+      File.readChunks [path|input.txt|] -- Stream IO (Array Word8)+    & Stream.fold Stdio.writeChunks     -- IO ()  -- | Read all files from a directory and write to a single output file. -- > cat dir/* > output.txt catDirTo :: IO () catDirTo =-      Dir.readFiles "dir"                -- Stream IO String-    & Stream.unfoldMany File.chunkReader -- Stream IO (Array Word8)-    & File.fromChunks "output.txt"       -- IO ()+      Dir.readFiles id [path|dir|]       -- Stream IO Path+    -- NOTE: This is a redundant step that we are forced to do as we've still+    -- not introduced FileIO module that uses Path yet.+    & Stream.unfoldEach File.chunkReader -- Stream IO (Array Word8)+    & File.fromChunks [path|output.txt|] -- IO ()  -- | > cp input.txt output.txt cp :: IO () cp =-      File.readChunks "input.txt"    -- Stream IO (Array Word8)-    & File.fromChunks "output.txt"   -- IO ()+      File.readChunks [path|input.txt|]  -- Stream IO (Array Word8)+    & File.fromChunks [path|output.txt|] -- IO ()  -- | > cat input.txt >> output.txt append :: IO () append =-      File.readChunks "input.txt"      -- Stream IO (Array Word8)-    & File.writeAppendChunks "output.txt"   -- IO ()+      File.readChunks [path|input.txt|]         -- Stream IO (Array Word8)+    & File.writeAppendChunks [path|output.txt|] -- IO ()  -- | > cat input.txt | tee output1.txt > output.txt tap :: IO () tap =-      File.readChunks "input.txt"                   -- Stream IO (Array Word8)-    & Stream.tap (File.writeChunks "output1.txt") -- Stream IO (Array Word8)-    & File.fromChunks "output.txt"                -- IO ()+      File.readChunks [path|input.txt|]                 -- Stream IO (Array Word8)+    & Stream.tap (File.writeChunks [path|output1.txt|]) -- Stream IO (Array Word8)+    & File.fromChunks [path|output.txt|]                -- IO () {- -- | > cat input.txt | tee output1.txt > output.txt -- output1.txt is processed/written to in a separate thread.@@ -54,28 +59,28 @@       Stdio.readChunks ()   -- Stream IO (Array Word8)     & Stream.tapAsyncK           (File.fromChunks "output1.txt") -- Stream IO (Array Word8)-    & File.fromChunks "output.txt"        -- IO ()+    & File.fromChunks [path|output.txt|]  -- IO () -}  -- | > cat input.txt | tee output1.txt > output.txt tee :: IO () tee =-      File.readChunks "input.txt" -- Stream IO (Array Word8)-    & Stream.fold t               -- IO ((),())-    & void                        -- IO ()+      File.readChunks [path|input.txt|] -- Stream IO (Array Word8)+    & Stream.fold t                     -- IO ((),())+    & void                              -- IO ()      where      -- Combine two folds into a single fold     t :: Fold IO (Array Word8) ((),())     t = Fold.tee-            (File.writeChunks "output1.txt") -- Fold IO (Array Word8) ()-            (File.writeChunks "output.txt")  -- Fold IO (Array Word8) ()+            (File.writeChunks [path|output1.txt|]) -- Fold IO (Array Word8) ()+            (File.writeChunks [path|output.txt|])  -- Fold IO (Array Word8) () {- -- | > grep -c "the" input.txt grepc :: IO () grepc = do-    File.toBytes "input.txt"                                -- Stream IO Word8+    File.toBytes [path|input.txt|]                          -- Stream IO Word8         & Stream.splitOnSeq (Array.fromList pat) Fold.drain -- Stream IO ()         & Stream.length                                     -- IO Int         >>= print . subtract 1                              -- IO ()
examples/CoreUtilsHandle.hs view
@@ -9,8 +9,7 @@ import qualified Streamly.Data.Stream as Stream import qualified Streamly.FileSystem.Handle as Handle import qualified Streamly.Unicode.Stream as Unicode--import qualified Streamly.Internal.Data.Array.Stream as ArrayStream (splitOn)+import qualified Streamly.Internal.Data.Array as Array (compactSepByByte_)  -- | Read the contents of a file to stdout. --@@ -73,9 +72,9 @@ -- first. wcl :: Handle -> IO Int wcl src =-      Handle.readChunks src   -- Stream IO (Array Word8)-    & ArrayStream.splitOn 10  -- Stream IO (Array Word8)-    & Stream.fold Fold.length -- IO ()+      Handle.readChunks src      -- Stream IO (Array Word8)+    & Array.compactSepByByte_ 10 -- Stream IO (Array Word8)+    & Stream.fold Fold.length    -- IO ()  main :: IO () main = do
examples/DateTimeParser.hs view
@@ -19,7 +19,6 @@ import qualified Streamly.Data.Array as Array import qualified Streamly.Data.Fold as Fold import qualified Streamly.Data.Parser as Parser-import qualified Streamly.Data.ParserK as ParserK import qualified Streamly.Data.StreamK as StreamK import qualified Streamly.Data.Stream as Stream import qualified Streamly.Internal.Data.Fold as Fold (foldt', satisfy)@@ -153,19 +152,19 @@ _parseBreakDateTime :: Array Char -> IO Int _parseBreakDateTime arr = do     let s = StreamK.fromStream $ Stream.fromPure arr-        p = ParserK.adaptC . Parser.fromFold-    (Right year, s1) <- StreamK.parseBreakChunks (p $ decimal 4) s-    (_, s2) <- StreamK.parseBreakChunks (p $ char '-') s1-    (Right month, s3) <- StreamK.parseBreakChunks (p $ decimal 2) s2-    (_, s4) <- StreamK.parseBreakChunks (p $ char '-') s3-    (Right day, s5) <- StreamK.parseBreakChunks (p $ decimal 2) s4-    (_, s6) <- StreamK.parseBreakChunks (p $ char 'T') s5-    (Right hr, s7) <- StreamK.parseBreakChunks (p $ decimal 2) s6-    (_, s8) <- StreamK.parseBreakChunks (p $ char ':') s7-    (Right mn, s9) <- StreamK.parseBreakChunks (p $ decimal 2) s8-    (_, s10) <- StreamK.parseBreakChunks (p $ char ':') s9-    (Right sec, s11) <- StreamK.parseBreakChunks (p $ decimal 2) s10-    (_, _) <- StreamK.parseBreakChunks (p $ char 'Z') s11+        p = Array.toParserK . Parser.fromFold+    (Right year, s1) <- Array.parseBreak (p $ decimal 4) s+    (_, s2) <- Array.parseBreak (p $ char '-') s1+    (Right month, s3) <- Array.parseBreak (p $ decimal 2) s2+    (_, s4) <- Array.parseBreak (p $ char '-') s3+    (Right day, s5) <- Array.parseBreak (p $ decimal 2) s4+    (_, s6) <- Array.parseBreak (p $ char 'T') s5+    (Right hr, s7) <- Array.parseBreak (p $ decimal 2) s6+    (_, s8) <- Array.parseBreak (p $ char ':') s7+    (Right mn, s9) <- Array.parseBreak (p $ decimal 2) s8+    (_, s10) <- Array.parseBreak (p $ char ':') s9+    (Right sec, s11) <- Array.parseBreak (p $ decimal 2) s10+    (_, _) <- Array.parseBreak (p $ char 'Z') s11     return (year + month + day + hr + mn + sec)  -------------------------------------------------------------------------------@@ -175,12 +174,12 @@ {-# NOINLINE _parseKDateTime #-} _parseKDateTime :: Array Char -> IO Int _parseKDateTime arr = do-    r <- StreamK.parseChunks dateParser $ StreamK.fromPure arr+    r <- Array.parse dateParser $ StreamK.fromPure arr     return $ fromRight (error "failed") r      where -    p = ParserK.adaptC+    p = Array.toParserK      dateParser = do         year <- p $ Parser.decimal <* Parser.char '-'
+ examples/EchoClient.hs view
@@ -0,0 +1,36 @@+module Main (main) where++import Data.Function ((&))+import Data.Word (Word8)+import Network.Socket (PortNumber)+import Streamly.Data.Stream.Prelude (Stream)++import qualified Streamly.Console.Stdio as Stdio+import qualified Streamly.Data.Array as Array+import qualified Streamly.Data.Fold as Fold+import qualified Streamly.Data.Stream.Prelude as Stream+import qualified Streamly.Internal.Network.Inet.TCP as TCP+import qualified Streamly.Unicode.Stream as Unicode+++remoteAddr :: (Word8,Word8,Word8,Word8)+remoteAddr = (127, 0, 0, 1)++remotePort :: PortNumber+remotePort = 8091++echo :: Stream IO ()+echo =+      Stream.unfold Stdio.reader ()                -- Stream IO Word8+    & split (== 10) Array.create                   -- Stream IO (Array Word8)+    & TCP.pipeChunks remoteAddr remotePort         -- Stream IO (Array Word8)+    & Stream.unfoldEach Array.reader               -- Stream IO Word8+    & Unicode.decodeLatin1                         -- Stream IO Char+    & Stream.mapM putChar                          -- Stream IO ()++    where++    split p f = Stream.foldMany (Fold.takeEndBy p f)++main :: IO ()+main = Stream.fold Fold.drain echo
examples/FileSystemEvent.hs view
@@ -2,42 +2,31 @@ -- Report all events recursively under the paths provided as arguments module Main (main) where -import Control.Monad.IO.Class (MonadIO) import Data.Function ((&))-import Data.Word (Word8) import System.Environment (getArgs)-import Streamly.Data.Array (Array)  import qualified Data.List.NonEmpty as NonEmpty-import qualified Streamly.Data.Array as Array import qualified Streamly.Data.Fold as Fold import qualified Streamly.Data.Stream as Stream-import qualified Streamly.Unicode.Stream as Unicode+import qualified Streamly.FileSystem.Path as Path  #if darwin_HOST_OS-import qualified Streamly.Internal.FileSystem.Event.Darwin as Event+import qualified Streamly.Internal.FS.Event.Darwin as Event #elif linux_HOST_OS-import qualified Streamly.Internal.FileSystem.Event.Linux as Event+import qualified Streamly.Internal.FS.Event.Linux as Event #elif mingw32_HOST_OS-import qualified Streamly.Internal.FileSystem.Event.Windows as Event+import qualified Streamly.Internal.FS.Event.Windows as Event #else #error "FS Events not supported on this platform" #endif  ---------------------------------------------------------------------------------- Utilities----------------------------------------------------------------------------------toUtf8 :: MonadIO m => String -> m (Array Word8)-toUtf8 s = Stream.fold Array.write (Unicode.encodeUtf8' $ Stream.fromList s)--------------------------------------------------------------------------------- -- Main -------------------------------------------------------------------------------  main :: IO () main = do     args <- getArgs-    paths <- mapM toUtf8 args+    paths <- mapM Path.fromString args     Event.watch (NonEmpty.fromList paths)         & Stream.fold (Fold.drainMapM (putStrLn . Event.showEvent))
examples/Intro.hs view
@@ -1,3 +1,5 @@+{-# LANGUAGE QuasiQuotes #-}+ -- Introductory example programs  import Control.Concurrent (threadDelay)@@ -11,14 +13,15 @@ import Streamly.Data.Fold (Fold, Tee(..)) import Streamly.Data.Stream.Prelude (Stream) import Streamly.Data.Unfold (Unfold)-import Streamly.Internal.Data.Stream (CrossStream, mkCross, unCross)+import Streamly.FileSystem.Path (path)+import Streamly.Internal.Data.Stream (Nested(..))  import qualified Streamly.Data.Array as Array import qualified Streamly.Data.Fold as Fold import qualified Streamly.Data.Stream.Prelude as Stream import qualified Streamly.Data.Unfold as Unfold import qualified Streamly.FileSystem.Handle as Handle-import qualified Streamly.FileSystem.File as File+import qualified Streamly.FileSystem.FileIO as File import qualified Streamly.Unicode.Stream as Unicode  -------------------------------------------------------------------------------@@ -63,11 +66,11 @@ -- efficient. The second stream may depend on the first stream. The loops -- cannot fuse completely. ---nestedLoops :: CrossStream IO ()+nestedLoops :: Nested IO () nestedLoops = do-    x <- mkCross $  Stream.fromList [3,4 :: Int]-    y <- mkCross $  Stream.fromList [1..x]-    mkCross $  Stream.fromEffect $ print (x, y)+    x <- Nested $  Stream.fromList [3,4 :: Int]+    y <- Nested $  Stream.fromList [1..x]+    Nested $  Stream.fromEffect $ print (x, y)  ------------------------------------------------------------------------------- -- Text processing@@ -79,7 +82,7 @@ -- | Find average line length for lines in a text file avgLineLength :: IO Double avgLineLength =-      File.read "input.txt"                    -- Stream IO Word8+      File.read [path|input.txt|]                 -- Stream IO Word8     & splitOn isNewLine Fold.length               -- Stream IO Int     & Stream.fold avg                             -- IO Double @@ -103,7 +106,7 @@ -- | Read text from a file and generate a histogram of line length lineLengthHistogram :: IO (Map Int Int) lineLengthHistogram =-      File.read "input.txt"                      -- Stream IO Word8+      File.read [path|input.txt|]                -- Stream IO Word8     & splitOn isNewLine Fold.length              -- Stream IO Int     & fmap bucket                                -- Stream IO (Int, Int)     & Stream.fold (kvMap Fold.length)            -- IO (Map Int Int)@@ -119,7 +122,7 @@ -- | Read text from a file and generate a histogram of word length wordLengthHistogram :: IO (Map Int Int) wordLengthHistogram =-      File.read "input.txt"                   -- Stream IO Word8+      File.read [path|input.txt|]             -- Stream IO Word8     & Unicode.decodeLatin1                    -- Stream IO Char     & Stream.wordsBy isSpace Fold.length      -- Stream IO Int     & fmap bucket                             -- Stream IO (Int, Int)@@ -157,7 +160,7 @@     & fmap Array.fromList                     -- Stream IO (Array Word8)     & Stream.fold (Handle.writeChunks stdout) -- IO () -    where unlinesBy = Stream.intercalateSuffix (Unfold.function id)+    where unlinesBy k = Stream.unfoldEachEndBySeq k (Unfold.function id)  ------------------------------------------------------------------------------- -- Main@@ -184,7 +187,7 @@             print $ runIdentity $ crossProduct (1,1000) (1000,2000)         "nestedLoops" -> do             putStrLn "nestedLoops"-            Stream.fold Fold.drain $ unCross nestedLoops+            Stream.fold Fold.drain $ unNested nestedLoops         "avgLineLength" -> do             putStrLn "avgLineLength"             avgLineLength >>= print
examples/ListDir.hs view
@@ -1,36 +1,212 @@+{-# LANGUAGE CPP #-} {-# LANGUAGE FlexibleContexts #-}-{-# OPTIONS_GHC -Wno-deprecations #-}+{-# OPTIONS_GHC -Wno-unused-binds #-}+{-# OPTIONS_GHC -Wno-unused-imports #-} +-- This is faster than the rust "fd". To compare listing the entire tree+-- recursively, use the following commands:+--+-- $ time fd -u > /dev/null+-- $ time ListDir > /dev/null+--+-- Running on a sample directory tree the concurrent rust "fd" tool took 150 ms+-- (real time). On the same tree the fastest variant using Haskell streamly+-- below took 94 ms. The time taken by other variants on the same tree is noted+-- in the comments. The fastest serial implementation using Haskell streamly+-- takes similar time as the concurrent rust "fd".+--+-- The code for directory traversal is just a few lines. This file is bigger+-- because we have implemented it in around 27 possible ways. To try other+-- variants just uncomment the relevant line and comment the currently enabled+-- line.+ module Main (main) where +import Data.Maybe (fromJust)+import Data.Word (Word8)+import Streamly.Data.Array (Array)+import Streamly.Data.Stream (Stream)+import Streamly.Data.Unfold (Unfold)+import Streamly.FileSystem.DirIO (ReadOptions)+import Streamly.FileSystem.Path (Path) import System.IO (stdout, hSetBuffering, BufferMode(LineBuffering)) -import qualified Streamly.Data.Fold as Fold import qualified Streamly.Data.Stream.Prelude as Stream--- import qualified Streamly.Internal.Data.Stream as Stream--- import qualified Streamly.Internal.Data.Unfold as Unfold (either, nil)-import qualified Streamly.Internal.FileSystem.Dir as Dir+import qualified Streamly.Data.Array as Array+import qualified Streamly.FileSystem.DirIO as DirIO+import qualified Streamly.Internal.Data.Array as Array (compactMax')+import qualified Streamly.Internal.Data.Stream as Stream+    (unfoldEachEndBy, concatIterateDfs, concatIterateBfs, concatIterateBfsRev)+import qualified Streamly.Data.StreamK as StreamK+import qualified Streamly.Internal.Data.StreamK as StreamK+    (concatIterateWith, mergeIterateWith)+import qualified Streamly.Data.Unfold as Unfold+import qualified Streamly.Internal.Data.Unfold as Unfold+    (either, nil)+import qualified Streamly.Internal.FileSystem.DirIO as Dir+    (readEitherChunks, readEitherPaths, eitherReaderPaths)+import qualified Streamly.FileSystem.Handle as Handle+import qualified Streamly.FileSystem.Path as Path+import qualified Streamly.Internal.FileSystem.Path as Path (toArray)+#if !defined(mingw32_HOST_OS) && !defined(__MINGW32__)+import qualified Streamly.Internal.FileSystem.Posix.ReadDir as Dir+    (readEitherByteChunks)+#endif --- | List the current directory recursively using concurrent processing-main :: IO ()-main = do-    hSetBuffering stdout LineBuffering-    Stream.fold (Fold.drainMapM print)-        -- $ Stream.unfoldIterateDfs unfoldOne-        -- $ Stream.unfoldIterateBfs unfoldOne-        -- $ Stream.unfoldIterateBfsRev unfoldOne-        -- $ Stream.concatIterateDfs streamOneMaybe-        -- $ Stream.concatIterateBfs streamOneMaybe-        -- $ Stream.concatIterateBfsRev streamOneMaybe-        -- $ Stream.concatIterateWith Stream.append streamOne-        -- $ Stream.mergeIterateWith Stream.interleave streamOne-        $ Stream.parConcatIterate id streamOne-        -- $ Stream.parConcatIterate (Stream.interleaved True) streamOne-        -- $ Stream.parConcatIterate (Stream.ordered True) streamOne-        $ Stream.fromPure (Left ".")+{-# INLINE recReadOpts #-}+recReadOpts :: ReadOptions -> ReadOptions+recReadOpts =+      DirIO.followSymlinks True+    . DirIO.ignoreSymlinkLoops False+    . DirIO.ignoreMissing True+    . DirIO.ignoreInaccessible True +#if !defined(mingw32_HOST_OS) && !defined(__MINGW32__)+-- Fastest implementation, only works for posix as of now.+listDirByteChunked :: IO ()+listDirByteChunked = do+    Stream.fold (Handle.writeChunks stdout)+        -- $ Array.compactMax' 32000+        $ Stream.catRights++        -- Serial+        -- $ Stream.concatIterateDfs streamDirMaybe -- 154 ms+        -- $ Stream.concatIterateBfs streamDirMaybe -- 154 ms+        -- $ Stream.concatIterateBfsRev streamDirMaybe -- 154 ms++        -- Serial using stream append and interleave+        -- $ 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 (Stream.interleaved True) streamDir -- 94 ms+        -- $ Stream.parConcatIterate (Stream.ordered True) streamDir -- 154 ms++        $ Stream.fromPure (Left [fromJust $ Path.fromString "."])+     where -    -- unfoldOne = Unfold.either Dir.eitherReaderPaths Unfold.nil-    -- streamOneMaybe = either (Just . Dir.readEitherPaths) (const Nothing)-    streamOne = either Dir.readEitherPaths (const Stream.nil)+    concatIterateWith f =+          StreamK.toStream+        . StreamK.concatIterateWith f (StreamK.fromStream . streamDir)+        . StreamK.fromStream++    mergeIterateWith f =+          StreamK.toStream+        . StreamK.mergeIterateWith f (StreamK.fromStream . streamDir)+        . 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 recReadOpts) (const Stream.nil)++    streamDirMaybe :: Either [Path] b -> Maybe (Stream IO (Either [Path] (Array Word8)))+    streamDirMaybe = either (Just . Dir.readEitherByteChunks recReadOpts) (const Nothing)+#endif++-- Faster than the listDir implementation below+listDirChunked :: IO ()+listDirChunked = do+    Stream.fold (Handle.writeWith 32000 stdout)+        $ Stream.unfoldEachEndBy 10 Array.reader+        $ fmap Path.toArray+        $ Stream.unfoldEach Unfold.fromList+        $ fmap (either id id)++        -- Serial using streams+        -- $ Stream.concatIterateDfs streamDirMaybe    -- 264 ms+        -- $ Stream.concatIterateBfs streamDirMaybe    -- 264 ms+        -- $ Stream.concatIterateBfsRev streamDirMaybe -- 264 ms++        -- Serial using stream append and interleave+        -- $ concatIterateWith StreamK.append     -- 164 ms+        -- $ mergeIterateWith StreamK.interleave  -- 194 ms++        -- Concurrent+        $ Stream.parConcatIterate id streamDir -- 124 ms+        -- $ Stream.parConcatIterate (Stream.interleaved True) streamDir -- 134 ms+        -- $ Stream.parConcatIterate (Stream.ordered True) streamDir -- 174 ms++        $ Stream.fromPure (Left [fromJust $ Path.fromString "."])++    where++    concatIterateWith f =+          StreamK.toStream+        . StreamK.concatIterateWith f (StreamK.fromStream . streamDir)+        . StreamK.fromStream++    mergeIterateWith f =+          StreamK.toStream+        . StreamK.mergeIterateWith f (StreamK.fromStream . streamDir)+        . StreamK.fromStream++    streamDir :: Either [Path] b -> Stream IO (Either [Path] [Path])+    streamDir = either (Dir.readEitherChunks recReadOpts) (const Stream.nil)++    streamDirMaybe :: Either [Path] b -> Maybe (Stream IO (Either [Path] [Path]))+    streamDirMaybe = either (Just . Dir.readEitherChunks recReadOpts) (const Nothing)++listDir :: IO ()+listDir = do+    Stream.fold (Handle.writeWith 32000 stdout)+        $ Stream.unfoldEachEndBy 10 Array.reader+        $ fmap (Path.toArray . either id id)++        -- Serial using unfolds+        -- $ Stream.unfoldIterateDfs unfoldDir -- 284 ms+        -- May fail with too many open files+        -- $ Stream.unfoldIterateBfs unfoldDir+        -- $ Stream.unfoldIterateBfsRev unfoldDir -- 344 ms++        -- Serial using streams+        -- $ Stream.concatIterateDfs streamDirMaybe -- 274 ms+        -- $ Stream.concatIterateBfs streamDirMaybe -- 274 ms+        -- $ Stream.concatIterateBfsRev streamDirMaybe -- 264 ms++        -- Serial using stream append and interleave+        -- $ concatIterateWith StreamK.append -- 204 ms+        -- $ mergeIterateWith StreamK.interleave  -- 304 ms++        -- Concurrent+        $ Stream.parConcatIterate id streamDir -- 174 ms+        -- $ Stream.parConcatIterate (Stream.interleaved True) streamDir -- 224 ms+        -- $ Stream.parConcatIterate (Stream.ordered True) streamDir -- 234 ms++        $ Stream.fromPure (Left (fromJust $ Path.fromString "."))++    where++    concatIterateWith f =+          StreamK.toStream+        . StreamK.concatIterateWith f (StreamK.fromStream . streamDir)+        . StreamK.fromStream++    mergeIterateWith f =+          StreamK.toStream+        . StreamK.mergeIterateWith f (StreamK.fromStream . streamDir)+        . StreamK.fromStream++    streamDir :: Either Path b -> Stream IO (Either Path Path)+    streamDir = either (Dir.readEitherPaths recReadOpts) (const Stream.nil)++    unfoldDir :: Unfold IO (Either Path b) (Either Path Path)+    unfoldDir = Unfold.either (Dir.eitherReaderPaths recReadOpts) Unfold.nil++    streamDirMaybe :: Either Path b -> Maybe (Stream IO (Either Path Path))+    streamDirMaybe = either (Just . Dir.readEitherPaths recReadOpts) (const Nothing)++-- | List the current directory recursively+main :: IO ()+main = do+    hSetBuffering stdout LineBuffering+#if !defined(mingw32_HOST_OS) && !defined(__MINGW32__)+    listDirByteChunked+#else+    listDirChunked+#endif+    -- listDirChunked+    -- listDir
+ examples/ListDirBasic.c view
@@ -0,0 +1,36 @@+#include <stdio.h>+#include <stdlib.h>+#include <dirent.h>+#include <string.h>++void list_directory(const char *path) {+    DIR *dir = opendir(path);+    if (dir == NULL) {+        perror("opendir");+        return;+    }++    struct dirent *entry;+    char full_path[1024];++    while ((entry = readdir(dir)) != NULL) {+        if (strcmp(entry->d_name, ".") == 0 || strcmp(entry->d_name, "..") == 0) {+            continue;+        }++        snprintf(full_path, sizeof(full_path), "%s/%s", path, entry->d_name);+        printf("%s\n", full_path);++        if (entry->d_type == DT_DIR) {+            list_directory(full_path);+        }+    }++    closedir(dir);+}++int main() {+    const char *path = ".";+    list_directory(path);+    return 0;+}
+ examples/ListDirBasic.hs view
@@ -0,0 +1,16 @@+module Main (main) where++import BasicDirStream (loopDir0)+import Streamly.Internal.Data.Array (arrContents)++import qualified Streamly.Internal.Data.MutByteArray as MutByteArray+import qualified Streamly.Internal.FileSystem.Path as Path+import qualified Streamly.Internal.FileSystem.Posix.ReadDir as ReadDir++main :: IO ()+main = do+    name <- Path.fromString "."+    let arr = Path.toArray name+    buffer <- MutByteArray.new' 1024+    ReadDir.openDirStream name+        >>= loopDir0 buffer (arrContents arr)
+ examples/ListDirBasic.rs view
@@ -0,0 +1,25 @@+use std::fs;+use std::path::Path;++fn list_dir(path: &Path) {+    if let Ok(entries) = fs::read_dir(path) {+        for entry in entries.filter_map(|e| e.ok()) {+            let entry_path = entry.path();+            println!("{}", entry_path.display());++            if let Ok(metadata) = entry.metadata() {+                if metadata.is_dir() && !metadata.file_type().is_symlink() {+                    list_dir(&entry_path);+                }+            }+        }+    } else {+        eprintln!("Could not access: {}", path.display());+    }+}++fn main() {+    let dir = ".";+    let path = Path::new(dir);+    list_dir(path);+}
+ examples/ListDirNaive.hs view
@@ -0,0 +1,21 @@+import Control.Monad+import System.Directory+import System.FilePath++listDir :: FilePath -> IO ()+listDir dir = do+    contents <- listDirectory dir+    let fullPaths = map (dir </>) contents+    forM_ fullPaths $ \path -> do+        isDir <- doesDirectoryExist path+        if isDir+        then do+            symlink <- pathIsSymbolicLink path+            if symlink+            then putStrLn path+            else listDir path -- depth first+        else putStrLn path+    putStrLn dir++main :: IO ()+main = listDir "."
+ examples/ListDirOsPath.hs view
@@ -0,0 +1,25 @@+{-# LANGUAGE QuasiQuotes #-}+module Main (main) where++import Control.Monad+import System.Directory.OsPath+import System.OsPath+import Data.Maybe++listDir :: OsPath -> IO ()+listDir dir = do+    contents <- listDirectory dir+    let fullPaths = map (dir </>) contents+    forM_ fullPaths $ \path -> do+        isDir <- doesDirectoryExist path+        if isDir+        then do+            symlink <- pathIsSymbolicLink path+            if symlink+            then putStrLn $ fromJust $ decodeUtf path+            else listDir path -- depth first+        else putStrLn $ fromJust $ decodeUtf path+    putStrLn $ fromJust $ decodeUtf dir++main :: IO ()+main = listDir [osp|.|]
examples/MergeServer.hs view
@@ -17,9 +17,9 @@ -- a limit to the buffering for safety. readLines :: Socket -> Stream IO (Array Char) readLines sk =-    Socket.read sk              -- Stream IO Word8-  & Unicode.decodeLatin1        -- Stream IO Char-  & split (== '\n') Array.write -- Stream IO String+    Socket.read sk               -- Stream IO Word8+  & Unicode.decodeLatin1         -- Stream IO Char+  & split (== '\n') Array.create -- Stream IO String    where @@ -36,7 +36,7 @@ server file =       TCP.accept 8090                               -- Stream IO Socket     & Stream.parConcatMap (Stream.eager True) recv  -- Stream IO (Array Char)-    & Stream.unfoldMany Array.reader                -- Stream IO Char+    & Stream.unfoldEach Array.reader                -- Stream IO Char     & Unicode.encodeLatin1                          -- Stream IO Word8     & Stream.fold (Handle.write file)               -- IO ()
examples/MergeSort.hs view
@@ -15,7 +15,7 @@ import qualified Streamly.Data.Stream.Prelude as Stream import qualified Streamly.Data.StreamK as K -import qualified Streamly.Internal.Data.Stream as Stream (reduceIterateBfs)+import qualified Streamly.Internal.Data.Stream as Stream (bfsReduceIterate)  input :: [Int] input = [1000000,999999..1]@@ -31,7 +31,7 @@         . Array.read  sortChunk :: Array Int -> IO (Array Int)-sortChunk = Stream.fold Array.write . streamChunk+sortChunk = Stream.fold Array.create . streamChunk  ------------------------------------------------------------------------------- -- Stream the unsorted chunks and sort, merge those streams.@@ -41,7 +41,7 @@ sortMergeCombined :: (Array Int -> Stream IO Int) -> IO () sortMergeCombined f =     Stream.fromList input-        & Stream.chunksOf chunkSize+        & Array.chunksOf chunkSize         & K.fromStream         & K.mergeMapWith (K.mergeBy compare) (K.fromStream . f)         & K.toStream@@ -60,7 +60,7 @@     -> IO () sortMergeSeparate f =     Stream.fromList input-        & Stream.chunksOf chunkSize+        & Array.chunksOf chunkSize         & f sortChunk         & K.fromStream         & K.mergeMapWith (K.mergeBy compare) (K.fromStream . Array.read)@@ -78,7 +78,7 @@         compare         (Array.read arr1)         (Array.read arr2)-        & Stream.fold Array.write+        & Stream.fold Array.create  sortMergeChunks ::        (  (Array Int -> IO (Array Int))@@ -88,9 +88,9 @@     -> IO () sortMergeChunks f =     Stream.fromList input-        & Stream.chunksOf chunkSize+        & Array.chunksOf chunkSize         & f sortChunk-        & Stream.reduceIterateBfs reduce+        & Stream.bfsReduceIterate reduce         & void  -- | Divide a stream in chunks, sort the chunks and merge them.
examples/Rate.hs view
@@ -7,5 +7,5 @@ main =       Stream.sequence (Stream.repeat (pure "tick"))     & Stream.timestamped-    & Stream.parEval (Stream.avgRate 1)+    & Stream.parBuffered (Stream.avgRate 1)     & Stream.fold (Fold.drainMapM print)
+ examples/WalkDirBasic.rs view
@@ -0,0 +1,10 @@+use walkdir::WalkDir;+use std::env;++fn main() {+    dir = ".";++    for entry in WalkDir::new(dir).into_iter().filter_map(|e| e.ok()) {+        println!("{}", entry.path().display());+    }+}
examples/WordCount.hs view
@@ -7,10 +7,12 @@ import Data.Char (ord) import Data.Function ((&)) import System.Environment (getArgs)+import Streamly.FileSystem.Path (Path)  import qualified Streamly.Data.Fold as Fold import qualified Streamly.Data.Stream as Stream-import qualified Streamly.FileSystem.File as File+import qualified Streamly.FileSystem.FileIO as File+import qualified Streamly.FileSystem.Path as Path import qualified Streamly.Unicode.Stream as Stream  -------------------------------------------------------------------------------@@ -39,7 +41,7 @@             else (if wasSpace then w + 1 else w, False)     in Counts l1 w1 (c + 1) wasSpace1 -wc :: String -> IO Counts+wc :: Path -> IO Counts wc file =       File.read file                                      -- Stream IO Word8     & Stream.decodeLatin1                                 -- Stream IO Char@@ -53,5 +55,6 @@ main :: IO () main = do     name <- fmap head getArgs-    Counts l w c _ <- wc name+    nameP <- Path.fromString name+    Counts l w c _ <- wc nameP     putStrLn $ show l ++ " " ++ show w ++ " " ++ show c ++ " " ++ name
examples/WordCountModular.hs view
@@ -9,10 +9,12 @@ import Data.Word (Word8) import Streamly.Data.Fold (Fold, Tee(..)) import System.Environment (getArgs)+import Streamly.FileSystem.Path (Path)  import qualified Streamly.Data.Fold as Fold import qualified Streamly.Data.Stream as Stream-import qualified Streamly.FileSystem.File as File+import qualified Streamly.FileSystem.FileIO as File+import qualified Streamly.FileSystem.Path as Path  {-# INLINE isSpace #-} isSpace :: Char -> Bool@@ -24,7 +26,7 @@ -------------------------------------------------------------------------------  -- The fold accepts a stream of `Word8` and returns a value of type "a".-foldWith :: Fold IO Word8 a -> String -> IO a+foldWith :: Fold IO Word8 a -> Path -> IO a foldWith f file =     File.read file    -- Stream IO Word8   & Stream.fold f     -- IO a@@ -69,8 +71,9 @@ main :: IO () main = do     name <- fmap head getArgs+    nameP <- Path.fromString name     -- foldWith Fold.length name >>= print -- count bytes only     -- foldWith nlines name >>= print      -- count lines only     -- foldWith nwords name >>= print      -- count words only-    (c, l, w) <- foldWith countAll name    -- count all at once+    (c, l, w) <- foldWith countAll nameP   -- count all at once     putStrLn $ show l ++ " " ++ show w ++ " " ++ show c ++ " " ++ name
examples/WordCountParallel.hs view
@@ -9,11 +9,13 @@ import System.Environment (getArgs) import Streamly.Data.Array (Array) import WordCount (count, Counts(..), isSpace)+import Streamly.FileSystem.Path (Path)  import qualified Streamly.Data.Array as Array import qualified Streamly.Data.Fold as Fold import qualified Streamly.Data.Stream.Prelude as Stream-import qualified Streamly.FileSystem.File as File (readChunks)+import qualified Streamly.FileSystem.Path as Path+import qualified Streamly.FileSystem.FileIO as File (readChunks) import qualified Streamly.Unicode.Stream as Stream  -- Get the line, word, char counts in one chunk.@@ -48,7 +50,7 @@  -- Now put it all together, we only need to divide the stream into arrays, -- apply our counting function to each array and then combine all the counts.-wc :: String -> IO (Bool, Counts)+wc :: Path -> IO (Bool, Counts) wc file = do       File.readChunks file               -- Stream IO (Array Word8)     & Stream.parMapM@@ -69,5 +71,6 @@ main :: IO () main = do     name <- fmap head getArgs-    (_, Counts l w c _) <- wc name+    nameP <- Path.fromString name+    (_, Counts l w c _) <- wc nameP     putStrLn $ show l ++ " " ++ show w ++ " " ++ show c ++ " " ++ name
examples/WordCountParallelUTF8.hs view
@@ -180,19 +180,19 @@ -------------------------------------------------------------------------------  readField :: MutArray Int -> Field -> IO Int-readField v fld = MArray.getIndexUnsafe (fromEnum fld) v+readField v fld = MArray.unsafeGetIndex (fromEnum fld) v  writeField :: MutArray Int -> Field -> Int -> IO ()-writeField v fld i = void $ MArray.putIndexUnsafe i v (fromEnum fld)+writeField v fld i = void $ MArray.unsafePutIndex i v (fromEnum fld)  modifyField :: MutArray Int -> Field -> (Int -> Int) -> IO () modifyField v fld f = do   let index = fromEnum fld-  MArray.modifyIndexUnsafe index v (\x -> (f x, ()))+  MArray.unsafeModifyIndex index v (\x -> (f x, ()))  newCounts :: IO (MutArray Int) newCounts = do-    counts <- MArray.pinnedNew (fromEnum (maxBound :: Field) + 1)+    counts <- MArray.emptyOf' (fromEnum (maxBound :: Field) + 1)     writeField counts LineCount 0     writeField counts WordCount 0     writeField counts CharCount 0
examples/WordFrequency.hs view
@@ -17,7 +17,8 @@ import qualified Streamly.Data.Array as Array import qualified Streamly.Data.Fold.Prelude as Fold import qualified Streamly.Data.Stream as Stream-import qualified Streamly.FileSystem.File as File+import qualified Streamly.FileSystem.Path as Path+import qualified Streamly.FileSystem.FileIO as File import qualified Streamly.Unicode.Stream as Unicode  hashArray :: (Unbox a, Integral b, Num c) =>@@ -56,12 +57,13 @@ main :: IO () main = do     inFile <- fmap head getArgs+    inFileP <- Path.fromString inFile      let counter = Fold.foldl' (\n _ -> n + 1) (0 :: Int)         classifier = Fold.toHashMapIO id counter     -- Write the stream to a hashmap consisting of word counts     mp <--        File.read inFile                       -- Stream IO Word8+        File.read inFileP                      -- Stream IO Word8          & Unicode.decodeLatin1                -- Stream IO Char          & fmap toLower                        -- Stream IO Char          & Stream.wordsBy isSpace Fold.toList  -- Stream IO String
streamly-examples.cabal view
@@ -1,6 +1,6 @@ cabal-version: 2.2 name:          streamly-examples-version:       0.2.0+version:       0.3.0 license:       Apache-2.0 license-file:  LICENSE author:        Composewell Technologies@@ -25,10 +25,14 @@ tested-with: GHC==9.4.4, GHC==9.2.7, GHC==9.0.1, GHC==8.10.7 build-type:    Simple extra-source-files:+  Cargo.toml   Changelog.md   Makefile   NOTICE   README.md+  examples/ListDirBasic.c+  examples/ListDirBasic.rs+  examples/WalkDirBasic.rs   examples/WordCount.c  source-repository head@@ -60,9 +64,11 @@  common exe-dependencies   build-depends:-      streamly              == 0.10.0-    , streamly-core         == 0.2.0-    , base                  >= 4.9   && < 4.20+      streamly              == 0.11.0+    -- , streamly-fsevents     == 0.1.0+    , streamly-core         == 0.3.0+    , streamly-fsevents     == 0.1.0+    , base                  >= 4.9   && < 4.22     , directory             >= 1.2   && < 1.4     , transformers          >= 0.4   && < 0.7     , containers            >= 0.5   && < 0.8@@ -70,10 +76,12 @@     , exceptions            >= 0.8   && < 0.11     , transformers-base     >= 0.4   && < 0.5     , network               >= 2.6   && < 4-    , hashable              >= 1.2   && < 1.5+    , hashable              >= 1.2   && < 1.6     , unordered-containers  >= 0.2   && < 0.3     , vector                >= 0.12  && < 0.14     , mtl                   >= 2.2   && < 3+    -- For the ListDirOsPath example+    , filepath              >= 1.4.2 && < 1.6  common exe-options   import: exe-dependencies@@ -197,6 +205,23 @@   import: exe-options-threaded   main-is: ListDir.hs +executable ListDirNaive+  import: exe-options+  main-is: ListDirNaive.hs++executable ListDirOsPath+  import: exe-options+  main-is: ListDirOsPath.hs+  if impl(ghc < 9.6)+    buildable: False++executable ListDirBasic+  import: exe-options+  main-is: ListDirBasic.hs+  other-modules: BasicDirStream+  if os(windows)+    buildable: False+ executable MergeSort   import: exe-options-threaded   main-is: MergeSort.hs@@ -238,6 +263,14 @@   else     buildable: False +executable EchoClient+  import: exe-options-threaded+  main-is: EchoClient.hs+  if !impl(ghcjs)+    buildable: True+  else+    buildable: False+ executable FileSender   import: exe-options-threaded   main-is: FileSender.hs@@ -333,7 +366,7 @@   main-is: DateTimeParser.hs   if !impl(ghcjs)     buildable: True-    build-depends: tasty-bench >= 0.3 && < 0.4+    build-depends: tasty-bench >= 0.3 && < 0.5   else     buildable: False @@ -342,6 +375,6 @@   main-is: LogParser.hs   if !impl(ghcjs)     buildable: True-    build-depends: tasty-bench >= 0.3 && < 0.4+    build-depends: tasty-bench >= 0.3 && < 0.5   else     buildable: False