Streamly.Internal.Data.Stream.Parallel

To run examples in this module:

>>> import qualified Streamly.Prelude as Stream
>>> import Control.Concurrent (threadDelay)
>>> :{
 delay n = do
     threadDelay (n * 1000000)   -- sleep for n seconds
     putStrLn (show n ++ " sec") -- print "n sec"
     return n                    -- IO Int
:}

Parallel Stream Type

data ParallelT m a

For ParallelT streams:

(<>) = parallel
(>>=) = flip . concatMapWith parallel

See AsyncT, ParallelT is similar except that all iterations are strictly concurrent while in AsyncT it depends on the consumer demand and available threads. See parallel for more details.

Since: 0.1.0 (Streamly)

Since: 0.7.0 (maxBuffer applies to ParallelT streams)

Since: 0.8.0

Instances

MonadTrans ParallelT

Defined in Streamly.Internal.Data.Stream.Parallel

Methods

lift :: Monad m => m a -> ParallelT m a

IsStream ParallelT

Defined in Streamly.Internal.Data.Stream.Parallel

Methods

toStream :: forall (m :: Type -> Type) a. ParallelT m a -> Stream m a

fromStream :: forall (m :: Type -> Type) a. Stream m a -> ParallelT m a

consM :: MonadAsync m => m a -> ParallelT m a -> ParallelT m a

(|:) :: MonadAsync m => m a -> ParallelT m a -> ParallelT m a

(MonadBase b m, Monad m, MonadAsync m) => MonadBase b (ParallelT m)

Defined in Streamly.Internal.Data.Stream.Parallel

Methods

liftBase :: b α -> ParallelT m α

(MonadState s m, MonadAsync m) => MonadState s (ParallelT m)

Defined in Streamly.Internal.Data.Stream.Parallel

Methods

get :: ParallelT m s

put :: s -> ParallelT m ()

state :: (s -> (a, s)) -> ParallelT m a

(MonadReader r m, MonadAsync m) => MonadReader r (ParallelT m)

Defined in Streamly.Internal.Data.Stream.Parallel

Methods

ask :: ParallelT m r

local :: (r -> r) -> ParallelT m a -> ParallelT m a

reader :: (r -> a) -> ParallelT m a

MonadAsync m => Monad (ParallelT m)

Defined in Streamly.Internal.Data.Stream.Parallel

Methods

(>>=) :: ParallelT m a -> (a -> ParallelT m b) -> ParallelT m b

(>>) :: ParallelT m a -> ParallelT m b -> ParallelT m b

return :: a -> ParallelT m a

Monad m => Functor (ParallelT m)

Defined in Streamly.Internal.Data.Stream.Parallel

Methods

fmap :: (a -> b) -> ParallelT m a -> ParallelT m b

(<$) :: a -> ParallelT m b -> ParallelT m a

(Monad m, MonadAsync m) => Applicative (ParallelT m)

Defined in Streamly.Internal.Data.Stream.Parallel

Methods

pure :: a -> ParallelT m a

(<*>) :: ParallelT m (a -> b) -> ParallelT m a -> ParallelT m b

liftA2 :: (a -> b -> c) -> ParallelT m a -> ParallelT m b -> ParallelT m c

(*>) :: ParallelT m a -> ParallelT m b -> ParallelT m b

(<*) :: ParallelT m a -> ParallelT m b -> ParallelT m a

(MonadIO m, MonadAsync m) => MonadIO (ParallelT m)

Defined in Streamly.Internal.Data.Stream.Parallel

Methods

liftIO :: IO a -> ParallelT m a

(MonadThrow m, MonadAsync m) => MonadThrow (ParallelT m)

Defined in Streamly.Internal.Data.Stream.Parallel

Methods

throwM :: Exception e => e -> ParallelT m a

MonadAsync m => Semigroup (ParallelT m a)

Defined in Streamly.Internal.Data.Stream.Parallel

Methods

(<>) :: ParallelT m a -> ParallelT m a -> ParallelT m a

sconcat :: NonEmpty (ParallelT m a) -> ParallelT m a

stimes :: Integral b => b -> ParallelT m a -> ParallelT m a

MonadAsync m => Monoid (ParallelT m a)

Defined in Streamly.Internal.Data.Stream.Parallel

Methods

mempty :: ParallelT m a

mappend :: ParallelT m a -> ParallelT m a -> ParallelT m a

mconcat :: [ParallelT m a] -> ParallelT m a

type Parallel = ParallelT IO

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

Since: 0.2.0 (Streamly)

Since: 0.8.0

fromParallel :: IsStream t => ParallelT m a -> t m a

Fix the type of a polymorphic stream as ParallelT.

Since: 0.1.0 (Streamly)

Since: 0.8.0

Merge Concurrently

parallel :: (IsStream t, MonadAsync m) => t m a -> t m a -> t m a infixr 6

Like async except that the execution is much more strict. There is no limit on the number of threads. While async may not schedule a stream if there is no demand from the consumer, parallel always evaluates both the streams immediately. The only limit that applies to parallel is maxBuffer. Evaluation may block if the output buffer becomes full.

>>> import Streamly.Prelude (parallel)
>>> stream = Stream.fromEffect (delay 2) `parallel` Stream.fromEffect (delay 1)
>>> Stream.toList stream -- IO [Int]
1 sec
2 sec
[1,2]

parallel guarantees that all the streams are scheduled for execution immediately, therefore, we could use things like starting timers inside the streams and relying on the fact that all timers were started at the same time.

Unlike async this operation cannot be used to fold an infinite lazy container of streams, because it schedules all the streams strictly concurrently.

Since: 0.2.0 (Streamly)

Since: 0.8.0

parallelFst :: (IsStream t, MonadAsync m) => t m a -> t m a -> t m a

Like parallel but stops the output as soon as the first stream stops.

Pre-release

parallelMin :: (IsStream t, MonadAsync m) => t m a -> t m a -> t m a

Like parallel but stops the output as soon as any of the two streams stops.

Pre-release

Evaluate Concurrently

mkParallel :: (IsStream t, MonadAsync m) => t m a -> t m a

Make the stream producer and consumer run concurrently by introducing a buffer between them. The producer thread evaluates the input stream until the buffer fills, it blocks if the buffer is full until there is space in the buffer. The consumer consumes the stream lazily from the buffer.

mkParallel = D.fromStreamD . mkParallelD . D.toStreamD

Pre-release

mkParallelD :: MonadAsync m => Stream m a -> Stream m a

Same as mkParallel but for StreamD stream.

mkParallelK :: (IsStream t, MonadAsync m) => t m a -> t m a

Like mkParallel but uses StreamK internally.

Pre-release

Tap Concurrently

tapAsync :: (IsStream t, MonadAsync m) => (t m a -> m b) -> t m a -> t m a

Redirect a copy of the stream to a supplied fold and run it concurrently in an independent thread. The fold may buffer some elements. The buffer size is determined by the prevailing maxBuffer setting.

              Stream m a -> m b
                      |
-----stream m a ---------------stream m a-----

> S.drain $ S.tapAsync (S.mapM_ print) (S.enumerateFromTo 1 2)
1
2

Exceptions from the concurrently running fold are propagated to the current computation. Note that, because of buffering in the fold, exceptions may be delayed and may not correspond to the current element being processed in the parent stream, but we guarantee that before the parent stream stops the tap finishes and all exceptions from it are drained.

Compare with tap.

Pre-release

tapAsyncF :: MonadAsync m => Fold m a b -> Stream m a -> Stream m a

Like tapAsync but uses a Fold instead of a fold function.

distributeAsync_ :: (Foldable f, IsStream t, MonadAsync m) => f (t m a -> m b) -> t m a -> t m a

Concurrently distribute a stream to a collection of fold functions, discarding the outputs of the folds.

> Stream.drain $ Stream.distributeAsync_ [Stream.mapM_ print, Stream.mapM_ print] (Stream.enumerateFromTo 1 2)
1
2
1
2

distributeAsync_ = flip (foldr tapAsync)

Pre-release

Callbacks

newCallbackStream :: (IsStream t, MonadAsync m) => m (a -> m (), t m a)

Generates a callback and a stream pair. The callback returned is used to queue values to the stream. The stream is infinite, there is no way for the callback to indicate that it is done now.

Pre-release