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
55 -- XXX move to Streams.hs
60 -- * Concurrent Stream Vars (SVars)
69-- MVar diagnostics has some overhead - around 5% on asyncly null benchmark,
70-- but keep it on for now to debug problems quickly if and when they happen.
73import Control.Concurrent (ThreadId, myThreadId, threadDelay)
74import Control.Concurrent.MVar (MVar, newEmptyMVar, tryTakeMVar,
75 tryPutMVar, takeMVar, putMVar)
76import Control.Exception (SomeException (..))
78import Control.Exception (catch,
79 BlockedIndefinitelyOnMVar(..))
81import qualified Control.Exception.Lifted as EL
82import Control.Monad (when)
83import Control.Monad.Catch (MonadThrow, throwM)
84import Control.Monad.Error.Class (MonadError(..))
85import Control.Monad.IO.Class (MonadIO(..))
86import Control.Monad.Reader.Class (MonadReader(..))
87import Control.Monad.Trans.Class (MonadTrans (lift))
88import Control.Monad.Trans.Control (MonadBaseControl, liftBaseWith)
89import Data.Atomics (casIORef, readForCAS, peekTicket
90 ,atomicModifyIORefCAS_)
91import Data.Concurrent.Queue.MichaelScott (LinkedQueue, newQ, pushL,
93import Data.Functor (void)
94import Data.Heap (Heap, Entry(..))
95import qualified Data.Heap as H
96import Data.IORef (IORef, modifyIORef, newIORef,
97 readIORef, atomicModifyIORef)
98import Data.Maybe (isNothing, fromJust)
99import Data.Semigroup (Semigroup(..))
101import qualified Data.Set as S
102import Prelude hiding (repeat, zipWith)
105import GHC.Conc (ThreadId(..))
106import GHC.IO (IO(..))
108------------------------------------------------------------------------------
109-- Parent child thread communication type
110------------------------------------------------------------------------------
112-- | Events that a child thread may send to a parent thread.
115 | ChildStop ThreadId (Maybe SomeException)
117-- | Sorting out of turn outputs in a heap for Ahead style streams
118data AheadHeapEntry a =
120 | AheadYieldMany a (MVar ()) -- value, MVar to wakeup for more values
122------------------------------------------------------------------------------
123-- State threaded around the monad for thread management
124------------------------------------------------------------------------------
126-- XXX use a separate data structure for each type of SVar
127-- | Identify the type of the SVar. Two computations using the same style can
128-- be scheduled on the same SVar.
130 AsyncVar -- depth first concurrent
131 | WAsyncVar -- breadth first concurrent
132 | ParallelVar -- all parallel
133 | AheadVar -- Concurrent look ahead
136-- | An SVar or a Stream Var is a conduit to the output from multiple streams
137-- running concurrently and asynchronously. An SVar can be thought of as an
138-- asynchronous IO handle. We can write any number of streams to an SVar in a
139-- non-blocking manner and then read them back at any time at any pace. The
140-- SVar would run the streams asynchronously and accumulate results. An SVar
141-- may not really execute the stream completely and accumulate all the results.
142-- However, it ensures that the reader can read the results at whatever paces
143-- it wants to read. The SVar monitors and adapts to the consumer's pace.
145-- An SVar is a mini scheduler, it has an associated runqueue that holds the
146-- stream tasks to be picked and run by a pool of worker threads. It has an
147-- associated output queue where the output stream elements are placed by the
148-- worker threads. A doorBell is used by the worker threads to intimate the
149-- consumer thread about availability of new results in the output queue. More
150-- workers are added to the SVar by 'fromStreamVar' on demand if the output
151-- produced is not keeping pace with the consumer. On bounded SVars, workers
152-- block on the output queue to provide throttling of the producer when the
153-- consumer is not pulling fast enough. The number of workers may even get
154-- reduced depending on the consuming pace.
156-- New work is enqueued either at the time of creation of the SVar or as a
157-- result of executing the parallel combinators i.e. '<|' and '<|>' when the
158-- already enqueued computations get evaluated. See 'joinStreamVarAsync'.
163 svarStyle :: SVarStyle
165 -- Shared output queue (events, length)
166 , outputQueue :: IORef ([ChildEvent a], Int)
167 , doorBell :: MVar () -- signal the consumer about output
168 , siren :: MVar () -- hooter for workers to begin work
170 -- Output synchronization mechanism for Ahead streams (Ahead and
171 -- wAhead). We maintain a heap of out of sequence ahead of time
172 -- generated outputs and the sequence number of the task that is
173 -- currently at the head of the stream. Concurrent execute ahead
174 -- tasks that have a sequence number greater than the task at the
175 -- head should add their output to the heap.
176 , outputHeap :: IORef (Heap (Entry Int (AheadHeapEntry a)), Int)
177 , heapSiren :: MVar () -- hooter for workers to continue
178 -- adding to the heap
181 , workQueue :: IORef ([Stream m a], Int)
182 , enqueue :: Stream m a -> IO ()
183 , queueEmpty :: m Bool
184 , waitingForWork :: IORef Bool
187 -- Shared, thread tracking
188 , runningThreads :: IORef (Set ThreadId)
189 , activeWorkers :: IORef Int
193{-# NOINLINE dumpSVar #-}
194dumpSVar :: SVar m a -> IO String
196 (oqList, oqLen) <- readIORef $ outputQueue sv
197 db <- tryTakeMVar $ doorBell sv
198 srn <- tryTakeMVar $ siren sv
199 (oheap, oheapSize) <- readIORef $ outputHeap sv
200 hsrn <- tryTakeMVar $ heapSiren sv
201 (wq, wqSize) <- readIORef $ workQueue sv
202 waiting <- readIORef $ waitingForWork sv
203 rthread <- readIORef $ runningThreads sv
204 workers <- readIORef $ activeWorkers sv
205 -- XXX queueEmpty should be made IO return type
208 [ "style = " ++ show (svarStyle sv)
209 , "outputQueue length computed = " ++ show (length oqList)
210 , "outputQueue length maintained = " ++ show oqLen
211 , "output doorBell = " ++ show db
212 , "worker siren = " ++ show srn
213 , "heap length computed = " ++ show (H.size oheap)
214 , "heap length maintained = " ++ show oheapSize
215 , "heap siren = " ++ show hsrn
216 , "work queue length computed = " ++ show (length wq)
217 , "work queue length maintained = " ++ show wqSize
218 , "waitingForWork = " ++ show waiting
219 , "running threads = " ++ show rthread
220 , "running thread count = " ++ show workers
223{-# NOINLINE mvarExcHandler #-}
224mvarExcHandler :: SVar m a -> String -> BlockedIndefinitelyOnMVar -> IO ()
225mvarExcHandler sv label BlockedIndefinitelyOnMVar = do
226 svInfo <- dumpSVar sv
227 error $ label ++ " " ++ "BlockedIndefinitelyOnMVar\n" ++ svInfo
229withDBGMVar :: SVar m a -> String -> IO () -> IO ()
230withDBGMVar sv label action = action `catch` mvarExcHandler sv label
232withDBGMVar :: SVar m a -> String -> IO () -> IO ()
233withDBGMVar _ _ action = action
236-- Slightly faster version of CAS. Gained some improvement by avoiding the use
237-- of "evaluate" because we know we do not have exceptions in fn.
238{-# INLINE atomicModifyIORefCAS #-}
239atomicModifyIORefCAS :: IORef a -> (a -> (a,b)) -> IO b
240atomicModifyIORefCAS ref fn = do
241 tkt <- readForCAS ref
247 loop _ 0 = atomicModifyIORef ref fn
249 let (new, result) = fn $ peekTicket old
250 (success, tkt) <- casIORef ref old new
253 else loop tkt (tries - 1)
255------------------------------------------------------------------------------
257------------------------------------------------------------------------------
259-- | The type 'Stream m a' represents a monadic stream of values of type 'a'
260-- constructed using actions in monad 'm'. It uses stop, singleton and yield
261-- continuations equivalent to the following direct style type:
263-- data Stream m a = Stop | Singleton a | Yield a (Stream m a)
265-- To facilitate parallel composition we maintain a local state in an SVar that
266-- is shared across and is used for synchronization of the streams being
269-- The singleton case can be expressed in terms of stop and yield but we have
270-- it as a separate case to optimize composition operations for streams with
271-- single element. We build singleton streams in the implementation of 'pure'
272-- for Applicative and Monad, and in 'lift' for MonadTrans.
276 runStream :: forall r.
277 Maybe (SVar m a) -- local state
279 -> (a -> m r) -- singleton
280 -> (a -> Stream m a -> m r) -- yield
285nil = Stream $ \_ stp _ _ -> stp
287-- | faster than consM because there is no bind.
288cons :: a -> Stream m a -> Stream m a
289cons a r = Stream $ \_ _ _ yld -> yld a r
291-- | Same as @once . return@ but may be faster because there is no bind
292singleton :: a -> Stream m a
293singleton a = Stream $ \_ _ single _ -> single a
296once :: Monad m => m a -> Stream m a
297once m = Stream $ \_ _ single _ -> m >>= single
300consM :: Monad m => m a -> Stream m a -> Stream m a
301consM m r = Stream $ \_ _ _ yld -> m >>= \a -> yld a r
303repeat :: a -> Stream m a
304repeat a = let x = cons a x in x
306------------------------------------------------------------------------------
308------------------------------------------------------------------------------
310-- | Concatenates two streams sequentially i.e. the first stream is
311-- exhausted completely before yielding any element from the second stream.
313serial :: Stream m a -> Stream m a -> Stream m a
316 go (Stream m) = Stream $ \_ stp sng yld ->
317 let stop = (runStream m2) Nothing stp sng yld
319 yield a r = yld a (go r)
320 in m Nothing stop single yield
322instance Semigroup (Stream m a) where
325------------------------------------------------------------------------------
327------------------------------------------------------------------------------
329instance Monoid (Stream m a) where
333------------------------------------------------------------------------------
335------------------------------------------------------------------------------
337{-# INLINE wSerial #-}
338wSerial :: Stream m a -> Stream m a -> Stream m a
339wSerial m1 m2 = Stream $ \_ stp sng yld -> do
340 let stop = (runStream m2) Nothing stp sng yld
342 yield a r = yld a (wSerial m2 r)
343 (runStream m1) Nothing stop single yield
345------------------------------------------------------------------------------
346-- Spawning threads and collecting result in streamed fashion
347------------------------------------------------------------------------------
349-- | A monad that can perform concurrent or parallel IO operations. Streams
350-- that can be composed concurrently require the underlying monad to be
354type MonadAsync m = (MonadIO m, MonadBaseControl IO m, MonadThrow m)
356-- Stolen from the async package. The perf improvement is modest, 2% on a
357-- thread heavy benchmark (parallel composition using noop computations).
358-- A version of forkIO that does not include the outer exception
359-- handler: saves a bit of time when we will be installing our own
361{-# INLINE rawForkIO #-}
362rawForkIO :: IO () -> IO ThreadId
363rawForkIO action = IO $ \ s ->
364 case (fork# action s) of (# s1, tid #) -> (# s1, ThreadId tid #)
367doFork :: MonadBaseControl IO m
369 -> (SomeException -> m ())
371doFork action exHandler =
372 EL.mask $ \restore ->
373 liftBaseWith $ \runInIO -> rawForkIO $ do
374 _ <- runInIO $ EL.catch (restore action) exHandler
375 -- XXX restore state here?
378-- XXX exception safety of all atomic/MVar operations
380-- TBD Each worker can have their own queue and the consumer can empty one
381-- queue at a time, that way contention can be reduced.
383-- | This function is used by the producer threads to queue output for the
384-- consumer thread to consume.
386send :: SVar m a -> ChildEvent a -> IO ()
388 len <- atomicModifyIORefCAS (outputQueue sv) $ \(es, n) ->
389 ((msg : es, n + 1), n)
390 if (len <= 0) then do
391 -- XXX need a memory barrier? The wake up must happen only after the
392 -- store has finished otherwise we can have lost wakeup problems.
394 -- Since multiple workers can try this at the same time, it is possible
395 -- that we may put a spurious MVar after the consumer has already seen
396 -- the output. But that's harmless, at worst it may cause the consumer
397 -- to read the queue again and find it empty.
398 -- The important point is that the consumer is guaranteed to receive a
399 -- doorbell if something was added to the queue after it empties it.
400 void $ tryPutMVar (doorBell sv) ()
401 -- those who were waiting because of buffer limit should now go ahead
402 void $ tryPutMVar (siren sv) ()
403 -- The only thread that reduces the length of the queue is the consumer
404 -- thread. It is possible that the queue may have been emptied in the
405 -- window when we acquired "len" and now when we are using it. In that case
406 -- we will be woken up by the consumer blowing the siren.
407 -- Note: current length after adding our output is len + 1
408 else when (len + 1 >= 1500) $ do
409 withDBGMVar sv "send: wait for siren" $ takeMVar (siren sv)
410 (_, n) <- readIORef (outputQueue sv)
411 when (n <= 1500) $ void $ tryPutMVar (siren sv) ()
413{-# INLINE sendStop #-}
414sendStop :: SVar m a -> IO ()
415sendStop sv = myThreadId >>= \tid -> send sv (ChildStop tid Nothing)
417-------------------------------------------------------------------------------
419-------------------------------------------------------------------------------
421-- Note: For purely right associated expressions this queue should have at most
422-- one element. It grows to more than one when we have left associcated
423-- expressions. Large left associated compositions can grow this to a
425{-# INLINE enqueueLIFO #-}
426enqueueLIFO :: SVar m a -> IORef [Stream m a] -> Stream m a -> IO ()
427enqueueLIFO sv q m = do
428 w <- readIORef $ waitingForWork sv
431 else atomicModifyIORefCAS_ q $ \ms -> m : ms
436 v <- atomicModifyIORefCAS q $ \ ms ->
440 when v $ void $ tryPutMVar (doorBell sv) ()
441 atomicModifyIORefCAS_ (waitingForWork sv) (const False)
443runqueueLIFO :: MonadIO m => SVar m a -> IORef [Stream m a] -> m ()
444runqueueLIFO sv q = run
451 Nothing -> liftIO $ sendStop sv
452 Just m -> (runStream m) (Just sv) run single yield
454 sendit a = liftIO $ send sv (ChildYield a)
455 single a = sendit a >> run
456 yield a r = sendit a >> (runStream r) (Just sv) run single yield
458 dequeue = liftIO $ atomicModifyIORefCAS q $ \case
460 x : xs -> (xs, Just x)
462-------------------------------------------------------------------------------
464-------------------------------------------------------------------------------
466{-# INLINE enqueueFIFO #-}
467enqueueFIFO :: SVar m a -> LinkedQueue (Stream m a) -> Stream m a -> IO ()
468enqueueFIFO sv q m = do
469 w <- readIORef $ waitingForWork sv
479 when emp $ void $ tryPutMVar (doorBell sv) ()
481runqueueFIFO :: MonadIO m => SVar m a -> LinkedQueue (Stream m a) -> m ()
482runqueueFIFO sv q = run
489 Nothing -> liftIO $ sendStop sv
490 Just m -> (runStream m) (Just sv) run single yield
492 dequeue = liftIO $ tryPopR q
493 sendit a = liftIO $ send sv (ChildYield a)
494 single a = sendit a >> run
495 yield a r = sendit a >> liftIO (enqueueFIFO sv q r) >> run
497-------------------------------------------------------------------------------
499-------------------------------------------------------------------------------
502runOne :: MonadIO m => SVar m a -> Stream m a -> m ()
503runOne sv m = (runStream m) (Just sv) stop single yield
507 stop = liftIO $ sendStop sv
508 sendit a = liftIO $ send sv (ChildYield a)
509 single a = sendit a >> stop
510 yield a r = sendit a >> runOne sv r
512-------------------------------------------------------------------------------
514-------------------------------------------------------------------------------
516-- Lookahead streams can execute multiple tasks concurrently, ahead of time but
517-- always serves them in the same order as they appear in the stream. To
518-- implement lookahead streams efficiently we assign a sequence number to each
519-- task when the task is picked up for execution. When the task finishes the
520-- output is tagged with the same sequence number and we rearrange the outputs
521-- in sequence based on that number.
523-- To explain the mechanism imagine that the current task at the head of the
524-- stream has a "token" to yield to the output. The ownership of the token is
525-- determined by the current sequence number maintained in outputHeap. Sequence
526-- number is assigned when a task is queued. When a thread dequeues a task it
527-- picks up the sequence number as well and when the output is ready it uses
528-- the sequence number to queue the output to the outputQueue.
530-- The thread with current sequence number sends the output directly to the
531-- outputQueue. Other threads push the output to the outputHeap. When the task
532-- being queued on the heap is a stream of many elements we yield the first
533-- element on the heap and an MVar to wakeup for more elements. When such a
534-- task gets the "token" for outputQueue it can directly keep yielding all the
535-- elements to the outputQueue without checking for the "token".
537-- Note that no two outputs in the heap can have the same sequence numbers and
538-- therefore we do not need a stable heap. We have also separated the buffer
539-- for the current task (outputQueue) and the pending tasks (outputHeap) so
540-- that the pending tasks cannot interfere with the current task. Note that for
541-- a single task just the outputQueue is enough and for the case of many
542-- threads just a heap is good enough. However we balance between these two
543-- cases, so that both are efficient.
545-- For bigger streams it may make sense to have separate buffers for each
546-- stream. However, for singleton streams this may become inefficient. However,
547-- if we do not have separate buffers, then the streams that come later in
548-- sequence may hog the buffer, hindering the streams that are ahead. For this
549-- reason we have a single element buffer limitation for the streams being
550-- executed in advance.
552-- This scheme works pretty efficiently with only 50% more overhead compared to
553-- the Async streams where we do not have any kind of sequencing of the
554-- outputs. It is especially devised so that we are most efficient when we have
555-- short tasks and need just a single thread. Also when a thread yields many
556-- items it can hold lockfree access to the outputQueue and do it efficiently.
558-- XXX Maybe we can start the ahead threads at a lower cpu and IO priority so
559-- that they do not hog the resources and hinder the progress of the threads in
562{-# INLINE sendToHeap #-}
563sendToHeap :: SVar m a -> AheadHeapEntry a -> Int -> IO Bool
564sendToHeap sv msg seqNo = do
565 let maxHeapSize = 1500
567 -- Maybe we can use a lock instead of using CAS
568 -- insertions in the heap may be expensive if the heap is big
569 -- Note: len is valid only when added is True
570 (len, added) <- atomicModifyIORefCAS (outputHeap sv) $
573 then ((H.insert (Entry seqNo msg) h, snum), (H.size h, True))
574 else (ref, (0, False))
576 -- this is only for the case when the consumer generated siren can get
577 -- missed due to race.
578 when (len < maxHeapSize) $ void $ tryPutMVar (heapSiren sv) ()
582 when (len + 1 >= maxHeapSize) $ do
583 withDBGMVar sv "sendToHeap: wait for heapSiren"
584 $ takeMVar (heapSiren sv)
585 (h, _) <- readIORef (outputHeap sv)
586 when (H.size h <= maxHeapSize) $ void $ tryPutMVar (siren sv) ()
590 AheadYieldSingle a -> send sv (ChildYield a)
591 AheadYieldMany a mv -> do
592 send sv (ChildYield a)
593 withDBGMVar sv "sendToHeap: wakeup stream" $ putMVar mv ()
597 :: IORef (Heap (Entry Int (AheadHeapEntry a)), Int)
599 -> IO (Maybe (AheadHeapEntry a))
600popFromHeap hp nextSeqNo =
601 atomicModifyIORefCAS hp $ \(h, snum) -> do
603 let n = max nextSeqNo snum
605 Nothing -> ((h, n), Nothing)
606 Just (Entry seqNo ev, hp') ->
608 then ((hp', seqNo), Just ev)
609 else ((h, n), Nothing)
611-- | As soon as we set the nextSeqNo in outputHeap we have transferred the
612-- output "token" to the thread executing the task with that sequence number.
613-- We have to make sure that there is no entry in the heap that has nextSeqNo
614-- before we set nextSeqNo as the expected sequence number in the heap. It is
615-- possible that when we are clearing the heap entry of the current sequence
616-- number the next sequence number gets added and then we have to clear that,
617-- this can keep going on and on.
618drainHeap :: SVar m a -> Int -> IO ()
619drainHeap sv nextSeqNo = do
620 r <- popFromHeap (outputHeap sv) nextSeqNo
625 AheadYieldSingle a -> do
626 send sv (ChildYield a)
627 atomicModifyIORefCAS_ (outputHeap sv) $ \(h, snum) ->
629 drainHeap sv nextSeqNo
630 AheadYieldMany a mv -> do
631 send sv (ChildYield a)
632 withDBGMVar sv "drainHeap: wakeup stream" $ putMVar mv ()
634-- | The thread that has the token has the responsibility to move any pending
635-- entries in the heap to the outputQueue. When the token is assigned to a
636-- sequence number we must not have any entries for that sequence number in the
638releaseToken :: SVar m a -> Int -> IO ()
639releaseToken sv nextSeqNo = do
640 drainHeap sv nextSeqNo
641 void $ tryPutMVar (heapSiren sv) ()
643-- | When we have the output "token" we are allowed to append to the
644-- outputQueue otherwise we have to insert the output in the outputHeap.
645-- Returns whether we still have the token.
646handleOutput :: SVar m a
653handleOutput sv haveToken prevSeqNo seqNo a ev = do
656 let nextSeqNo = prevSeqNo + 1
657 if seqNo == nextSeqNo
658 then send sv (ChildYield a) >> return True
660 releaseToken sv nextSeqNo
663 where toHeap = sendToHeap sv ev seqNo
665-- Left associated ahead expressions are expensive. We start a new SVar for
666-- each left associative expression. The queue is used only for right
667-- associated expression, we queue the right expression and execute the left.
668-- Thererefore the queue never has more than on item in it.
669{-# INLINE enqueueAhead #-}
670enqueueAhead :: SVar m a -> IORef ([Stream m a], Int) -> Stream m a -> IO ()
671enqueueAhead sv q m = do
672 w <- readIORef $ waitingForWork sv
673 atomicModifyIORefCAS_ q $ \ case
674 ([], n) -> ([m], n + 1) -- increment sequence
675 _ -> error "not empty"
677 void $ tryPutMVar (doorBell sv) ()
678 atomicModifyIORefCAS_ (waitingForWork sv) (const False)
680runqueueAhead :: MonadIO m => SVar m a -> IORef ([Stream m a], Int) -> m ()
681runqueueAhead sv q = run False (-1)
685 run haveToken prevSeqNo = do
688 Nothing -> liftIO $ do
689 when haveToken $ releaseToken sv (prevSeqNo + 1)
691 Just (m, seqNo) -> do
692 let tStatus = haveToken || seqNo == 0
694 let nextSeqNo = prevSeqNo + 1
695 if tStatus && seqNo /= nextSeqNo
697 liftIO $ releaseToken sv nextSeqNo
699 else run tStatus seqNo
701 tokenStatus <- liftIO $
702 handleOutput sv tStatus prevSeqNo seqNo a
704 run tokenStatus seqNo
706 -- If we are yielding more than one item and we do not have the
707 -- token then we insert the current item in the pending
708 -- outputHeap and wait for our turn.
710 stopNext = run True seqNo
712 liftIO $ send sv (ChildYield a)
715 liftIO $ send sv (ChildYield a)
716 (runStream r) (Just sv) stopNext singleNext yieldNext
719 -- XXX we can allocate the MVar when actually inserting
720 -- in the heap and then return it.
724 handleOutput sv tStatus prevSeqNo seqNo a
725 (AheadYieldMany a mv)
726 when (not gotToken) $ void $
727 withDBGMVar sv "runqueueAhead: stream waiting"
729 (runStream r) (Just sv) stopNext singleNext yieldNext
731 (runStream m) (Just sv) stop single yield
733 dequeue = liftIO $ do
734 atomicModifyIORefCAS q $ \case
735 ([], n) -> (([], n), Nothing)
736 (x : [], n) -> (([], n), Just (x, n))
737 _ -> error "more than one item on queue"
739-- Thread tracking is needed for two reasons:
741-- 1) Killing threads on exceptions. Threads may not be left to go away by
742-- themselves because they may run for significant times before going away or
743-- worse they may be stuck in IO and never go away.
745-- 2) To know when all threads are done and the stream has ended.
747{-# NOINLINE addThread #-}
748addThread :: MonadIO m => SVar m a -> ThreadId -> m ()
750 liftIO $ modifyIORef (runningThreads sv) (S.insert tid)
753{-# INLINE delThread #-}
754delThread :: MonadIO m => SVar m a -> ThreadId -> m ()
756 liftIO $ modifyIORef (runningThreads sv) $ (\s -> S.delete tid s)
759-- If present then delete else add. This takes care of out of order add and
760-- delete i.e. a delete arriving before we even added a thread.
761-- This occurs when the forked thread is done even before the 'addThread' right
762-- after the fork is executed.
763{-# INLINE modifyThread #-}
764modifyThread :: MonadIO m => SVar m a -> ThreadId -> m ()
765modifyThread sv tid = do
766 changed <- liftIO $ atomicModifyIORefCAS (runningThreads sv) $ \old ->
767 if (S.member tid old)
768 then let new = (S.delete tid old) in (new, new)
769 else let new = (S.insert tid old) in (new, old)
771 then liftIO $ void $ tryPutMVar (doorBell sv) ()
774-- | This is safe even if we are adding more threads concurrently because if
775-- a child thread is adding another thread then anyway 'runningThreads' will
777{-# INLINE allThreadsDone #-}
778allThreadsDone :: MonadIO m => SVar m a -> m Bool
779allThreadsDone sv = liftIO $ S.null <$> readIORef (runningThreads sv)
781{-# NOINLINE handleChildException #-}
782handleChildException :: MonadIO m => SVar m a -> SomeException -> m ()
783handleChildException sv e = liftIO $ do
785 send sv (ChildStop tid (Just e))
787{-# NOINLINE pushWorker #-}
788pushWorker :: MonadAsync m => SVar m a -> m ()
790 liftIO $ atomicModifyIORefCAS_ (activeWorkers sv) $ \n -> n + 1
791 doFork (runqueue sv) (handleChildException sv) >>= addThread sv
793-- | In contrast to pushWorker which always happens only from the consumer
794-- thread, a pushWorkerPar can happen concurrently from multiple threads on the
795-- producer side. So we need to use a thread safe modification of
796-- runningThreads. Alternatively, we can use a CreateThread event to avoid
797-- using a CAS based modification.
798{-# NOINLINE pushWorkerPar #-}
799pushWorkerPar :: MonadAsync m => SVar m a -> Stream m a -> m ()
801 doFork (runOne sv m) (handleChildException sv) >>= modifyThread sv
803-- XXX When the queue is LIFO we can put a limit on the number of dispatches.
804-- Also, if a worker blocks on the output queue we can decide if we want to
805-- block or make it go away entirely, depending on the number of workers and
806-- the type of the queue.
807{-# INLINE sendWorkerWait #-}
808sendWorkerWait :: MonadAsync m => SVar m a -> m ()
809sendWorkerWait sv = do
810 -- When there is no output seen we dispatch more workers to help out if
811 -- there is work pending in the work queue. But we wait a little while
812 -- and check the output again so that we are not too aggressive.
813 -- If there is no output pending to process and there is no worker to be
814 -- sent then we block, so that we do not keep looping fruitlessly.
816 -- Note that we are guaranteed to have at least one outstanding worker when
817 -- we enter this function. So if we sleep we are guaranteed to be woken up.
818 -- Note that the worker count is only decremented during event processing
819 -- in fromStreamVar and therefore it is safe to read and use it without a
822 -- XXX we should wait in such a way so that we can be immediately
823 -- interrupted by a doorBell. We can sleep on the doorbell and use a
824 -- monitor thread to send us a doorbell.
825 liftIO $ threadDelay 200
826 (_, n) <- liftIO $ readIORef (outputQueue sv)
828 -- The queue may be empty temporarily if the worker has dequeued the
829 -- work item but has not enqueued the remaining part yet. For the same
830 -- reason, a worker may come back if it tries to dequeue and finds the
831 -- queue empty, even though the whole work has not finished yet.
832 done <- queueEmpty sv
835 cnt <- liftIO $ readIORef $ activeWorkers sv
836 -- Note that we may deadlock if the previous workers (tasks in the
837 -- stream) wait/depend on the future workers (tasks in the stream)
838 -- executing. In that case we should either configure the maxWorker
839 -- count to higher or use parallel style instead of ahead or async
845 else liftIO $ withDBGMVar sv "sendWorkerWait: wait for workers"
846 $ takeMVar (doorBell sv)
848 -- We found that the queue was empty, but it may be empty
849 -- temporarily, when we checked it. If that's the case we might
850 -- sleep indefinitely unless the active workers produce some
851 -- output. We may deadlock specially if the otuput from the other
852 -- workers depends on the future workers that we may never send.
853 -- So in case the queue was temporarily empty set a flag to inform
854 -- the enqueue to send us a doorbell.
856 liftIO $ atomicModifyIORefCAS_ (waitingForWork sv) $ const True
858 -- check again, this time we have set the waitingForWork flag so we
859 -- are guaranteed to get a doorbell in case the status changed
860 -- before we could sleep.
862 -- Note that this is just a best effort mechanism to avoid a
863 -- deadlock. Deadlocks may still happen if for some weird reason
864 -- the consuming computation shares an MVar or some other resource
865 -- with the producing computation and gets blocked on that resource
866 -- and therefore cannot do any pushworker to add more threads to
867 -- the producer. In such cases the programmer should use a parallel
868 -- style so that all the producers are scheduled immediately and
869 -- unconditionally. We can also use a separate monitor thread to
870 -- push workers instead of pushing them from the consumer, but then
871 -- we are no longer using pull based concurrency rate adaptation.
873 -- XXX update this in the tutorial.
876 liftIO $ withDBGMVar sv "sendWorkerWait: nothing to do"
877 $ takeMVar (doorBell sv)
879-- | Pull a stream from an SVar.
880{-# NOINLINE fromStreamVar #-}
881fromStreamVar :: MonadAsync m => SVar m a -> Stream m a
882fromStreamVar sv = Stream $ \_ stp sng yld -> do
883 if svarStyle sv == ParallelVar
884 then liftIO $ withDBGMVar sv "fromStreamVar: doorbell"
885 $ takeMVar (doorBell sv)
887 res <- liftIO $ tryTakeMVar (doorBell sv)
888 when (isNothing res) $ sendWorkerWait sv
890 (list, _) <- liftIO $ atomicModifyIORefCAS (outputQueue sv) $ \x -> (([],0), x)
891 void $ liftIO $ tryPutMVar (siren sv) ()
892 -- Reversing the output is important to guarantee that we process the
893 -- outputs in the same order as they were generated by the constituent
895 (runStream $ processEvents (reverse list)) Nothing stp sng yld
899 handleException e tid = do
900 liftIO $ atomicModifyIORefCAS_ (activeWorkers sv) $ \n -> n - 1
902 -- XXX implement kill async exception handling
903 -- liftIO $ readIORef (runningThreads sv) >>= mapM_ killThread
906 {-# INLINE processEvents #-}
907 processEvents [] = Stream $ \_ stp sng yld -> do
908 done <- allThreadsDone sv
910 then (runStream (fromStreamVar sv)) Nothing stp sng yld
913 processEvents (ev : es) = Stream $ \_ stp sng yld -> do
914 let continue = (runStream (processEvents es)) Nothing stp sng yld
915 yield a = yld a (processEvents es)
918 ChildYield a -> yield a
922 let active = activeWorkers sv
923 liftIO $ atomicModifyIORefCAS_ active $ \n -> n - 1
924 modifyThread sv tid >> continue
925 Just ex -> handleException ex tid
927getFifoSVar :: MonadIO m => SVarStyle -> IO (SVar m a)
928getFifoSVar ctype = do
929 outQ <- newIORef ([], 0)
930 outQMv <- newEmptyMVar
931 hooter <- newEmptyMVar
933 wfw <- newIORef False
934 running <- newIORef S.empty
937 SVar { outputQueue = outQ
940 , outputHeap = undefined
941 , heapSiren = undefined
942 , runningThreads = running
943 , workQueue = undefined
944 , runqueue = runqueueFIFO sv q
945 , enqueue = enqueueFIFO sv q
946 , queueEmpty = liftIO $ nullQ q
947 , waitingForWork = wfw
949 , activeWorkers = active
953getLifoSVar :: MonadIO m => SVarStyle -> IO (SVar m a)
954getLifoSVar ctype = do
955 outQ <- newIORef ([], 0)
956 outQMv <- newEmptyMVar
957 hooter <- newEmptyMVar
959 wfw <- newIORef False
960 running <- newIORef S.empty
962 let checkEmpty = null <$> liftIO (readIORef q)
964 SVar { outputQueue = outQ
967 , outputHeap = undefined
968 , heapSiren = undefined
969 , runningThreads = running
970 , workQueue = undefined
971 , runqueue = runqueueLIFO sv q
972 , enqueue = enqueueLIFO sv q
973 , queueEmpty = checkEmpty
974 , waitingForWork = wfw
976 , activeWorkers = active
980getParSVar :: SVarStyle -> IO (SVar m a)
982 outQ <- newIORef ([], 0)
983 outQMv <- newEmptyMVar
984 hooter <- newEmptyMVar
986 wfw <- newIORef False
987 running <- newIORef S.empty
989 SVar { outputQueue = outQ
992 , outputHeap = undefined
993 , heapSiren = undefined
994 , runningThreads = running
995 , workQueue = undefined
996 , runqueue = undefined
997 , enqueue = undefined
998 , queueEmpty = undefined
999 , waitingForWork = wfw
1001 , activeWorkers = active
1005getAheadSVar :: MonadIO m => SVarStyle -> IO (SVar m a)
1006getAheadSVar style = do
1007 outQ <- newIORef ([], 0)
1008 outH <- newIORef (H.empty, 0)
1009 outQMv <- newEmptyMVar
1010 hooter <- newEmptyMVar
1011 hsiren <- newEmptyMVar
1012 active <- newIORef 0
1013 wfw <- newIORef False
1014 running <- newIORef S.empty
1015 q <- newIORef ([], -1)
1016 let checkEmpty = liftIO $ do
1017 (xs, _) <- readIORef q
1020 SVar { outputQueue = outQ
1024 , heapSiren = hsiren
1025 , runningThreads = running
1027 , runqueue = runqueueAhead sv q
1028 , enqueue = undefined
1029 , queueEmpty = checkEmpty
1030 , waitingForWork = wfw
1032 , activeWorkers = active
1036-- | Create a new empty SVar.
1037newEmptySVar :: MonadAsync m => SVarStyle -> m (SVar m a)
1038newEmptySVar style = do
1041 WAsyncVar -> getFifoSVar style
1042 AsyncVar -> getLifoSVar style
1043 ParallelVar -> getParSVar style
1044 AheadVar -> getAheadSVar style
1046-- | Create a new SVar and enqueue one stream computation on it.
1047newStreamVar1 :: MonadAsync m => SVarStyle -> Stream m a -> m (SVar m a)
1048newStreamVar1 style m = do
1049 sv <- newEmptySVar style
1050 -- Note: We must have all the work on the queue before sending the
1051 -- pushworker, otherwise the pushworker may exit before we even get a
1053 liftIO $ (enqueue sv) m
1057-- | Create a new SVar and enqueue one stream computation on it.
1058newStreamVarAhead :: MonadAsync m => Stream m a -> m (SVar m a)
1059newStreamVarAhead m = do
1060 sv <- newEmptySVar AheadVar
1061 -- Note: We must have all the work on the queue before sending the
1062 -- pushworker, otherwise the pushworker may exit before we even get a
1064 liftIO $ enqueueAhead sv (workQueue sv) m
1068-- | Write a stream to an 'SVar' in a non-blocking manner. The stream can then
1069-- be read back from the SVar using 'fromSVar'.
1070toStreamVar :: MonadAsync m => SVar m a -> Stream m a -> m ()
1071toStreamVar sv m = do
1072 liftIO $ (enqueue sv) m
1073 done <- allThreadsDone sv
1074 -- XXX This is safe only when called from the consumer thread or when no
1075 -- consumer is present. There may be a race if we are not running in the
1077 when done $ pushWorker sv
1079------------------------------------------------------------------------------
1080-- Running streams concurrently
1081------------------------------------------------------------------------------
1083-- Concurrency rate control.
1085-- Our objective is to create more threads on demand if the consumer is running
1086-- faster than us. As soon as we encounter a concurrent composition we create a
1087-- push pull pair of threads. We use an SVar for communication between the
1088-- consumer, pulling from the SVar and the producer who is pushing to the SVar.
1089-- The producer creates more threads if the SVar drains and becomes empty, that
1090-- is the consumer is running faster.
1092-- XXX Note 1: This mechanism can be problematic if the initial production
1093-- latency is high, we may end up creating too many threads. So we need some
1094-- way to monitor and use the latency as well. Having a limit on the dispatches
1095-- (programmer controlled) may also help.
1097-- TBD Note 2: We may want to run computations at the lower level of the
1098-- composition tree serially even when they are composed using a parallel
1099-- combinator. We can use 'serial' in place of 'async' and 'wSerial' in
1100-- place of 'wAsync'. If we find that an SVar immediately above a computation
1101-- gets drained empty we can switch to parallelizing the computation. For that
1102-- we can use a state flag to fork the rest of the computation at any point of
1103-- time inside the Monad bind operation if the consumer is running at a faster
1106-- TBD Note 3: the binary operation ('parallel') composition allows us to
1107-- dispatch a chunkSize of only 1. If we have to dispatch in arbitrary
1108-- chunksizes we will need to compose the parallel actions using a data
1109-- constructor (A Free container) instead so that we can divide it in chunks of
1110-- arbitrary size before dispatching. If the stream is composed of
1111-- hierarchically composed grains of different sizes then we can always switch
1112-- to a desired granularity depending on the consumer speed.
1114-- TBD Note 4: for pure work (when we are not in the IO monad) we can divide it
1115-- into just the number of CPUs.
1117-- | Join two computations on the currently running 'SVar' queue for concurrent
1118-- execution. When we are using parallel composition, an SVar is passed around
1119-- as a state variable. We try to schedule a new parallel computation on the
1120-- SVar passed to us. The first time, when no SVar exists, a new SVar is
1121-- created. Subsequently, 'joinStreamVarAsync' may get called when a computation
1122-- already scheduled on the SVar is further evaluated. For example, when (a
1123-- `parallel` b) is evaluated it calls a 'joinStreamVarAsync' to put 'a' and 'b' on
1124-- the current scheduler queue.
1126-- The 'SVarStyle' required by the current composition context is passed as one
1127-- of the parameters. If the scheduling and composition style of the new
1128-- computation being scheduled is different than the style of the current SVar,
1129-- then we create a new SVar and schedule it on that. The newly created SVar
1130-- joins as one of the computations on the current SVar queue.
1132-- Cases when we need to switch to a new SVar:
1134-- * (x `parallel` y) `parallel` (t `parallel` u) -- all of them get scheduled on the same SVar
1135-- * (x `parallel` y) `parallel` (t `async` u) -- @t@ and @u@ get scheduled on a new child SVar
1136-- because of the scheduling policy change.
1137-- * if we 'adapt' a stream of type 'async' to a stream of type
1138-- 'Parallel', we create a new SVar at the transitioning bind.
1139-- * When the stream is switching from disjunctive composition to conjunctive
1140-- composition and vice-versa we create a new SVar to isolate the scheduling
1143{-# INLINE joinStreamVarAsync #-}
1144joinStreamVarAsync :: MonadAsync m
1145 => SVarStyle -> Stream m a -> Stream m a -> Stream m a
1146joinStreamVarAsync style m1 m2 = Stream $ \svr stp sng yld ->
1148 Just sv | svarStyle sv == style ->
1149 liftIO ((enqueue sv) m2) >> (runStream m1) svr stp sng yld
1151 sv <- newStreamVar1 style (concurrently m1 m2)
1152 (runStream (fromStreamVar sv)) Nothing stp sng yld
1154 concurrently ma mb = Stream $ \svr stp sng yld -> do
1155 liftIO $ enqueue (fromJust svr) mb
1156 (runStream ma) svr stp sng yld
1158{-# INLINE joinStreamVarPar #-}
1159joinStreamVarPar :: MonadAsync m
1160 => SVarStyle -> Stream m a -> Stream m a -> Stream m a
1161joinStreamVarPar style m1 m2 = Stream $ \svr stp sng yld ->
1163 Just sv | svarStyle sv == style -> do
1164 pushWorkerPar sv m1 >> (runStream m2) svr stp sng yld
1166 sv <- newEmptySVar style
1169 (runStream (fromStreamVar sv)) Nothing stp sng yld
1172ahead :: MonadAsync m => Stream m a -> Stream m a -> Stream m a
1173ahead m1 m2 = Stream $ \svr stp sng yld -> do
1175 Just sv | svarStyle sv == AheadVar -> do
1176 liftIO $ enqueueAhead sv (workQueue sv) m2
1177 -- Always run the left side on a new SVar to avoid complexity in
1178 -- sequencing results. This means the left side cannot further
1179 -- split into more ahead computations on the same SVar.
1180 (runStream m1) Nothing stp sng yld
1182 sv <- newStreamVarAhead (concurrently m1 m2)
1183 (runStream (fromStreamVar sv)) Nothing stp sng yld
1185 concurrently ma mb = Stream $ \svr stp sng yld -> do
1186 liftIO $ enqueueAhead (fromJust svr) (workQueue (fromJust svr)) mb
1187 (runStream ma) Nothing stp sng yld
1189-- | XXX we can implement it more efficienty by directly implementing instead
1190-- of combining streams using ahead.
1191{-# INLINE consMAhead #-}
1192consMAhead :: MonadAsync m => m a -> Stream m a -> Stream m a
1193consMAhead m r = once m `ahead` r
1195------------------------------------------------------------------------------
1196-- Semigroup and Monoid style compositions for parallel actions
1197------------------------------------------------------------------------------
1200async :: MonadAsync m => Stream m a -> Stream m a -> Stream m a
1201async = joinStreamVarAsync AsyncVar
1203-- | XXX we can implement it more efficienty by directly implementing instead
1204-- of combining streams using async.
1205{-# INLINE consMAsync #-}
1206consMAsync :: MonadAsync m => m a -> Stream m a -> Stream m a
1207consMAsync m r = once m `async` r
1209{-# INLINE wAsync #-}
1210wAsync :: MonadAsync m => Stream m a -> Stream m a -> Stream m a
1211wAsync = joinStreamVarAsync WAsyncVar
1213-- | XXX we can implement it more efficienty by directly implementing instead
1214-- of combining streams using wAsync.
1215{-# INLINE consMWAsync #-}
1216consMWAsync :: MonadAsync m => m a -> Stream m a -> Stream m a
1217consMWAsync m r = once m `wAsync` r
1219{-# INLINE parallel #-}
1220parallel :: MonadAsync m => Stream m a -> Stream m a -> Stream m a
1221parallel = joinStreamVarPar ParallelVar
1223-- | XXX we can implement it more efficienty by directly implementing instead
1224-- of combining streams using parallel.
1225{-# INLINE consMParallel #-}
1226consMParallel :: MonadAsync m => m a -> Stream m a -> Stream m a
1227consMParallel m r = once m `parallel` r
1229-------------------------------------------------------------------------------
1230-- Functor instace is the same for all types
1231-------------------------------------------------------------------------------
1233instance Monad m => Functor (Stream m) where
1234 fmap f m = Stream $ \_ stp sng yld ->
1235 let single = sng . f
1236 yield a r = yld (f a) (fmap f r)
1237 in (runStream m) Nothing stp single yield
1239------------------------------------------------------------------------------
1240-- Alternative & MonadPlus
1241------------------------------------------------------------------------------
1243_alt :: Stream m a -> Stream m a -> Stream m a
1244_alt m1 m2 = Stream $ \_ stp sng yld ->
1245 let stop = runStream m2 Nothing stp sng yld
1246 in runStream m1 Nothing stop sng yld
1248------------------------------------------------------------------------------
1250------------------------------------------------------------------------------
1252zipWith :: (a -> b -> c) -> Stream m a -> Stream m b -> Stream m c
1253zipWith f m1 m2 = go m1 m2
1255 go mx my = Stream $ \_ stp sng yld -> do
1257 let single2 b = sng (f a b)
1258 yield2 b rb = yld (f a b) (go ra rb)
1259 in (runStream my) Nothing stp single2 yield2
1260 let single1 a = merge a nil
1261 yield1 a ra = merge a ra
1262 (runStream mx) Nothing stp single1 yield1
1264mkAsync :: MonadAsync m => Stream m a -> m (Stream m a)
1265mkAsync m = newStreamVar1 AsyncVar m
1266 >>= return . fromStreamVar
1268zipAsyncWith :: MonadAsync m
1269 => (a -> b -> c) -> Stream m a -> Stream m b -> Stream m c
1270zipAsyncWith f m1 m2 = Stream $ \_ stp sng yld -> do
1273 (runStream (zipWith f ma mb)) Nothing stp sng yld
1275-------------------------------------------------------------------------------
1277-------------------------------------------------------------------------------
1279instance MonadTrans Stream where
1282withLocal :: MonadReader r m => (r -> r) -> Stream m a -> Stream m a
1284 Stream $ \svr stp sng yld ->
1285 let single = local f . sng
1286 yield a r = local f $ yld a (withLocal f r)
1287 in (runStream m) svr (local f stp) single yield
1289-- XXX handle and test cross thread state transfer
1292 => Stream m a -> (e -> Stream m a) -> Stream m a
1294 Stream $ \svr stp sng yld ->
1295 let run x = runStream x svr stp sng yield
1296 handle r = r `catchError` \e -> run $ h e
1297 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