streamly-examples (empty) → 0.1.0
raw patch · 34 files changed
+3065/−0 lines, 34 filesdep +basedep +conduitdep +containers
Dependencies added: base, conduit, containers, directory, exceptions, fusion-plugin, hashable, mtl, network, pipes, random, sdl2, streaming, streamly, text, transformers, transformers-base, unordered-containers, vector
Files
- Changelog.md +5/−0
- LICENSE +175/−0
- Makefile +10/−0
- NOTICE +10/−0
- README.md +115/−0
- examples/AcidRain.hs +69/−0
- examples/CamelCase.hs +41/−0
- examples/CirclingSquare.hs +86/−0
- examples/CmdClient.hs +50/−0
- examples/CmdServer.hs +95/−0
- examples/ControlFlow.hs +309/−0
- examples/CoreUtils.hs +102/−0
- examples/CoreUtilsHandle.hs +90/−0
- examples/EchoServer.hs +27/−0
- examples/FileSender.hs +45/−0
- examples/FileSystemEvent.hs +42/−0
- examples/Interop/Conduit.hs +19/−0
- examples/Interop/Pipes.hs +26/−0
- examples/Interop/Streaming.hs +23/−0
- examples/Interop/Vector.hs +26/−0
- examples/Intro.hs +205/−0
- examples/ListDir.hs +27/−0
- examples/MergeServer.hs +39/−0
- examples/MergeSort.hs +26/−0
- examples/Rate.hs +11/−0
- examples/Split.hs +39/−0
- examples/WordCount.c +111/−0
- examples/WordCount.hs +56/−0
- examples/WordCountModular.hs +78/−0
- examples/WordCountParallel.hs +66/−0
- examples/WordCountParallelUTF8.hs +600/−0
- examples/WordFrequency.hs +78/−0
- examples/WordServer.hs +46/−0
- streamly-examples.cabal +318/−0
+ Changelog.md view
@@ -0,0 +1,5 @@+# Changelog++## 0.1.0 (Jun 2021)++* Initial version
+ LICENSE view
@@ -0,0 +1,175 @@++ Apache License+ Version 2.0, January 2004+ http://www.apache.org/licenses/++ TERMS AND CONDITIONS FOR USE, REPRODUCTION, AND DISTRIBUTION++ 1. Definitions.++ "License" shall mean the terms and conditions for use, reproduction,+ and distribution as defined by Sections 1 through 9 of this document.++ "Licensor" shall mean the copyright owner or entity authorized by+ the copyright owner that is granting the License.++ "Legal Entity" shall mean the union of the acting entity and all+ other entities that control, are controlled by, or are under common+ control with that entity. For the purposes of this definition,+ "control" means (i) the power, direct or indirect, to cause the+ direction or management of such entity, whether by contract or+ otherwise, or (ii) ownership of fifty percent (50%) or more of the+ outstanding shares, or (iii) beneficial ownership of such entity.++ "You" (or "Your") shall mean an individual or Legal Entity+ exercising permissions granted by this License.++ "Source" form shall mean the preferred form for making modifications,+ including but not limited to software source code, documentation+ source, and configuration files.++ "Object" form shall mean any form resulting from mechanical+ transformation or translation of a Source form, including but+ not limited to compiled object code, generated documentation,+ and conversions to other media types.++ "Work" shall mean the work of authorship, whether in Source or+ Object form, made available under the License, as indicated by a+ copyright notice that is included in or attached to the work+ (an example is provided in the Appendix below).++ "Derivative Works" shall mean any work, whether in Source or Object+ form, that is based on (or derived from) the Work and for which the+ editorial revisions, annotations, elaborations, or other modifications+ represent, as a whole, an original work of authorship. For the purposes+ of this License, Derivative Works shall not include works that remain+ separable from, or merely link (or bind by name) to the interfaces of,+ the Work and Derivative Works thereof.++ "Contribution" shall mean any work of authorship, including+ the original version of the Work and any modifications or additions+ to that Work or Derivative Works thereof, that is intentionally+ submitted to Licensor for inclusion in the Work by the copyright owner+ or by an individual or Legal Entity authorized to submit on behalf of+ the copyright owner. For the purposes of this definition, "submitted"+ means any form of electronic, verbal, or written communication sent+ to the Licensor or its representatives, including but not limited to+ communication on electronic mailing lists, source code control systems,+ and issue tracking systems that are managed by, or on behalf of, the+ Licensor for the purpose of discussing and improving the Work, but+ excluding communication that is conspicuously marked or otherwise+ designated in writing by the copyright owner as "Not a Contribution."++ "Contributor" shall mean Licensor and any individual or Legal Entity+ on behalf of whom a Contribution has been received by Licensor and+ subsequently incorporated within the Work.++ 2. Grant of Copyright License. Subject to the terms and conditions of+ this License, each Contributor hereby grants to You a perpetual,+ worldwide, non-exclusive, no-charge, royalty-free, irrevocable+ copyright license to reproduce, prepare Derivative Works of,+ publicly display, publicly perform, sublicense, and distribute the+ Work and such Derivative Works in Source or Object form.++ 3. Grant of Patent License. Subject to the terms and conditions of+ this License, each Contributor hereby grants to You a perpetual,+ worldwide, non-exclusive, no-charge, royalty-free, irrevocable+ (except as stated in this section) patent license to make, have made,+ use, offer to sell, sell, import, and otherwise transfer the Work,+ where such license applies only to those patent claims licensable+ by such Contributor that are necessarily infringed by their+ Contribution(s) alone or by combination of their Contribution(s)+ with the Work to which such Contribution(s) was submitted. If You+ institute patent litigation against any entity (including a+ cross-claim or counterclaim in a lawsuit) alleging that the Work+ or a Contribution incorporated within the Work constitutes direct+ or contributory patent infringement, then any patent licenses+ granted to You under this License for that Work shall terminate+ as of the date such litigation is filed.++ 4. Redistribution. You may reproduce and distribute copies of the+ Work or Derivative Works thereof in any medium, with or without+ modifications, and in Source or Object form, provided that You+ meet the following conditions:++ (a) You must give any other recipients of the Work or+ Derivative Works a copy of this License; and++ (b) You must cause any modified files to carry prominent notices+ stating that You changed the files; and++ (c) You must retain, in the Source form of any Derivative Works+ that You distribute, all copyright, patent, trademark, and+ attribution notices from the Source form of the Work,+ excluding those notices that do not pertain to any part of+ the Derivative Works; and++ (d) If the Work includes a "NOTICE" text file as part of its+ distribution, then any Derivative Works that You distribute must+ include a readable copy of the attribution notices contained+ within such NOTICE file, excluding those notices that do not+ pertain to any part of the Derivative Works, in at least one+ of the following places: within a NOTICE text file distributed+ as part of the Derivative Works; within the Source form or+ documentation, if provided along with the Derivative Works; or,+ within a display generated by the Derivative Works, if and+ wherever such third-party notices normally appear. The contents+ of the NOTICE file are for informational purposes only and+ do not modify the License. You may add Your own attribution+ notices within Derivative Works that You distribute, alongside+ or as an addendum to the NOTICE text from the Work, provided+ that such additional attribution notices cannot be construed+ as modifying the License.++ You may add Your own copyright statement to Your modifications and+ may provide additional or different license terms and conditions+ for use, reproduction, or distribution of Your modifications, or+ for any such Derivative Works as a whole, provided Your use,+ reproduction, and distribution of the Work otherwise complies with+ the conditions stated in this License.++ 5. Submission of Contributions. Unless You explicitly state otherwise,+ any Contribution intentionally submitted for inclusion in the Work+ by You to the Licensor shall be under the terms and conditions of+ this License, without any additional terms or conditions.+ Notwithstanding the above, nothing herein shall supersede or modify+ the terms of any separate license agreement you may have executed+ with Licensor regarding such Contributions.++ 6. Trademarks. This License does not grant permission to use the trade+ names, trademarks, service marks, or product names of the Licensor,+ except as required for reasonable and customary use in describing the+ origin of the Work and reproducing the content of the NOTICE file.++ 7. Disclaimer of Warranty. Unless required by applicable law or+ agreed to in writing, Licensor provides the Work (and each+ Contributor provides its Contributions) on an "AS IS" BASIS,+ WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or+ implied, including, without limitation, any warranties or conditions+ of TITLE, NON-INFRINGEMENT, MERCHANTABILITY, or FITNESS FOR A+ PARTICULAR PURPOSE. You are solely responsible for determining the+ appropriateness of using or redistributing the Work and assume any+ risks associated with Your exercise of permissions under this License.++ 8. Limitation of Liability. In no event and under no legal theory,+ whether in tort (including negligence), contract, or otherwise,+ unless required by applicable law (such as deliberate and grossly+ negligent acts) or agreed to in writing, shall any Contributor be+ liable to You for damages, including any direct, indirect, special,+ incidental, or consequential damages of any character arising as a+ result of this License or out of the use or inability to use the+ Work (including but not limited to damages for loss of goodwill,+ work stoppage, computer failure or malfunction, or any and all+ other commercial damages or losses), even if such Contributor+ has been advised of the possibility of such damages.++ 9. Accepting Warranty or Additional Liability. While redistributing+ the Work or Derivative Works thereof, You may choose to offer,+ and charge a fee for, acceptance of support, warranty, indemnity,+ or other liability obligations and/or rights consistent with this+ License. However, in accepting such obligations, You may act only+ on Your own behalf and on Your sole responsibility, not on behalf+ of any other Contributor, and only if You agree to indemnify,+ defend, and hold each Contributor harmless for any liability+ incurred by, or claims asserted against, such Contributor by reason+ of your accepting any such warranty or additional liability.
+ Makefile view
@@ -0,0 +1,10 @@+CFLAGS += -std=gnu11 -Wall -Wextra -pedantic -march=native -Ofast++all: examples/WordCount++examples/WordCount: examples/WordCount.c+ $(CC) $(CFLAGS) $< -o $@ $(LDFLAGS)++.PHONY: clean+clean:+ rm -f examples/WordCount
+ NOTICE view
@@ -0,0 +1,10 @@+streamly-examples+Copyright 2019 Composewell Technologies++This product includes software developed at+Composewell Technologies (http://www.composewell.com).++The example AcidRain is adapted from Gabriel Gonzalez's pipes-concurrency+package, which is available under BSD-3-Clause License. The source code of which+can be found at+https://hackage.haskell.org/package/pipes-concurrency-2.0.8/docs/Pipes-Concurrent-Tutorial.html
+ README.md view
@@ -0,0 +1,115 @@+# Streamly Examples++Practical examples to demonstrate the features and performance of+Streamly. Includes examples about file IO, text processing, networking,+concurrent programming, reactive programming and more.++These examples also serve as a beginner's guide to express practical+programs using the dataflow programming (streaming) model. Please visit+the [Streamly homepage](https://streamly.composewell.com) for more+details and comprehensive documentation.++## Running The Examples++Executable names are the same as the filenames. To run an example:++```+$ cabal run AcidRain+```++For performance sensitive examples use fusion-plugin for best performance:++```+$ cabal run --flag fusion-plugin WordCount -- streamly-examples.cabal+```++To run SDL2 based examples, make sure that you have the OS package for+the sdl2 library installed on your system and the headers are visible to+cabal:++```+$ cabal run --flag sdl2 CirclingSquare+```++To run interop examples, use the `interop` build flag:++```+$ cabal run --flag interop Interop.Pipes+```++## List of examples++### General++* [Intro](examples/Intro.hs): Simple, introductory examples - loops, text+ processing, networking, concurrency.+* [MergeSort](examples/MergeSort.hs): Merge sorted streams concurrently.+* [Rate](examples/Rate.hs): Run an action at a given rate.++### FileSystem++* [CoreUtils](examples/CoreUtils.hs): Implement simplified coreutils+ like `cat`, `cp`, `tee`, `grep` using `Streamly.FileSystem.File` API.+* [CoreUtilsHandle](examples/CoreUtilsHandle.hs): Implement simplified+ coreutils using `Streamly.FileSystem.Handle` API.+* [Split](examples/Split.hs): Implement file splitting utility `split`.+* [FileSystemEvent](examples/FileSystemEvent.hs): File watching/fsnotify API+ example.+* [ListDir](examples/ListDir.hs): List a directory tree recursively and+ concurrently.++### Text Processing++* [CamelCase](examples/CamelCase.hs): Convert a file to camel case.+* [WordCount](examples/WordCount.hs): Simple word counting (`wc`) program.+* [WordCount.c](examples/WordCount.c): C equivalent for perf comparison.+* [WordCountModular](examples/WordCountModular.hs): Modular version.+* [WordCountParallel](examples/WordCountParallel.hs): Concurrent version.+* [WordCountParallelUTF8](examples/WordCountParallelUTF8.hs): Concurrent + version with full UTF8 handling.+* [WordFrequency](examples/WordFrequency.hs): Count word frequency in+ a file and print top 25 words.++### Networking++* [EchoServer](examples/EchoServer.hs): A concurrent TCP server that+ echoes everything that it receives.+* [MergeServer](examples/MergeServer.hs): Merges lines received from+ several client connections and writes them to a file.+* [FileSender](examples/FileSender.hs): Send many files concurrently to+ a server over multiple connections. Can be used to test `MergeServer`.+* [CmdServer](examples/CmdServer.hs): Receive a stream of commands from many+ clients and respond to them using command handlers.+* [CmdClient](examples/CmdClient.hs): Run multiple concurrent clients sending+ streams of commands to a server and receiving responses. Can be used to test+ `CmdServer`.+* [WordServer](examples/WordServer.hs): A word look up (dictionary)+ server, instead of performing a real DB query the server just adds a+ time delay to simulate the IO.++### FRP/Games/Animation++* [AcidRain](examples/AcidRain.hs): A console game with deteriorating health+ that can be modified by typing "harm" and "potion" commands.+* [CirclingSquare](examples/CirclingSquare.hs): Use SDL2 to display a+ square that moves in a circle, and follows the mouse pointer.++### Monad Transformers++* [ControlFlow](examples/ControlFlow.hs): Combining control flow manipulating+ transformers (`MaybeT`, `ExceptT`, `ContT`) with streamly.++### Interoperation++* [Interop.Streaming](examples/Interop/Streaming.hs): Converting streamly+ stream type to and from `streaming` stream type.+* [Interop.Pipes](examples/Interop/Pipes.hs): Converting streamly+ stream type to and from `pipes` stream type.+* [Interop.Conduit](examples/Interop/Conduit.hs): Converting streamly+ stream type to and from `conduit` stream type.+* [Interop.Vector](examples/Interop/Vector.hs): Converting streamly+ stream type to and from `vector` stream type.++## Licensing++Available under [Apache-2.0 license](LICENSE).
+ examples/AcidRain.hs view
@@ -0,0 +1,69 @@+{-# LANGUAGE CPP #-}+{-# LANGUAGE FlexibleContexts #-}++-- Copyright : (c) 2017 Harendra Kumar+-- (c) 2013, 2014 Gabriel Gonzalez+--+-- This example is adapted from Gabriel Gonzalez's pipes-concurrency package.+-- https://hackage.haskell.org/package/pipes-concurrency-2.0.8/docs/Pipes-Concurrent-Tutorial.html++#if !(MIN_VERSION_base(4,11,0))+import Data.Semigroup ((<>))+#endif+import Control.Monad (void)+import Control.Monad.IO.Class (MonadIO(liftIO))+import Control.Monad.State (MonadState, get, modify, runStateT)+import Data.Function ((&))+import Streamly.Prelude (MonadAsync, SerialT)++import qualified Streamly.Prelude as Stream++data Event = Quit | Harm Int | Heal Int deriving (Show)++userAction :: MonadAsync m => SerialT m Event+userAction = Stream.repeatM $ liftIO askUser+ where+ askUser = do+ command <- getLine+ case command of+ "potion" -> return (Heal 10)+ "harm" -> return (Harm 10)+ "quit" -> return Quit+ _ -> putStrLn "Type potion or harm or quit" >> askUser++acidRain :: MonadAsync m => SerialT m Event+acidRain =+ Stream.repeatM (liftIO $ return $ Harm 1) -- AsyncT m Event+ & Stream.constRate 1 -- AsyncT m Event+ & Stream.fromAsync -- SerialT m Event++data Result = Check | Done++runEvents :: (MonadAsync m, MonadState Int m) => SerialT m Result+runEvents = do+ event <- userAction `Stream.parallel` acidRain -- SerialT m Event+ case event of+ Harm n -> modify (\h -> h - n) >> return Check+ Heal n -> modify (\h -> h + n) >> return Check+ Quit -> return Done++data Status = Alive | GameOver deriving Eq++getStatus :: (MonadAsync m, MonadState Int m) => Result -> m Status+getStatus result =+ case result of+ Done -> liftIO $ putStrLn "You quit!" >> return GameOver+ Check -> do+ h <- get+ liftIO $ if (h <= 0)+ then putStrLn "You die!" >> return GameOver+ else putStrLn ("Health = " <> show h) >> return Alive++main :: IO ()+main = do+ putStrLn "Your health is deteriorating due to acid rain,\\+ \ type \"potion\" or \"quit\""+ let runGame =+ Stream.mapM getStatus runEvents -- SerialT (StateT Int IO) Status+ & Stream.drainWhile (== Alive) -- StateT Int IO ()+ void $ runStateT runGame 60
+ examples/CamelCase.hs view
@@ -0,0 +1,41 @@+-- ghc -O2 -fspec-constr-recursive=10 -fmax-worker-args=16+-- Convert the input file to camel case and write to stdout++import Data.Function ((&))+import Data.Word (Word8)+import System.Environment (getArgs)+import System.IO (Handle, IOMode(..), openFile, stdout)++import qualified Streamly.Prelude as Stream+import qualified Streamly.FileSystem.Handle as Handle++-- | @camelCase source-file dest-file@+camelCase :: Handle -> Handle -> IO ()+camelCase src dst =+ Stream.unfold Handle.read src -- SerialT IO Word8+ & Stream.scanl' step (True, Nothing) -- SerialT IO (Bool, Maybe Word8)+ & Stream.mapMaybe snd -- SerialT IO Word8+ & Stream.fold (Handle.write dst) -- IO ()++ where++ isNewline x = x == 0x0a+ isUpper x = x >= 0x41 && x <= 0x5a+ isLower x = x >= 0x61 && x <= 0x7a++ -- Scan accumulator @(wasSpace, output)@ contains whether the previous+ -- character was white space, and if the current char should be emitted in+ -- the output.+ step :: (Bool, Maybe Word8) -> Word8 -> (Bool, Maybe Word8)+ step (wasSpace, _) x+ -- Newline or upper case chars go unmodified+ | isNewline x || isUpper x = (False, Just x)+ -- Convert lower to upper if preceded by whitespace+ | isLower x = (False, Just $ if wasSpace then x - 32 else x)+ | otherwise = (True, Nothing)++main :: IO ()+main = do+ name <- fmap head getArgs+ src <- openFile name ReadMode+ camelCase src stdout
+ examples/CirclingSquare.hs view
@@ -0,0 +1,86 @@+-- Adapted from the Yampa package.+-- Displays a square moving in a circle. To move the position drag it with the+-- mouse.++import Control.Concurrent (forkIO)+import Data.Function ((&))+import Data.IORef (IORef, newIORef, readIORef, writeIORef)+import SDL (($=), V2(..), V4(..), Point(..))++import qualified Data.Text as Text+import qualified SDL+import qualified Streamly.Prelude as Stream++------------------------------------------------------------------------------+-- SDL Graphics Init+------------------------------------------------------------------------------++sdlInit :: IO (SDL.Window, SDL.Renderer)+sdlInit = do+ SDL.initializeAll+ window <- SDL.createWindow (Text.pack "Circling Square") SDL.defaultWindow+ rend <- SDL.createRenderer window (-1) SDL.defaultRenderer+ return (window, rend)++------------------------------------------------------------------------------+-- Wait and update Controller Position if it changes+------------------------------------------------------------------------------++handleEvents :: IORef (Double, Double) -> IO Bool+handleEvents ref = do+ e <- SDL.waitEvent+ case SDL.eventPayload e of+ SDL.MouseMotionEvent (SDL.MouseMotionEventData _ _ _ (P (V2 x y)) _) -> do+ writeIORef ref (fromIntegral x, fromIntegral y)+ return True+ SDL.WindowClosedEvent _ ->+ return False+ _ -> return True++------------------------------------------------------------------------------+-- Display a rectangle at given coordinates+------------------------------------------------------------------------------++display :: (Double, Double) -> SDL.Renderer -> IO ()+display (playerX, playerY) rend = do+ -- Paint the screen with background color+ SDL.rendererDrawColor rend $= V4 55 60 64 255+ SDL.fillRect rend Nothing++ -- Paint a small square at the current position+ SDL.rendererDrawColor rend $= V4 212 108 73 255+ let diag = 15+ x = round playerX+ y = round playerY+ SDL.fillRect rend (Just (SDL.Rectangle (P (V2 x y)) diag))++------------------------------------------------------------------------------+-- Periodically refresh the output display+------------------------------------------------------------------------------++updateDisplay :: IORef (Double, Double) -> SDL.Renderer -> IO ()+updateDisplay ref rend = do+ time <- SDL.ticks+ (x, y) <- readIORef ref+ let t = fromIntegral time * speed / 1000+ in display (x + cos t * radius, y + sin t * radius) rend+ SDL.present rend++ where++ speed = 6+ radius = 60++main :: IO ()+main = do+ (window, rend) <- sdlInit+ ref <- newIORef (0,0)++ _ <- forkIO $ do+ Stream.repeatM (updateDisplay ref rend) -- SerialT IO ()+ & Stream.delay (1/60) -- SerialT IO ()+ & Stream.drain -- IO ()++ -- MacOS requires pollEvents to run in the Main thread.+ Stream.repeatM (handleEvents ref) & Stream.drainWhile (== True)+ SDL.destroyWindow window
+ examples/CmdClient.hs view
@@ -0,0 +1,50 @@+module Main (main) where++import Control.Monad (when)+import Data.Function ((&))+import Data.Word (Word8)+import Network.Socket (PortNumber)+import Streamly.Prelude (SerialT)++import qualified Streamly.Data.Fold as Fold+import qualified Streamly.Data.Unfold as Unfold+import qualified Streamly.Internal.Network.Inet.TCP as TCP (processBytes)+import qualified Streamly.Prelude as Stream+import qualified Streamly.Unicode.Stream as Unicode++remoteAddr :: (Word8,Word8,Word8,Word8)+remoteAddr = (192, 168, 1, 4)++remotePort :: PortNumber+remotePort = 8091++chunkSize :: Int+chunkSize = 10000++nChunks :: Int+nChunks = 10++counter :: String -> Int -> () -> IO Int+counter tag n () = do+ let i = n + 1+ when (i `mod` nChunks == 0) $+ putStrLn $ tag ++ show (i * chunkSize)+ return i++sender :: SerialT IO ()+sender =+ Stream.repeat "time\nrandom\n" -- SerialT IO String+ & Stream.unfoldMany Unfold.fromList -- SerialT IO Char+ & Unicode.encodeLatin1 -- SerialT IO Word8+ & TCP.processBytes remoteAddr remotePort -- SerialT IO Word8+ & Unicode.decodeLatin1 -- SerialT IO Char+ & Stream.splitOnSuffix (== '\n') Fold.drain -- SerialT IO String+ & Stream.chunksOf chunkSize Fold.drain -- SerialT IO ()++main :: IO ()+main = do+ Stream.replicate 4 sender -- SerialT IO (SerialT IO ())+ & Stream.concatMapWith Stream.async id -- SerialT IO ()+ & Stream.postscanlM' (counter "rcvd: ")+ (return 0 :: IO Int) -- SerialT IO Int+ & Stream.drain -- IO ()
+ examples/CmdServer.hs view
@@ -0,0 +1,95 @@+-- The server accepts a stream of command words separated by space characters+-- and responds with outputs of the commands.+--+-- To try this example, use "telnet 127.0.0.1 8091" on a terminal and type one+-- or more space separated commands e.g. "time random time" followed by a+-- newline.++{-# LANGUAGE ScopedTypeVariables #-}+{-# LANGUAGE TupleSections #-}++module Main (main) where++import Control.Monad (void)+import Control.Monad.Catch (catch, SomeException)+import Data.Char (isSpace)+import Data.Function ((&))+import Network.Socket (Socket)+import Streamly.Data.Fold (Fold)+import System.Random (randomIO)++import qualified Data.Map.Strict as Map+import qualified Streamly.Data.Fold as Fold+import qualified Streamly.Data.Array.Foreign as Array+import qualified Streamly.Network.Inet.TCP as TCP+import qualified Streamly.Network.Socket as Socket+import qualified Streamly.Prelude as Stream+import qualified Streamly.Unicode.Stream as Unicode++import qualified Streamly.Internal.Data.Fold as Fold (demuxDefault)+import qualified Streamly.Internal.Data.Time.Clock as Clock (getTime, Clock(..))++------------------------------------------------------------------------------+-- Utility functions+------------------------------------------------------------------------------++sendValue :: Show a => Socket -> a -> IO ()+sendValue sk x =+ Stream.fromList (show x ++ "\n")+ & Unicode.encodeLatin1+ & Stream.fold (Array.writeN 60)+ >>= Socket.writeChunk sk++------------------------------------------------------------------------------+-- Command Handlers+------------------------------------------------------------------------------++time :: Socket -> IO ()+time sk = Clock.getTime Clock.Monotonic >>= sendValue sk++random :: Socket -> IO ()+random sk = (randomIO :: IO Int) >>= sendValue sk++def :: (String, Socket) -> IO ()+def (str, sk) = sendValue sk ("Unknown command: " ++ str)++commands :: Map.Map String (Fold IO Socket ())+commands = Map.fromList+ [ ("time" , Fold.drainBy time)+ , ("random", Fold.drainBy random)+ ]++demux :: Fold IO (String, Socket) ()+demux = snd <$> Fold.demuxDefault commands (Fold.drainBy def)++------------------------------------------------------------------------------+-- Parse and handle commands on a socket+------------------------------------------------------------------------------++handler :: Socket -> IO ()+handler sk =+ Stream.unfold Socket.read sk -- SerialT IO Word8+ & Unicode.decodeLatin1 -- SerialT IO Char+ & Stream.wordsBy isSpace Fold.toList -- SerialT IO String+ & Stream.map (, sk) -- SerialT IO (String, Socket)+ & Stream.fold demux -- IO () + Exceptions+ & discard -- IO ()++ where++ discard action = void action `catch` (\(_ :: SomeException) -> return ())++------------------------------------------------------------------------------+-- Accept connecttions and handle connected sockets+------------------------------------------------------------------------------++server :: IO ()+server =+ Stream.unfold TCP.acceptOnPort 8091 -- SerialT IO Socket+ & Stream.fromSerial -- AsyncT IO Socket+ & Stream.mapM (Socket.forSocketM handler) -- AsyncT IO ()+ & Stream.fromAsync -- SerialT IO ()+ & Stream.drain -- IO ()++main :: IO ()+main = server
+ examples/ControlFlow.hs view
@@ -0,0 +1,309 @@+{-# LANGUAGE CPP #-}+{-# LANGUAGE FlexibleContexts #-}+{-# LANGUAGE ScopedTypeVariables #-}++-------------------------------------------------------------------------------+-- Combining control flow manipulating monad transformers (MaybeT, exceptT,+-- ContT) with Streamly+-------------------------------------------------------------------------------+--+-- Streamly streams are non-determinism (nested looping) monads. We can use a+-- control flow monad on top or streamly on top depending on whether we want to+-- superimpose control flow manipulation on top of non-deterministic+-- composition or vice-versa.+--+-- This file provides an example where we enter a sequence of characters "x",+-- and "y" on separate lines, on the command line. When any other sequence is+-- entered the control flow short circuits at the first non-matching char and++-- exits.++#if !(MIN_VERSION_base(4,11,0))+import Data.Semigroup (Semigroup(..))+#endif+import Control.Concurrent (threadDelay)+import Control.Exception (catch, SomeException)+import Control.Monad (when, mzero)+import Control.Monad.Catch (MonadThrow, throwM, Exception)+import Control.Monad.IO.Class (MonadIO (..))+import Control.Monad.Trans.Class (MonadTrans(..))+import Control.Monad.Trans.Maybe (MaybeT (..))+import Control.Monad.Trans.Except (ExceptT, runExceptT, throwE, catchE)+import Control.Monad.Trans.Cont (ContT(..), callCC)+import Streamly.Prelude (IsStream)++import qualified Streamly.Prelude as Stream++-------------------------------------------------------------------------------+-- Using MaybeT below streamly+-------------------------------------------------------------------------------+--+-- When streamly is on top MaybeT would terminate all iterations of+-- non-determinism.+--+getSequenceMaybeBelow+ :: ( IsStream t+ , Monad m+ , MonadTrans t+ , MonadIO (t (MaybeT m))+ )+ => t (MaybeT m) ()+getSequenceMaybeBelow = do+ liftIO $ putStrLn "MaybeT below streamly: Enter one char per line: "++ i <- Stream.fromFoldable [1..2 :: Int]+ liftIO $ putStrLn $ "iteration = " <> show i++ r1 <- liftIO getLine+ when (r1 /= "x") $ lift mzero++ r2 <- liftIO getLine+ when (r2 /= "y") $ lift mzero++mainMaybeBelow :: IO ()+mainMaybeBelow = do+ r <- runMaybeT (Stream.drain getSequenceMaybeBelow)+ case r of+ Just _ -> putStrLn "Bingo"+ Nothing -> putStrLn "Wrong"++-------------------------------------------------------------------------------+-- Using MaybeT above streamly+-------------------------------------------------------------------------------+--+-- When MaybeT is on top a Nothing would terminate only the current iteration+-- of non-determinism below.+--+-- Note that this is redundant configuration as the same behavior can be+-- achieved with just streamly, using mzero.+--+getSequenceMaybeAbove :: (IsStream t, MonadIO (t m)) => MaybeT (t m) ()+getSequenceMaybeAbove = do+ liftIO $ putStrLn "MaybeT above streamly: Enter one char per line: "++ i <- lift $ Stream.fromFoldable [1..2 :: Int]+ liftIO $ putStrLn $ "iteration = " <> show i++ r1 <- liftIO getLine+ when (r1 /= "x") mzero++ r2 <- liftIO getLine+ when (r2 /= "y") mzero++mainMaybeAbove :: (IsStream t, MonadIO (t m)) => MaybeT (t m) ()+mainMaybeAbove = do+ getSequenceMaybeAbove+ liftIO $ putStrLn "Bingo"++-------------------------------------------------------------------------------+-- Using ExceptT below streamly+-------------------------------------------------------------------------------+--+-- XXX need to have a specialized liftCatch to lift catchE+--+-- Note that throwE would terminate all iterations of non-determinism+-- altogether.+getSequenceEitherBelow+ :: ( IsStream t+ , MonadTrans t+ , Monad m+ , MonadIO (t (ExceptT String m))+ )+ => t (ExceptT String m) ()+getSequenceEitherBelow = do+ liftIO $ putStrLn "ExceptT below streamly: Enter one char per line: "++ i <- Stream.fromFoldable [1..2 :: Int]+ liftIO $ putStrLn $ "iteration = " <> show i++ r1 <- liftIO getLine+ when (r1 /= "x") $ lift $ throwE $ "Expecting x got: " <> r1++ r2 <- liftIO getLine+ when (r2 /= "y") $ lift $ throwE $ "Expecting y got: " <> r2++mainEitherBelow :: IO ()+mainEitherBelow = do+ -- XXX Cannot lift catchE+ r <- runExceptT (Stream.drain getSequenceEitherBelow)+ case r of+ Right _ -> liftIO $ putStrLn "Bingo"+ Left s -> liftIO $ putStrLn s++-------------------------------------------------------------------------------+-- Using ExceptT below concurrent streamly+-------------------------------------------------------------------------------+--+-- XXX does not work correctly yet+--+getSequenceEitherAsyncBelow+ :: ( IsStream t+ , MonadTrans t+ , MonadIO m+ , MonadIO (t (ExceptT String m))+ , Semigroup (t (ExceptT String m) Integer)+ )+ => t (ExceptT String m) ()+getSequenceEitherAsyncBelow = do+ liftIO $ putStrLn "ExceptT below concurrent streamly: "++ i <- (liftIO (threadDelay 1000)+ >> lift (throwE "First task")+ >> return 1)+ <> (lift (throwE "Second task") >> return 2)+ <> Stream.fromPure (3 :: Integer)+ liftIO $ putStrLn $ "iteration = " <> show i++mainEitherAsyncBelow :: IO ()+mainEitherAsyncBelow = do+ r <- runExceptT (Stream.drain $ Stream.fromAsync getSequenceEitherAsyncBelow)+ case r of+ Right _ -> liftIO $ putStrLn "Bingo"+ Left s -> liftIO $ putStrLn s++-------------------------------------------------------------------------------+-- Using ExceptT above streamly+-------------------------------------------------------------------------------+--+-- When ExceptT is on top, we can lift the non-determinism of stream from+-- below.+--+-- Note that throwE would terminate/break only current iteration of+-- non-determinism and not all of them altogether.+--+-- Here we can use catchE directly but will have to use monad-control to lift+-- stream operations with stream arguments.+getSequenceEitherAbove :: (IsStream t, MonadIO (t m))+ => ExceptT String (t m) ()+getSequenceEitherAbove = do+ liftIO $ putStrLn "ExceptT above streamly: Enter one char per line: "++ i <- lift $ Stream.fromFoldable [1..2 :: Int]+ liftIO $ putStrLn $ "iteration = " <> show i++ r1 <- liftIO getLine+ when (r1 /= "x") $ throwE $ "Expecting x got: " <> r1++ r2 <- liftIO getLine+ when (r2 /= "y") $ throwE $ "Expecting y got: " <> r2++mainEitherAbove :: (IsStream t, MonadIO (t m)) => ExceptT String (t m) ()+mainEitherAbove =+ catchE (getSequenceEitherAbove >> liftIO (putStrLn "Bingo"))+ (liftIO . putStrLn)++-------------------------------------------------------------------------------+-- Using MonadThrow to throw exceptions in streamly+-------------------------------------------------------------------------------+--+newtype Unexpected = Unexpected String deriving Show++instance Exception Unexpected++-- Note that unlike when ExceptT is used on top, MonadThrow terminates all+-- iterations of non-determinism rather then just the current iteration.+--+getSequenceMonadThrow :: (IsStream t, MonadIO (t m), MonadThrow (t m))+ => t m ()+getSequenceMonadThrow = do+ liftIO $ putStrLn "MonadThrow in streamly: Enter one char per line: "++ i <- Stream.fromFoldable [1..2 :: Int]+ liftIO $ putStrLn $ "iteration = " <> show i++ r1 <- liftIO getLine+ when (r1 /= "x") $ throwM $ Unexpected $ "Expecting x got: " <> r1++ r2 <- liftIO getLine+ when (r2 /= "y") $ throwM $ Unexpected $ "Expecting y got: " <> r2++mainMonadThrow :: IO ()+mainMonadThrow =+ catch (Stream.drain getSequenceMonadThrow >> liftIO (putStrLn "Bingo"))+ (\(e :: SomeException) -> liftIO $ print e)++-------------------------------------------------------------------------------+-- Using ContT below streamly+-------------------------------------------------------------------------------+--+-- CallCC is the goto/setjmp/longjmp equivalent+-- Allows us to manipulate the control flow in arbitrary ways+--+-- XXX need to have a specialized liftCallCC to actually lift callCC+--+getSequenceContBelow+ :: (IsStream t, MonadTrans t, MonadIO m, MonadIO (t (ContT r m)))+ => t (ContT r m) (Either String ())+getSequenceContBelow = do+ liftIO $ putStrLn "ContT below streamly: Enter one char per line: "++ i <- Stream.fromFoldable [1..2 :: Int]+ liftIO $ putStrLn $ "iteration = " <> show i++ r <- lift $ callCC $ \exit -> do+ r1 <- liftIO getLine+ _ <- if r1 /= "x"+ then exit $ Left $ "Expecting x got: " <> r1+ else return $ Right ()++ r2 <- liftIO getLine+ if r2 /= "y"+ then exit $ Left $ "Expecting y got: " <> r2+ else return $ Right ()+ liftIO $ putStrLn $ "done iteration = " <> show i+ return r++mainContBelow+ :: (IsStream t, MonadIO m, MonadTrans t, MonadIO (t (ContT r m)))+ => t (ContT r m) ()+mainContBelow = do+ r <- getSequenceContBelow+ case r of+ Right _ -> liftIO $ putStrLn "Bingo"+ Left s -> liftIO $ putStrLn s++-------------------------------------------------------------------------------+-- Using ContT above streamly+-------------------------------------------------------------------------------+--+getSequenceContAbove :: (IsStream t, MonadIO (t m))+ => ContT r (t m) (Either String ())+getSequenceContAbove = do+ liftIO $ putStrLn "ContT above streamly: Enter one char per line: "++ i <- lift $ Stream.fromFoldable [1..2 :: Int]+ liftIO $ putStrLn $ "iteration = " <> show i++ callCC $ \exit -> do+ r1 <- liftIO getLine+ _ <- if r1 /= "x"+ then exit $ Left $ "Expecting x got: " <> r1+ else return $ Right ()++ r2 <- liftIO getLine+ if r2 /= "y"+ then exit $ Left $ "Expecting y got: " <> r2+ else return $ Right ()++mainContAbove :: (IsStream t, MonadIO (t m)) => ContT r (t m) ()+mainContAbove = do+ r <- getSequenceContAbove+ case r of+ Right _ -> liftIO $ putStrLn "Bingo"+ Left s -> liftIO $ putStrLn s++-------------------------------------------------------------------------------+-- Combining control flow manipulating monad transformers (MaybeT, exceptT,+-- ContT) with Streamly+-------------------------------------------------------------------------------++main :: IO ()+main = do+ mainMaybeBelow+ Stream.drain $ runMaybeT mainMaybeAbove+ runContT (Stream.drain mainContBelow) return+ Stream.drain (runContT mainContAbove return)+ mainEitherBelow+ Stream.drain (runExceptT mainEitherAbove)+ mainMonadThrow+ mainEitherAsyncBelow
+ examples/CoreUtils.hs view
@@ -0,0 +1,102 @@+-- Implement some simple coreutils commands++import Control.Monad (void)+import Data.Char (ord)+import Data.Function ((&))+import Data.Word (Word8)+import System.Environment (getArgs)+import Streamly.Data.Array.Foreign (Array)+import Streamly.Data.Fold (Fold)++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.FileSystem.Dir as Dir (toFiles)+import qualified Streamly.Internal.FileSystem.File as File+import qualified Streamly.Prelude as Stream++-- | > cat input.txt+cat :: IO ()+cat =+ File.toChunks "input.txt" -- SerialT 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.toFiles "dir" -- SerialT IO String+ & Stream.unfoldMany File.readChunks -- SerialT IO (Array Word8)+ & File.fromChunks "output.txt" -- IO ()++-- | > cp input.txt output.txt+cp :: IO ()+cp =+ File.toChunks "input.txt" -- SerialT IO (Array Word8)+ & File.fromChunks "output.txt" -- IO ()++-- | > cat input.txt >> output.txt+append :: IO ()+append =+ File.toChunks "input.txt" -- SerialT IO (Array Word8)+ & File.appendChunks "output.txt" -- IO ()++-- | > cat input.txt | tee output1.txt > output.txt+tap :: IO ()+tap =+ File.toChunks "input.txt" -- SerialT IO (Array Word8)+ & Stream.tap (File.writeChunks "output1.txt") -- SerialT IO (Array Word8)+ & File.fromChunks "output.txt" -- IO ()++-- | > cat input.txt | tee output1.txt > output.txt+-- 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 ()++-- | > cat input.txt | tee output1.txt > output.txt+tee :: IO ()+tee =+ File.toChunks "input.txt" -- SerialT 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) ()++-- | > grep -c "the" input.txt+grepc :: IO ()+grepc = do+ File.toBytes "input.txt" -- SerialT IO Word8+ & Stream.splitOnSeq (Array.fromList pat) Fold.drain -- SerialT IO ()+ & Stream.length -- IO Int+ >>= print . subtract 1 -- IO ()++ where++ pat :: [Word8]+ pat = map (fromIntegral . ord) "the"++main :: IO ()+main = do+ cmd <- fmap head getArgs++ case cmd of+ "cat" -> putStrLn "cat" >> cat+ "catDirTo" -> putStrLn "catDirTo" >> catDirTo+ "cp" -> putStrLn "cp" >> cp+ "append" -> putStrLn "append" >> append+ "tap" -> putStrLn "tap" >> tap+ "tapAsync" -> putStrLn "tapAsync" >> tapAsync+ "tee" -> putStrLn "tee" >> tee+ "grepc" -> putStrLn "grepc" >> grepc+ _ -> putStrLn $ "Unknown command: " ++ cmd
+ examples/CoreUtilsHandle.hs view
@@ -0,0 +1,90 @@+-- This is same as the examples in CoreUtils but uses the Handle API instead of+-- the FilePath based APIs.++import Data.Function ((&))+import System.Environment (getArgs)+import System.IO (IOMode(..), stdin, stdout, Handle, openFile)++import qualified Streamly.Data.Fold as Fold+import qualified Streamly.FileSystem.Handle as Handle+import qualified Streamly.Prelude as Stream+import qualified Streamly.Unicode.Stream as Unicode++import qualified Streamly.Internal.Data.Array.Stream.Foreign as ArrayStream (splitOn)++-- | Read the contents of a file to stdout.+--+-- Handle.read reads the file in 32KB chunks and converts the chunks into a byte+-- stream. FH.write takes the byte stream as input, converts it into chunks of+-- 32KB and writes those chunks to stdout.+--+catBytes :: Handle -> IO ()+catBytes src =+ Stream.unfold Handle.read src -- SerialT IO Word8+ & Stream.fold (Handle.write stdout) -- IO ()++-- | Chunked version, more efficient than the byte stream version above. Reads+-- the file in 256KB chunks and writes those chunks to stdout.+cat :: Handle -> IO ()+cat src =+ Stream.unfold Handle.readChunksWithBufferOf (256*1024, src) -- SerialT IO (Array Word8)+ & Stream.fold (Handle.writeChunks stdout) -- IO ()++-- | Read from standard input write to standard output+echo :: IO ()+echo =+ Stream.unfold Handle.readChunks stdin -- SerialT IO (Array Word8)+ & Stream.fold (Handle.writeChunks stdout) -- IO ()++-- | Copy a source file to a destination file.+--+-- Handle.read reads the file in 32KB chunks and converts the chunks into a byte+-- stream. FH.write takes the byte stream as input, converts it into chunks of+-- 32KB and writes those chunks to the destination file.+cpBytes :: Handle -> Handle -> IO ()+cpBytes src dst =+ Stream.unfold Handle.read src -- SerialT IO Word8+ & Stream.fold (Handle.write dst) -- IO ()++-- | Chunked version, more efficient than the byte stream version above. Reads+-- the file in 256KB chunks and writes those chunks to stdout.+cp :: Handle -> Handle -> IO ()+cp src dst =+ Stream.fold (Handle.writeChunks dst)+ $ Stream.unfold Handle.readChunksWithBufferOf (256*1024, src)++-- | Count lines like wc -l.+--+-- Char stream version. Reads the input as a byte stream, splits it into lines+-- and counts the lines..+wclChar :: Handle -> IO Int+wclChar src =+ Stream.unfold Handle.read src -- SerialT IO Word8+ & Unicode.decodeLatin1 -- SerialT IO Char+ & Stream.splitOnSuffix (== '\n') Fold.drain -- SerialT IO ()+ & Stream.length -- IO ()++-- | More efficient chunked version. Reads chunks from the input handles and+-- splits the chunks directly instead of converting them into byte stream+-- first.+wcl :: Handle -> IO Int+wcl src =+ Stream.unfold Handle.readChunks src -- SerialT IO (Array Word8)+ & ArrayStream.splitOn 10 -- SerialT IO (Array Word8)+ & Stream.length -- IO ()++main :: IO ()+main = do+ cmd <- fmap head getArgs+ src <- openFile "input.txt" ReadMode+ dst <- openFile "output.txt" WriteMode++ case cmd of+ "cat" -> putStrLn "cat" >> cat src+ "catBytes" -> putStrLn "catBytes" >> catBytes src+ "echo" -> putStrLn "echo" >> echo+ "cp" -> putStrLn "cp" >> cp src dst+ "cpBytes" -> putStrLn "cpBytes" >> cpBytes src dst+ "wcl" -> putStrLn "wcl" >> (wcl src >>= print)+ "wclChar" -> putStrLn "wclChar" >> (wclChar src >>= print)+ _ -> putStrLn $ "Unknown command: " ++ cmd
+ examples/EchoServer.hs view
@@ -0,0 +1,27 @@+-- A concurrent TCP server that echoes everything that it receives.++import Control.Monad.Catch (finally)+import Data.Function ((&))+import Network.Socket (Socket)++import qualified Network.Socket as Net+import qualified Streamly.Network.Inet.TCP as TCP+import qualified Streamly.Network.Socket as Socket+import qualified Streamly.Prelude as Stream++main :: IO ()+main =+ Stream.unfold TCP.acceptOnPort 8091 -- ParallelT IO Socket+ & Stream.mapM (handleWithM echo) -- ParallelT IO ()+ & Stream.fromParallel -- SerialT IO ()+ & Stream.drain -- IO ()++ where++ echo :: Socket -> IO ()+ echo sk =+ Stream.unfold Socket.readChunksWithBufferOf (32768, sk) -- SerialT IO (Array Word8)+ & Stream.fold (Socket.writeChunks sk) -- IO ()++ handleWithM :: (Socket -> IO ()) -> Socket -> IO ()+ handleWithM f sk = finally (f sk) (Net.close sk)
+ examples/FileSender.hs view
@@ -0,0 +1,45 @@+{-# LANGUAGE FlexibleContexts #-}+-- A TCP client that does the following:+--+-- o Reads multiple filenames passed on the command line+-- o Opens as many concurrent connections to the server+-- o Sends all the files concurrently to the server++import Data.Function ((&))+import Network.Socket (Socket)+import System.Environment (getArgs)+import System.IO (Handle, withFile, IOMode(..))++import qualified Control.Monad.Catch as Catch+import qualified Network.Socket as Net+import qualified Streamly.FileSystem.Handle as Handle+import qualified Streamly.Network.Inet.TCP as TCP+import qualified Streamly.Network.Socket as Socket+import qualified Streamly.Prelude as Stream++main :: IO ()+main = do+ files <- getArgs+ sendAll files++ where++ connect :: IO Socket+ connect = TCP.connect (127,0,0,1) 8090++ fileToSocket :: Handle -> Socket -> IO ()+ fileToSocket fh sk =+ Stream.unfold Handle.readChunks fh -- SerialT IO (Array Word8)+ & Stream.fold (Socket.writeChunks sk) -- IO ()++ sendFile :: String -> IO ()+ sendFile file =+ withFile file ReadMode $ \fh ->+ Catch.bracket connect Net.close (fileToSocket fh)++ sendAll :: [String] -> IO ()+ sendAll files =+ Stream.fromList files -- ParallelT IO String+ & Stream.mapM sendFile -- ParallelT IO ()+ & Stream.fromParallel -- SerialT IO ()+ & Stream.drain -- IO ()
+ examples/FileSystemEvent.hs view
@@ -0,0 +1,42 @@+{-# LANGUAGE CPP #-}+-- 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.Foreign (Array)++import qualified Data.List.NonEmpty as NonEmpty+import qualified Streamly.Data.Array.Foreign as Array+import qualified Streamly.Prelude as Stream+import qualified Streamly.Unicode.Stream as Unicode++#if darwin_HOST_OS+import qualified Streamly.Internal.FileSystem.Event.Darwin as Event+#elif linux_HOST_OS+import qualified Streamly.Internal.FileSystem.Event.Linux as Event+#elif mingw32_HOST_OS+import qualified Streamly.Internal.FileSystem.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+ Event.watchTrees (NonEmpty.fromList paths)+ & Stream.mapM_ (putStrLn . Event.showEvent)
+ examples/Interop/Conduit.hs view
@@ -0,0 +1,19 @@+{-# LANGUAGE FlexibleContexts #-}++import Streamly.Prelude (IsStream, MonadAsync, SerialT)+import qualified Streamly.Prelude as Stream+import qualified Data.Conduit as Conduit+import qualified Data.Conduit.List as Conduit++-- | conduit to streamly+fromConduit :: (IsStream t, MonadAsync m) => Conduit.ConduitT () a m () -> t m a+fromConduit = Stream.unfoldrM Conduit.unconsM . Conduit.sealConduitT++-- | streamly to conduit+toConduit :: Monad m => SerialT m a -> Conduit.ConduitT i a m ()+toConduit = Conduit.unfoldM Stream.uncons++main :: IO ()+main = do+ Stream.toList (fromConduit (Conduit.sourceList ([1..3]::[Int]))) >>= print+ Conduit.runConduit (toConduit (Stream.fromFoldable ([1..3]::[Int])) Conduit..| Conduit.consume) >>= print
+ examples/Interop/Pipes.hs view
@@ -0,0 +1,26 @@+{-# LANGUAGE FlexibleContexts #-}++import Streamly.Prelude (IsStream, MonadAsync, SerialT)+import qualified Streamly.Prelude as Stream+import qualified Pipes as Pipe+import qualified Pipes.Prelude as Pipe++-- | pipes to streamly+fromPipes :: (IsStream t, MonadAsync m) => Pipe.Producer a m () -> t m a+fromPipes = Stream.unfoldrM unconsP+ where+ -- Adapt Pipe.next to return a Maybe instead of Either+ unconsP p = either (const Nothing) Just <$> Pipe.next p++-- | streamly to pipes+toPipes :: Monad m => SerialT m a -> Pipe.Producer a m ()+toPipes = Pipe.unfoldr unconsEither+ where+ -- Adapt Stream.uncons to return an Either instead of Maybe+ unconsEither :: Monad m => SerialT m a -> m (Either () (a, SerialT m a))+ unconsEither s = maybe (Left ()) Right <$> Stream.uncons s++main :: IO ()+main = do+ Stream.toList (fromPipes (Pipe.each ([1..3]::[Int]))) >>= print+ Pipe.toListM (toPipes (Stream.fromFoldable ([1..3]::[Int]))) >>= print
+ examples/Interop/Streaming.hs view
@@ -0,0 +1,23 @@+{-# LANGUAGE FlexibleContexts #-}++import Streamly.Prelude (IsStream, MonadAsync, SerialT)+import qualified Streamly.Prelude as Stream+import qualified Streaming+import qualified Streaming.Prelude as Streaming++-- | streaming to streamly+fromStreaming :: (IsStream t, MonadAsync m) => Streaming.Stream (Streaming.Of a) m () -> t m a+fromStreaming = Stream.unfoldrM Streaming.uncons+--+-- | streamly to streaming+toStreaming :: Monad m => SerialT m a -> Streaming.Stream (Streaming.Of a) m ()+toStreaming = Streaming.unfoldr unconsEither+ where+ -- Adapt Stream.uncons to return an Either instead of Maybe+ unconsEither :: Monad m => SerialT m a -> m (Either () (a, SerialT m a))+ unconsEither s = maybe (Left ()) Right <$> Stream.uncons s++main :: IO ()+main = do+ Stream.toList (fromStreaming (Streaming.each ([1..3]::[Int]))) >>= print+ Streaming.toList (toStreaming (Stream.fromFoldable ([1..3]::[Int]))) >>= print
+ examples/Interop/Vector.hs view
@@ -0,0 +1,26 @@+{-# LANGUAGE FlexibleContexts #-}++import Streamly.Prelude (MonadAsync, SerialT)+import qualified Streamly.Prelude as Stream+import qualified Data.Vector.Fusion.Stream.Monadic as Vector++-- | vector to streamly+fromVector :: (Stream.IsStream t, MonadAsync m) => Vector.Stream m a -> t m a+fromVector = Stream.unfoldrM unconsV+ where+ unconsV v = do+ r <- Vector.null v+ if r+ then return Nothing+ else do+ h <- Vector.head v+ return $ Just (h, Vector.tail v)++-- | streamly to vector+toVector :: Monad m => SerialT m a -> Vector.Stream m a+toVector = Vector.unfoldrM (Stream.uncons . Stream.adapt)++main :: IO ()+main = do+ Stream.toList (fromVector (Vector.fromList ([1..3]::[Int]))) >>= print+ Vector.toList (toVector (Stream.fromFoldable ([1..3]::[Int]))) >>= print
+ examples/Intro.hs view
@@ -0,0 +1,205 @@+-- Introductory example programs++import Control.Concurrent (threadDelay)+import Data.Char (ord, isSpace)+import Data.Functor.Identity (Identity(..))+import Data.Function ((&))+import Data.Map.Strict (Map)+import Data.Word (Word8)+import System.Environment (getArgs)+import System.IO (stdout)++import Streamly.Prelude (SerialT)+import Streamly.Data.Fold (Fold)+import Streamly.Data.Fold.Tee (Tee(..))+import Streamly.Data.Unfold (Unfold)++import qualified Streamly.Data.Array.Foreign as Array+import qualified Streamly.Data.Fold as Fold+import qualified Streamly.Data.Unfold as Unfold+import qualified Streamly.Prelude as Stream+import qualified Streamly.FileSystem.Handle as Handle+import qualified Streamly.Unicode.Stream as Unicode++import qualified Streamly.Internal.Data.Unfold as Unfold (enumerateFromToIntegral)+import qualified Streamly.Internal.Data.Fold as Fold (classify)+import qualified Streamly.Internal.FileSystem.File as File (toBytes)++-------------------------------------------------------------------------------+-- Simple loops+-------------------------------------------------------------------------------++-- | Sum a list of Int+sumInt :: Identity Int+sumInt =+ Stream.unfold Unfold.fromList [1..10] -- SerialT Identity Int+ & Stream.fold Fold.sum -- Identity Int++-- | Sum a list of Int+sumInt1 :: Identity Int+sumInt1 =+ Stream.fromList [1..10] -- SerialT Identity Int+ & Stream.sum -- Identity Int++-------------------------------------------------------------------------------+-- Nested loops+-------------------------------------------------------------------------------++-- | Nested looping example using unfolds. This is the most efficient way to do+-- 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) =+ 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++ -- 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++-- | Nested looping similar to 'cross' above but more general and less+-- efficient. The second stream may depend on the first stream. The loops+-- cannot fuse completely.+--+nestedLoops :: SerialT IO ()+nestedLoops = do+ x <- Stream.fromList [3,4 :: Int]+ y <- Stream.fromList [1..x]+ Stream.fromEffect $ print (x, y)++-------------------------------------------------------------------------------+-- Text processing+-------------------------------------------------------------------------------++-- | Find average line length for lines in a text file+avgLineLength :: IO Double+avgLineLength =+ File.toBytes "input.txt" -- SerialT IO Word8+ & Stream.splitOnSuffix isNewLine Fold.length -- SerialT IO Int+ & Stream.fold avg -- IO Double++ where++ isNewLine :: Word8 -> Bool+ isNewLine = (== (fromIntegral . ord) '\n')++ toDouble :: Fold IO Int Int -> Fold IO Int Double+ toDouble = fmap (fromIntegral :: Int -> Double)++ avg :: Fold IO Int Double+ avg = toFold $ (/)+ <$> Tee (toDouble Fold.sum)+ <*> Tee (toDouble Fold.length)++-- | Read text from a file and generate a histogram of line length+lineLengthHistogram :: IO (Map Int Int)+lineLengthHistogram =+ File.toBytes "input.txt" -- SerialT IO Word8+ & Stream.splitOnSuffix isNewLine Fold.length -- SerialT IO Int+ & Stream.map bucket -- SerialT IO (Int, Int)+ & Stream.fold (Fold.classify Fold.length) -- IO (Map Int Int)++ where++ isNewLine :: Word8 -> Bool+ isNewLine = (== (fromIntegral . ord) '\n')++ bucket :: Int -> (Int, Int)+ bucket n = let i = n `mod` 10 in if i > 9 then (9,n) else (i,n)++-- | Read text from a file and generate a histogram of word length+wordLengthHistogram :: IO (Map Int Int)+wordLengthHistogram =+ File.toBytes "input.txt" -- SerialT IO Word8+ & Unicode.decodeLatin1 -- SerialT IO Char+ & Stream.wordsBy isSpace Fold.length -- SerialT IO Int+ & Stream.map bucket -- SerialT IO (Int, Int)+ & Stream.fold (Fold.classify Fold.length) -- IO (Map (Int, Int))++ where++ bucket :: Int -> (Int, Int)+ bucket n = let i = n `mod` 10 in if i > 9 then (9,n) else (i,n)++-------------------------------------------------------------------------------+-- Network/Concurrency+-------------------------------------------------------------------------------++-- Simulate network/db query by adding a delay+fetch :: String -> IO (String, String)+fetch w = threadDelay 1000000 >> return (w,w)++wordList :: [String]+wordList = ["cat", "dog", "mouse"]++meanings :: [IO (String, String)]+meanings = map fetch wordList++-- | Fetch word meanings for words in 'wordList'. All searches are performed+-- concurrently.+--+getWords :: IO ()+getWords =+ Stream.fromListM meanings -- AheadT IO (String, String)+ & Stream.fromAhead -- SerialT IO (String, String)+ & Stream.map show -- SerialT IO String+ & unlinesBy "\n" -- SerialT IO String+ & Stream.map Array.fromList -- SerialT IO (Array Word8)+ & Stream.fold (Handle.writeChunks stdout) -- IO ()++ where unlinesBy = Stream.intercalateSuffix (Unfold.function id)++-------------------------------------------------------------------------------+-- Main+-------------------------------------------------------------------------------++-- Usage example: Intro sumInt+--+-- Some examples (e.g. avgLineLength) require a text file "input.txt" in+-- the current directory.+--+main :: IO ()+main = do+ cmd <- fmap head getArgs++ case cmd of+ "sumInt" -> do+ putStrLn "sumInt"+ print (runIdentity sumInt)+ "sumInt1" -> do+ putStrLn "sumInt1"+ print (runIdentity sumInt1)+ "crossProduct" -> do+ putStrLn "crossProduct"+ print $ runIdentity $ crossProduct (1,1000) (1000,2000)+ "nestedLoops" -> do+ putStrLn "nestedLoops"+ Stream.drain nestedLoops+ "avgLineLength" -> do+ putStrLn "avgLineLength"+ avgLineLength >>= print+ "lineLengthHistogram" -> do+ putStrLn "lineLengthHistogram"+ lineLengthHistogram >>= print+ "wordLengthHistogram" -> do+ putStrLn "wordLengthHistogram"+ wordLengthHistogram >>= print+ "getWords" -> do+ putStrLn "getWords"+ getWords >>= print+ _ -> putStrLn $ "Unknown command: " ++ cmd
+ examples/ListDir.hs view
@@ -0,0 +1,27 @@+module Main (main) where++import Data.Bifunctor (bimap)+import Data.Function ((&))+import Streamly.Prelude (SerialT)+import System.IO (stdout, hSetBuffering, BufferMode(LineBuffering))++import qualified Streamly.Prelude as Stream+import qualified Streamly.Internal.Data.Stream.IsStream as Stream+ (iterateMapLeftsWith)+import qualified Streamly.Internal.FileSystem.Dir as Dir (toEither)++-- Lists a dir as a stream of (Either Dir File)+listDir :: String -> SerialT IO (Either String String)+listDir dir =+ Dir.toEither dir -- SerialT IO (Either String String)+ & Stream.map (bimap mkAbs mkAbs) -- SerialT IO (Either String String)++ where mkAbs x = dir ++ "/" ++ x++-- | List the current directory recursively using concurrent processing+main :: IO ()+main = do+ hSetBuffering stdout LineBuffering+ let start = Stream.fromPure (Left ".")+ Stream.iterateMapLeftsWith Stream.ahead listDir start+ & Stream.mapM_ print
+ examples/MergeServer.hs view
@@ -0,0 +1,39 @@+import Control.Monad.IO.Class (liftIO)+import Data.Function ((&))+import Network.Socket (Socket, close)+import Streamly.Data.Array.Foreign (Array)+import Streamly.Prelude (SerialT)+import System.IO (Handle, withFile, IOMode(..))++import qualified Streamly.Data.Array.Foreign as Array+import qualified Streamly.FileSystem.Handle as Handle+import qualified Streamly.Network.Socket as Socket+import qualified Streamly.Network.Inet.TCP as TCP+import qualified Streamly.Prelude as Stream+import qualified Streamly.Unicode.Stream as Unicode++-- | Read a line stream from a socket. Note, lines are buffered, we could add+-- a limit to the buffering for safety.+readLines :: Socket -> SerialT IO (Array Char)+readLines sk =+ Stream.unfold Socket.read sk -- SerialT IO Word8+ & Unicode.decodeLatin1 -- SerialT IO Char+ & Stream.splitWithSuffix (== '\n') Array.write -- SerialT IO String++recv :: Socket -> SerialT IO (Array Char)+recv sk = Stream.finally (liftIO $ close sk) (readLines sk)++-- | Starts a server at port 8091 listening for lines with space separated+-- words. Multiple clients can connect to the server and send streams of lines.+-- The server handles all the connections concurrently, merges the incoming+-- streams at line boundaries and writes the merged stream to a file.+server :: Handle -> IO ()+server file =+ Stream.unfold TCP.acceptOnPort 8090 -- SerialT IO Socket+ & Stream.concatMapWith Stream.parallel recv -- SerialT IO (Array Char)+ & Stream.unfoldMany Array.read -- SerialT IO Char+ & Unicode.encodeLatin1 -- SerialT IO Word8+ & Stream.fold (Handle.write file) -- IO ()++main :: IO ()+main = withFile "output.txt" AppendMode server
+ examples/MergeSort.hs view
@@ -0,0 +1,26 @@+{-# LANGUAGE FlexibleContexts #-}++-- This example generates two random streams sorted in ascending order and+-- merges them in ascending order, concurrently.+--+-- Compile with '-threaded -with-rtsopts "-N"' GHC options to use the+-- parallelism.++import Data.Word (Word16)+import Streamly.Prelude (SerialT)+import System.Random (getStdGen, randoms)++import qualified Data.List as List+import qualified Streamly.Prelude as Stream++getRandomSorted :: IO (SerialT IO Word16)+getRandomSorted = do+ g <- getStdGen+ let ls = take 100000 (randoms g) :: [Word16]+ return $ Stream.fromList (List.sort ls)++main :: IO ()+main = do+ s1 <- getRandomSorted+ s2 <- getRandomSorted+ Stream.last (Stream.mergeAsyncBy compare s1 s2) >>= print
+ examples/Rate.hs view
@@ -0,0 +1,11 @@+import Data.Function ((&))+import qualified Streamly.Prelude as Stream+import qualified Streamly.Internal.Data.Stream.IsStream as Stream (timestamped)++main :: IO ()+main =+ Stream.repeatM (pure "tick") -- AsyncT IO String+ & Stream.timestamped -- AsyncT IO (AbsTime, String)+ & Stream.avgRate 1 -- AsyncT IO (AbsTime, String)+ & Stream.fromAsync -- SerialT IO (AbsTime, String)+ & Stream.mapM_ print -- IO ()
+ examples/Split.hs view
@@ -0,0 +1,39 @@+import Control.Monad.IO.Class (liftIO)+import Control.Monad.State.Strict (StateT(..), get, put)+import Data.Function ((&))+import System.Environment (getArgs)+import System.IO (Handle, IOMode(..), openFile, hClose)++import qualified Streamly.Prelude as Stream+import qualified Streamly.Internal.Data.Stream.IsStream as Stream (chunksOf2)+import qualified Streamly.FileSystem.Handle as Handle+import qualified Streamly.Internal.FileSystem.Handle as Handle (write2)++newHandle :: StateT (Maybe (Handle, Int)) IO Handle+newHandle = do+ old <- get+ idx <- case old of+ Nothing -> return 0+ Just (h, i) -> liftIO (hClose h) >> return (i + 1)+ h <- liftIO $ openFile ("output-" ++ show idx ++ ".txt") WriteMode+ put (Just (h, idx))+ return h++-- XXX reduce the input stream to a stream of file names+-- The fold can return the file name/handle after it is done.+-- similarly the files can written to directories and we can generate a stream+-- of directory names.+splitFile :: Handle -> IO ()+splitFile inHandle =+ Stream.unfold Handle.read inHandle -- 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) ()+ & Stream.runStateT (return Nothing) -- SerialT IO (Maybe (Handle, Int), ())+ & Stream.map snd -- SerialT IO ()+ & Stream.drain -- SerialT IO ()++main :: IO ()+main = do+ name <- fmap head getArgs+ src <- openFile name ReadMode+ splitFile src
+ examples/WordCount.c view
@@ -0,0 +1,111 @@+/*+ * Copyright (c) 1980, 1987, 1991, 1993+ * The Regents of the University of California. All rights reserved.+ *+ * Redistribution and use in source and binary forms, with or without+ * modification, are permitted provided that the following conditions+ * are met:+ * 1. Redistributions of source code must retain the above copyright+ * notice, this list of conditions and the following disclaimer.+ * 2. Redistributions in binary form must reproduce the above copyright+ * notice, this list of conditions and the following disclaimer in the+ * documentation and/or other materials provided with the distribution.+ * 3. All advertising materials mentioning features or use of this software+ * must display the following acknowledgement:+ * This product includes software developed by the University of+ * California, Berkeley and its contributors.+ * 4. Neither the name of the University nor the names of its contributors+ * may be used to endorse or promote products derived from this software+ * without specific prior written permission.+ *+ * THIS SOFTWARE IS PROVIDED BY THE REGENTS AND CONTRIBUTORS ``AS IS'' AND+ * ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE+ * IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE+ * ARE DISCLAIMED. IN NO EVENT SHALL THE REGENTS OR CONTRIBUTORS BE LIABLE+ * FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL+ * DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS+ * OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION)+ * HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT+ * LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY+ * OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF+ * SUCH DAMAGE.+ */++#include <sys/param.h>+#include <sys/mount.h>+#include <sys/stat.h>+#include <sys/statfs.h>++#include <ctype.h>+#include <err.h>+#include <errno.h>+#include <fcntl.h>+#include <stdint.h>+#include <stdio.h>+#include <stdlib.h>+#include <string.h>+#include <unistd.h>++int main(int argc, char *argv[])+{+ if (argc != 2) {+ puts("usage: bsd-wc FILE");+ exit(EXIT_FAILURE);+ }++ struct statfs fsb;+ uintmax_t linect, wordct, charct;+ int fd, len;+ short gotsp;+ uint8_t *p;+ uint8_t *buf;++ linect = wordct = charct = 0;++ if ((fd = open(argv[1], O_RDONLY, 0)) < 0) {+ perror("open");+ exit(EXIT_FAILURE);+ }++ if (fstatfs(fd, &fsb)) {+ perror("fstatfs");+ exit(EXIT_FAILURE);+ }++ buf = malloc(fsb.f_bsize);+ if (!buf) {+ perror("malloc");+ exit(EXIT_FAILURE);+ }++ gotsp = 1;+ while ((len = read(fd, buf, fsb.f_bsize)) != 0) {+ if (len == -1) {+ perror("read");+ exit(EXIT_FAILURE);+ }+ p = buf;+ while (len > 0) {+ uint8_t ch = *p;+ charct++;+ len -= 1;+ p += 1;+ if (ch == '\n')+ ++linect;+ if (isspace(ch))+ gotsp = 1;+ else if (gotsp) {+ gotsp = 0;+ ++wordct;+ }+ }+ }++ printf(" %7ju", linect);+ printf(" %7ju", wordct);+ printf(" %7ju", charct);++ printf(" %s\n", argv[1]);++ close(fd);+}
+ examples/WordCount.hs view
@@ -0,0 +1,56 @@+-- To run this program:+--+-- cabal run --flag fusion-plugin WordCount test-data.txt+--+module WordCount (main, count, Counts(..), isSpace) where++import Data.Char (ord)+import Data.Function ((&))+import System.Environment (getArgs)++import qualified Streamly.Internal.FileSystem.File as File+import qualified Streamly.Prelude as Stream+import qualified Streamly.Unicode.Stream as Stream++-------------------------------------------------------------------------------+-- C compatible isSpace+-------------------------------------------------------------------------------++{-# INLINE isSpace #-}+isSpace :: Char -> Bool+isSpace c = uc == 0x20 || uc - 0x9 <= 4+ where uc = fromIntegral (ord c) :: Word++-------------------------------------------------------------------------------+-- Counting+-------------------------------------------------------------------------------++-- Counts lines words chars lastCharWasSpace+data Counts = Counts !Int !Int !Int !Bool deriving Show++{-# INLINE count #-}+count :: Counts -> Char -> Counts+count (Counts l w c wasSpace) ch =+ let l1 = if ch == '\n' then l + 1 else l+ (w1, wasSpace1) =+ if isSpace ch+ then (w, True)+ else (if wasSpace then w + 1 else w, False)+ in Counts l1 w1 (c + 1) wasSpace1++wc :: String -> IO Counts+wc file =+ Stream.unfold File.read file -- SerialT IO Word8+ & Stream.decodeLatin1 -- SerialT IO Char+ -- & Stream.decodeUtf8 -- SerialT IO Char+ & Stream.foldl' count (Counts 0 0 0 True) -- IO Counts++-------------------------------------------------------------------------------+-- Main+-------------------------------------------------------------------------------++main :: IO ()+main = do+ name <- fmap head getArgs+ Counts l w c _ <- wc name+ putStrLn $ show l ++ " " ++ show w ++ " " ++ show c ++ " " ++ name
+ examples/WordCountModular.hs view
@@ -0,0 +1,78 @@+-- To run this program:+--+-- cabal run --flag fusion-plugin WordCountModular test-data.txt+--+module Main (main) where++import Data.Char (chr, ord)+import Data.Function ((&))+import Data.Word (Word8)+import Streamly.Data.Fold (Fold)+import Streamly.Data.Fold.Tee (Tee(..))+import System.Environment (getArgs)++import qualified Streamly.Data.Fold as Fold+import qualified Streamly.Data.Fold.Tee as Tee+import qualified Streamly.Internal.FileSystem.File as File (toBytes)+import qualified Streamly.Prelude as Stream++{-# INLINE isSpace #-}+isSpace :: Char -> Bool+isSpace c = uc == 0x20 || uc - 0x9 <= 4+ where uc = fromIntegral (ord c) :: Word++-------------------------------------------------------------------------------+-- Fold the bytes from an input file+-------------------------------------------------------------------------------++-- The fold accepts a stream of `Word8` and returns a value of type "a".+foldWith :: Fold IO Word8 a -> String -> IO a+foldWith f file =+ File.toBytes file -- SerialT IO Word8+ & Stream.fold f -- IO a++-------------------------------------------------------------------------------+-- Count Lines+-------------------------------------------------------------------------------++-- ASCII character 10 is newline+countl :: Int -> Word8 -> Int+countl n ch = if ch == 10 then n + 1 else n++nlines :: Monad m => Fold m Word8 Int+nlines = Fold.foldl' countl 0++-------------------------------------------------------------------------------+-- Count Words+-------------------------------------------------------------------------------++countw :: (Int, Bool) -> Word8 -> (Int, Bool)+countw (n, wasSpace) ch =+ if isSpace $ chr $ fromIntegral ch+ then (n, True)+ else (if wasSpace then n + 1 else n, False)++-- The fold accepts a stream of `Word8` and returns a word count (`Int`)+nwords :: Monad m => Fold m Word8 Int+nwords = fst <$> Fold.foldl' countw (0, True)++-------------------------------------------------------------------------------+-- Count Bytes, Lines, Words+-------------------------------------------------------------------------------++-- The fold accepts a stream of `Word8` and returns the three counts+countAll :: Fold IO Word8 (Int, Int, Int)+countAll = Tee.toFold $ (,,) <$> Tee Fold.length <*> Tee nlines <*> Tee nwords++-------------------------------------------------------------------------------+-- Main+-------------------------------------------------------------------------------++main :: IO ()+main = do+ name <- fmap head getArgs+ -- 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+ putStrLn $ show l ++ " " ++ show w ++ " " ++ show c ++ " " ++ name
+ examples/WordCountParallel.hs view
@@ -0,0 +1,66 @@+-- To run this program:+--+-- cabal run --flag fusion-plugin WordCountParallel test-data.txt+--+import Data.Char (chr)+import Data.Function ((&))+import Data.Word (Word8)+import GHC.Conc (numCapabilities)+import System.Environment (getArgs)+import Streamly.Data.Array.Foreign (Array)+import WordCount (count, Counts(..), isSpace)++import qualified Streamly.Data.Array.Foreign as Array+import qualified Streamly.Internal.FileSystem.File as File (readChunks)+import qualified Streamly.Prelude as Stream+import qualified Streamly.Unicode.Stream as Stream++-- Get the line, word, char counts in one chunk.+countArray :: Array Word8 -> IO Counts+countArray arr =+ Stream.unfold Array.read arr -- SerialT IO Word8+ & Stream.decodeLatin1 -- SerialT IO Char+ & Stream.foldl' count (Counts 0 0 0 True) -- IO Counts++-- When combining the counts in two contiguous chunks, we would also need to+-- know whether the first element of the next chunk was a space char or+-- non-space to know whether the same word is continuing to the next chunk or+-- if it is a new word. So add that too, giving (firstCharWasSpace, Counts).+{-# NOINLINE partialCounts #-}+partialCounts :: Array Word8 -> IO (Bool, Counts)+partialCounts arr = do+ let r = Array.getIndex arr 0+ case r of+ Just x -> do+ counts <- countArray arr+ return (isSpace (chr (fromIntegral x)), counts)+ Nothing -> return (False, Counts 0 0 0 True)++-- Combine the counts from two consecutive chunks+addCounts :: (Bool, Counts) -> (Bool, Counts) -> (Bool, Counts)+addCounts (sp1, Counts l1 w1 c1 ws1) (sp2, Counts l2 w2 c2 ws2) =+ let wcount =+ if not ws1 && not sp2 -- no space between two chunks+ then w1 + w2 - 1+ else w1 + w2+ in (sp1, Counts (l1 + l2) wcount (c1 + c2) ws2)++-- 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 file = do+ Stream.unfold File.readChunks file -- AheadT IO (Array Word8)+ & Stream.mapM partialCounts -- AheadT IO (Bool, Counts)+ & Stream.maxThreads numCapabilities -- AheadT IO (Bool, Counts)+ & Stream.fromAhead -- SerialT IO (Bool, Counts)+ & Stream.foldl' addCounts (False, Counts 0 0 0 True) -- IO (Bool, Counts)++-------------------------------------------------------------------------------+-- Main+-------------------------------------------------------------------------------++main :: IO ()+main = do+ name <- fmap head getArgs+ (_, Counts l w c _) <- wc name+ putStrLn $ show l ++ " " ++ show w ++ " " ++ show c ++ " " ++ name
+ examples/WordCountParallelUTF8.hs view
@@ -0,0 +1,600 @@+-------------------------------------------------------------------------------+-- Fast, streaming and parallel word counting (wc) program.+-------------------------------------------------------------------------------+-- 1) On utf8 inputs the serial version is around 3x faster than MacOS wc+-- 2) It can run parallely on multiple cores providing further speedup+-- 3) Parallel version works efficiently on stdin/streaming input as well+-- 4) Parallel version handles utf8 input correctly (including multi-byte space+-- chars) and gives the same output as the serial version on all inputs.+-- 5) There may be differences in word/char counts when there are invalid utf8+-- byte sequences present in the input because of different styles of error+-- handling.++-------------------------------------------------------------------------------+-- Build with the following options:+-------------------------------------------------------------------------------+-- streamly optimization plugin is required for best performance+--+-- cabal build --flag fusion-plugin WordCountUTF8++-------------------------------------------------------------------------------+-- Comparing with "wc -mwl" command:+-------------------------------------------------------------------------------+--+-- 1) To enable UTF8 with wc: export LANG=en_US.UTF-8; export LC_ALL=$LANG+-- 2) To test whether it is actually using utf8, copy and paste this string+-- "U+1680 U+2000 U+2001 U+2002" and run "wc -mwl" on this. Without proper UTF8+-- handling word count would be 1, with proper UTF8 handling word count would+-- be 4. Note that the spaces in this string are not regular space chars they+-- are different unicode space chars.++module Main (main, wcMwlParserial) where++import Control.Monad (when, unless)+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.Internal.Unicode.Stream as Unicode+ (DecodeState, DecodeError(..), CodePoint, decodeUtf8Either+ , resumeDecodeUtf8Either)+import qualified Streamly.Prelude as Stream++-------------------------------------------------------------------------------+-- Parallel char, line and word counting+-------------------------------------------------------------------------------++-- We process individual chunks in the stream independently and parallely and+-- the combine the chunks to combine what they have counted.+--+-------------------------------------------------------------------------------+-- Char counting+-------------------------------------------------------------------------------++-- To count chars each block needs the following:+--+-- -- | header | char counts | trailer |+--+-- header and trailer are incomplete utf8 byte sequences that may be combined+-- with the previous or the next block to complete them later.+--+-- The trailer may have one or more bytes in a valid utf8 sequence and is+-- expecting more bytes to complete the sequence. The header stores any+-- possible continuation from the previous block. It contains a maximum of 3+-- bytes which all must be non-starter bytes.+--+-- When two blocks are combined, the trailer of the first block is combined+-- with the header of the next block and then utf8 decoded. The combined+-- header+trailer may yield:+--+-- * Nothing - when there is no trailer and header+-- * All errors - when there is no trailer in the previous block, and there is+-- a header in the next block. In this case there is no starting char which+-- means all header bytes are errors.+-- * It can yield at most one valid character followed by 0, 1 or 2 errors.+--+-- We count an incomplete utf8 sequence of 2 or more bytes starting with a+-- valid starter byte as a single codepoint. Bytes not following a valid+-- starter byte are treated as individual codepoints for counting.+--+-------------------------------------------------------------------------------+-- Word counting+-------------------------------------------------------------------------------++-- For word counting we need the following in each block:+--+-- -- | header | startsWithSpace | word counts | endsWithSpace | trailer |+--+-- The word counts in individual blocks are performed assuming that the+-- previous char before the block is a space.+-- When combining two blocks, after combining the trailer of previous blocks+-- with the header of the next we determine if the resulting char is a space or+-- not.+--+-- 1) If there is no new char joining the two blocks then we use endsWithSpace+-- of the previous block and startsWithSpace of the next block to determine if+-- the word counts are to be adjusted. If the previous block ends with+-- non-space and the next block starts with non-space we need to decrement the+-- word count by one if it is non-zero in the next block.+--+-- 2) If the new joining char is a space then we combine it with+-- startsWithSpace and endsWithSpace to determine the+-- startsWithSpace/endsWithSpace of the combined block and adjust the word+-- counts appropriately.+--+-------------------------------------------------------------------------------+-- Line counting+-------------------------------------------------------------------------------++-- Line counting is performed by counting "\n" in the stream. No new "\n" can+-- result from patching the trailer and header as it is always a single byte.++-------------------------------------------------------------------------------+-- Counting state+-------------------------------------------------------------------------------++-- We use a mutable vector for the counting state. A fold using an immutable+-- structure for such a large state does not perform well. However, mutability+-- is confined to just the accumulator.+--+-- XXX we need convenient mutable records (like C structs) to handle things+-- like this. It may be possible to achieve the same performance with an+-- immutable accumulator, but that will require more research. Since we are+-- always discarding the previous state, we can perhaps make use of that memory+-- using safe in-place modifications, without having to allocate new memory.++-- XXX we can also count the number of decoding errors separately+data Field =+ -- The number of "\n" characters found in the block.+ LineCount+ -- Number of full words found in the block, words are counted on a+ -- transition from space char to a non-space char. We always assume the+ -- char before the first starter char in a block is a space. If this is+ -- found to be incorrect when joining two blocks then we fix the counts+ -- accordingly.+ | WordCount+ -- The number of successfully decoded characters plus the number of+ -- decoding failures in the block. Each byte or sequence of bytes on which+ -- decoding fails is also counted as one char. The header and trailer bytes+ -- are not accounted in this, they are accounted only when we join two+ -- blocks.+ | CharCount+ -- whether the last counted char in this block was a space char+ | WasSpace+ -- whether the first successfully decoded char in this block is a space. A+ -- decoding failure, after the trailing bytes from the previous block are+ -- accounted, is also considered as space.+ | FirstIsSpace+ -- If no starter byte is found in the first three bytes in the block then+ -- store those bytes to possibly combine them with the trailing incomplete+ -- byte sequence in the previous block. We mark it done when either we have+ -- stored three bytes or we have found a starter byte.+ --+ -- XXX This is ugly to manipulate, we can implement a statically max sized+ -- mutable ring structure within this record.+ | HeaderDone+ | HeaderWordCount+ | HeaderWord1+ | HeaderWord2+ | HeaderWord3+ -- If a byte sequence at the end of the block is not complete then store+ -- the current state of the utf8 decoder to continue it later using the+ -- incomplete leading byte sequence in the next block.+ | TrailerPresent+ | TrailerState+ | TrailerCodePoint+ deriving (Show, Enum, Bounded)++-------------------------------------------------------------------------------+-- Default/initial state of the block+-------------------------------------------------------------------------------++readField :: V.IOVector Int -> Field -> IO Int+readField v fld = V.read v (fromEnum fld)++writeField :: V.IOVector Int -> Field -> Int -> IO ()+writeField v fld = V.write v (fromEnum fld)++modifyField :: V.IOVector Int -> Field -> (Int -> Int) -> IO ()+modifyField v fld f = V.modify v f (fromEnum fld)++newCounts :: IO (V.IOVector Int)+newCounts = do+ counts <- V.new (fromEnum (maxBound :: Field) + 1)+ writeField counts LineCount 0+ writeField counts WordCount 0+ writeField counts CharCount 0+ writeField counts WasSpace 1+ writeField counts FirstIsSpace 0+ writeField counts HeaderDone 0+ writeField counts HeaderWordCount 0+ writeField counts TrailerPresent 0+ return counts++-------------------------------------------------------------------------------+-- Counting chars+-------------------------------------------------------------------------------++accountChar :: V.IOVector Int -> Bool -> IO ()+accountChar counts isSp = do+ c <- readField counts CharCount+ let space = if isSp then 1 else 0+ when (c == 0) $ writeField counts FirstIsSpace space+ writeField counts CharCount (c + 1)+ writeField counts WasSpace space++-------------------------------------------------------------------------------+-- Manipulating the header bytes+-------------------------------------------------------------------------------++addToHeader :: V.IOVector Int -> Int -> IO Bool+addToHeader counts cp = do+ cnt <- readField counts HeaderWordCount+ case cnt of+ 0 -> do+ writeField counts HeaderWord1 cp+ writeField counts HeaderWordCount 1+ return True+ 1 -> do+ writeField counts HeaderWord2 cp+ writeField counts HeaderWordCount 2+ return True+ 2 -> do+ writeField counts HeaderWord3 cp+ writeField counts HeaderWordCount 3+ writeField counts HeaderDone 1+ return True+ _ -> return False++resetHeaderOnNewChar :: V.IOVector Int -> IO ()+resetHeaderOnNewChar counts = do+ hdone <- readField counts HeaderDone+ when (hdone == 0) $ writeField counts HeaderDone 1++-------------------------------------------------------------------------------+-- Manipulating the trailer+-------------------------------------------------------------------------------++setTrailer :: V.IOVector 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 counts = do+ trailer <- readField counts TrailerPresent+ when (trailer /= 0) $ do+ writeField counts TrailerPresent 0+ accountChar counts True++-------------------------------------------------------------------------------+-- Counting the stream+-------------------------------------------------------------------------------++{-# INLINE countChar #-}+countChar :: V.IOVector Int -> Either Unicode.DecodeError Char -> IO ()+countChar counts inp =+ case inp of+ Right ch -> do+ resetHeaderOnNewChar counts+ -- account the last stored error as whitespace and clear it+ resetTrailerOnNewChar counts++ when (ch == '\n') $ modifyField counts LineCount (+ 1)+ if isSpace ch+ then accountChar counts True+ else do+ wasSpace <- readField counts WasSpace+ when (wasSpace /= 0) $ modifyField counts WordCount (+ 1)+ accountChar counts False+ Left (Unicode.DecodeError st cp) -> do+ hdone <- readField counts HeaderDone+ if hdone == 0+ then do+ if st == 0+ then do+ -- We got a non-starter in initial decoder state, there may+ -- be something that comes before this to complete it.+ r <- addToHeader counts cp+ unless r $ error "countChar: Bug addToHeader failed"+ else do+ -- We got an error in a non-initial decoder state, it may+ -- be an input underflow error, keep it as incomplete in+ -- the trailer.+ writeField counts HeaderDone 1+ setTrailer counts st cp+ else do+ resetTrailerOnNewChar counts+ if st == 0+ then accountChar counts True+ else setTrailer counts st cp++printCounts :: V.IOVector Int -> IO ()+printCounts v = do+ l <- readField v LineCount+ w <- readField v WordCount+ c <- readField v CharCount+ putStrLn $ show l ++ " " ++ show w ++ " " ++ show c++-------------------------------------------------------------------------------+-- Parallel counting+-------------------------------------------------------------------------------++-- XXX we need a better data structure to store the header bytes to make these+-- routines simpler.+--+-- 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+ -> IO (Stream.SerialT IO (Either Unicode.DecodeError Char))+reconstructChar hdrCnt v1 v2 = do+ when (hdrCnt > 3 || hdrCnt < 0) $ error "reconstructChar: hdrCnt > 3"+ stream1 <-+ if hdrCnt > 2+ then do+ x <- readField v2 HeaderWord3+ return $ (fromIntegral x :: Word8) `Stream.cons` Stream.nil+ else return Stream.nil+ stream2 <-+ if hdrCnt > 1+ then do+ x <- readField v2 HeaderWord2+ return $ fromIntegral x `Stream.cons` stream1+ else return stream1+ stream3 <-+ if hdrCnt > 0+ then do+ x <- readField v2 HeaderWord1+ return $ fromIntegral x `Stream.cons` stream2+ else return stream2++ state <- readField v1 TrailerState+ cp <- readField v1 TrailerCodePoint+ return $ Unicode.resumeDecodeUtf8Either (fromIntegral state) cp stream3++getHdrChar :: V.IOVector Int -> IO (Maybe Int)+getHdrChar v = do+ hdrCnt <- readField v HeaderWordCount+ case hdrCnt of+ 0 -> return Nothing+ 1 -> do+ writeField v HeaderWordCount 0+ Just <$> readField v HeaderWord1+ 2 -> do+ x1 <- readField v HeaderWord1+ x2 <- readField v HeaderWord2+ writeField v HeaderWord1 x2+ writeField v HeaderWordCount 1+ return $ Just x1+ 3 -> do+ x1 <- readField v HeaderWord1+ x2 <- readField v HeaderWord2+ x3 <- readField v HeaderWord3+ writeField v HeaderWord1 x2+ writeField v HeaderWord2 x3+ writeField v HeaderWordCount 2+ return $ Just x1+ _ -> error "getHdrChar: Bug, hdrCnt not in range 0-3"++-- 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 v1 v2 = do+ hdone1 <- readField v1 HeaderDone+ when (hdone1 == 0) $ do+ res <- getHdrChar v2+ case res of+ Nothing -> return ()+ Just x -> do+ r <- addToHeader v1 x+ unless r $ error "combineHeaders: Bug, addToHeader failed"++-- We combine the contents of the second vector into the first vector, mutating+-- 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 v1 v2 = do+ hdone1 <- readField v1 HeaderDone+ hdone2 <- readField v2 HeaderDone+ hdrCnt2_0 <- readField v2 HeaderWordCount+ if hdone1 == 0 && (hdrCnt2_0 /= 0 || hdone2 /= 0)+ then do+ combineHeaders v1 v2+ if hdone2 == 0+ then return v1+ else do+ writeField v1 HeaderDone 1+ addCounts v1 v2+ else do+ trailerPresent2 <- readField v2 TrailerPresent+ trailerState2 <- readField v2 TrailerState+ trailerCodePoint2 <- readField v2 TrailerCodePoint+ if hdone1 == 0+ then error "addCounts: Bug, trying to add completely empty second block"+ else do+ l1 <- readField v1 LineCount+ w1 <- readField v1 WordCount+ c1 <- readField v1 CharCount+ wasSpace1 <- readField v1 WasSpace++ l2 <- readField v2 LineCount+ w2 <- readField v2 WordCount+ c2 <- readField v2 CharCount+ wasSpace2 <- readField v2 WasSpace+ firstIsSpace2 <- readField v2 FirstIsSpace+ hdrCnt2 <- readField v2 HeaderWordCount++ trailer1 <- readField v1 TrailerPresent+ if trailer1 == 0 -- no trailer in the first block+ then do+ -- header2, if any, complete or incomplete, is just invalid+ -- bytes, count them as whitespace+ let firstIsSpace2' = firstIsSpace2 /= 0 || hdrCnt2 /= 0+ w <-+ if w2 /= 0 && wasSpace1 == 0 && not firstIsSpace2'+ then return $ w1 + w2 - 1+ else return $ w1 + w2+ writeField v1 LineCount (l1 + l2)+ writeField v1 WordCount w+ writeField v1 CharCount (c1 + c2 + hdrCnt2)++ when (c1 == 0) $ do+ if c2 == 0 && hdrCnt2 /= 0+ then writeField v1 FirstIsSpace 1+ else writeField v1 FirstIsSpace firstIsSpace2++ if c2 == 0 && hdrCnt2 /= 0+ then writeField v1 WasSpace 1+ else writeField v1 WasSpace wasSpace2++ writeField v1 TrailerPresent trailerPresent2+ writeField v1 TrailerState trailerState2+ writeField v1 TrailerCodePoint trailerCodePoint2+ return v1+ else do+ if hdrCnt2 == 0+ then do+ when (hdone2 /= 0) $ do -- empty and Done header+ -- count trailer as whitespace, its counted as one char+ -- Note: hdrCnt2 == 0 means either header is not done+ -- or c2 /= 0+ writeField v1 LineCount (l1 + l2)+ writeField v1 WordCount (w1 + w2)+ writeField v1 CharCount (c1 + c2 + 1)++ when (c1 == 0) $ writeField v1 FirstIsSpace 1+ if c2 == 0+ then writeField v1 WasSpace 1+ else writeField v1 WasSpace wasSpace2++ writeField v1 TrailerPresent trailerPresent2+ writeField v1 TrailerState trailerState2+ writeField v1 TrailerCodePoint trailerCodePoint2+ -- If header of the next block is not done we continue the+ -- trailer from the previous block instead of treating it+ -- as whitespace+ return v1+ else do+ -- join the trailing part of the first block with the+ -- header of the next+ decoded <- reconstructChar hdrCnt2 v1 v2+ res <- Stream.uncons decoded+ case res of+ Nothing -> error "addCounts: Bug. empty reconstructed char"+ Just (h, t) -> do+ tlength <- Stream.length t+ case h of+ Right ch -> do+ -- If we have an error case after this+ -- char then that would be treated+ -- as whitespace+ let lcount = l1 + l2+ lcount1 = if ch == '\n' then lcount + 1 else lcount+ wcount = w1 + w2+ firstSpace = isSpace ch+ wasSpace = firstSpace || tlength /= 0+ wcount1+ | wasSpace = wcount+ | w2 == 0 || firstIsSpace2 /= 0 = wcount+ | otherwise = wcount - 1+ writeField v1 LineCount lcount1+ writeField v1 WordCount wcount1+ writeField v1 CharCount (c1 + c2 + 1 + tlength)++ when (c1 == 0) $+ writeField v1 FirstIsSpace+ (if firstSpace then 1 else 0)++ if c2 == 0+ then do+ if wasSpace+ then writeField v1 WasSpace 1+ else writeField v1 WasSpace 0+ else writeField v1 WasSpace wasSpace2++ writeField v1 TrailerPresent trailerPresent2+ writeField v1 TrailerState trailerState2+ writeField v1 TrailerCodePoint trailerCodePoint2+ return v1+ Left (Unicode.DecodeError st cp) -> do+ -- if header was incomplete it may result+ -- in partially decoded char to be written+ -- as trailer. Check if the last error is+ -- an incomplete decode.+ r <- Stream.last t+ let (st', cp') =+ case r of+ Nothing -> (st, cp)+ Just lst -> case lst of+ Right _ -> error "addCounts: Bug"+ Left (Unicode.DecodeError st1 cp1) -> (st1, cp1)+ if hdone2 == 0 && st' /= 12+ then do+ -- all elements before the last one must be errors+ writeField v1 CharCount (c1 + tlength)++ when (c1 == 0) $+ writeField v1 FirstIsSpace 1++ writeField v1 WasSpace 1++ writeField v1 TrailerState (fromIntegral st')+ writeField v1 TrailerCodePoint cp'+ else do+ -- all elements must be errors+ -- treat them as whitespace+ writeField v1 LineCount (l1 + l2)+ writeField v1 WordCount (w1 + w2)+ writeField v1 CharCount (c1 + c2 + tlength + 1)++ when (c1 == 0) $+ writeField v1 FirstIsSpace 1++ if c2 == 0+ then writeField v1 WasSpace 1+ else writeField v1 WasSpace wasSpace2++ writeField v1 TrailerPresent trailerPresent2+ writeField v1 TrailerState trailerState2+ writeField v1 TrailerCodePoint trailerCodePoint2+ return v1++-- 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 src = do+ counts <- newCounts+ Stream.mapM_ (countChar counts)+ $ Unicode.decodeUtf8Either+ $ Stream.unfold Array.read src+ return counts++{-# INLINE wcMwlParallel #-}+wcMwlParallel :: Handle -> Int -> IO (V.IOVector Int)+wcMwlParallel src n = do+ Stream.foldlM' addCounts newCounts+ $ Stream.fromAhead+ $ Stream.maxThreads numCapabilities+ $ Stream.mapM countArray+ $ Stream.unfold Handle.readChunksWithBufferOf (n, src)++-------------------------------------------------------------------------------+-- Serial counting using parallel version of countChar+-------------------------------------------------------------------------------+--+-- This is only for perf comparison+wcMwlParserial :: Handle -> IO (V.IOVector Int)+wcMwlParserial src = do+ counts <- newCounts+ Stream.mapM_ (countChar counts)+ $ Unicode.decodeUtf8Either+ $ Stream.unfold Handle.read src+ return counts++-------------------------------------------------------------------------------+-- Main+-------------------------------------------------------------------------------++main :: IO ()+main = do+ name <- fmap head getArgs+ src <- openFile name ReadMode+ -- printCounts =<< wcMwlParserial src -- Unix wc -l program+ -- Using different sizes of chunks (1,2,3,4,5,10,128,256) is a good testing+ -- mechanism for parallel counting code.+ {-+ args <- getArgs+ let chunkSize = read $ args !! 1+ -}+ let chunkSize = 32 * 1024+ printCounts =<< wcMwlParallel src chunkSize -- Unix wc -l program
+ examples/WordFrequency.hs view
@@ -0,0 +1,78 @@+{-# LANGUAGE FlexibleInstances #-}++{-# OPTIONS_GHC -Wno-orphans #-}++import Data.Char (isSpace)+import Data.Foldable (traverse_)+import Data.Function ((&))+import Data.Functor.Identity (Identity(..))+import Data.Hashable (Hashable(..))+import Data.IORef (newIORef, readIORef, modifyIORef')+import Foreign.Storable (Storable(..))+import System.Environment (getArgs)++import qualified Data.Char as Char+import qualified Data.HashMap.Strict as Map+import qualified Data.List as List+import qualified Data.Ord as Ord+import qualified Streamly.Data.Array.Foreign as Array+import qualified Streamly.Data.Fold as Fold+import qualified Streamly.Internal.FileSystem.File as File (toBytes)+import qualified Streamly.Prelude as Stream+import qualified Streamly.Unicode.Stream as Unicode++hashArray :: (Storable a, Integral b, Num c) =>+ Fold.Fold Identity a b -> Array.Array a -> c+hashArray f arr =+ Stream.unfold Array.read arr+ & Stream.fold f+ & fromIntegral . runIdentity++instance (Enum a, Storable a) => Hashable (Array.Array a) where+ hash = hashArray Fold.rollingHash++ hashWithSalt salt =+ hashArray (Fold.rollingHashWithSalt (fromIntegral salt))++{-# INLINE toLower #-}+toLower :: Char -> Char+toLower c+ | uc >= 0x61 && uc <= 0x7a = c+ | otherwise = Char.toLower c+ where+ uc = fromIntegral (Char.ord c) :: Word++{-# INLINE isAlpha #-}+isAlpha :: Char -> Bool+isAlpha c+ | uc >= 0x61 && uc <= 0x7a = True+ | otherwise = Char.isAlpha c+ where+ uc = fromIntegral (Char.ord c) :: Word++main :: IO ()+main = do+ inFile <- fmap head getArgs++ -- Write the stream to a hashmap consisting of word counts+ mp <-+ let+ alter Nothing = Just <$> newIORef (1 :: Int)+ alter (Just ref) = modifyIORef' ref (+ 1) >> return (Just ref)+ in File.toBytes inFile -- SerialT IO Word8+ & Unicode.decodeLatin1 -- SerialT IO Char+ & Stream.map toLower -- SerialT IO Char+ & Stream.wordsBy isSpace Fold.toList -- SerialT IO String+ & Stream.filter (all isAlpha) -- SerialT IO String+ & Stream.foldlM' (flip (Map.alterF alter)) (return Map.empty) -- IO (Map String (IORef Int))++ -- Print the top hashmap entries+ counts <-+ let readRef (w, ref) = do+ cnt <- readIORef ref+ return (w, cnt)+ in Map.toList mp+ & mapM readRef++ traverse_ print $ List.sortOn (Ord.Down . snd) counts+ & List.take 25
+ examples/WordServer.hs view
@@ -0,0 +1,46 @@+import Control.Concurrent (threadDelay)+import Control.Exception (finally)+import Data.Char (isSpace)+import Data.Function ((&))+import Network.Socket (Socket, close)++import qualified Streamly.Prelude as Stream+import qualified Streamly.Data.Fold as Fold+import qualified Streamly.Network.Socket as Socket+import qualified Streamly.Network.Inet.TCP as TCP+import qualified Streamly.Unicode.Stream as Unicode++-- Simulate network/db query by adding a delay+fetch :: String -> IO (String, String)+fetch w = threadDelay 1000000 >> return (w,w)++-- Read lines of whitespace separated list of words from a socket, fetch the+-- meanings of each word concurrently and return the meanings separated by+-- newlines, in same order as the words were received. Repeat until the+-- connection is closed.+lookupWords :: Socket -> IO ()+lookupWords sk =+ Stream.unfold Socket.read sk -- SerialT IO Word8+ & Unicode.decodeLatin1 -- SerialT IO Char+ & Stream.wordsBy isSpace Fold.toList -- SerialT IO String+ & Stream.fromSerial -- AheadT IO String+ & Stream.mapM fetch -- AheadT IO (String, String)+ & Stream.fromAhead -- SerialT IO (String, String)+ & Stream.map show -- SerialT IO String+ & Stream.intersperse "\n" -- SerialT IO String+ & Unicode.encodeStrings Unicode.encodeLatin1 -- SerialT IO (Array Word8)+ & Stream.fold (Socket.writeChunks sk) -- IO ()++serve :: Socket -> IO ()+serve sk = finally (lookupWords sk) (close sk)++-- | Run a server on port 8091. Accept and handle connections concurrently. The+-- connection handler is "serve" (i.e. lookupWords). You can use "telnet" or+-- "nc" as a client to try it out.+main :: IO ()+main =+ Stream.unfold TCP.acceptOnPort 8091 -- SerialT IO Socket+ & Stream.fromSerial -- AsyncT IO ()+ & Stream.mapM serve -- AsyncT IO ()+ & Stream.fromAsync -- SerialT IO ()+ & Stream.drain -- IO ()
+ streamly-examples.cabal view
@@ -0,0 +1,318 @@+cabal-version: 2.2+name: streamly-examples+version: 0.1.0+license: Apache-2.0+license-file: LICENSE+author: Composewell Technologies+maintainer: streamly@composewell.com+homepage: https://streamly.composewell.com+bug-reports: http://github.com/composewell/streamly-examples/issues+copyright: Copyright (c) 2019 Composewell Technologies+synopsis: Examples for Streamly+description:+ Practical examples to demonstrate the features and performance+ of Streamly. Includes examples about file IO, text processing,+ networking, concurrent programming, reactive programming and+ more.+ .+ 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+ more details and comprehensive documentation.+category:+ Streamly, Streaming, Concurrency, Text, Filesystem, Network, Reactivity+stability: provisional+tested-with: GHC==9.0.1, GHC==8.10.5, GHC==8.8.4, GHC==8.6.5+build-type: Simple+extra-source-files:+ Changelog.md+ Makefile+ NOTICE+ README.md+ examples/WordCount.c++source-repository head+ type: git+ location: git://github.com/composewell/streamly-examples.git++------------------------------------------------------------------------------+-- Flags --+------------------------------------------------------------------------------++flag fusion-plugin+ description: Use fusion plugin for for best performance+ manual: True+ default: False++flag sdl2+ description: Include graphics examples using SDL2+ manual: True+ default: False++flag interop+ description: Include examples of inter-operation with other libraries+ manual: True+ default: False++------------------------------------------------------------------------------+-- Common Stanzas --+------------------------------------------------------------------------------++common exe-dependencies+ build-depends:+ streamly >= 0.8.0 && < 0.9+ , base >= 4.9 && < 5+ , directory >= 1.2 && < 1.4+ , transformers >= 0.4 && < 0.6+ , containers >= 0.5 && < 0.7+ , random >= 1.0.0 && < 2+ , exceptions >= 0.8 && < 0.11+ , transformers-base >= 0.4 && < 0.5+ , network >= 2.6 && < 4+ , hashable >= 1.2 && < 1.4+ , unordered-containers >= 0.2 && < 0.3+ , vector >= 0.12 && < 0.13+ , mtl >= 2.2 && < 3++common exe-options+ import: exe-dependencies+ default-language: Haskell2010+ hs-source-dirs: examples++ ghc-options: -O2+ -fdicts-strict+ -fspec-constr-recursive=16+ -fmax-worker-args=16++ -Wall+ -Wcompat+ -Wunrecognised-warning-flags+ -Widentities+ -Wincomplete-record-updates+ -Wincomplete-uni-patterns+ -Wredundant-constraints+ -Wnoncanonical-monad-instances++ if flag(fusion-plugin) && !impl(ghcjs) && !impl(ghc < 8.6)+ ghc-options: -fplugin Fusion.Plugin+ build-depends:+ fusion-plugin >= 0.2 && < 0.3++common exe-options-threaded+ import: exe-options+ ghc-options: -threaded+ -with-rtsopts=-N++------------------------------------------------------------------------------+-- Examples --+------------------------------------------------------------------------------++executable Interop.Vector+ import: exe-options-threaded+ main-is: Interop/Vector.hs+ if !impl(ghcjs) && flag(interop)+ buildable: True+ else+ buildable: False++executable Interop.Pipes+ import: exe-options-threaded+ main-is: Interop/Pipes.hs+ if !impl(ghcjs) && flag(interop)+ buildable: True+ build-depends: pipes >= 4.3.16 && < 4.4+ else+ buildable: False++executable Interop.Streaming+ import: exe-options-threaded+ main-is: Interop/Streaming.hs+ if !impl(ghcjs) && flag(interop)+ buildable: True+ build-depends: streaming >= 0.2.3.0 && < 0.3+ else+ buildable: False++executable Interop.Conduit+ import: exe-options-threaded+ main-is: Interop/Conduit.hs+ if !impl(ghcjs) && flag(interop)+ buildable: True+ build-depends: conduit >= 1.3.4.1 && < 1.3.5+ else+ buildable: False++executable Intro+ import: exe-options-threaded+ main-is: Intro.hs+ if !impl(ghcjs)+ buildable: True+ else+ buildable: False++executable WordServer+ import: exe-options-threaded+ main-is: WordServer.hs+ if !impl(ghcjs)+ buildable: True+ else+ buildable: False++executable MergeServer+ import: exe-options-threaded+ main-is: MergeServer.hs+ if !impl(ghcjs)+ buildable: True+ else+ buildable: False++executable CmdServer+ import: exe-options-threaded+ main-is: CmdServer.hs+ if !impl(ghcjs)+ buildable: True+ else+ buildable: False++executable CmdClient+ import: exe-options-threaded+ main-is: CmdClient.hs+ if !impl(ghcjs)+ buildable: True+ else+ buildable: False++-- executable SearchQuery+-- import: exe-options-threaded+-- main-is: SearchQuery.hs+-- if !impl(ghcjs)+-- buildable: True+-- else+-- buildable: False++executable ListDir+ import: exe-options-threaded+ main-is: ListDir.hs++executable MergeSort+ import: exe-options-threaded+ main-is: MergeSort.hs++executable AcidRain+ import: exe-options-threaded+ main-is: AcidRain.hs++executable CirclingSquare+ import: exe-options-threaded+ main-is: CirclingSquare.hs+ if flag(sdl2)+ buildable: True+ build-Depends:+ text >= 1.2.3 && < 1.2.5+ , sdl2 >= 2.5.0 && < 2.6+ if os(darwin)+ frameworks: Cocoa+ else+ buildable: False++executable ControlFlow+ import: exe-options+ main-is: ControlFlow.hs++executable CoreUtilsHandle+ import: exe-options+ main-is: CoreUtilsHandle.hs++executable CoreUtils+ import: exe-options+ main-is: CoreUtils.hs++executable EchoServer+ import: exe-options-threaded+ main-is: EchoServer.hs+ if !impl(ghcjs)+ buildable: True+ else+ buildable: False++executable FileSender+ import: exe-options-threaded+ main-is: FileSender.hs+ if !impl(ghcjs)+ buildable: True+ else+ buildable: False++executable WordFrequency+ import: exe-options+ main-is: WordFrequency.hs+ if !impl(ghcjs)+ buildable: True+ else+ buildable: False++executable WordCount+ import: exe-options+ main-is: WordCount.hs+ ghc-options: -main-is WordCount+ if !impl(ghcjs)+ buildable: True+ else+ buildable: False++executable WordCountParallel+ import: exe-options-threaded+ main-is: WordCountParallel.hs+ other-modules: WordCount+ if !impl(ghcjs)+ buildable: True+ else+ buildable: False++executable WordCountModular+ import: exe-options+ main-is: WordCountModular.hs+ if !impl(ghcjs)+ buildable: True+ else+ buildable: False++executable WordCountParallelUTF8+ import: exe-options-threaded+ main-is: WordCountParallelUTF8.hs+ if !impl(ghcjs)+ buildable: True+ else+ buildable: False++executable CamelCase+ import: exe-options+ main-is: CamelCase.hs+ if !impl(ghcjs)+ buildable: True+ else+ buildable: False++executable Rate+ import: exe-options-threaded+ main-is: Rate.hs+ if !impl(ghcjs)+ buildable: True+ else+ buildable: False++executable Split+ import: exe-options+ main-is: Split.hs+ if !impl(ghcjs)+ buildable: True+ else+ buildable: False++executable FileSystemEvent+ import: exe-options-threaded+ main-is: FileSystemEvent.hs+ if !impl(ghcjs)+ buildable: True+ else+ buildable: False