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
65 -- * Stream Transformers
67 , StreamT -- deprecated
69 , InterleavedT -- deprecated
74 , ZipStream -- deprecated
79 , serially -- deprecated
81 , interleaving -- deprecated
87 , zipping -- deprecated
89 , zippingAsync -- deprecated
93 , runStreamT -- deprecated
94 , runInterleavedT -- deprecated
95 , runAsyncT -- deprecated
96 , runParallelT -- deprecated
97 , runZipStream -- deprecated
98 , runZipAsync -- deprecated
107import Control.Monad (ap)
108import Control.Monad.Base (MonadBase (..), liftBaseDefault)
109import Control.Monad.Catch (MonadThrow, throwM)
110-- import Control.Monad.Error.Class (MonadError(..))
111import Control.Monad.IO.Class (MonadIO(..))
112import Control.Monad.Reader.Class (MonadReader(..))
113import Control.Monad.State.Class (MonadState(..))
114import Control.Monad.Trans.Class (MonadTrans (lift))
115import Data.Semigroup (Semigroup(..))
116import Streamly.Core ( MonadAsync , SVar,
118import qualified Streamly.Core as S
120------------------------------------------------------------------------------
121-- Types that can behave as a Stream
122------------------------------------------------------------------------------
127-- | Class of types that can represent a stream of elements of some type 'a' in
131class IsStream t where
132 toStream :: t m a -> S.Stream m a
133 fromStream :: S.Stream m a -> t m a
134 -- | Constructs a stream by adding a monadic action at the head of an existing
135 -- stream. For example:
138 -- > toList $ getLine \`consM` getLine \`consM` nil
145 consM :: MonadAsync m => m a -> t m a -> t m a
146 -- | Operator equivalent of 'consM'.
149 -- > toList $ getLine |: getLine |: nil
156 (|:) :: MonadAsync m => m a -> t m a -> t m a
158-- | Same as 'IsStream'.
161{-# DEPRECATED Streaming "Please use IsStream instead." #-}
162type Streaming = IsStream
164------------------------------------------------------------------------------
165-- Constructing a stream
166------------------------------------------------------------------------------
176nil :: IsStream t => t m a
177nil = fromStream S.nil
179-- | Constructs a stream by adding a monadic action at the head of an existing
180-- stream. For example:
183-- > toList $ getLine \`consM` getLine \`consM` nil
190consMSerial :: (IsStream t, Monad m) => m a -> t m a -> t m a
191consMSerial m r = fromStream $ S.consM m (toStream r)
195-- | Construct a stream by adding a pure value at the head of an existing
196-- stream. For pure values it can be faster than 'consM'. For example:
199-- > toList $ 1 \`cons` 2 \`cons` 3 \`cons` nil
204cons :: IsStream t => a -> t m a -> t m a
205cons a r = fromStream $ S.cons a (toStream r)
209-- | Operator equivalent of 'cons'.
212-- > toList $ 1 .: 2 .: 3 .: nil
217(.:) :: IsStream t => a -> t m a -> t m a
220-- | Build a stream from its church encoding. The function passed maps
221-- directly to the underlying representation of the stream type. The second
222-- parameter to the function is the "yield" function yielding a value and the
223-- remaining stream if any otherwise 'Nothing'. The third parameter is to
224-- represent an "empty" stream.
225streamBuild :: IsStream t
226 => (forall r. Maybe (SVar m a)
227 -> (a -> t m a -> m r)
232streamBuild k = fromStream $ S.Stream $ \svr stp sng yld ->
233 let yield a r = yld a (toStream r)
234 in k svr yield sng stp
236-- | Build a singleton stream from a callback function.
237fromCallback :: IsStream t => (forall r. (a -> m r) -> m r) -> t m a
238fromCallback k = fromStream $ S.Stream $ \_ _ sng _ -> k sng
240-- | Read an SVar to get a stream.
241fromSVar :: (MonadAsync m, IsStream t) => SVar m a -> t m a
242fromSVar sv = fromStream $ S.fromStreamVar sv
244------------------------------------------------------------------------------
245-- Destroying a stream
246------------------------------------------------------------------------------
248-- | Fold a stream using its church encoding. The second argument is the "step"
249-- function consuming an element and the remaining stream, if any. The third
250-- argument is for consuming an "empty" stream that yields nothing.
254 -> (a -> t m a -> m r)
259streamFold svr step single blank m =
260 let yield a x = step a (fromStream x)
261 in (S.runStream (toStream m)) svr blank single yield
263-- | Run a streaming composition, discard the results. By default it interprets
264-- the stream as 'SerialT', to run other types of streams use the type adapting
265-- combinators for example @runStream . 'asyncly'@.
268runStream :: Monad m => SerialT m a -> m ()
269runStream m = go (toStream m)
275 in (S.runStream m1) Nothing stop single yield
277-- | Same as 'runStream'
280{-# DEPRECATED runStreaming "Please use runStream instead." #-}
281runStreaming :: (Monad m, IsStream t) => t m a -> m ()
282runStreaming = runStream . adapt
284-- | Write a stream to an 'SVar' in a non-blocking manner. The stream can then
285-- be read back from the SVar using 'fromSVar'.
286toSVar :: (IsStream t, MonadAsync m) => SVar m a -> t m a -> m ()
287toSVar sv m = S.toStreamVar sv (toStream m)
289------------------------------------------------------------------------------
291------------------------------------------------------------------------------
293-- XXX Get rid of this?
294-- | Make a stream asynchronous, triggers the computation and returns a stream
295-- in the underlying monad representing the output generated by the original
296-- computation. The returned action is exhaustible and must be drained once. If
297-- not drained fully we may have a thread blocked forever and once exhausted it
298-- will always return 'empty'.
301mkAsync :: (IsStream t, MonadAsync m) => t m a -> m (t m a)
303 sv <- S.newStreamVar1 AsyncVar (toStream m)
306------------------------------------------------------------------------------
307-- CPP macros for common instances
308------------------------------------------------------------------------------
310-- XXX use template haskell instead and include Monoid and IsStream instances
313#define MONADPARALLEL , MonadAsync m
315#define MONAD_APPLICATIVE_INSTANCE(STREAM,CONSTRAINT) \
316instance (Monad m CONSTRAINT) => Applicative (STREAM m) where { \
317 pure = STREAM . S.singleton; \
320#define MONAD_COMMON_INSTANCES(STREAM,CONSTRAINT) \
321instance (MonadBase b m, Monad m CONSTRAINT) => MonadBase b (STREAM m) where {\
322 liftBase = liftBaseDefault }; \
324instance (MonadIO m CONSTRAINT) => MonadIO (STREAM m) where { \
325 liftIO = lift . liftIO }; \
327instance (MonadThrow m CONSTRAINT) => MonadThrow (STREAM m) where { \
328 throwM = lift . throwM }; \
331instance (MonadError e m CONSTRAINT) => MonadError e (STREAM m) where { \
332 throwError = lift . throwError; \
334 fromStream $ S.withCatchError (toStream m) (\e -> toStream $ h e) }; \
337instance (MonadReader r m CONSTRAINT) => MonadReader r (STREAM m) where { \
339 local f m = fromStream $ S.withLocal f (toStream m) }; \
341instance (MonadState s m CONSTRAINT) => MonadState s (STREAM m) where { \
343 put x = lift (put x); \
344 state k = lift (state k) }
346------------------------------------------------------------------------------
348------------------------------------------------------------------------------
350-- | Deep serial composition or serial composition with depth first traversal.
351-- The 'Semigroup' instance of 'SerialT' appends two streams serially in a
352-- depth first manner, yielding all elements from the first stream, and then
353-- all elements from the second stream.
357-- import qualified "Streamly.Prelude" as S
359-- main = ('toList' . 'serially' $ (fromFoldable [1,2]) \<\> (fromFoldable [3,4])) >>= print
365-- The 'Monad' instance runs the /monadic continuation/ for each
366-- element of the stream, serially.
369-- main = 'runStream' . 'serially' $ do
370-- x <- return 1 \<\> return 2
378-- 'SerialT' nests streams serially in a depth first manner.
381-- main = 'runStream' . 'serially' $ do
382-- x <- return 1 \<\> return 2
383-- y <- return 3 \<\> return 4
384-- S.once $ print (x, y)
393-- This behavior of 'SerialT' is exactly like a list transformer. We call the
394-- monadic code being run for each element of the stream a monadic
395-- continuation. In imperative paradigm we can think of this composition as
396-- nested @for@ loops and the monadic continuation is the body of the loop. The
397-- loop iterates for all elements of the stream.
399-- The 'serially' combinator can be omitted as the default stream type is
401-- Note that serial composition with depth first traversal can be used to
402-- combine an infinite number of streams as it explores only one stream at a
406newtype SerialT m a = SerialT {getSerialT :: S.Stream m a}
407 deriving (Semigroup, Monoid, Functor, MonadTrans)
411{-# DEPRECATED StreamT "Please use 'SerialT' instead." #-}
412type StreamT = SerialT
414instance IsStream SerialT where
415 toStream = getSerialT
419 {-# SPECIALIZE consM :: IO a -> SerialT IO a -> SerialT IO a #-}
420 consM :: Monad m => m a -> SerialT m a -> SerialT m a
424 {-# SPECIALIZE (|:) :: IO a -> SerialT IO a -> SerialT IO a #-}
425 (|:) :: Monad m => m a -> SerialT m a -> SerialT m a
428------------------------------------------------------------------------------
430------------------------------------------------------------------------------
432-- | Polymorphic version of the 'Semigroup' operation '<>' of 'SerialT'.
433-- Appends two streams sequentially, yielding all elements from the first
434-- stream, and then all elements from the second stream.
438serial :: IsStream t => t m a -> t m a -> t m a
439serial m1 m2 = fromStream $ S.serial (toStream m1) (toStream m2)
441------------------------------------------------------------------------------
443------------------------------------------------------------------------------
445instance Monad m => Monad (SerialT m) where
447 (SerialT (S.Stream m)) >>= f = SerialT $ S.Stream $ \_ stp sng yld ->
448 let run x = (S.runStream x) Nothing stp sng yld
449 single a = run $ toStream (f a)
450 yield a r = run $ toStream $ f a <> (fromStream r >>= f)
451 in m Nothing stp single yield
453------------------------------------------------------------------------------
455------------------------------------------------------------------------------
457MONAD_APPLICATIVE_INSTANCE(SerialT,)
458MONAD_COMMON_INSTANCES(SerialT,)
460------------------------------------------------------------------------------
462------------------------------------------------------------------------------
464-- | Wide serial composition or serial composition with a breadth first
465-- traversal. The 'Semigroup' instance of 'WSerialT' traverses
466-- the two streams in a breadth first manner. In other words, it interleaves
467-- two streams, yielding one element from each stream alternately.
471-- import qualified "Streamly.Prelude" as S
473-- main = ('toList' . 'wSerially' $ (fromFoldable [1,2]) \<\> (fromFoldable [3,4])) >>= print
479-- Similarly, the 'Monad' instance interleaves the iterations of the
480-- inner and the outer loop, nesting loops in a breadth first manner.
484-- main = 'runStream' . 'wSerially' $ do
485-- x <- return 1 \<\> return 2
486-- y <- return 3 \<\> return 4
487-- S.once $ print (x, y)
496-- Note that a serial composition with breadth first traversal can only combine
497-- a finite number of streams as it needs to retain state for each unfinished
501newtype WSerialT m a = WSerialT {getWSerialT :: S.Stream m a}
502 deriving (Functor, MonadTrans)
506{-# DEPRECATED InterleavedT "Please use 'WSerialT' instead." #-}
507type InterleavedT = WSerialT
509instance IsStream WSerialT where
510 toStream = getWSerialT
511 fromStream = WSerialT
514 {-# SPECIALIZE consM :: IO a -> WSerialT IO a -> WSerialT IO a #-}
515 consM :: Monad m => m a -> WSerialT m a -> WSerialT m a
519 {-# SPECIALIZE (|:) :: IO a -> WSerialT IO a -> WSerialT IO a #-}
520 (|:) :: Monad m => m a -> WSerialT m a -> WSerialT m a
523------------------------------------------------------------------------------
525------------------------------------------------------------------------------
527-- | Polymorphic version of the 'Semigroup' operation '<>' of 'WSerialT'.
528-- Interleaves two streams, yielding one element from each stream alternately.
531{-# INLINE wSerial #-}
532wSerial :: IsStream t => t m a -> t m a -> t m a
533wSerial m1 m2 = fromStream $ S.wSerial (toStream m1) (toStream m2)
535instance Semigroup (WSerialT m a) where
540-- | Same as 'wSerial'.
543{-# DEPRECATED (<=>) "Please use 'wSerial' instead." #-}
545(<=>) :: IsStream t => t m a -> t m a -> t m a
548------------------------------------------------------------------------------
550------------------------------------------------------------------------------
552instance Monoid (WSerialT m a) where
556------------------------------------------------------------------------------
558------------------------------------------------------------------------------
560instance Monad m => Monad (WSerialT m) where
562 (WSerialT (S.Stream m)) >>= f = WSerialT $ 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(WSerialT,)
573MONAD_COMMON_INSTANCES(WSerialT,)
575------------------------------------------------------------------------------
577------------------------------------------------------------------------------
579-- | Deep ahead composition or ahead composition with depth first traversal.
580-- The semigroup composition of 'AheadT' appends streams in a depth first
581-- manner just like 'SerialT' except that it can produce elements concurrently
582-- ahead of time. It is like 'AsyncT' except that 'AsyncT' produces the output
583-- as it arrives whereas 'AheadT' orders the output in the traversal order.
586-- main = ('toList' . 'aheadly' $ (fromFoldable [1,2]) \<> (fromFoldable [3,4])) >>= print
592-- Any exceptions generated by a constituent stream are propagated to the
595-- Similarly, the monad instance of 'AheadT' may run each iteration
596-- concurrently ahead of time but presents the results in the same order as
601-- import qualified "Streamly.Prelude" as S
602-- import Control.Concurrent
604-- main = 'runStream' . 'aheadly' $ do
605-- n <- return 3 \<\> return 2 \<\> return 1
607-- threadDelay (n * 1000000)
608-- myThreadId >>= \\tid -> putStrLn (show tid ++ ": Delay " ++ show n)
611-- ThreadId 40: Delay 1
612-- ThreadId 39: Delay 2
613-- ThreadId 38: Delay 3
616-- All iterations may run in the same thread if they do not block.
618-- Note that ahead composition with depth first traversal can be used to
619-- combine infinite number of streams as it explores only a bounded number of
623newtype AheadT m a = AheadT {getAheadT :: S.Stream m a}
624 deriving (Functor, MonadTrans)
626instance IsStream AheadT where
631 {-# SPECIALIZE consM :: IO a -> AheadT IO a -> AheadT IO a #-}
632 consM m r = fromStream $ S.consMAhead m (toStream r)
635 {-# SPECIALIZE (|:) :: IO a -> AheadT IO a -> AheadT IO a #-}
638------------------------------------------------------------------------------
640------------------------------------------------------------------------------
642-- | Polymorphic version of the 'Semigroup' operation '<>' of 'AheadT'.
643-- Merges two streams sequentially but with concurrent lookahead.
647ahead :: (IsStream t, MonadAsync m) => t m a -> t m a -> t m a
648ahead m1 m2 = fromStream $ S.ahead (toStream m1) (toStream m2)
650instance MonadAsync m => Semigroup (AheadT m a) where
653------------------------------------------------------------------------------
655------------------------------------------------------------------------------
657instance MonadAsync m => Monoid (AheadT m a) where
661------------------------------------------------------------------------------
663------------------------------------------------------------------------------
665{-# INLINE aheadbind #-}
669 -> (a -> S.Stream m b)
674 S.Stream $ \ctx stp sng yld ->
675 let run x = (S.runStream x) ctx stp sng yld
677 yield a r = run $ f a `S.ahead` go r
678 in g Nothing stp single yield
680instance MonadAsync m => Monad (AheadT m) where
682 (AheadT m) >>= f = AheadT $ aheadbind m (getAheadT . f)
684------------------------------------------------------------------------------
686------------------------------------------------------------------------------
688MONAD_APPLICATIVE_INSTANCE(AheadT,MONADPARALLEL)
689MONAD_COMMON_INSTANCES(AheadT, MONADPARALLEL)
691------------------------------------------------------------------------------
693------------------------------------------------------------------------------
695-- | Deep async composition or async composition with depth first traversal. In
696-- a left to right 'Semigroup' composition it tries to yield elements from the
697-- left stream as long as it can, but it can run the right stream in parallel
698-- if it needs to, based on demand. The right stream can be run if the left
699-- stream blocks on IO or cannot produce elements fast enough for the consumer.
702-- main = ('toList' . 'asyncly' $ (fromFoldable [1,2]) \<> (fromFoldable [3,4])) >>= print
708-- Any exceptions generated by a constituent stream are propagated to the
709-- output stream. The output and exceptions from a single stream are guaranteed
710-- to arrive in the same order in the resulting stream as they were generated
711-- in the input stream. However, the relative ordering of elements from
712-- different streams in the resulting stream can vary depending on scheduling
713-- and generation delays.
715-- Similarly, the monad instance of 'AsyncT' /may/ run each iteration
716-- concurrently based on demand. More concurrent iterations are started only
717-- if the previous iterations are not able to produce enough output for the
722-- import qualified "Streamly.Prelude" as S
723-- import Control.Concurrent
725-- main = 'runStream' . 'asyncly' $ do
726-- n <- return 3 \<\> return 2 \<\> return 1
728-- threadDelay (n * 1000000)
729-- myThreadId >>= \\tid -> putStrLn (show tid ++ ": Delay " ++ show n)
732-- ThreadId 40: Delay 1
733-- ThreadId 39: Delay 2
734-- ThreadId 38: Delay 3
737-- All iterations may run in the same thread if they do not block.
739-- Note that async composition with depth first traversal can be used to
740-- combine infinite number of streams as it explores only a bounded number of
744newtype AsyncT m a = AsyncT {getAsyncT :: S.Stream m a}
745 deriving (Functor, MonadTrans)
747instance IsStream AsyncT where
752 {-# SPECIALIZE consM :: IO a -> AsyncT IO a -> AsyncT IO a #-}
753 consM m r = fromStream $ S.consMAsync m (toStream r)
756 {-# SPECIALIZE (|:) :: IO a -> AsyncT IO a -> AsyncT IO a #-}
759------------------------------------------------------------------------------
761------------------------------------------------------------------------------
763-- | Polymorphic version of the 'Semigroup' operation '<>' of 'AsyncT'.
764-- Merges two streams possibly concurrently, preferring the
765-- elements from the left one when available.
769async :: (IsStream t, MonadAsync m) => t m a -> t m a -> t m a
770async m1 m2 = fromStream $ S.async (toStream m1) (toStream m2)
772instance MonadAsync m => Semigroup (AsyncT m a) where
778{-# DEPRECATED (<|) "Please use 'async' instead." #-}
780(<|) :: (IsStream t, MonadAsync m) => t m a -> t m a -> t m a
783------------------------------------------------------------------------------
785------------------------------------------------------------------------------
787instance MonadAsync m => Monoid (AsyncT m a) where
791------------------------------------------------------------------------------
793------------------------------------------------------------------------------
795{-# INLINE parbind #-}
797 :: (forall c. S.Stream m c -> S.Stream m c -> S.Stream m c)
799 -> (a -> S.Stream m b)
801parbind par m f = go m
804 S.Stream $ \ctx stp sng yld ->
805 let run x = (S.runStream x) ctx stp sng yld
807 yield a r = run $ f a `par` go r
808 in g Nothing stp single yield
810instance MonadAsync m => Monad (AsyncT m) where
812 (AsyncT m) >>= f = AsyncT $ parbind S.async m (getAsyncT . f)
814------------------------------------------------------------------------------
816------------------------------------------------------------------------------
818MONAD_APPLICATIVE_INSTANCE(AsyncT,MONADPARALLEL)
819MONAD_COMMON_INSTANCES(AsyncT, MONADPARALLEL)
821------------------------------------------------------------------------------
823------------------------------------------------------------------------------
825-- | Wide async composition or async composition with breadth first traversal.
826-- The Semigroup instance of 'WAsyncT' concurrently /traverses/ the composed
827-- streams using a depth first travesal or in a round robin fashion, yielding
828-- elements from both streams alternately.
831-- main = ('toList' . 'wAsyncly' $ (fromFoldable [1,2]) \<> (fromFoldable [3,4])) >>= print
837-- Any exceptions generated by a constituent stream are propagated to the
838-- output stream. The output and exceptions from a single stream are guaranteed
839-- to arrive in the same order in the resulting stream as they were generated
840-- in the input stream. However, the relative ordering of elements from
841-- different streams in the resulting stream can vary depending on scheduling
842-- and generation delays.
844-- Similarly, the 'Monad' instance of 'WAsyncT' runs /all/ iterations fairly
845-- concurrently using a round robin scheduling.
849-- import qualified "Streamly.Prelude" as S
850-- import Control.Concurrent
852-- main = 'runStream' . 'wAsyncly' $ do
853-- n <- return 3 \<\> return 2 \<\> return 1
855-- threadDelay (n * 1000000)
856-- myThreadId >>= \\tid -> putStrLn (show tid ++ ": Delay " ++ show n)
859-- ThreadId 40: Delay 1
860-- ThreadId 39: Delay 2
861-- ThreadId 38: Delay 3
864-- Unlike 'AsyncT' all iterations are guaranteed to run fairly
865-- concurrently, unconditionally.
867-- Note that async composition with breadth first traversal can only combine a
868-- finite number of streams as it needs to retain state for each unfinished
872newtype WAsyncT m a = WAsyncT {getWAsyncT :: S.Stream m a}
873 deriving (Functor, MonadTrans)
875instance IsStream WAsyncT where
876 toStream = getWAsyncT
880 {-# SPECIALIZE consM :: IO a -> WAsyncT IO a -> WAsyncT IO a #-}
881 consM m r = fromStream $ S.consMWAsync m (toStream r)
884 {-# SPECIALIZE (|:) :: IO a -> WAsyncT IO a -> WAsyncT IO a #-}
887------------------------------------------------------------------------------
889------------------------------------------------------------------------------
891-- | Polymorphic version of the 'Semigroup' operation '<>' of 'WAsyncT'.
892-- Merges two streams concurrently choosing elements from both fairly.
896wAsync :: (IsStream t, MonadAsync m) => t m a -> t m a -> t m a
897wAsync m1 m2 = fromStream $ S.wAsync (toStream m1) (toStream m2)
899instance MonadAsync m => Semigroup (WAsyncT m a) where
902------------------------------------------------------------------------------
904------------------------------------------------------------------------------
906instance MonadAsync m => Monoid (WAsyncT m a) where
910------------------------------------------------------------------------------
912------------------------------------------------------------------------------
914instance MonadAsync m => Monad (WAsyncT m) where
917 WAsyncT $ parbind S.wAsync m (getWAsyncT . f)
919------------------------------------------------------------------------------
921------------------------------------------------------------------------------
923MONAD_APPLICATIVE_INSTANCE(WAsyncT,MONADPARALLEL)
924MONAD_COMMON_INSTANCES(WAsyncT, MONADPARALLEL)
926------------------------------------------------------------------------------
928------------------------------------------------------------------------------
930-- | Async composition with simultaneous traversal of all streams.
932-- The Semigroup instance of 'ParallelT' concurrently /merges/ two streams,
933-- running both strictly concurrently and yielding elements from both streams
934-- as they arrive. When multiple streams are combined using 'ParallelT' each
935-- one is evaluated in its own thread and the results produced are presented in
936-- the combined stream on a first come first serve basis.
938-- 'AsyncT' and 'WAsyncT' are /concurrent lookahead streams/ each with a
939-- specific type of consumption pattern (depth first or breadth first). Since
940-- they are lookahead, they may introduce certain default latency in starting
941-- more concurrent tasks for efficiency reasons or may put a default limitation
942-- on the resource consumption (e.g. number of concurrent threads for
943-- lookahead). If we look at the implementation detail, they both can share a
944-- pool of worker threads to evaluate the streams in the desired pattern and at
945-- the desired rate. However, 'ParallelT' uses a separate runtime thread to
946-- evaluate each stream.
948-- 'WAsyncT' is similar to 'ParallelT', as both of them evaluate the
949-- constituent streams fairly in a round robin fashion.
950-- However, the key difference is that 'WAsyncT' is lazy or pull driven
951-- whereas 'ParallelT' is strict or push driven. 'ParallelT' immediately
952-- starts concurrent evaluation of both the streams (in separate threads) and
953-- later picks the results whereas 'WAsyncT' may wait for a certain latency
954-- threshold before initiating concurrent evaluation of the next stream. The
955-- concurrent scheduling of the next stream or the degree of concurrency is
956-- driven by the feedback from the consumer. In case of 'ParallelT' each stream
957-- is evaluated in a separate thread and results are /pushed/ to a shared
958-- output buffer, the evaluation rate is controlled by blocking when the buffer
961-- Concurrent lookahead streams are generally more efficient than
962-- 'ParallelT' and can work pretty efficiently even for smaller tasks because
963-- they do not necessarily use a separate thread for each task. So they should
964-- be preferred over 'ParallelT' especially when efficiency is a concern and
965-- simultaneous strict evaluation is not a requirement. 'ParallelT' is useful
966-- for cases when the streams are required to be evaluated simultaneously
967-- irrespective of how the consumer consumes them e.g. when we want to race
968-- two tasks and want to start both strictly at the same time or if we have
969-- timers in the parallel tasks and our results depend on the timers being
970-- started at the same time. We can say that 'ParallelT' is almost the same
971-- (modulo some implementation differences) as 'WAsyncT' when the latter is
972-- used with unlimited lookahead and zero latency in initiating lookahead.
975-- main = ('toList' . 'parallely' $ (fromFoldable [1,2]) \<> (fromFoldable [3,4])) >>= print
981-- When streams with more than one element are merged, it yields whichever
982-- stream yields first without any bias, unlike the 'Async' style streams.
984-- Any exceptions generated by a constituent stream are propagated to the
985-- output stream. The output and exceptions from a single stream are guaranteed
986-- to arrive in the same order in the resulting stream as they were generated
987-- in the input stream. However, the relative ordering of elements from
988-- different streams in the resulting stream can vary depending on scheduling
989-- and generation delays.
991-- Similarly, the 'Monad' instance of 'ParallelT' runs /all/ iterations
992-- of the loop concurrently.
996-- import qualified "Streamly.Prelude" as S
997-- import Control.Concurrent
999-- main = 'runStream' . 'parallely' $ do
1000-- n <- return 3 \<\> return 2 \<\> return 1
1002-- threadDelay (n * 1000000)
1003-- myThreadId >>= \\tid -> putStrLn (show tid ++ ": Delay " ++ show n)
1006-- ThreadId 40: Delay 1
1007-- ThreadId 39: Delay 2
1008-- ThreadId 38: Delay 3
1011-- Note that parallel composition can only combine a finite number of
1012-- streams as it needs to retain state for each unfinished stream.
1015newtype ParallelT m a = ParallelT {getParallelT :: S.Stream m a}
1016 deriving (Functor, MonadTrans)
1018instance IsStream ParallelT where
1019 toStream = getParallelT
1020 fromStream = ParallelT
1022 {-# INLINE consM #-}
1023 {-# SPECIALIZE consM :: IO a -> ParallelT IO a -> ParallelT IO a #-}
1024 consM m r = fromStream $ S.consMParallel m (toStream r)
1027 {-# SPECIALIZE (|:) :: IO a -> ParallelT IO a -> ParallelT IO a #-}
1030------------------------------------------------------------------------------
1032------------------------------------------------------------------------------
1034-- | Polymorphic version of the 'Semigroup' operation '<>' of 'ParallelT'
1035-- Merges two streams concurrently.
1038{-# INLINE parallel #-}
1039parallel :: (IsStream t, MonadAsync m) => t m a -> t m a -> t m a
1040parallel m1 m2 = fromStream $ S.parallel (toStream m1) (toStream m2)
1042instance MonadAsync m => Semigroup (ParallelT m a) where
1045------------------------------------------------------------------------------
1047------------------------------------------------------------------------------
1049instance MonadAsync m => Monoid (ParallelT m a) where
1053------------------------------------------------------------------------------
1055------------------------------------------------------------------------------
1057instance MonadAsync m => Monad (ParallelT m) where
1059 (ParallelT m) >>= f = ParallelT $ parbind S.parallel m (getParallelT . f)
1061------------------------------------------------------------------------------
1063------------------------------------------------------------------------------
1065MONAD_APPLICATIVE_INSTANCE(ParallelT,MONADPARALLEL)
1066MONAD_COMMON_INSTANCES(ParallelT, MONADPARALLEL)
1068------------------------------------------------------------------------------
1069-- Serially Zipping Streams
1070------------------------------------------------------------------------------
1072-- | The applicative instance of 'ZipSerialM' zips a number of streams serially
1073-- i.e. it produces one element from each stream serially and then zips all
1077-- main = (toList . 'zipSerially' $ (,,) \<$\> s1 \<*\> s2 \<*\> s3) >>= print
1078-- where s1 = fromFoldable [1, 2]
1079-- s2 = fromFoldable [3, 4]
1080-- s3 = fromFoldable [5, 6]
1086-- The 'Semigroup' instance of this type works the same way as that of
1090newtype ZipSerialM m a = ZipSerialM {getZipSerialM :: S.Stream m a}
1091 deriving (Functor, Semigroup, Monoid)
1095{-# DEPRECATED ZipStream "Please use 'ZipSerialM' instead." #-}
1096type ZipStream = ZipSerialM
1098instance IsStream ZipSerialM where
1099 toStream = getZipSerialM
1100 fromStream = ZipSerialM
1102 {-# INLINE consM #-}
1103 {-# SPECIALIZE consM :: IO a -> ZipSerialM IO a -> ZipSerialM IO a #-}
1104 consM :: Monad m => m a -> ZipSerialM m a -> ZipSerialM m a
1108 {-# SPECIALIZE (|:) :: IO a -> ZipSerialM IO a -> ZipSerialM IO a #-}
1109 (|:) :: Monad m => m a -> ZipSerialM m a -> ZipSerialM m a
1112instance Monad m => Applicative (ZipSerialM m) where
1113 pure = ZipSerialM . S.repeat
1114 m1 <*> m2 = fromStream $ S.zipWith id (toStream m1) (toStream m2)
1116------------------------------------------------------------------------------
1117-- Parallely Zipping Streams
1118------------------------------------------------------------------------------
1120-- | Like 'ZipSerialM' but zips in parallel, it generates all the elements to
1121-- be zipped concurrently.
1124-- main = (toList . 'zipAsyncly' $ (,,) \<$\> s1 \<*\> s2 \<*\> s3) >>= print
1125-- where s1 = fromFoldable [1, 2]
1126-- s2 = fromFoldable [3, 4]
1127-- s3 = fromFoldable [5, 6]
1133-- The 'Semigroup' instance of this type works the same way as that of
1137newtype ZipAsyncM m a = ZipAsyncM {getZipAsyncM :: S.Stream m a}
1138 deriving (Functor, Semigroup, Monoid)
1140instance IsStream ZipAsyncM where
1141 toStream = getZipAsyncM
1142 fromStream = ZipAsyncM
1144 {-# INLINE consM #-}
1145 {-# SPECIALIZE consM :: IO a -> ZipAsyncM IO a -> ZipAsyncM IO a #-}
1146 consM :: Monad m => m a -> ZipAsyncM m a -> ZipAsyncM m a
1150 {-# SPECIALIZE (|:) :: IO a -> ZipAsyncM IO a -> ZipAsyncM IO a #-}
1151 (|:) :: Monad m => m a -> ZipAsyncM m a -> ZipAsyncM m a
1154instance MonadAsync m => Applicative (ZipAsyncM m) where
1155 pure = ZipAsyncM . S.repeat
1156 m1 <*> m2 = fromStream $ S.zipAsyncWith id (toStream m1) (toStream m2)
1158-------------------------------------------------------------------------------
1159-- Type adapting combinators
1160-------------------------------------------------------------------------------
1162-- | Adapt any specific stream type to any other specific stream type.
1165adapt :: (IsStream t1, IsStream t2) => t1 m a -> t2 m a
1166adapt = fromStream . toStream
1168-- | Fix the type of a polymorphic stream as 'SerialT'.
1171serially :: IsStream t => SerialT m a -> t m a
1174-- | Fix the type of a polymorphic stream as 'WSerialT'.
1177wSerially :: IsStream t => WSerialT m a -> t m a
1180-- | Same as 'wSerially'.
1183{-# DEPRECATED interleaving "Please use wSerially instead." #-}
1184interleaving :: IsStream t => WSerialT m a -> t m a
1185interleaving = wSerially
1187-- | Fix the type of a polymorphic stream as 'AheadT'.
1190aheadly :: IsStream t => AheadT m a -> t m a
1193-- | Fix the type of a polymorphic stream as 'AsyncT'.
1196asyncly :: IsStream t => AsyncT m a -> t m a
1199-- | Fix the type of a polymorphic stream as 'WAsyncT'.
1202wAsyncly :: IsStream t => WAsyncT m a -> t m a
1205-- | Fix the type of a polymorphic stream as 'ParallelT'.
1208parallely :: IsStream t => ParallelT m a -> t m a
1211-- | Fix the type of a polymorphic stream as 'ZipSerialM'.
1214zipSerially :: IsStream t => ZipSerialM m a -> t m a
1217-- | Same as 'zipSerially'.
1220{-# DEPRECATED zipping "Please use zipSerially instead." #-}
1221zipping :: IsStream t => ZipSerialM m a -> t m a
1222zipping = zipSerially
1224-- | Fix the type of a polymorphic stream as 'ZipAsyncM'.
1227zipAsyncly :: IsStream t => ZipAsyncM m a -> t m a
1230-- | Same as 'zipAsyncly'.
1233{-# DEPRECATED zippingAsync "Please use zipAsyncly instead." #-}
1234zippingAsync :: IsStream t => ZipAsyncM m a -> t m a
1235zippingAsync = zipAsyncly
1237-------------------------------------------------------------------------------
1238-- Running Streams, convenience functions specialized to types
1239-------------------------------------------------------------------------------
1241-- | Same as @runStream@.
1244{-# DEPRECATED runStreamT "Please use runStream instead." #-}
1245runStreamT :: Monad m => SerialT m a -> m ()
1246runStreamT = runStream
1248-- | Same as @runStream . wSerially@.
1251{-# DEPRECATED runInterleavedT "Please use 'runStream . interleaving' instead." #-}
1252runInterleavedT :: Monad m => InterleavedT m a -> m ()
1253runInterleavedT = runStream . wSerially
1255-- | Same as @runStream . asyncly@.
1258{-# DEPRECATED runAsyncT "Please use 'runStream . asyncly' instead." #-}
1259runAsyncT :: Monad m => AsyncT m a -> m ()
1260runAsyncT = runStream . asyncly
1262-- | Same as @runStream . parallely@.
1265{-# DEPRECATED runParallelT "Please use 'runStream . parallely' instead." #-}
1266runParallelT :: Monad m => ParallelT m a -> m ()
1267runParallelT = runStream . parallely
1269-- | Same as @runStream . zipping@.
1272{-# DEPRECATED runZipStream "Please use 'runStream . zipSerially instead." #-}
1273runZipStream :: Monad m => ZipSerialM m a -> m ()
1274runZipStream = runStream . zipSerially
1276-- | Same as @runStream . zippingAsync@.
1279{-# DEPRECATED runZipAsync "Please use 'runStream . zipAsyncly instead." #-}
1280runZipAsync :: Monad m => ZipAsyncM m a -> m ()
1281runZipAsync = runStream . zipAsyncly
1283------------------------------------------------------------------------------
1285------------------------------------------------------------------------------
1287-- | A serial IO stream of elements of type @a@. See 'SerialT' documentation
1291type Serial a = SerialT IO a
1293-- | An interleaving serial IO stream of elements of type @a@. See 'WSerialT'
1294-- documentation for more details.
1297type WSerial a = WSerialT IO a
1299-- | A serial IO stream of elements of type @a@ with concurrent lookahead. See
1300-- 'AheadT' documentation for more details.
1303type Ahead a = AheadT IO a
1305-- | A demand driven left biased parallely composing IO stream of elements of
1306-- type @a@. See 'AsyncT' documentation for more details.
1309type Async a = AsyncT IO a
1311-- | A round robin parallely composing IO stream of elements of type @a@.
1312-- See 'WAsyncT' documentation for more details.
1315type WAsync a = WAsyncT IO a
1317-- | A parallely composing IO stream of elements of type @a@.
1318-- See 'ParallelT' documentation for more details.
1321type Parallel a = ParallelT IO a
1323-- | An IO stream whose applicative instance zips streams serially.
1326type ZipSerial a = ZipSerialM IO a
1328-- | An IO stream whose applicative instance zips streams wAsyncly.
1331type ZipAsync a = ZipAsyncM IO a
1333------------------------------------------------------------------------------
1335------------------------------------------------------------------------------
1337-- | A variant of 'Data.Foldable.fold' that allows you to fold a 'Foldable'
1338-- container of streams using the specified stream sum operation.
1340-- @foldWith 'async' $ map return [1..3]@
1343{-# INLINABLE foldWith #-}
1344foldWith :: (IsStream t, Foldable f)
1345 => (t m a -> t m a -> t m a) -> f (t m a) -> t m a
1346foldWith f = foldr f nil
1348-- | A variant of 'foldMap' that allows you to map a monadic streaming action
1349-- on a 'Foldable' container and then fold it using the specified stream sum
1352-- @foldMapWith 'async' return [1..3]@
1355{-# INLINABLE foldMapWith #-}
1356foldMapWith :: (IsStream t, Foldable f)
1357 => (t m b -> t m b -> t m b) -> (a -> t m b) -> f a -> t m b
1358foldMapWith f g = foldr (f . g) nil
1360-- | Like 'foldMapWith' but with the last two arguments reversed i.e. the
1361-- monadic streaming function is the last argument.
1364{-# INLINABLE forEachWith #-}
1365forEachWith :: (IsStream t, Foldable f)
1366 => (t m b -> t m b -> t m b) -> f a -> (a -> t m b) -> t m b
1367forEachWith 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