streamly-examples 0.1.0 → 0.1.1
raw patch · 9 files changed
+113/−59 lines, 9 filesdep ~basedep ~streamlydep ~textnew-component:exe:CSVParsernew-uploader
Dependency ranges changed: base, streamly, text
Files
- Changelog.md +4/−0
- README.md +1/−0
- examples/CSVParser.hs +35/−0
- examples/CoreUtils.hs +6/−5
- examples/FileSystemEvent.hs +1/−1
- examples/Intro.hs +8/−17
- examples/Split.hs +12/−4
- examples/WordCountParallelUTF8.hs +32/−26
- streamly-examples.cabal +14/−6
Changelog.md view
@@ -1,5 +1,9 @@ # Changelog +## 0.1.1 (Jan 2022)++* Update the examples to work with the latest streamly version+ ## 0.1.0 (Jun 2021) * Initial version
README.md view
@@ -69,6 +69,7 @@ version with full UTF8 handling. * [WordFrequency](examples/WordFrequency.hs): Count word frequency in a file and print top 25 words.+* [CSVParser](examples/CSVParser.hs): Process a CSV file ### Networking
+ examples/CSVParser.hs view
@@ -0,0 +1,35 @@+-- ghc -O2 -fspec-constr-recursive=10 -fmax-worker-args=16+-- A simplistic CSV processing example.++import Data.Char (chr)+import Data.Function ((&))+import Data.Word (Word8)+import Streamly.Prelude (SerialT)+import System.Environment (getArgs)+import System.IO (IOMode(..))++import qualified Streamly.Data.Array.Foreign as Array+import qualified Streamly.FileSystem.Handle as Handle+import qualified Streamly.Data.Fold as Fold+import qualified Streamly.Prelude as Stream+import qualified System.IO as IO+import qualified Streamly.Internal.Data.Array.Stream.Foreign as ArrayStream++main :: IO ()+main = do+ inFile <- fmap head getArgs+ src <- IO.openFile inFile ReadMode++ Stream.unfold Handle.readChunks src -- SerialT IO (Array Word8)+ & ArrayStream.splitOn 10 -- SerialT IO (Array Word8)+ & Stream.mapM_ parseLine -- IO ()++ where++ printList = putStr . map (chr . fromIntegral)+ parseLine arr =+ (Stream.unfold Array.read arr :: SerialT IO Word8)+ & Stream.splitOn (== 44) Fold.toList -- SerialT IO [Word8]+ & Stream.intersperse [32] -- SerialT IO [Word8]+ & Stream.mapM_ printList -- IO ()+ >> putStrLn ""
examples/CoreUtils.hs view
@@ -11,8 +11,8 @@ import qualified Streamly.Console.Stdio as Stdio import qualified Streamly.Data.Array.Foreign as Array import qualified Streamly.Data.Fold as Fold-import qualified Streamly.Internal.Data.Stream.IsStream as Stream (splitOnSeq)-import qualified Streamly.Internal.Data.Stream.Parallel as Par (tapAsync)+import qualified Streamly.Internal.Data.Stream.IsStream as Stream+ (splitOnSeq, tapAsyncK) import qualified Streamly.Internal.FileSystem.Dir as Dir (toFiles) import qualified Streamly.Internal.FileSystem.File as File import qualified Streamly.Prelude as Stream@@ -54,9 +54,10 @@ -- output1.txt is processed/written to in a separate thread. tapAsync :: IO () tapAsync =- Stream.unfold Stdio.readChunks () -- SerialT IO (Array Word8)- & Par.tapAsync (File.fromChunks "output1.txt") -- SerialT IO (Array Word8)- & File.fromChunks "output.txt" -- IO ()+ Stream.unfold Stdio.readChunks () -- SerialT IO (Array Word8)+ & Stream.tapAsyncK+ (File.fromChunks "output1.txt") -- SerialT IO (Array Word8)+ & File.fromChunks "output.txt" -- IO () -- | > cat input.txt | tee output1.txt > output.txt tee :: IO ()
examples/FileSystemEvent.hs view
@@ -38,5 +38,5 @@ main = do args <- getArgs paths <- mapM toUtf8 args- Event.watchTrees (NonEmpty.fromList paths)+ Event.watch (NonEmpty.fromList paths) & Stream.mapM_ (putStrLn . Event.showEvent)
examples/Intro.hs view
@@ -49,28 +49,19 @@ -- nested loops (or cross product) when the two streams do not depend on each -- other. The loops fuse completely generating code equivalent to C. crossProduct :: (Int,Int) -> (Int,Int) -> Identity Int-crossProduct (from1,to1) (from2,to2) =+crossProduct range1 range2 = let -- cross multiply src1 and src2 e.g. -- if src1 = [1,2], and src2 = [3,4] then src1 x src2 = -- [(1*3),(1*4),(2*3),(2*4)]- xmult :: Unfold Identity (Int,Int) Int- xmult = Unfold.crossWith (*) src1 src2-- in Stream.unfold xmult (from1,from2) -- SerialT Identity Int- & Stream.fold Fold.sum -- Identity Int-- where-- -- The input to the unfold is (from1,from2)- -- Generate a stream from from1..to1- src1 :: Monad m => Unfold m (Int,Int) Int- src1 = Unfold.lmap fst $ Unfold.enumerateFromToIntegral to1+ xmult :: Unfold Identity ((Int, Int), (Int, Int)) Int+ xmult =+ Unfold.crossWith (*)+ (Unfold.lmap fst Unfold.enumerateFromToIntegral)+ (Unfold.lmap snd Unfold.enumerateFromToIntegral) - -- The input to the unfold is (from1,from2)- -- Generate a stream from from2..to2- src2 :: Monad m => Unfold m (Int,Int) Int- src2 = Unfold.lmap snd $ Unfold.enumerateFromToIntegral to2+ in Stream.unfold xmult (range1,range2) -- SerialT Identity Int+ & Stream.fold Fold.sum -- Identity Int -- | Nested looping similar to 'cross' above but more general and less -- efficient. The second stream may depend on the first stream. The loops
examples/Split.hs view
@@ -1,13 +1,16 @@ import Control.Monad.IO.Class (liftIO) import Control.Monad.State.Strict (StateT(..), get, put) import Data.Function ((&))+import Data.Word (Word8) import System.Environment (getArgs) import System.IO (Handle, IOMode(..), openFile, hClose)+import Streamly.Prelude (SerialT) import qualified Streamly.Prelude as Stream-import qualified Streamly.Internal.Data.Stream.IsStream as Stream (chunksOf2)+import qualified Streamly.Internal.Data.Stream.IsStream as Stream (refoldMany)+import qualified Streamly.Internal.Data.Refold.Type as Refold (take) import qualified Streamly.FileSystem.Handle as Handle-import qualified Streamly.Internal.FileSystem.Handle as Handle (write2)+import qualified Streamly.Internal.FileSystem.Handle as Handle (consumer) newHandle :: StateT (Maybe (Handle, Int)) IO Handle newHandle = do@@ -25,12 +28,17 @@ -- of directory names. splitFile :: Handle -> IO () splitFile inHandle =- Stream.unfold Handle.read inHandle -- SerialT IO Word8+ (Stream.unfold Handle.read inHandle :: SerialT IO Word8) -- SerialT IO Word8 & Stream.liftInner -- SerialT (StateT (Maybe (Handle, Int)) IO) Word8- & Stream.chunksOf2 (180 * 1024 * 1024) newHandle Handle.write2 -- SerialT (StateT (Maybe (Handle, Int)) IO) ()+ -- SerialT (StateT (Maybe (Handle, Int)) IO) ()+ & Stream.refoldMany (Refold.take (180 * mb) Handle.consumer) newHandle & Stream.runStateT (return Nothing) -- SerialT IO (Maybe (Handle, Int), ()) & Stream.map snd -- SerialT IO () & Stream.drain -- SerialT IO ()++ where++ mb = 1024 * 1024 main :: IO () main = do
examples/WordCountParallelUTF8.hs view
@@ -30,21 +30,25 @@ module Main (main, wcMwlParserial) where -import Control.Monad (when, unless)+import Control.Monad (when, unless, void) import Data.Char (isSpace) import Data.Word (Word8) import GHC.Conc (numCapabilities) import System.Environment (getArgs) import System.IO (Handle, openFile, IOMode(..)) -import qualified Data.Vector.Storable.Mutable as V import qualified Streamly.Data.Array.Foreign as Array import qualified Streamly.FileSystem.Handle as Handle+import qualified Streamly.Prelude as Stream++-- Internal modules import qualified Streamly.Internal.Unicode.Stream as Unicode (DecodeState, DecodeError(..), CodePoint, decodeUtf8Either , resumeDecodeUtf8Either)-import qualified Streamly.Prelude as Stream+import qualified Streamly.Internal.Data.Array.Foreign.Mut.Type as MArray+ (getIndexUnsafe, putIndexUnsafe, modifyIndexUnsafe, Array, newArray) + ------------------------------------------------------------------------------- -- Parallel char, line and word counting -------------------------------------------------------------------------------@@ -174,18 +178,20 @@ -- Default/initial state of the block ------------------------------------------------------------------------------- -readField :: V.IOVector Int -> Field -> IO Int-readField v fld = V.read v (fromEnum fld)+readField :: MArray.Array Int -> Field -> IO Int+readField v fld = MArray.getIndexUnsafe v (fromEnum fld) -writeField :: V.IOVector Int -> Field -> Int -> IO ()-writeField v fld = V.write v (fromEnum fld)+writeField :: MArray.Array Int -> Field -> Int -> IO ()+writeField v fld = void . MArray.putIndexUnsafe v (fromEnum fld) -modifyField :: V.IOVector Int -> Field -> (Int -> Int) -> IO ()-modifyField v fld f = V.modify v f (fromEnum fld)+modifyField :: MArray.Array Int -> Field -> (Int -> Int) -> IO ()+modifyField v fld f = do+ let index = fromEnum fld+ MArray.modifyIndexUnsafe v index (\x -> (f x, ())) -newCounts :: IO (V.IOVector Int)+newCounts :: IO (MArray.Array Int) newCounts = do- counts <- V.new (fromEnum (maxBound :: Field) + 1)+ counts <- MArray.newArray (fromEnum (maxBound :: Field) + 1) writeField counts LineCount 0 writeField counts WordCount 0 writeField counts CharCount 0@@ -200,7 +206,7 @@ -- Counting chars ------------------------------------------------------------------------------- -accountChar :: V.IOVector Int -> Bool -> IO ()+accountChar :: MArray.Array Int -> Bool -> IO () accountChar counts isSp = do c <- readField counts CharCount let space = if isSp then 1 else 0@@ -212,7 +218,7 @@ -- Manipulating the header bytes ------------------------------------------------------------------------------- -addToHeader :: V.IOVector Int -> Int -> IO Bool+addToHeader :: MArray.Array Int -> Int -> IO Bool addToHeader counts cp = do cnt <- readField counts HeaderWordCount case cnt of@@ -231,7 +237,7 @@ return True _ -> return False -resetHeaderOnNewChar :: V.IOVector Int -> IO ()+resetHeaderOnNewChar :: MArray.Array Int -> IO () resetHeaderOnNewChar counts = do hdone <- readField counts HeaderDone when (hdone == 0) $ writeField counts HeaderDone 1@@ -240,13 +246,13 @@ -- Manipulating the trailer ------------------------------------------------------------------------------- -setTrailer :: V.IOVector Int -> Unicode.DecodeState -> Unicode.CodePoint -> IO ()+setTrailer :: MArray.Array Int -> Unicode.DecodeState -> Unicode.CodePoint -> IO () setTrailer counts st cp = do writeField counts TrailerState (fromIntegral st) writeField counts TrailerCodePoint cp writeField counts TrailerPresent 1 -resetTrailerOnNewChar :: V.IOVector Int -> IO ()+resetTrailerOnNewChar :: MArray.Array Int -> IO () resetTrailerOnNewChar counts = do trailer <- readField counts TrailerPresent when (trailer /= 0) $ do@@ -258,7 +264,7 @@ ------------------------------------------------------------------------------- {-# INLINE countChar #-}-countChar :: V.IOVector Int -> Either Unicode.DecodeError Char -> IO ()+countChar :: MArray.Array Int -> Either Unicode.DecodeError Char -> IO () countChar counts inp = case inp of Right ch -> do@@ -295,7 +301,7 @@ then accountChar counts True else setTrailer counts st cp -printCounts :: V.IOVector Int -> IO ()+printCounts :: MArray.Array Int -> IO () printCounts v = do l <- readField v LineCount w <- readField v WordCount@@ -312,8 +318,8 @@ -- combine trailing bytes in preceding block with leading bytes in the next -- block and decode them into a codepoint reconstructChar :: Int- -> V.IOVector Int- -> V.IOVector Int+ -> MArray.Array Int+ -> MArray.Array Int -> IO (Stream.SerialT IO (Either Unicode.DecodeError Char)) reconstructChar hdrCnt v1 v2 = do when (hdrCnt > 3 || hdrCnt < 0) $ error "reconstructChar: hdrCnt > 3"@@ -340,7 +346,7 @@ cp <- readField v1 TrailerCodePoint return $ Unicode.resumeDecodeUtf8Either (fromIntegral state) cp stream3 -getHdrChar :: V.IOVector Int -> IO (Maybe Int)+getHdrChar :: MArray.Array Int -> IO (Maybe Int) getHdrChar v = do hdrCnt <- readField v HeaderWordCount case hdrCnt of@@ -366,7 +372,7 @@ -- If the header of the first block is not done then combine the header -- with the header of the next block.-combineHeaders :: V.IOVector Int -> V.IOVector Int -> IO ()+combineHeaders :: MArray.Array Int -> MArray.Array Int -> IO () combineHeaders v1 v2 = do hdone1 <- readField v1 HeaderDone when (hdone1 == 0) $ do@@ -381,7 +387,7 @@ -- the first vector and returning it. -- XXX This is a quick hack and can be refactored to reduce the size -- and understandability considerably.-addCounts :: V.IOVector Int -> V.IOVector Int -> IO (V.IOVector Int)+addCounts :: MArray.Array Int -> MArray.Array Int -> IO (MArray.Array Int) addCounts v1 v2 = do hdone1 <- readField v1 HeaderDone hdone2 <- readField v2 HeaderDone@@ -551,7 +557,7 @@ -- Individual array processing is an isolated loop, fusing it with the bigger -- loop may be counter productive. {-# NOINLINE countArray #-}-countArray :: Array.Array Word8 -> IO (V.IOVector Int)+countArray :: Array.Array Word8 -> IO (MArray.Array Int) countArray src = do counts <- newCounts Stream.mapM_ (countChar counts)@@ -560,7 +566,7 @@ return counts {-# INLINE wcMwlParallel #-}-wcMwlParallel :: Handle -> Int -> IO (V.IOVector Int)+wcMwlParallel :: Handle -> Int -> IO (MArray.Array Int) wcMwlParallel src n = do Stream.foldlM' addCounts newCounts $ Stream.fromAhead@@ -573,7 +579,7 @@ ------------------------------------------------------------------------------- -- -- This is only for perf comparison-wcMwlParserial :: Handle -> IO (V.IOVector Int)+wcMwlParserial :: Handle -> IO (MArray.Array Int) wcMwlParserial src = do counts <- newCounts Stream.mapM_ (countChar counts)
streamly-examples.cabal view
@@ -1,6 +1,6 @@ cabal-version: 2.2 name: streamly-examples-version: 0.1.0+version: 0.1.1 license: Apache-2.0 license-file: LICENSE author: Composewell Technologies@@ -17,7 +17,7 @@ . These examples also serve as a beginner's guide to express practical programs using the dataflow programming (streaming) model. Please- visit the <https://streamly.composewell.com> Streamly homepage> for+ visit the <https://streamly.composewell.com Streamly homepage> for more details and comprehensive documentation. category: Streamly, Streaming, Concurrency, Text, Filesystem, Network, Reactivity@@ -60,8 +60,8 @@ common exe-dependencies build-depends:- streamly >= 0.8.0 && < 0.9- , base >= 4.9 && < 5+ streamly >= 0.8.1 && < 0.8.2+ , base >= 4.9 && < 4.17 , directory >= 1.2 && < 1.4 , transformers >= 0.4 && < 0.6 , containers >= 0.5 && < 0.7@@ -208,8 +208,8 @@ if flag(sdl2) buildable: True build-Depends:- text >= 1.2.3 && < 1.2.5- , sdl2 >= 2.5.0 && < 2.6+ text >= 1.2.3 && < 1.3+ , sdl2 >= 2.5.0 && < 2.6 if os(darwin) frameworks: Cocoa else@@ -288,6 +288,14 @@ executable CamelCase import: exe-options main-is: CamelCase.hs+ if !impl(ghcjs)+ buildable: True+ else+ buildable: False++executable CSVParser+ import: exe-options+ main-is: CSVParser.hs if !impl(ghcjs) buildable: True else