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

newtype 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

Constructors

ParallelT

Fields

• getParallelT :: Stream m a

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.IsStream.Type

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

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

XXX we can implement it more efficienty by directly implementing instead of combining streams using parallel.

Merge Concurrently

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

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

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

Pre-release

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

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

Pre-release

Evaluate Concurrently

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

Same as mkParallel but for StreamD stream.

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

Like mkParallel but uses StreamK internally.

Pre-release

Tap Concurrently

tapAsyncK :: MonadAsync m => (Stream m a -> m b) -> Stream m a -> Stream 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.

Callbacks

newCallbackStream :: MonadAsync m => m (a -> m (), Stream 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