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
36 -- * Semigroup Style Composition
54 -- * Concurrent Stream Vars (SVars)
68import Control.Concurrent (ThreadId, myThreadId, threadDelay)
69import Control.Concurrent.MVar (MVar, newEmptyMVar, tryTakeMVar,
70 tryPutMVar, takeMVar, readMVar)
71import Control.Exception (SomeException (..))
72import qualified Control.Exception.Lifted as EL
73import Control.Monad (when)
74import Control.Monad.Catch (MonadThrow, throwM)
75import Control.Monad.Error.Class (MonadError(..))
76import Control.Monad.IO.Class (MonadIO(..))
77import Control.Monad.Reader.Class (MonadReader(..))
78import Control.Monad.Trans.Class (MonadTrans (lift))
79import Control.Monad.Trans.Control (MonadBaseControl, liftBaseWith)
80import Data.Atomics (casIORef, readForCAS, peekTicket
81 ,atomicModifyIORefCAS_)
82import Data.Concurrent.Queue.MichaelScott (LinkedQueue, newQ, pushL,
84import Data.Functor (void)
85import Data.IORef (IORef, modifyIORef, newIORef,
86 readIORef, atomicModifyIORef)
87import Data.Maybe (isNothing, fromJust)
88import Data.Semigroup (Semigroup(..))
90import qualified Data.Set as S
91import Prelude hiding (repeat, zipWith)
94import GHC.Conc (ThreadId(..))
97------------------------------------------------------------------------------
98-- Parent child thread communication type
99------------------------------------------------------------------------------
101-- | Events that a child thread may send to a parent thread.
104 | ChildStop ThreadId (Maybe SomeException)
106------------------------------------------------------------------------------
107-- State threaded around the monad for thread management
108------------------------------------------------------------------------------
110-- | Conjunction is used for monadic/product style composition. Disjunction is
111-- used for fold/sum style composition. We need to distinguish the two types of
112-- SVars so that the scheduling of the two is independent.
113data SVarTag = Conjunction | Disjunction deriving Eq
116 LIFO -- depth first concurrent
117 | FIFO -- breadth first concurrent
118 | Par -- all parallel
121-- | Identify the type of the SVar. Two computations using the same style can
122-- be scheduled on the same SVar.
123data SVarStyle = SVarStyle SVarTag SVarSched deriving Eq
125-- | An SVar or a Stream Var is a conduit to the output from multiple streams
126-- running concurrently and asynchronously. An SVar can be thought of as an
127-- asynchronous IO handle. We can write any number of streams to an SVar in a
128-- non-blocking manner and then read them back at any time at any pace. The
129-- SVar would run the streams asynchronously and accumulate results. An SVar
130-- may not really execute the stream completely and accumulate all the results.
131-- However, it ensures that the reader can read the results at whatever paces
132-- it wants to read. The SVar monitors and adapts to the consumer's pace.
134-- An SVar is a mini scheduler, it has an associated runqueue that holds the
135-- stream tasks to be picked and run by a pool of worker threads. It has an
136-- associated output queue where the output stream elements are placed by the
137-- worker threads. A doorBell is used by the worker threads to intimate the
138-- consumer thread about availability of new results in the output queue. More
139-- workers are added to the SVar by 'fromStreamVar' on demand if the output
140-- produced is not keeping pace with the consumer. On bounded SVars, workers
141-- block on the output queue to provide throttling of the producer when the
142-- consumer is not pulling fast enough. The number of workers may even get
143-- reduced depending on the consuming pace.
145-- New work is enqueued either at the time of creation of the SVar or as a
146-- result of executing the parallel combinators i.e. '<|' and '<|>' when the
147-- already enqueued computations get evaluated. See 'joinStreamVar2'.
150 SVar { outputQueue :: IORef ([ChildEvent a], Int)
151 , doorBell :: MVar () -- signal the consumer about output
152 , siren :: MVar () -- hooter for workers to begin work
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.
236serial :: 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------------------------------------------------------------------------------
260wSerial :: Stream m a -> Stream m a -> Stream m a
261wSerial m1 m2 = Stream $ \_ stp sng yld -> do
262 let stop = (runStream m2) Nothing stp sng yld
264 yield a r = yld a (wSerial 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 MonadAsync m = (MonadIO m, MonadBaseControl IO m, MonadThrow m)
278-- Stolen from the async package. The perf improvement is modest, 2% on a
279-- thread heavy benchmark (parallel composition using noop computations).
280-- A version of forkIO that does not include the outer exception
281-- handler: saves a bit of time when we will be installing our own
283{-# INLINE rawForkIO #-}
284rawForkIO :: IO () -> IO ThreadId
285rawForkIO action = IO $ \ s ->
286 case (fork# action s) of (# s1, tid #) -> (# s1, ThreadId tid #)
289doFork :: MonadBaseControl IO m
291 -> (SomeException -> m ())
293doFork action exHandler =
294 EL.mask $ \restore ->
295 liftBaseWith $ \runInIO -> rawForkIO $ do
296 _ <- runInIO $ EL.catch (restore action) exHandler
297 -- XXX restore state here?
300-- XXX exception safety of all atomic/MVar operations
302-- TBD Each worker can have their own queue and the consumer can empty one
303-- queue at a time, that way contention can be reduced.
305send :: MonadIO m => SVar m a -> ChildEvent a -> m ()
306send sv msg = liftIO $ do
307 len <- atomicModifyIORefCAS (outputQueue sv) $ \(es, n) ->
308 ((msg : es, n + 1), n)
309 if (len <= 0) then do
310 -- XXX need a memory barrier? The wake up must happen only after the
311 -- store has finished otherwise we can have lost wakeup problems.
313 -- Since multiple workers can try this at the same time, it is possible
314 -- that we may put a spurious MVar after the consumer has already seen
315 -- the output. But that's harmless, at worst it may cause the consumer
316 -- to read the queue again and find it empty.
317 -- The important point is that the consumer is guaranteed to receive a
318 -- doorbell if something was added to the queue after it empties it.
319 void $ tryPutMVar (doorBell sv) ()
320 -- The first worker who notices the output queue was emptied puts the
322 void $ tryTakeMVar (siren sv)
323 else if (len + 1 >= 1500) then do
324 -- We are guaranteed to receive the siren if the consumer reads the
325 -- queue because the consumer puts the siren on before reading the
328 -- We may get the siren between the siren being swicthed on and the
329 -- queue getting read but that's harmless, at amost everyone will go
330 -- back to work and will have to sleep again if queue was still not
333 -- If even before a worker could read the MVar, the queue gets emptied
334 -- and another worker queuing on it switches off the siren, then we may
335 -- sleep here. In that case we are guaranteed to be woken up on the
336 -- next siren. The next siren is guaranteed as we send a doorbell
337 -- before switching off the siren, and the consumer switches on the
338 -- siren after receiving the doorbell.
342{-# INLINE sendStop #-}
343sendStop :: MonadIO m => SVar m a -> m ()
344sendStop sv = liftIO myThreadId >>= \tid -> send sv (ChildStop tid Nothing)
346-- Note: Left associated compositions can grow this queue to a large size
347{-# INLINE enqueueLIFO #-}
348enqueueLIFO :: IORef [Stream m a] -> Stream m a -> IO ()
349enqueueLIFO q m = atomicModifyIORefCAS_ q $ \ ms -> m : ms
351runqueueLIFO :: MonadIO m => SVar m a -> IORef [Stream m a] -> m ()
352runqueueLIFO sv q = run
359 Nothing -> sendStop sv
360 Just m -> (runStream m) (Just sv) run single yield
362 sendit a = send sv (ChildYield a)
363 single a = sendit a >> run
364 yield a r = sendit a >> (runStream r) (Just sv) run single yield
366 dequeue = liftIO $ atomicModifyIORefCAS q $ \case
368 x : xs -> (xs, Just x)
370{-# INLINE enqueueFIFO #-}
371enqueueFIFO :: LinkedQueue (Stream m a) -> Stream m a -> IO ()
374runqueueFIFO :: MonadIO m => SVar m a -> LinkedQueue (Stream m a) -> m ()
375runqueueFIFO sv q = run
382 Nothing -> sendStop sv
383 Just m -> (runStream m) (Just sv) run single yield
385 dequeue = liftIO $ tryPopR q
386 sendit a = send sv (ChildYield a)
387 single a = sendit a >> run
388 yield a r = sendit a >> liftIO (enqueueFIFO q r) >> run
391runOne :: MonadIO m => SVar m a -> Stream m a -> m ()
392runOne sv m = (runStream m) (Just sv) stop single yield
397 sendit a = send sv (ChildYield a)
398 single a = sendit a >> stop
399 yield a r = sendit a >> runOne sv r
401-- Thread tracking is needed for two reasons:
403-- 1) Killing threads on exceptions. Threads may not be left to go away by
404-- themselves because they may run for significant times before going away or
405-- worse they may be stuck in IO and never go away.
407-- 2) To know when all threads are done and the stream has ended.
409{-# NOINLINE addThread #-}
410addThread :: MonadIO m => SVar m a -> ThreadId -> m ()
412 liftIO $ modifyIORef (runningThreads sv) (S.insert tid)
415{-# INLINE delThread #-}
416delThread :: MonadIO m => SVar m a -> ThreadId -> m ()
418 liftIO $ modifyIORef (runningThreads sv) $ (\s -> S.delete tid s)
421-- If present then delete else add. This takes care of out of order add and
422-- delete i.e. a delete arriving before we even added a thread.
423-- This occurs when the forked thread is done even before the 'addThread' right
424-- after the fork is executed.
425{-# INLINE modifyThread #-}
426modifyThread :: MonadIO m => SVar m a -> ThreadId -> m ()
427modifyThread sv tid = do
428 changed <- liftIO $ atomicModifyIORefCAS (runningThreads sv) $ \old ->
429 if (S.member tid old)
430 then let new = (S.delete tid old) in (new, new)
431 else let new = (S.insert tid old) in (new, old)
433 then liftIO $ void $ tryPutMVar (doorBell sv) ()
436-- | This is safe even if we are adding more threads concurrently because if
437-- a child thread is adding another thread then anyway 'runningThreads' will
439{-# INLINE allThreadsDone #-}
440allThreadsDone :: MonadIO m => SVar m a -> m Bool
441allThreadsDone sv = liftIO $ S.null <$> readIORef (runningThreads sv)
443{-# NOINLINE handleChildException #-}
444handleChildException :: MonadIO m => SVar m a -> SomeException -> m ()
445handleChildException sv e = do
446 tid <- liftIO myThreadId
447 send sv (ChildStop tid (Just e))
449{-# NOINLINE pushWorker #-}
450pushWorker :: MonadAsync m => SVar m a -> m ()
452 liftIO $ atomicModifyIORefCAS_ (activeWorkers sv) $ \n -> n + 1
453 doFork (runqueue sv) (handleChildException sv) >>= addThread sv
455-- | In contrast to pushWorker which always happens only from the consumer
456-- thread, a pushWorkerPar can happen concurrently from multiple threads on the
457-- producer side. So we need to use a thread safe modification of
458-- runningThreads. Alternatively, we can use a CreateThread event to avoid
459-- using a CAS based modification.
460{-# NOINLINE pushWorkerPar #-}
461pushWorkerPar :: MonadAsync m => SVar m a -> Stream m a -> m ()
463 doFork (runOne sv m) (handleChildException sv) >>= modifyThread sv
465-- XXX When the queue is LIFO we can put a limit on the number of dispatches.
466-- Also, if a worker blocks on the output queue we can decide if we want to
467-- block or make it go away entirely, depending on the number of workers and
468-- the type of the queue.
469{-# INLINE sendWorkerWait #-}
470sendWorkerWait :: MonadAsync m => SVar m a -> m ()
471sendWorkerWait sv = do
472 -- When there is no output seen we dispatch more workers to help out if
473 -- there is work pending in the work queue. But we wait a little while
474 -- and check the output again so that we are not too aggressive.
475 -- If there is no output pending to process and there is no worker to be
476 -- sent then we block, so that we do not keep looping fruitlessly.
478 liftIO $ threadDelay 200
479 (_, n) <- liftIO $ readIORef (outputQueue sv)
481 done <- queueEmpty sv
484 cnt <- liftIO $ readIORef $ activeWorkers sv
489 else liftIO $ takeMVar (doorBell sv)
490 else liftIO $ takeMVar (doorBell sv)
492-- | Pull a stream from an SVar.
493{-# NOINLINE fromStreamVar #-}
494fromStreamVar :: MonadAsync m => SVar m a -> Stream m a
495fromStreamVar sv = Stream $ \_ stp sng yld -> do
496 let SVarStyle _ sched = svarStyle sv
498 then liftIO $ takeMVar (doorBell sv)
500 res <- liftIO $ tryTakeMVar (doorBell sv)
501 when (isNothing res) $ sendWorkerWait sv
503 void $ liftIO $ tryPutMVar (siren sv) ()
504 (list, _) <- liftIO $ atomicModifyIORefCAS (outputQueue sv) $ \x -> (([],0), x)
505 -- Reversing the output is important to guarantee that we process the
506 -- outputs in the same order as they were generated by the constituent
508 (runStream $ processEvents (reverse list)) Nothing stp sng yld
512 handleException e tid = do
513 liftIO $ atomicModifyIORefCAS_ (activeWorkers sv) $ \n -> n - 1
515 -- XXX implement kill async exception handling
516 -- liftIO $ readIORef (runningThreads sv) >>= mapM_ killThread
519 {-# INLINE processEvents #-}
520 processEvents [] = Stream $ \_ stp sng yld -> do
521 done <- allThreadsDone sv
523 then (runStream (fromStreamVar sv)) Nothing stp sng yld
526 processEvents (ev : es) = Stream $ \_ stp sng yld -> do
527 let continue = (runStream (processEvents es)) Nothing stp sng yld
528 yield a = yld a (processEvents es)
531 ChildYield a -> yield a
535 let active = activeWorkers sv
536 liftIO $ atomicModifyIORefCAS_ active $ \n -> n - 1
537 modifyThread sv tid >> continue
538 Just ex -> handleException ex tid
540getFifoSVar :: MonadIO m => SVarStyle -> IO (SVar m a)
541getFifoSVar ctype = do
542 outQ <- newIORef ([], 0)
543 outQMv <- newEmptyMVar
544 hooter <- newEmptyMVar
546 running <- newIORef S.empty
549 SVar { outputQueue = outQ
552 , runningThreads = running
553 , runqueue = runqueueFIFO sv q
555 , queueEmpty = liftIO $ nullQ q
557 , activeWorkers = active
561getLifoSVar :: MonadIO m => SVarStyle -> IO (SVar m a)
562getLifoSVar ctype = do
563 outQ <- newIORef ([], 0)
564 outQMv <- newEmptyMVar
565 hooter <- newEmptyMVar
567 running <- newIORef S.empty
569 let checkEmpty = null <$> liftIO (readIORef q)
571 SVar { outputQueue = outQ
574 , runningThreads = running
575 , runqueue = runqueueLIFO sv q
576 , enqueue = enqueueLIFO q
577 , queueEmpty = checkEmpty
579 , activeWorkers = active
583getParSVar :: SVarStyle -> IO (SVar m a)
585 outQ <- newIORef ([], 0)
586 outQMv <- newEmptyMVar
587 hooter <- newEmptyMVar
589 running <- newIORef S.empty
591 SVar { outputQueue = outQ
594 , runningThreads = running
595 , runqueue = undefined
596 , enqueue = undefined
597 , queueEmpty = undefined
599 , activeWorkers = active
603-- | Create a new empty SVar.
604newEmptySVar :: MonadAsync m => SVarStyle -> m (SVar m a)
605newEmptySVar style = do
608 SVarStyle _ FIFO -> getFifoSVar style
609 SVarStyle _ LIFO -> getLifoSVar style
610 SVarStyle _ Par -> getParSVar style
612-- | Create a new SVar and enqueue one stream computation on it.
613newStreamVar1 :: MonadAsync m => SVarStyle -> Stream m a -> m (SVar m a)
614newStreamVar1 style m = do
615 sv <- newEmptySVar style
616 -- Note: We must have all the work on the queue before sending the
617 -- pushworker, otherwise the pushworker may exit before we even get a
619 liftIO $ (enqueue sv) m
623-- | Create a new SVar and enqueue two stream computations on it.
624newStreamVar2 :: MonadAsync m
625 => SVarStyle -> Stream m a -> Stream m a -> m (SVar m a)
626newStreamVar2 style m1 m2 = do
627 -- Note: We must have all the work on the queue before sending the
628 -- pushworker, otherwise the pushworker may exit before we even get a
632 SVarStyle _ FIFO -> do
633 c <- getFifoSVar style
634 (enqueue c) m1 >> (enqueue c) m2
636 SVarStyle _ LIFO -> do
637 c <- getLifoSVar style
638 (enqueue c) m2 >> (enqueue c) m1
640 SVarStyle _ Par -> undefined
644-- | Write a stream to an 'SVar' in a non-blocking manner. The stream can then
645-- be read back from the SVar using 'fromSVar'.
646toStreamVar :: MonadAsync m => SVar m a -> Stream m a -> m ()
648 liftIO $ (enqueue sv) m
649 done <- allThreadsDone sv
650 -- XXX This is safe only when called from the consumer thread or when no
651 -- consumer is present. There may be a race if we are not running in the
653 when done $ pushWorker sv
655------------------------------------------------------------------------------
656-- Running streams concurrently
657------------------------------------------------------------------------------
659-- Concurrency rate control.
661-- Our objective is to create more threads on demand if the consumer is running
662-- faster than us. As soon as we encounter a concurrent composition we create a
663-- push pull pair of threads. We use an SVar for communication between the
664-- consumer, pulling from the SVar and the producer who is pushing to the SVar.
665-- The producer creates more threads if the SVar drains and becomes empty, that
666-- is the consumer is running faster.
668-- XXX Note 1: This mechanism can be problematic if the initial production
669-- latency is high, we may end up creating too many threads. So we need some
670-- way to monitor and use the latency as well. Having a limit on the dispatches
671-- (programmer controlled) may also help.
673-- TBD Note 2: We may want to run computations at the lower level of the
674-- composition tree serially even when they are composed using a parallel
675-- combinator. We can use 'splice' in place of 'async' and 'cosplice' in
676-- place of 'coParAhead'. If we find that an SVar immediately above a computation
677-- gets drained empty we can switch to parallelizing the computation. For that
678-- we can use a state flag to fork the rest of the computation at any point of
679-- time inside the Monad bind operation if the consumer is running at a faster
682-- TBD Note 3: the binary operation ('parallel') composition allows us to
683-- dispatch a chunkSize of only 1. If we have to dispatch in arbitrary
684-- chunksizes we will need to compose the parallel actions using a data
685-- constructor (A Free container) instead so that we can divide it in chunks of
686-- arbitrary size before dispatching. If the stream is composed of
687-- hierarchically composed grains of different sizes then we can always switch
688-- to a desired granularity depending on the consumer speed.
690-- TBD Note 4: for pure work (when we are not in the IO monad) we can divide it
691-- into just the number of CPUs.
693-- | Join two computations on the currently running 'SVar' queue for concurrent
694-- execution. When we are using parallel composition, an SVar is passed around
695-- as a state variable. We try to schedule a new parallel computation on the
696-- SVar passed to us. The first time, when no SVar exists, a new SVar is
697-- created. Subsequently, 'joinStreamVar2' may get called when a computation
698-- already scheduled on the SVar is further evaluated. For example, when (a
699-- `parallel` b) is evaluated it calls a 'joinStreamVar2' to put 'a' and 'b' on
700-- the current scheduler queue.
702-- The 'SVarStyle' required by the current composition context is passed as one
703-- of the parameters. If the scheduling and composition style of the new
704-- computation being scheduled is different than the style of the current SVar,
705-- then we create a new SVar and schedule it on that. The newly created SVar
706-- joins as one of the computations on the current SVar queue.
708-- Cases when we need to switch to a new SVar:
710-- * (x `parallel` y) `parallel` (t `parallel` u) -- all of them get scheduled on the same SVar
711-- * (x `parallel` y) `parallel` (t `async` u) -- @t@ and @u@ get scheduled on a new child SVar
712-- because of the scheduling policy change.
713-- * if we 'adapt' a stream of type 'async' to a stream of type
714-- 'Parallel', we create a new SVar at the transitioning bind.
715-- * When the stream is switching from disjunctive composition to conjunctive
716-- composition and vice-versa we create a new SVar to isolate the scheduling
719{-# INLINE joinStreamVar2 #-}
720joinStreamVar2 :: MonadAsync m
721 => SVarStyle -> Stream m a -> Stream m a -> Stream m a
722joinStreamVar2 style m1 m2 = Stream $ \svr stp sng yld ->
724 Just sv | svarStyle sv == style ->
725 liftIO ((enqueue sv) m2) >> (runStream m1) svr stp sng yld
727 sv <- newStreamVar1 style (concurrently m1 m2)
728 (runStream (fromStreamVar sv)) Nothing stp sng yld
730 concurrently ma mb = Stream $ \svr stp sng yld -> do
731 liftIO $ (enqueue (fromJust svr)) mb
732 (runStream ma) svr stp sng yld
734{-# INLINE joinStreamVarPar #-}
735joinStreamVarPar :: MonadAsync m
736 => SVarStyle -> Stream m a -> Stream m a -> Stream m a
737joinStreamVarPar style m1 m2 = Stream $ \svr stp sng yld ->
739 Just sv | svarStyle sv == style -> do
740 pushWorkerPar sv m1 >> (runStream m2) svr stp sng yld
742 sv <- newEmptySVar style
745 (runStream (fromStreamVar sv)) Nothing stp sng yld
747------------------------------------------------------------------------------
748-- Semigroup and Monoid style compositions for parallel actions
749------------------------------------------------------------------------------
752async :: MonadAsync m => Stream m a -> Stream m a -> Stream m a
753async = joinStreamVar2 (SVarStyle Disjunction LIFO)
756wAsync :: MonadAsync m => Stream m a -> Stream m a -> Stream m a
757wAsync = joinStreamVar2 (SVarStyle Disjunction FIFO)
759{-# INLINE parallel #-}
760parallel :: MonadAsync m => Stream m a -> Stream m a -> Stream m a
761parallel = joinStreamVarPar (SVarStyle Disjunction Par)
763-------------------------------------------------------------------------------
764-- Functor instace is the same for all types
765-------------------------------------------------------------------------------
767instance Monad m => Functor (Stream m) where
768 fmap f m = Stream $ \_ stp sng yld ->
770 yield a r = yld (f a) (fmap f r)
771 in (runStream m) Nothing stp single yield
773------------------------------------------------------------------------------
774-- Alternative & MonadPlus
775------------------------------------------------------------------------------
777alt :: Stream m a -> Stream m a -> Stream m a
778alt m1 m2 = Stream $ \_ stp sng yld ->
779 let stop = runStream m2 Nothing stp sng yld
780 in runStream m1 Nothing stop sng yld
782------------------------------------------------------------------------------
784------------------------------------------------------------------------------
786zipWith :: (a -> b -> c) -> Stream m a -> Stream m b -> Stream m c
787zipWith f m1 m2 = go m1 m2
789 go mx my = Stream $ \_ stp sng yld -> do
791 let single2 b = sng (f a b)
792 yield2 b rb = yld (f a b) (go ra rb)
793 in (runStream my) Nothing stp single2 yield2
794 let single1 a = merge a nil
795 yield1 a ra = merge a ra
796 (runStream mx) Nothing stp single1 yield1
798mkAsync :: MonadAsync m => Stream m a -> m (Stream m a)
799mkAsync m = newStreamVar1 (SVarStyle Disjunction LIFO) m
800 >>= return . fromStreamVar
802zipAsyncWith :: MonadAsync m
803 => (a -> b -> c) -> Stream m a -> Stream m b -> Stream m c
804zipAsyncWith f m1 m2 = Stream $ \_ stp sng yld -> do
807 (runStream (zipWith f ma mb)) Nothing stp sng yld
809-------------------------------------------------------------------------------
811-------------------------------------------------------------------------------
813instance MonadTrans Stream where
816withLocal :: MonadReader r m => (r -> r) -> Stream m a -> Stream m a
818 Stream $ \svr stp sng yld ->
819 let single = local f . sng
820 yield a r = local f $ yld a (withLocal f r)
821 in (runStream m) svr (local f stp) single yield
823-- XXX handle and test cross thread state transfer
826 => Stream m a -> (e -> Stream m a) -> Stream m a
828 Stream $ \svr stp sng yld ->
829 let run x = runStream x svr stp sng yield
830 handle r = r `catchError` \e -> run $ h e
831 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