diff --git a/Changelog.md b/Changelog.md new file mode 100644 --- /dev/null +++ b/Changelog.md @@ -0,0 +1,5 @@ +# Changelog + +## 0.1.0 (Jun 2021) + +* Initial version diff --git a/LICENSE b/LICENSE new file mode 100644 --- /dev/null +++ b/LICENSE @@ -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. diff --git a/Makefile b/Makefile new file mode 100644 --- /dev/null +++ b/Makefile @@ -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 diff --git a/NOTICE b/NOTICE new file mode 100644 --- /dev/null +++ b/NOTICE @@ -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 diff --git a/README.md b/README.md new file mode 100644 --- /dev/null +++ b/README.md @@ -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). diff --git a/examples/AcidRain.hs b/examples/AcidRain.hs new file mode 100644 --- /dev/null +++ b/examples/AcidRain.hs @@ -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 diff --git a/examples/CamelCase.hs b/examples/CamelCase.hs new file mode 100644 --- /dev/null +++ b/examples/CamelCase.hs @@ -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 diff --git a/examples/CirclingSquare.hs b/examples/CirclingSquare.hs new file mode 100644 --- /dev/null +++ b/examples/CirclingSquare.hs @@ -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 diff --git a/examples/CmdClient.hs b/examples/CmdClient.hs new file mode 100644 --- /dev/null +++ b/examples/CmdClient.hs @@ -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 () diff --git a/examples/CmdServer.hs b/examples/CmdServer.hs new file mode 100644 --- /dev/null +++ b/examples/CmdServer.hs @@ -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 diff --git a/examples/ControlFlow.hs b/examples/ControlFlow.hs new file mode 100644 --- /dev/null +++ b/examples/ControlFlow.hs @@ -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 diff --git a/examples/CoreUtils.hs b/examples/CoreUtils.hs new file mode 100644 --- /dev/null +++ b/examples/CoreUtils.hs @@ -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 diff --git a/examples/CoreUtilsHandle.hs b/examples/CoreUtilsHandle.hs new file mode 100644 --- /dev/null +++ b/examples/CoreUtilsHandle.hs @@ -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 diff --git a/examples/EchoServer.hs b/examples/EchoServer.hs new file mode 100644 --- /dev/null +++ b/examples/EchoServer.hs @@ -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) diff --git a/examples/FileSender.hs b/examples/FileSender.hs new file mode 100644 --- /dev/null +++ b/examples/FileSender.hs @@ -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 () diff --git a/examples/FileSystemEvent.hs b/examples/FileSystemEvent.hs new file mode 100644 --- /dev/null +++ b/examples/FileSystemEvent.hs @@ -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) diff --git a/examples/Interop/Conduit.hs b/examples/Interop/Conduit.hs new file mode 100644 --- /dev/null +++ b/examples/Interop/Conduit.hs @@ -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 diff --git a/examples/Interop/Pipes.hs b/examples/Interop/Pipes.hs new file mode 100644 --- /dev/null +++ b/examples/Interop/Pipes.hs @@ -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 diff --git a/examples/Interop/Streaming.hs b/examples/Interop/Streaming.hs new file mode 100644 --- /dev/null +++ b/examples/Interop/Streaming.hs @@ -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 diff --git a/examples/Interop/Vector.hs b/examples/Interop/Vector.hs new file mode 100644 --- /dev/null +++ b/examples/Interop/Vector.hs @@ -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 diff --git a/examples/Intro.hs b/examples/Intro.hs new file mode 100644 --- /dev/null +++ b/examples/Intro.hs @@ -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 diff --git a/examples/ListDir.hs b/examples/ListDir.hs new file mode 100644 --- /dev/null +++ b/examples/ListDir.hs @@ -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 diff --git a/examples/MergeServer.hs b/examples/MergeServer.hs new file mode 100644 --- /dev/null +++ b/examples/MergeServer.hs @@ -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 diff --git a/examples/MergeSort.hs b/examples/MergeSort.hs new file mode 100644 --- /dev/null +++ b/examples/MergeSort.hs @@ -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 diff --git a/examples/Rate.hs b/examples/Rate.hs new file mode 100644 --- /dev/null +++ b/examples/Rate.hs @@ -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 () diff --git a/examples/Split.hs b/examples/Split.hs new file mode 100644 --- /dev/null +++ b/examples/Split.hs @@ -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 diff --git a/examples/WordCount.c b/examples/WordCount.c new file mode 100644 --- /dev/null +++ b/examples/WordCount.c @@ -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 +#include +#include +#include + +#include +#include +#include +#include +#include +#include +#include +#include +#include + +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); +} diff --git a/examples/WordCount.hs b/examples/WordCount.hs new file mode 100644 --- /dev/null +++ b/examples/WordCount.hs @@ -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 diff --git a/examples/WordCountModular.hs b/examples/WordCountModular.hs new file mode 100644 --- /dev/null +++ b/examples/WordCountModular.hs @@ -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 diff --git a/examples/WordCountParallel.hs b/examples/WordCountParallel.hs new file mode 100644 --- /dev/null +++ b/examples/WordCountParallel.hs @@ -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 diff --git a/examples/WordCountParallelUTF8.hs b/examples/WordCountParallelUTF8.hs new file mode 100644 --- /dev/null +++ b/examples/WordCountParallelUTF8.hs @@ -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 diff --git a/examples/WordFrequency.hs b/examples/WordFrequency.hs new file mode 100644 --- /dev/null +++ b/examples/WordFrequency.hs @@ -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 diff --git a/examples/WordServer.hs b/examples/WordServer.hs new file mode 100644 --- /dev/null +++ b/examples/WordServer.hs @@ -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 () diff --git a/streamly-examples.cabal b/streamly-examples.cabal new file mode 100644 --- /dev/null +++ b/streamly-examples.cabal @@ -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 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