Deprecated: Please use Streamly.Internal.Data.Stream.Concurrent instead.

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 Source #

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

Instances
Instances details
IsStream ParallelT Source # 
Instance details

Defined in Streamly.Internal.Data.Stream.IsStream.Type

Methods

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

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

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

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

MonadTrans ParallelT Source # 
Instance details

Defined in Streamly.Internal.Data.Stream.Parallel

Methods

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

(MonadReader r m, MonadAsync m) => MonadReader r (ParallelT m) Source # 
Instance details

Defined in Streamly.Internal.Data.Stream.Parallel

Methods

ask :: ParallelT m r Source #

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

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

(MonadState s m, MonadAsync m) => MonadState s (ParallelT m) Source # 
Instance details

Defined in Streamly.Internal.Data.Stream.Parallel

Methods

get :: ParallelT m s Source #

put :: s -> ParallelT m () Source #

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

(MonadBase b m, Monad m, MonadAsync m) => MonadBase b (ParallelT m) Source # 
Instance details

Defined in Streamly.Internal.Data.Stream.Parallel

Methods

liftBase :: b α -> ParallelT m α Source #

(MonadIO m, MonadAsync m) => MonadIO (ParallelT m) Source # 
Instance details

Defined in Streamly.Internal.Data.Stream.Parallel

Methods

liftIO :: IO a -> ParallelT m a Source #

(Monad m, MonadAsync m) => Applicative (ParallelT m) Source # 
Instance details

Defined in Streamly.Internal.Data.Stream.Parallel

Methods

pure :: a -> ParallelT m a Source #

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

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

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

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

Monad m => Functor (ParallelT m) Source # 
Instance details

Defined in Streamly.Internal.Data.Stream.Parallel

Methods

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

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

MonadAsync m => Monad (ParallelT m) Source # 
Instance details

Defined in Streamly.Internal.Data.Stream.Parallel

Methods

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

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

return :: a -> ParallelT m a Source #

(MonadThrow m, MonadAsync m) => MonadThrow (ParallelT m) Source # 
Instance details

Defined in Streamly.Internal.Data.Stream.Parallel

Methods

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

MonadAsync m => Monoid (ParallelT m a) Source # 
Instance details

Defined in Streamly.Internal.Data.Stream.Parallel

Methods

mempty :: ParallelT m a Source #

mappend :: ParallelT m a -> ParallelT m a -> ParallelT m a Source #

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

MonadAsync m => Semigroup (ParallelT m a) Source # 
Instance details

Defined in Streamly.Internal.Data.Stream.Parallel

Methods

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

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

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

type Parallel = ParallelT IO Source #

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 Source #

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

Merge Concurrently

parallelK :: MonadAsync m => StreamK m a -> StreamK m a -> StreamK m a Source #

parallelFstK :: MonadAsync m => StreamK m a -> StreamK m a -> StreamK m a Source #

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

Pre-release

parallelMinK :: MonadAsync m => StreamK m a -> StreamK m a -> StreamK m a Source #

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 Source #

Same as mkParallel but for StreamD stream.

mkParallelK :: MonadAsync m => StreamK m a -> StreamK m a Source #

Like mkParallel but uses StreamK internally.

Pre-release

Tap Concurrently

tapAsyncK :: MonadAsync m => (StreamK m a -> m b) -> StreamK m a -> StreamK m a Source #

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.

              StreamK 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 Source #

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

Callbacks

newCallbackStream :: MonadAsync m => m (a -> m (), StreamK m a) Source #

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