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 existing
138 -- 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'.
154 -- > toList $ getLine |: getLine |: nil
161 -- let delay = threadDelay 1000000 >> print 1
162 -- runStream $ serially $ delay |: delay |: delay |: nil
163 -- runStream $ parallely $ delay |: delay |: delay |: nil
166 -- /Concurrent (do not use 'parallely' to construct infinite streams)/
169 (|:) :: MonadAsync m => m a -> t m a -> t m a
170 -- We can define (|:) just as 'consM' but it is defined explicitly for each
171 -- type because we want to use SPECIALIZE pragma on the definition.
173-- | Same as 'IsStream'.
176{-# DEPRECATED Streaming "Please use IsStream instead." #-}
177type Streaming = IsStream
179------------------------------------------------------------------------------
180-- Constructing a stream
181------------------------------------------------------------------------------
191nil :: IsStream t => t m a
192nil = fromStream S.nil
194-- | Constructs a stream by adding a monadic action at the head of an existing
195-- stream. For example:
198-- > toList $ getLine \`consM` getLine \`consM` nil
205consMSerial :: (IsStream t, Monad m) => m a -> t m a -> t m a
206consMSerial m r = fromStream $ S.consM m (toStream r)
210-- | Construct a stream by adding a pure value at the head of an existing
211-- stream. For pure values it can be faster than 'consM'. For example:
214-- > toList $ 1 \`cons` 2 \`cons` 3 \`cons` nil
219cons :: IsStream t => a -> t m a -> t m a
220cons a r = fromStream $ S.cons a (toStream r)
224-- | Operator equivalent of 'cons'.
227-- > toList $ 1 .: 2 .: 3 .: nil
232(.:) :: IsStream t => a -> t m a -> t m a
235-- | Build a stream from its church encoding. The function passed maps
236-- directly to the underlying representation of the stream type. The second
237-- parameter to the function is the "yield" function yielding a value and the
238-- remaining stream if any otherwise 'Nothing'. The third parameter is to
239-- represent an "empty" stream.
240streamBuild :: IsStream t
241 => (forall r. Maybe (SVar m a)
242 -> (a -> t m a -> m r)
247streamBuild k = fromStream $ S.Stream $ \svr stp sng yld ->
248 let yield a r = yld a (toStream r)
249 in k svr yield sng stp
251-- | Build a singleton stream from a callback function.
252fromCallback :: IsStream t => (forall r. (a -> m r) -> m r) -> t m a
253fromCallback k = fromStream $ S.Stream $ \_ _ sng _ -> k sng
255-- | Read an SVar to get a stream.
256fromSVar :: (MonadAsync m, IsStream t) => SVar m a -> t m a
257fromSVar sv = fromStream $ S.fromStreamVar sv
259------------------------------------------------------------------------------
260-- Destroying a stream
261------------------------------------------------------------------------------
263-- | Fold a stream using its church encoding. The second argument is the "step"
264-- function consuming an element and the remaining stream, if any. The third
265-- argument is for consuming an "empty" stream that yields nothing.
269 -> (a -> t m a -> m r)
274streamFold svr step single blank m =
275 let yield a x = step a (fromStream x)
276 in (S.runStream (toStream m)) svr blank single yield
278-- | Run a streaming composition, discard the results. By default it interprets
279-- the stream as 'SerialT', to run other types of streams use the type adapting
280-- combinators for example @runStream . 'asyncly'@.
283runStream :: Monad m => SerialT m a -> m ()
284runStream m = go (toStream m)
290 in (S.runStream m1) Nothing stop single yield
292-- | Same as 'runStream'
295{-# DEPRECATED runStreaming "Please use runStream instead." #-}
296runStreaming :: (Monad m, IsStream t) => t m a -> m ()
297runStreaming = runStream . adapt
299-- | Write a stream to an 'SVar' in a non-blocking manner. The stream can then
300-- be read back from the SVar using 'fromSVar'.
301toSVar :: (IsStream t, MonadAsync m) => SVar m a -> t m a -> m ()
302toSVar sv m = S.toStreamVar sv (toStream m)
304------------------------------------------------------------------------------
306------------------------------------------------------------------------------
308-- XXX Get rid of this?
309-- | Make a stream asynchronous, triggers the computation and returns a stream
310-- in the underlying monad representing the output generated by the original
311-- computation. The returned action is exhaustible and must be drained once. If
312-- not drained fully we may have a thread blocked forever and once exhausted it
313-- will always return 'empty'.
316mkAsync :: (IsStream t, MonadAsync m) => t m a -> m (t m a)
318 sv <- S.newStreamVar1 AsyncVar (toStream m)
321{-# INLINE applyWith #-}
322applyWith :: (IsStream t, MonadAsync m)
323 => SVarStyle -> (t m a -> t m b) -> t m a -> t m b
324applyWith style f x = fromStream $
325 S.applyWith style (toStream . f . fromStream) (toStream x)
327{-# INLINE runWith #-}
328runWith :: (IsStream t, MonadAsync m)
329 => SVarStyle -> (t m a -> m b) -> t m a -> m b
330runWith style f x = S.runWith style (f . fromStream) (toStream x)
332-- | Parallel function application operator for streams; just like the regular
333-- function application operator '$' except that it is concurrent. The
334-- following code prints a value every second even though each stage adds a 1
340-- S.mapM (\\x -> threadDelay 1000000 >> print x)
341-- |$ S.repeatM (threadDelay 1000000 >> return 1)
348(|$) :: (IsStream t, MonadAsync m) => (t m a -> t m b) -> t m a -> t m b
349f |$ x = applyWith ParallelVar f x
351-- | Parallel reverse function application operator for streams; just like the
352-- regular reverse function application operator '&' except that it is
357-- S.repeatM (threadDelay 1000000 >> return 1)
358-- |& S.mapM (\\x -> threadDelay 1000000 >> print x)
365(|&) :: (IsStream t, MonadAsync m) => t m a -> (t m a -> t m b) -> t m b
368-- | Parallel function application operator; applies a @run@ or @fold@ function
369-- to a stream such that the fold consumer and the stream producer run in
370-- parallel. A @run@ or @fold@ function reduces the stream to a value in the
374-- S.foldlM' (\\_ a -> threadDelay 1000000 >> print a) ()
375-- <| S.repeatM (threadDelay 1000000 >> return 1)
382(<|) :: (IsStream t, MonadAsync m) => (t m a -> m b) -> t m a -> m b
383f <| x = runWith ParallelVar f x
385-- | Parallel reverse function application operator for applying a run or fold
386-- functions to a stream. Just like '<|' except that the operands are reversed.
389-- S.repeatM (threadDelay 1000000 >> return 1)
390-- |> S.foldlM' (\\_ a -> threadDelay 1000000 >> print a) ()
397(|>) :: (IsStream t, MonadAsync m) => t m a -> (t m a -> m b) -> m b
400------------------------------------------------------------------------------
401-- CPP macros for common instances
402------------------------------------------------------------------------------
404-- XXX use template haskell instead and include Monoid and IsStream instances
407withLocal :: MonadReader r m => (r -> r) -> S.Stream m a -> S.Stream m a
409 S.Stream $ \svr stp sng yld ->
410 let single = local f . sng
411 yield a r = local f $ yld a (withLocal f r)
412 in (S.runStream m) svr (local f stp) single yield
415-- XXX handle and test cross thread state transfer
418 => S.Stream m a -> (e -> S.Stream m a) -> S.Stream m a
420 S.Stream $ \svr stp sng yld ->
421 let run x = S.runStream x svr stp sng yield
422 handle r = r `catchError` \e -> run $ h e
423 yield a r = yld a (withCatchError r h)
427#define MONADPARALLEL , MonadAsync m
429#define MONAD_APPLICATIVE_INSTANCE(STREAM,CONSTRAINT) \
430instance (Monad m CONSTRAINT) => Applicative (STREAM m) where { \
431 pure = STREAM . S.singleton; \
434#define MONAD_COMMON_INSTANCES(STREAM,CONSTRAINT) \
435instance (MonadBase b m, Monad m CONSTRAINT) => MonadBase b (STREAM m) where {\
436 liftBase = liftBaseDefault }; \
438instance (MonadIO m CONSTRAINT) => MonadIO (STREAM m) where { \
439 liftIO = lift . liftIO }; \
441instance (MonadThrow m CONSTRAINT) => MonadThrow (STREAM m) where { \
442 throwM = lift . throwM }; \
445instance (MonadError e m CONSTRAINT) => MonadError e (STREAM m) where { \
446 throwError = lift . throwError; \
448 fromStream $ withCatchError (toStream m) (\e -> toStream $ h e) }; \
451instance (MonadReader r m CONSTRAINT) => MonadReader r (STREAM m) where { \
453 local f m = fromStream $ withLocal f (toStream m) }; \
455instance (MonadState s m CONSTRAINT) => MonadState s (STREAM m) where { \
457 put x = lift (put x); \
458 state k = lift (state k) }
460------------------------------------------------------------------------------
462------------------------------------------------------------------------------
464-- | Deep serial composition or serial composition with depth first traversal.
465-- The 'Semigroup' instance of 'SerialT' appends two streams serially in a
466-- depth first manner, yielding all elements from the first stream, and then
467-- all elements from the second stream.
471-- import qualified "Streamly.Prelude" as S
473-- main = ('toList' . 'serially' $ (fromFoldable [1,2]) \<\> (fromFoldable [3,4])) >>= print
479-- The 'Monad' instance runs the /monadic continuation/ for each
480-- element of the stream, serially.
483-- main = 'runStream' . 'serially' $ do
484-- x <- return 1 \<\> return 2
492-- 'SerialT' nests streams serially in a depth first manner.
495-- main = 'runStream' . 'serially' $ do
496-- x <- return 1 \<\> return 2
497-- y <- return 3 \<\> return 4
498-- S.once $ print (x, y)
507-- This behavior of 'SerialT' is exactly like a list transformer. We call the
508-- monadic code being run for each element of the stream a monadic
509-- continuation. In imperative paradigm we can think of this composition as
510-- nested @for@ loops and the monadic continuation is the body of the loop. The
511-- loop iterates for all elements of the stream.
513-- The 'serially' combinator can be omitted as the default stream type is
515-- Note that serial composition with depth first traversal can be used to
516-- combine an infinite number of streams as it explores only one stream at a
520newtype SerialT m a = SerialT {getSerialT :: S.Stream m a}
521 deriving (Semigroup, Monoid, Functor, MonadTrans)
525{-# DEPRECATED StreamT "Please use 'SerialT' instead." #-}
526type StreamT = SerialT
528instance IsStream SerialT where
529 toStream = getSerialT
533 {-# SPECIALIZE consM :: IO a -> SerialT IO a -> SerialT IO a #-}
534 consM :: Monad m => m a -> SerialT m a -> SerialT m a
538 {-# SPECIALIZE (|:) :: IO a -> SerialT IO a -> SerialT IO a #-}
539 (|:) :: Monad m => m a -> SerialT m a -> SerialT m a
542------------------------------------------------------------------------------
544------------------------------------------------------------------------------
546-- | Polymorphic version of the 'Semigroup' operation '<>' of 'SerialT'.
547-- Appends two streams sequentially, yielding all elements from the first
548-- stream, and then all elements from the second stream.
552serial :: IsStream t => t m a -> t m a -> t m a
553serial m1 m2 = fromStream $ S.serial (toStream m1) (toStream m2)
555------------------------------------------------------------------------------
557------------------------------------------------------------------------------
559instance Monad m => Monad (SerialT m) where
561 (SerialT (S.Stream m)) >>= f = SerialT $ S.Stream $ \_ stp sng yld ->
562 let run x = (S.runStream x) Nothing stp sng yld
563 single a = run $ toStream (f a)
564 yield a r = run $ toStream $ f a <> (fromStream r >>= f)
565 in m Nothing stp single yield
567------------------------------------------------------------------------------
569------------------------------------------------------------------------------
571MONAD_APPLICATIVE_INSTANCE(SerialT,)
572MONAD_COMMON_INSTANCES(SerialT,)
574------------------------------------------------------------------------------
576------------------------------------------------------------------------------
578-- | Wide serial composition or serial composition with a breadth first
579-- traversal. The 'Semigroup' instance of 'WSerialT' traverses
580-- the two streams in a breadth first manner. In other words, it interleaves
581-- two streams, yielding one element from each stream alternately.
585-- import qualified "Streamly.Prelude" as S
587-- main = ('toList' . 'wSerially' $ (fromFoldable [1,2]) \<\> (fromFoldable [3,4])) >>= print
593-- Similarly, the 'Monad' instance interleaves the iterations of the
594-- inner and the outer loop, nesting loops in a breadth first manner.
598-- main = 'runStream' . 'wSerially' $ do
599-- x <- return 1 \<\> return 2
600-- y <- return 3 \<\> return 4
601-- S.once $ print (x, y)
610-- Note that a serial composition with breadth first traversal can only combine
611-- a finite number of streams as it needs to retain state for each unfinished
615newtype WSerialT m a = WSerialT {getWSerialT :: S.Stream m a}
616 deriving (Functor, MonadTrans)
620{-# DEPRECATED InterleavedT "Please use 'WSerialT' instead." #-}
621type InterleavedT = WSerialT
623instance IsStream WSerialT where
624 toStream = getWSerialT
625 fromStream = WSerialT
628 {-# SPECIALIZE consM :: IO a -> WSerialT IO a -> WSerialT IO a #-}
629 consM :: Monad m => m a -> WSerialT m a -> WSerialT m a
633 {-# SPECIALIZE (|:) :: IO a -> WSerialT IO a -> WSerialT IO a #-}
634 (|:) :: Monad m => m a -> WSerialT m a -> WSerialT m a
637------------------------------------------------------------------------------
639------------------------------------------------------------------------------
641-- | Polymorphic version of the 'Semigroup' operation '<>' of 'WSerialT'.
642-- Interleaves two streams, yielding one element from each stream alternately.
645{-# INLINE wSerial #-}
646wSerial :: IsStream t => t m a -> t m a -> t m a
647wSerial m1 m2 = fromStream $ S.wSerial (toStream m1) (toStream m2)
649instance Semigroup (WSerialT m a) where
654-- | Same as 'wSerial'.
657{-# DEPRECATED (<=>) "Please use 'wSerial' instead." #-}
659(<=>) :: IsStream t => t m a -> t m a -> t m a
662------------------------------------------------------------------------------
664------------------------------------------------------------------------------
666instance Monoid (WSerialT m a) where
670------------------------------------------------------------------------------
672------------------------------------------------------------------------------
674instance Monad m => Monad (WSerialT m) where
676 (WSerialT (S.Stream m)) >>= f = WSerialT $ S.Stream $ \_ stp sng yld ->
677 let run x = (S.runStream x) Nothing stp sng yld
678 single a = run $ toStream (f a)
679 yield a r = run $ toStream $ f a <> (fromStream r >>= f)
680 in m Nothing stp single yield
682------------------------------------------------------------------------------
684------------------------------------------------------------------------------
686MONAD_APPLICATIVE_INSTANCE(WSerialT,)
687MONAD_COMMON_INSTANCES(WSerialT,)
689------------------------------------------------------------------------------
691------------------------------------------------------------------------------
693-- | Deep ahead composition or ahead composition with depth first traversal.
694-- The semigroup composition of 'AheadT' appends streams in a depth first
695-- manner just like 'SerialT' except that it can produce elements concurrently
696-- ahead of time. It is like 'AsyncT' except that 'AsyncT' produces the output
697-- as it arrives whereas 'AheadT' orders the output in the traversal order.
700-- main = ('toList' . 'aheadly' $ (fromFoldable [1,2]) \<> (fromFoldable [3,4])) >>= print
706-- Any exceptions generated by a constituent stream are propagated to the
709-- Similarly, the monad instance of 'AheadT' may run each iteration
710-- concurrently ahead of time but presents the results in the same order as
715-- import qualified "Streamly.Prelude" as S
716-- import Control.Concurrent
718-- main = 'runStream' . 'aheadly' $ do
719-- n <- return 3 \<\> return 2 \<\> return 1
721-- threadDelay (n * 1000000)
722-- myThreadId >>= \\tid -> putStrLn (show tid ++ ": Delay " ++ show n)
725-- ThreadId 40: Delay 1
726-- ThreadId 39: Delay 2
727-- ThreadId 38: Delay 3
730-- All iterations may run in the same thread if they do not block.
732-- Note that ahead composition with depth first traversal can be used to
733-- combine infinite number of streams as it explores only a bounded number of
737newtype AheadT m a = AheadT {getAheadT :: S.Stream m a}
738 deriving (Functor, MonadTrans)
740instance IsStream AheadT where
745 {-# SPECIALIZE consM :: IO a -> AheadT IO a -> AheadT IO a #-}
746 consM m r = fromStream $ S.consMAhead m (toStream r)
749 {-# SPECIALIZE (|:) :: IO a -> AheadT IO a -> AheadT IO a #-}
752------------------------------------------------------------------------------
754------------------------------------------------------------------------------
756-- | Polymorphic version of the 'Semigroup' operation '<>' of 'AheadT'.
757-- Merges two streams sequentially but with concurrent lookahead.
761ahead :: (IsStream t, MonadAsync m) => t m a -> t m a -> t m a
762ahead m1 m2 = fromStream $ S.ahead (toStream m1) (toStream m2)
764instance MonadAsync m => Semigroup (AheadT m a) where
767------------------------------------------------------------------------------
769------------------------------------------------------------------------------
771instance MonadAsync m => Monoid (AheadT m a) where
775------------------------------------------------------------------------------
777------------------------------------------------------------------------------
779{-# INLINE aheadbind #-}
783 -> (a -> S.Stream m b)
788 S.Stream $ \ctx stp sng yld ->
789 let run x = (S.runStream x) ctx stp sng yld
791 yield a r = run $ f a `S.ahead` go r
792 in g Nothing stp single yield
794instance MonadAsync m => Monad (AheadT m) where
796 (AheadT m) >>= f = AheadT $ aheadbind m (getAheadT . f)
798------------------------------------------------------------------------------
800------------------------------------------------------------------------------
802MONAD_APPLICATIVE_INSTANCE(AheadT,MONADPARALLEL)
803MONAD_COMMON_INSTANCES(AheadT, MONADPARALLEL)
805------------------------------------------------------------------------------
807------------------------------------------------------------------------------
809-- | Deep async composition or async composition with depth first traversal. In
810-- a left to right 'Semigroup' composition it tries to yield elements from the
811-- left stream as long as it can, but it can run the right stream in parallel
812-- if it needs to, based on demand. The right stream can be run if the left
813-- stream blocks on IO or cannot produce elements fast enough for the consumer.
816-- main = ('toList' . 'asyncly' $ (fromFoldable [1,2]) \<> (fromFoldable [3,4])) >>= print
822-- Any exceptions generated by a constituent stream are propagated to the
823-- output stream. The output and exceptions from a single stream are guaranteed
824-- to arrive in the same order in the resulting stream as they were generated
825-- in the input stream. However, the relative ordering of elements from
826-- different streams in the resulting stream can vary depending on scheduling
827-- and generation delays.
829-- Similarly, the monad instance of 'AsyncT' /may/ run each iteration
830-- concurrently based on demand. More concurrent iterations are started only
831-- if the previous iterations are not able to produce enough output for the
836-- import qualified "Streamly.Prelude" as S
837-- import Control.Concurrent
839-- main = 'runStream' . 'asyncly' $ do
840-- n <- return 3 \<\> return 2 \<\> return 1
842-- threadDelay (n * 1000000)
843-- myThreadId >>= \\tid -> putStrLn (show tid ++ ": Delay " ++ show n)
846-- ThreadId 40: Delay 1
847-- ThreadId 39: Delay 2
848-- ThreadId 38: Delay 3
851-- All iterations may run in the same thread if they do not block.
853-- Note that async composition with depth first traversal can be used to
854-- combine infinite number of streams as it explores only a bounded number of
858newtype AsyncT m a = AsyncT {getAsyncT :: S.Stream m a}
859 deriving (Functor, MonadTrans)
861instance IsStream AsyncT where
866 {-# SPECIALIZE consM :: IO a -> AsyncT IO a -> AsyncT IO a #-}
867 consM m r = fromStream $ S.consMAsync m (toStream r)
870 {-# SPECIALIZE (|:) :: IO a -> AsyncT IO a -> AsyncT IO a #-}
873------------------------------------------------------------------------------
875------------------------------------------------------------------------------
877-- | Polymorphic version of the 'Semigroup' operation '<>' of 'AsyncT'.
878-- Merges two streams possibly concurrently, preferring the
879-- elements from the left one when available.
883async :: (IsStream t, MonadAsync m) => t m a -> t m a -> t m a
884async m1 m2 = fromStream $ S.async (toStream m1) (toStream m2)
886instance MonadAsync m => Semigroup (AsyncT m a) where
889------------------------------------------------------------------------------
891------------------------------------------------------------------------------
893instance MonadAsync m => Monoid (AsyncT m a) where
897------------------------------------------------------------------------------
899------------------------------------------------------------------------------
901{-# INLINE parbind #-}
903 :: (forall c. S.Stream m c -> S.Stream m c -> S.Stream m c)
905 -> (a -> S.Stream m b)
907parbind par m f = go m
910 S.Stream $ \ctx stp sng yld ->
911 let run x = (S.runStream x) ctx stp sng yld
913 yield a r = run $ f a `par` go r
914 in g Nothing stp single yield
916instance MonadAsync m => Monad (AsyncT m) where
918 (AsyncT m) >>= f = AsyncT $ parbind S.async m (getAsyncT . f)
920------------------------------------------------------------------------------
922------------------------------------------------------------------------------
924MONAD_APPLICATIVE_INSTANCE(AsyncT,MONADPARALLEL)
925MONAD_COMMON_INSTANCES(AsyncT, MONADPARALLEL)
927------------------------------------------------------------------------------
929------------------------------------------------------------------------------
931-- | Wide async composition or async composition with breadth first traversal.
932-- The Semigroup instance of 'WAsyncT' concurrently /traverses/ the composed
933-- streams using a depth first travesal or in a round robin fashion, yielding
934-- elements from both streams alternately.
937-- main = ('toList' . 'wAsyncly' $ (fromFoldable [1,2]) \<> (fromFoldable [3,4])) >>= print
943-- Any exceptions generated by a constituent stream are propagated to the
944-- output stream. The output and exceptions from a single stream are guaranteed
945-- to arrive in the same order in the resulting stream as they were generated
946-- in the input stream. However, the relative ordering of elements from
947-- different streams in the resulting stream can vary depending on scheduling
948-- and generation delays.
950-- Similarly, the 'Monad' instance of 'WAsyncT' runs /all/ iterations fairly
951-- concurrently using a round robin scheduling.
955-- import qualified "Streamly.Prelude" as S
956-- import Control.Concurrent
958-- main = 'runStream' . 'wAsyncly' $ do
959-- n <- return 3 \<\> return 2 \<\> return 1
961-- threadDelay (n * 1000000)
962-- myThreadId >>= \\tid -> putStrLn (show tid ++ ": Delay " ++ show n)
965-- ThreadId 40: Delay 1
966-- ThreadId 39: Delay 2
967-- ThreadId 38: Delay 3
970-- Unlike 'AsyncT' all iterations are guaranteed to run fairly
971-- concurrently, unconditionally.
973-- Note that async composition with breadth first traversal can only combine a
974-- finite number of streams as it needs to retain state for each unfinished
978newtype WAsyncT m a = WAsyncT {getWAsyncT :: S.Stream m a}
979 deriving (Functor, MonadTrans)
981instance IsStream WAsyncT where
982 toStream = getWAsyncT
986 {-# SPECIALIZE consM :: IO a -> WAsyncT IO a -> WAsyncT IO a #-}
987 consM m r = fromStream $ S.consMWAsync m (toStream r)
990 {-# SPECIALIZE (|:) :: IO a -> WAsyncT IO a -> WAsyncT IO a #-}
993------------------------------------------------------------------------------
995------------------------------------------------------------------------------
997-- | Polymorphic version of the 'Semigroup' operation '<>' of 'WAsyncT'.
998-- Merges two streams concurrently choosing elements from both fairly.
1001{-# INLINE wAsync #-}
1002wAsync :: (IsStream t, MonadAsync m) => t m a -> t m a -> t m a
1003wAsync m1 m2 = fromStream $ S.wAsync (toStream m1) (toStream m2)
1005instance MonadAsync m => Semigroup (WAsyncT m a) where
1008------------------------------------------------------------------------------
1010------------------------------------------------------------------------------
1012instance MonadAsync m => Monoid (WAsyncT m a) where
1016------------------------------------------------------------------------------
1018------------------------------------------------------------------------------
1020instance MonadAsync m => Monad (WAsyncT m) where
1023 WAsyncT $ parbind S.wAsync m (getWAsyncT . f)
1025------------------------------------------------------------------------------
1027------------------------------------------------------------------------------
1029MONAD_APPLICATIVE_INSTANCE(WAsyncT,MONADPARALLEL)
1030MONAD_COMMON_INSTANCES(WAsyncT, MONADPARALLEL)
1032------------------------------------------------------------------------------
1034------------------------------------------------------------------------------
1036-- | Async composition with simultaneous traversal of all streams.
1038-- The Semigroup instance of 'ParallelT' concurrently /merges/ two streams,
1039-- running both strictly concurrently and yielding elements from both streams
1040-- as they arrive. When multiple streams are combined using 'ParallelT' each
1041-- one is evaluated in its own thread and the results produced are presented in
1042-- the combined stream on a first come first serve basis.
1044-- 'AsyncT' and 'WAsyncT' are /concurrent lookahead streams/ each with a
1045-- specific type of consumption pattern (depth first or breadth first). Since
1046-- they are lookahead, they may introduce certain default latency in starting
1047-- more concurrent tasks for efficiency reasons or may put a default limitation
1048-- on the resource consumption (e.g. number of concurrent threads for
1049-- lookahead). If we look at the implementation detail, they both can share a
1050-- pool of worker threads to evaluate the streams in the desired pattern and at
1051-- the desired rate. However, 'ParallelT' uses a separate runtime thread to
1052-- evaluate each stream.
1054-- 'WAsyncT' is similar to 'ParallelT', as both of them evaluate the
1055-- constituent streams fairly in a round robin fashion.
1056-- However, the key difference is that 'WAsyncT' is lazy or pull driven
1057-- whereas 'ParallelT' is strict or push driven. 'ParallelT' immediately
1058-- starts concurrent evaluation of both the streams (in separate threads) and
1059-- later picks the results whereas 'WAsyncT' may wait for a certain latency
1060-- threshold before initiating concurrent evaluation of the next stream. The
1061-- concurrent scheduling of the next stream or the degree of concurrency is
1062-- driven by the feedback from the consumer. In case of 'ParallelT' each stream
1063-- is evaluated in a separate thread and results are /pushed/ to a shared
1064-- output buffer, the evaluation rate is controlled by blocking when the buffer
1067-- Concurrent lookahead streams are generally more efficient than
1068-- 'ParallelT' and can work pretty efficiently even for smaller tasks because
1069-- they do not necessarily use a separate thread for each task. So they should
1070-- be preferred over 'ParallelT' especially when efficiency is a concern and
1071-- simultaneous strict evaluation is not a requirement. 'ParallelT' is useful
1072-- for cases when the streams are required to be evaluated simultaneously
1073-- irrespective of how the consumer consumes them e.g. when we want to race
1074-- two tasks and want to start both strictly at the same time or if we have
1075-- timers in the parallel tasks and our results depend on the timers being
1076-- started at the same time. We can say that 'ParallelT' is almost the same
1077-- (modulo some implementation differences) as 'WAsyncT' when the latter is
1078-- used with unlimited lookahead and zero latency in initiating lookahead.
1081-- main = ('toList' . 'parallely' $ (fromFoldable [1,2]) \<> (fromFoldable [3,4])) >>= print
1087-- When streams with more than one element are merged, it yields whichever
1088-- stream yields first without any bias, unlike the 'Async' style streams.
1090-- Any exceptions generated by a constituent stream are propagated to the
1091-- output stream. The output and exceptions from a single stream are guaranteed
1092-- to arrive in the same order in the resulting stream as they were generated
1093-- in the input stream. However, the relative ordering of elements from
1094-- different streams in the resulting stream can vary depending on scheduling
1095-- and generation delays.
1097-- Similarly, the 'Monad' instance of 'ParallelT' runs /all/ iterations
1098-- of the loop concurrently.
1102-- import qualified "Streamly.Prelude" as S
1103-- import Control.Concurrent
1105-- main = 'runStream' . 'parallely' $ do
1106-- n <- return 3 \<\> return 2 \<\> return 1
1108-- threadDelay (n * 1000000)
1109-- myThreadId >>= \\tid -> putStrLn (show tid ++ ": Delay " ++ show n)
1112-- ThreadId 40: Delay 1
1113-- ThreadId 39: Delay 2
1114-- ThreadId 38: Delay 3
1117-- Note that parallel composition can only combine a finite number of
1118-- streams as it needs to retain state for each unfinished stream.
1121newtype ParallelT m a = ParallelT {getParallelT :: S.Stream m a}
1122 deriving (Functor, MonadTrans)
1124instance IsStream ParallelT where
1125 toStream = getParallelT
1126 fromStream = ParallelT
1128 {-# INLINE consM #-}
1129 {-# SPECIALIZE consM :: IO a -> ParallelT IO a -> ParallelT IO a #-}
1130 consM m r = fromStream $ S.consMParallel m (toStream r)
1133 {-# SPECIALIZE (|:) :: IO a -> ParallelT IO a -> ParallelT IO a #-}
1136------------------------------------------------------------------------------
1138------------------------------------------------------------------------------
1140-- | Polymorphic version of the 'Semigroup' operation '<>' of 'ParallelT'
1141-- Merges two streams concurrently.
1144{-# INLINE parallel #-}
1145parallel :: (IsStream t, MonadAsync m) => t m a -> t m a -> t m a
1146parallel m1 m2 = fromStream $ S.parallel (toStream m1) (toStream m2)
1148instance MonadAsync m => Semigroup (ParallelT m a) where
1151------------------------------------------------------------------------------
1153------------------------------------------------------------------------------
1155instance MonadAsync m => Monoid (ParallelT m a) where
1159------------------------------------------------------------------------------
1161------------------------------------------------------------------------------
1163instance MonadAsync m => Monad (ParallelT m) where
1165 (ParallelT m) >>= f = ParallelT $ parbind S.parallel m (getParallelT . f)
1167------------------------------------------------------------------------------
1169------------------------------------------------------------------------------
1171MONAD_APPLICATIVE_INSTANCE(ParallelT,MONADPARALLEL)
1172MONAD_COMMON_INSTANCES(ParallelT, MONADPARALLEL)
1174------------------------------------------------------------------------------
1175-- Serially Zipping Streams
1176------------------------------------------------------------------------------
1178-- | The applicative instance of 'ZipSerialM' zips a number of streams serially
1179-- i.e. it produces one element from each stream serially and then zips all
1183-- main = (toList . 'zipSerially' $ (,,) \<$\> s1 \<*\> s2 \<*\> s3) >>= print
1184-- where s1 = fromFoldable [1, 2]
1185-- s2 = fromFoldable [3, 4]
1186-- s3 = fromFoldable [5, 6]
1192-- The 'Semigroup' instance of this type works the same way as that of
1196newtype ZipSerialM m a = ZipSerialM {getZipSerialM :: S.Stream m a}
1197 deriving (Functor, Semigroup, Monoid)
1201{-# DEPRECATED ZipStream "Please use 'ZipSerialM' instead." #-}
1202type ZipStream = ZipSerialM
1204instance IsStream ZipSerialM where
1205 toStream = getZipSerialM
1206 fromStream = ZipSerialM
1208 {-# INLINE consM #-}
1209 {-# SPECIALIZE consM :: IO a -> ZipSerialM IO a -> ZipSerialM IO a #-}
1210 consM :: Monad m => m a -> ZipSerialM m a -> ZipSerialM m a
1214 {-# SPECIALIZE (|:) :: IO a -> ZipSerialM IO a -> ZipSerialM IO a #-}
1215 (|:) :: Monad m => m a -> ZipSerialM m a -> ZipSerialM m a
1218instance Monad m => Applicative (ZipSerialM m) where
1219 pure = ZipSerialM . S.repeat
1220 m1 <*> m2 = fromStream $ S.zipWith id (toStream m1) (toStream m2)
1222------------------------------------------------------------------------------
1223-- Parallely Zipping Streams
1224------------------------------------------------------------------------------
1226-- | Like 'ZipSerialM' but zips in parallel, it generates all the elements to
1227-- be zipped concurrently.
1230-- main = (toList . 'zipAsyncly' $ (,,) \<$\> s1 \<*\> s2 \<*\> s3) >>= print
1231-- where s1 = fromFoldable [1, 2]
1232-- s2 = fromFoldable [3, 4]
1233-- s3 = fromFoldable [5, 6]
1239-- The 'Semigroup' instance of this type works the same way as that of
1243newtype ZipAsyncM m a = ZipAsyncM {getZipAsyncM :: S.Stream m a}
1244 deriving (Functor, Semigroup, Monoid)
1246instance IsStream ZipAsyncM where
1247 toStream = getZipAsyncM
1248 fromStream = ZipAsyncM
1250 {-# INLINE consM #-}
1251 {-# SPECIALIZE consM :: IO a -> ZipAsyncM IO a -> ZipAsyncM IO a #-}
1252 consM :: Monad m => m a -> ZipAsyncM m a -> ZipAsyncM m a
1256 {-# SPECIALIZE (|:) :: IO a -> ZipAsyncM IO a -> ZipAsyncM IO a #-}
1257 (|:) :: Monad m => m a -> ZipAsyncM m a -> ZipAsyncM m a
1260instance MonadAsync m => Applicative (ZipAsyncM m) where
1261 pure = ZipAsyncM . S.repeat
1262 m1 <*> m2 = fromStream $ S.zipAsyncWith id (toStream m1) (toStream m2)
1264-------------------------------------------------------------------------------
1265-- Type adapting combinators
1266-------------------------------------------------------------------------------
1268-- | Adapt any specific stream type to any other specific stream type.
1271adapt :: (IsStream t1, IsStream t2) => t1 m a -> t2 m a
1272adapt = fromStream . toStream
1274-- | Fix the type of a polymorphic stream as 'SerialT'.
1277serially :: IsStream t => SerialT m a -> t m a
1280-- | Fix the type of a polymorphic stream as 'WSerialT'.
1283wSerially :: IsStream t => WSerialT m a -> t m a
1286-- | Same as 'wSerially'.
1289{-# DEPRECATED interleaving "Please use wSerially instead." #-}
1290interleaving :: IsStream t => WSerialT m a -> t m a
1291interleaving = wSerially
1293-- | Fix the type of a polymorphic stream as 'AheadT'.
1296aheadly :: IsStream t => AheadT m a -> t m a
1299-- | Fix the type of a polymorphic stream as 'AsyncT'.
1302asyncly :: IsStream t => AsyncT m a -> t m a
1305-- | Fix the type of a polymorphic stream as 'WAsyncT'.
1308wAsyncly :: IsStream t => WAsyncT m a -> t m a
1311-- | Fix the type of a polymorphic stream as 'ParallelT'.
1314parallely :: IsStream t => ParallelT m a -> t m a
1317-- | Fix the type of a polymorphic stream as 'ZipSerialM'.
1320zipSerially :: IsStream t => ZipSerialM m a -> t m a
1323-- | Same as 'zipSerially'.
1326{-# DEPRECATED zipping "Please use zipSerially instead." #-}
1327zipping :: IsStream t => ZipSerialM m a -> t m a
1328zipping = zipSerially
1330-- | Fix the type of a polymorphic stream as 'ZipAsyncM'.
1333zipAsyncly :: IsStream t => ZipAsyncM m a -> t m a
1336-- | Same as 'zipAsyncly'.
1339{-# DEPRECATED zippingAsync "Please use zipAsyncly instead." #-}
1340zippingAsync :: IsStream t => ZipAsyncM m a -> t m a
1341zippingAsync = zipAsyncly
1343-------------------------------------------------------------------------------
1344-- Running Streams, convenience functions specialized to types
1345-------------------------------------------------------------------------------
1347-- | Same as @runStream@.
1350{-# DEPRECATED runStreamT "Please use runStream instead." #-}
1351runStreamT :: Monad m => SerialT m a -> m ()
1352runStreamT = runStream
1354-- | Same as @runStream . wSerially@.
1357{-# DEPRECATED runInterleavedT "Please use 'runStream . interleaving' instead." #-}
1358runInterleavedT :: Monad m => InterleavedT m a -> m ()
1359runInterleavedT = runStream . wSerially
1361-- | Same as @runStream . asyncly@.
1364{-# DEPRECATED runAsyncT "Please use 'runStream . asyncly' instead." #-}
1365runAsyncT :: Monad m => AsyncT m a -> m ()
1366runAsyncT = runStream . asyncly
1368-- | Same as @runStream . parallely@.
1371{-# DEPRECATED runParallelT "Please use 'runStream . parallely' instead." #-}
1372runParallelT :: Monad m => ParallelT m a -> m ()
1373runParallelT = runStream . parallely
1375-- | Same as @runStream . zipping@.
1378{-# DEPRECATED runZipStream "Please use 'runStream . zipSerially instead." #-}
1379runZipStream :: Monad m => ZipSerialM m a -> m ()
1380runZipStream = runStream . zipSerially
1382-- | Same as @runStream . zippingAsync@.
1385{-# DEPRECATED runZipAsync "Please use 'runStream . zipAsyncly instead." #-}
1386runZipAsync :: Monad m => ZipAsyncM m a -> m ()
1387runZipAsync = runStream . zipAsyncly
1389------------------------------------------------------------------------------
1391------------------------------------------------------------------------------
1393-- | A serial IO stream of elements of type @a@. See 'SerialT' documentation
1397type Serial a = SerialT IO a
1399-- | An interleaving serial IO stream of elements of type @a@. See 'WSerialT'
1400-- documentation for more details.
1403type WSerial a = WSerialT IO a
1405-- | A serial IO stream of elements of type @a@ with concurrent lookahead. See
1406-- 'AheadT' documentation for more details.
1409type Ahead a = AheadT IO a
1411-- | A demand driven left biased parallely composing IO stream of elements of
1412-- type @a@. See 'AsyncT' documentation for more details.
1415type Async a = AsyncT IO a
1417-- | A round robin parallely composing IO stream of elements of type @a@.
1418-- See 'WAsyncT' documentation for more details.
1421type WAsync a = WAsyncT IO a
1423-- | A parallely composing IO stream of elements of type @a@.
1424-- See 'ParallelT' documentation for more details.
1427type Parallel a = ParallelT IO a
1429-- | An IO stream whose applicative instance zips streams serially.
1432type ZipSerial a = ZipSerialM IO a
1434-- | An IO stream whose applicative instance zips streams wAsyncly.
1437type ZipAsync a = ZipAsyncM IO a
1439------------------------------------------------------------------------------
1441------------------------------------------------------------------------------
1443-- | A variant of 'Data.Foldable.fold' that allows you to fold a 'Foldable'
1444-- container of streams using the specified stream sum operation.
1446-- @foldWith 'async' $ map return [1..3]@
1449{-# INLINABLE foldWith #-}
1450foldWith :: (IsStream t, Foldable f)
1451 => (t m a -> t m a -> t m a) -> f (t m a) -> t m a
1452foldWith f = foldr f nil
1454-- | A variant of 'foldMap' that allows you to map a monadic streaming action
1455-- on a 'Foldable' container and then fold it using the specified stream sum
1458-- @foldMapWith 'async' return [1..3]@
1461{-# INLINABLE foldMapWith #-}
1462foldMapWith :: (IsStream t, Foldable f)
1463 => (t m b -> t m b -> t m b) -> (a -> t m b) -> f a -> t m b
1464foldMapWith f g = foldr (f . g) nil
1466-- | Like 'foldMapWith' but with the last two arguments reversed i.e. the
1467-- monadic streaming function is the last argument.
1470{-# INLINABLE forEachWith #-}
1471forEachWith :: (IsStream t, Foldable f)
1472 => (t m b -> t m b -> t m b) -> f a -> (a -> t m b) -> t m b
1473forEachWith 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