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)
68import Control.Concurrent (ThreadId, myThreadId, threadDelay)
69import Control.Concurrent.MVar (MVar, newEmptyMVar, tryTakeMVar,
70 tryPutMVar, takeMVar, putMVar)
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.Heap (Heap, Entry(..))
86import qualified Data.Heap as H
87import Data.IORef (IORef, modifyIORef, newIORef,
88 readIORef, atomicModifyIORef)
89import Data.Maybe (isNothing, fromJust)
90import Data.Semigroup (Semigroup(..))
92import qualified Data.Set as S
93import Prelude hiding (repeat, zipWith)
96import GHC.Conc (ThreadId(..))
99------------------------------------------------------------------------------
100-- Parent child thread communication type
101------------------------------------------------------------------------------
103-- | Events that a child thread may send to a parent thread.
106 | ChildStop ThreadId (Maybe SomeException)
108-- | Sorting out of turn outputs in a heap for Ahead style streams
109data AheadHeapEntry a =
111 | AheadYieldMany a (MVar ()) -- value, MVar to wakeup for more values
113------------------------------------------------------------------------------
114-- State threaded around the monad for thread management
115------------------------------------------------------------------------------
117-- XXX this is redundant as we always start a new SVar on bind
118-- | Conjunction is used for monadic/product style composition. Disjunction is
119-- used for fold/sum style composition. We need to distinguish the two types of
120-- SVars so that the scheduling of the two is independent.
121data SVarTag = Conjunction | Disjunction deriving Eq
123-- XXX use a separate data structure for each type of SVar
125 LIFO -- depth first concurrent
126 | FIFO -- breadth first concurrent
127 | Par -- all parallel
131-- | Identify the type of the SVar. Two computations using the same style can
132-- be scheduled on the same SVar.
133data SVarStyle = SVarStyle SVarTag SVarSched deriving Eq
135-- | An SVar or a Stream Var is a conduit to the output from multiple streams
136-- running concurrently and asynchronously. An SVar can be thought of as an
137-- asynchronous IO handle. We can write any number of streams to an SVar in a
138-- non-blocking manner and then read them back at any time at any pace. The
139-- SVar would run the streams asynchronously and accumulate results. An SVar
140-- may not really execute the stream completely and accumulate all the results.
141-- However, it ensures that the reader can read the results at whatever paces
142-- it wants to read. The SVar monitors and adapts to the consumer's pace.
144-- An SVar is a mini scheduler, it has an associated runqueue that holds the
145-- stream tasks to be picked and run by a pool of worker threads. It has an
146-- associated output queue where the output stream elements are placed by the
147-- worker threads. A doorBell is used by the worker threads to intimate the
148-- consumer thread about availability of new results in the output queue. More
149-- workers are added to the SVar by 'fromStreamVar' on demand if the output
150-- produced is not keeping pace with the consumer. On bounded SVars, workers
151-- block on the output queue to provide throttling of the producer when the
152-- consumer is not pulling fast enough. The number of workers may even get
153-- reduced depending on the consuming pace.
155-- New work is enqueued either at the time of creation of the SVar or as a
156-- result of executing the parallel combinators i.e. '<|' and '<|>' when the
157-- already enqueued computations get evaluated. See 'joinStreamVar2'.
162 svarStyle :: SVarStyle
164 -- Shared output queue (events, length)
165 , outputQueue :: IORef ([ChildEvent a], Int)
166 , doorBell :: MVar () -- signal the consumer about output
167 , siren :: MVar () -- hooter for workers to begin work
169 -- Output synchronization mechanism for Ahead streams (Ahead and
170 -- wAhead). We maintain a heap of out of sequence ahead of time
171 -- generated outputs and the sequence number of the task that is
172 -- currently at the head of the stream. Concurrent execute ahead
173 -- tasks that have a sequence number greater than the task at the
174 -- head should add their output to the heap.
175 , outputHeap :: IORef (Heap (Entry Int (AheadHeapEntry a)), Int)
176 , heapSiren :: MVar () -- hooter for workers to continue
177 -- adding to the heap
180 , workQueue :: IORef ([Stream m a], Int)
181 , enqueue :: Stream m a -> IO ()
182 , queueEmpty :: m Bool
185 -- Shared, thread tracking
186 , runningThreads :: IORef (Set ThreadId)
187 , activeWorkers :: IORef Int
190-- Slightly faster version of CAS. Gained some improvement by avoiding the use
191-- of "evaluate" because we know we do not have exceptions in fn.
192{-# INLINE atomicModifyIORefCAS #-}
193atomicModifyIORefCAS :: IORef a -> (a -> (a,b)) -> IO b
194atomicModifyIORefCAS ref fn = do
195 tkt <- readForCAS ref
201 loop _ 0 = atomicModifyIORef ref fn
203 let (new, result) = fn $ peekTicket old
204 (success, tkt) <- casIORef ref old new
207 else loop tkt (tries - 1)
209------------------------------------------------------------------------------
211------------------------------------------------------------------------------
213-- | The type 'Stream m a' represents a monadic stream of values of type 'a'
214-- constructed using actions in monad 'm'. It uses stop, singleton and yield
215-- continuations equivalent to the following direct style type:
217-- data Stream m a = Stop | Singleton a | Yield a (Stream m a)
219-- To facilitate parallel composition we maintain a local state in an SVar that
220-- is shared across and is used for synchronization of the streams being
223-- The singleton case can be expressed in terms of stop and yield but we have
224-- it as a separate case to optimize composition operations for streams with
225-- single element. We build singleton streams in the implementation of 'pure'
226-- for Applicative and Monad, and in 'lift' for MonadTrans.
230 runStream :: forall r.
231 Maybe (SVar m a) -- local state
233 -> (a -> m r) -- singleton
234 -> (a -> Stream m a -> m r) -- yield
239nil = Stream $ \_ stp _ _ -> stp
241once :: Monad m => m a -> Stream m a
242once m = Stream $ \_ _ single _ -> m >>= single
244{-# INLINE singleton #-}
245-- | Same as @once . return@
246singleton :: a -> Stream m a
247singleton a = Stream $ \_ _ single _ -> single a
249consM :: Monad m => m a -> Stream m a -> Stream m a
250consM m r = Stream $ \_ _ _ yld -> m >>= \a -> yld a r
252-- | Same as @consM . return@
253cons :: a -> Stream m a -> Stream m a
254cons a r = Stream $ \_ _ _ yld -> yld a r
256repeat :: a -> Stream m a
257repeat a = let x = cons a x in x
259------------------------------------------------------------------------------
261------------------------------------------------------------------------------
263-- | Concatenates two streams sequentially i.e. the first stream is
264-- exhausted completely before yielding any element from the second stream.
265serial :: Stream m a -> Stream m a -> Stream m a
268 go (Stream m) = Stream $ \_ stp sng yld ->
269 let stop = (runStream m2) Nothing stp sng yld
271 yield a r = yld a (go r)
272 in m Nothing stop single yield
274instance Semigroup (Stream m a) where
277------------------------------------------------------------------------------
279------------------------------------------------------------------------------
281instance Monoid (Stream m a) where
285------------------------------------------------------------------------------
287------------------------------------------------------------------------------
289wSerial :: Stream m a -> Stream m a -> Stream m a
290wSerial m1 m2 = Stream $ \_ stp sng yld -> do
291 let stop = (runStream m2) Nothing stp sng yld
293 yield a r = yld a (wSerial m2 r)
294 (runStream m1) Nothing stop single yield
296------------------------------------------------------------------------------
297-- Spawning threads and collecting result in streamed fashion
298------------------------------------------------------------------------------
300-- | A monad that can perform concurrent or parallel IO operations. Streams
301-- that can be composed concurrently require the underlying monad to be
305type MonadAsync m = (MonadIO m, MonadBaseControl IO m, MonadThrow m)
307-- Stolen from the async package. The perf improvement is modest, 2% on a
308-- thread heavy benchmark (parallel composition using noop computations).
309-- A version of forkIO that does not include the outer exception
310-- handler: saves a bit of time when we will be installing our own
312{-# INLINE rawForkIO #-}
313rawForkIO :: IO () -> IO ThreadId
314rawForkIO action = IO $ \ s ->
315 case (fork# action s) of (# s1, tid #) -> (# s1, ThreadId tid #)
318doFork :: MonadBaseControl IO m
320 -> (SomeException -> m ())
322doFork action exHandler =
323 EL.mask $ \restore ->
324 liftBaseWith $ \runInIO -> rawForkIO $ do
325 _ <- runInIO $ EL.catch (restore action) exHandler
326 -- XXX restore state here?
329-- XXX exception safety of all atomic/MVar operations
331-- TBD Each worker can have their own queue and the consumer can empty one
332-- queue at a time, that way contention can be reduced.
334-- | This function is used by the producer threads to queue output for the
335-- consumer thread to consume.
337send :: SVar m a -> ChildEvent a -> IO ()
339 len <- atomicModifyIORefCAS (outputQueue sv) $ \(es, n) ->
340 ((msg : es, n + 1), n)
341 if (len <= 0) then do
342 -- XXX need a memory barrier? The wake up must happen only after the
343 -- store has finished otherwise we can have lost wakeup problems.
345 -- Since multiple workers can try this at the same time, it is possible
346 -- that we may put a spurious MVar after the consumer has already seen
347 -- the output. But that's harmless, at worst it may cause the consumer
348 -- to read the queue again and find it empty.
349 -- The important point is that the consumer is guaranteed to receive a
350 -- doorbell if something was added to the queue after it empties it.
351 void $ tryPutMVar (doorBell sv) ()
352 -- those who were waiting because of buffer limit should now go ahead
353 void $ tryPutMVar (siren sv) ()
354 -- The only thread that reduces the length of the queue is the consumer
355 -- thread. It is possible that the queue may have been emptied in the
356 -- window when we acquired "len" and now when we are using it. In that case
357 -- we will be woken up by the consumer blowing the siren.
358 -- Note: current length after adding our output is len + 1
359 else when (len + 1 >= 1500) $ do
361 (_, n) <- readIORef (outputQueue sv)
362 when (n <= 1500) $ void $ tryPutMVar (siren sv) ()
364{-# INLINE sendStop #-}
365sendStop :: SVar m a -> IO ()
366sendStop sv = myThreadId >>= \tid -> send sv (ChildStop tid Nothing)
368-------------------------------------------------------------------------------
370-------------------------------------------------------------------------------
372-- Note: For purely right associated expressions this queue should have at most
373-- one element. It grows to more than one when we have left associcated
374-- expressions. Large left associated compositions can grow this to a
376{-# INLINE enqueueLIFO #-}
377enqueueLIFO :: IORef [Stream m a] -> Stream m a -> IO ()
378enqueueLIFO q m = atomicModifyIORefCAS_ q $ \ ms -> m : ms
380runqueueLIFO :: MonadIO m => SVar m a -> IORef [Stream m a] -> m ()
381runqueueLIFO sv q = run
388 Nothing -> liftIO $ sendStop sv
389 Just m -> (runStream m) (Just sv) run single yield
391 sendit a = liftIO $ send sv (ChildYield a)
392 single a = sendit a >> run
393 yield a r = sendit a >> (runStream r) (Just sv) run single yield
395 dequeue = liftIO $ atomicModifyIORefCAS q $ \case
397 x : xs -> (xs, Just x)
399-------------------------------------------------------------------------------
401-------------------------------------------------------------------------------
403{-# INLINE enqueueFIFO #-}
404enqueueFIFO :: LinkedQueue (Stream m a) -> Stream m a -> IO ()
407runqueueFIFO :: MonadIO m => SVar m a -> LinkedQueue (Stream m a) -> m ()
408runqueueFIFO sv q = run
415 Nothing -> liftIO $ sendStop sv
416 Just m -> (runStream m) (Just sv) run single yield
418 dequeue = liftIO $ tryPopR q
419 sendit a = liftIO $ send sv (ChildYield a)
420 single a = sendit a >> run
421 yield a r = sendit a >> liftIO (enqueueFIFO q r) >> run
423-------------------------------------------------------------------------------
425-------------------------------------------------------------------------------
428runOne :: MonadIO m => SVar m a -> Stream m a -> m ()
429runOne sv m = (runStream m) (Just sv) stop single yield
433 stop = liftIO $ sendStop sv
434 sendit a = liftIO $ send sv (ChildYield a)
435 single a = sendit a >> stop
436 yield a r = sendit a >> runOne sv r
438-------------------------------------------------------------------------------
440-------------------------------------------------------------------------------
442-- Lookahead streams can execute multiple tasks concurrently, ahead of time but
443-- always serves them in the same order as they appear in the stream. To
444-- implement lookahead streams efficiently we assign a sequence number to each
445-- task when the task is picked up for execution. When the task finishes the
446-- output is tagged with the same sequence number and we rearrange the outputs
447-- in sequence based on that number.
449-- To explain the mechanism imagine that the current task at the head of the
450-- stream has a "token" to yield to the output. The ownership of the token is
451-- determined by the current sequence number maintained in outputHeap. Sequence
452-- number is assigned when a task is queued. When a thread dequeues a task it
453-- picks up the sequence number as well and when the output is ready it uses
454-- the sequence number to queue the output to the outputQueue.
456-- The thread with current sequence number sends the output directly to the
457-- outputQueue. Other threads push the output to the outputHeap. When the task
458-- being queued on the heap is a stream of many elements we yield the first
459-- element on the heap and an MVar to wakeup for more elements. When such a
460-- task gets the "token" for outputQueue it can directly keep yielding all the
461-- elements to the outputQueue without checking for the "token".
463-- Note that no two outputs in the heap can have the same sequence numbers and
464-- therefore we do not need a stable heap. We have also separated the buffer
465-- for the current task (outputQueue) and the pending tasks (outputHeap) so
466-- that the pending tasks cannot interfere with the current task. Note that for
467-- a single task just the outputQueue is enough and for the case of many
468-- threads just a heap is good enough. However we balance between these two
469-- cases, so that both are efficient.
471-- For bigger streams it may make sense to have separate buffers for each
472-- stream. However, for singleton streams this may become inefficient. However,
473-- if we do not have separate buffers, then the streams that come later in
474-- sequence may hog the buffer, hindering the streams that are ahead. For this
475-- reason we have a single element buffer limitation for the streams being
476-- executed in advance.
478-- This scheme works pretty efficiently with only 50% more overhead compared to
479-- the Async streams where we do not have any kind of sequencing of the
480-- outputs. It is especially devised so that we are most efficient when we have
481-- short tasks and need just a single thread. Also when a thread yields many
482-- items it can hold lockfree access to the outputQueue and do it efficiently.
484-- XXX Maybe we can start the ahead threads at a lower cpu and IO priority so
485-- that they do not hog the resources and hinder the progress of the threads in
488{-# INLINE sendToHeap #-}
489sendToHeap :: SVar m a -> AheadHeapEntry a -> Int -> IO Bool
490sendToHeap sv msg seqNo = do
491 let maxHeapSize = 1500
493 -- Maybe we can use a lock instead of using CAS
494 -- insertions in the heap may be expensive if the heap is big
495 -- Note: len is valid only when added is True
496 (len, added) <- atomicModifyIORefCAS (outputHeap sv) $
499 then ((H.insert (Entry seqNo msg) h, snum), (H.size h, True))
500 else (ref, (0, False))
502 -- this is only for the case when the consumer generated siren can get
503 -- missed due to race.
504 when (len < maxHeapSize) $ void $ tryPutMVar (heapSiren sv) ()
508 when (len + 1 >= maxHeapSize) $ do
509 takeMVar (heapSiren sv)
510 (h, _) <- readIORef (outputHeap sv)
511 when (H.size h <= maxHeapSize) $ void $ tryPutMVar (siren sv) ()
515 AheadYieldSingle a -> send sv (ChildYield a)
516 AheadYieldMany a mv -> do
517 send sv (ChildYield a)
522 :: IORef (Heap (Entry Int (AheadHeapEntry a)), Int)
524 -> IO (Maybe (AheadHeapEntry a))
525popFromHeap hp nextSeqNo =
526 atomicModifyIORefCAS hp $ \(h, snum) -> do
528 let n = max nextSeqNo snum
530 Nothing -> ((h, n), Nothing)
531 Just (Entry seqNo ev, hp') ->
533 then ((hp', seqNo), Just ev)
534 else ((h, n), Nothing)
536-- | As soon as we set the nextSeqNo in outputHeap we have transferred the
537-- output "token" to the thread executing the task with that sequence number.
538-- We have to make sure that there is no entry in the heap that has nextSeqNo
539-- before we set nextSeqNo as the expected sequence number in the heap. It is
540-- possible that when we are clearing the heap entry of the current sequence
541-- number the next sequence number gets added and then we have to clear that,
542-- this can keep going on and on.
543drainHeap :: SVar m a -> Int -> IO ()
544drainHeap sv nextSeqNo = do
545 r <- popFromHeap (outputHeap sv) nextSeqNo
550 AheadYieldSingle a -> do
551 send sv (ChildYield a)
552 atomicModifyIORefCAS_ (outputHeap sv) $ \(h, snum) ->
554 drainHeap sv nextSeqNo
555 AheadYieldMany a mv -> do
556 send sv (ChildYield a)
559-- | The thread that has the token has the responsibility to move any pending
560-- entries in the heap to the outputQueue. When the token is assigned to a
561-- sequence number we must not have any entries for that sequence number in the
563releaseToken :: SVar m a -> Int -> IO ()
564releaseToken sv nextSeqNo = do
565 drainHeap sv nextSeqNo
566 void $ tryPutMVar (heapSiren sv) ()
568-- | When we have the output "token" we are allowed to append to the
569-- outputQueue otherwise we have to insert the output in the outputHeap.
570-- Returns whether we still have the token.
571handleOutput :: SVar m a
578handleOutput sv haveToken prevSeqNo seqNo a ev = do
581 let nextSeqNo = prevSeqNo + 1
582 if seqNo == nextSeqNo
583 then send sv (ChildYield a) >> return True
585 releaseToken sv nextSeqNo
588 where toHeap = sendToHeap sv ev seqNo
590-- Left associated ahead expressions are expensive. We start a new SVar for
591-- each left associative expression. The queue is used only for right
592-- associated expression, we queue the right expression and execute the left.
593-- Thererefore the queue never has more than on item in it.
594{-# INLINE enqueueAhead #-}
595enqueueAhead :: IORef ([Stream m a], Int) -> Stream m a -> IO ()
597 atomicModifyIORefCAS_ q $ \ case
598 ([], n) -> ([m], n + 1) -- increment sequence
599 _ -> error "not empty"
601runqueueAhead :: MonadIO m => SVar m a -> IORef ([Stream m a], Int) -> m ()
602runqueueAhead sv q = run False (-1)
606 run haveToken prevSeqNo = do
609 Nothing -> liftIO $ do
610 when haveToken $ releaseToken sv (prevSeqNo + 1)
612 Just (m, seqNo) -> do
613 let tStatus = haveToken || seqNo == 0
615 let nextSeqNo = prevSeqNo + 1
616 if tStatus && seqNo /= nextSeqNo
618 liftIO $ releaseToken sv nextSeqNo
620 else run tStatus seqNo
622 tokenStatus <- liftIO $
623 handleOutput sv tStatus prevSeqNo seqNo a
625 run tokenStatus seqNo
627 -- If we are yielding more than one item and we do not have the
628 -- token then we insert the current item in the pending
629 -- outputHeap and wait for our turn.
631 stopNext = run True seqNo
633 liftIO $ send sv (ChildYield a)
636 liftIO $ send sv (ChildYield a)
637 (runStream r) (Just sv) stopNext singleNext yieldNext
640 -- XXX we can allocate the MVar when actually inserting
641 -- in the heap and then return it.
645 handleOutput sv tStatus prevSeqNo seqNo a
646 (AheadYieldMany a mv)
647 when (not gotToken) $ void $ takeMVar mv
648 (runStream r) (Just sv) stopNext singleNext yieldNext
650 (runStream m) (Just sv) stop single yield
652 dequeue = liftIO $ do
653 atomicModifyIORefCAS q $ \case
654 ([], n) -> (([], n), Nothing)
655 (x : [], n) -> (([], n), Just (x, n))
656 _ -> error "more than one item on queue"
658-- Thread tracking is needed for two reasons:
660-- 1) Killing threads on exceptions. Threads may not be left to go away by
661-- themselves because they may run for significant times before going away or
662-- worse they may be stuck in IO and never go away.
664-- 2) To know when all threads are done and the stream has ended.
666{-# NOINLINE addThread #-}
667addThread :: MonadIO m => SVar m a -> ThreadId -> m ()
669 liftIO $ modifyIORef (runningThreads sv) (S.insert tid)
672{-# INLINE delThread #-}
673delThread :: MonadIO m => SVar m a -> ThreadId -> m ()
675 liftIO $ modifyIORef (runningThreads sv) $ (\s -> S.delete tid s)
678-- If present then delete else add. This takes care of out of order add and
679-- delete i.e. a delete arriving before we even added a thread.
680-- This occurs when the forked thread is done even before the 'addThread' right
681-- after the fork is executed.
682{-# INLINE modifyThread #-}
683modifyThread :: MonadIO m => SVar m a -> ThreadId -> m ()
684modifyThread sv tid = do
685 changed <- liftIO $ atomicModifyIORefCAS (runningThreads sv) $ \old ->
686 if (S.member tid old)
687 then let new = (S.delete tid old) in (new, new)
688 else let new = (S.insert tid old) in (new, old)
690 then liftIO $ void $ tryPutMVar (doorBell sv) ()
693-- | This is safe even if we are adding more threads concurrently because if
694-- a child thread is adding another thread then anyway 'runningThreads' will
696{-# INLINE allThreadsDone #-}
697allThreadsDone :: MonadIO m => SVar m a -> m Bool
698allThreadsDone sv = liftIO $ S.null <$> readIORef (runningThreads sv)
700{-# NOINLINE handleChildException #-}
701handleChildException :: MonadIO m => SVar m a -> SomeException -> m ()
702handleChildException sv e = liftIO $ do
704 send sv (ChildStop tid (Just e))
706{-# NOINLINE pushWorker #-}
707pushWorker :: MonadAsync m => SVar m a -> m ()
709 liftIO $ atomicModifyIORefCAS_ (activeWorkers sv) $ \n -> n + 1
710 doFork (runqueue sv) (handleChildException sv) >>= addThread sv
712-- | In contrast to pushWorker which always happens only from the consumer
713-- thread, a pushWorkerPar can happen concurrently from multiple threads on the
714-- producer side. So we need to use a thread safe modification of
715-- runningThreads. Alternatively, we can use a CreateThread event to avoid
716-- using a CAS based modification.
717{-# NOINLINE pushWorkerPar #-}
718pushWorkerPar :: MonadAsync m => SVar m a -> Stream m a -> m ()
720 doFork (runOne sv m) (handleChildException sv) >>= modifyThread sv
722-- XXX When the queue is LIFO we can put a limit on the number of dispatches.
723-- Also, if a worker blocks on the output queue we can decide if we want to
724-- block or make it go away entirely, depending on the number of workers and
725-- the type of the queue.
726{-# INLINE sendWorkerWait #-}
727sendWorkerWait :: MonadAsync m => SVar m a -> m ()
728sendWorkerWait sv = do
729 -- When there is no output seen we dispatch more workers to help out if
730 -- there is work pending in the work queue. But we wait a little while
731 -- and check the output again so that we are not too aggressive.
732 -- If there is no output pending to process and there is no worker to be
733 -- sent then we block, so that we do not keep looping fruitlessly.
735 -- XXX we should wait in such a way so that we can be immediately
736 -- interrupted by a doorBell.
737 liftIO $ threadDelay 200
738 (_, n) <- liftIO $ readIORef (outputQueue sv)
740 -- XXX the queue may be empty temporarily if the worker has dequeued
741 -- the work item but has not enqueued the remaining part yet. For the
742 -- same reason, a worker may come back if it tries to dequeue when the
743 -- queue is empty, even though the whole work has not finished yet.
744 done <- queueEmpty sv
747 cnt <- liftIO $ readIORef $ activeWorkers sv
752 else liftIO $ takeMVar (doorBell sv)
753 else liftIO $ takeMVar (doorBell sv)
755-- | Pull a stream from an SVar.
756{-# NOINLINE fromStreamVar #-}
757fromStreamVar :: MonadAsync m => SVar m a -> Stream m a
758fromStreamVar sv = Stream $ \_ stp sng yld -> do
759 let SVarStyle _ sched = svarStyle sv
761 then liftIO $ takeMVar (doorBell sv)
763 res <- liftIO $ tryTakeMVar (doorBell sv)
764 when (isNothing res) $ sendWorkerWait sv
766 (list, _) <- liftIO $ atomicModifyIORefCAS (outputQueue sv) $ \x -> (([],0), x)
767 void $ liftIO $ tryPutMVar (siren sv) ()
768 -- Reversing the output is important to guarantee that we process the
769 -- outputs in the same order as they were generated by the constituent
771 (runStream $ processEvents (reverse list)) Nothing stp sng yld
775 handleException e tid = do
776 liftIO $ atomicModifyIORefCAS_ (activeWorkers sv) $ \n -> n - 1
778 -- XXX implement kill async exception handling
779 -- liftIO $ readIORef (runningThreads sv) >>= mapM_ killThread
782 {-# INLINE processEvents #-}
783 processEvents [] = Stream $ \_ stp sng yld -> do
784 done <- allThreadsDone sv
786 then (runStream (fromStreamVar sv)) Nothing stp sng yld
789 processEvents (ev : es) = Stream $ \_ stp sng yld -> do
790 let continue = (runStream (processEvents es)) Nothing stp sng yld
791 yield a = yld a (processEvents es)
794 ChildYield a -> yield a
798 let active = activeWorkers sv
799 liftIO $ atomicModifyIORefCAS_ active $ \n -> n - 1
800 modifyThread sv tid >> continue
801 Just ex -> handleException ex tid
803getFifoSVar :: MonadIO m => SVarStyle -> IO (SVar m a)
804getFifoSVar ctype = do
805 outQ <- newIORef ([], 0)
806 outQMv <- newEmptyMVar
807 hooter <- newEmptyMVar
809 running <- newIORef S.empty
812 SVar { outputQueue = outQ
815 , outputHeap = undefined
816 , heapSiren = undefined
817 , runningThreads = running
818 , workQueue = undefined
819 , runqueue = runqueueFIFO sv q
821 , queueEmpty = liftIO $ nullQ q
823 , activeWorkers = active
827getLifoSVar :: MonadIO m => SVarStyle -> IO (SVar m a)
828getLifoSVar ctype = do
829 outQ <- newIORef ([], 0)
830 outQMv <- newEmptyMVar
831 hooter <- newEmptyMVar
833 running <- newIORef S.empty
835 let checkEmpty = null <$> liftIO (readIORef q)
837 SVar { outputQueue = outQ
840 , outputHeap = undefined
841 , heapSiren = undefined
842 , runningThreads = running
843 , workQueue = undefined
844 , runqueue = runqueueLIFO sv q
845 , enqueue = enqueueLIFO q
846 , queueEmpty = checkEmpty
848 , activeWorkers = active
852getParSVar :: SVarStyle -> IO (SVar m a)
854 outQ <- newIORef ([], 0)
855 outQMv <- newEmptyMVar
856 hooter <- newEmptyMVar
858 running <- newIORef S.empty
860 SVar { outputQueue = outQ
863 , outputHeap = undefined
864 , heapSiren = undefined
865 , runningThreads = running
866 , workQueue = undefined
867 , runqueue = undefined
868 , enqueue = undefined
869 , queueEmpty = undefined
871 , activeWorkers = active
875getAheadSVar :: MonadIO m => SVarStyle -> IO (SVar m a)
876getAheadSVar style = do
877 outQ <- newIORef ([], 0)
878 outH <- newIORef (H.empty, 0)
879 outQMv <- newEmptyMVar
880 hooter <- newEmptyMVar
881 hsiren <- newEmptyMVar
883 running <- newIORef S.empty
884 q <- newIORef ([], -1)
885 let checkEmpty = liftIO $ do
886 (xs, _) <- readIORef q
889 SVar { outputQueue = outQ
894 , runningThreads = running
896 , runqueue = runqueueAhead sv q
897 , enqueue = undefined
898 , queueEmpty = checkEmpty
900 , activeWorkers = active
904-- | Create a new empty SVar.
905newEmptySVar :: MonadAsync m => SVarStyle -> m (SVar m a)
906newEmptySVar style = do
909 SVarStyle _ FIFO -> getFifoSVar style
910 SVarStyle _ LIFO -> getLifoSVar style
911 SVarStyle _ Par -> getParSVar style
912 SVarStyle _ SVarAhead -> getAheadSVar style
914-- | Create a new SVar and enqueue one stream computation on it.
915newStreamVar1 :: MonadAsync m => SVarStyle -> Stream m a -> m (SVar m a)
916newStreamVar1 style m = do
917 sv <- newEmptySVar style
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 $ (enqueue sv) m
925-- | Create a new SVar and enqueue one stream computation on it.
926newStreamVarAhead :: MonadAsync m => Stream m a -> m (SVar m a)
927newStreamVarAhead m = do
928 sv <- newEmptySVar (SVarStyle Disjunction SVarAhead)
929 -- Note: We must have all the work on the queue before sending the
930 -- pushworker, otherwise the pushworker may exit before we even get a
932 liftIO $ enqueueAhead (workQueue sv) m
936-- | Write a stream to an 'SVar' in a non-blocking manner. The stream can then
937-- be read back from the SVar using 'fromSVar'.
938toStreamVar :: MonadAsync m => SVar m a -> Stream m a -> m ()
940 liftIO $ (enqueue sv) m
941 done <- allThreadsDone sv
942 -- XXX This is safe only when called from the consumer thread or when no
943 -- consumer is present. There may be a race if we are not running in the
945 when done $ pushWorker sv
947------------------------------------------------------------------------------
948-- Running streams concurrently
949------------------------------------------------------------------------------
951-- Concurrency rate control.
953-- Our objective is to create more threads on demand if the consumer is running
954-- faster than us. As soon as we encounter a concurrent composition we create a
955-- push pull pair of threads. We use an SVar for communication between the
956-- consumer, pulling from the SVar and the producer who is pushing to the SVar.
957-- The producer creates more threads if the SVar drains and becomes empty, that
958-- is the consumer is running faster.
960-- XXX Note 1: This mechanism can be problematic if the initial production
961-- latency is high, we may end up creating too many threads. So we need some
962-- way to monitor and use the latency as well. Having a limit on the dispatches
963-- (programmer controlled) may also help.
965-- TBD Note 2: We may want to run computations at the lower level of the
966-- composition tree serially even when they are composed using a parallel
967-- combinator. We can use 'serial' in place of 'async' and 'wSerial' in
968-- place of 'wAsync'. If we find that an SVar immediately above a computation
969-- gets drained empty we can switch to parallelizing the computation. For that
970-- we can use a state flag to fork the rest of the computation at any point of
971-- time inside the Monad bind operation if the consumer is running at a faster
974-- TBD Note 3: the binary operation ('parallel') composition allows us to
975-- dispatch a chunkSize of only 1. If we have to dispatch in arbitrary
976-- chunksizes we will need to compose the parallel actions using a data
977-- constructor (A Free container) instead so that we can divide it in chunks of
978-- arbitrary size before dispatching. If the stream is composed of
979-- hierarchically composed grains of different sizes then we can always switch
980-- to a desired granularity depending on the consumer speed.
982-- TBD Note 4: for pure work (when we are not in the IO monad) we can divide it
983-- into just the number of CPUs.
985-- | Join two computations on the currently running 'SVar' queue for concurrent
986-- execution. When we are using parallel composition, an SVar is passed around
987-- as a state variable. We try to schedule a new parallel computation on the
988-- SVar passed to us. The first time, when no SVar exists, a new SVar is
989-- created. Subsequently, 'joinStreamVar2' may get called when a computation
990-- already scheduled on the SVar is further evaluated. For example, when (a
991-- `parallel` b) is evaluated it calls a 'joinStreamVar2' to put 'a' and 'b' on
992-- the current scheduler queue.
994-- The 'SVarStyle' required by the current composition context is passed as one
995-- of the parameters. If the scheduling and composition style of the new
996-- computation being scheduled is different than the style of the current SVar,
997-- then we create a new SVar and schedule it on that. The newly created SVar
998-- joins as one of the computations on the current SVar queue.
1000-- Cases when we need to switch to a new SVar:
1002-- * (x `parallel` y) `parallel` (t `parallel` u) -- all of them get scheduled on the same SVar
1003-- * (x `parallel` y) `parallel` (t `async` u) -- @t@ and @u@ get scheduled on a new child SVar
1004-- because of the scheduling policy change.
1005-- * if we 'adapt' a stream of type 'async' to a stream of type
1006-- 'Parallel', we create a new SVar at the transitioning bind.
1007-- * When the stream is switching from disjunctive composition to conjunctive
1008-- composition and vice-versa we create a new SVar to isolate the scheduling
1011{-# INLINE joinStreamVar2 #-}
1012joinStreamVar2 :: MonadAsync m
1013 => SVarStyle -> Stream m a -> Stream m a -> Stream m a
1014joinStreamVar2 style m1 m2 = Stream $ \svr stp sng yld ->
1016 Just sv | svarStyle sv == style ->
1017 liftIO ((enqueue sv) m2) >> (runStream m1) svr stp sng yld
1019 sv <- newStreamVar1 style (concurrently m1 m2)
1020 (runStream (fromStreamVar sv)) Nothing stp sng yld
1022 concurrently ma mb = Stream $ \svr stp sng yld -> do
1023 liftIO $ enqueue (fromJust svr) mb
1024 (runStream ma) svr stp sng yld
1026{-# INLINE joinStreamVarPar #-}
1027joinStreamVarPar :: MonadAsync m
1028 => SVarStyle -> Stream m a -> Stream m a -> Stream m a
1029joinStreamVarPar style m1 m2 = Stream $ \svr stp sng yld ->
1031 Just sv | svarStyle sv == style -> do
1032 pushWorkerPar sv m1 >> (runStream m2) svr stp sng yld
1034 sv <- newEmptySVar style
1037 (runStream (fromStreamVar sv)) Nothing stp sng yld
1040ahead :: MonadAsync m => Stream m a -> Stream m a -> Stream m a
1041ahead m1 m2 = Stream $ \svr stp sng yld -> do
1042 let style = SVarStyle Disjunction SVarAhead
1044 Just sv | svarStyle sv == style -> do
1045 liftIO $ enqueueAhead (workQueue sv) m2
1046 -- Always run the left side on a new SVar to avoid complexity in
1047 -- sequencing results. This means the left side cannot further
1048 -- split into more ahead computations on the same SVar.
1049 (runStream m1) Nothing stp sng yld
1051 sv <- newStreamVarAhead (concurrently m1 m2)
1052 (runStream (fromStreamVar sv)) Nothing stp sng yld
1054 concurrently ma mb = Stream $ \svr stp sng yld -> do
1055 liftIO $ enqueueAhead (workQueue (fromJust svr)) mb
1056 (runStream ma) Nothing stp sng yld
1058------------------------------------------------------------------------------
1059-- Semigroup and Monoid style compositions for parallel actions
1060------------------------------------------------------------------------------
1063async :: MonadAsync m => Stream m a -> Stream m a -> Stream m a
1064async = joinStreamVar2 (SVarStyle Disjunction LIFO)
1066{-# INLINE wAsync #-}
1067wAsync :: MonadAsync m => Stream m a -> Stream m a -> Stream m a
1068wAsync = joinStreamVar2 (SVarStyle Disjunction FIFO)
1070{-# INLINE parallel #-}
1071parallel :: MonadAsync m => Stream m a -> Stream m a -> Stream m a
1072parallel = joinStreamVarPar (SVarStyle Disjunction Par)
1074-------------------------------------------------------------------------------
1075-- Functor instace is the same for all types
1076-------------------------------------------------------------------------------
1078instance Monad m => Functor (Stream m) where
1079 fmap f m = Stream $ \_ stp sng yld ->
1080 let single = sng . f
1081 yield a r = yld (f a) (fmap f r)
1082 in (runStream m) Nothing stp single yield
1084------------------------------------------------------------------------------
1085-- Alternative & MonadPlus
1086------------------------------------------------------------------------------
1088alt :: Stream m a -> Stream m a -> Stream m a
1089alt m1 m2 = Stream $ \_ stp sng yld ->
1090 let stop = runStream m2 Nothing stp sng yld
1091 in runStream m1 Nothing stop sng yld
1093------------------------------------------------------------------------------
1095------------------------------------------------------------------------------
1097zipWith :: (a -> b -> c) -> Stream m a -> Stream m b -> Stream m c
1098zipWith f m1 m2 = go m1 m2
1100 go mx my = Stream $ \_ stp sng yld -> do
1102 let single2 b = sng (f a b)
1103 yield2 b rb = yld (f a b) (go ra rb)
1104 in (runStream my) Nothing stp single2 yield2
1105 let single1 a = merge a nil
1106 yield1 a ra = merge a ra
1107 (runStream mx) Nothing stp single1 yield1
1109mkAsync :: MonadAsync m => Stream m a -> m (Stream m a)
1110mkAsync m = newStreamVar1 (SVarStyle Disjunction LIFO) m
1111 >>= return . fromStreamVar
1113zipAsyncWith :: MonadAsync m
1114 => (a -> b -> c) -> Stream m a -> Stream m b -> Stream m c
1115zipAsyncWith f m1 m2 = Stream $ \_ stp sng yld -> do
1118 (runStream (zipWith f ma mb)) Nothing stp sng yld
1120-------------------------------------------------------------------------------
1122-------------------------------------------------------------------------------
1124instance MonadTrans Stream where
1127withLocal :: MonadReader r m => (r -> r) -> Stream m a -> Stream m a
1129 Stream $ \svr stp sng yld ->
1130 let single = local f . sng
1131 yield a r = local f $ yld a (withLocal f r)
1132 in (runStream m) svr (local f stp) single yield
1134-- XXX handle and test cross thread state transfer
1137 => Stream m a -> (e -> Stream m a) -> Stream m a
1139 Stream $ \svr stp sng yld ->
1140 let run x = runStream x svr stp sng yield
1141 handle r = r `catchError` \e -> run $ h e
1142 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