1{-# LANGUAGE ConstraintKinds #-}
2{-# LANGUAGE FlexibleContexts #-}
3{-# LANGUAGE FlexibleInstances #-}
4{-# LANGUAGE LambdaCase #-}
5{-# LANGUAGE MagicHash #-}
6{-# LANGUAGE MultiParamTypeClasses #-}
7{-# LANGUAGE RankNTypes #-}
8{-# LANGUAGE UnboxedTuples #-}
9{-# LANGUAGE UndecidableInstances #-} -- XXX
12-- Module : Streamly.Core
13-- Copyright : (c) 2017 Harendra Kumar
16-- Maintainer : harendra.kumar@gmail.com
17-- Stability : experimental
24 , MonadAsync -- deprecated
37 -- * Semigroup Style Composition
55 -- * Concurrent Stream Vars (SVars)
69import Control.Concurrent (ThreadId, myThreadId, threadDelay)
70import Control.Concurrent.MVar (MVar, newEmptyMVar, tryTakeMVar,
72import Control.Exception (SomeException (..))
73import qualified Control.Exception.Lifted as EL
74import Control.Monad (when)
75import Control.Monad.Catch (MonadThrow, throwM)
76import Control.Monad.Error.Class (MonadError(..))
77import Control.Monad.IO.Class (MonadIO(..))
78import Control.Monad.Reader.Class (MonadReader(..))
79import Control.Monad.Trans.Class (MonadTrans (lift))
80import Control.Monad.Trans.Control (MonadBaseControl, liftBaseWith)
81import Data.Atomics (casIORef, readForCAS, peekTicket
82 ,atomicModifyIORefCAS_)
83import Data.Concurrent.Queue.MichaelScott (LinkedQueue, newQ, pushL,
85import Data.Functor (void)
86import Data.IORef (IORef, modifyIORef, newIORef,
87 readIORef, atomicModifyIORef)
88import Data.Maybe (isNothing, fromJust)
89import Data.Semigroup (Semigroup(..))
91import qualified Data.Set as S
92import Prelude hiding (repeat, zipWith)
95import GHC.Conc (ThreadId(..))
98------------------------------------------------------------------------------
99-- Parent child thread communication type
100------------------------------------------------------------------------------
102-- | Events that a child thread may send to a parent thread.
105 | ChildStop ThreadId (Maybe SomeException)
107------------------------------------------------------------------------------
108-- State threaded around the monad for thread management
109------------------------------------------------------------------------------
111-- | Conjunction is used for monadic/product style composition. Disjunction is
112-- used for fold/sum style composition. We need to distinguish the two types of
113-- SVars so that the scheduling of the two is independent.
114data SVarTag = Conjunction | Disjunction deriving Eq
117 LIFO -- depth first concurrent
118 | FIFO -- breadth first concurrent
119 | Par -- all parallel
122-- | Identify the type of the SVar. Two computations using the same style can
123-- be scheduled on the same SVar.
124data SVarStyle = SVarStyle SVarTag SVarSched deriving Eq
126-- | An SVar or a Stream Var is a conduit to the output from multiple streams
127-- running concurrently and asynchronously. An SVar can be thought of as an
128-- asynchronous IO handle. We can write any number of streams to an SVar in a
129-- non-blocking manner and then read them back at any time at any pace. The
130-- SVar would run the streams asynchronously and accumulate results. An SVar
131-- may not really execute the stream completely and accumulate all the results.
132-- However, it ensures that the reader can read the results at whatever paces
133-- it wants to read. The SVar monitors and adapts to the consumer's pace.
135-- An SVar is a mini scheduler, it has an associated runqueue that holds the
136-- stream tasks to be picked and run by a pool of worker threads. It has an
137-- associated output queue where the output stream elements are placed by the
138-- worker threads. A doorBell is used by the worker threads to intimate the
139-- consumer thread about availability of new results in the output queue. More
140-- workers are added to the SVar by 'fromStreamVar' on demand if the output
141-- produced is not keeping pace with the consumer. On bounded SVars, workers
142-- block on the output queue to provide throttling of the producer when the
143-- consumer is not pulling fast enough. The number of workers may even get
144-- reduced depending on the consuming pace.
146-- New work is enqueued either at the time of creation of the SVar or as a
147-- result of executing the parallel combinators i.e. '<|' and '<|>' when the
148-- already enqueued computations get evaluated. See 'joinStreamVar2'.
151 SVar { outputQueue :: IORef [ChildEvent a]
152 , doorBell :: MVar Bool -- wakeup mechanism for outQ
153 , enqueue :: Stream m a -> IO ()
155 , runningThreads :: IORef (Set ThreadId)
156 , queueEmpty :: m Bool
157 , activeWorkers :: IORef Int
158 , svarStyle :: SVarStyle
161-- Slightly faster version of CAS. Gained some improvement by avoiding the use
162-- of "evaluate" because we know we do not have exceptions in fn.
163{-# INLINE atomicModifyIORefCAS #-}
164atomicModifyIORefCAS :: IORef a -> (a -> (a,b)) -> IO b
165atomicModifyIORefCAS ref fn = do
166 tkt <- readForCAS ref
172 loop _ 0 = atomicModifyIORef ref fn
174 let (new, result) = fn $ peekTicket old
175 (success, tkt) <- casIORef ref old new
178 else loop tkt (tries - 1)
180------------------------------------------------------------------------------
182------------------------------------------------------------------------------
184-- | The type 'Stream m a' represents a monadic stream of values of type 'a'
185-- constructed using actions in monad 'm'. It uses stop, singleton and yield
186-- continuations equivalent to the following direct style type:
188-- data Stream m a = Stop | Singleton a | Yield a (Stream m a)
190-- To facilitate parallel composition we maintain a local state in an SVar that
191-- is shared across and is used for synchronization of the streams being
194-- The singleton case can be expressed in terms of stop and yield but we have
195-- it as a separate case to optimize composition operations for streams with
196-- single element. We build singleton streams in the implementation of 'pure'
197-- for Applicative and Monad, and in 'lift' for MonadTrans.
201 runStream :: forall r.
202 Maybe (SVar m a) -- local state
204 -> (a -> m r) -- singleton
205 -> (a -> Stream m a -> m r) -- yield
210nil = Stream $ \_ stp _ _ -> stp
212once :: Monad m => m a -> Stream m a
213once m = Stream $ \_ _ single _ -> m >>= single
215{-# INLINE singleton #-}
216-- | Same as @once . return@
217singleton :: a -> Stream m a
218singleton a = Stream $ \_ _ single _ -> single a
220consM :: Monad m => m a -> Stream m a -> Stream m a
221consM m r = Stream $ \_ _ _ yld -> m >>= \a -> yld a r
223-- | Same as @consM . return@
224cons :: a -> Stream m a -> Stream m a
225cons a r = Stream $ \_ _ _ yld -> yld a r
227repeat :: a -> Stream m a
228repeat a = let x = cons a x in x
230------------------------------------------------------------------------------
232------------------------------------------------------------------------------
234-- | Concatenates two streams sequentially i.e. the first stream is
235-- exhausted completely before yielding any element from the second stream.
236splice :: Stream m a -> Stream m a -> Stream m a
239 go (Stream m) = Stream $ \_ stp sng yld ->
240 let stop = (runStream m2) Nothing stp sng yld
242 yield a r = yld a (go r)
243 in m Nothing stop single yield
245instance Semigroup (Stream m a) where
248------------------------------------------------------------------------------
250------------------------------------------------------------------------------
252instance Monoid (Stream m a) where
256------------------------------------------------------------------------------
258------------------------------------------------------------------------------
260cosplice :: Stream m a -> Stream m a -> Stream m a
261cosplice m1 m2 = Stream $ \_ stp sng yld -> do
262 let stop = (runStream m2) Nothing stp sng yld
264 yield a r = yld a (cosplice m2 r)
265 (runStream m1) Nothing stop single yield
267------------------------------------------------------------------------------
268-- Spawning threads and collecting result in streamed fashion
269------------------------------------------------------------------------------
271-- | A monad that can perform concurrent or parallel IO operations. Streams
272-- that can be composed concurrently require the underlying monad to be
276type MonadParallel m = (MonadIO m, MonadBaseControl IO m, MonadThrow m)
280{-# DEPRECATED MonadAsync "Please use MonadParallel instead." #-}
281type MonadAsync m = MonadParallel m
283-- Stolen from the async package. The perf improvement is modest, 2% on a
284-- thread heavy benchmark (parallel composition using noop computations).
285-- A version of forkIO that does not include the outer exception
286-- handler: saves a bit of time when we will be installing our own
288{-# INLINE rawForkIO #-}
289rawForkIO :: IO () -> IO ThreadId
290rawForkIO action = IO $ \ s ->
291 case (fork# action s) of (# s1, tid #) -> (# s1, ThreadId tid #)
294doFork :: MonadBaseControl IO m
296 -> (SomeException -> m ())
298doFork action exHandler =
299 EL.mask $ \restore ->
300 liftBaseWith $ \runInIO -> rawForkIO $ do
301 _ <- runInIO $ EL.catch (restore action) exHandler
302 -- XXX restore state here?
305-- XXX exception safety of all atomic/MVar operations
308send :: MonadIO m => SVar m a -> ChildEvent a -> m ()
309send sv msg = liftIO $ do
310 old <- atomicModifyIORefCAS (outputQueue sv) $ \es -> (msg : es, es)
312 -- XXX need a memory barrier? The wake up must happen only after the
313 -- store has finished otherwise we can have lost wakeup problems.
314 then void $ tryPutMVar (doorBell sv) True
317{-# INLINE sendStop #-}
318sendStop :: MonadIO m => SVar m a -> m ()
319sendStop sv = liftIO myThreadId >>= \tid -> send sv (ChildStop tid Nothing)
321-- Note: Left associated compositions can grow this queue to a large size
322{-# INLINE enqueueLIFO #-}
323enqueueLIFO :: IORef [Stream m a] -> Stream m a -> IO ()
324enqueueLIFO q m = atomicModifyIORefCAS_ q $ \ ms -> m : ms
326runqueueLIFO :: MonadIO m => SVar m a -> IORef [Stream m a] -> m ()
327runqueueLIFO sv q = run
334 Nothing -> sendStop sv
335 Just m -> (runStream m) (Just sv) run single yield
337 sendit a = send sv (ChildYield a)
338 single a = sendit a >> run
339 yield a r = sendit a >> (runStream r) (Just sv) run single yield
341 dequeue = liftIO $ atomicModifyIORefCAS q $ \case
343 x : xs -> (xs, Just x)
345{-# INLINE enqueueFIFO #-}
346enqueueFIFO :: LinkedQueue (Stream m a) -> Stream m a -> IO ()
349runqueueFIFO :: MonadIO m => SVar m a -> LinkedQueue (Stream m a) -> m ()
350runqueueFIFO sv q = run
357 Nothing -> sendStop sv
358 Just m -> (runStream m) (Just sv) run single yield
360 dequeue = liftIO $ tryPopR q
361 sendit a = send sv (ChildYield a)
362 single a = sendit a >> run
363 yield a r = sendit a >> liftIO (enqueueFIFO q r) >> run
366runOne :: MonadIO m => SVar m a -> Stream m a -> m ()
367runOne sv m = (runStream m) (Just sv) stop single yield
372 sendit a = send sv (ChildYield a)
373 single a = sendit a >> stop
374 yield a r = sendit a >> runOne sv r
376-- Thread tracking is needed for two reasons:
378-- 1) Killing threads on exceptions. Threads may not be left to go away by
379-- themselves because they may run for significant times before going away or
380-- worse they may be stuck in IO and never go away.
382-- 2) To know when all threads are done and the stream has ended.
384{-# NOINLINE addThread #-}
385addThread :: MonadIO m => SVar m a -> ThreadId -> m ()
387 liftIO $ modifyIORef (runningThreads sv) (S.insert tid)
390{-# INLINE delThread #-}
391delThread :: MonadIO m => SVar m a -> ThreadId -> m ()
393 liftIO $ modifyIORef (runningThreads sv) $ (\s -> S.delete tid s)
396-- If present then delete else add. This takes care of out of order add and
397-- delete i.e. a delete arriving before we even added a thread.
398-- This occurs when the forked thread is done even before the 'addThread' right
399-- after the fork is executed.
400{-# INLINE modifyThread #-}
401modifyThread :: MonadIO m => SVar m a -> ThreadId -> m ()
402modifyThread sv tid = do
403 changed <- liftIO $ atomicModifyIORefCAS (runningThreads sv) $ \old ->
404 if (S.member tid old)
405 then let new = (S.delete tid old) in (new, new)
406 else let new = (S.insert tid old) in (new, old)
408 then liftIO $ void $ tryPutMVar (doorBell sv) True
411-- | This is safe even if we are adding more threads concurrently because if
412-- a child thread is adding another thread then anyway 'runningThreads' will
414{-# INLINE allThreadsDone #-}
415allThreadsDone :: MonadIO m => SVar m a -> m Bool
416allThreadsDone sv = liftIO $ S.null <$> readIORef (runningThreads sv)
418{-# NOINLINE handleChildException #-}
419handleChildException :: MonadIO m => SVar m a -> SomeException -> m ()
420handleChildException sv e = do
421 tid <- liftIO myThreadId
422 send sv (ChildStop tid (Just e))
424{-# NOINLINE pushWorker #-}
425pushWorker :: MonadParallel m => SVar m a -> m ()
427 liftIO $ atomicModifyIORefCAS_ (activeWorkers sv) $ \n -> n + 1
428 doFork (runqueue sv) (handleChildException sv) >>= addThread sv
430-- | In contrast to pushWorker which always happens only from the consumer
431-- thread, a pushWorkerPar can happen concurrently from multiple threads on the
432-- producer side. So we need to use a thread safe modification of
433-- runningThreads. Alternatively, we can use a CreateThread event to avoid
434-- using a CAS based modification.
435{-# NOINLINE pushWorkerPar #-}
436pushWorkerPar :: MonadParallel m => SVar m a -> Stream m a -> m ()
438 doFork (runOne sv m) (handleChildException sv) >>= modifyThread sv
440-- XXX When the queue is LIFO we can put a limit on the number of dispatches.
441-- Also, if a worker blocks on the output queue we can decide if we want to
442-- block or make it go away entirely, depending on the number of workers and
443-- the type of the queue.
444{-# INLINE sendWorkerWait #-}
445sendWorkerWait :: MonadParallel m => SVar m a -> m ()
446sendWorkerWait sv = do
447 -- When there is no output seen we dispatch more workers to help out if
448 -- there is work pending in the work queue. But we wait a little while
449 -- and check the output again so that we are not too aggressive.
450 -- If there is no output pending to process and there is no worker to be
451 -- sent then we block, so that we do not keep looping fruitlessly.
453 liftIO $ threadDelay 200
454 output <- liftIO $ readIORef (outputQueue sv)
455 when (null output) $ do
456 done <- queueEmpty sv
458 then pushWorker sv >> sendWorkerWait sv
459 else void (liftIO $ takeMVar (doorBell sv))
461-- | Pull a stream from an SVar.
462{-# NOINLINE fromStreamVar #-}
463fromStreamVar :: MonadParallel m => SVar m a -> Stream m a
464fromStreamVar sv = Stream $ \_ stp sng yld -> do
465 let SVarStyle _ sched = svarStyle sv
467 then void $ liftIO $ takeMVar (doorBell sv)
469 res <- liftIO $ tryTakeMVar (doorBell sv)
470 when (isNothing res) $ sendWorkerWait sv
472 list <- liftIO $ atomicModifyIORefCAS (outputQueue sv) $ \x -> ([], x)
473 -- Reversing the output is important to guarantee that we process the
474 -- outputs in the same order as they were generated by the constituent
476 (runStream $ processEvents (reverse list)) Nothing stp sng yld
480 handleException e tid = do
481 liftIO $ atomicModifyIORefCAS_ (activeWorkers sv) $ \n -> n - 1
483 -- XXX implement kill async exception handling
484 -- liftIO $ readIORef (runningThreads sv) >>= mapM_ killThread
487 {-# INLINE processEvents #-}
488 processEvents [] = Stream $ \_ stp sng yld -> do
489 done <- allThreadsDone sv
491 then (runStream (fromStreamVar sv)) Nothing stp sng yld
494 processEvents (ev : es) = Stream $ \_ stp sng yld -> do
495 let continue = (runStream (processEvents es)) Nothing stp sng yld
496 yield a = yld a (processEvents es)
499 ChildYield a -> yield a
503 let active = activeWorkers sv
504 liftIO $ atomicModifyIORefCAS_ active $ \n -> n - 1
505 modifyThread sv tid >> continue
506 Just ex -> handleException ex tid
508getFifoSVar :: MonadIO m => SVarStyle -> IO (SVar m a)
509getFifoSVar ctype = do
511 outQMv <- newEmptyMVar
513 running <- newIORef S.empty
516 SVar { outputQueue = outQ
518 , runningThreads = running
519 , runqueue = runqueueFIFO sv q
521 , queueEmpty = liftIO $ nullQ q
523 , activeWorkers = active
527getLifoSVar :: MonadIO m => SVarStyle -> IO (SVar m a)
528getLifoSVar ctype = do
530 outQMv <- newEmptyMVar
532 running <- newIORef S.empty
534 let checkEmpty = null <$> liftIO (readIORef q)
536 SVar { outputQueue = outQ
538 , runningThreads = running
539 , runqueue = runqueueLIFO sv q
540 , enqueue = enqueueLIFO q
541 , queueEmpty = checkEmpty
543 , activeWorkers = active
547getParSVar :: SVarStyle -> IO (SVar m a)
550 outQMv <- newEmptyMVar
552 running <- newIORef S.empty
554 SVar { outputQueue = outQ
556 , runningThreads = running
557 , runqueue = undefined
558 , enqueue = undefined
559 , queueEmpty = undefined
561 , activeWorkers = active
565-- | Create a new empty SVar.
566newEmptySVar :: MonadParallel m => SVarStyle -> m (SVar m a)
567newEmptySVar style = do
570 SVarStyle _ FIFO -> getFifoSVar style
571 SVarStyle _ LIFO -> getLifoSVar style
572 SVarStyle _ Par -> getParSVar style
574-- | Create a new SVar and enqueue one stream computation on it.
575newStreamVar1 :: MonadParallel m => SVarStyle -> Stream m a -> m (SVar m a)
576newStreamVar1 style m = do
577 sv <- newEmptySVar style
578 -- Note: We must have all the work on the queue before sending the
579 -- pushworker, otherwise the pushworker may exit before we even get a
581 liftIO $ (enqueue sv) m
585-- | Create a new SVar and enqueue two stream computations on it.
586newStreamVar2 :: MonadParallel m
587 => SVarStyle -> Stream m a -> Stream m a -> m (SVar m a)
588newStreamVar2 style m1 m2 = do
589 -- Note: We must have all the work on the queue before sending the
590 -- pushworker, otherwise the pushworker may exit before we even get a
594 SVarStyle _ FIFO -> do
595 c <- getFifoSVar style
596 (enqueue c) m1 >> (enqueue c) m2
598 SVarStyle _ LIFO -> do
599 c <- getLifoSVar style
600 (enqueue c) m2 >> (enqueue c) m1
602 SVarStyle _ Par -> undefined
606-- | Write a stream to an 'SVar' in a non-blocking manner. The stream can then
607-- be read back from the SVar using 'fromSVar'.
608toStreamVar :: MonadParallel m => SVar m a -> Stream m a -> m ()
610 liftIO $ (enqueue sv) m
611 done <- allThreadsDone sv
612 -- XXX This is safe only when called from the consumer thread or when no
613 -- consumer is present. There may be a race if we are not running in the
615 when done $ pushWorker sv
617------------------------------------------------------------------------------
618-- Running streams concurrently
619------------------------------------------------------------------------------
621-- Concurrency rate control.
623-- Our objective is to create more threads on demand if the consumer is running
624-- faster than us. As soon as we encounter a concurrent composition we create a
625-- push pull pair of threads. We use an SVar for communication between the
626-- consumer, pulling from the SVar and the producer who is pushing to the SVar.
627-- The producer creates more threads if the SVar drains and becomes empty, that
628-- is the consumer is running faster.
630-- XXX Note 1: This mechanism can be problematic if the initial production
631-- latency is high, we may end up creating too many threads. So we need some
632-- way to monitor and use the latency as well. Having a limit on the dispatches
633-- (programmer controlled) may also help.
635-- TBD Note 2: We may want to run computations at the lower level of the
636-- composition tree serially even when they are composed using a parallel
637-- combinator. We can use 'splice' in place of 'parAhead' and 'cosplice' in
638-- place of 'coParAhead'. If we find that an SVar immediately above a computation
639-- gets drained empty we can switch to parallelizing the computation. For that
640-- we can use a state flag to fork the rest of the computation at any point of
641-- time inside the Monad bind operation if the consumer is running at a faster
644-- TBD Note 3: the binary operation ('parallel') composition allows us to
645-- dispatch a chunkSize of only 1. If we have to dispatch in arbitrary
646-- chunksizes we will need to compose the parallel actions using a data
647-- constructor (A Free container) instead so that we can divide it in chunks of
648-- arbitrary size before dispatching. If the stream is composed of
649-- hierarchically composed grains of different sizes then we can always switch
650-- to a desired granularity depending on the consumer speed.
652-- TBD Note 4: for pure work (when we are not in the IO monad) we can divide it
653-- into just the number of CPUs.
655-- | Join two computations on the currently running 'SVar' queue for concurrent
656-- execution. When we are using parallel composition, an SVar is passed around
657-- as a state variable. We try to schedule a new parallel computation on the
658-- SVar passed to us. The first time, when no SVar exists, a new SVar is
659-- created. Subsequently, 'joinStreamVar2' may get called when a computation
660-- already scheduled on the SVar is further evaluated. For example, when (a
661-- `parallel` b) is evaluated it calls a 'joinStreamVar2' to put 'a' and 'b' on
662-- the current scheduler queue.
664-- The 'SVarStyle' required by the current composition context is passed as one
665-- of the parameters. If the scheduling and composition style of the new
666-- computation being scheduled is different than the style of the current SVar,
667-- then we create a new SVar and schedule it on that. The newly created SVar
668-- joins as one of the computations on the current SVar queue.
670-- Cases when we need to switch to a new SVar:
672-- * (x `parallel` y) `parallel` (t `parallel` u) -- all of them get scheduled on the same SVar
673-- * (x `parallel` y) `parallel` (t `parAhead` u) -- @t@ and @u@ get scheduled on a new child SVar
674-- because of the scheduling policy change.
675-- * if we 'adapt' a stream of type 'parAhead' to a stream of type
676-- 'Parallel', we create a new SVar at the transitioning bind.
677-- * When the stream is switching from disjunctive composition to conjunctive
678-- composition and vice-versa we create a new SVar to isolate the scheduling
681{-# INLINE joinStreamVar2 #-}
682joinStreamVar2 :: MonadParallel m
683 => SVarStyle -> Stream m a -> Stream m a -> Stream m a
684joinStreamVar2 style m1 m2 = Stream $ \svr stp sng yld ->
686 Just sv | svarStyle sv == style ->
687 liftIO ((enqueue sv) m2) >> (runStream m1) svr stp sng yld
689 sv <- newStreamVar1 style (concurrently m1 m2)
690 (runStream (fromStreamVar sv)) Nothing stp sng yld
692 concurrently ma mb = Stream $ \svr stp sng yld -> do
693 liftIO $ (enqueue (fromJust svr)) mb
694 (runStream ma) svr stp sng yld
696{-# INLINE joinStreamVarPar #-}
697joinStreamVarPar :: MonadParallel m
698 => SVarStyle -> Stream m a -> Stream m a -> Stream m a
699joinStreamVarPar style m1 m2 = Stream $ \svr stp sng yld ->
701 Just sv | svarStyle sv == style -> do
702 pushWorkerPar sv m1 >> (runStream m2) svr stp sng yld
704 sv <- newEmptySVar style
707 (runStream (fromStreamVar sv)) Nothing stp sng yld
709------------------------------------------------------------------------------
710-- Semigroup and Monoid style compositions for parallel actions
711------------------------------------------------------------------------------
713{-# INLINE parAhead #-}
714parAhead :: MonadParallel m => Stream m a -> Stream m a -> Stream m a
715parAhead = joinStreamVar2 (SVarStyle Disjunction LIFO)
717{-# INLINE coparAhead #-}
718coparAhead :: MonadParallel m => Stream m a -> Stream m a -> Stream m a
719coparAhead = joinStreamVar2 (SVarStyle Disjunction FIFO)
721{-# INLINE parallel #-}
722parallel :: MonadParallel m => Stream m a -> Stream m a -> Stream m a
723parallel = joinStreamVarPar (SVarStyle Disjunction Par)
725-------------------------------------------------------------------------------
726-- Functor instace is the same for all types
727-------------------------------------------------------------------------------
729instance Monad m => Functor (Stream m) where
730 fmap f m = Stream $ \_ stp sng yld ->
732 yield a r = yld (f a) (fmap f r)
733 in (runStream m) Nothing stp single yield
735------------------------------------------------------------------------------
736-- Alternative & MonadPlus
737------------------------------------------------------------------------------
739alt :: Stream m a -> Stream m a -> Stream m a
740alt m1 m2 = Stream $ \_ stp sng yld ->
741 let stop = runStream m2 Nothing stp sng yld
742 in runStream m1 Nothing stop sng yld
744------------------------------------------------------------------------------
746------------------------------------------------------------------------------
748zipWith :: (a -> b -> c) -> Stream m a -> Stream m b -> Stream m c
749zipWith f m1 m2 = go m1 m2
751 go mx my = Stream $ \_ stp sng yld -> do
753 let single2 b = sng (f a b)
754 yield2 b rb = yld (f a b) (go ra rb)
755 in (runStream my) Nothing stp single2 yield2
756 let single1 a = merge a nil
757 yield1 a ra = merge a ra
758 (runStream mx) Nothing stp single1 yield1
760async :: MonadParallel m => Stream m a -> m (Stream m a)
761async m = newStreamVar1 (SVarStyle Disjunction LIFO) m
762 >>= return . fromStreamVar
764zipParallelWith :: MonadParallel m
765 => (a -> b -> c) -> Stream m a -> Stream m b -> Stream m c
766zipParallelWith f m1 m2 = Stream $ \_ stp sng yld -> do
769 (runStream (zipWith f ma mb)) Nothing stp sng yld
771-------------------------------------------------------------------------------
773-------------------------------------------------------------------------------
775instance MonadTrans Stream where
778withLocal :: MonadReader r m => (r -> r) -> Stream m a -> Stream m a
780 Stream $ \svr stp sng yld ->
781 let single = local f . sng
782 yield a r = local f $ yld a (withLocal f r)
783 in (runStream m) svr (local f stp) single yield
785-- XXX handle and test cross thread state transfer
788 => Stream m a -> (e -> Stream m a) -> Stream m a
790 Stream $ \svr stp sng yld ->
791 let run x = runStream x svr stp sng yield
792 handle r = r `catchError` \e -> run $ h e
793 yield a r = yld a (withCatchError r h)
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