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 #-}
60module Streamly {-# DEPRECATED "Please use \"Streamly.Prelude\" instead." #-}
62 -- -- * Concepts Overview
78 -- * Stream transformers
79 -- | A stream represents a sequence of pure or effectful actions. The
80 -- `cons` and `consM` operations and the corresponding operators '.:' and
81 -- '|:' can be used to join pure values or effectful actions in a sequence.
82 -- The effects in the stream can be executed in many different ways
83 -- depending on the type of stream. In other words, the behavior of 'consM'
84 -- depends on the type of the stream.
86 -- There are three high level categories of streams, /spatially ordered/
87 -- streams, /speculative/ streams and /time ordered/ streams. Spatially
88 -- ordered streams, 'SerialT' and 'WSerialT', execute the effects in serial
89 -- order i.e. one at a time and present the outputs of those effects to the
90 -- consumer in the same order. Speculative streams, 'AheadT', may execute
91 -- many effects concurrently but present the outputs to the consumer in the
92 -- specified spatial order. Time ordered streams, 'AsyncT', 'WAsyncT' and
93 -- 'ParallelT', may execute many effects concurrently and present the
94 -- outputs of those effects to the consumer in time order i.e. as soon as
95 -- the output is generated.
97 -- We described above how the effects in a sequence are executed for
98 -- different types of streams. The behvavior of the 'Semigroup' and 'Monad'
99 -- instances follow the behavior of 'consM'. Stream generation operations
100 -- like 'repeatM' also execute the effects differently for different
101 -- streams, providing a concurrent generation capability when used with
102 -- stream types that execute effects concurrently. Similarly, effectful
103 -- transformation operations like 'mapM' also execute the transforming
104 -- effects differently for different types of streams.
111 -- ** Speculative Streams
115 -- ** Asynchronous Streams
121 -- ** Zipping Streams
126 -- * Parallel Function Application
143 -- * Concurrency Control
156 -- * Stream Type Adapters
180 -- ** Folding Containers of Streams
181 -- | These are variants of standard 'Foldable' fold functions that use a
182 -- polymorphic stream sum operation (e.g. 'async' or 'wSerial') to fold a
183 -- finite container of streams. Note that these are just special cases of
184 -- the more general 'concatMapWith' operation.
197import Data.Semigroup (Semigroup(..))
198import Streamly.Internal.Control.Concurrent (MonadAsync)
199import Streamly.Internal.Data.SVar (Rate(..))
200import Streamly.Internal.Data.Stream.Ahead
201import Streamly.Internal.Data.Stream.Async
202import Streamly.Internal.Data.Stream.IsStream.Combinators
203import Streamly.Internal.Data.Stream.IsStream.Expand
204import Streamly.Internal.Data.Stream.IsStream.Type
206import qualified Streamly.Prelude as P
207import qualified Streamly.Internal.Data.Stream.IsStream as IP
208import qualified Streamly.Internal.Data.Stream.IsStream.Transform as Transform
210-- XXX provide good succinct examples of pipelining, merging, splitting etc.
215-- A program is expressed as a network of streams and folds. A stream is a
216-- source or generator of data elements and a fold is a consumer of data elements that
217-- reduces multiple input elements to a single value.
219-- In the following example, a 'Word8' stream is generated by using
220-- 'Streamly.FileSystem.Handle.read' on a file handle, then the
221-- 'Streamly.Prelude.splitBySuffix' transformation splits the stream on
222-- newlines (ascii value 10); it uses the 'Streamly.Data.Fold.drain' fold to reduce
223-- the resulting lines to unit values (@()@), 'Streamly.Prelude.length' fold
224-- then counts the unit elements in the resulting stream which gives us the
225-- number of lines in the file:
227-- > S.length $ S.splitOnSuffix FL.drain 10 $ FH.read fh
229-- The following example folds the lines to arrays of 'Word8' using the
230-- 'Streamly.Data.Array.Foreign.writeF' fold and then wraps the lines in square
231-- brackets before writing them to standard output using
232-- 'Streamly.FileSystem.Handle.write':
234-- > wrapLine ln = S.fromList "[" <> A.read ln <> S.fromList "]\n"
235-- > readLines = S.splitOnSuffix A.writeF 10
236-- > FH.write stdout $ S.concatMap wrapLine $ readLines fh1
238-- One stream can be appended after another:
240-- > FH.write stdout $ S.concatMap wrapLine $ readLines fh1 <> readLines fh2
242-- The following example reads two files concurrently, merges the lines from
243-- the two streams and writes the resulting stream to another file:
245-- > FH.write stdout $ S.concatMap wrapLine $ readLines fh1 `parallel` readLines fh2
247-- There are many ways to generate, merge, zip, transform and fold data
248-- streams. Many transformations can be chained in a stream pipeline. See
249-- "Streamly.Prelude" module for combinators to manipulate streams.
253-- The way stream types in this module like 'SerialT' represent data sources,
254-- the same way the 'Fold' type from "Streamly.Data.Fold" represents data sinks or
255-- reducers of streams. Reducers can be combined to consume a stream source in
256-- many ways. The simplest is to reduce a stream source using a fold e.g.:
258-- > S.fold FL.length $ S.enumerateTo 100
260-- Folds are consumers of streams and can be used to split a stream into
261-- multiple independent flows. Grouping transforms a stream by applying a fold
262-- on segments of a stream, distributing applies multiple folds in parallel on
263-- the same stream and combines them, partitioning sends different elements of
264-- a stream to different folds, unzipping divides the elements of a stream into
265-- parts and sends them through different folds. Parsers are nothing but a
266-- particular type of folds. Transformations can be applied contravariantly on
267-- the input of a fold.
271-- |---transform----Fold m a b--------|
272-- ---stream m a-->transform-->| |---f b c ...
273-- |---transform----Fold m a c--------|
281-- Streamly arrays (See "Streamly.Data.Array.Foreign") complement streams to provide an
282-- efficient computing paradigm. Streams are suitable for immutable
283-- transformations of /potentially infinite/ data using /sequential access/ and
284-- pipelined transformations whereas arrays are suitable for in-place
285-- transformations of /necessarily finite/ data using /random access/. Streams
286-- are synonymous with /sequential pipelined processing/ whereas arrays are
287-- synonymous with /efficient buffering and random access/.
289-- In general, a data processing pipeline reads data from some IO device, does
290-- some processing on it and finally writes the output to another IO device.
291-- Streams provide the overall framework of sequential processing pipeline in
292-- which arrays are used as buffering elements in the middle. In addition to
293-- buffering in the middle, arrays can also be used at the boundaries of the
294-- pipeline to efficiently interface with external storage systems like memory,
295-- files and network. If streams are the pipes in a water pipeline network
296-- then arrays are like the storage tanks in the middle. On the input side,
297-- think of arrays as buckets to fetch water to feed the pipeline and on the
298-- output side buckets to remove the processed water.
300-- 'ByteString' data type from the 'bytestring' package and the 'Text' data
301-- type from the 'text' package are special cases of arrays. 'ByteString' is
302-- like @Array Word8@ and 'Text' is like @utf16@ encoded @Array Word8@.
303-- Streamly arrays can be transformed as efficiently as @bytestring@ or @text@
304-- by using stream operations on them.
306-- Streams and arrays are equally important in computing. They are computing
307-- duals of each other.
310-- The basic stream type is 'Serial', it represents a sequence of IO actions,
311-- and is a 'Monad'. The type 'SerialT' is a monad transformer that can
312-- represent a sequence of actions in an arbitrary monad. The type 'Serial' is
313-- in fact a synonym for @SerialT IO@. There are a few more types similar to
314-- 'SerialT', all of them represent a stream and differ only in the
315-- 'Semigroup', 'Applicative' and 'Monad' compositions of the stream. 'Serial'
316-- and 'WSerial' types compose serially whereas 'Async' and 'WAsync'
317-- types compose concurrently. All these types can be freely inter-converted
318-- using type combinators without any cost. You can freely switch to any type
319-- of composition at any point in the program. When no type annotation or
320-- explicit stream type combinators are used, the default stream type is
321-- inferred as 'Serial'.
323-- This module exports stream types, instances and combinators for:
325-- * converting between different stream types
326-- * appending and concurrently merging streams
327-- * Concurrency control
328-- * Concurrent function application
329-- * Stream rate control
331-- This module is designed to be imported unqualified:
337-- See the "Streamly.Prelude" module for APIs for construction,
338-- generation, elimination and transformation of streams.
340------------------------------------------------------------------------------
341-- Eliminating a stream
342------------------------------------------------------------------------------
344-- | Same as "Streamly.Prelude.runStream".
346{-# DEPRECATED runStream "Please use Streamly.Prelude.drain instead." #-}
347runStream :: Monad m => SerialT m a -> m ()
351-- | Same as "Streamly.Prelude.foldWith".
353{-# DEPRECATED foldWith "Please use Streamly.Prelude.foldWith instead." #-}
354{-# INLINABLE foldWith #-}
355foldWith :: (IsStream t, Foldable f)
356 => (t m a -> t m a -> t m a) -> f (t m a) -> t m a
359-- | Same as "Streamly.Prelude.foldMapWith".
361{-# DEPRECATED foldMapWith "Please use Streamly.Prelude.foldMapWith instead." #-}
362{-# INLINABLE foldMapWith #-}
363foldMapWith :: (IsStream t, Foldable f)
364 => (t m b -> t m b -> t m b) -> (a -> t m b) -> f a -> t m b
365foldMapWith = P.foldMapWith
367-- | Same as "Streamly.Prelude.forEachWith".
369{-# DEPRECATED forEachWith "Please use Streamly.Prelude.forEachWith instead." #-}
370{-# INLINABLE forEachWith #-}
371forEachWith :: (IsStream t, Foldable f)
372 => (t m b -> t m b -> t m b) -> f a -> (a -> t m b) -> t m b
373forEachWith = P.forEachWith
376-- XXX Deprecate it in 0.8.0
378-- | Make a stream asynchronous, triggers the computation and returns a stream
379-- in the underlying monad representing the output generated by the original
380-- computation. The returned action is exhaustible and must be drained once. If
381-- not drained fully we may have a thread blocked forever and once exhausted it
382-- will always return 'empty'.
385{-# INLINABLE mkAsync #-}
386mkAsync :: (IsStream t, MonadAsync m) => t m a -> m (t m a)
387mkAsync = return . Transform.mkAsync
389------------------------------------------------------------------------------
391------------------------------------------------------------------------------
395-- When a stream consumer demands an element from a serial stream constructed
396-- as @a \`consM` b \`consM` ... nil@, the action @a@ at the head of the stream
397-- sequence is executed and the result is supplied to the consumer. When the
398-- next element is demanded, the action @b@ is executed and its result is
399-- supplied. Thus, the effects are performed and results are consumed strictly
400-- in a serial order. Serial streams can be considered as /spatially ordered/
401-- streams as the order of execution and consumption is the same as the spatial
402-- order in which the actions are composed by the programmer.
404-- Serial streams enforce the side effects as well as the results of the
405-- actions to be in the same order in which the actions are added to the
406-- stream. Therefore, the semigroup operation for serial streams is not
410-- a <> b is not the same as b <> a
413-- There are two serial stream types 'SerialT' and 'WSerialT'. The stream
414-- evaluation of both the variants works in the same way as described above,
415-- they differ only in the 'Semigroup' and 'Monad' implementaitons.
419-- When a stream consumer demands an element from a speculative stream
420-- constructed as @a \`consM` b \`consM` ... nil@, the action @a@ at the head
421-- of the stream is executed and the output of the action is supplied to the
422-- consumer. However, in addition to the action at the head multiple actions
423-- following it may also be executed concurrently and the results buffered.
424-- When the next element is demanded it may be served from the buffer and we
425-- may execute the next action in the sequence to keep the buffer adequately
426-- filled. Thus, the actions are executed concurrently but results consumed in
427-- serial order just like serial streams. `consM` can be used to fold an
428-- infinite lazy container of effects, as the number of concurrent executions
431-- Similar to 'consM', the monadic stream generation (e.g. replicateM) and
432-- transformation operations (e.g. mapM) on speculative streams can execute
433-- multiple effects concurrently in a speculative manner.
435-- How many effects can be executed concurrently and how many results can be
436-- buffered are controlled by 'maxThreads' and 'maxBuffer' combinators
437-- respectively. The actual number of concurrent threads is adjusted according
438-- to the rate at which the consumer is consuming the stream. It may even
439-- execute actions serially in a single thread if that is enough to match the
442-- Speculative streams enforce ordering of the results of actions in the stream
443-- but the side effects are only partially ordered. Therefore, the semigroup
444-- operation for speculative streams is not commutative from the pure outputs
445-- perspective but commutative from side effects perspective.
449-- /Scheduling and execution:/ In an asynchronous stream @a \`consM` b \`consM`
450-- c ...@, the actions @a@, @b@, and @c@ are executed concurrently with the
451-- consumer of the stream. The actions are /scheduled/ for execution in the
452-- same order as they are specified in the stream. Multiple scheduled actions
453-- may be /executed/ concurrently in parallel threads of execution. The
454-- actions may be executed out of order and they may complete at arbitrary
455-- times. Therefore, the /effects/ of the actions may be observed out of
458-- /Buffering:/ The /results/ from multiple threads of execution are queued in
459-- a buffer as soon as they become available. The consumer of the stream is
460-- served from this buffer. Therefore, the consumer may observe the results to
461-- be out of order. In other words, an asynchronous stream is an unordered
462-- stream i.e. order does not matter.
464-- /Concurrency control:/ Threads are suspended if the `maxBuffer` limit is
465-- reached, and resumed when the consumer makes space in the buffer. The
466-- maximum number of concurrent threads depends on `maxThreads`. Number of
467-- threads is increased or decreased based on the speed of the consumer.
469-- /Generation operations:/ Concurrent stream generation operations e.g.
470-- 'Streamly.Prelude.replicateM' when used in async style schedule and execute
471-- the stream generating actions in the manner described above. The generation
472-- actions run concurrently, effects and results of the actions as observed by
473-- the consumer of the stream may be out of order.
475-- /Transformation operations:/ Concurrent stream transformation operations
476-- e.g. 'Streamly.Prelude.mapM', when used in async style, schedule and
477-- execute transformation actions in the manner described above. Transformation
478-- actions run concurrently, effects and results of the actions may be
479-- observed by the consumer out of order.
481-- /Variants:/ There are two asynchronous stream types 'AsyncT' and 'WAsyncT'.
482-- They are identical with respect to single stream evaluation behavior. Their
483-- behaviors differ in how they combine multiple streams using 'Semigroup' or
484-- 'Monad' composition. Since the order of elements does not matter in
485-- asynchronous streams the 'Semigroup' operation is effectively commutative.
489-- 'ZipSerialM' and 'ZipAsyncM', provide 'Applicative' instances for zipping the
490-- corresponding elements of two streams together. Note that these types are
495-- Stream processing functions can be composed in a chain using function
496-- application with or without the '$' operator, or with reverse function
497-- application operator '&'. Streamly provides concurrent versions of these
498-- operators applying stream processing functions such that each stage of the
499-- stream can run in parallel. The operators start with a @|@; we can read '|$'
500-- as "@parallel dollar@" to remember that @|@ comes before '$'.
502-- Imports for the code snippets below:
506-- import qualified Streamly.Prelude as S
507-- import Control.Concurrent
511-- The 'Semigroup' operation '<>' of each stream type combines two streams in a
512-- type specific manner. This section provides polymorphic versions of '<>'
513-- which can be used to combine two streams in a predetermined way irrespective
516-- XXX An alternative design choice would be to let a control parameter affect
517-- the nearest SVar only and then it gets cleared. The benefit of the current
518-- choice is that it is simply just like global configuration, just like state
519-- behaves, so should be easy to comprehend. But it has the downside of leaking
520-- to undesired actions, that is we can forget to reset it.
524-- These combinators can be used at any point in a stream composition to set
525-- parameters to control the concurrency of the /argument stream/. A control
526-- parameter set at any point remains effective for any concurrent combinators
527-- used in the argument stream until it is reset by using the combinator again.
528-- These control parameters have no effect on non-concurrent combinators in the
529-- stream, or on non-concurrent streams.
531-- /Pitfall:/ Remember that 'maxBuffer' in the following example applies to
532-- 'mapM' and any other combinators that may follow it, and it does not apply
533-- to the combinators before it:
542-- If we use '&' instead of '$' the situation will reverse, in the following
543-- example, 'maxBuffer' does not apply to 'mapM', it applies to combinators
544-- that come before it, because those are the arguments to 'maxBuffer':
555-- You may want to use different stream composition styles at different points
556-- in your program. Stream types can be freely converted or adapted from one
557-- type to another. The 'IsStream' type class facilitates type conversion of
558-- one stream type to another. It is not used directly, instead the type
559-- combinators provided below are used for conversions.
561-- To adapt from one monomorphic type (e.g. 'AsyncT') to another monomorphic
562-- type (e.g. 'SerialT') use the 'adapt' combinator. To give a polymorphic code
563-- a specific interpretation or to adapt a specific type to a polymorphic type
564-- use the type specific combinators e.g. 'fromAsync' or 'fromWSerial'. You
565-- cannot adapt polymorphic code to polymorphic code, as the compiler would not know
566-- which specific type you are converting from or to. If you see a an
567-- @ambiguous type variable@ error then most likely you are using 'adapt'
568-- unnecessarily on polymorphic code.
570-- | Same as 'Streamly.Prelude.concatFoldableWith'
573{-# DEPRECATED foldWith "Please use 'Streamly.Prelude.concatFoldableWith' instead." #-}
574{-# INLINEABLE foldWith #-}
575foldWith :: (IsStream t, Foldable f) => (t m a -> t m a -> t m a) -> f (t m a) -> t m a
576foldWith = P.concatFoldableWith
578-- | Same as 'Streamly.Prelude.concatMapFoldableWith'
581{-# DEPRECATED foldMapWith "Please use 'Streamly.Prelude.concatMapFoldableWith' instead." #-}
582{-# INLINEABLE foldMapWith #-}
583foldMapWith :: (IsStream t, Foldable f) => (t m b -> t m b -> t m b) -> (a -> t m b) -> f a -> t m b
584foldMapWith = P.concatMapFoldableWith
586-- | Same as 'Streamly.Prelude.concatForFoldableWith'
589{-# DEPRECATED forEachWith "Please use 'Streamly.Prelude.concatForFoldableWith' instead." #-}
590{-# INLINEABLE forEachWith #-}
591forEachWith :: (IsStream t, Foldable f) => (t m b -> t m b -> t m b) -> f a -> (a -> t m b) -> t m b
592forEachWith = P.concatForFoldableWith
594{-# DEPRECATED serially "Please use 'Streamly.Prelude.fromSerial' instead." #-}
595serially :: IsStream t => SerialT m a -> t m a
598{-# DEPRECATED wSerially "Please use 'Streamly.Prelude.fromWSerial' instead." #-}
599wSerially :: IsStream t => WSerialT m a -> t m a
600wSerially = fromWSerial
602{-# DEPRECATED asyncly "Please use 'Streamly.Prelude.fromAsync' instead." #-}
603asyncly :: IsStream t => AsyncT m a -> t m a
606{-# DEPRECATED aheadly "Please use 'Streamly.Prelude.fromAhead' instead." #-}
607aheadly :: IsStream t => AheadT m a -> t m a
610{-# DEPRECATED wAsyncly "Please use 'Streamly.Prelude.fromWAsync' instead." #-}
611wAsyncly :: IsStream t => WAsyncT m a -> t m a
614{-# DEPRECATED parallely "Please use 'Streamly.Prelude.fromParallel' instead." #-}
615parallely :: IsStream t => ParallelT m a -> t m a
616parallely = fromParallel
618{-# DEPRECATED zipSerially "Please use 'Streamly.Prelude.fromZipSerial' instead." #-}
619zipSerially :: IsStream t => ZipSerialM m a -> t m a
620zipSerially = fromZipSerial
622{-# DEPRECATED zipAsyncly "Please use 'Streamly.Prelude.fromZipAsync' instead." #-}
623zipAsyncly :: IsStream t => ZipAsyncM m a -> t m a
624zipAsyncly = 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