A monad that can perform concurrent or parallel IO operations. Streams that can be composed concurrently require the underlying monad to be MonadAsync.

Since: 0.1.0

The Semigroup operation for SerialT behaves like a regular append operation. Therefore, when a <> b is evaluated, stream a is evaluated first until it exhausts and then stream b is evaluated. In other words, the elements of stream b are appended to the elements of stream a. This operation can be used to fold an infinite lazy container of streams.

import Streamly
import qualified Streamly.Prelude as S

main = (S.toList . serially $ (S.fromList [1,2]) <> (S.fromList [3,4])) >>= print

[1,2,3,4]

The Monad instance runs the monadic continuation for each element of the stream, serially.

main = S.drain . serially $ do
    x <- return 1 <> return 2
    S.yieldM $ print x

1
2

SerialT nests streams serially in a depth first manner.

main = S.drain . serially $ do
    x <- return 1 <> return 2
    y <- return 3 <> return 4
    S.yieldM $ print (x, y)

(1,3)
(1,4)
(2,3)
(2,4)

We call the monadic code being run for each element of the stream a monadic continuation. In imperative paradigm we can think of this composition as nested for loops and the monadic continuation is the body of the loop. The loop iterates for all elements of the stream.

Note that the behavior and semantics of SerialT, including Semigroup and Monad instances are exactly like Haskell lists except that SerialT can contain effectful actions while lists are pure.

In the code above, the serially combinator can be omitted as the default stream type is SerialT.

Since: 0.2.0

The Semigroup operation for WSerialT interleaves the elements from the two streams. Therefore, when a <> b is evaluated, stream a is evaluated first to produce the first element of the combined stream and then stream b is evaluated to produce the next element of the combined stream, and then we go back to evaluating stream a and so on. In other words, the elements of stream a are interleaved with the elements of stream b.

Note that when multiple actions are combined like a <> b <> c ... <> z we interleave them in a binary fashion i.e. a and b are interleaved with each other and the result is interleaved with c and so on. This will not act as a true round-robin scheduling across all the streams. Note that this operation cannot be used to fold a container of infinite streams as the state that it needs to maintain is proportional to the number of streams.

import Streamly
import qualified Streamly.Prelude as S

main = (S.toList . wSerially $ (S.fromList [1,2]) <> (S.fromList [3,4])) >>= print

[1,3,2,4]

Similarly, the Monad instance interleaves the iterations of the inner and the outer loop, nesting loops in a breadth first manner.

main = S.drain . wSerially $ do
    x <- return 1 <> return 2
    y <- return 3 <> return 4
    S.yieldM $ print (x, y)

(1,3)
(2,3)
(1,4)
(2,4)

Since: 0.2.0

The Semigroup operation for AheadT appends two streams. The combined stream behaves like a single stream with the actions from the second stream appended to the first stream. The combined stream is evaluated in the speculative style. This operation can be used to fold an infinite lazy container of streams.

import Streamly
import qualified Streamly.Prelude as S
import Control.Concurrent

main = do
 xs <- S.toList . aheadly $ (p 1 |: p 2 |: nil) <> (p 3 |: p 4 |: nil)
 print xs
 where p n = threadDelay 1000000 >> return n

[1,2,3,4]

Any exceptions generated by a constituent stream are propagated to the output stream.

The monad instance of AheadT may run each monadic continuation (bind) concurrently in a speculative manner, performing side effects in a partially ordered manner but producing the outputs in an ordered manner like SerialT.

main = S.drain . aheadly $ do
    n <- return 3 <> return 2 <> return 1
    S.yieldM $ do
         threadDelay (n * 1000000)
         myThreadId >>= \tid -> putStrLn (show tid ++ ": Delay " ++ show n)

ThreadId 40: Delay 1
ThreadId 39: Delay 2
ThreadId 38: Delay 3

Since: 0.3.0

The Semigroup operation for AsyncT appends two streams. The combined stream behaves like a single stream with the actions from the second stream appended to the first stream. The combined stream is evaluated in the asynchronous style. This operation can be used to fold an infinite lazy container of streams.

import Streamly
import qualified Streamly.Prelude as S
import Control.Concurrent

main = (S.toList . asyncly $ (S.fromList [1,2]) <> (S.fromList [3,4])) >>= print

[1,2,3,4]

Any exceptions generated by a constituent stream are propagated to the output stream. The output and exceptions from a single stream are guaranteed to arrive in the same order in the resulting stream as they were generated in the input stream. However, the relative ordering of elements from different streams in the resulting stream can vary depending on scheduling and generation delays.

Similarly, the monad instance of AsyncT may run each iteration concurrently based on demand. More concurrent iterations are started only if the previous iterations are not able to produce enough output for the consumer.

main = drain . asyncly $ do
    n <- return 3 <> return 2 <> return 1
    S.yieldM $ do
         threadDelay (n * 1000000)
         myThreadId >>= \tid -> putStrLn (show tid ++ ": Delay " ++ show n)

ThreadId 40: Delay 1
ThreadId 39: Delay 2
ThreadId 38: Delay 3

Since: 0.1.0

The Semigroup operation for WAsyncT interleaves the elements from the two streams. Therefore, when a <> b is evaluated, one action is picked from stream a for evaluation and then the next action is picked from stream b and then the next action is again picked from stream a, going around in a round-robin fashion. Many such actions are executed concurrently depending on maxThreads and maxBuffer limits. Results are served to the consumer in the order completion of the actions.

Note that when multiple actions are combined like a <> b <> c ... <> z we go in a round-robin fasion across all of them picking one action from each up to z and then come back to a. Note that this operation cannot be used to fold a container of infinite streams as the state that it needs to maintain is proportional to the number of streams.

import Streamly
import qualified Streamly.Prelude as S
import Control.Concurrent

main = (S.toList . wAsyncly $ (S.fromList [1,2]) <> (S.fromList [3,4])) >>= print

[1,3,2,4]

Any exceptions generated by a constituent stream are propagated to the output stream. The output and exceptions from a single stream are guaranteed to arrive in the same order in the resulting stream as they were generated in the input stream. However, the relative ordering of elements from different streams in the resulting stream can vary depending on scheduling and generation delays.

Similarly, the Monad instance of WAsyncT runs all iterations fairly concurrently using a round robin scheduling.

main = drain . wAsyncly $ do
    n <- return 3 <> return 2 <> return 1
    S.yieldM $ do
         threadDelay (n * 1000000)
         myThreadId >>= \tid -> putStrLn (show tid ++ ": Delay " ++ show n)

ThreadId 40: Delay 1
ThreadId 39: Delay 2
ThreadId 38: Delay 3

Since: 0.2.0

Async composition with strict concurrent execution of all streams.

The Semigroup instance of ParallelT executes both the streams concurrently without any delay or without waiting for the consumer demand and merges the results as they arrive. If the consumer does not consume the results, they are buffered upto a configured maximum, controlled by the maxBuffer primitive. If the buffer becomes full the concurrent tasks will block until there is space in the buffer.

Both WAsyncT and ParallelT, evaluate the constituent streams fairly in a round robin fashion. The key difference is that WAsyncT might wait for the consumer demand before it executes the tasks whereas ParallelT starts executing all the tasks immediately without waiting for the consumer demand. For WAsyncT the maxThreads limit applies whereas for ParallelT it does not apply. In other words, WAsyncT can be lazy whereas ParallelT is strict.

ParallelT is useful for cases when the streams are required to be evaluated simultaneously irrespective of how the consumer consumes them e.g. when we want to race two tasks and want to start both strictly at the same time or if we have timers in the parallel tasks and our results depend on the timers being started at the same time. If we do not have such requirements then AsyncT or AheadT are recommended as they can be more efficient than ParallelT.

main = (toList . parallely $ (fromFoldable [1,2]) <> (fromFoldable [3,4])) >>= print

[1,3,2,4]

When streams with more than one element are merged, it yields whichever stream yields first without any bias, unlike the Async style streams.

Any exceptions generated by a constituent stream are propagated to the output stream. The output and exceptions from a single stream are guaranteed to arrive in the same order in the resulting stream as they were generated in the input stream. However, the relative ordering of elements from different streams in the resulting stream can vary depending on scheduling and generation delays.

Similarly, the Monad instance of ParallelT runs all iterations of the loop concurrently.

import Streamly
import qualified Streamly.Prelude as S
import Control.Concurrent

main = drain . parallely $ do
    n <- return 3 <> return 2 <> return 1
    S.yieldM $ do
         threadDelay (n * 1000000)
         myThreadId >>= \tid -> putStrLn (show tid ++ ": Delay " ++ show n)

ThreadId 40: Delay 1
ThreadId 39: Delay 2
ThreadId 38: Delay 3

Note that parallel composition can only combine a finite number of streams as it needs to retain state for each unfinished stream.

Since: 0.7.0 (maxBuffer applies to ParallelT streams)

Since: 0.1.0

The applicative instance of ZipSerialM zips a number of streams serially i.e. it produces one element from each stream serially and then zips all those elements.

main = (toList . zipSerially $ (,,) <$> s1 <*> s2 <*> s3) >>= print
    where s1 = fromFoldable [1, 2]
          s2 = fromFoldable [3, 4]
          s3 = fromFoldable [5, 6]

[(1,3,5),(2,4,6)]

The Semigroup instance of this type works the same way as that of SerialT.

Since: 0.2.0

Like ZipSerialM but zips in parallel, it generates all the elements to be zipped concurrently.

main = (toList . zipAsyncly $ (,,) <$> s1 <*> s2 <*> s3) >>= print
    where s1 = fromFoldable [1, 2]
          s2 = fromFoldable [3, 4]
          s3 = fromFoldable [5, 6]

[(1,3,5),(2,4,6)]

The Semigroup instance of this type works the same way as that of SerialT.

Since: 0.2.0

Parallel function application operator for streams; just like the regular function application operator $ except that it is concurrent. The following code prints a value every second even though each stage adds a 1 second delay.

drain $
   S.mapM (\x -> threadDelay 1000000 >> print x)
     |$ S.repeatM (threadDelay 1000000 >> return 1)

Concurrent

Since: 0.3.0

Parallel reverse function application operator for streams; just like the regular reverse function application operator & except that it is concurrent.

drain $
      S.repeatM (threadDelay 1000000 >> return 1)
   |& S.mapM (\x -> threadDelay 1000000 >> print x)

Concurrent

Since: 0.3.0

Parallel function application operator; applies a run or fold function to a stream such that the fold consumer and the stream producer run in parallel. A run or fold function reduces the stream to a value in the underlying monad. The . at the end of the operator is a mnemonic for termination of the stream.

   S.foldlM' (\_ a -> threadDelay 1000000 >> print a) ()
      |$. S.repeatM (threadDelay 1000000 >> return 1)

Concurrent

Since: 0.3.0

Parallel reverse function application operator for applying a run or fold functions to a stream. Just like |$. except that the operands are reversed.

       S.repeatM (threadDelay 1000000 >> return 1)
   |&. S.foldlM' (\_ a -> threadDelay 1000000 >> print a) ()

Concurrent

Since: 0.3.0

Make a stream asynchronous, triggers the computation and returns a stream in the underlying monad representing the output generated by the original computation. The returned action is exhaustible and must be drained once. If not drained fully we may have a thread blocked forever and once exhausted it will always return empty.

Since: 0.2.0

Polymorphic version of the Semigroup operation <> of SerialT. Appends two streams sequentially, yielding all elements from the first stream, and then all elements from the second stream.

Since: 0.2.0

Polymorphic version of the Semigroup operation <> of WSerialT. Interleaves two streams, yielding one element from each stream alternately. When one stream stops the rest of the other stream is used in the output stream.

Since: 0.2.0

Polymorphic version of the Semigroup operation <> of AheadT. Merges two streams sequentially but with concurrent lookahead.

Since: 0.3.0

Polymorphic version of the Semigroup operation <> of AsyncT. Merges two streams possibly concurrently, preferring the elements from the left one when available.

Since: 0.2.0

Polymorphic version of the Semigroup operation <> of WAsyncT. Merges two streams concurrently choosing elements from both fairly.

Since: 0.2.0

Polymorphic version of the Semigroup operation <> of ParallelT Merges two streams concurrently.

Since: 0.2.0

Specify the maximum number of threads that can be spawned concurrently for any concurrent combinator in a stream. A value of 0 resets the thread limit to default, a negative value means there is no limit. The default value is 1500. maxThreads does not affect ParallelT streams as they can use unbounded number of threads.

When the actions in a stream are IO bound, having blocking IO calls, this option can be used to control the maximum number of in-flight IO requests. When the actions are CPU bound this option can be used to control the amount of CPU used by the stream.

Since: 0.4.0

Specify the maximum size of the buffer for storing the results from concurrent computations. If the buffer becomes full we stop spawning more concurrent tasks until there is space in the buffer. A value of 0 resets the buffer size to default, a negative value means there is no limit. The default value is 1500.

CAUTION! using an unbounded maxBuffer value (i.e. a negative value) coupled with an unbounded maxThreads value is a recipe for disaster in presence of infinite streams, or very large streams. Especially, it must not be used when pure is used in ZipAsyncM streams as pure in applicative zip streams generates an infinite stream causing unbounded concurrent generation with no limit on the buffer or threads.

Since: 0.4.0

Specifies the stream yield rate in yields per second (Hertz). We keep accumulating yield credits at rateGoal. At any point of time we allow only as many yields as we have accumulated as per rateGoal since the start of time. If the consumer or the producer is slower or faster, the actual rate may fall behind or exceed rateGoal. We try to recover the gap between the two by increasing or decreasing the pull rate from the producer. However, if the gap becomes more than rateBuffer we try to recover only as much as rateBuffer.

rateLow puts a bound on how low the instantaneous rate can go when recovering the rate gap. In other words, it determines the maximum yield latency. Similarly, rateHigh puts a bound on how high the instantaneous rate can go when recovering the rate gap. In other words, it determines the minimum yield latency. We reduce the latency by increasing concurrency, therefore we can say that it puts an upper bound on concurrency.

If the rateGoal is 0 or negative the stream never yields a value. If the rateBuffer is 0 or negative we do not attempt to recover.

Since: 0.5.0

Specify the pull rate of a stream. A Nothing value resets the rate to default which is unlimited. When the rate is specified, concurrent production may be ramped up or down automatically to achieve the specified yield rate. The specific behavior for different styles of Rate specifications is documented under Rate. The effective maximum production rate achieved by a stream is governed by:

• The maxThreads limit
• The maxBuffer limit
• The maximum rate that the stream producer can achieve
• The maximum rate that the stream consumer can achieve

Since: 0.5.0

Same as rate (Just $ Rate (r/2) r (2*r) maxBound)

Specifies the average production rate of a stream in number of yields per second (i.e. Hertz). Concurrent production is ramped up or down automatically to achieve the specified average yield rate. The rate can go down to half of the specified rate on the lower side and double of the specified rate on the higher side.

Since: 0.5.0

Same as rate (Just $ Rate r r (2*r) maxBound)

Specifies the minimum rate at which the stream should yield values. As far as possible the yield rate would never be allowed to go below the specified rate, even though it may possibly go above it at times, the upper limit is double of the specified rate.

Since: 0.5.0

Same as rate (Just $ Rate (r/2) r r maxBound)

Specifies the maximum rate at which the stream should yield values. As far as possible the yield rate would never be allowed to go above the specified rate, even though it may possibly go below it at times, the lower limit is half of the specified rate. This can be useful in applications where certain resource usage must not be allowed to go beyond certain limits.

Since: 0.5.0

Same as rate (Just $ Rate r r r 0)

Specifies a constant yield rate. If for some reason the actual rate goes above or below the specified rate we do not try to recover it by increasing or decreasing the rate in future. This can be useful in applications like graphics frame refresh where we need to maintain a constant refresh rate.

Since: 0.5.0

Class of types that can represent a stream of elements of some type a in some monad m.

Since: 0.2.0

Fix the type of a polymorphic stream as SerialT.

Since: 0.1.0

Fix the type of a polymorphic stream as WSerialT.

Since: 0.2.0

Fix the type of a polymorphic stream as AsyncT.

Since: 0.1.0

Fix the type of a polymorphic stream as AheadT.

Since: 0.3.0

Fix the type of a polymorphic stream as WAsyncT.

Since: 0.2.0

Fix the type of a polymorphic stream as ParallelT.

Since: 0.1.0

Fix the type of a polymorphic stream as ZipSerialM.

Since: 0.2.0

Fix the type of a polymorphic stream as ZipAsyncM.

Since: 0.2.0

Adapt any specific stream type to any other specific stream type.

Since: 0.1.0

A serial IO stream of elements of type a. See SerialT documentation for more details.

Since: 0.2.0

An interleaving serial IO stream of elements of type a. See WSerialT documentation for more details.

Since: 0.2.0

A serial IO stream of elements of type a with concurrent lookahead. See AheadT documentation for more details.

Since: 0.3.0

A demand driven left biased parallely composing IO stream of elements of type a. See AsyncT documentation for more details.

Since: 0.2.0

A round robin parallely composing IO stream of elements of type a. See WAsyncT documentation for more details.

Since: 0.2.0

A parallely composing IO stream of elements of type a. See ParallelT documentation for more details.

Since: 0.2.0

An IO stream whose applicative instance zips streams serially.

Since: 0.2.0

An IO stream whose applicative instance zips streams wAsyncly.

Since: 0.2.0

A variant of fold that allows you to fold a Foldable container of streams using the specified stream sum operation.

foldWith async $ map return [1..3]

Equivalent to:

foldWith f = S.foldMapWith f id

Since: 0.1.0 (Streamly)

A variant of foldMap that allows you to map a monadic streaming action on a Foldable container and then fold it using the specified stream merge operation.

foldMapWith async return [1..3]

Equivalent to:

foldMapWith f g xs = S.concatMapWith f g (S.fromFoldable xs)

Since: 0.1.0 (Streamly)

Like foldMapWith but with the last two arguments reversed i.e. the monadic streaming function is the last argument.

Equivalent to:

forEachWith = flip S.foldMapWith

Since: 0.1.0 (Streamly)

The class of semigroups (types with an associative binary operation).

Instances should satisfy the associativity law:

• x <> (y <> z) = (x <> y) <> z

Since: base-4.9.0.0