2{-# LANGUAGE ConstraintKinds #-}
3{-# LANGUAGE FlexibleContexts #-}
4{-# LANGUAGE FlexibleInstances #-}
5{-# LANGUAGE LambdaCase #-}
6{-# LANGUAGE MagicHash #-}
7{-# LANGUAGE MultiParamTypeClasses #-}
8{-# LANGUAGE RankNTypes #-}
9{-# LANGUAGE UnboxedTuples #-}
10{-# LANGUAGE UndecidableInstances #-} -- XXX
13-- Module : Streamly.Core
14-- Copyright : (c) 2017 Harendra Kumar
17-- Maintainer : harendra.kumar@gmail.com
18-- Stability : experimental
29 -- * Construction (pure)
36 -- * Construction (monadic)
43 -- * Semigroup Style Composition
59 -- * Concurrent Stream Vars (SVars)
68import Control.Concurrent (ThreadId, myThreadId,
69 threadDelay, getNumCapabilities)
70import Control.Concurrent.MVar (MVar, newEmptyMVar,
72import Control.Exception (SomeException (..), catch, mask)
73import Control.Monad (when)
74import Control.Monad.Catch (MonadThrow, throwM)
75import Control.Monad.IO.Class (MonadIO(..))
76import Control.Monad.Trans.Class (MonadTrans (lift))
77import Control.Monad.Trans.Control (MonadBaseControl, control)
78import Data.Atomics (casIORef, readForCAS, peekTicket
79 ,atomicModifyIORefCAS_
80 ,writeBarrier,storeLoadBarrier)
81import Data.Concurrent.Queue.MichaelScott (LinkedQueue, newQ, pushL,
83import Data.Functor (void)
84import Data.Heap (Heap, Entry(..))
85import qualified Data.Heap as H
86import Data.IORef (IORef, modifyIORef, newIORef,
87 readIORef, atomicModifyIORef
92import Data.Maybe (fromJust)
93import Data.Semigroup (Semigroup(..))
95import qualified Data.Set as S
96import Prelude hiding (repeat, zipWith)
99import GHC.Conc (ThreadId(..))
100import GHC.IO (IO(..))
102-- MVar diagnostics has some overhead - around 5% on asyncly null benchmark, we
103-- can keep it on in production to debug problems quickly if and when they
104-- happen, but it may result in unexpected output when threads are left hanging
105-- until they are GCed because the consumer went away.
108import Control.Concurrent.MVar (tryTakeMVar)
109import Control.Exception (catches, throwIO, Handler(..),
110 BlockedIndefinitelyOnMVar(..),
111 BlockedIndefinitelyOnSTM(..))
112import System.IO (hPutStrLn, stderr)
115------------------------------------------------------------------------------
116-- Parent child thread communication type
117------------------------------------------------------------------------------
119-- | Events that a child thread may send to a parent thread.
122 | ChildStop ThreadId (Maybe SomeException)
124-- | Sorting out-of-turn outputs in a heap for Ahead style streams
125data AheadHeapEntry m a =
127 | AheadEntryStream (Stream m a)
129------------------------------------------------------------------------------
130-- State threaded around the monad for thread management
131------------------------------------------------------------------------------
133-- XXX use a separate data structure for each type of SVar
134-- | Identify the type of the SVar. Two computations using the same style can
135-- be scheduled on the same SVar.
137 AsyncVar -- depth first concurrent
138 | WAsyncVar -- breadth first concurrent
139 | ParallelVar -- all parallel
140 | AheadVar -- Concurrent look ahead
143-- | An SVar or a Stream Var is a conduit to the output from multiple streams
144-- running concurrently and asynchronously. An SVar can be thought of as an
145-- asynchronous IO handle. We can write any number of streams to an SVar in a
146-- non-blocking manner and then read them back at any time at any pace. The
147-- SVar would run the streams asynchronously and accumulate results. An SVar
148-- may not really execute the stream completely and accumulate all the results.
149-- However, it ensures that the reader can read the results at whatever paces
150-- it wants to read. The SVar monitors and adapts to the consumer's pace.
152-- An SVar is a mini scheduler, it has an associated runqueue that holds the
153-- stream tasks to be picked and run by a pool of worker threads. It has an
154-- associated output queue where the output stream elements are placed by the
155-- worker threads. A doorBell is used by the worker threads to intimate the
156-- consumer thread about availability of new results in the output queue. More
157-- workers are added to the SVar by 'fromStreamVar' on demand if the output
158-- produced is not keeping pace with the consumer. On bounded SVars, workers
159-- block on the output queue to provide throttling of the producer when the
160-- consumer is not pulling fast enough. The number of workers may even get
161-- reduced depending on the consuming pace.
163-- New work is enqueued either at the time of creation of the SVar or as a
164-- result of executing the parallel combinators i.e. '<|' and '<|>' when the
165-- already enqueued computations get evaluated. See 'joinStreamVarAsync'.
170 svarStyle :: SVarStyle
172 -- Shared output queue (events, length)
173 , outputQueue :: IORef ([ChildEvent a], Int)
174 , doorBell :: MVar () -- signal the consumer about output
176 -- Output synchronization mechanism for Ahead streams (Ahead and
177 -- wAhead). We maintain a heap of out of sequence ahead of time
178 -- generated outputs and the sequence number of the task that is
179 -- currently at the head of the stream. Concurrent execute ahead
180 -- tasks that have a sequence number greater than the task at the
181 -- head should add their output to the heap.
182 , outputHeap :: IORef (Heap (Entry Int (AheadHeapEntry m a))
186 -- Shared work queue (stream, seqNo)
187 , workQueue :: IORef ([Stream m a], Int)
188 , enqueue :: Stream m a -> IO ()
189 , queueEmpty :: m Bool
190 , waitingForWork :: IORef Bool
193 -- Shared, thread tracking
194 , runningThreads :: IORef (Set ThreadId)
195 , activeWorkers :: IORef Int
197 , totalDispatches :: IORef Int
198 , maxWorkers :: IORef Int
199 , maxOutQSize :: IORef Int
200 , maxHeapSize :: IORef Int
201 , maxWorkQSize :: IORef Int
206{-# NOINLINE dumpSVar #-}
207dumpSVar :: SVar m a -> IO String
210 (oqList, oqLen) <- readIORef $ outputQueue sv
211 db <- tryTakeMVar $ doorBell sv
213 if svarStyle sv == AheadVar
215 (oheap, oheapSeq) <- readIORef $ outputHeap sv
216 (wq, wqSeq) <- readIORef $ workQueue sv
217 maxHp <- readIORef $ maxHeapSize sv
219 [ "heap length = " ++ show (H.size oheap)
220 , "heap seqeunce = " ++ show oheapSeq
221 , "work queue length = " ++ show (length wq)
222 , "work queue sequence = " ++ show wqSeq
223 , "heap max size = " ++ show maxHp
227 waiting <- readIORef $ waitingForWork sv
228 rthread <- readIORef $ runningThreads sv
229 workers <- readIORef $ activeWorkers sv
230 maxWrk <- readIORef $ maxWorkers sv
231 dispatches <- readIORef $ totalDispatches sv
232 maxOq <- readIORef $ maxOutQSize sv
233 -- XXX queueEmpty should be made IO return type
236 [ "tid = " ++ show tid
237 , "style = " ++ show (svarStyle sv)
238 , "outputQueue length computed = " ++ show (length oqList)
239 , "outputQueue length maintained = " ++ show oqLen
240 , "output doorBell = " ++ show db
241 , "total dispatches = " ++ show dispatches
242 , "max workers = " ++ show maxWrk
243 , "max outQSize = " ++ show maxOq
245 ++ aheadDump ++ unlines
246 [ "waitingForWork = " ++ show waiting
247 , "running threads = " ++ show rthread
248 , "running thread count = " ++ show workers
251{-# NOINLINE mvarExcHandler #-}
252mvarExcHandler :: SVar m a -> String -> BlockedIndefinitelyOnMVar -> IO ()
253mvarExcHandler sv label e@BlockedIndefinitelyOnMVar = do
254 svInfo <- dumpSVar sv
255 hPutStrLn stderr $ label ++ " " ++ "BlockedIndefinitelyOnMVar\n" ++ svInfo
258{-# NOINLINE stmExcHandler #-}
259stmExcHandler :: SVar m a -> String -> BlockedIndefinitelyOnSTM -> IO ()
260stmExcHandler sv label e@BlockedIndefinitelyOnSTM = do
261 svInfo <- dumpSVar sv
262 hPutStrLn stderr $ label ++ " " ++ "BlockedIndefinitelyOnSTM\n" ++ svInfo
265withDBGMVar :: SVar m a -> String -> IO () -> IO ()
266withDBGMVar sv label action =
267 action `catches` [ Handler (mvarExcHandler sv label)
268 , Handler (stmExcHandler sv label)
271withDBGMVar :: SVar m a -> String -> IO () -> IO ()
272withDBGMVar _ _ action = action
275-- Slightly faster version of CAS. Gained some improvement by avoiding the use
276-- of "evaluate" because we know we do not have exceptions in fn.
277{-# INLINE atomicModifyIORefCAS #-}
278atomicModifyIORefCAS :: IORef a -> (a -> (a,b)) -> IO b
279atomicModifyIORefCAS ref fn = do
280 tkt <- readForCAS ref
286 loop _ 0 = atomicModifyIORef ref fn
288 let (new, result) = fn $ peekTicket old
289 (success, tkt) <- casIORef ref old new
292 else loop tkt (tries - 1)
294------------------------------------------------------------------------------
296------------------------------------------------------------------------------
298-- | The type 'Stream m a' represents a monadic stream of values of type 'a'
299-- constructed using actions in monad 'm'. It uses stop, singleton and yield
300-- continuations equivalent to the following direct style type:
302-- data Stream m a = Stop | Singleton a | Yield a (Stream m a)
304-- To facilitate parallel composition we maintain a local state in an SVar that
305-- is shared across and is used for synchronization of the streams being
308-- The singleton case can be expressed in terms of stop and yield but we have
309-- it as a separate case to optimize composition operations for streams with
310-- single element. We build singleton streams in the implementation of 'pure'
311-- for Applicative and Monad, and in 'lift' for MonadTrans.
315 runStream :: forall r.
316 Maybe (SVar m a) -- local state
318 -> (a -> m r) -- singleton
319 -> (a -> Stream m a -> m r) -- yield
324nil = Stream $ \_ stp _ _ -> stp
326-- | faster than consM because there is no bind.
327cons :: a -> Stream m a -> Stream m a
328cons a r = Stream $ \_ _ _ yld -> yld a r
330-- | Same as @once . return@ but may be faster because there is no bind
331singleton :: a -> Stream m a
332singleton a = Stream $ \_ _ single _ -> single a
335once :: Monad m => m a -> Stream m a
336once m = Stream $ \_ _ single _ -> m >>= single
339consM :: Monad m => m a -> Stream m a -> Stream m a
340consM m r = Stream $ \_ _ _ yld -> m >>= \a -> yld a r
342repeat :: a -> Stream m a
343repeat a = let x = cons a x in x
345------------------------------------------------------------------------------
347------------------------------------------------------------------------------
349-- | Concatenates two streams sequentially i.e. the first stream is
350-- exhausted completely before yielding any element from the second stream.
352serial :: Stream m a -> Stream m a -> Stream m a
355 go (Stream m) = Stream $ \_ stp sng yld ->
356 let stop = (runStream m2) Nothing stp sng yld
358 yield a r = yld a (go r)
359 in m Nothing stop single yield
361instance Semigroup (Stream m a) where
364------------------------------------------------------------------------------
366------------------------------------------------------------------------------
368instance Monoid (Stream m a) where
372------------------------------------------------------------------------------
374------------------------------------------------------------------------------
376{-# INLINE wSerial #-}
377wSerial :: Stream m a -> Stream m a -> Stream m a
378wSerial m1 m2 = Stream $ \_ stp sng yld -> do
379 let stop = (runStream m2) Nothing stp sng yld
381 yield a r = yld a (wSerial m2 r)
382 (runStream m1) Nothing stop single yield
384------------------------------------------------------------------------------
385-- Spawning threads and collecting result in streamed fashion
386------------------------------------------------------------------------------
388-- | A monad that can perform concurrent or parallel IO operations. Streams
389-- that can be composed concurrently require the underlying monad to be
393type MonadAsync m = (MonadIO m, MonadBaseControl IO m, MonadThrow m)
395-- Stolen from the async package. The perf improvement is modest, 2% on a
396-- thread heavy benchmark (parallel composition using noop computations).
397-- A version of forkIO that does not include the outer exception
398-- handler: saves a bit of time when we will be installing our own
400{-# INLINE rawForkIO #-}
401rawForkIO :: IO () -> IO ThreadId
402rawForkIO action = IO $ \ s ->
403 case (fork# action s) of (# s1, tid #) -> (# s1, ThreadId tid #)
406doFork :: MonadBaseControl IO m
408 -> (SomeException -> IO ())
410doFork action exHandler =
411 control $ \runInIO ->
412 mask $ \restore -> do
413 tid <- rawForkIO $ catch (restore $ void $ runInIO action)
417-- XXX exception safety of all atomic/MVar operations
419-- TBD Each worker can have their own queue and the consumer can empty one
420-- queue at a time, that way contention can be reduced.
425-- | This function is used by the producer threads to queue output for the
426-- consumer thread to consume. Returns whether the queue has more space.
428send :: SVar m a -> ChildEvent a -> IO Bool
430 len <- atomicModifyIORefCAS (outputQueue sv) $ \(es, n) ->
431 ((msg : es, n + 1), n)
433 -- The wake up must happen only after the store has finished otherwise
434 -- we can have lost wakeup problems.
436 -- Since multiple workers can try this at the same time, it is possible
437 -- that we may put a spurious MVar after the consumer has already seen
438 -- the output. But that's harmless, at worst it may cause the consumer
439 -- to read the queue again and find it empty.
440 -- The important point is that the consumer is guaranteed to receive a
441 -- doorbell if something was added to the queue after it empties it.
442 void $ tryPutMVar (doorBell sv) ()
443 return (len < maxOutputQLen)
445{-# NOINLINE sendStop #-}
446sendStop :: SVar m a -> IO ()
448 liftIO $ atomicModifyIORefCAS_ (activeWorkers sv) $ \n -> n - 1
449 myThreadId >>= \tid -> void $ send sv (ChildStop tid Nothing)
451-------------------------------------------------------------------------------
453-------------------------------------------------------------------------------
455-- Note: For purely right associated expressions this queue should have at most
456-- one element. It grows to more than one when we have left associcated
457-- expressions. Large left associated compositions can grow this to a
459{-# INLINE enqueueLIFO #-}
460enqueueLIFO :: SVar m a -> IORef [Stream m a] -> Stream m a -> IO ()
461enqueueLIFO sv q m = do
462 atomicModifyIORefCAS_ q $ \ms -> m : ms
464 w <- readIORef $ waitingForWork sv
466 -- Note: the sequence of operations is important for correctness here.
467 -- We need to set the flag to false strictly before sending the
468 -- doorBell, otherwise the doorBell may get processed too early and
469 -- then we may set the flag to False to later making the consumer lose
470 -- the flag, even without receiving a doorBell.
471 atomicModifyIORefCAS_ (waitingForWork sv) (const False)
472 void $ tryPutMVar (doorBell sv) ()
474runqueueLIFO :: MonadIO m => SVar m a -> IORef [Stream m a] -> m ()
475runqueueLIFO sv q = run
482 Nothing -> liftIO $ sendStop sv
483 Just m -> (runStream m) (Just sv) run single yield
486 res <- liftIO $ send sv (ChildYield a)
487 if res then run else liftIO $ sendStop sv
489 res <- liftIO $ send sv (ChildYield a)
491 then (runStream r) (Just sv) run single yield
492 else liftIO $ enqueueLIFO sv q r >> sendStop sv
494 dequeue = liftIO $ atomicModifyIORefCAS q $ \case
496 x : xs -> (xs, Just x)
498-------------------------------------------------------------------------------
500-------------------------------------------------------------------------------
502-- XXX we can use the Ahead style sequence/heap mechanism to make the best
503-- effort to always try to finish the streams on the left side of an expression
504-- first as long as possible.
506{-# INLINE enqueueFIFO #-}
507enqueueFIFO :: SVar m a -> LinkedQueue (Stream m a) -> Stream m a -> IO ()
508enqueueFIFO sv q m = do
511 w <- readIORef $ waitingForWork sv
513 -- Note: the sequence of operations is important for correctness here.
514 -- We need to set the flag to false strictly before sending the
515 -- doorBell, otherwise the doorBell may get processed too early and
516 -- then we may set the flag to False to later making the consumer lose
517 -- the flag, even without receiving a doorBell.
518 atomicModifyIORefCAS_ (waitingForWork sv) (const False)
519 void $ tryPutMVar (doorBell sv) ()
521runqueueFIFO :: MonadIO m => SVar m a -> LinkedQueue (Stream m a) -> m ()
522runqueueFIFO sv q = run
529 Nothing -> liftIO $ sendStop sv
530 Just m -> (runStream m) (Just sv) run single yield
532 dequeue = liftIO $ tryPopR q
534 res <- liftIO $ send sv (ChildYield a)
535 if res then run else liftIO $ sendStop sv
537 res <- liftIO $ send sv (ChildYield a)
538 liftIO (enqueueFIFO sv q r)
539 if res then run else liftIO $ sendStop sv
541-------------------------------------------------------------------------------
543-------------------------------------------------------------------------------
545{-# NOINLINE runOne #-}
546runOne :: MonadIO m => SVar m a -> Stream m a -> m ()
547runOne sv m = (runStream m) (Just sv) stop single yield
551 stop = liftIO $ sendStop sv
552 sendit a = liftIO $ send sv (ChildYield a)
553 single a = sendit a >> stop
554 -- XXX there is no flow control in parallel case. We should perhaps use a
555 -- queue and queue it back on that and exit the thread when the outputQueue
556 -- overflows. Parallel is dangerous because it can accumulate unbounded
557 -- output in the buffer.
558 yield a r = void (sendit a) >> runOne sv r
560-------------------------------------------------------------------------------
562-------------------------------------------------------------------------------
564-- Lookahead streams can execute multiple tasks concurrently, ahead of time,
565-- but always serve them in the same order as they appear in the stream. To
566-- implement lookahead streams efficiently we assign a sequence number to each
567-- task when the task is picked up for execution. When the task finishes, the
568-- output is tagged with the same sequence number and we rearrange the outputs
569-- in sequence based on that number.
571-- To explain the mechanism imagine that the current task at the head of the
572-- stream has a "token" to yield to the outputQueue. The ownership of the token
573-- is determined by the current sequence number is maintained in outputHeap.
574-- Sequence number is assigned when a task is queued. When a thread dequeues a
575-- task it picks up the sequence number as well and when the output is ready it
576-- uses the sequence number to queue the output to the outputQueue.
578-- The thread with current sequence number sends the output directly to the
579-- outputQueue. Other threads push the output to the outputHeap. When the task
580-- being queued on the heap is a stream of many elements we evaluate only the
581-- first element and keep the rest of the unevaluated computation in the heap.
582-- When such a task gets the "token" for outputQueue it evaluates and directly
583-- yields all the elements to the outputQueue without checking for the
586-- Note that no two outputs in the heap can have the same sequence numbers and
587-- therefore we do not need a stable heap. We have also separated the buffer
588-- for the current task (outputQueue) and the pending tasks (outputHeap) so
589-- that the pending tasks cannot interfere with the current task. Note that for
590-- a single task just the outputQueue is enough and for the case of many
591-- threads just a heap is good enough. However we balance between these two
592-- cases, so that both are efficient.
594-- For bigger streams it may make sense to have separate buffers for each
595-- stream. However, for singleton streams this may become inefficient. However,
596-- if we do not have separate buffers, then the streams that come later in
597-- sequence may hog the buffer, hindering the streams that are ahead. For this
598-- reason we have a single element buffer limitation for the streams being
599-- executed in advance.
601-- This scheme works pretty efficiently with less than 40% extra overhead
602-- compared to the Async streams where we do not have any kind of sequencing of
603-- the outputs. It is especially devised so that we are most efficient when we
604-- have short tasks and need just a single thread. Also when a thread yields
605-- many items it can hold lockfree access to the outputQueue and do it
608-- XXX Maybe we can start the ahead threads at a lower cpu and IO priority so
609-- that they do not hog the resources and hinder the progress of the threads in
612-- Left associated ahead expressions are expensive. We start a new SVar for
613-- each left associative expression. The queue is used only for right
614-- associated expression, we queue the right expression and execute the left.
615-- Thererefore the queue never has more than on item in it.
616{-# INLINE enqueueAhead #-}
617enqueueAhead :: SVar m a -> IORef ([Stream m a], Int) -> Stream m a -> IO ()
618enqueueAhead sv q m = do
619 atomicModifyIORefCAS_ q $ \ case
620 ([], n) -> ([m], n + 1) -- increment sequence
621 _ -> error "not empty"
623 w <- readIORef $ waitingForWork sv
625 -- Note: the sequence of operations is important for correctness here.
626 -- We need to set the flag to false strictly before sending the
627 -- doorBell, otherwise the doorBell may get processed too early and
628 -- then we may set the flag to False to later making the consumer lose
629 -- the flag, even without receiving a doorBell.
630 atomicModifyIORefCAS_ (waitingForWork sv) (const False)
631 void $ tryPutMVar (doorBell sv) ()
633-- Normally the thread that has the token should never go away. The token gets
634-- handed over to another thread, but someone or the other has the token at any
635-- point of time. But if the task that has the token finds that the outputQueue
636-- is full, in that case it can go away without even handing over the token to
637-- another thread. In that case it sets the nextSequence number in the heap its
638-- own sequence number before going away. To handle this case, any task that
639-- does not have the token tries to dequeue from the heap first before
640-- dequeuing from the work queue. If it finds that the task at the top of the
641-- heap is the one that owns the current sequence number then it grabs the
642-- token and starts with that.
644-- XXX instead of queueing just the head element and the remaining computation
645-- on the heap, evaluate as many as we can and place them on the heap. But we
646-- need to give higher priority to the lower sequence numbers so that lower
647-- priority tasks do not fill up the heap making higher priority tasks block
648-- due to full heap. Maybe we can have a weighted space for them in the heap.
649-- The weight is inversely proportional to the sequence number.
651-- XXX review for livelock
653runqueueAhead :: MonadIO m => SVar m a -> IORef ([Stream m a], Int) -> m ()
654runqueueAhead sv q = runHeap
660 toHeap seqNo ent = do
661 hp <- liftIO $ atomicModifyIORefCAS (outputHeap sv) $ \(h, snum) ->
662 ((H.insert (Entry seqNo ent) h, snum), h)
663 if H.size hp <= maxHeap
665 else liftIO $ sendStop sv
667 singleToHeap seqNo a = toHeap seqNo (AheadEntryPure a)
668 yieldToHeap seqNo a r = toHeap seqNo (AheadEntryStream (a `cons` r))
670 singleOutput seqNo a = do
671 continue <- liftIO $ send sv (ChildYield a)
673 then runQueueToken seqNo
675 atomicModifyIORefCAS_ (outputHeap sv) $ \(h, _) -> (h, seqNo + 1)
678 yieldOutput seqNo a r = do
679 continue <- liftIO $ send sv (ChildYield a)
681 then (runStream r) (Just sv) (runQueueToken seqNo)
685 atomicModifyIORefCAS_ (outputHeap sv) $ \(h, _) ->
686 (H.insert (Entry seqNo (AheadEntryStream r)) h, seqNo)
689 {-# INLINE runQueueToken #-}
690 runQueueToken prevSeqNo = do
694 liftIO $ atomicModifyIORefCAS_ (outputHeap sv) $ \(h, _) ->
697 Just (m, seqNo) -> do
698 if seqNo == prevSeqNo + 1
700 (runStream m) (Just sv) (runQueueToken seqNo)
704 liftIO $ atomicModifyIORefCAS_ (outputHeap sv) $ \(h, _) ->
706 (runStream m) (Just sv) runHeap
713 Just (m, seqNo) -> do
716 (runStream m) (Just sv) (runQueueToken seqNo)
720 (runStream m) (Just sv) runHeap
724 {-# NOINLINE runHeap #-}
728 maxHp <- readIORef (maxHeapSize sv)
729 (hp, _) <- readIORef (outputHeap sv)
730 when (H.size hp > maxHp) $ writeIORef (maxHeapSize sv) (H.size hp)
732 ent <- liftIO $ dequeueFromHeap (outputHeap sv)
735 done <- queueEmpty sv
737 then liftIO $ sendStop sv
739 Just (Entry seqNo hent) -> do
741 AheadEntryPure a -> singleOutput seqNo a
742 AheadEntryStream r ->
743 (runStream r) (Just sv) (runQueueToken seqNo)
747 dequeue = liftIO $ do
748 atomicModifyIORefCAS q $ \case
749 ([], n) -> (([], n), Nothing)
750 (x : [], n) -> (([], n), Just (x, n))
751 _ -> error "more than one item on queue"
754 :: IORef (Heap (Entry Int (AheadHeapEntry m a)), Int)
755 -> IO (Maybe (Entry Int (AheadHeapEntry m a)))
756 dequeueFromHeap hpRef = do
757 atomicModifyIORefCAS hpRef $ \hp@(h, snum) -> do
760 Nothing -> (hp, Nothing)
761 Just (ent@(Entry seqNo _ev), hp') ->
763 then ((hp', seqNo), Just ent)
766-------------------------------------------------------------------------------
768-------------------------------------------------------------------------------
770-- XXX To be implemented. Use a linked queue like WAsync and put back the
771-- remaining computation at the back of the queue instead of the heap, and
772-- increment the sequence number.
774-- Thread tracking is needed for two reasons:
776-- 1) Killing threads on exceptions. Threads may not be left to go away by
777-- themselves because they may run for significant times before going away or
778-- worse they may be stuck in IO and never go away.
780-- 2) To know when all threads are done and the stream has ended.
782{-# NOINLINE addThread #-}
783addThread :: MonadIO m => SVar m a -> ThreadId -> m ()
785 liftIO $ modifyIORef (runningThreads sv) (S.insert tid)
787-- This is cheaper than modifyThread because we do not have to send a doorBell
788-- This can make a difference when more workers are being dispatched.
789{-# INLINE delThread #-}
790delThread :: MonadIO m => SVar m a -> ThreadId -> m ()
792 liftIO $ modifyIORef (runningThreads sv) $ (\s -> S.delete tid s)
794-- If present then delete else add. This takes care of out of order add and
795-- delete i.e. a delete arriving before we even added a thread.
796-- This occurs when the forked thread is done even before the 'addThread' right
797-- after the fork gets a chance to run.
798{-# INLINE modifyThread #-}
799modifyThread :: MonadIO m => SVar m a -> ThreadId -> m ()
800modifyThread sv tid = do
801 changed <- liftIO $ atomicModifyIORefCAS (runningThreads sv) $ \old ->
802 if (S.member tid old)
803 then let new = (S.delete tid old) in (new, new)
804 else let new = (S.insert tid old) in (new, old)
808 void $ tryPutMVar (doorBell sv) ()
811-- | This is safe even if we are adding more threads concurrently because if
812-- a child thread is adding another thread then anyway 'runningThreads' will
814{-# INLINE allThreadsDone #-}
815allThreadsDone :: MonadIO m => SVar m a -> m Bool
816allThreadsDone sv = liftIO $ S.null <$> readIORef (runningThreads sv)
818{-# NOINLINE handleChildException #-}
819handleChildException :: SVar m a -> SomeException -> IO ()
820handleChildException sv e = do
822 void $ send sv (ChildStop tid (Just e))
825recordMaxWorkers :: MonadIO m => SVar m a -> m ()
826recordMaxWorkers sv = liftIO $ do
827 active <- readIORef (activeWorkers sv)
828 maxWrk <- readIORef (maxWorkers sv)
829 when (active > maxWrk) $ writeIORef (maxWorkers sv) active
830 modifyIORef (totalDispatches sv) (+1)
833{-# NOINLINE pushWorker #-}
834pushWorker :: MonadAsync m => SVar m a -> m ()
836 liftIO $ atomicModifyIORefCAS_ (activeWorkers sv) $ \n -> n + 1
840 doFork (runqueue sv) (handleChildException sv) >>= addThread sv
842-- | In contrast to pushWorker which always happens only from the consumer
843-- thread, a pushWorkerPar can happen concurrently from multiple threads on the
844-- producer side. So we need to use a thread safe modification of
845-- runningThreads. Alternatively, we can use a CreateThread event to avoid
846-- using a CAS based modification.
847{-# NOINLINE pushWorkerPar #-}
848pushWorkerPar :: MonadAsync m => SVar m a -> Stream m a -> m ()
849pushWorkerPar sv m = do
850 -- We do not use activeWorkers in case of ParallelVar but still there is no
851 -- harm in maintaining it correctly.
853 liftIO $ atomicModifyIORefCAS_ (activeWorkers sv) $ \n -> n + 1
856 doFork (runOne sv m) (handleChildException sv) >>= modifyThread sv
858{-# INLINE workDone #-}
859workDone :: MonadIO m => SVar m a -> m Bool
862 if (svarStyle sv == AheadVar)
864 (hp, _) <- liftIO $ readIORef (outputHeap sv)
865 return (H.size hp <= 0)
867 queueDone <- queueEmpty sv
868 return $ queueDone && heapDone
873dispatchWorker :: MonadAsync m => SVar m a -> m ()
874dispatchWorker sv = do
877 -- Note that the worker count is only decremented during event
878 -- processing in fromStreamVar and therefore it is safe to read and
879 -- use it without a lock.
880 cnt <- liftIO $ readIORef $ activeWorkers sv
881 -- Note that we may deadlock if the previous workers (tasks in the
882 -- stream) wait/depend on the future workers (tasks in the stream)
883 -- executing. In that case we should either configure the maxWorker
884 -- count to higher or use parallel style instead of ahead or async
886 when (cnt < maxWorkerLimit) $ pushWorker sv
888{-# NOINLINE sendWorkerWait #-}
889sendWorkerWait :: MonadAsync m => SVar m a -> m ()
890sendWorkerWait sv = do
891 -- Note that we are guaranteed to have at least one outstanding worker when
892 -- we enter this function. So if we sleep we are guaranteed to be woken up
893 -- by a doorBell, when the worker exits.
895 -- XXX we need a better way to handle this than hardcoded delays. The
896 -- delays may be different for different systems.
897 ncpu <- liftIO $ getNumCapabilities
900 if (svarStyle sv == AheadVar)
901 then liftIO $ threadDelay 100
902 else liftIO $ threadDelay 25
904 if (svarStyle sv == AheadVar)
905 then liftIO $ threadDelay 100
906 else liftIO $ threadDelay 10
908 (_, n) <- liftIO $ readIORef (outputQueue sv)
910 -- The queue may be empty temporarily if the worker has dequeued the
911 -- work item but has not enqueued the remaining part yet. For the same
912 -- reason, a worker may come back if it tries to dequeue and finds the
913 -- queue empty, even though the whole work has not finished yet.
915 -- If we find that the queue is empty, but it may be empty
916 -- temporarily, when we checked it. If that's the case we might
917 -- sleep indefinitely unless the active workers produce some
918 -- output. We may deadlock specially if the otuput from the active
919 -- workers depends on the future workers that we may never send.
920 -- So in case the queue was temporarily empty set a flag to inform
921 -- the enqueue to send us a doorbell.
923 -- Note that this is just a best effort mechanism to avoid a
924 -- deadlock. Deadlocks may still happen if for some weird reason
925 -- the consuming computation shares an MVar or some other resource
926 -- with the producing computation and gets blocked on that resource
927 -- and therefore cannot do any pushworker to add more threads to
928 -- the producer. In such cases the programmer should use a parallel
929 -- style so that all the producers are scheduled immediately and
930 -- unconditionally. We can also use a separate monitor thread to
931 -- push workers instead of pushing them from the consumer, but then
932 -- we are no longer using pull based concurrency rate adaptation.
934 -- XXX update this in the tutorial.
936 -- register for the doorBell before we check the queue so that if we
937 -- sleep because the queue was empty we are guaranteed to get a
938 -- doorbell on the next enqueue.
940 liftIO $ atomicModifyIORefCAS_ (waitingForWork sv) $ const True
941 liftIO $ storeLoadBarrier
944 -- XXX test for the case when we miss sending a worker when the worker
945 -- count is more than 1500.
947 -- XXX Assert here that if the heap is not empty then there is at
948 -- least one outstanding worker. Otherwise we could be sleeping
954 liftIO $ withDBGMVar sv "sendWorkerWait: nothing to do"
955 $ takeMVar (doorBell sv)
956 (_, len) <- liftIO $ readIORef (outputQueue sv)
957 when (len <= 0) $ sendWorkerWait sv
958 else sendWorkerWait sv
960-- | Pull a stream from an SVar.
961{-# NOINLINE fromStreamVar #-}
962fromStreamVar :: MonadAsync m => SVar m a -> Stream m a
963fromStreamVar sv = Stream $ \_ stp sng yld -> do
965 -- XXX we can set this in SVar
966 if svarStyle sv == ParallelVar
968 liftIO $ withDBGMVar sv "fromStreamVar: doorbell"
969 $ takeMVar (doorBell sv)
972 res@(_, len) <- readOutputQ sv
973 -- When there is no output seen we dispatch more workers to help
974 -- out if there is work pending in the work queue.
978 -- send a worker proactively, if needed, even before we start
979 -- processing the output. This may degrade single processor
980 -- perf but improves multi-processor, because of more
985 -- Reversing the output is important to guarantee that we process the
986 -- outputs in the same order as they were generated by the constituent
988 runStream (processEvents $ reverse list) Nothing stp sng yld
992 {-# INLINE readOutputQ #-}
993 readOutputQ svr = liftIO $ do
994 (list, len) <- atomicModifyIORefCAS (outputQueue svr) $
997 oqLen <- readIORef (maxOutQSize svr)
998 when (len > oqLen) $ writeIORef (maxOutQSize svr) len
1003 cnt <- liftIO $ readIORef $ activeWorkers sv
1004 when (cnt <= 0) $ do
1006 when (not done) $ pushWorker sv
1008 {-# INLINE blockingRead #-}
1015#ifdef DIAGNOSTICS_VERBOSE
1016 svInfo <- liftIO $ dumpSVar sv
1017 liftIO $ hPutStrLn stderr $ "fromStreamVar done\n" ++ svInfo
1022 {-# INLINE processEvents #-}
1023 processEvents [] = Stream $ \_ stp sng yld -> do
1024 workersDone <- allThreadsDone sv
1026 -- XXX we can set this in SVar
1027 if svarStyle sv == ParallelVar
1028 then return workersDone
1030 -- There may still be work pending even if there are no workers
1031 -- pending because all the workers may return if the
1032 -- outputQueue becomes full. In that case send off a worker to
1033 -- kickstart the work again.
1037 when (not r) $ pushWorker sv
1043 else runStream (fromStreamVar sv) Nothing stp sng yld
1045 processEvents (ev : es) = Stream $ \_ stp sng yld -> do
1046 let rest = processEvents es
1048 ChildYield a -> yld a rest
1049 ChildStop tid e -> do
1050 if svarStyle sv == ParallelVar
1051 then modifyThread sv tid
1052 else delThread sv tid
1054 Nothing -> runStream rest Nothing stp sng yld
1055 Just ex -> throwM ex
1057getFifoSVar :: MonadIO m => SVarStyle -> IO (SVar m a)
1058getFifoSVar ctype = do
1059 outQ <- newIORef ([], 0)
1060 outQMv <- newEmptyMVar
1061 active <- newIORef 0
1062 wfw <- newIORef False
1063 running <- newIORef S.empty
1067 maxWrk <- newIORef 0
1073 SVar { outputQueue = outQ
1075 , outputHeap = undefined
1076 , runningThreads = running
1077 , workQueue = undefined
1078 , runqueue = runqueueFIFO sv q
1079 , enqueue = enqueueFIFO sv q
1080 , queueEmpty = liftIO $ nullQ q
1081 , waitingForWork = wfw
1083 , activeWorkers = active
1085 , totalDispatches = disp
1086 , maxWorkers = maxWrk
1087 , maxOutQSize = maxOq
1088 , maxHeapSize = maxHs
1089 , maxWorkQSize = maxWq
1094getLifoSVar :: MonadIO m => SVarStyle -> IO (SVar m a)
1095getLifoSVar ctype = do
1096 outQ <- newIORef ([], 0)
1097 outQMv <- newEmptyMVar
1098 active <- newIORef 0
1099 wfw <- newIORef False
1100 running <- newIORef S.empty
1104 maxWrk <- newIORef 0
1109 let checkEmpty = null <$> liftIO (readIORef q)
1111 SVar { outputQueue = outQ
1113 , outputHeap = undefined
1114 , runningThreads = running
1115 , workQueue = undefined
1116 , runqueue = runqueueLIFO sv q
1117 , enqueue = enqueueLIFO sv q
1118 , queueEmpty = checkEmpty
1119 , waitingForWork = wfw
1121 , activeWorkers = active
1123 , maxWorkers = maxWrk
1124 , totalDispatches = disp
1125 , maxOutQSize = maxOq
1126 , maxHeapSize = maxHs
1127 , maxWorkQSize = maxWq
1132getParSVar :: SVarStyle -> IO (SVar m a)
1133getParSVar style = do
1134 outQ <- newIORef ([], 0)
1135 outQMv <- newEmptyMVar
1136 active <- newIORef 0
1137 wfw <- newIORef False
1138 running <- newIORef S.empty
1141 maxWrk <- newIORef 0
1147 SVar { outputQueue = outQ
1149 , outputHeap = undefined
1150 , runningThreads = running
1151 , workQueue = undefined
1152 , runqueue = undefined
1153 , enqueue = undefined
1154 , queueEmpty = undefined
1155 , waitingForWork = wfw
1157 , activeWorkers = active
1159 , totalDispatches = disp
1160 , maxWorkers = maxWrk
1161 , maxOutQSize = maxOq
1162 , maxHeapSize = maxHs
1163 , maxWorkQSize = maxWq
1168getAheadSVar :: MonadIO m => SVarStyle -> IO (SVar m a)
1169getAheadSVar style = do
1170 outQ <- newIORef ([], 0)
1171 outH <- newIORef (H.empty, 0)
1172 outQMv <- newEmptyMVar
1173 active <- newIORef 0
1174 wfw <- newIORef False
1175 running <- newIORef S.empty
1176 q <- newIORef ([], -1)
1180 maxWrk <- newIORef 0
1186 let checkEmpty = liftIO $ do
1187 (xs, _) <- readIORef q
1190 SVar { outputQueue = outQ
1193 , runningThreads = running
1195 , runqueue = runqueueAhead sv q
1196 , enqueue = undefined
1197 , queueEmpty = checkEmpty
1198 , waitingForWork = wfw
1200 , activeWorkers = active
1203 , totalDispatches = disp
1204 , maxWorkers = maxWrk
1205 , maxOutQSize = maxOq
1206 , maxHeapSize = maxHs
1207 , maxWorkQSize = maxWq
1212-- | Create a new empty SVar.
1213newEmptySVar :: MonadAsync m => SVarStyle -> m (SVar m a)
1214newEmptySVar style = do
1217 WAsyncVar -> getFifoSVar style
1218 AsyncVar -> getLifoSVar style
1219 ParallelVar -> getParSVar style
1220 AheadVar -> getAheadSVar style
1222-- | Create a new SVar and enqueue one stream computation on it.
1223{-# INLINABLE newStreamVar1 #-}
1224newStreamVar1 :: MonadAsync m => SVarStyle -> Stream m a -> m (SVar m a)
1225newStreamVar1 style m = do
1226 sv <- newEmptySVar style
1227 -- Note: We must have all the work on the queue before sending the
1228 -- pushworker, otherwise the pushworker may exit before we even get a
1230 if style == ParallelVar
1231 then pushWorkerPar sv m
1233 liftIO $ (enqueue sv) m
1237-- | Create a new SVar and enqueue one stream computation on it.
1238{-# INLINABLE newStreamVarAhead #-}
1239newStreamVarAhead :: MonadAsync m => Stream m a -> m (SVar m a)
1240newStreamVarAhead m = do
1241 sv <- newEmptySVar AheadVar
1242 -- Note: We must have all the work on the queue before sending the
1243 -- pushworker, otherwise the pushworker may exit before we even get a
1245 liftIO $ enqueueAhead sv (workQueue sv) m
1249-- | Write a stream to an 'SVar' in a non-blocking manner. The stream can then
1250-- be read back from the SVar using 'fromSVar'.
1251toStreamVar :: MonadAsync m => SVar m a -> Stream m a -> m ()
1252toStreamVar sv m = do
1253 liftIO $ (enqueue sv) m
1254 done <- allThreadsDone sv
1255 -- XXX This is safe only when called from the consumer thread or when no
1256 -- consumer is present. There may be a race if we are not running in the
1258 when done $ pushWorker sv
1260------------------------------------------------------------------------------
1261-- Running streams concurrently
1262------------------------------------------------------------------------------
1264-- Concurrency rate control.
1266-- Our objective is to create more threads on demand if the consumer is running
1267-- faster than us. As soon as we encounter a concurrent composition we create a
1268-- push pull pair of threads. We use an SVar for communication between the
1269-- consumer, pulling from the SVar and the producer who is pushing to the SVar.
1270-- The producer creates more threads if the SVar drains and becomes empty, that
1271-- is the consumer is running faster.
1273-- XXX Note 1: This mechanism can be problematic if the initial production
1274-- latency is high, we may end up creating too many threads. So we need some
1275-- way to monitor and use the latency as well. Having a limit on the dispatches
1276-- (programmer controlled) may also help.
1278-- TBD Note 2: We may want to run computations at the lower level of the
1279-- composition tree serially even when they are composed using a parallel
1280-- combinator. We can use 'serial' in place of 'async' and 'wSerial' in
1281-- place of 'wAsync'. If we find that an SVar immediately above a computation
1282-- gets drained empty we can switch to parallelizing the computation. For that
1283-- we can use a state flag to fork the rest of the computation at any point of
1284-- time inside the Monad bind operation if the consumer is running at a faster
1287-- TBD Note 3: the binary operation ('parallel') composition allows us to
1288-- dispatch a chunkSize of only 1. If we have to dispatch in arbitrary
1289-- chunksizes we will need to compose the parallel actions using a data
1290-- constructor (A Free container) instead so that we can divide it in chunks of
1291-- arbitrary size before dispatching. If the stream is composed of
1292-- hierarchically composed grains of different sizes then we can always switch
1293-- to a desired granularity depending on the consumer speed.
1295-- TBD Note 4: for pure work (when we are not in the IO monad) we can divide it
1296-- into just the number of CPUs.
1298-- | Join two computations on the currently running 'SVar' queue for concurrent
1299-- execution. When we are using parallel composition, an SVar is passed around
1300-- as a state variable. We try to schedule a new parallel computation on the
1301-- SVar passed to us. The first time, when no SVar exists, a new SVar is
1302-- created. Subsequently, 'joinStreamVarAsync' may get called when a computation
1303-- already scheduled on the SVar is further evaluated. For example, when (a
1304-- `parallel` b) is evaluated it calls a 'joinStreamVarAsync' to put 'a' and 'b' on
1305-- the current scheduler queue.
1307-- The 'SVarStyle' required by the current composition context is passed as one
1308-- of the parameters. If the scheduling and composition style of the new
1309-- computation being scheduled is different than the style of the current SVar,
1310-- then we create a new SVar and schedule it on that. The newly created SVar
1311-- joins as one of the computations on the current SVar queue.
1313-- Cases when we need to switch to a new SVar:
1315-- * (x `parallel` y) `parallel` (t `parallel` u) -- all of them get scheduled on the same SVar
1316-- * (x `parallel` y) `parallel` (t `async` u) -- @t@ and @u@ get scheduled on a new child SVar
1317-- because of the scheduling policy change.
1318-- * if we 'adapt' a stream of type 'async' to a stream of type
1319-- 'Parallel', we create a new SVar at the transitioning bind.
1320-- * When the stream is switching from disjunctive composition to conjunctive
1321-- composition and vice-versa we create a new SVar to isolate the scheduling
1324forkSVarAsync :: MonadAsync m => SVarStyle -> Stream m a -> Stream m a -> Stream m a
1325forkSVarAsync style m1 m2 = Stream $ \_ stp sng yld -> do
1326 sv <- newStreamVar1 style (concurrently m1 m2)
1327 (runStream (fromStreamVar sv)) Nothing stp sng yld
1329 concurrently ma mb = Stream $ \svr stp sng yld -> do
1330 liftIO $ enqueue (fromJust svr) mb
1331 (runStream ma) svr stp sng yld
1333{-# INLINE joinStreamVarAsync #-}
1334joinStreamVarAsync :: MonadAsync m
1335 => SVarStyle -> Stream m a -> Stream m a -> Stream m a
1336joinStreamVarAsync style m1 m2 = Stream $ \svr stp sng yld ->
1338 Just sv | svarStyle sv == style ->
1339 liftIO ((enqueue sv) m2) >> (runStream m1) svr stp sng yld
1340 _ -> runStream (forkSVarAsync style m1 m2) Nothing stp sng yld
1342{-# NOINLINE forkSVarPar #-}
1343forkSVarPar :: MonadAsync m => Stream m a -> Stream m a -> Stream m a
1344forkSVarPar m r = Stream $ \_ stp sng yld -> do
1345 sv <- newEmptySVar ParallelVar
1348 (runStream (fromStreamVar sv)) Nothing stp sng yld
1350{-# INLINE joinStreamVarPar #-}
1351joinStreamVarPar :: MonadAsync m
1352 => SVarStyle -> Stream m a -> Stream m a -> Stream m a
1353joinStreamVarPar style m1 m2 = Stream $ \svr stp sng yld ->
1355 Just sv | svarStyle sv == style -> do
1356 pushWorkerPar sv m1 >> (runStream m2) svr stp sng yld
1357 _ -> runStream (forkSVarPar m1 m2) Nothing stp sng yld
1359forkSVarAhead :: MonadAsync m => Stream m a -> Stream m a -> Stream m a
1360forkSVarAhead m1 m2 = Stream $ \_ stp sng yld -> do
1361 sv <- newStreamVarAhead (concurrently m1 m2)
1362 (runStream (fromStreamVar sv)) Nothing stp sng yld
1364 concurrently ma mb = Stream $ \svr stp sng yld -> do
1365 liftIO $ enqueueAhead (fromJust svr) (workQueue (fromJust svr)) mb
1366 (runStream ma) Nothing stp sng yld
1369ahead :: MonadAsync m => Stream m a -> Stream m a -> Stream m a
1370ahead m1 m2 = Stream $ \svr stp sng yld -> do
1372 Just sv | svarStyle sv == AheadVar -> do
1373 liftIO $ enqueueAhead sv (workQueue sv) m2
1374 -- Always run the left side on a new SVar to avoid complexity in
1375 -- sequencing results. This means the left side cannot further
1376 -- split into more ahead computations on the same SVar.
1377 (runStream m1) Nothing stp sng yld
1378 _ -> runStream (forkSVarAhead m1 m2) Nothing stp sng yld
1380-- | XXX we can implement it more efficienty by directly implementing instead
1381-- of combining streams using ahead.
1382{-# INLINE consMAhead #-}
1383consMAhead :: MonadAsync m => m a -> Stream m a -> Stream m a
1384consMAhead m r = once m `ahead` r
1386------------------------------------------------------------------------------
1387-- Semigroup and Monoid style compositions for parallel actions
1388------------------------------------------------------------------------------
1391async :: MonadAsync m => Stream m a -> Stream m a -> Stream m a
1392async = joinStreamVarAsync AsyncVar
1394-- | XXX we can implement it more efficienty by directly implementing instead
1395-- of combining streams using async.
1396{-# INLINE consMAsync #-}
1397consMAsync :: MonadAsync m => m a -> Stream m a -> Stream m a
1398consMAsync m r = once m `async` r
1400{-# INLINE wAsync #-}
1401wAsync :: MonadAsync m => Stream m a -> Stream m a -> Stream m a
1402wAsync = joinStreamVarAsync WAsyncVar
1404-- | XXX we can implement it more efficienty by directly implementing instead
1405-- of combining streams using wAsync.
1406{-# INLINE consMWAsync #-}
1407consMWAsync :: MonadAsync m => m a -> Stream m a -> Stream m a
1408consMWAsync m r = once m `wAsync` r
1410{-# INLINE parallel #-}
1411parallel :: MonadAsync m => Stream m a -> Stream m a -> Stream m a
1412parallel = joinStreamVarPar ParallelVar
1414-- | XXX we can implement it more efficienty by directly implementing instead
1415-- of combining streams using parallel.
1416{-# INLINE consMParallel #-}
1417consMParallel :: MonadAsync m => m a -> Stream m a -> Stream m a
1418consMParallel m r = once m `parallel` r
1420-------------------------------------------------------------------------------
1421-- Functor instace is the same for all types
1422-------------------------------------------------------------------------------
1424instance Monad m => Functor (Stream m) where
1425 fmap f m = Stream $ \_ stp sng yld ->
1426 let single = sng . f
1427 yield a r = yld (f a) (fmap f r)
1428 in (runStream m) Nothing stp single yield
1430------------------------------------------------------------------------------
1431-- Alternative & MonadPlus
1432------------------------------------------------------------------------------
1434_alt :: Stream m a -> Stream m a -> Stream m a
1435_alt m1 m2 = Stream $ \_ stp sng yld ->
1436 let stop = runStream m2 Nothing stp sng yld
1437 in runStream m1 Nothing stop sng yld
1439------------------------------------------------------------------------------
1440-- Stream to stream function application
1441------------------------------------------------------------------------------
1443applyWith :: MonadAsync m
1444 => SVarStyle -> (Stream m a -> Stream m b) -> Stream m a -> Stream m b
1445applyWith style f m = Stream $ \svr stp sng yld -> do
1446 sv <- newStreamVar1 style m
1447 runStream (f $ fromStreamVar sv) svr stp sng yld
1449------------------------------------------------------------------------------
1450-- Stream runner function application
1451------------------------------------------------------------------------------
1453runWith :: MonadAsync m
1454 => SVarStyle -> (Stream m a -> m b) -> Stream m a -> m b
1455runWith style f m = do
1456 sv <- newStreamVar1 style m
1457 f $ fromStreamVar sv
1459------------------------------------------------------------------------------
1461------------------------------------------------------------------------------
1463{-# INLINE zipWith #-}
1464zipWith :: (a -> b -> c) -> Stream m a -> Stream m b -> Stream m c
1465zipWith f m1 m2 = go m1 m2
1467 go mx my = Stream $ \_ stp sng yld -> do
1469 let single2 b = sng (f a b)
1470 yield2 b rb = yld (f a b) (go ra rb)
1471 in (runStream my) Nothing stp single2 yield2
1472 let single1 a = merge a nil
1473 yield1 a ra = merge a ra
1474 (runStream mx) Nothing stp single1 yield1
1476{-# INLINE zipAsyncWith #-}
1477zipAsyncWith :: MonadAsync m
1478 => (a -> b -> c) -> Stream m a -> Stream m b -> Stream m c
1479zipAsyncWith f m1 m2 = Stream $ \_ stp sng yld -> do
1482 (runStream (zipWith f ma mb)) Nothing stp sng yld
1486 mkAsync :: MonadAsync m => Stream m a -> m (Stream m a)
1487 mkAsync m = newStreamVar1 AsyncVar m
1488 >>= return . fromStreamVar
1490-------------------------------------------------------------------------------
1492-------------------------------------------------------------------------------
1494instance MonadTrans Stream where
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