1{-# OPTIONS_GHC -Wno-deprecations #-}
5-- Copyright : (c) 2017 Composewell Technologies
8-- Maintainer : streamly@composewell.com
9-- Stability : experimental
12-- Streamly is a general purpose programming framework using cocnurrent data
13-- flow programming paradigm. It can be considered as a generalization of
14-- Haskell lists to monadic streaming with concurrent composition capability.
15-- The serial stream type in streamly @SerialT m a@ is like the list type @[a]@
16-- parameterized by the monad @m@. For example, @SerialT IO a@ is a moral
17-- equivalent of @[a]@ in the IO monad. Streams are constructed very much like
18-- lists, except that they use 'nil' and 'cons' instead of '[]' and ':'.
22-- > import "Streamly.Prelude" (cons, consM)
23-- > import qualified "Streamly.Prelude" as S
25-- > S.toList $ 1 \`cons` 2 \`cons` 3 \`cons` nil
29-- Unlike lists, streams can be constructed from monadic effects:
32-- > S.'toList' $ 'getLine' \`consM` 'getLine' \`consM` S.'nil'
38-- Streams are processed just like lists, with list like combinators, except
39-- that they are monadic and work in a streaming fashion. Here is a simple
40-- console echo program example:
43-- > S.drain $ S.repeatM getLine & S.mapM putStrLn
46-- @SerialT Identity a@ is a moral equivalent of pure lists. Streamly utilizes
47-- fusion for high performance, therefore, we can represent and process strings
48-- as streams of 'Char', encode and decode the streams to/from UTF8 and
49-- serialize them to @Array Word8@ obviating the need for special purpose
50-- libraries like @bytestring@ and @text@.
53-- please see the "Streamly.Tutorial" module and the examples directory in this
56{-# LANGUAGE NoMonomorphismRestriction #-}
57{-# OPTIONS_GHC -Wno-orphans #-}
61module Streamly {-# DEPRECATED "Please use \"Streamly.Prelude\" instead." #-}
63 -- -- * Concepts Overview
79 -- * Stream transformers
80 -- | A stream represents a sequence of pure or effectful actions. The
81 -- `cons` and `consM` operations and the corresponding operators '.:' and
82 -- '|:' can be used to join pure values or effectful actions in a sequence.
83 -- The effects in the stream can be executed in many different ways
84 -- depending on the type of stream. In other words, the behavior of 'consM'
85 -- depends on the type of the stream.
87 -- There are three high level categories of streams, /spatially ordered/
88 -- streams, /speculative/ streams and /time ordered/ streams. Spatially
89 -- ordered streams, 'SerialT' and 'WSerialT', execute the effects in serial
90 -- order i.e. one at a time and present the outputs of those effects to the
91 -- consumer in the same order. Speculative streams, 'AheadT', may execute
92 -- many effects concurrently but present the outputs to the consumer in the
93 -- specified spatial order. Time ordered streams, 'AsyncT', 'WAsyncT' and
94 -- 'ParallelT', may execute many effects concurrently and present the
95 -- outputs of those effects to the consumer in time order i.e. as soon as
96 -- the output is generated.
98 -- We described above how the effects in a sequence are executed for
99 -- different types of streams. The behvavior of the 'Semigroup' and 'Monad'
100 -- instances follow the behavior of 'consM'. Stream generation operations
101 -- like 'repeatM' also execute the effects differently for different
102 -- streams, providing a concurrent generation capability when used with
103 -- stream types that execute effects concurrently. Similarly, effectful
104 -- transformation operations like 'mapM' also execute the transforming
105 -- effects differently for different types of streams.
112 -- ** Speculative Streams
116 -- ** Asynchronous Streams
122 -- ** Zipping Streams
127 -- * Parallel Function Application
144 -- * Concurrency Control
157 -- * Stream Type Adapters
181 -- ** Folding Containers of Streams
182 -- | These are variants of standard 'Foldable' fold functions that use a
183 -- polymorphic stream sum operation (e.g. 'async' or 'wSerial') to fold a
184 -- finite container of streams. Note that these are just special cases of
185 -- the more general 'concatMapWith' operation.
198import Data.Semigroup (Semigroup(..))
199import Streamly.Internal.Control.Concurrent (MonadAsync)
200import Streamly.Internal.Data.SVar (Rate(..))
201import Streamly.Internal.Data.Stream.Ahead
202import Streamly.Internal.Data.Stream.Async
203import Streamly.Internal.Data.Stream.IsStream.Combinators
204import Streamly.Internal.Data.Stream.IsStream.Expand
205import Streamly.Internal.Data.Stream.IsStream.Type
207import qualified Streamly.Prelude as P
208import qualified Streamly.Internal.Data.Stream.IsStream as IP
209import qualified Streamly.Internal.Data.Stream.IsStream.Transform as Transform
211-- XXX provide good succinct examples of pipelining, merging, splitting etc.
216-- A program is expressed as a network of streams and folds. A stream is a
217-- source or generator of data elements and a fold is a consumer of data elements that
218-- reduces multiple input elements to a single value.
220-- In the following example, a 'Word8' stream is generated by using
221-- 'Streamly.FileSystem.Handle.read' on a file handle, then the
222-- 'Streamly.Prelude.splitBySuffix' transformation splits the stream on
223-- newlines (ascii value 10); it uses the 'Streamly.Data.Fold.drain' fold to reduce
224-- the resulting lines to unit values (@()@), 'Streamly.Prelude.length' fold
225-- then counts the unit elements in the resulting stream which gives us the
226-- number of lines in the file:
228-- > S.length $ S.splitOnSuffix FL.drain 10 $ FH.read fh
230-- The following example folds the lines to arrays of 'Word8' using the
231-- 'Streamly.Data.Array.Foreign.writeF' fold and then wraps the lines in square
232-- brackets before writing them to standard output using
233-- 'Streamly.FileSystem.Handle.write':
235-- > wrapLine ln = S.fromList "[" <> A.read ln <> S.fromList "]\n"
236-- > readLines = S.splitOnSuffix A.writeF 10
237-- > FH.write stdout $ S.concatMap wrapLine $ readLines fh1
239-- One stream can be appended after another:
241-- > FH.write stdout $ S.concatMap wrapLine $ readLines fh1 <> readLines fh2
243-- The following example reads two files concurrently, merges the lines from
244-- the two streams and writes the resulting stream to another file:
246-- > FH.write stdout $ S.concatMap wrapLine $ readLines fh1 `parallel` readLines fh2
248-- There are many ways to generate, merge, zip, transform and fold data
249-- streams. Many transformations can be chained in a stream pipeline. See
250-- "Streamly.Prelude" module for combinators to manipulate streams.
254-- The way stream types in this module like 'SerialT' represent data sources,
255-- the same way the 'Fold' type from "Streamly.Data.Fold" represents data sinks or
256-- reducers of streams. Reducers can be combined to consume a stream source in
257-- many ways. The simplest is to reduce a stream source using a fold e.g.:
259-- > S.fold FL.length $ S.enumerateTo 100
261-- Folds are consumers of streams and can be used to split a stream into
262-- multiple independent flows. Grouping transforms a stream by applying a fold
263-- on segments of a stream, distributing applies multiple folds in parallel on
264-- the same stream and combines them, partitioning sends different elements of
265-- a stream to different folds, unzipping divides the elements of a stream into
266-- parts and sends them through different folds. Parsers are nothing but a
267-- particular type of folds. Transformations can be applied contravariantly on
268-- the input of a fold.
272-- |---transform----Fold m a b--------|
273-- ---stream m a-->transform-->| |---f b c ...
274-- |---transform----Fold m a c--------|
282-- Streamly arrays (See "Streamly.Data.Array.Foreign") complement streams to provide an
283-- efficient computing paradigm. Streams are suitable for immutable
284-- transformations of /potentially infinite/ data using /sequential access/ and
285-- pipelined transformations whereas arrays are suitable for in-place
286-- transformations of /necessarily finite/ data using /random access/. Streams
287-- are synonymous with /sequential pipelined processing/ whereas arrays are
288-- synonymous with /efficient buffering and random access/.
290-- In general, a data processing pipeline reads data from some IO device, does
291-- some processing on it and finally writes the output to another IO device.
292-- Streams provide the overall framework of sequential processing pipeline in
293-- which arrays are used as buffering elements in the middle. In addition to
294-- buffering in the middle, arrays can also be used at the boundaries of the
295-- pipeline to efficiently interface with external storage systems like memory,
296-- files and network. If streams are the pipes in a water pipeline network
297-- then arrays are like the storage tanks in the middle. On the input side,
298-- think of arrays as buckets to fetch water to feed the pipeline and on the
299-- output side buckets to remove the processed water.
301-- 'ByteString' data type from the 'bytestring' package and the 'Text' data
302-- type from the 'text' package are special cases of arrays. 'ByteString' is
303-- like @Array Word8@ and 'Text' is like @utf16@ encoded @Array Word8@.
304-- Streamly arrays can be transformed as efficiently as @bytestring@ or @text@
305-- by using stream operations on them.
307-- Streams and arrays are equally important in computing. They are computing
308-- duals of each other.
311-- The basic stream type is 'Serial', it represents a sequence of IO actions,
312-- and is a 'Monad'. The type 'SerialT' is a monad transformer that can
313-- represent a sequence of actions in an arbitrary monad. The type 'Serial' is
314-- in fact a synonym for @SerialT IO@. There are a few more types similar to
315-- 'SerialT', all of them represent a stream and differ only in the
316-- 'Semigroup', 'Applicative' and 'Monad' compositions of the stream. 'Serial'
317-- and 'WSerial' types compose serially whereas 'Async' and 'WAsync'
318-- types compose concurrently. All these types can be freely inter-converted
319-- using type combinators without any cost. You can freely switch to any type
320-- of composition at any point in the program. When no type annotation or
321-- explicit stream type combinators are used, the default stream type is
322-- inferred as 'Serial'.
324-- This module exports stream types, instances and combinators for:
326-- * converting between different stream types
327-- * appending and concurrently merging streams
328-- * Concurrency control
329-- * Concurrent function application
330-- * Stream rate control
332-- This module is designed to be imported unqualified:
338-- See the "Streamly.Prelude" module for APIs for construction,
339-- generation, elimination and transformation of streams.
341------------------------------------------------------------------------------
342-- Eliminating a stream
343------------------------------------------------------------------------------
345-- | Same as "Streamly.Prelude.runStream".
347{-# DEPRECATED runStream "Please use Streamly.Prelude.drain instead." #-}
348runStream :: Monad m => SerialT m a -> m ()
352-- | Same as "Streamly.Prelude.foldWith".
354{-# DEPRECATED foldWith "Please use Streamly.Prelude.foldWith instead." #-}
355{-# INLINABLE foldWith #-}
356foldWith :: (IsStream t, Foldable f)
357 => (t m a -> t m a -> t m a) -> f (t m a) -> t m a
360-- | Same as "Streamly.Prelude.foldMapWith".
362{-# DEPRECATED foldMapWith "Please use Streamly.Prelude.foldMapWith instead." #-}
363{-# INLINABLE foldMapWith #-}
364foldMapWith :: (IsStream t, Foldable f)
365 => (t m b -> t m b -> t m b) -> (a -> t m b) -> f a -> t m b
366foldMapWith = P.foldMapWith
368-- | Same as "Streamly.Prelude.forEachWith".
370{-# DEPRECATED forEachWith "Please use Streamly.Prelude.forEachWith instead." #-}
371{-# INLINABLE forEachWith #-}
372forEachWith :: (IsStream t, Foldable f)
373 => (t m b -> t m b -> t m b) -> f a -> (a -> t m b) -> t m b
374forEachWith = P.forEachWith
377-- XXX Deprecate it in 0.8.0
379-- | Make a stream asynchronous, triggers the computation and returns a stream
380-- in the underlying monad representing the output generated by the original
381-- computation. The returned action is exhaustible and must be drained once. If
382-- not drained fully we may have a thread blocked forever and once exhausted it
383-- will always return 'empty'.
386{-# INLINABLE mkAsync #-}
387mkAsync :: (IsStream t, MonadAsync m) => t m a -> m (t m a)
388mkAsync = return . Transform.mkAsync
390------------------------------------------------------------------------------
392------------------------------------------------------------------------------
396-- When a stream consumer demands an element from a serial stream constructed
397-- as @a \`consM` b \`consM` ... nil@, the action @a@ at the head of the stream
398-- sequence is executed and the result is supplied to the consumer. When the
399-- next element is demanded, the action @b@ is executed and its result is
400-- supplied. Thus, the effects are performed and results are consumed strictly
401-- in a serial order. Serial streams can be considered as /spatially ordered/
402-- streams as the order of execution and consumption is the same as the spatial
403-- order in which the actions are composed by the programmer.
405-- Serial streams enforce the side effects as well as the results of the
406-- actions to be in the same order in which the actions are added to the
407-- stream. Therefore, the semigroup operation for serial streams is not
411-- a <> b is not the same as b <> a
414-- There are two serial stream types 'SerialT' and 'WSerialT'. The stream
415-- evaluation of both the variants works in the same way as described above,
416-- they differ only in the 'Semigroup' and 'Monad' implementaitons.
420-- When a stream consumer demands an element from a speculative stream
421-- constructed as @a \`consM` b \`consM` ... nil@, the action @a@ at the head
422-- of the stream is executed and the output of the action is supplied to the
423-- consumer. However, in addition to the action at the head multiple actions
424-- following it may also be executed concurrently and the results buffered.
425-- When the next element is demanded it may be served from the buffer and we
426-- may execute the next action in the sequence to keep the buffer adequately
427-- filled. Thus, the actions are executed concurrently but results consumed in
428-- serial order just like serial streams. `consM` can be used to fold an
429-- infinite lazy container of effects, as the number of concurrent executions
432-- Similar to 'consM', the monadic stream generation (e.g. replicateM) and
433-- transformation operations (e.g. mapM) on speculative streams can execute
434-- multiple effects concurrently in a speculative manner.
436-- How many effects can be executed concurrently and how many results can be
437-- buffered are controlled by 'maxThreads' and 'maxBuffer' combinators
438-- respectively. The actual number of concurrent threads is adjusted according
439-- to the rate at which the consumer is consuming the stream. It may even
440-- execute actions serially in a single thread if that is enough to match the
443-- Speculative streams enforce ordering of the results of actions in the stream
444-- but the side effects are only partially ordered. Therefore, the semigroup
445-- operation for speculative streams is not commutative from the pure outputs
446-- perspective but commutative from side effects perspective.
450-- /Scheduling and execution:/ In an asynchronous stream @a \`consM` b \`consM`
451-- c ...@, the actions @a@, @b@, and @c@ are executed concurrently with the
452-- consumer of the stream. The actions are /scheduled/ for execution in the
453-- same order as they are specified in the stream. Multiple scheduled actions
454-- may be /executed/ concurrently in parallel threads of execution. The
455-- actions may be executed out of order and they may complete at arbitrary
456-- times. Therefore, the /effects/ of the actions may be observed out of
459-- /Buffering:/ The /results/ from multiple threads of execution are queued in
460-- a buffer as soon as they become available. The consumer of the stream is
461-- served from this buffer. Therefore, the consumer may observe the results to
462-- be out of order. In other words, an asynchronous stream is an unordered
463-- stream i.e. order does not matter.
465-- /Concurrency control:/ Threads are suspended if the `maxBuffer` limit is
466-- reached, and resumed when the consumer makes space in the buffer. The
467-- maximum number of concurrent threads depends on `maxThreads`. Number of
468-- threads is increased or decreased based on the speed of the consumer.
470-- /Generation operations:/ Concurrent stream generation operations e.g.
471-- 'Streamly.Prelude.replicateM' when used in async style schedule and execute
472-- the stream generating actions in the manner described above. The generation
473-- actions run concurrently, effects and results of the actions as observed by
474-- the consumer of the stream may be out of order.
476-- /Transformation operations:/ Concurrent stream transformation operations
477-- e.g. 'Streamly.Prelude.mapM', when used in async style, schedule and
478-- execute transformation actions in the manner described above. Transformation
479-- actions run concurrently, effects and results of the actions may be
480-- observed by the consumer out of order.
482-- /Variants:/ There are two asynchronous stream types 'AsyncT' and 'WAsyncT'.
483-- They are identical with respect to single stream evaluation behavior. Their
484-- behaviors differ in how they combine multiple streams using 'Semigroup' or
485-- 'Monad' composition. Since the order of elements does not matter in
486-- asynchronous streams the 'Semigroup' operation is effectively commutative.
490-- 'ZipSerialM' and 'ZipAsyncM', provide 'Applicative' instances for zipping the
491-- corresponding elements of two streams together. Note that these types are
496-- Stream processing functions can be composed in a chain using function
497-- application with or without the '$' operator, or with reverse function
498-- application operator '&'. Streamly provides concurrent versions of these
499-- operators applying stream processing functions such that each stage of the
500-- stream can run in parallel. The operators start with a @|@; we can read '|$'
501-- as "@parallel dollar@" to remember that @|@ comes before '$'.
503-- Imports for the code snippets below:
507-- import qualified Streamly.Prelude as S
508-- import Control.Concurrent
512-- The 'Semigroup' operation '<>' of each stream type combines two streams in a
513-- type specific manner. This section provides polymorphic versions of '<>'
514-- which can be used to combine two streams in a predetermined way irrespective
517-- XXX An alternative design choice would be to let a control parameter affect
518-- the nearest SVar only and then it gets cleared. The benefit of the current
519-- choice is that it is simply just like global configuration, just like state
520-- behaves, so should be easy to comprehend. But it has the downside of leaking
521-- to undesired actions, that is we can forget to reset it.
525-- These combinators can be used at any point in a stream composition to set
526-- parameters to control the concurrency of the /argument stream/. A control
527-- parameter set at any point remains effective for any concurrent combinators
528-- used in the argument stream until it is reset by using the combinator again.
529-- These control parameters have no effect on non-concurrent combinators in the
530-- stream, or on non-concurrent streams.
532-- /Pitfall:/ Remember that 'maxBuffer' in the following example applies to
533-- 'mapM' and any other combinators that may follow it, and it does not apply
534-- to the combinators before it:
543-- If we use '&' instead of '$' the situation will reverse, in the following
544-- example, 'maxBuffer' does not apply to 'mapM', it applies to combinators
545-- that come before it, because those are the arguments to 'maxBuffer':
556-- You may want to use different stream composition styles at different points
557-- in your program. Stream types can be freely converted or adapted from one
558-- type to another. The 'IsStream' type class facilitates type conversion of
559-- one stream type to another. It is not used directly, instead the type
560-- combinators provided below are used for conversions.
562-- To adapt from one monomorphic type (e.g. 'AsyncT') to another monomorphic
563-- type (e.g. 'SerialT') use the 'adapt' combinator. To give a polymorphic code
564-- a specific interpretation or to adapt a specific type to a polymorphic type
565-- use the type specific combinators e.g. 'fromAsync' or 'fromWSerial'. You
566-- cannot adapt polymorphic code to polymorphic code, as the compiler would not know
567-- which specific type you are converting from or to. If you see a an
568-- @ambiguous type variable@ error then most likely you are using 'adapt'
569-- unnecessarily on polymorphic code.
571-- | Same as 'Streamly.Prelude.concatFoldableWith'
574{-# DEPRECATED foldWith "Please use 'Streamly.Prelude.concatFoldableWith' instead." #-}
575{-# INLINEABLE foldWith #-}
576foldWith :: (IsStream t, Foldable f) => (t m a -> t m a -> t m a) -> f (t m a) -> t m a
577foldWith = P.concatFoldableWith
579-- | Same as 'Streamly.Prelude.concatMapFoldableWith'
582{-# DEPRECATED foldMapWith "Please use 'Streamly.Prelude.concatMapFoldableWith' instead." #-}
583{-# INLINEABLE foldMapWith #-}
584foldMapWith :: (IsStream t, Foldable f) => (t m b -> t m b -> t m b) -> (a -> t m b) -> f a -> t m b
585foldMapWith = P.concatMapFoldableWith
587-- | Same as 'Streamly.Prelude.concatForFoldableWith'
590{-# DEPRECATED forEachWith "Please use 'Streamly.Prelude.concatForFoldableWith' instead." #-}
591{-# INLINEABLE forEachWith #-}
592forEachWith :: (IsStream t, Foldable f) => (t m b -> t m b -> t m b) -> f a -> (a -> t m b) -> t m b
593forEachWith = P.concatForFoldableWith
595{-# DEPRECATED serially "Please use 'Streamly.Prelude.fromSerial' instead." #-}
596serially :: IsStream t => SerialT m a -> t m a
599{-# DEPRECATED wSerially "Please use 'Streamly.Prelude.fromWSerial' instead." #-}
600wSerially :: IsStream t => WSerialT m a -> t m a
601wSerially = fromWSerial
603{-# DEPRECATED asyncly "Please use 'Streamly.Prelude.fromAsync' instead." #-}
604asyncly :: IsStream t => AsyncT m a -> t m a
607{-# DEPRECATED aheadly "Please use 'Streamly.Prelude.fromAhead' instead." #-}
608aheadly :: IsStream t => AheadT m a -> t m a
611{-# DEPRECATED wAsyncly "Please use 'Streamly.Prelude.fromWAsync' instead." #-}
612wAsyncly :: IsStream t => WAsyncT m a -> t m a
615{-# DEPRECATED parallely "Please use 'Streamly.Prelude.fromParallel' instead." #-}
616parallely :: IsStream t => ParallelT m a -> t m a
617parallely = fromParallel
619{-# DEPRECATED zipSerially "Please use 'Streamly.Prelude.fromZipSerial' instead." #-}
620zipSerially :: IsStream t => ZipSerialM m a -> t m a
621zipSerially = fromZipSerial
623{-# DEPRECATED zipAsyncly "Please use 'Streamly.Prelude.fromZipAsync' instead." #-}
624zipAsyncly :: IsStream t => ZipAsyncM m a -> t m a
625zipAsyncly = 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