2{-# LANGUAGE ConstraintKinds #-}
3{-# LANGUAGE FlexibleContexts #-}
4{-# LANGUAGE FlexibleInstances #-}
5{-# LANGUAGE GeneralizedNewtypeDeriving#-}
6{-# LANGUAGE InstanceSigs #-}
7{-# LANGUAGE MultiParamTypeClasses #-}
8{-# LANGUAGE RankNTypes #-}
9{-# LANGUAGE StandaloneDeriving #-}
10{-# LANGUAGE UndecidableInstances #-} -- XXX
13-- Module : Streamly.Streams
14-- Copyright : (c) 2017 Harendra Kumar
17-- Maintainer : harendra.kumar@gmail.com
18-- Stability : experimental
22module Streamly.Streams
25 , Streaming -- deprecated
39 , runStreaming -- deprecated
68 -- * Stream Transformers
70 , StreamT -- deprecated
72 , InterleavedT -- deprecated
77 , ZipStream -- deprecated
82 , serially -- deprecated
84 , interleaving -- deprecated
90 , zipping -- deprecated
92 , zippingAsync -- deprecated
96 , runStreamT -- deprecated
97 , runInterleavedT -- deprecated
98 , runAsyncT -- deprecated
99 , runParallelT -- deprecated
100 , runZipStream -- deprecated
101 , runZipAsync -- deprecated
110import Control.Monad (ap)
111import Control.Monad.Base (MonadBase (..), liftBaseDefault)
112import Control.Monad.Catch (MonadThrow, throwM)
113-- import Control.Monad.Error.Class (MonadError(..))
114import Control.Monad.IO.Class (MonadIO(..))
115import Control.Monad.Reader.Class (MonadReader(..))
116import Control.Monad.State.Class (MonadState(..))
117import Control.Monad.Trans.Class (MonadTrans (lift))
118import Data.Semigroup (Semigroup(..))
119import Streamly.Core ( MonadAsync , SVar,
121import qualified Streamly.Core as S
123------------------------------------------------------------------------------
124-- Types that can behave as a Stream
125------------------------------------------------------------------------------
130-- | Class of types that can represent a stream of elements of some type 'a' in
134class IsStream t where
135 toStream :: t m a -> S.Stream m a
136 fromStream :: S.Stream m a -> t m a
137 -- | Constructs a stream by adding a monadic action at the head of an
138 -- existing stream. For example:
141 -- > toList $ getLine \`consM` getLine \`consM` nil
147 -- /Concurrent (do not use 'parallely' to construct infinite streams)/
150 consM :: MonadAsync m => m a -> t m a -> t m a
151 -- | Operator equivalent of 'consM'. We can read it as "@parallel colon@"
152 -- to remember that @|@ comes before ':'.
155 -- > toList $ getLine |: getLine |: nil
162 -- let delay = threadDelay 1000000 >> print 1
163 -- runStream $ serially $ delay |: delay |: delay |: nil
164 -- runStream $ parallely $ delay |: delay |: delay |: nil
167 -- /Concurrent (do not use 'parallely' to construct infinite streams)/
170 (|:) :: MonadAsync m => m a -> t m a -> t m a
171 -- We can define (|:) just as 'consM' but it is defined explicitly for each
172 -- type because we want to use SPECIALIZE pragma on the definition.
174-- | Same as 'IsStream'.
177{-# DEPRECATED Streaming "Please use IsStream instead." #-}
178type Streaming = IsStream
180------------------------------------------------------------------------------
181-- Constructing a stream
182------------------------------------------------------------------------------
192nil :: IsStream t => t m a
193nil = fromStream S.nil
195-- | Constructs a stream by adding a monadic action at the head of an existing
196-- stream. For example:
199-- > toList $ getLine \`consM` getLine \`consM` nil
206consMSerial :: (IsStream t, Monad m) => m a -> t m a -> t m a
207consMSerial m r = fromStream $ S.consM m (toStream r)
211-- | Construct a stream by adding a pure value at the head of an existing
212-- stream. For pure values it can be faster than 'consM'. For example:
215-- > toList $ 1 \`cons` 2 \`cons` 3 \`cons` nil
220cons :: IsStream t => a -> t m a -> t m a
221cons a r = fromStream $ S.cons a (toStream r)
225-- | Operator equivalent of 'cons'.
228-- > toList $ 1 .: 2 .: 3 .: nil
233(.:) :: IsStream t => a -> t m a -> t m a
236-- | Build a stream from its church encoding. The function passed maps
237-- directly to the underlying representation of the stream type. The second
238-- parameter to the function is the "yield" function yielding a value and the
239-- remaining stream if any otherwise 'Nothing'. The third parameter is to
240-- represent an "empty" stream.
241streamBuild :: IsStream t
242 => (forall r. Maybe (SVar m a)
243 -> (a -> t m a -> m r)
248streamBuild k = fromStream $ S.Stream $ \svr stp sng yld ->
249 let yield a r = yld a (toStream r)
250 in k svr yield sng stp
252-- | Build a singleton stream from a callback function.
253fromCallback :: IsStream t => (forall r. (a -> m r) -> m r) -> t m a
254fromCallback k = fromStream $ S.Stream $ \_ _ sng _ -> k sng
256-- | Read an SVar to get a stream.
257fromSVar :: (MonadAsync m, IsStream t) => SVar m a -> t m a
258fromSVar sv = fromStream $ S.fromStreamVar sv
260------------------------------------------------------------------------------
261-- Destroying a stream
262------------------------------------------------------------------------------
264-- | Fold a stream using its church encoding. The second argument is the "step"
265-- function consuming an element and the remaining stream, if any. The third
266-- argument is for consuming an "empty" stream that yields nothing.
270 -> (a -> t m a -> m r)
275streamFold svr step single blank m =
276 let yield a x = step a (fromStream x)
277 in (S.runStream (toStream m)) svr blank single yield
279-- | Run a streaming composition, discard the results. By default it interprets
280-- the stream as 'SerialT', to run other types of streams use the type adapting
281-- combinators for example @runStream . 'asyncly'@.
284runStream :: Monad m => SerialT m a -> m ()
285runStream m = go (toStream m)
291 in (S.runStream m1) Nothing stop single yield
293-- | Same as 'runStream'
296{-# DEPRECATED runStreaming "Please use runStream instead." #-}
297runStreaming :: (Monad m, IsStream t) => t m a -> m ()
298runStreaming = runStream . adapt
300-- | Write a stream to an 'SVar' in a non-blocking manner. The stream can then
301-- be read back from the SVar using 'fromSVar'.
302toSVar :: (IsStream t, MonadAsync m) => SVar m a -> t m a -> m ()
303toSVar sv m = S.toStreamVar sv (toStream m)
305------------------------------------------------------------------------------
307------------------------------------------------------------------------------
309-- XXX Get rid of this?
310-- | Make a stream asynchronous, triggers the computation and returns a stream
311-- in the underlying monad representing the output generated by the original
312-- computation. The returned action is exhaustible and must be drained once. If
313-- not drained fully we may have a thread blocked forever and once exhausted it
314-- will always return 'empty'.
317mkAsync :: (IsStream t, MonadAsync m) => t m a -> m (t m a)
319 sv <- S.newStreamVar1 AsyncVar (toStream m)
322{-# INLINE applyWith #-}
323applyWith :: (IsStream t, MonadAsync m)
324 => SVarStyle -> (t m a -> t m b) -> t m a -> t m b
325applyWith style f x = fromStream $
326 S.applyWith style (toStream . f . fromStream) (toStream x)
328{-# INLINE runWith #-}
329runWith :: (IsStream t, MonadAsync m)
330 => SVarStyle -> (t m a -> m b) -> t m a -> m b
331runWith style f x = S.runWith style (f . fromStream) (toStream x)
333-- | Parallel function application operator for streams; just like the regular
334-- function application operator '$' except that it is concurrent. The
335-- following code prints a value every second even though each stage adds a 1
341-- S.mapM (\\x -> threadDelay 1000000 >> print x)
342-- |$ S.repeatM (threadDelay 1000000 >> return 1)
349(|$) :: (IsStream t, MonadAsync m) => (t m a -> t m b) -> t m a -> t m b
350f |$ x = applyWith ParallelVar f x
352-- | Parallel reverse function application operator for streams; just like the
353-- regular reverse function application operator '&' except that it is
358-- S.repeatM (threadDelay 1000000 >> return 1)
359-- |& S.mapM (\\x -> threadDelay 1000000 >> print x)
366(|&) :: (IsStream t, MonadAsync m) => t m a -> (t m a -> t m b) -> t m b
369-- | Parallel function application operator; applies a @run@ or @fold@ function
370-- to a stream such that the fold consumer and the stream producer run in
371-- parallel. A @run@ or @fold@ function reduces the stream to a value in the
375-- S.foldlM' (\\_ a -> threadDelay 1000000 >> print a) ()
376-- <| S.repeatM (threadDelay 1000000 >> return 1)
383(<|) :: (IsStream t, MonadAsync m) => (t m a -> m b) -> t m a -> m b
384f <| x = runWith ParallelVar f x
386-- | Parallel reverse function application operator for applying a run or fold
387-- functions to a stream. Just like '<|' except that the operands are reversed.
390-- S.repeatM (threadDelay 1000000 >> return 1)
391-- |> S.foldlM' (\\_ a -> threadDelay 1000000 >> print a) ()
398(|>) :: (IsStream t, MonadAsync m) => t m a -> (t m a -> m b) -> m b
401------------------------------------------------------------------------------
402-- CPP macros for common instances
403------------------------------------------------------------------------------
405-- XXX use template haskell instead and include Monoid and IsStream instances
408withLocal :: MonadReader r m => (r -> r) -> S.Stream m a -> S.Stream m a
410 S.Stream $ \svr stp sng yld ->
411 let single = local f . sng
412 yield a r = local f $ yld a (withLocal f r)
413 in (S.runStream m) svr (local f stp) single yield
416-- XXX handle and test cross thread state transfer
419 => S.Stream m a -> (e -> S.Stream m a) -> S.Stream m a
421 S.Stream $ \svr stp sng yld ->
422 let run x = S.runStream x svr stp sng yield
423 handle r = r `catchError` \e -> run $ h e
424 yield a r = yld a (withCatchError r h)
428#define MONADPARALLEL , MonadAsync m
430#define MONAD_APPLICATIVE_INSTANCE(STREAM,CONSTRAINT) \
431instance (Monad m CONSTRAINT) => Applicative (STREAM m) where { \
432 pure = STREAM . S.singleton; \
435#define MONAD_COMMON_INSTANCES(STREAM,CONSTRAINT) \
436instance (MonadBase b m, Monad m CONSTRAINT) => MonadBase b (STREAM m) where {\
437 liftBase = liftBaseDefault }; \
439instance (MonadIO m CONSTRAINT) => MonadIO (STREAM m) where { \
440 liftIO = lift . liftIO }; \
442instance (MonadThrow m CONSTRAINT) => MonadThrow (STREAM m) where { \
443 throwM = lift . throwM }; \
446instance (MonadError e m CONSTRAINT) => MonadError e (STREAM m) where { \
447 throwError = lift . throwError; \
449 fromStream $ withCatchError (toStream m) (\e -> toStream $ h e) }; \
452instance (MonadReader r m CONSTRAINT) => MonadReader r (STREAM m) where { \
454 local f m = fromStream $ withLocal f (toStream m) }; \
456instance (MonadState s m CONSTRAINT) => MonadState s (STREAM m) where { \
458 put x = lift (put x); \
459 state k = lift (state k) }
461------------------------------------------------------------------------------
463------------------------------------------------------------------------------
465-- | Deep serial composition or serial composition with depth first traversal.
466-- The 'Semigroup' instance of 'SerialT' appends two streams serially in a
467-- depth first manner, yielding all elements from the first stream, and then
468-- all elements from the second stream.
472-- import qualified "Streamly.Prelude" as S
474-- main = ('toList' . 'serially' $ (fromFoldable [1,2]) \<\> (fromFoldable [3,4])) >>= print
480-- The 'Monad' instance runs the /monadic continuation/ for each
481-- element of the stream, serially.
484-- main = 'runStream' . 'serially' $ do
485-- x <- return 1 \<\> return 2
493-- 'SerialT' nests streams serially in a depth first manner.
496-- main = 'runStream' . 'serially' $ do
497-- x <- return 1 \<\> return 2
498-- y <- return 3 \<\> return 4
499-- S.once $ print (x, y)
508-- This behavior of 'SerialT' is exactly like a list transformer. We call the
509-- monadic code being run for each element of the stream a monadic
510-- continuation. In imperative paradigm we can think of this composition as
511-- nested @for@ loops and the monadic continuation is the body of the loop. The
512-- loop iterates for all elements of the stream.
514-- The 'serially' combinator can be omitted as the default stream type is
516-- Note that serial composition with depth first traversal can be used to
517-- combine an infinite number of streams as it explores only one stream at a
521newtype SerialT m a = SerialT {getSerialT :: S.Stream m a}
522 deriving (Semigroup, Monoid, Functor, MonadTrans)
526{-# DEPRECATED StreamT "Please use 'SerialT' instead." #-}
527type StreamT = SerialT
529instance IsStream SerialT where
530 toStream = getSerialT
534 {-# SPECIALIZE consM :: IO a -> SerialT IO a -> SerialT IO a #-}
535 consM :: Monad m => m a -> SerialT m a -> SerialT m a
539 {-# SPECIALIZE (|:) :: IO a -> SerialT IO a -> SerialT IO a #-}
540 (|:) :: Monad m => m a -> SerialT m a -> SerialT m a
543------------------------------------------------------------------------------
545------------------------------------------------------------------------------
547-- | Polymorphic version of the 'Semigroup' operation '<>' of 'SerialT'.
548-- Appends two streams sequentially, yielding all elements from the first
549-- stream, and then all elements from the second stream.
553serial :: IsStream t => t m a -> t m a -> t m a
554serial m1 m2 = fromStream $ S.serial (toStream m1) (toStream m2)
556------------------------------------------------------------------------------
558------------------------------------------------------------------------------
560instance Monad m => Monad (SerialT m) where
562 (SerialT (S.Stream m)) >>= f = SerialT $ S.Stream $ \_ stp sng yld ->
563 let run x = (S.runStream x) Nothing stp sng yld
564 single a = run $ toStream (f a)
565 yield a r = run $ toStream $ f a <> (fromStream r >>= f)
566 in m Nothing stp single yield
568------------------------------------------------------------------------------
570------------------------------------------------------------------------------
572MONAD_APPLICATIVE_INSTANCE(SerialT,)
573MONAD_COMMON_INSTANCES(SerialT,)
575------------------------------------------------------------------------------
577------------------------------------------------------------------------------
579-- | Wide serial composition or serial composition with a breadth first
580-- traversal. The 'Semigroup' instance of 'WSerialT' traverses
581-- the two streams in a breadth first manner. In other words, it interleaves
582-- two streams, yielding one element from each stream alternately.
586-- import qualified "Streamly.Prelude" as S
588-- main = ('toList' . 'wSerially' $ (fromFoldable [1,2]) \<\> (fromFoldable [3,4])) >>= print
594-- Similarly, the 'Monad' instance interleaves the iterations of the
595-- inner and the outer loop, nesting loops in a breadth first manner.
599-- main = 'runStream' . 'wSerially' $ do
600-- x <- return 1 \<\> return 2
601-- y <- return 3 \<\> return 4
602-- S.once $ print (x, y)
611-- Note that a serial composition with breadth first traversal can only combine
612-- a finite number of streams as it needs to retain state for each unfinished
616newtype WSerialT m a = WSerialT {getWSerialT :: S.Stream m a}
617 deriving (Functor, MonadTrans)
621{-# DEPRECATED InterleavedT "Please use 'WSerialT' instead." #-}
622type InterleavedT = WSerialT
624instance IsStream WSerialT where
625 toStream = getWSerialT
626 fromStream = WSerialT
629 {-# SPECIALIZE consM :: IO a -> WSerialT IO a -> WSerialT IO a #-}
630 consM :: Monad m => m a -> WSerialT m a -> WSerialT m a
634 {-# SPECIALIZE (|:) :: IO a -> WSerialT IO a -> WSerialT IO a #-}
635 (|:) :: Monad m => m a -> WSerialT m a -> WSerialT m a
638------------------------------------------------------------------------------
640------------------------------------------------------------------------------
642-- | Polymorphic version of the 'Semigroup' operation '<>' of 'WSerialT'.
643-- Interleaves two streams, yielding one element from each stream alternately.
646{-# INLINE wSerial #-}
647wSerial :: IsStream t => t m a -> t m a -> t m a
648wSerial m1 m2 = fromStream $ S.wSerial (toStream m1) (toStream m2)
650instance Semigroup (WSerialT m a) where
655-- | Same as 'wSerial'.
658{-# DEPRECATED (<=>) "Please use 'wSerial' instead." #-}
660(<=>) :: IsStream t => t m a -> t m a -> t m a
663------------------------------------------------------------------------------
665------------------------------------------------------------------------------
667instance Monoid (WSerialT m a) where
671------------------------------------------------------------------------------
673------------------------------------------------------------------------------
675instance Monad m => Monad (WSerialT m) where
677 (WSerialT (S.Stream m)) >>= f = WSerialT $ S.Stream $ \_ stp sng yld ->
678 let run x = (S.runStream x) Nothing stp sng yld
679 single a = run $ toStream (f a)
680 yield a r = run $ toStream $ f a <> (fromStream r >>= f)
681 in m Nothing stp single yield
683------------------------------------------------------------------------------
685------------------------------------------------------------------------------
687MONAD_APPLICATIVE_INSTANCE(WSerialT,)
688MONAD_COMMON_INSTANCES(WSerialT,)
690------------------------------------------------------------------------------
692------------------------------------------------------------------------------
694-- | Deep ahead composition or ahead composition with depth first traversal.
695-- The semigroup composition of 'AheadT' appends streams in a depth first
696-- manner just like 'SerialT' except that it can produce elements concurrently
697-- ahead of time. It is like 'AsyncT' except that 'AsyncT' produces the output
698-- as it arrives whereas 'AheadT' orders the output in the traversal order.
701-- main = ('toList' . 'aheadly' $ (fromFoldable [1,2]) \<> (fromFoldable [3,4])) >>= print
707-- Any exceptions generated by a constituent stream are propagated to the
710-- Similarly, the monad instance of 'AheadT' may run each iteration
711-- concurrently ahead of time but presents the results in the same order as
716-- import qualified "Streamly.Prelude" as S
717-- import Control.Concurrent
719-- main = 'runStream' . 'aheadly' $ do
720-- n <- return 3 \<\> return 2 \<\> return 1
722-- threadDelay (n * 1000000)
723-- myThreadId >>= \\tid -> putStrLn (show tid ++ ": Delay " ++ show n)
726-- ThreadId 40: Delay 1
727-- ThreadId 39: Delay 2
728-- ThreadId 38: Delay 3
731-- All iterations may run in the same thread if they do not block.
733-- Note that ahead composition with depth first traversal can be used to
734-- combine infinite number of streams as it explores only a bounded number of
738newtype AheadT m a = AheadT {getAheadT :: S.Stream m a}
739 deriving (Functor, MonadTrans)
741instance IsStream AheadT where
746 {-# SPECIALIZE consM :: IO a -> AheadT IO a -> AheadT IO a #-}
747 consM m r = fromStream $ S.consMAhead m (toStream r)
750 {-# SPECIALIZE (|:) :: IO a -> AheadT IO a -> AheadT IO a #-}
753------------------------------------------------------------------------------
755------------------------------------------------------------------------------
757-- | Polymorphic version of the 'Semigroup' operation '<>' of 'AheadT'.
758-- Merges two streams sequentially but with concurrent lookahead.
762ahead :: (IsStream t, MonadAsync m) => t m a -> t m a -> t m a
763ahead m1 m2 = fromStream $ S.ahead (toStream m1) (toStream m2)
765instance MonadAsync m => Semigroup (AheadT m a) where
768------------------------------------------------------------------------------
770------------------------------------------------------------------------------
772instance MonadAsync m => Monoid (AheadT m a) where
776------------------------------------------------------------------------------
778------------------------------------------------------------------------------
780{-# INLINE aheadbind #-}
784 -> (a -> S.Stream m b)
789 S.Stream $ \ctx stp sng yld ->
790 let run x = (S.runStream x) ctx stp sng yld
792 yield a r = run $ f a `S.ahead` go r
793 in g Nothing stp single yield
795instance MonadAsync m => Monad (AheadT m) where
797 (AheadT m) >>= f = AheadT $ aheadbind m (getAheadT . f)
799------------------------------------------------------------------------------
801------------------------------------------------------------------------------
803MONAD_APPLICATIVE_INSTANCE(AheadT,MONADPARALLEL)
804MONAD_COMMON_INSTANCES(AheadT, MONADPARALLEL)
806------------------------------------------------------------------------------
808------------------------------------------------------------------------------
810-- | Deep async composition or async composition with depth first traversal. In
811-- a left to right 'Semigroup' composition it tries to yield elements from the
812-- left stream as long as it can, but it can run the right stream in parallel
813-- if it needs to, based on demand. The right stream can be run if the left
814-- stream blocks on IO or cannot produce elements fast enough for the consumer.
817-- main = ('toList' . 'asyncly' $ (fromFoldable [1,2]) \<> (fromFoldable [3,4])) >>= print
823-- Any exceptions generated by a constituent stream are propagated to the
824-- output stream. The output and exceptions from a single stream are guaranteed
825-- to arrive in the same order in the resulting stream as they were generated
826-- in the input stream. However, the relative ordering of elements from
827-- different streams in the resulting stream can vary depending on scheduling
828-- and generation delays.
830-- Similarly, the monad instance of 'AsyncT' /may/ run each iteration
831-- concurrently based on demand. More concurrent iterations are started only
832-- if the previous iterations are not able to produce enough output for the
837-- import qualified "Streamly.Prelude" as S
838-- import Control.Concurrent
840-- main = 'runStream' . 'asyncly' $ do
841-- n <- return 3 \<\> return 2 \<\> return 1
843-- threadDelay (n * 1000000)
844-- myThreadId >>= \\tid -> putStrLn (show tid ++ ": Delay " ++ show n)
847-- ThreadId 40: Delay 1
848-- ThreadId 39: Delay 2
849-- ThreadId 38: Delay 3
852-- All iterations may run in the same thread if they do not block.
854-- Note that async composition with depth first traversal can be used to
855-- combine infinite number of streams as it explores only a bounded number of
859newtype AsyncT m a = AsyncT {getAsyncT :: S.Stream m a}
860 deriving (Functor, MonadTrans)
862instance IsStream AsyncT where
867 {-# SPECIALIZE consM :: IO a -> AsyncT IO a -> AsyncT IO a #-}
868 consM m r = fromStream $ S.consMAsync m (toStream r)
871 {-# SPECIALIZE (|:) :: IO a -> AsyncT IO a -> AsyncT IO a #-}
874------------------------------------------------------------------------------
876------------------------------------------------------------------------------
878-- | Polymorphic version of the 'Semigroup' operation '<>' of 'AsyncT'.
879-- Merges two streams possibly concurrently, preferring the
880-- elements from the left one when available.
884async :: (IsStream t, MonadAsync m) => t m a -> t m a -> t m a
885async m1 m2 = fromStream $ S.async (toStream m1) (toStream m2)
887instance MonadAsync m => Semigroup (AsyncT m a) where
890------------------------------------------------------------------------------
892------------------------------------------------------------------------------
894instance MonadAsync m => Monoid (AsyncT m a) where
898------------------------------------------------------------------------------
900------------------------------------------------------------------------------
902{-# INLINE parbind #-}
904 :: (forall c. S.Stream m c -> S.Stream m c -> S.Stream m c)
906 -> (a -> S.Stream m b)
908parbind par m f = go m
911 S.Stream $ \ctx stp sng yld ->
912 let run x = (S.runStream x) ctx stp sng yld
914 yield a r = run $ f a `par` go r
915 in g Nothing stp single yield
917instance MonadAsync m => Monad (AsyncT m) where
919 (AsyncT m) >>= f = AsyncT $ parbind S.async m (getAsyncT . f)
921------------------------------------------------------------------------------
923------------------------------------------------------------------------------
925MONAD_APPLICATIVE_INSTANCE(AsyncT,MONADPARALLEL)
926MONAD_COMMON_INSTANCES(AsyncT, MONADPARALLEL)
928------------------------------------------------------------------------------
930------------------------------------------------------------------------------
932-- | Wide async composition or async composition with breadth first traversal.
933-- The Semigroup instance of 'WAsyncT' concurrently /traverses/ the composed
934-- streams using a depth first travesal or in a round robin fashion, yielding
935-- elements from both streams alternately.
938-- main = ('toList' . 'wAsyncly' $ (fromFoldable [1,2]) \<> (fromFoldable [3,4])) >>= print
944-- Any exceptions generated by a constituent stream are propagated to the
945-- output stream. The output and exceptions from a single stream are guaranteed
946-- to arrive in the same order in the resulting stream as they were generated
947-- in the input stream. However, the relative ordering of elements from
948-- different streams in the resulting stream can vary depending on scheduling
949-- and generation delays.
951-- Similarly, the 'Monad' instance of 'WAsyncT' runs /all/ iterations fairly
952-- concurrently using a round robin scheduling.
956-- import qualified "Streamly.Prelude" as S
957-- import Control.Concurrent
959-- main = 'runStream' . 'wAsyncly' $ do
960-- n <- return 3 \<\> return 2 \<\> return 1
962-- threadDelay (n * 1000000)
963-- myThreadId >>= \\tid -> putStrLn (show tid ++ ": Delay " ++ show n)
966-- ThreadId 40: Delay 1
967-- ThreadId 39: Delay 2
968-- ThreadId 38: Delay 3
971-- Unlike 'AsyncT' all iterations are guaranteed to run fairly
972-- concurrently, unconditionally.
974-- Note that async composition with breadth first traversal can only combine a
975-- finite number of streams as it needs to retain state for each unfinished
979newtype WAsyncT m a = WAsyncT {getWAsyncT :: S.Stream m a}
980 deriving (Functor, MonadTrans)
982instance IsStream WAsyncT where
983 toStream = getWAsyncT
987 {-# SPECIALIZE consM :: IO a -> WAsyncT IO a -> WAsyncT IO a #-}
988 consM m r = fromStream $ S.consMWAsync m (toStream r)
991 {-# SPECIALIZE (|:) :: IO a -> WAsyncT IO a -> WAsyncT IO a #-}
994------------------------------------------------------------------------------
996------------------------------------------------------------------------------
998-- | Polymorphic version of the 'Semigroup' operation '<>' of 'WAsyncT'.
999-- Merges two streams concurrently choosing elements from both fairly.
1002{-# INLINE wAsync #-}
1003wAsync :: (IsStream t, MonadAsync m) => t m a -> t m a -> t m a
1004wAsync m1 m2 = fromStream $ S.wAsync (toStream m1) (toStream m2)
1006instance MonadAsync m => Semigroup (WAsyncT m a) where
1009------------------------------------------------------------------------------
1011------------------------------------------------------------------------------
1013instance MonadAsync m => Monoid (WAsyncT m a) where
1017------------------------------------------------------------------------------
1019------------------------------------------------------------------------------
1021instance MonadAsync m => Monad (WAsyncT m) where
1024 WAsyncT $ parbind S.wAsync m (getWAsyncT . f)
1026------------------------------------------------------------------------------
1028------------------------------------------------------------------------------
1030MONAD_APPLICATIVE_INSTANCE(WAsyncT,MONADPARALLEL)
1031MONAD_COMMON_INSTANCES(WAsyncT, MONADPARALLEL)
1033------------------------------------------------------------------------------
1035------------------------------------------------------------------------------
1037-- | Async composition with simultaneous traversal of all streams.
1039-- The Semigroup instance of 'ParallelT' concurrently /merges/ two streams,
1040-- running both strictly concurrently and yielding elements from both streams
1041-- as they arrive. When multiple streams are combined using 'ParallelT' each
1042-- one is evaluated in its own thread and the results produced are presented in
1043-- the combined stream on a first come first serve basis.
1045-- 'AsyncT' and 'WAsyncT' are /concurrent lookahead streams/ each with a
1046-- specific type of consumption pattern (depth first or breadth first). Since
1047-- they are lookahead, they may introduce certain default latency in starting
1048-- more concurrent tasks for efficiency reasons or may put a default limitation
1049-- on the resource consumption (e.g. number of concurrent threads for
1050-- lookahead). If we look at the implementation detail, they both can share a
1051-- pool of worker threads to evaluate the streams in the desired pattern and at
1052-- the desired rate. However, 'ParallelT' uses a separate runtime thread to
1053-- evaluate each stream.
1055-- 'WAsyncT' is similar to 'ParallelT', as both of them evaluate the
1056-- constituent streams fairly in a round robin fashion.
1057-- However, the key difference is that 'WAsyncT' is lazy or pull driven
1058-- whereas 'ParallelT' is strict or push driven. 'ParallelT' immediately
1059-- starts concurrent evaluation of both the streams (in separate threads) and
1060-- later picks the results whereas 'WAsyncT' may wait for a certain latency
1061-- threshold before initiating concurrent evaluation of the next stream. The
1062-- concurrent scheduling of the next stream or the degree of concurrency is
1063-- driven by the feedback from the consumer. In case of 'ParallelT' each stream
1064-- is evaluated in a separate thread and results are /pushed/ to a shared
1065-- output buffer, the evaluation rate is controlled by blocking when the buffer
1068-- Concurrent lookahead streams are generally more efficient than
1069-- 'ParallelT' and can work pretty efficiently even for smaller tasks because
1070-- they do not necessarily use a separate thread for each task. So they should
1071-- be preferred over 'ParallelT' especially when efficiency is a concern and
1072-- simultaneous strict evaluation is not a requirement. 'ParallelT' is useful
1073-- for cases when the streams are required to be evaluated simultaneously
1074-- irrespective of how the consumer consumes them e.g. when we want to race
1075-- two tasks and want to start both strictly at the same time or if we have
1076-- timers in the parallel tasks and our results depend on the timers being
1077-- started at the same time. We can say that 'ParallelT' is almost the same
1078-- (modulo some implementation differences) as 'WAsyncT' when the latter is
1079-- used with unlimited lookahead and zero latency in initiating lookahead.
1082-- main = ('toList' . 'parallely' $ (fromFoldable [1,2]) \<> (fromFoldable [3,4])) >>= print
1088-- When streams with more than one element are merged, it yields whichever
1089-- stream yields first without any bias, unlike the 'Async' style streams.
1091-- Any exceptions generated by a constituent stream are propagated to the
1092-- output stream. The output and exceptions from a single stream are guaranteed
1093-- to arrive in the same order in the resulting stream as they were generated
1094-- in the input stream. However, the relative ordering of elements from
1095-- different streams in the resulting stream can vary depending on scheduling
1096-- and generation delays.
1098-- Similarly, the 'Monad' instance of 'ParallelT' runs /all/ iterations
1099-- of the loop concurrently.
1103-- import qualified "Streamly.Prelude" as S
1104-- import Control.Concurrent
1106-- main = 'runStream' . 'parallely' $ do
1107-- n <- return 3 \<\> return 2 \<\> return 1
1109-- threadDelay (n * 1000000)
1110-- myThreadId >>= \\tid -> putStrLn (show tid ++ ": Delay " ++ show n)
1113-- ThreadId 40: Delay 1
1114-- ThreadId 39: Delay 2
1115-- ThreadId 38: Delay 3
1118-- Note that parallel composition can only combine a finite number of
1119-- streams as it needs to retain state for each unfinished stream.
1122newtype ParallelT m a = ParallelT {getParallelT :: S.Stream m a}
1123 deriving (Functor, MonadTrans)
1125instance IsStream ParallelT where
1126 toStream = getParallelT
1127 fromStream = ParallelT
1129 {-# INLINE consM #-}
1130 {-# SPECIALIZE consM :: IO a -> ParallelT IO a -> ParallelT IO a #-}
1131 consM m r = fromStream $ S.consMParallel m (toStream r)
1134 {-# SPECIALIZE (|:) :: IO a -> ParallelT IO a -> ParallelT IO a #-}
1137------------------------------------------------------------------------------
1139------------------------------------------------------------------------------
1141-- | Polymorphic version of the 'Semigroup' operation '<>' of 'ParallelT'
1142-- Merges two streams concurrently.
1145{-# INLINE parallel #-}
1146parallel :: (IsStream t, MonadAsync m) => t m a -> t m a -> t m a
1147parallel m1 m2 = fromStream $ S.parallel (toStream m1) (toStream m2)
1149instance MonadAsync m => Semigroup (ParallelT m a) where
1152------------------------------------------------------------------------------
1154------------------------------------------------------------------------------
1156instance MonadAsync m => Monoid (ParallelT m a) where
1160------------------------------------------------------------------------------
1162------------------------------------------------------------------------------
1164instance MonadAsync m => Monad (ParallelT m) where
1166 (ParallelT m) >>= f = ParallelT $ parbind S.parallel m (getParallelT . f)
1168------------------------------------------------------------------------------
1170------------------------------------------------------------------------------
1172MONAD_APPLICATIVE_INSTANCE(ParallelT,MONADPARALLEL)
1173MONAD_COMMON_INSTANCES(ParallelT, MONADPARALLEL)
1175------------------------------------------------------------------------------
1176-- Serially Zipping Streams
1177------------------------------------------------------------------------------
1179-- | The applicative instance of 'ZipSerialM' zips a number of streams serially
1180-- i.e. it produces one element from each stream serially and then zips all
1184-- main = (toList . 'zipSerially' $ (,,) \<$\> s1 \<*\> s2 \<*\> s3) >>= print
1185-- where s1 = fromFoldable [1, 2]
1186-- s2 = fromFoldable [3, 4]
1187-- s3 = fromFoldable [5, 6]
1193-- The 'Semigroup' instance of this type works the same way as that of
1197newtype ZipSerialM m a = ZipSerialM {getZipSerialM :: S.Stream m a}
1198 deriving (Functor, Semigroup, Monoid)
1202{-# DEPRECATED ZipStream "Please use 'ZipSerialM' instead." #-}
1203type ZipStream = ZipSerialM
1205instance IsStream ZipSerialM where
1206 toStream = getZipSerialM
1207 fromStream = ZipSerialM
1209 {-# INLINE consM #-}
1210 {-# SPECIALIZE consM :: IO a -> ZipSerialM IO a -> ZipSerialM IO a #-}
1211 consM :: Monad m => m a -> ZipSerialM m a -> ZipSerialM m a
1215 {-# SPECIALIZE (|:) :: IO a -> ZipSerialM IO a -> ZipSerialM IO a #-}
1216 (|:) :: Monad m => m a -> ZipSerialM m a -> ZipSerialM m a
1219instance Monad m => Applicative (ZipSerialM m) where
1220 pure = ZipSerialM . S.repeat
1221 m1 <*> m2 = fromStream $ S.zipWith id (toStream m1) (toStream m2)
1223------------------------------------------------------------------------------
1224-- Parallely Zipping Streams
1225------------------------------------------------------------------------------
1227-- | Like 'ZipSerialM' but zips in parallel, it generates all the elements to
1228-- be zipped concurrently.
1231-- main = (toList . 'zipAsyncly' $ (,,) \<$\> s1 \<*\> s2 \<*\> s3) >>= print
1232-- where s1 = fromFoldable [1, 2]
1233-- s2 = fromFoldable [3, 4]
1234-- s3 = fromFoldable [5, 6]
1240-- The 'Semigroup' instance of this type works the same way as that of
1244newtype ZipAsyncM m a = ZipAsyncM {getZipAsyncM :: S.Stream m a}
1245 deriving (Functor, Semigroup, Monoid)
1247instance IsStream ZipAsyncM where
1248 toStream = getZipAsyncM
1249 fromStream = ZipAsyncM
1251 {-# INLINE consM #-}
1252 {-# SPECIALIZE consM :: IO a -> ZipAsyncM IO a -> ZipAsyncM IO a #-}
1253 consM :: Monad m => m a -> ZipAsyncM m a -> ZipAsyncM m a
1257 {-# SPECIALIZE (|:) :: IO a -> ZipAsyncM IO a -> ZipAsyncM IO a #-}
1258 (|:) :: Monad m => m a -> ZipAsyncM m a -> ZipAsyncM m a
1261instance MonadAsync m => Applicative (ZipAsyncM m) where
1262 pure = ZipAsyncM . S.repeat
1263 m1 <*> m2 = fromStream $ S.zipAsyncWith id (toStream m1) (toStream m2)
1265-------------------------------------------------------------------------------
1266-- Type adapting combinators
1267-------------------------------------------------------------------------------
1269-- | Adapt any specific stream type to any other specific stream type.
1272adapt :: (IsStream t1, IsStream t2) => t1 m a -> t2 m a
1273adapt = fromStream . toStream
1275-- | Fix the type of a polymorphic stream as 'SerialT'.
1278serially :: IsStream t => SerialT m a -> t m a
1281-- | Fix the type of a polymorphic stream as 'WSerialT'.
1284wSerially :: IsStream t => WSerialT m a -> t m a
1287-- | Same as 'wSerially'.
1290{-# DEPRECATED interleaving "Please use wSerially instead." #-}
1291interleaving :: IsStream t => WSerialT m a -> t m a
1292interleaving = wSerially
1294-- | Fix the type of a polymorphic stream as 'AheadT'.
1297aheadly :: IsStream t => AheadT m a -> t m a
1300-- | Fix the type of a polymorphic stream as 'AsyncT'.
1303asyncly :: IsStream t => AsyncT m a -> t m a
1306-- | Fix the type of a polymorphic stream as 'WAsyncT'.
1309wAsyncly :: IsStream t => WAsyncT m a -> t m a
1312-- | Fix the type of a polymorphic stream as 'ParallelT'.
1315parallely :: IsStream t => ParallelT m a -> t m a
1318-- | Fix the type of a polymorphic stream as 'ZipSerialM'.
1321zipSerially :: IsStream t => ZipSerialM m a -> t m a
1324-- | Same as 'zipSerially'.
1327{-# DEPRECATED zipping "Please use zipSerially instead." #-}
1328zipping :: IsStream t => ZipSerialM m a -> t m a
1329zipping = zipSerially
1331-- | Fix the type of a polymorphic stream as 'ZipAsyncM'.
1334zipAsyncly :: IsStream t => ZipAsyncM m a -> t m a
1337-- | Same as 'zipAsyncly'.
1340{-# DEPRECATED zippingAsync "Please use zipAsyncly instead." #-}
1341zippingAsync :: IsStream t => ZipAsyncM m a -> t m a
1342zippingAsync = zipAsyncly
1344-------------------------------------------------------------------------------
1345-- Running Streams, convenience functions specialized to types
1346-------------------------------------------------------------------------------
1348-- | Same as @runStream@.
1351{-# DEPRECATED runStreamT "Please use runStream instead." #-}
1352runStreamT :: Monad m => SerialT m a -> m ()
1353runStreamT = runStream
1355-- | Same as @runStream . wSerially@.
1358{-# DEPRECATED runInterleavedT "Please use 'runStream . interleaving' instead." #-}
1359runInterleavedT :: Monad m => InterleavedT m a -> m ()
1360runInterleavedT = runStream . wSerially
1362-- | Same as @runStream . asyncly@.
1365{-# DEPRECATED runAsyncT "Please use 'runStream . asyncly' instead." #-}
1366runAsyncT :: Monad m => AsyncT m a -> m ()
1367runAsyncT = runStream . asyncly
1369-- | Same as @runStream . parallely@.
1372{-# DEPRECATED runParallelT "Please use 'runStream . parallely' instead." #-}
1373runParallelT :: Monad m => ParallelT m a -> m ()
1374runParallelT = runStream . parallely
1376-- | Same as @runStream . zipping@.
1379{-# DEPRECATED runZipStream "Please use 'runStream . zipSerially instead." #-}
1380runZipStream :: Monad m => ZipSerialM m a -> m ()
1381runZipStream = runStream . zipSerially
1383-- | Same as @runStream . zippingAsync@.
1386{-# DEPRECATED runZipAsync "Please use 'runStream . zipAsyncly instead." #-}
1387runZipAsync :: Monad m => ZipAsyncM m a -> m ()
1388runZipAsync = runStream . zipAsyncly
1390------------------------------------------------------------------------------
1392------------------------------------------------------------------------------
1394-- | A serial IO stream of elements of type @a@. See 'SerialT' documentation
1398type Serial a = SerialT IO a
1400-- | An interleaving serial IO stream of elements of type @a@. See 'WSerialT'
1401-- documentation for more details.
1404type WSerial a = WSerialT IO a
1406-- | A serial IO stream of elements of type @a@ with concurrent lookahead. See
1407-- 'AheadT' documentation for more details.
1410type Ahead a = AheadT IO a
1412-- | A demand driven left biased parallely composing IO stream of elements of
1413-- type @a@. See 'AsyncT' documentation for more details.
1416type Async a = AsyncT IO a
1418-- | A round robin parallely composing IO stream of elements of type @a@.
1419-- See 'WAsyncT' documentation for more details.
1422type WAsync a = WAsyncT IO a
1424-- | A parallely composing IO stream of elements of type @a@.
1425-- See 'ParallelT' documentation for more details.
1428type Parallel a = ParallelT IO a
1430-- | An IO stream whose applicative instance zips streams serially.
1433type ZipSerial a = ZipSerialM IO a
1435-- | An IO stream whose applicative instance zips streams wAsyncly.
1438type ZipAsync a = ZipAsyncM IO a
1440------------------------------------------------------------------------------
1442------------------------------------------------------------------------------
1444-- | A variant of 'Data.Foldable.fold' that allows you to fold a 'Foldable'
1445-- container of streams using the specified stream sum operation.
1447-- @foldWith 'async' $ map return [1..3]@
1450{-# INLINABLE foldWith #-}
1451foldWith :: (IsStream t, Foldable f)
1452 => (t m a -> t m a -> t m a) -> f (t m a) -> t m a
1453foldWith f = foldr f nil
1455-- | A variant of 'foldMap' that allows you to map a monadic streaming action
1456-- on a 'Foldable' container and then fold it using the specified stream sum
1459-- @foldMapWith 'async' return [1..3]@
1462{-# INLINABLE foldMapWith #-}
1463foldMapWith :: (IsStream t, Foldable f)
1464 => (t m b -> t m b -> t m b) -> (a -> t m b) -> f a -> t m b
1465foldMapWith f g = foldr (f . g) nil
1467-- | Like 'foldMapWith' but with the last two arguments reversed i.e. the
1468-- monadic streaming function is the last argument.
1471{-# INLINABLE forEachWith #-}
1472forEachWith :: (IsStream t, Foldable f)
1473 => (t m b -> t m b -> t m b) -> f a -> (a -> t m b) -> t m b
1474forEachWith f xs g = foldr (f . g) nil xs
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