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
55 -- * Concurrent Stream Vars (SVars)
66import Control.Concurrent (ThreadId, myThreadId, threadDelay)
67import Control.Concurrent.MVar (MVar, newEmptyMVar, tryTakeMVar,
68 tryPutMVar, takeMVar, putMVar)
69import Control.Exception (SomeException (..))
70import qualified Control.Exception.Lifted as EL
71import Control.Monad (when)
72import Control.Monad.Catch (MonadThrow, throwM)
73import Control.Monad.Error.Class (MonadError(..))
74import Control.Monad.IO.Class (MonadIO(..))
75import Control.Monad.Reader.Class (MonadReader(..))
76import Control.Monad.Trans.Class (MonadTrans (lift))
77import Control.Monad.Trans.Control (MonadBaseControl, liftBaseWith)
78import Data.Atomics (casIORef, readForCAS, peekTicket
79 ,atomicModifyIORefCAS_)
80import Data.Concurrent.Queue.MichaelScott (LinkedQueue, newQ, pushL,
82import Data.Functor (void)
83import Data.Heap (Heap, Entry(..))
84import qualified Data.Heap as H
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-- | Sorting out of turn outputs in a heap for Ahead style streams
107data AheadHeapEntry a =
109 | AheadYieldMany a (MVar ()) -- value, MVar to wakeup for more values
111------------------------------------------------------------------------------
112-- State threaded around the monad for thread management
113------------------------------------------------------------------------------
115-- XXX use a separate data structure for each type of SVar
116-- | Identify the type of the SVar. Two computations using the same style can
117-- be scheduled on the same SVar.
119 AsyncVar -- depth first concurrent
120 | WAsyncVar -- breadth first concurrent
121 | ParallelVar -- all parallel
122 | AheadVar -- Concurrent look ahead
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'.
152 svarStyle :: SVarStyle
154 -- Shared output queue (events, length)
155 , outputQueue :: IORef ([ChildEvent a], Int)
156 , doorBell :: MVar () -- signal the consumer about output
157 , siren :: MVar () -- hooter for workers to begin work
159 -- Output synchronization mechanism for Ahead streams (Ahead and
160 -- wAhead). We maintain a heap of out of sequence ahead of time
161 -- generated outputs and the sequence number of the task that is
162 -- currently at the head of the stream. Concurrent execute ahead
163 -- tasks that have a sequence number greater than the task at the
164 -- head should add their output to the heap.
165 , outputHeap :: IORef (Heap (Entry Int (AheadHeapEntry a)), Int)
166 , heapSiren :: MVar () -- hooter for workers to continue
167 -- adding to the heap
170 , workQueue :: IORef ([Stream m a], Int)
171 , enqueue :: Stream m a -> IO ()
172 , queueEmpty :: m Bool
175 -- Shared, thread tracking
176 , runningThreads :: IORef (Set ThreadId)
177 , activeWorkers :: IORef Int
180-- Slightly faster version of CAS. Gained some improvement by avoiding the use
181-- of "evaluate" because we know we do not have exceptions in fn.
182{-# INLINE atomicModifyIORefCAS #-}
183atomicModifyIORefCAS :: IORef a -> (a -> (a,b)) -> IO b
184atomicModifyIORefCAS ref fn = do
185 tkt <- readForCAS ref
191 loop _ 0 = atomicModifyIORef ref fn
193 let (new, result) = fn $ peekTicket old
194 (success, tkt) <- casIORef ref old new
197 else loop tkt (tries - 1)
199------------------------------------------------------------------------------
201------------------------------------------------------------------------------
203-- | The type 'Stream m a' represents a monadic stream of values of type 'a'
204-- constructed using actions in monad 'm'. It uses stop, singleton and yield
205-- continuations equivalent to the following direct style type:
207-- data Stream m a = Stop | Singleton a | Yield a (Stream m a)
209-- To facilitate parallel composition we maintain a local state in an SVar that
210-- is shared across and is used for synchronization of the streams being
213-- The singleton case can be expressed in terms of stop and yield but we have
214-- it as a separate case to optimize composition operations for streams with
215-- single element. We build singleton streams in the implementation of 'pure'
216-- for Applicative and Monad, and in 'lift' for MonadTrans.
220 runStream :: forall r.
221 Maybe (SVar m a) -- local state
223 -> (a -> m r) -- singleton
224 -> (a -> Stream m a -> m r) -- yield
229nil = Stream $ \_ stp _ _ -> stp
231once :: Monad m => m a -> Stream m a
232once m = Stream $ \_ _ single _ -> m >>= single
234{-# INLINE singleton #-}
235-- | Same as @once . return@
236singleton :: a -> Stream m a
237singleton a = Stream $ \_ _ single _ -> single a
239consM :: Monad m => m a -> Stream m a -> Stream m a
240consM m r = Stream $ \_ _ _ yld -> m >>= \a -> yld a r
242-- | Same as @consM . return@
243cons :: a -> Stream m a -> Stream m a
244cons a r = Stream $ \_ _ _ yld -> yld a r
246repeat :: a -> Stream m a
247repeat a = let x = cons a x in x
249------------------------------------------------------------------------------
251------------------------------------------------------------------------------
253-- | Concatenates two streams sequentially i.e. the first stream is
254-- exhausted completely before yielding any element from the second stream.
255serial :: Stream m a -> Stream m a -> Stream m a
258 go (Stream m) = Stream $ \_ stp sng yld ->
259 let stop = (runStream m2) Nothing stp sng yld
261 yield a r = yld a (go r)
262 in m Nothing stop single yield
264instance Semigroup (Stream m a) where
267------------------------------------------------------------------------------
269------------------------------------------------------------------------------
271instance Monoid (Stream m a) where
275------------------------------------------------------------------------------
277------------------------------------------------------------------------------
279wSerial :: Stream m a -> Stream m a -> Stream m a
280wSerial m1 m2 = Stream $ \_ stp sng yld -> do
281 let stop = (runStream m2) Nothing stp sng yld
283 yield a r = yld a (wSerial m2 r)
284 (runStream m1) Nothing stop single yield
286------------------------------------------------------------------------------
287-- Spawning threads and collecting result in streamed fashion
288------------------------------------------------------------------------------
290-- | A monad that can perform concurrent or parallel IO operations. Streams
291-- that can be composed concurrently require the underlying monad to be
295type MonadAsync m = (MonadIO m, MonadBaseControl IO m, MonadThrow m)
297-- Stolen from the async package. The perf improvement is modest, 2% on a
298-- thread heavy benchmark (parallel composition using noop computations).
299-- A version of forkIO that does not include the outer exception
300-- handler: saves a bit of time when we will be installing our own
302{-# INLINE rawForkIO #-}
303rawForkIO :: IO () -> IO ThreadId
304rawForkIO action = IO $ \ s ->
305 case (fork# action s) of (# s1, tid #) -> (# s1, ThreadId tid #)
308doFork :: MonadBaseControl IO m
310 -> (SomeException -> m ())
312doFork action exHandler =
313 EL.mask $ \restore ->
314 liftBaseWith $ \runInIO -> rawForkIO $ do
315 _ <- runInIO $ EL.catch (restore action) exHandler
316 -- XXX restore state here?
319-- XXX exception safety of all atomic/MVar operations
321-- TBD Each worker can have their own queue and the consumer can empty one
322-- queue at a time, that way contention can be reduced.
324-- | This function is used by the producer threads to queue output for the
325-- consumer thread to consume.
327send :: SVar m a -> ChildEvent a -> IO ()
329 len <- atomicModifyIORefCAS (outputQueue sv) $ \(es, n) ->
330 ((msg : es, n + 1), n)
331 if (len <= 0) then do
332 -- XXX need a memory barrier? The wake up must happen only after the
333 -- store has finished otherwise we can have lost wakeup problems.
335 -- Since multiple workers can try this at the same time, it is possible
336 -- that we may put a spurious MVar after the consumer has already seen
337 -- the output. But that's harmless, at worst it may cause the consumer
338 -- to read the queue again and find it empty.
339 -- The important point is that the consumer is guaranteed to receive a
340 -- doorbell if something was added to the queue after it empties it.
341 void $ tryPutMVar (doorBell sv) ()
342 -- those who were waiting because of buffer limit should now go ahead
343 void $ tryPutMVar (siren sv) ()
344 -- The only thread that reduces the length of the queue is the consumer
345 -- thread. It is possible that the queue may have been emptied in the
346 -- window when we acquired "len" and now when we are using it. In that case
347 -- we will be woken up by the consumer blowing the siren.
348 -- Note: current length after adding our output is len + 1
349 else when (len + 1 >= 1500) $ do
351 (_, n) <- readIORef (outputQueue sv)
352 when (n <= 1500) $ void $ tryPutMVar (siren sv) ()
354{-# INLINE sendStop #-}
355sendStop :: SVar m a -> IO ()
356sendStop sv = myThreadId >>= \tid -> send sv (ChildStop tid Nothing)
358-------------------------------------------------------------------------------
360-------------------------------------------------------------------------------
362-- Note: For purely right associated expressions this queue should have at most
363-- one element. It grows to more than one when we have left associcated
364-- expressions. Large left associated compositions can grow this to a
366{-# INLINE enqueueLIFO #-}
367enqueueLIFO :: IORef [Stream m a] -> Stream m a -> IO ()
368enqueueLIFO q m = atomicModifyIORefCAS_ q $ \ ms -> m : ms
370runqueueLIFO :: MonadIO m => SVar m a -> IORef [Stream m a] -> m ()
371runqueueLIFO sv q = run
378 Nothing -> liftIO $ sendStop sv
379 Just m -> (runStream m) (Just sv) run single yield
381 sendit a = liftIO $ send sv (ChildYield a)
382 single a = sendit a >> run
383 yield a r = sendit a >> (runStream r) (Just sv) run single yield
385 dequeue = liftIO $ atomicModifyIORefCAS q $ \case
387 x : xs -> (xs, Just x)
389-------------------------------------------------------------------------------
391-------------------------------------------------------------------------------
393{-# INLINE enqueueFIFO #-}
394enqueueFIFO :: LinkedQueue (Stream m a) -> Stream m a -> IO ()
397runqueueFIFO :: MonadIO m => SVar m a -> LinkedQueue (Stream m a) -> m ()
398runqueueFIFO sv q = run
405 Nothing -> liftIO $ sendStop sv
406 Just m -> (runStream m) (Just sv) run single yield
408 dequeue = liftIO $ tryPopR q
409 sendit a = liftIO $ send sv (ChildYield a)
410 single a = sendit a >> run
411 yield a r = sendit a >> liftIO (enqueueFIFO q r) >> run
413-------------------------------------------------------------------------------
415-------------------------------------------------------------------------------
418runOne :: MonadIO m => SVar m a -> Stream m a -> m ()
419runOne sv m = (runStream m) (Just sv) stop single yield
423 stop = liftIO $ sendStop sv
424 sendit a = liftIO $ send sv (ChildYield a)
425 single a = sendit a >> stop
426 yield a r = sendit a >> runOne sv r
428-------------------------------------------------------------------------------
430-------------------------------------------------------------------------------
432-- Lookahead streams can execute multiple tasks concurrently, ahead of time but
433-- always serves them in the same order as they appear in the stream. To
434-- implement lookahead streams efficiently we assign a sequence number to each
435-- task when the task is picked up for execution. When the task finishes the
436-- output is tagged with the same sequence number and we rearrange the outputs
437-- in sequence based on that number.
439-- To explain the mechanism imagine that the current task at the head of the
440-- stream has a "token" to yield to the output. The ownership of the token is
441-- determined by the current sequence number maintained in outputHeap. Sequence
442-- number is assigned when a task is queued. When a thread dequeues a task it
443-- picks up the sequence number as well and when the output is ready it uses
444-- the sequence number to queue the output to the outputQueue.
446-- The thread with current sequence number sends the output directly to the
447-- outputQueue. Other threads push the output to the outputHeap. When the task
448-- being queued on the heap is a stream of many elements we yield the first
449-- element on the heap and an MVar to wakeup for more elements. When such a
450-- task gets the "token" for outputQueue it can directly keep yielding all the
451-- elements to the outputQueue without checking for the "token".
453-- Note that no two outputs in the heap can have the same sequence numbers and
454-- therefore we do not need a stable heap. We have also separated the buffer
455-- for the current task (outputQueue) and the pending tasks (outputHeap) so
456-- that the pending tasks cannot interfere with the current task. Note that for
457-- a single task just the outputQueue is enough and for the case of many
458-- threads just a heap is good enough. However we balance between these two
459-- cases, so that both are efficient.
461-- For bigger streams it may make sense to have separate buffers for each
462-- stream. However, for singleton streams this may become inefficient. However,
463-- if we do not have separate buffers, then the streams that come later in
464-- sequence may hog the buffer, hindering the streams that are ahead. For this
465-- reason we have a single element buffer limitation for the streams being
466-- executed in advance.
468-- This scheme works pretty efficiently with only 50% more overhead compared to
469-- the Async streams where we do not have any kind of sequencing of the
470-- outputs. It is especially devised so that we are most efficient when we have
471-- short tasks and need just a single thread. Also when a thread yields many
472-- items it can hold lockfree access to the outputQueue and do it efficiently.
474-- XXX Maybe we can start the ahead threads at a lower cpu and IO priority so
475-- that they do not hog the resources and hinder the progress of the threads in
478{-# INLINE sendToHeap #-}
479sendToHeap :: SVar m a -> AheadHeapEntry a -> Int -> IO Bool
480sendToHeap sv msg seqNo = do
481 let maxHeapSize = 1500
483 -- Maybe we can use a lock instead of using CAS
484 -- insertions in the heap may be expensive if the heap is big
485 -- Note: len is valid only when added is True
486 (len, added) <- atomicModifyIORefCAS (outputHeap sv) $
489 then ((H.insert (Entry seqNo msg) h, snum), (H.size h, True))
490 else (ref, (0, False))
492 -- this is only for the case when the consumer generated siren can get
493 -- missed due to race.
494 when (len < maxHeapSize) $ void $ tryPutMVar (heapSiren sv) ()
498 when (len + 1 >= maxHeapSize) $ do
499 takeMVar (heapSiren sv)
500 (h, _) <- readIORef (outputHeap sv)
501 when (H.size h <= maxHeapSize) $ void $ tryPutMVar (siren sv) ()
505 AheadYieldSingle a -> send sv (ChildYield a)
506 AheadYieldMany a mv -> do
507 send sv (ChildYield a)
512 :: IORef (Heap (Entry Int (AheadHeapEntry a)), Int)
514 -> IO (Maybe (AheadHeapEntry a))
515popFromHeap hp nextSeqNo =
516 atomicModifyIORefCAS hp $ \(h, snum) -> do
518 let n = max nextSeqNo snum
520 Nothing -> ((h, n), Nothing)
521 Just (Entry seqNo ev, hp') ->
523 then ((hp', seqNo), Just ev)
524 else ((h, n), Nothing)
526-- | As soon as we set the nextSeqNo in outputHeap we have transferred the
527-- output "token" to the thread executing the task with that sequence number.
528-- We have to make sure that there is no entry in the heap that has nextSeqNo
529-- before we set nextSeqNo as the expected sequence number in the heap. It is
530-- possible that when we are clearing the heap entry of the current sequence
531-- number the next sequence number gets added and then we have to clear that,
532-- this can keep going on and on.
533drainHeap :: SVar m a -> Int -> IO ()
534drainHeap sv nextSeqNo = do
535 r <- popFromHeap (outputHeap sv) nextSeqNo
540 AheadYieldSingle a -> do
541 send sv (ChildYield a)
542 atomicModifyIORefCAS_ (outputHeap sv) $ \(h, snum) ->
544 drainHeap sv nextSeqNo
545 AheadYieldMany a mv -> do
546 send sv (ChildYield a)
549-- | The thread that has the token has the responsibility to move any pending
550-- entries in the heap to the outputQueue. When the token is assigned to a
551-- sequence number we must not have any entries for that sequence number in the
553releaseToken :: SVar m a -> Int -> IO ()
554releaseToken sv nextSeqNo = do
555 drainHeap sv nextSeqNo
556 void $ tryPutMVar (heapSiren sv) ()
558-- | When we have the output "token" we are allowed to append to the
559-- outputQueue otherwise we have to insert the output in the outputHeap.
560-- Returns whether we still have the token.
561handleOutput :: SVar m a
568handleOutput sv haveToken prevSeqNo seqNo a ev = do
571 let nextSeqNo = prevSeqNo + 1
572 if seqNo == nextSeqNo
573 then send sv (ChildYield a) >> return True
575 releaseToken sv nextSeqNo
578 where toHeap = sendToHeap sv ev seqNo
580-- Left associated ahead expressions are expensive. We start a new SVar for
581-- each left associative expression. The queue is used only for right
582-- associated expression, we queue the right expression and execute the left.
583-- Thererefore the queue never has more than on item in it.
584{-# INLINE enqueueAhead #-}
585enqueueAhead :: IORef ([Stream m a], Int) -> Stream m a -> IO ()
587 atomicModifyIORefCAS_ q $ \ case
588 ([], n) -> ([m], n + 1) -- increment sequence
589 _ -> error "not empty"
591runqueueAhead :: MonadIO m => SVar m a -> IORef ([Stream m a], Int) -> m ()
592runqueueAhead sv q = run False (-1)
596 run haveToken prevSeqNo = do
599 Nothing -> liftIO $ do
600 when haveToken $ releaseToken sv (prevSeqNo + 1)
602 Just (m, seqNo) -> do
603 let tStatus = haveToken || seqNo == 0
605 let nextSeqNo = prevSeqNo + 1
606 if tStatus && seqNo /= nextSeqNo
608 liftIO $ releaseToken sv nextSeqNo
610 else run tStatus seqNo
612 tokenStatus <- liftIO $
613 handleOutput sv tStatus prevSeqNo seqNo a
615 run tokenStatus seqNo
617 -- If we are yielding more than one item and we do not have the
618 -- token then we insert the current item in the pending
619 -- outputHeap and wait for our turn.
621 stopNext = run True seqNo
623 liftIO $ send sv (ChildYield a)
626 liftIO $ send sv (ChildYield a)
627 (runStream r) (Just sv) stopNext singleNext yieldNext
630 -- XXX we can allocate the MVar when actually inserting
631 -- in the heap and then return it.
635 handleOutput sv tStatus prevSeqNo seqNo a
636 (AheadYieldMany a mv)
637 when (not gotToken) $ void $ takeMVar mv
638 (runStream r) (Just sv) stopNext singleNext yieldNext
640 (runStream m) (Just sv) stop single yield
642 dequeue = liftIO $ do
643 atomicModifyIORefCAS q $ \case
644 ([], n) -> (([], n), Nothing)
645 (x : [], n) -> (([], n), Just (x, n))
646 _ -> error "more than one item on queue"
648-- Thread tracking is needed for two reasons:
650-- 1) Killing threads on exceptions. Threads may not be left to go away by
651-- themselves because they may run for significant times before going away or
652-- worse they may be stuck in IO and never go away.
654-- 2) To know when all threads are done and the stream has ended.
656{-# NOINLINE addThread #-}
657addThread :: MonadIO m => SVar m a -> ThreadId -> m ()
659 liftIO $ modifyIORef (runningThreads sv) (S.insert tid)
662{-# INLINE delThread #-}
663delThread :: MonadIO m => SVar m a -> ThreadId -> m ()
665 liftIO $ modifyIORef (runningThreads sv) $ (\s -> S.delete tid s)
668-- If present then delete else add. This takes care of out of order add and
669-- delete i.e. a delete arriving before we even added a thread.
670-- This occurs when the forked thread is done even before the 'addThread' right
671-- after the fork is executed.
672{-# INLINE modifyThread #-}
673modifyThread :: MonadIO m => SVar m a -> ThreadId -> m ()
674modifyThread sv tid = do
675 changed <- liftIO $ atomicModifyIORefCAS (runningThreads sv) $ \old ->
676 if (S.member tid old)
677 then let new = (S.delete tid old) in (new, new)
678 else let new = (S.insert tid old) in (new, old)
680 then liftIO $ void $ tryPutMVar (doorBell sv) ()
683-- | This is safe even if we are adding more threads concurrently because if
684-- a child thread is adding another thread then anyway 'runningThreads' will
686{-# INLINE allThreadsDone #-}
687allThreadsDone :: MonadIO m => SVar m a -> m Bool
688allThreadsDone sv = liftIO $ S.null <$> readIORef (runningThreads sv)
690{-# NOINLINE handleChildException #-}
691handleChildException :: MonadIO m => SVar m a -> SomeException -> m ()
692handleChildException sv e = liftIO $ do
694 send sv (ChildStop tid (Just e))
696{-# NOINLINE pushWorker #-}
697pushWorker :: MonadAsync m => SVar m a -> m ()
699 liftIO $ atomicModifyIORefCAS_ (activeWorkers sv) $ \n -> n + 1
700 doFork (runqueue sv) (handleChildException sv) >>= addThread sv
702-- | In contrast to pushWorker which always happens only from the consumer
703-- thread, a pushWorkerPar can happen concurrently from multiple threads on the
704-- producer side. So we need to use a thread safe modification of
705-- runningThreads. Alternatively, we can use a CreateThread event to avoid
706-- using a CAS based modification.
707{-# NOINLINE pushWorkerPar #-}
708pushWorkerPar :: MonadAsync m => SVar m a -> Stream m a -> m ()
710 doFork (runOne sv m) (handleChildException sv) >>= modifyThread sv
712-- XXX When the queue is LIFO we can put a limit on the number of dispatches.
713-- Also, if a worker blocks on the output queue we can decide if we want to
714-- block or make it go away entirely, depending on the number of workers and
715-- the type of the queue.
716{-# INLINE sendWorkerWait #-}
717sendWorkerWait :: MonadAsync m => SVar m a -> m ()
718sendWorkerWait sv = do
719 -- When there is no output seen we dispatch more workers to help out if
720 -- there is work pending in the work queue. But we wait a little while
721 -- and check the output again so that we are not too aggressive.
722 -- If there is no output pending to process and there is no worker to be
723 -- sent then we block, so that we do not keep looping fruitlessly.
725 -- XXX we should wait in such a way so that we can be immediately
726 -- interrupted by a doorBell.
727 liftIO $ threadDelay 200
728 (_, n) <- liftIO $ readIORef (outputQueue sv)
730 -- XXX the queue may be empty temporarily if the worker has dequeued
731 -- the work item but has not enqueued the remaining part yet. For the
732 -- same reason, a worker may come back if it tries to dequeue when the
733 -- queue is empty, even though the whole work has not finished yet.
734 done <- queueEmpty sv
737 cnt <- liftIO $ readIORef $ activeWorkers sv
742 else liftIO $ takeMVar (doorBell sv)
743 else liftIO $ takeMVar (doorBell sv)
745-- | Pull a stream from an SVar.
746{-# NOINLINE fromStreamVar #-}
747fromStreamVar :: MonadAsync m => SVar m a -> Stream m a
748fromStreamVar sv = Stream $ \_ stp sng yld -> do
749 if svarStyle sv == ParallelVar
750 then liftIO $ takeMVar (doorBell sv)
752 res <- liftIO $ tryTakeMVar (doorBell sv)
753 when (isNothing res) $ sendWorkerWait sv
755 (list, _) <- liftIO $ atomicModifyIORefCAS (outputQueue sv) $ \x -> (([],0), x)
756 void $ liftIO $ tryPutMVar (siren sv) ()
757 -- Reversing the output is important to guarantee that we process the
758 -- outputs in the same order as they were generated by the constituent
760 (runStream $ processEvents (reverse list)) Nothing stp sng yld
764 handleException e tid = do
765 liftIO $ atomicModifyIORefCAS_ (activeWorkers sv) $ \n -> n - 1
767 -- XXX implement kill async exception handling
768 -- liftIO $ readIORef (runningThreads sv) >>= mapM_ killThread
771 {-# INLINE processEvents #-}
772 processEvents [] = Stream $ \_ stp sng yld -> do
773 done <- allThreadsDone sv
775 then (runStream (fromStreamVar sv)) Nothing stp sng yld
778 processEvents (ev : es) = Stream $ \_ stp sng yld -> do
779 let continue = (runStream (processEvents es)) Nothing stp sng yld
780 yield a = yld a (processEvents es)
783 ChildYield a -> yield a
787 let active = activeWorkers sv
788 liftIO $ atomicModifyIORefCAS_ active $ \n -> n - 1
789 modifyThread sv tid >> continue
790 Just ex -> handleException ex tid
792getFifoSVar :: MonadIO m => SVarStyle -> IO (SVar m a)
793getFifoSVar ctype = do
794 outQ <- newIORef ([], 0)
795 outQMv <- newEmptyMVar
796 hooter <- newEmptyMVar
798 running <- newIORef S.empty
801 SVar { outputQueue = outQ
804 , outputHeap = undefined
805 , heapSiren = undefined
806 , runningThreads = running
807 , workQueue = undefined
808 , runqueue = runqueueFIFO sv q
810 , queueEmpty = liftIO $ nullQ q
812 , activeWorkers = active
816getLifoSVar :: MonadIO m => SVarStyle -> IO (SVar m a)
817getLifoSVar ctype = do
818 outQ <- newIORef ([], 0)
819 outQMv <- newEmptyMVar
820 hooter <- newEmptyMVar
822 running <- newIORef S.empty
824 let checkEmpty = null <$> liftIO (readIORef q)
826 SVar { outputQueue = outQ
829 , outputHeap = undefined
830 , heapSiren = undefined
831 , runningThreads = running
832 , workQueue = undefined
833 , runqueue = runqueueLIFO sv q
834 , enqueue = enqueueLIFO q
835 , queueEmpty = checkEmpty
837 , activeWorkers = active
841getParSVar :: SVarStyle -> IO (SVar m a)
843 outQ <- newIORef ([], 0)
844 outQMv <- newEmptyMVar
845 hooter <- newEmptyMVar
847 running <- newIORef S.empty
849 SVar { outputQueue = outQ
852 , outputHeap = undefined
853 , heapSiren = undefined
854 , runningThreads = running
855 , workQueue = undefined
856 , runqueue = undefined
857 , enqueue = undefined
858 , queueEmpty = undefined
860 , activeWorkers = active
864getAheadSVar :: MonadIO m => SVarStyle -> IO (SVar m a)
865getAheadSVar style = do
866 outQ <- newIORef ([], 0)
867 outH <- newIORef (H.empty, 0)
868 outQMv <- newEmptyMVar
869 hooter <- newEmptyMVar
870 hsiren <- newEmptyMVar
872 running <- newIORef S.empty
873 q <- newIORef ([], -1)
874 let checkEmpty = liftIO $ do
875 (xs, _) <- readIORef q
878 SVar { outputQueue = outQ
883 , runningThreads = running
885 , runqueue = runqueueAhead sv q
886 , enqueue = undefined
887 , queueEmpty = checkEmpty
889 , activeWorkers = active
893-- | Create a new empty SVar.
894newEmptySVar :: MonadAsync m => SVarStyle -> m (SVar m a)
895newEmptySVar style = do
898 WAsyncVar -> getFifoSVar style
899 AsyncVar -> getLifoSVar style
900 ParallelVar -> getParSVar style
901 AheadVar -> getAheadSVar style
903-- | Create a new SVar and enqueue one stream computation on it.
904newStreamVar1 :: MonadAsync m => SVarStyle -> Stream m a -> m (SVar m a)
905newStreamVar1 style m = do
906 sv <- newEmptySVar style
907 -- Note: We must have all the work on the queue before sending the
908 -- pushworker, otherwise the pushworker may exit before we even get a
910 liftIO $ (enqueue sv) m
914-- | Create a new SVar and enqueue one stream computation on it.
915newStreamVarAhead :: MonadAsync m => Stream m a -> m (SVar m a)
916newStreamVarAhead m = do
917 sv <- newEmptySVar AheadVar
918 -- Note: We must have all the work on the queue before sending the
919 -- pushworker, otherwise the pushworker may exit before we even get a
921 liftIO $ enqueueAhead (workQueue sv) m
925-- | Write a stream to an 'SVar' in a non-blocking manner. The stream can then
926-- be read back from the SVar using 'fromSVar'.
927toStreamVar :: MonadAsync m => SVar m a -> Stream m a -> m ()
929 liftIO $ (enqueue sv) m
930 done <- allThreadsDone sv
931 -- XXX This is safe only when called from the consumer thread or when no
932 -- consumer is present. There may be a race if we are not running in the
934 when done $ pushWorker sv
936------------------------------------------------------------------------------
937-- Running streams concurrently
938------------------------------------------------------------------------------
940-- Concurrency rate control.
942-- Our objective is to create more threads on demand if the consumer is running
943-- faster than us. As soon as we encounter a concurrent composition we create a
944-- push pull pair of threads. We use an SVar for communication between the
945-- consumer, pulling from the SVar and the producer who is pushing to the SVar.
946-- The producer creates more threads if the SVar drains and becomes empty, that
947-- is the consumer is running faster.
949-- XXX Note 1: This mechanism can be problematic if the initial production
950-- latency is high, we may end up creating too many threads. So we need some
951-- way to monitor and use the latency as well. Having a limit on the dispatches
952-- (programmer controlled) may also help.
954-- TBD Note 2: We may want to run computations at the lower level of the
955-- composition tree serially even when they are composed using a parallel
956-- combinator. We can use 'serial' in place of 'async' and 'wSerial' in
957-- place of 'wAsync'. If we find that an SVar immediately above a computation
958-- gets drained empty we can switch to parallelizing the computation. For that
959-- we can use a state flag to fork the rest of the computation at any point of
960-- time inside the Monad bind operation if the consumer is running at a faster
963-- TBD Note 3: the binary operation ('parallel') composition allows us to
964-- dispatch a chunkSize of only 1. If we have to dispatch in arbitrary
965-- chunksizes we will need to compose the parallel actions using a data
966-- constructor (A Free container) instead so that we can divide it in chunks of
967-- arbitrary size before dispatching. If the stream is composed of
968-- hierarchically composed grains of different sizes then we can always switch
969-- to a desired granularity depending on the consumer speed.
971-- TBD Note 4: for pure work (when we are not in the IO monad) we can divide it
972-- into just the number of CPUs.
974-- | Join two computations on the currently running 'SVar' queue for concurrent
975-- execution. When we are using parallel composition, an SVar is passed around
976-- as a state variable. We try to schedule a new parallel computation on the
977-- SVar passed to us. The first time, when no SVar exists, a new SVar is
978-- created. Subsequently, 'joinStreamVar2' may get called when a computation
979-- already scheduled on the SVar is further evaluated. For example, when (a
980-- `parallel` b) is evaluated it calls a 'joinStreamVar2' to put 'a' and 'b' on
981-- the current scheduler queue.
983-- The 'SVarStyle' required by the current composition context is passed as one
984-- of the parameters. If the scheduling and composition style of the new
985-- computation being scheduled is different than the style of the current SVar,
986-- then we create a new SVar and schedule it on that. The newly created SVar
987-- joins as one of the computations on the current SVar queue.
989-- Cases when we need to switch to a new SVar:
991-- * (x `parallel` y) `parallel` (t `parallel` u) -- all of them get scheduled on the same SVar
992-- * (x `parallel` y) `parallel` (t `async` u) -- @t@ and @u@ get scheduled on a new child SVar
993-- because of the scheduling policy change.
994-- * if we 'adapt' a stream of type 'async' to a stream of type
995-- 'Parallel', we create a new SVar at the transitioning bind.
996-- * When the stream is switching from disjunctive composition to conjunctive
997-- composition and vice-versa we create a new SVar to isolate the scheduling
1000{-# INLINE joinStreamVar2 #-}
1001joinStreamVar2 :: MonadAsync m
1002 => SVarStyle -> Stream m a -> Stream m a -> Stream m a
1003joinStreamVar2 style m1 m2 = Stream $ \svr stp sng yld ->
1005 Just sv | svarStyle sv == style ->
1006 liftIO ((enqueue sv) m2) >> (runStream m1) svr stp sng yld
1008 sv <- newStreamVar1 style (concurrently m1 m2)
1009 (runStream (fromStreamVar sv)) Nothing stp sng yld
1011 concurrently ma mb = Stream $ \svr stp sng yld -> do
1012 liftIO $ enqueue (fromJust svr) mb
1013 (runStream ma) svr stp sng yld
1015{-# INLINE joinStreamVarPar #-}
1016joinStreamVarPar :: MonadAsync m
1017 => SVarStyle -> Stream m a -> Stream m a -> Stream m a
1018joinStreamVarPar style m1 m2 = Stream $ \svr stp sng yld ->
1020 Just sv | svarStyle sv == style -> do
1021 pushWorkerPar sv m1 >> (runStream m2) svr stp sng yld
1023 sv <- newEmptySVar style
1026 (runStream (fromStreamVar sv)) Nothing stp sng yld
1029ahead :: MonadAsync m => Stream m a -> Stream m a -> Stream m a
1030ahead m1 m2 = Stream $ \svr stp sng yld -> do
1032 Just sv | svarStyle sv == AheadVar -> do
1033 liftIO $ enqueueAhead (workQueue sv) m2
1034 -- Always run the left side on a new SVar to avoid complexity in
1035 -- sequencing results. This means the left side cannot further
1036 -- split into more ahead computations on the same SVar.
1037 (runStream m1) Nothing stp sng yld
1039 sv <- newStreamVarAhead (concurrently m1 m2)
1040 (runStream (fromStreamVar sv)) Nothing stp sng yld
1042 concurrently ma mb = Stream $ \svr stp sng yld -> do
1043 liftIO $ enqueueAhead (workQueue (fromJust svr)) mb
1044 (runStream ma) Nothing stp sng yld
1046------------------------------------------------------------------------------
1047-- Semigroup and Monoid style compositions for parallel actions
1048------------------------------------------------------------------------------
1051async :: MonadAsync m => Stream m a -> Stream m a -> Stream m a
1052async = joinStreamVar2 AsyncVar
1054{-# INLINE wAsync #-}
1055wAsync :: MonadAsync m => Stream m a -> Stream m a -> Stream m a
1056wAsync = joinStreamVar2 WAsyncVar
1058{-# INLINE parallel #-}
1059parallel :: MonadAsync m => Stream m a -> Stream m a -> Stream m a
1060parallel = joinStreamVarPar ParallelVar
1062-------------------------------------------------------------------------------
1063-- Functor instace is the same for all types
1064-------------------------------------------------------------------------------
1066instance Monad m => Functor (Stream m) where
1067 fmap f m = Stream $ \_ stp sng yld ->
1068 let single = sng . f
1069 yield a r = yld (f a) (fmap f r)
1070 in (runStream m) Nothing stp single yield
1072------------------------------------------------------------------------------
1073-- Alternative & MonadPlus
1074------------------------------------------------------------------------------
1076alt :: Stream m a -> Stream m a -> Stream m a
1077alt m1 m2 = Stream $ \_ stp sng yld ->
1078 let stop = runStream m2 Nothing stp sng yld
1079 in runStream m1 Nothing stop sng yld
1081------------------------------------------------------------------------------
1083------------------------------------------------------------------------------
1085zipWith :: (a -> b -> c) -> Stream m a -> Stream m b -> Stream m c
1086zipWith f m1 m2 = go m1 m2
1088 go mx my = Stream $ \_ stp sng yld -> do
1090 let single2 b = sng (f a b)
1091 yield2 b rb = yld (f a b) (go ra rb)
1092 in (runStream my) Nothing stp single2 yield2
1093 let single1 a = merge a nil
1094 yield1 a ra = merge a ra
1095 (runStream mx) Nothing stp single1 yield1
1097mkAsync :: MonadAsync m => Stream m a -> m (Stream m a)
1098mkAsync m = newStreamVar1 AsyncVar m
1099 >>= return . fromStreamVar
1101zipAsyncWith :: MonadAsync m
1102 => (a -> b -> c) -> Stream m a -> Stream m b -> Stream m c
1103zipAsyncWith f m1 m2 = Stream $ \_ stp sng yld -> do
1106 (runStream (zipWith f ma mb)) Nothing stp sng yld
1108-------------------------------------------------------------------------------
1110-------------------------------------------------------------------------------
1112instance MonadTrans Stream where
1115withLocal :: MonadReader r m => (r -> r) -> Stream m a -> Stream m a
1117 Stream $ \svr stp sng yld ->
1118 let single = local f . sng
1119 yield a r = local f $ yld a (withLocal f r)
1120 in (runStream m) svr (local f stp) single yield
1122-- XXX handle and test cross thread state transfer
1125 => Stream m a -> (e -> Stream m a) -> Stream m a
1127 Stream $ \svr stp sng yld ->
1128 let run x = runStream x svr stp sng yield
1129 handle r = r `catchError` \e -> run $ h e
1130 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