3-- Copyright : (c) 2017 Composewell Technologies
6-- Maintainer : streamly@composewell.com
7-- Stability : experimental
10-- Streamly is a general purpose programming framework using cocnurrent data
11-- flow programming paradigm. It can be considered as a generalization of
12-- Haskell lists to monadic streaming with concurrent composition capability.
13-- The serial stream type in streamly @SerialT m a@ is like the list type @[a]@
14-- parameterized by the monad @m@. For example, @SerialT IO a@ is a moral
15-- equivalent of @[a]@ in the IO monad. Streams are constructed very much like
16-- lists, except that they use 'nil' and 'cons' instead of '[]' and ':'.
20-- > import "Streamly.Prelude" (cons, consM)
21-- > import qualified "Streamly.Prelude" as S
23-- > S.toList $ 1 \`cons` 2 \`cons` 3 \`cons` nil
27-- Unlike lists, streams can be constructed from monadic effects:
30-- > S.'toList' $ 'getLine' \`consM` 'getLine' \`consM` S.'nil'
36-- Streams are processed just like lists, with list like combinators, except
37-- that they are monadic and work in a streaming fashion. Here is a simple
38-- console echo program example:
41-- > S.drain $ S.repeatM getLine & S.mapM putStrLn
44-- @SerialT Identity a@ is a moral equivalent of pure lists. Streamly utilizes
45-- fusion for high performance, therefore, we can represent and process strings
46-- as streams of 'Char', encode and decode the streams to/from UTF8 and
47-- serialize them to @Array Word8@ obviating the need for special purpose
48-- libraries like @bytestring@ and @text@.
51-- please see the "Streamly.Tutorial" module and the examples directory in this
54{-# LANGUAGE NoMonomorphismRestriction #-}
55{-# OPTIONS_GHC -Wno-orphans #-}
59module Streamly {-# DEPRECATED "Please use \"Streamly.Prelude\" instead." #-}
61 -- -- * Concepts Overview
77 -- * Stream transformers
78 -- | A stream represents a sequence of pure or effectful actions. The
79 -- `cons` and `consM` operations and the corresponding operators '.:' and
80 -- '|:' can be used to join pure values or effectful actions in a sequence.
81 -- The effects in the stream can be executed in many different ways
82 -- depending on the type of stream. In other words, the behavior of 'consM'
83 -- depends on the type of the stream.
85 -- There are three high level categories of streams, /spatially ordered/
86 -- streams, /speculative/ streams and /time ordered/ streams. Spatially
87 -- ordered streams, 'SerialT' and 'WSerialT', execute the effects in serial
88 -- order i.e. one at a time and present the outputs of those effects to the
89 -- consumer in the same order. Speculative streams, 'AheadT', may execute
90 -- many effects concurrently but present the outputs to the consumer in the
91 -- specified spatial order. Time ordered streams, 'AsyncT', 'WAsyncT' and
92 -- 'ParallelT', may execute many effects concurrently and present the
93 -- outputs of those effects to the consumer in time order i.e. as soon as
94 -- the output is generated.
96 -- We described above how the effects in a sequence are executed for
97 -- different types of streams. The behvavior of the 'Semigroup' and 'Monad'
98 -- instances follow the behavior of 'consM'. Stream generation operations
99 -- like 'repeatM' also execute the effects differently for different
100 -- streams, providing a concurrent generation capability when used with
101 -- stream types that execute effects concurrently. Similarly, effectful
102 -- transformation operations like 'mapM' also execute the transforming
103 -- effects differently for different types of streams.
110 -- ** Speculative Streams
114 -- ** Asynchronous Streams
120 -- ** Zipping Streams
125 -- * Parallel Function Application
142 -- * Concurrency Control
155 -- * Stream Type Adapters
179 -- ** Folding Containers of Streams
180 -- | These are variants of standard 'Foldable' fold functions that use a
181 -- polymorphic stream sum operation (e.g. 'async' or 'wSerial') to fold a
182 -- finite container of streams. Note that these are just special cases of
183 -- the more general 'concatMapWith' operation.
212import Data.Semigroup (Semigroup(..))
213import Streamly.Internal.Control.Concurrent (MonadAsync)
214import Streamly.Internal.Data.SVar (Rate(..))
215import Streamly.Internal.Data.Stream.Ahead
216import Streamly.Internal.Data.Stream.Async hiding (mkAsync)
217import Streamly.Internal.Data.Stream.IsStream.Combinators
218import Streamly.Internal.Data.Stream.Parallel
219import Streamly.Internal.Data.Stream.Serial
220import Streamly.Internal.Data.Stream.StreamK hiding (serial)
221import Streamly.Internal.Data.Stream.Zip
223import qualified Streamly.Prelude as P
224import qualified Streamly.Internal.Data.Stream.IsStream as IP
225import qualified Streamly.Internal.Data.Stream.StreamK as K
226import qualified Streamly.Internal.Data.Stream.Async as Async
228-- XXX provide good succinct examples of pipelining, merging, splitting etc.
233-- A program is expressed as a network of streams and folds. A stream is a
234-- source or generator of data elements and a fold is a consumer of data elements that
235-- reduces multiple input elements to a single value.
237-- In the following example, a 'Word8' stream is generated by using
238-- 'Streamly.FileSystem.Handle.read' on a file handle, then the
239-- 'Streamly.Prelude.splitBySuffix' transformation splits the stream on
240-- newlines (ascii value 10); it uses the 'Streamly.Data.Fold.drain' fold to reduce
241-- the resulting lines to unit values (@()@), 'Streamly.Prelude.length' fold
242-- then counts the unit elements in the resulting stream which gives us the
243-- number of lines in the file:
245-- > S.length $ S.splitOnSuffix FL.drain 10 $ FH.read fh
247-- The following example folds the lines to arrays of 'Word8' using the
248-- 'Streamly.Data.Array.Foreign.writeF' fold and then wraps the lines in square
249-- brackets before writing them to standard output using
250-- 'Streamly.FileSystem.Handle.write':
252-- > wrapLine ln = S.fromList "[" <> A.read ln <> S.fromList "]\n"
253-- > readLines = S.splitOnSuffix A.writeF 10
254-- > FH.write stdout $ S.concatMap wrapLine $ readLines fh1
256-- One stream can be appended after another:
258-- > FH.write stdout $ S.concatMap wrapLine $ readLines fh1 <> readLines fh2
260-- The following example reads two files concurrently, merges the lines from
261-- the two streams and writes the resulting stream to another file:
263-- > FH.write stdout $ S.concatMap wrapLine $ readLines fh1 `parallel` readLines fh2
265-- There are many ways to generate, merge, zip, transform and fold data
266-- streams. Many transformations can be chained in a stream pipeline. See
267-- "Streamly.Prelude" module for combinators to manipulate streams.
271-- The way stream types in this module like 'SerialT' represent data sources,
272-- the same way the 'Fold' type from "Streamly.Data.Fold" represents data sinks or
273-- reducers of streams. Reducers can be combined to consume a stream source in
274-- many ways. The simplest is to reduce a stream source using a fold e.g.:
276-- > S.fold FL.length $ S.enumerateTo 100
278-- Folds are consumers of streams and can be used to split a stream into
279-- multiple independent flows. Grouping transforms a stream by applying a fold
280-- on segments of a stream, distributing applies multiple folds in parallel on
281-- the same stream and combines them, partitioning sends different elements of
282-- a stream to different folds, unzipping divides the elements of a stream into
283-- parts and sends them through different folds. Parsers are nothing but a
284-- particular type of folds. Transformations can be applied contravariantly on
285-- the input of a fold.
289-- |---transform----Fold m a b--------|
290-- ---stream m a-->transform-->| |---f b c ...
291-- |---transform----Fold m a c--------|
299-- Streamly arrays (See "Streamly.Data.Array.Foreign") complement streams to provide an
300-- efficient computing paradigm. Streams are suitable for immutable
301-- transformations of /potentially infinite/ data using /sequential access/ and
302-- pipelined transformations whereas arrays are suitable for in-place
303-- transformations of /necessarily finite/ data using /random access/. Streams
304-- are synonymous with /sequential pipelined processing/ whereas arrays are
305-- synonymous with /efficient buffering and random access/.
307-- In general, a data processing pipeline reads data from some IO device, does
308-- some processing on it and finally writes the output to another IO device.
309-- Streams provide the overall framework of sequential processing pipeline in
310-- which arrays are used as buffering elements in the middle. In addition to
311-- buffering in the middle, arrays can also be used at the boundaries of the
312-- pipeline to efficiently interface with external storage systems like memory,
313-- files and network. If streams are the pipes in a water pipeline network
314-- then arrays are like the storage tanks in the middle. On the input side,
315-- think of arrays as buckets to fetch water to feed the pipeline and on the
316-- output side buckets to remove the processed water.
318-- 'ByteString' data type from the 'bytestring' package and the 'Text' data
319-- type from the 'text' package are special cases of arrays. 'ByteString' is
320-- like @Array Word8@ and 'Text' is like @utf16@ encoded @Array Word8@.
321-- Streamly arrays can be transformed as efficiently as @bytestring@ or @text@
322-- by using stream operations on them.
324-- Streams and arrays are equally important in computing. They are computing
325-- duals of each other.
328-- The basic stream type is 'Serial', it represents a sequence of IO actions,
329-- and is a 'Monad'. The type 'SerialT' is a monad transformer that can
330-- represent a sequence of actions in an arbitrary monad. The type 'Serial' is
331-- in fact a synonym for @SerialT IO@. There are a few more types similar to
332-- 'SerialT', all of them represent a stream and differ only in the
333-- 'Semigroup', 'Applicative' and 'Monad' compositions of the stream. 'Serial'
334-- and 'WSerial' types compose serially whereas 'Async' and 'WAsync'
335-- types compose concurrently. All these types can be freely inter-converted
336-- using type combinators without any cost. You can freely switch to any type
337-- of composition at any point in the program. When no type annotation or
338-- explicit stream type combinators are used, the default stream type is
339-- inferred as 'Serial'.
341-- This module exports stream types, instances and combinators for:
343-- * converting between different stream types
344-- * appending and concurrently merging streams
345-- * Concurrency control
346-- * Concurrent function application
347-- * Stream rate control
349-- This module is designed to be imported unqualified:
355-- See the "Streamly.Prelude" module for APIs for construction,
356-- generation, elimination and transformation of streams.
358------------------------------------------------------------------------------
359-- Eliminating a stream
360------------------------------------------------------------------------------
362-- | Same as 'runStream'
365{-# DEPRECATED runStreaming "Please use runStream instead." #-}
366runStreaming :: (Monad m, IsStream t) => t m a -> m ()
367runStreaming = P.drain . K.adapt
369-- | Same as @runStream@.
372{-# DEPRECATED runStreamT "Please use runStream instead." #-}
373runStreamT :: Monad m => SerialT m a -> m ()
376-- | Same as "Streamly.Prelude.runStream".
378{-# DEPRECATED runStream "Please use Streamly.Prelude.drain instead." #-}
379runStream :: Monad m => SerialT m a -> m ()
382-- | Same as @drain . fromWSerial@.
385{-# DEPRECATED runInterleavedT "Please use 'drain . interleaving' instead." #-}
386runInterleavedT :: Monad m => WSerialT m a -> m ()
387runInterleavedT = P.drain . K.adapt
389-- | Same as @drain . fromParallel@.
392{-# DEPRECATED runParallelT "Please use 'drain . fromParallel' instead." #-}
393runParallelT :: Monad m => ParallelT m a -> m ()
394runParallelT = P.drain . K.adapt
396-- | Same as @drain . fromAsync@.
399{-# DEPRECATED runAsyncT "Please use 'drain . fromAsync' instead." #-}
400runAsyncT :: Monad m => AsyncT m a -> m ()
401runAsyncT = P.drain . K.adapt
403-- | Same as @drain . zipping@.
406{-# DEPRECATED runZipStream "Please use 'drain . fromZipSerial instead." #-}
407runZipStream :: Monad m => ZipSerialM m a -> m ()
408runZipStream = P.drain . K.adapt
410-- | Same as @drain . zippingAsync@.
413{-# DEPRECATED runZipAsync "Please use 'drain . fromZipAsync instead." #-}
414runZipAsync :: Monad m => ZipAsyncM m a -> m ()
415runZipAsync = P.drain . K.adapt
418-- | Same as "Streamly.Prelude.foldWith".
420{-# DEPRECATED foldWith "Please use Streamly.Prelude.foldWith instead." #-}
421{-# INLINABLE foldWith #-}
422foldWith :: (IsStream t, Foldable f)
423 => (t m a -> t m a -> t m a) -> f (t m a) -> t m a
426-- | Same as "Streamly.Prelude.foldMapWith".
428{-# DEPRECATED foldMapWith "Please use Streamly.Prelude.foldMapWith instead." #-}
429{-# INLINABLE foldMapWith #-}
430foldMapWith :: (IsStream t, Foldable f)
431 => (t m b -> t m b -> t m b) -> (a -> t m b) -> f a -> t m b
432foldMapWith = P.foldMapWith
434-- | Same as "Streamly.Prelude.forEachWith".
436{-# DEPRECATED forEachWith "Please use Streamly.Prelude.forEachWith instead." #-}
437{-# INLINABLE forEachWith #-}
438forEachWith :: (IsStream t, Foldable f)
439 => (t m b -> t m b -> t m b) -> f a -> (a -> t m b) -> t m b
440forEachWith = P.forEachWith
443-- XXX Deprecate it in 0.8.0
445-- | Make a stream asynchronous, triggers the computation and returns a stream
446-- in the underlying monad representing the output generated by the original
447-- computation. The returned action is exhaustible and must be drained once. If
448-- not drained fully we may have a thread blocked forever and once exhausted it
449-- will always return 'empty'.
452{-# INLINABLE mkAsync #-}
453mkAsync :: (IsStream t, MonadAsync m) => t m a -> m (t m a)
454mkAsync = return . Async.mkAsync
456------------------------------------------------------------------------------
458------------------------------------------------------------------------------
462-- When a stream consumer demands an element from a serial stream constructed
463-- as @a \`consM` b \`consM` ... nil@, the action @a@ at the head of the stream
464-- sequence is executed and the result is supplied to the consumer. When the
465-- next element is demanded, the action @b@ is executed and its result is
466-- supplied. Thus, the effects are performed and results are consumed strictly
467-- in a serial order. Serial streams can be considered as /spatially ordered/
468-- streams as the order of execution and consumption is the same as the spatial
469-- order in which the actions are composed by the programmer.
471-- Serial streams enforce the side effects as well as the results of the
472-- actions to be in the same order in which the actions are added to the
473-- stream. Therefore, the semigroup operation for serial streams is not
477-- a <> b is not the same as b <> a
480-- There are two serial stream types 'SerialT' and 'WSerialT'. The stream
481-- evaluation of both the variants works in the same way as described above,
482-- they differ only in the 'Semigroup' and 'Monad' implementaitons.
486-- When a stream consumer demands an element from a speculative stream
487-- constructed as @a \`consM` b \`consM` ... nil@, the action @a@ at the head
488-- of the stream is executed and the output of the action is supplied to the
489-- consumer. However, in addition to the action at the head multiple actions
490-- following it may also be executed concurrently and the results buffered.
491-- When the next element is demanded it may be served from the buffer and we
492-- may execute the next action in the sequence to keep the buffer adequately
493-- filled. Thus, the actions are executed concurrently but results consumed in
494-- serial order just like serial streams. `consM` can be used to fold an
495-- infinite lazy container of effects, as the number of concurrent executions
498-- Similar to 'consM', the monadic stream generation (e.g. replicateM) and
499-- transformation operations (e.g. mapM) on speculative streams can execute
500-- multiple effects concurrently in a speculative manner.
502-- How many effects can be executed concurrently and how many results can be
503-- buffered are controlled by 'maxThreads' and 'maxBuffer' combinators
504-- respectively. The actual number of concurrent threads is adjusted according
505-- to the rate at which the consumer is consuming the stream. It may even
506-- execute actions serially in a single thread if that is enough to match the
509-- Speculative streams enforce ordering of the results of actions in the stream
510-- but the side effects are only partially ordered. Therefore, the semigroup
511-- operation for speculative streams is not commutative from the pure outputs
512-- perspective but commutative from side effects perspective.
516-- /Scheduling and execution:/ In an asynchronous stream @a \`consM` b \`consM`
517-- c ...@, the actions @a@, @b@, and @c@ are executed concurrently with the
518-- consumer of the stream. The actions are /scheduled/ for execution in the
519-- same order as they are specified in the stream. Multiple scheduled actions
520-- may be /executed/ concurrently in parallel threads of execution. The
521-- actions may be executed out of order and they may complete at arbitrary
522-- times. Therefore, the /effects/ of the actions may be observed out of
525-- /Buffering:/ The /results/ from multiple threads of execution are queued in
526-- a buffer as soon as they become available. The consumer of the stream is
527-- served from this buffer. Therefore, the consumer may observe the results to
528-- be out of order. In other words, an asynchronous stream is an unordered
529-- stream i.e. order does not matter.
531-- /Concurrency control:/ Threads are suspended if the `maxBuffer` limit is
532-- reached, and resumed when the consumer makes space in the buffer. The
533-- maximum number of concurrent threads depends on `maxThreads`. Number of
534-- threads is increased or decreased based on the speed of the consumer.
536-- /Generation operations:/ Concurrent stream generation operations e.g.
537-- 'Streamly.Prelude.replicateM' when used in async style schedule and execute
538-- the stream generating actions in the manner described above. The generation
539-- actions run concurrently, effects and results of the actions as observed by
540-- the consumer of the stream may be out of order.
542-- /Transformation operations:/ Concurrent stream transformation operations
543-- e.g. 'Streamly.Prelude.mapM', when used in async style, schedule and
544-- execute transformation actions in the manner described above. Transformation
545-- actions run concurrently, effects and results of the actions may be
546-- observed by the consumer out of order.
548-- /Variants:/ There are two asynchronous stream types 'AsyncT' and 'WAsyncT'.
549-- They are identical with respect to single stream evaluation behavior. Their
550-- behaviors differ in how they combine multiple streams using 'Semigroup' or
551-- 'Monad' composition. Since the order of elements does not matter in
552-- asynchronous streams the 'Semigroup' operation is effectively commutative.
556-- 'ZipSerialM' and 'ZipAsyncM', provide 'Applicative' instances for zipping the
557-- corresponding elements of two streams together. Note that these types are
562-- Stream processing functions can be composed in a chain using function
563-- application with or without the '$' operator, or with reverse function
564-- application operator '&'. Streamly provides concurrent versions of these
565-- operators applying stream processing functions such that each stage of the
566-- stream can run in parallel. The operators start with a @|@; we can read '|$'
567-- as "@parallel dollar@" to remember that @|@ comes before '$'.
569-- Imports for the code snippets below:
573-- import qualified Streamly.Prelude as S
574-- import Control.Concurrent
578-- The 'Semigroup' operation '<>' of each stream type combines two streams in a
579-- type specific manner. This section provides polymorphic versions of '<>'
580-- which can be used to combine two streams in a predetermined way irrespective
583-- XXX An alternative design choice would be to let a control parameter affect
584-- the nearest SVar only and then it gets cleared. The benefit of the current
585-- choice is that it is simply just like global configuration, just like state
586-- behaves, so should be easy to comprehend. But it has the downside of leaking
587-- to undesired actions, that is we can forget to reset it.
591-- These combinators can be used at any point in a stream composition to set
592-- parameters to control the concurrency of the /argument stream/. A control
593-- parameter set at any point remains effective for any concurrent combinators
594-- used in the argument stream until it is reset by using the combinator again.
595-- These control parameters have no effect on non-concurrent combinators in the
596-- stream, or on non-concurrent streams.
598-- /Pitfall:/ Remember that 'maxBuffer' in the following example applies to
599-- 'mapM' and any other combinators that may follow it, and it does not apply
600-- to the combinators before it:
609-- If we use '&' instead of '$' the situation will reverse, in the following
610-- example, 'maxBuffer' does not apply to 'mapM', it applies to combinators
611-- that come before it, because those are the arguments to 'maxBuffer':
622-- You may want to use different stream composition styles at different points
623-- in your program. Stream types can be freely converted or adapted from one
624-- type to another. The 'IsStream' type class facilitates type conversion of
625-- one stream type to another. It is not used directly, instead the type
626-- combinators provided below are used for conversions.
628-- To adapt from one monomorphic type (e.g. 'AsyncT') to another monomorphic
629-- type (e.g. 'SerialT') use the 'adapt' combinator. To give a polymorphic code
630-- a specific interpretation or to adapt a specific type to a polymorphic type
631-- use the type specific combinators e.g. 'fromAsync' or 'fromWSerial'. You
632-- cannot adapt polymorphic code to polymorphic code, as the compiler would not know
633-- which specific type you are converting from or to. If you see a an
634-- @ambiguous type variable@ error then most likely you are using 'adapt'
635-- unnecessarily on polymorphic code.
637-- | Same as 'Streamly.Prelude.concatFoldableWith'
640{-# DEPRECATED foldWith "Please use 'Streamly.Prelude.concatFoldableWith' instead." #-}
641{-# INLINEABLE foldWith #-}
642foldWith :: (IsStream t, Foldable f) => (t m a -> t m a -> t m a) -> f (t m a) -> t m a
643foldWith = P.concatFoldableWith
645-- | Same as 'Streamly.Prelude.concatMapFoldableWith'
648{-# DEPRECATED foldMapWith "Please use 'Streamly.Prelude.concatMapFoldableWith' instead." #-}
649{-# INLINEABLE foldMapWith #-}
650foldMapWith :: (IsStream t, Foldable f) => (t m b -> t m b -> t m b) -> (a -> t m b) -> f a -> t m b
651foldMapWith = P.concatMapFoldableWith
653-- | Same as 'Streamly.Prelude.concatForFoldableWith'
656{-# DEPRECATED forEachWith "Please use 'Streamly.Prelude.concatForFoldableWith' instead." #-}
657{-# INLINEABLE forEachWith #-}
658forEachWith :: (IsStream t, Foldable f) => (t m b -> t m b -> t m b) -> f a -> (a -> t m b) -> t m b
659forEachWith = P.concatForFoldableWith
661{-# DEPRECATED serially "Please use 'Streamly.Prelude.fromSerial' instead." #-}
662serially :: IsStream t => SerialT m a -> t m a
665{-# DEPRECATED wSerially "Please use 'Streamly.Prelude.fromWSerial' instead." #-}
666wSerially :: IsStream t => WSerialT m a -> t m a
667wSerially = fromWSerial
669{-# DEPRECATED asyncly "Please use 'Streamly.Prelude.fromAsync' instead." #-}
670asyncly :: IsStream t => AsyncT m a -> t m a
673{-# DEPRECATED aheadly "Please use 'Streamly.Prelude.fromAhead' instead." #-}
674aheadly :: IsStream t => AheadT m a -> t m a
677{-# DEPRECATED wAsyncly "Please use 'Streamly.Prelude.fromWAsync' instead." #-}
678wAsyncly :: IsStream t => WAsyncT m a -> t m a
681{-# DEPRECATED parallely "Please use 'Streamly.Prelude.fromParallel' instead." #-}
682parallely :: IsStream t => ParallelT m a -> t m a
683parallely = fromParallel
685{-# DEPRECATED zipSerially "Please use 'Streamly.Prelude.fromZipSerial' instead." #-}
686zipSerially :: IsStream t => ZipSerialM m a -> t m a
687zipSerially = fromZipSerial
689{-# DEPRECATED zipAsyncly "Please use 'Streamly.Prelude.fromZipAsync' instead." #-}
690zipAsyncly :: IsStream t => ZipAsyncM m a -> t m a
691zipAsyncly = fromZipAsync
keysesc to close
- j k
- move down, up
- enter
- open
- esc
- back, or up a level
- 1 2 3 4
- readme, files, commits, refs
- /
- filter
- y
- copy the clone command
- c
- compare the chosen commit with another (or shift-click both)
- ?
- show or hide this list