@@ -55,9 +55,7 @@ As an expository device, we have indicated the types at the intermediate
5555stages of stream computations as comments in the examples below.
5656The meaning of these types are:
5757
58−* A `SerialT IO a` is a serial stream of values of type `a` in the IO Monad.
59−* An `AsyncT IO a` is a concurrent (asynchronous) stream of values of type
60− `a` in the IO Monad.
58+* A `Stream IO a` is a serial stream of values of type `a` in the IO Monad.
6159* An `Unfold IO a b` is a representation of a function that converts a seed
6260 value of type `a` into a stream of values of type `b` in the IO Monad.
6361* A `Fold IO a b` is a representation of a function that converts a stream of
@@ -98,14 +96,16 @@ program, including the imports that we have omitted here.
9896
9997We start with a code fragment that counts the number of bytes in a file:
10098
101−```haskell
99+```haskell ghci
100+import Data.Function ((&))
101+
102102import qualified Streamly.Data.Fold as Fold
103+import qualified Streamly.Data.Stream as Stream
103104import qualified Streamly.Internal.FileSystem.File as File
104−import qualified Streamly.Prelude as Stream
105105
106106wcb :: String -> IO Int
107107wcb file =
108− File.toBytes file -- SerialT IO Word8
108+ File.read file -- Stream IO Word8
109109 & Stream.fold Fold.length -- IO Int
110110```
111111
@@ -113,7 +113,10 @@ wcb file =
113113
114114The next code fragment shows how to count the number of lines in a file:
115115
116−```haskell
116+```haskell ghci
117+import Data.Word (Word8)
118+import Streamly.Data.Fold (Fold)
119+
117120-- ASCII character 10 is a newline.
118121countl :: Int -> Word8 -> Int
119122countl n ch = if ch == 10 then n + 1 else n
@@ -124,7 +127,7 @@ nlines = Fold.foldl' countl 0
124127
125128wcl :: String -> IO Int
126129wcl file =
127− File.toBytes file -- SerialT IO Word8
130+ File.read file -- Stream IO Word8
128131 & Stream.fold nlines -- IO Int
129132```
130133
@@ -133,7 +136,9 @@ wcl file =
133136Our final code fragment counts the number of whitespace-separated words
134137in a stream:
135138
136−```haskell
139+```haskell ghci
140+import Data.Char (chr, isSpace)
141+
137142countw :: (Int, Bool) -> Word8 -> (Int, Bool)
138143countw (n, wasSpace) ch =
139144 if isSpace $ chr $ fromIntegral ch
@@ -146,7 +151,7 @@ nwords = fst <$> Fold.foldl' countw (0, True)
146151
147152wcw :: String -> IO Int
148153wcw file =
149− File.toBytes file -- SerialT IO Word8
154+ File.read file -- Stream IO Word8
150155 & Stream.fold nwords -- IO Int
151156```
152157
@@ -159,7 +164,8 @@ to all the supplied folds (`Fold.length`, `nlines`, and `nwords`) and
159164then combines the outputs from the folds using the supplied combiner
160165function (`(,,)`).
161166
162−```haskell
167+```haskell ghci
168+import Streamly.Internal.Data.Fold.Tee (Tee(Tee))
163169import qualified Streamly.Internal.Data.Fold.Tee as Tee
164170
165171-- The fold accepts a stream of `Word8` and returns the three counts.
@@ -168,7 +174,7 @@ countAll = Tee.toFold $ (,,) <$> Tee Fold.length <*> Tee nlines <*> Tee nwords
168174
169175wc :: String -> IO (Int, Int, Int)
170176wc file =
171− File.toBytes file -- SerialT IO Word8
177+ File.read file -- Stream IO Word8
172178 & Stream.fold countAll -- IO (Int, Int, Int)
173179```
174180
@@ -218,16 +224,36 @@ arrays for our input data.
218224Please see the file [WordCountParallel.hs][] for the complete working
219225code for this example, including the imports that we have omitted below.
220226
227+First we create a new data type `Counts` that holds all the context.
228+
229+```haskell ghci
230+-- Counts lines words chars lastCharWasSpace
231+data Counts = Counts !Int !Int !Int !Bool deriving Show
232+
233+{-# INLINE count #-}
234+count :: Counts -> Char -> Counts
235+count (Counts l w c wasSpace) ch =
236+ let l1 = if ch == '\n' then l + 1 else l
237+ (w1, wasSpace1) =
238+ if isSpace ch
239+ then (w, True)
240+ else (if wasSpace then w + 1 else w, False)
241+ in Counts l1 w1 (c + 1) wasSpace1
242+```
243+
221244The `countArray` function counts the line, word, char counts in one chunk:
222245
223−```haskell
224−import qualified Streamly.Data.Array.Foreign as Array
246+```haskell ghci
247+import Streamly.Data.Array (Array)
248+
249+import qualified Streamly.Data.Array as Array
250+import qualified Streamly.Unicode.Stream as Unicode
225251
226252countArray :: Array Word8 -> IO Counts
227253countArray arr =
228− Stream.unfold Array.read arr -- SerialT IO Word8
229− & Stream.decodeLatin1 -- SerialT IO Char
230− & Stream.foldl' count (Counts 0 0 0 True) -- IO Counts
254+ Stream.unfold Array.reader arr -- Stream IO Word8
255+ & Unicode.decodeLatin1 -- Stream IO Char
256+ & Stream.fold (Fold.foldl' count (Counts 0 0 0 True)) -- IO Counts
231257```
232258
233259Here the function `count` and the `Counts` data type are defined in the
@@ -240,10 +266,10 @@ whether the chunk starts with a new word. The `partialCounts` function
240266adds a `Bool` flag to `Counts` returned by `countArray` to indicate
241267whether the first character in the chunk is a space.
242268
243−```haskell
269+```haskell ghci
244270partialCounts :: Array Word8 -> IO (Bool, Counts)
245271partialCounts arr = do
246− let r = Array.getIndex arr 0
272+ let r = Array.getIndex 0 arr
247273 case r of
248274 Just x -> do
249275 counts <- countArray arr
@@ -253,7 +279,7 @@ partialCounts arr = do
253279
254280`addCounts` then adds the counts from two consecutive chunks:
255281
256−```haskell
282+```haskell ghci
257283addCounts :: (Bool, Counts) -> (Bool, Counts) -> (Bool, Counts)
258284addCounts (sp1, Counts l1 w1 c1 ws1) (sp2, Counts l2 w2 c2 ws2) =
259285 let wcount =
@@ -267,22 +293,34 @@ To count in parallel we now only need to divide the stream into arrays,
267293apply our counting function to each array, and then combine the counts
268294from each chunk.
269295
270−```haskell
296+```haskell ghci
297+{-# LANGUAGE FlexibleContexts #-}
298+
299+import qualified Streamly.Data.Stream.Concurrent as Concur
300+
301+-- Number of threads we can afford
302+numCapabilities :: Int
303+numCapabilities = 10
304+
271305wc :: String -> IO (Bool, Counts)
272306wc file = do
273− Stream.unfold File.readChunks file -- AheadT IO (Array Word8)
274− & Stream.mapM partialCounts -- AheadT IO (Bool, Counts)
275− & Stream.maxThreads numCapabilities -- AheadT IO (Bool, Counts)
276− & Stream.fromAhead -- SerialT IO (Bool, Counts)
277− & Stream.foldl' addCounts (False, Counts 0 0 0 True) -- IO (Bool, Counts)
307+ Stream.unfold File.chunkReader file
308+ & Concur.mapMWith
309+ ( Concur.maxThreads numCapabilities
310+ . Concur.ordered True )
311+ partialCounts
312+ & Stream.fold
313+ (Fold.foldl' addCounts (False, Counts 0 0 0 True))
278314```
279315
280−Please note that the only difference between a concurrent and a
281−non-concurrent program lies in the use of the `Stream.fromAhead`
282−combinator. If we remove the call to `Stream.fromAhead`, we would
283−still have a perfectly valid and performant serial program. Notice
284−how succinctly and idiomatically we have expressed the concurrent word
285−counting problem.
316+Please note that we use the mapping function from
317+`Streamly.Data.Stream.Concurrent` module. We explicitly specify what parts of
318+our stream are concurrent.
319+
320+If we replace `Concur.mapMWith (Concur.maxThreads numCapabilities
321+. Concur.ordered True)` with `Stream.mapM`, we would still have a perfectly
322+valid and performant serial program. Notice how succinctly and idiomatically we
323+have expressed the concurrent word counting problem.
286324
287325A benchmark with 2 CPUs:
288326
@@ -313,14 +351,17 @@ with ease, with support for different types of concurrent scheduling.
313351
314352We now move to a slightly more complicated example: we simulate a
315353dictionary lookup server which can serve word meanings to multiple
316−clients concurrently. This example demonstrates the use of the concurrent
317−`mapM` combinator.
354+clients concurrently.
318355
319356Please see the file [WordServer.hs][] for the complete code for this
320357example, including the imports that we have omitted below.
321358
322−```haskell
323−import qualified Streamly.Data.Fold as Fold
359+```haskell ghci
360+import Control.Concurrent (threadDelay)
361+import Control.Exception (finally)
362+import Network.Socket (Socket, close)
363+
364+import qualified Streamly.Data.Parser as Parser
324365import qualified Streamly.Network.Inet.TCP as TCP
325366import qualified Streamly.Network.Socket as Socket
326367import qualified Streamly.Unicode.Stream as Unicode
@@ -335,16 +376,14 @@ fetch w = threadDelay 1000000 >> return (w,w)
335376-- connection is closed.
336377lookupWords :: Socket -> IO ()
337378lookupWords sk =
338− Stream.unfold Socket.read sk -- SerialT IO Word8
339− & Unicode.decodeLatin1 -- SerialT IO Char
340− & Stream.wordsBy isSpace Fold.toList -- SerialT IO String
341− & Stream.fromSerial -- AheadT IO String
342− & Stream.mapM fetch -- AheadT IO (String, String)
343− & Stream.fromAhead -- SerialT IO (String, String)
344− & Stream.map show -- SerialT IO String
345− & Stream.intersperse "\n" -- SerialT IO String
346− & Unicode.encodeStrings Unicode.encodeLatin1 -- SerialT IO (Array Word8)
347− & Stream.fold (Socket.writeChunks sk) -- IO ()
379+ Stream.unfold Socket.reader sk
380+ & Unicode.decodeLatin1
381+ & Stream.parseMany (Parser.wordBy isSpace Fold.toList)
382+ & Concur.mapMWith (Concur.ordered True) fetch
383+ & fmap show
384+ & Stream.intersperse "\n"
385+ & Unicode.encodeStrings Unicode.encodeLatin1
386+ & Stream.fold (Socket.writeChunks sk)
348387
349388serve :: Socket -> IO ()
350389serve sk = finally (lookupWords sk) (close sk)
@@ -354,11 +393,9 @@ serve sk = finally (lookupWords sk) (close sk)
354393-- "nc" as a client to try it out.
355394main :: IO ()
356395main =
357− Stream.unfold TCP.acceptOnPort 8091 -- SerialT IO Socket
358− & Stream.fromSerial -- AsyncT IO ()
359− & Stream.mapM serve -- AsyncT IO ()
360− & Stream.fromAsync -- SerialT IO ()
361− & Stream.drain -- IO ()
396+ Stream.unfold TCP.acceptorOnPort 8091
397+ & Concur.mapMWith id serve
398+ & Stream.fold Fold.drain
362399```
363400
364401### Merging Incoming Streams
@@ -372,21 +409,27 @@ streams concurrently.
372409Please see the file [MergeServer.hs][] for the complete working code,
373410including the imports that we have omitted below.
374411
375−```haskell
412+```haskell ghci
413+{-# LANGUAGE FlexibleContexts #-}
414+
415+import Streamly.Data.Stream (Stream)
416+import System.IO (IOMode(AppendMode), Handle, withFile)
417+
376418import qualified Streamly.Data.Unfold as Unfold
377419import qualified Streamly.Network.Socket as Socket
420+import qualified Streamly.FileSystem.Handle as Handle
378421
379422-- | Read a line stream from a socket.
380423-- Note: lines are buffered, and we could add a limit to the
381424-- buffering for safety.
382−readLines :: Socket -> SerialT IO (Array Char)
425+readLines :: Socket -> Stream IO (Array Char)
383426readLines sk =
384− Stream.unfold Socket.read sk -- SerialT IO Word8
385− & Unicode.decodeLatin1 -- SerialT IO Char
386− & Stream.splitWithSuffix (== '\n') Array.write -- SerialT IO String
427+ Stream.unfold Socket.reader sk
428+ & Unicode.decodeLatin1
429+ & Stream.foldMany (Fold.takeEndBy (== '\n') Array.write)
387430
388−recv :: Socket -> SerialT IO (Array Char)
389−recv sk = Stream.finally (liftIO $ close sk) (readLines sk)
431+recv :: Socket -> Stream IO (Array Char)
432+recv sk = Stream.finallyIO (close sk) (readLines sk)
390433
391434-- | Starts a server at port 8091 listening for lines with space separated
392435-- words. Multiple clients can connect to the server and send streams of lines.
@@ -394,11 +437,11 @@ recv sk = Stream.finally (liftIO $ close sk) (readLines sk)
394437-- streams at line boundaries and writes the merged stream to a file.
395438server :: Handle -> IO ()
396439server file =
397− Stream.unfold TCP.acceptOnPort 8090 -- SerialT IO Socket
398− & Stream.concatMapWith Stream.parallel recv -- SerialT IO (Array Char)
399− & Stream.unfoldMany Array.read -- SerialT IO Char
400− & Unicode.encodeLatin1 -- SerialT IO Word8
401− & Stream.fold (Handle.write file) -- IO ()
440+ Stream.unfold TCP.acceptorOnPort 8090
441+ & Concur.concatMapWith (Concur.eager True) recv
442+ & Stream.unfoldMany Array.reader
443+ & Unicode.encodeLatin1
444+ & Stream.fold (Handle.write file)
402445
403446main :: IO ()
404447main = withFile "output.txt" AppendMode server
@@ -409,37 +452,43 @@ main = withFile "output.txt" AppendMode server
409452Our next example lists a directory tree recursively, reading
410453multiple directories concurrently.
411454
412−This example uses the tree traversing combinator `iterateMapLeftsWith`.
413−This combinator maps a stream generator on the `Left` values in its
414−input stream (directory names in this case), feeding the resulting `Left`
415−values back to the input, while it lets the `Right` values (file names
416−in this case) pass through to the output. The `Stream.ahead` stream
417−joining combinator then makes it iterate on the directories concurrently.
455+This example uses the tree traversing combinator `iterateMapLeftsWith`. This
456+combinator maps a stream generator on the `Left` values in its input stream
457+(directory names in this case), feeding the resulting `Left` values back to the
458+input, while it lets the `Right` values (file names in this case) pass through
459+to the output. The joining combinator explicitly sends streams to a channel
460+making it iterate on the directories concurrently.
418461
419462Please see the file [ListDir.hs][] for the complete working code,
420463including the imports that we have omitted below.
421464
422−```haskell
423−import Streamly.Internal.Data.Stream.IsStream (iterateMapLeftsWith)
424−
425−import qualified Streamly.Prelude as Stream
426−import qualified Streamly.Internal.FileSystem.Dir as Dir (toEither)
465+```haskell ghci
466+import Data.Bifunctor (bimap)
467+import System.IO (stdout, hSetBuffering, BufferMode(LineBuffering))
468+import qualified Streamly.Internal.Data.Stream as Stream (iterateMapLeftsWith)
469+import qualified Streamly.Internal.FileSystem.Dir as Dir (readEither)
470+import qualified Streamly.Internal.Data.Stream.Concurrent.Channel as Concur
427471
428472-- Lists a directory as a stream of (Either Dir File).
429−listDir :: String -> SerialT IO (Either String String)
473+listDir :: String -> Stream IO (Either String String)
430474listDir dir =
431− Dir.toEither dir -- SerialT IO (Either String String)
432− & Stream.map (bimap mkAbs mkAbs) -- SerialT IO (Either String String)
475+ Dir.readEither dir
476+ & fmap (bimap mkAbs mkAbs)
433477
434478 where mkAbs x = dir ++ "/" ++ x
435479
436480-- | List the current directory recursively using concurrent processing.
437481main :: IO ()
438482main = do
483+ chan <- Concur.newChannel (Concur.ordered True)
484+ let combineOrdered s1 s2 = do
485+ Stream.fromEffect $ Concur.toChannel chan s1
486+ Stream.fromEffect $ Concur.toChannel chan s2
487+ Concur.fromChannel chan
439488 hSetBuffering stdout LineBuffering
440489 let start = Stream.fromPure (Left ".")
441− Stream.iterateMapLeftsWith Stream.ahead listDir start
442− & Stream.mapM_ print
490+ Stream.iterateMapLeftsWith combineOrdered listDir start
491+ & Stream.fold (Fold.drainMapM print)
443492```
444493
445494### Rate Limiting
@@ -448,14 +497,15 @@ For bounded concurrent streams, a stream yield rate can be specified
448497easily. For example, to print "tick" once every second you can simply
449498write:
450499
451−```haskell
500+```haskell ghci
501+import qualified Streamly.Internal.Data.Stream as Stream (timestamped)
502+
452503main :: IO ()
453504main =
454− Stream.repeatM (pure "tick") -- AsyncT IO String
455− & Stream.timestamped -- AsyncT IO (AbsTime, String)
456− & Stream.avgRate 1 -- AsyncT IO (AbsTime, String)
457− & Stream.fromAsync -- SerialT IO (AbsTime, String)
458− & Stream.mapM_ print -- IO ()
505+ Stream.repeat (pure "tick")
506+ & Concur.sequenceWith (Concur.avgRate 1)
507+ & Stream.timestamped
508+ & Stream.fold (Fold.drainMapM print)
459509```
460510
461511Please see the file [Rate.hs][] for the complete working code.
@@ -469,13 +519,9 @@ documentation][Streamly].
469519
470520### Reactive Programming
471521
472−Streamly supports reactive (time domain) programming because of its
473−support for declarative concurrency. Please see the `Streamly.Prelude`
474−module for time-specific combinators like `intervalsOf`, and
475−folds like `takeInterval` in `Streamly.Internal.Data.Fold`.
476−Please also see the pre-release sampling combinators in the
477−`Streamly.Internal.Data.Stream.IsStream.Top` module for `throttle` and
478−`debounce` like operations.
522+Streamly supports reactive (time domain) programming because of its support for
523+declarative concurrency. Please see the `Streamly.Internal.Data.Stream.Time`
524+module for time-specific and sampling combinators.
479525
480526The examples [AcidRain.hs][] and [CirclingSquare.hs][] demonstrate
481527reactive programming using [Streamly][].