2{-# LANGUAGE ConstraintKinds #-}
3{-# LANGUAGE FlexibleContexts #-}
4{-# LANGUAGE FlexibleInstances #-}
5{-# LANGUAGE LambdaCase #-}
6{-# LANGUAGE MagicHash #-}
7{-# LANGUAGE MultiParamTypeClasses #-}
8{-# LANGUAGE RankNTypes #-}
9{-# LANGUAGE UnboxedTuples #-}
10{-# LANGUAGE UndecidableInstances #-} -- XXX
13-- Module : Streamly.Core
14-- Copyright : (c) 2017 Harendra Kumar
17-- Maintainer : harendra.kumar@gmail.com
18-- Stability : experimental
29 -- * Construction (pure)
36 -- * Construction (monadic)
43 -- * Semigroup Style Composition
59 -- * Concurrent Stream Vars (SVars)
68import Control.Concurrent (ThreadId, myThreadId, threadDelay)
69import Control.Concurrent.MVar (MVar, newEmptyMVar, tryTakeMVar,
70 tryPutMVar, takeMVar, putMVar)
71import Control.Exception (SomeException (..), catch, mask)
72import Control.Monad (when)
73import Control.Monad.Catch (MonadThrow, throwM)
74import Control.Monad.IO.Class (MonadIO(..))
75import Control.Monad.Trans.Class (MonadTrans (lift))
76import Control.Monad.Trans.Control (MonadBaseControl, control)
77import Data.Atomics (casIORef, readForCAS, peekTicket
78 ,atomicModifyIORefCAS_)
79import Data.Concurrent.Queue.MichaelScott (LinkedQueue, newQ, pushL,
81import Data.Functor (void)
82import Data.Heap (Heap, Entry(..))
83import qualified Data.Heap as H
84import Data.IORef (IORef, modifyIORef, newIORef,
85 readIORef, atomicModifyIORef)
86import Data.Maybe (isNothing, fromJust)
87import Data.Semigroup (Semigroup(..))
89import qualified Data.Set as S
90import Prelude hiding (repeat, zipWith)
93import GHC.Conc (ThreadId(..))
96-- MVar diagnostics has some overhead - around 5% on asyncly null benchmark, we
97-- can keep it on in production to debug problems quickly if and when they
98-- happen, but it may result in unexpected output when threads are left hanging
99-- until they are GCed because the consumer went away.
102import Control.Exception (catches, throwIO, Handler(..),
103 BlockedIndefinitelyOnMVar(..),
104 BlockedIndefinitelyOnSTM(..))
105import System.IO (hPutStrLn, stderr)
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
197 (oqList, oqLen) <- readIORef $ outputQueue sv
198 db <- tryTakeMVar $ doorBell sv
199 srn <- tryTakeMVar $ siren sv
201 if svarStyle sv == AheadVar
203 (oheap, oheapSize) <- readIORef $ outputHeap sv
204 hsrn <- tryTakeMVar $ heapSiren sv
205 (wq, wqSize) <- readIORef $ workQueue sv
207 [ "heap length computed = " ++ show (H.size oheap)
208 , "heap length maintained = " ++ show oheapSize
209 , "heap siren = " ++ show hsrn
210 , "work queue length computed = " ++ show (length wq)
211 , "work queue length maintained = " ++ show wqSize
215 waiting <- readIORef $ waitingForWork sv
216 rthread <- readIORef $ runningThreads sv
217 workers <- readIORef $ activeWorkers sv
218 -- XXX queueEmpty should be made IO return type
221 [ "tid = " ++ show tid
222 , "style = " ++ show (svarStyle sv)
223 , "outputQueue length computed = " ++ show (length oqList)
224 , "outputQueue length maintained = " ++ show oqLen
225 , "output doorBell = " ++ show db
226 , "worker siren = " ++ show srn
228 ++ aheadDump ++ unlines
229 [ "waitingForWork = " ++ show waiting
230 , "running threads = " ++ show rthread
231 , "running thread count = " ++ show workers
234{-# NOINLINE mvarExcHandler #-}
235mvarExcHandler :: SVar m a -> String -> BlockedIndefinitelyOnMVar -> IO ()
236mvarExcHandler sv label e@BlockedIndefinitelyOnMVar = do
237 svInfo <- dumpSVar sv
238 hPutStrLn stderr $ label ++ " " ++ "BlockedIndefinitelyOnMVar\n" ++ svInfo
241{-# NOINLINE stmExcHandler #-}
242stmExcHandler :: SVar m a -> String -> BlockedIndefinitelyOnSTM -> IO ()
243stmExcHandler sv label e@BlockedIndefinitelyOnSTM = do
244 svInfo <- dumpSVar sv
245 hPutStrLn stderr $ label ++ " " ++ "BlockedIndefinitelyOnSTM\n" ++ svInfo
248withDBGMVar :: SVar m a -> String -> IO () -> IO ()
249withDBGMVar sv label action =
250 action `catches` [ Handler (mvarExcHandler sv label)
251 , Handler (stmExcHandler sv label)
254withDBGMVar :: SVar m a -> String -> IO () -> IO ()
255withDBGMVar _ _ action = action
258-- Slightly faster version of CAS. Gained some improvement by avoiding the use
259-- of "evaluate" because we know we do not have exceptions in fn.
260{-# INLINE atomicModifyIORefCAS #-}
261atomicModifyIORefCAS :: IORef a -> (a -> (a,b)) -> IO b
262atomicModifyIORefCAS ref fn = do
263 tkt <- readForCAS ref
269 loop _ 0 = atomicModifyIORef ref fn
271 let (new, result) = fn $ peekTicket old
272 (success, tkt) <- casIORef ref old new
275 else loop tkt (tries - 1)
277------------------------------------------------------------------------------
279------------------------------------------------------------------------------
281-- | The type 'Stream m a' represents a monadic stream of values of type 'a'
282-- constructed using actions in monad 'm'. It uses stop, singleton and yield
283-- continuations equivalent to the following direct style type:
285-- data Stream m a = Stop | Singleton a | Yield a (Stream m a)
287-- To facilitate parallel composition we maintain a local state in an SVar that
288-- is shared across and is used for synchronization of the streams being
291-- The singleton case can be expressed in terms of stop and yield but we have
292-- it as a separate case to optimize composition operations for streams with
293-- single element. We build singleton streams in the implementation of 'pure'
294-- for Applicative and Monad, and in 'lift' for MonadTrans.
298 runStream :: forall r.
299 Maybe (SVar m a) -- local state
301 -> (a -> m r) -- singleton
302 -> (a -> Stream m a -> m r) -- yield
307nil = Stream $ \_ stp _ _ -> stp
309-- | faster than consM because there is no bind.
310cons :: a -> Stream m a -> Stream m a
311cons a r = Stream $ \_ _ _ yld -> yld a r
313-- | Same as @once . return@ but may be faster because there is no bind
314singleton :: a -> Stream m a
315singleton a = Stream $ \_ _ single _ -> single a
318once :: Monad m => m a -> Stream m a
319once m = Stream $ \_ _ single _ -> m >>= single
322consM :: Monad m => m a -> Stream m a -> Stream m a
323consM m r = Stream $ \_ _ _ yld -> m >>= \a -> yld a r
325repeat :: a -> Stream m a
326repeat a = let x = cons a x in x
328------------------------------------------------------------------------------
330------------------------------------------------------------------------------
332-- | Concatenates two streams sequentially i.e. the first stream is
333-- exhausted completely before yielding any element from the second stream.
335serial :: Stream m a -> Stream m a -> Stream m a
338 go (Stream m) = Stream $ \_ stp sng yld ->
339 let stop = (runStream m2) Nothing stp sng yld
341 yield a r = yld a (go r)
342 in m Nothing stop single yield
344instance Semigroup (Stream m a) where
347------------------------------------------------------------------------------
349------------------------------------------------------------------------------
351instance Monoid (Stream m a) where
355------------------------------------------------------------------------------
357------------------------------------------------------------------------------
359{-# INLINE wSerial #-}
360wSerial :: Stream m a -> Stream m a -> Stream m a
361wSerial m1 m2 = Stream $ \_ stp sng yld -> do
362 let stop = (runStream m2) Nothing stp sng yld
364 yield a r = yld a (wSerial m2 r)
365 (runStream m1) Nothing stop single yield
367------------------------------------------------------------------------------
368-- Spawning threads and collecting result in streamed fashion
369------------------------------------------------------------------------------
371-- | A monad that can perform concurrent or parallel IO operations. Streams
372-- that can be composed concurrently require the underlying monad to be
376type MonadAsync m = (MonadIO m, MonadBaseControl IO m, MonadThrow m)
378-- Stolen from the async package. The perf improvement is modest, 2% on a
379-- thread heavy benchmark (parallel composition using noop computations).
380-- A version of forkIO that does not include the outer exception
381-- handler: saves a bit of time when we will be installing our own
383{-# INLINE rawForkIO #-}
384rawForkIO :: IO () -> IO ThreadId
385rawForkIO action = IO $ \ s ->
386 case (fork# action s) of (# s1, tid #) -> (# s1, ThreadId tid #)
389doFork :: MonadBaseControl IO m
391 -> (SomeException -> IO ())
393doFork action exHandler =
394 control $ \runInIO ->
395 mask $ \restore -> do
396 tid <- rawForkIO $ catch (restore $ void $ runInIO action)
400-- XXX exception safety of all atomic/MVar operations
402-- TBD Each worker can have their own queue and the consumer can empty one
403-- queue at a time, that way contention can be reduced.
405-- | This function is used by the producer threads to queue output for the
406-- consumer thread to consume.
408send :: SVar m a -> ChildEvent a -> IO ()
410 len <- atomicModifyIORefCAS (outputQueue sv) $ \(es, n) ->
411 ((msg : es, n + 1), n)
412 if (len <= 0) then do
413 -- XXX need a memory barrier? The wake up must happen only after the
414 -- store has finished otherwise we can have lost wakeup problems.
416 -- Since multiple workers can try this at the same time, it is possible
417 -- that we may put a spurious MVar after the consumer has already seen
418 -- the output. But that's harmless, at worst it may cause the consumer
419 -- to read the queue again and find it empty.
420 -- The important point is that the consumer is guaranteed to receive a
421 -- doorbell if something was added to the queue after it empties it.
422 void $ tryPutMVar (doorBell sv) ()
423 -- those who were waiting because of buffer limit should now go ahead
424 void $ tryPutMVar (siren sv) ()
425 -- The only thread that reduces the length of the queue is the consumer
426 -- thread. It is possible that the queue may have been emptied in the
427 -- window when we acquired "len" and now when we are using it. In that case
428 -- we will be woken up by the consumer blowing the siren.
429 -- Note: current length after adding our output is len + 1
430 else when (len + 1 >= 1500) $ do
431 withDBGMVar sv "send: wait for siren" $ takeMVar (siren sv)
432 (_, n) <- readIORef (outputQueue sv)
433 when (n <= 1500) $ void $ tryPutMVar (siren sv) ()
435{-# INLINE sendStop #-}
436sendStop :: SVar m a -> IO ()
437sendStop sv = myThreadId >>= \tid -> send sv (ChildStop tid Nothing)
439-------------------------------------------------------------------------------
441-------------------------------------------------------------------------------
443-- Note: For purely right associated expressions this queue should have at most
444-- one element. It grows to more than one when we have left associcated
445-- expressions. Large left associated compositions can grow this to a
447{-# INLINE enqueueLIFO #-}
448enqueueLIFO :: SVar m a -> IORef [Stream m a] -> Stream m a -> IO ()
449enqueueLIFO sv q m = do
450 w <- readIORef $ waitingForWork sv
453 else atomicModifyIORefCAS_ q $ \ms -> m : ms
458 v <- atomicModifyIORefCAS q $ \ ms ->
462 when v $ void $ tryPutMVar (doorBell sv) ()
463 atomicModifyIORefCAS_ (waitingForWork sv) (const False)
465runqueueLIFO :: MonadIO m => SVar m a -> IORef [Stream m a] -> m ()
466runqueueLIFO sv q = run
473 Nothing -> liftIO $ sendStop sv
474 Just m -> (runStream m) (Just sv) run single yield
476 sendit a = liftIO $ send sv (ChildYield a)
477 single a = sendit a >> run
478 yield a r = sendit a >> (runStream r) (Just sv) run single yield
480 dequeue = liftIO $ atomicModifyIORefCAS q $ \case
482 x : xs -> (xs, Just x)
484-------------------------------------------------------------------------------
486-------------------------------------------------------------------------------
488{-# INLINE enqueueFIFO #-}
489enqueueFIFO :: SVar m a -> LinkedQueue (Stream m a) -> Stream m a -> IO ()
490enqueueFIFO sv q m = do
491 w <- readIORef $ waitingForWork sv
501 when emp $ void $ tryPutMVar (doorBell sv) ()
503runqueueFIFO :: MonadIO m => SVar m a -> LinkedQueue (Stream m a) -> m ()
504runqueueFIFO sv q = run
511 Nothing -> liftIO $ sendStop sv
512 Just m -> (runStream m) (Just sv) run single yield
514 dequeue = liftIO $ tryPopR q
515 sendit a = liftIO $ send sv (ChildYield a)
516 single a = sendit a >> run
517 yield a r = sendit a >> liftIO (enqueueFIFO sv q r) >> run
519-------------------------------------------------------------------------------
521-------------------------------------------------------------------------------
524runOne :: MonadIO m => SVar m a -> Stream m a -> m ()
525runOne sv m = (runStream m) (Just sv) stop single yield
529 stop = liftIO $ sendStop sv
530 sendit a = liftIO $ send sv (ChildYield a)
531 single a = sendit a >> stop
532 yield a r = sendit a >> runOne sv r
534-------------------------------------------------------------------------------
536-------------------------------------------------------------------------------
538-- Lookahead streams can execute multiple tasks concurrently, ahead of time but
539-- always serves them in the same order as they appear in the stream. To
540-- implement lookahead streams efficiently we assign a sequence number to each
541-- task when the task is picked up for execution. When the task finishes the
542-- output is tagged with the same sequence number and we rearrange the outputs
543-- in sequence based on that number.
545-- To explain the mechanism imagine that the current task at the head of the
546-- stream has a "token" to yield to the output. The ownership of the token is
547-- determined by the current sequence number maintained in outputHeap. Sequence
548-- number is assigned when a task is queued. When a thread dequeues a task it
549-- picks up the sequence number as well and when the output is ready it uses
550-- the sequence number to queue the output to the outputQueue.
552-- The thread with current sequence number sends the output directly to the
553-- outputQueue. Other threads push the output to the outputHeap. When the task
554-- being queued on the heap is a stream of many elements we yield the first
555-- element on the heap and an MVar to wakeup for more elements. When such a
556-- task gets the "token" for outputQueue it can directly keep yielding all the
557-- elements to the outputQueue without checking for the "token".
559-- Note that no two outputs in the heap can have the same sequence numbers and
560-- therefore we do not need a stable heap. We have also separated the buffer
561-- for the current task (outputQueue) and the pending tasks (outputHeap) so
562-- that the pending tasks cannot interfere with the current task. Note that for
563-- a single task just the outputQueue is enough and for the case of many
564-- threads just a heap is good enough. However we balance between these two
565-- cases, so that both are efficient.
567-- For bigger streams it may make sense to have separate buffers for each
568-- stream. However, for singleton streams this may become inefficient. However,
569-- if we do not have separate buffers, then the streams that come later in
570-- sequence may hog the buffer, hindering the streams that are ahead. For this
571-- reason we have a single element buffer limitation for the streams being
572-- executed in advance.
574-- This scheme works pretty efficiently with only 50% more overhead compared to
575-- the Async streams where we do not have any kind of sequencing of the
576-- outputs. It is especially devised so that we are most efficient when we have
577-- short tasks and need just a single thread. Also when a thread yields many
578-- items it can hold lockfree access to the outputQueue and do it efficiently.
580-- XXX Maybe we can start the ahead threads at a lower cpu and IO priority so
581-- that they do not hog the resources and hinder the progress of the threads in
584{-# INLINE sendToHeap #-}
585sendToHeap :: SVar m a -> AheadHeapEntry a -> Int -> IO Bool
586sendToHeap sv msg seqNo = do
587 let maxHeapSize = 1500
589 -- Maybe we can use a lock instead of using CAS
590 -- insertions in the heap may be expensive if the heap is big
591 -- Note: len is valid only when added is True
592 (len, added) <- atomicModifyIORefCAS (outputHeap sv) $
595 then ((H.insert (Entry seqNo msg) h, snum), (H.size h, True))
596 else (ref, (0, False))
598 -- this is only for the case when the consumer generated siren can get
599 -- missed due to race.
600 when (len < maxHeapSize) $ void $ tryPutMVar (heapSiren sv) ()
604 when (len + 1 >= maxHeapSize) $ do
605 withDBGMVar sv "sendToHeap: wait for heapSiren"
606 $ takeMVar (heapSiren sv)
607 (h, _) <- readIORef (outputHeap sv)
608 when (H.size h <= maxHeapSize) $ void $ tryPutMVar (siren sv) ()
612 AheadYieldSingle a -> send sv (ChildYield a)
613 AheadYieldMany a mv -> do
614 send sv (ChildYield a)
615 withDBGMVar sv "sendToHeap: wakeup stream" $ putMVar mv ()
619 :: IORef (Heap (Entry Int (AheadHeapEntry a)), Int)
621 -> IO (Maybe (AheadHeapEntry a))
622popFromHeap hp nextSeqNo =
623 atomicModifyIORefCAS hp $ \(h, snum) -> do
625 let n = max nextSeqNo snum
627 Nothing -> ((h, n), Nothing)
628 Just (Entry seqNo ev, hp') ->
630 then ((hp', seqNo), Just ev)
631 else ((h, n), Nothing)
633-- | As soon as we set the nextSeqNo in outputHeap we have transferred the
634-- output "token" to the thread executing the task with that sequence number.
635-- We have to make sure that there is no entry in the heap that has nextSeqNo
636-- before we set nextSeqNo as the expected sequence number in the heap. It is
637-- possible that when we are clearing the heap entry of the current sequence
638-- number the next sequence number gets added and then we have to clear that,
639-- this can keep going on and on.
640drainHeap :: SVar m a -> Int -> IO ()
641drainHeap sv nextSeqNo = do
642 r <- popFromHeap (outputHeap sv) nextSeqNo
647 AheadYieldSingle a -> do
648 send sv (ChildYield a)
649 atomicModifyIORefCAS_ (outputHeap sv) $ \(h, snum) ->
651 drainHeap sv nextSeqNo
652 AheadYieldMany a mv -> do
653 send sv (ChildYield a)
654 withDBGMVar sv "drainHeap: wakeup stream" $ putMVar mv ()
656-- | The thread that has the token has the responsibility to move any pending
657-- entries in the heap to the outputQueue. When the token is assigned to a
658-- sequence number we must not have any entries for that sequence number in the
660releaseToken :: SVar m a -> Int -> IO ()
661releaseToken sv nextSeqNo = do
662 drainHeap sv nextSeqNo
663 void $ tryPutMVar (heapSiren sv) ()
665-- | When we have the output "token" we are allowed to append to the
666-- outputQueue otherwise we have to insert the output in the outputHeap.
667-- Returns whether we still have the token.
668handleOutput :: SVar m a
675handleOutput sv haveToken prevSeqNo seqNo a ev = do
678 let nextSeqNo = prevSeqNo + 1
679 if seqNo == nextSeqNo
680 then send sv (ChildYield a) >> return True
682 releaseToken sv nextSeqNo
685 where toHeap = sendToHeap sv ev seqNo
687-- Left associated ahead expressions are expensive. We start a new SVar for
688-- each left associative expression. The queue is used only for right
689-- associated expression, we queue the right expression and execute the left.
690-- Thererefore the queue never has more than on item in it.
691{-# INLINE enqueueAhead #-}
692enqueueAhead :: SVar m a -> IORef ([Stream m a], Int) -> Stream m a -> IO ()
693enqueueAhead sv q m = do
694 w <- readIORef $ waitingForWork sv
695 atomicModifyIORefCAS_ q $ \ case
696 ([], n) -> ([m], n + 1) -- increment sequence
697 _ -> error "not empty"
699 void $ tryPutMVar (doorBell sv) ()
700 atomicModifyIORefCAS_ (waitingForWork sv) (const False)
702runqueueAhead :: MonadIO m => SVar m a -> IORef ([Stream m a], Int) -> m ()
703runqueueAhead sv q = run False (-1)
707 run haveToken prevSeqNo = do
710 Nothing -> liftIO $ do
711 when haveToken $ releaseToken sv (prevSeqNo + 1)
713 Just (m, seqNo) -> do
714 let tStatus = haveToken || seqNo == 0
716 let nextSeqNo = prevSeqNo + 1
717 if tStatus && seqNo /= nextSeqNo
719 liftIO $ releaseToken sv nextSeqNo
721 else run tStatus seqNo
723 tokenStatus <- liftIO $
724 handleOutput sv tStatus prevSeqNo seqNo a
726 run tokenStatus seqNo
728 -- If we are yielding more than one item and we do not have the
729 -- token then we insert the current item in the pending
730 -- outputHeap and wait for our turn.
732 stopNext = run True seqNo
734 liftIO $ send sv (ChildYield a)
737 liftIO $ send sv (ChildYield a)
738 (runStream r) (Just sv) stopNext singleNext yieldNext
741 -- XXX we can allocate the MVar when actually inserting
742 -- in the heap and then return it.
746 handleOutput sv tStatus prevSeqNo seqNo a
747 (AheadYieldMany a mv)
748 when (not gotToken) $ void $
749 withDBGMVar sv "runqueueAhead: stream waiting"
751 (runStream r) (Just sv) stopNext singleNext yieldNext
753 (runStream m) (Just sv) stop single yield
755 dequeue = liftIO $ do
756 atomicModifyIORefCAS q $ \case
757 ([], n) -> (([], n), Nothing)
758 (x : [], n) -> (([], n), Just (x, n))
759 _ -> error "more than one item on queue"
761-- Thread tracking is needed for two reasons:
763-- 1) Killing threads on exceptions. Threads may not be left to go away by
764-- themselves because they may run for significant times before going away or
765-- worse they may be stuck in IO and never go away.
767-- 2) To know when all threads are done and the stream has ended.
769{-# NOINLINE addThread #-}
770addThread :: MonadIO m => SVar m a -> ThreadId -> m ()
772 liftIO $ modifyIORef (runningThreads sv) (S.insert tid)
775{-# INLINE delThread #-}
776delThread :: MonadIO m => SVar m a -> ThreadId -> m ()
778 liftIO $ modifyIORef (runningThreads sv) $ (\s -> S.delete tid s)
781-- If present then delete else add. This takes care of out of order add and
782-- delete i.e. a delete arriving before we even added a thread.
783-- This occurs when the forked thread is done even before the 'addThread' right
784-- after the fork gets a chance to run.
785{-# INLINE modifyThread #-}
786modifyThread :: MonadIO m => SVar m a -> ThreadId -> m ()
787modifyThread sv tid = do
788 changed <- liftIO $ atomicModifyIORefCAS (runningThreads sv) $ \old ->
789 if (S.member tid old)
790 then let new = (S.delete tid old) in (new, new)
791 else let new = (S.insert tid old) in (new, old)
793 then liftIO $ void $ tryPutMVar (doorBell sv) ()
796-- | This is safe even if we are adding more threads concurrently because if
797-- a child thread is adding another thread then anyway 'runningThreads' will
799{-# INLINE allThreadsDone #-}
800allThreadsDone :: MonadIO m => SVar m a -> m Bool
801allThreadsDone sv = liftIO $ S.null <$> readIORef (runningThreads sv)
803{-# NOINLINE handleChildException #-}
804handleChildException :: SVar m a -> SomeException -> IO ()
805handleChildException sv e = do
807 send sv (ChildStop tid (Just e))
809{-# NOINLINE pushWorker #-}
810pushWorker :: MonadAsync m => SVar m a -> m ()
812 liftIO $ atomicModifyIORefCAS_ (activeWorkers sv) $ \n -> n + 1
813 doFork (runqueue sv) (handleChildException sv) >>= addThread sv
815-- | In contrast to pushWorker which always happens only from the consumer
816-- thread, a pushWorkerPar can happen concurrently from multiple threads on the
817-- producer side. So we need to use a thread safe modification of
818-- runningThreads. Alternatively, we can use a CreateThread event to avoid
819-- using a CAS based modification.
820{-# NOINLINE pushWorkerPar #-}
821pushWorkerPar :: MonadAsync m => SVar m a -> Stream m a -> m ()
822pushWorkerPar sv m = do
823 -- We do not use activeWorkers in case of ParallelVar but still there is no
824 -- harm in maintaining it correctly.
825 liftIO $ atomicModifyIORefCAS_ (activeWorkers sv) $ \n -> n + 1
826 doFork (runOne sv m) (handleChildException sv) >>= modifyThread sv
828-- XXX When the queue is LIFO we can put a limit on the number of dispatches.
829-- Also, if a worker blocks on the output queue we can decide if we want to
830-- block or make it go away entirely, depending on the number of workers and
831-- the type of the queue.
832{-# INLINE sendWorkerWait #-}
833sendWorkerWait :: MonadAsync m => SVar m a -> m ()
834sendWorkerWait sv = do
835 -- When there is no output seen we dispatch more workers to help out if
836 -- there is work pending in the work queue. But we wait a little while
837 -- and check the output again so that we are not too aggressive.
838 -- If there is no output pending to process and there is no worker to be
839 -- sent then we block, so that we do not keep looping fruitlessly.
841 -- Note that we are guaranteed to have at least one outstanding worker when
842 -- we enter this function. So if we sleep we are guaranteed to be woken up.
843 -- Note that the worker count is only decremented during event processing
844 -- in fromStreamVar and therefore it is safe to read and use it without a
847 -- XXX we should wait in such a way so that we can be immediately
848 -- interrupted by a doorBell. We can sleep on the doorbell and use a
849 -- monitor thread to send us a doorbell.
850 liftIO $ threadDelay 200
851 (_, n) <- liftIO $ readIORef (outputQueue sv)
853 -- The queue may be empty temporarily if the worker has dequeued the
854 -- work item but has not enqueued the remaining part yet. For the same
855 -- reason, a worker may come back if it tries to dequeue and finds the
856 -- queue empty, even though the whole work has not finished yet.
857 done <- queueEmpty sv
860 cnt <- liftIO $ readIORef $ activeWorkers sv
861 -- Note that we may deadlock if the previous workers (tasks in the
862 -- stream) wait/depend on the future workers (tasks in the stream)
863 -- executing. In that case we should either configure the maxWorker
864 -- count to higher or use parallel style instead of ahead or async
870 else liftIO $ withDBGMVar sv "sendWorkerWait: wait for workers"
871 $ takeMVar (doorBell sv)
873 -- We found that the queue was empty, but it may be empty
874 -- temporarily, when we checked it. If that's the case we might
875 -- sleep indefinitely unless the active workers produce some
876 -- output. We may deadlock specially if the otuput from the other
877 -- workers depends on the future workers that we may never send.
878 -- So in case the queue was temporarily empty set a flag to inform
879 -- the enqueue to send us a doorbell.
881 liftIO $ atomicModifyIORefCAS_ (waitingForWork sv) $ const True
883 -- check again, this time we have set the waitingForWork flag so we
884 -- are guaranteed to get a doorbell in case the status changed
885 -- before we could sleep.
887 -- Note that this is just a best effort mechanism to avoid a
888 -- deadlock. Deadlocks may still happen if for some weird reason
889 -- the consuming computation shares an MVar or some other resource
890 -- with the producing computation and gets blocked on that resource
891 -- and therefore cannot do any pushworker to add more threads to
892 -- the producer. In such cases the programmer should use a parallel
893 -- style so that all the producers are scheduled immediately and
894 -- unconditionally. We can also use a separate monitor thread to
895 -- push workers instead of pushing them from the consumer, but then
896 -- we are no longer using pull based concurrency rate adaptation.
898 -- XXX update this in the tutorial.
901 liftIO $ withDBGMVar sv "sendWorkerWait: nothing to do"
902 $ takeMVar (doorBell sv)
904-- | Pull a stream from an SVar.
905{-# NOINLINE fromStreamVar #-}
906fromStreamVar :: MonadAsync m => SVar m a -> Stream m a
907fromStreamVar sv = Stream $ \_ stp sng yld -> do
908 if svarStyle sv == ParallelVar
909 then liftIO $ withDBGMVar sv "fromStreamVar: doorbell"
910 $ takeMVar (doorBell sv)
912 res <- liftIO $ tryTakeMVar (doorBell sv)
913 when (isNothing res) $ sendWorkerWait sv
915 (list, _) <- liftIO $ atomicModifyIORefCAS (outputQueue sv) $ \x -> (([],0), x)
916 void $ liftIO $ tryPutMVar (siren sv) ()
917 -- Reversing the output is important to guarantee that we process the
918 -- outputs in the same order as they were generated by the constituent
920 (runStream $ processEvents (reverse list)) Nothing stp sng yld
924 handleException e tid = do
925 liftIO $ atomicModifyIORefCAS_ (activeWorkers sv) $ \n -> n - 1
927 -- XXX implement kill async exception handling
928 -- liftIO $ readIORef (runningThreads sv) >>= mapM_ killThread
931 {-# INLINE processEvents #-}
932 processEvents [] = Stream $ \_ stp sng yld -> do
933 done <- allThreadsDone sv
935 then (runStream (fromStreamVar sv)) Nothing stp sng yld
938 processEvents (ev : es) = Stream $ \_ stp sng yld -> do
939 let continue = (runStream (processEvents es)) Nothing stp sng yld
940 yield a = yld a (processEvents es)
943 ChildYield a -> yield a
947 let active = activeWorkers sv
948 liftIO $ atomicModifyIORefCAS_ active $ \n -> n - 1
949 modifyThread sv tid >> continue
950 Just ex -> handleException ex tid
952getFifoSVar :: MonadIO m => SVarStyle -> IO (SVar m a)
953getFifoSVar ctype = do
954 outQ <- newIORef ([], 0)
955 outQMv <- newEmptyMVar
956 hooter <- newEmptyMVar
958 wfw <- newIORef False
959 running <- newIORef S.empty
962 SVar { outputQueue = outQ
965 , outputHeap = undefined
966 , heapSiren = undefined
967 , runningThreads = running
968 , workQueue = undefined
969 , runqueue = runqueueFIFO sv q
970 , enqueue = enqueueFIFO sv q
971 , queueEmpty = liftIO $ nullQ q
972 , waitingForWork = wfw
974 , activeWorkers = active
978getLifoSVar :: MonadIO m => SVarStyle -> IO (SVar m a)
979getLifoSVar ctype = do
980 outQ <- newIORef ([], 0)
981 outQMv <- newEmptyMVar
982 hooter <- newEmptyMVar
984 wfw <- newIORef False
985 running <- newIORef S.empty
987 let checkEmpty = null <$> liftIO (readIORef q)
989 SVar { outputQueue = outQ
992 , outputHeap = undefined
993 , heapSiren = undefined
994 , runningThreads = running
995 , workQueue = undefined
996 , runqueue = runqueueLIFO sv q
997 , enqueue = enqueueLIFO sv q
998 , queueEmpty = checkEmpty
999 , waitingForWork = wfw
1001 , activeWorkers = active
1005getParSVar :: SVarStyle -> IO (SVar m a)
1006getParSVar style = do
1007 outQ <- newIORef ([], 0)
1008 outQMv <- newEmptyMVar
1009 hooter <- newEmptyMVar
1010 active <- newIORef 0
1011 wfw <- newIORef False
1012 running <- newIORef S.empty
1014 SVar { outputQueue = outQ
1017 , outputHeap = undefined
1018 , heapSiren = undefined
1019 , runningThreads = running
1020 , workQueue = undefined
1021 , runqueue = undefined
1022 , enqueue = undefined
1023 , queueEmpty = undefined
1024 , waitingForWork = wfw
1026 , activeWorkers = active
1030getAheadSVar :: MonadIO m => SVarStyle -> IO (SVar m a)
1031getAheadSVar style = do
1032 outQ <- newIORef ([], 0)
1033 outH <- newIORef (H.empty, 0)
1034 outQMv <- newEmptyMVar
1035 hooter <- newEmptyMVar
1036 hsiren <- newEmptyMVar
1037 active <- newIORef 0
1038 wfw <- newIORef False
1039 running <- newIORef S.empty
1040 q <- newIORef ([], -1)
1041 let checkEmpty = liftIO $ do
1042 (xs, _) <- readIORef q
1045 SVar { outputQueue = outQ
1049 , heapSiren = hsiren
1050 , runningThreads = running
1052 , runqueue = runqueueAhead sv q
1053 , enqueue = undefined
1054 , queueEmpty = checkEmpty
1055 , waitingForWork = wfw
1057 , activeWorkers = active
1061-- | Create a new empty SVar.
1062newEmptySVar :: MonadAsync m => SVarStyle -> m (SVar m a)
1063newEmptySVar style = do
1066 WAsyncVar -> getFifoSVar style
1067 AsyncVar -> getLifoSVar style
1068 ParallelVar -> getParSVar style
1069 AheadVar -> getAheadSVar style
1071-- | Create a new SVar and enqueue one stream computation on it.
1072{-# INLINABLE newStreamVar1 #-}
1073newStreamVar1 :: MonadAsync m => SVarStyle -> Stream m a -> m (SVar m a)
1074newStreamVar1 style m = do
1075 sv <- newEmptySVar style
1076 -- Note: We must have all the work on the queue before sending the
1077 -- pushworker, otherwise the pushworker may exit before we even get a
1079 if style == ParallelVar
1080 then pushWorkerPar sv m
1082 liftIO $ (enqueue sv) m
1086-- | Create a new SVar and enqueue one stream computation on it.
1087{-# INLINABLE newStreamVarAhead #-}
1088newStreamVarAhead :: MonadAsync m => Stream m a -> m (SVar m a)
1089newStreamVarAhead m = do
1090 sv <- newEmptySVar AheadVar
1091 -- Note: We must have all the work on the queue before sending the
1092 -- pushworker, otherwise the pushworker may exit before we even get a
1094 liftIO $ enqueueAhead sv (workQueue sv) m
1098-- | Write a stream to an 'SVar' in a non-blocking manner. The stream can then
1099-- be read back from the SVar using 'fromSVar'.
1100toStreamVar :: MonadAsync m => SVar m a -> Stream m a -> m ()
1101toStreamVar sv m = do
1102 liftIO $ (enqueue sv) m
1103 done <- allThreadsDone sv
1104 -- XXX This is safe only when called from the consumer thread or when no
1105 -- consumer is present. There may be a race if we are not running in the
1107 when done $ pushWorker sv
1109------------------------------------------------------------------------------
1110-- Running streams concurrently
1111------------------------------------------------------------------------------
1113-- Concurrency rate control.
1115-- Our objective is to create more threads on demand if the consumer is running
1116-- faster than us. As soon as we encounter a concurrent composition we create a
1117-- push pull pair of threads. We use an SVar for communication between the
1118-- consumer, pulling from the SVar and the producer who is pushing to the SVar.
1119-- The producer creates more threads if the SVar drains and becomes empty, that
1120-- is the consumer is running faster.
1122-- XXX Note 1: This mechanism can be problematic if the initial production
1123-- latency is high, we may end up creating too many threads. So we need some
1124-- way to monitor and use the latency as well. Having a limit on the dispatches
1125-- (programmer controlled) may also help.
1127-- TBD Note 2: We may want to run computations at the lower level of the
1128-- composition tree serially even when they are composed using a parallel
1129-- combinator. We can use 'serial' in place of 'async' and 'wSerial' in
1130-- place of 'wAsync'. If we find that an SVar immediately above a computation
1131-- gets drained empty we can switch to parallelizing the computation. For that
1132-- we can use a state flag to fork the rest of the computation at any point of
1133-- time inside the Monad bind operation if the consumer is running at a faster
1136-- TBD Note 3: the binary operation ('parallel') composition allows us to
1137-- dispatch a chunkSize of only 1. If we have to dispatch in arbitrary
1138-- chunksizes we will need to compose the parallel actions using a data
1139-- constructor (A Free container) instead so that we can divide it in chunks of
1140-- arbitrary size before dispatching. If the stream is composed of
1141-- hierarchically composed grains of different sizes then we can always switch
1142-- to a desired granularity depending on the consumer speed.
1144-- TBD Note 4: for pure work (when we are not in the IO monad) we can divide it
1145-- into just the number of CPUs.
1147-- | Join two computations on the currently running 'SVar' queue for concurrent
1148-- execution. When we are using parallel composition, an SVar is passed around
1149-- as a state variable. We try to schedule a new parallel computation on the
1150-- SVar passed to us. The first time, when no SVar exists, a new SVar is
1151-- created. Subsequently, 'joinStreamVarAsync' may get called when a computation
1152-- already scheduled on the SVar is further evaluated. For example, when (a
1153-- `parallel` b) is evaluated it calls a 'joinStreamVarAsync' to put 'a' and 'b' on
1154-- the current scheduler queue.
1156-- The 'SVarStyle' required by the current composition context is passed as one
1157-- of the parameters. If the scheduling and composition style of the new
1158-- computation being scheduled is different than the style of the current SVar,
1159-- then we create a new SVar and schedule it on that. The newly created SVar
1160-- joins as one of the computations on the current SVar queue.
1162-- Cases when we need to switch to a new SVar:
1164-- * (x `parallel` y) `parallel` (t `parallel` u) -- all of them get scheduled on the same SVar
1165-- * (x `parallel` y) `parallel` (t `async` u) -- @t@ and @u@ get scheduled on a new child SVar
1166-- because of the scheduling policy change.
1167-- * if we 'adapt' a stream of type 'async' to a stream of type
1168-- 'Parallel', we create a new SVar at the transitioning bind.
1169-- * When the stream is switching from disjunctive composition to conjunctive
1170-- composition and vice-versa we create a new SVar to isolate the scheduling
1173{-# INLINE joinStreamVarAsync #-}
1174joinStreamVarAsync :: MonadAsync m
1175 => SVarStyle -> Stream m a -> Stream m a -> Stream m a
1176joinStreamVarAsync style m1 m2 = Stream $ \svr stp sng yld ->
1178 Just sv | svarStyle sv == style ->
1179 liftIO ((enqueue sv) m2) >> (runStream m1) svr stp sng yld
1181 sv <- newStreamVar1 style (concurrently m1 m2)
1182 (runStream (fromStreamVar sv)) Nothing stp sng yld
1184 concurrently ma mb = Stream $ \svr stp sng yld -> do
1185 liftIO $ enqueue (fromJust svr) mb
1186 (runStream ma) svr stp sng yld
1188{-# INLINE joinStreamVarPar #-}
1189joinStreamVarPar :: MonadAsync m
1190 => SVarStyle -> Stream m a -> Stream m a -> Stream m a
1191joinStreamVarPar style m1 m2 = Stream $ \svr stp sng yld ->
1193 Just sv | svarStyle sv == style -> do
1194 pushWorkerPar sv m1 >> (runStream m2) svr stp sng yld
1196 sv <- newEmptySVar style
1199 (runStream (fromStreamVar sv)) Nothing stp sng yld
1202ahead :: MonadAsync m => Stream m a -> Stream m a -> Stream m a
1203ahead m1 m2 = Stream $ \svr stp sng yld -> do
1205 Just sv | svarStyle sv == AheadVar -> do
1206 liftIO $ enqueueAhead sv (workQueue sv) m2
1207 -- Always run the left side on a new SVar to avoid complexity in
1208 -- sequencing results. This means the left side cannot further
1209 -- split into more ahead computations on the same SVar.
1210 (runStream m1) Nothing stp sng yld
1212 sv <- newStreamVarAhead (concurrently m1 m2)
1213 (runStream (fromStreamVar sv)) Nothing stp sng yld
1215 concurrently ma mb = Stream $ \svr stp sng yld -> do
1216 liftIO $ enqueueAhead (fromJust svr) (workQueue (fromJust svr)) mb
1217 (runStream ma) Nothing stp sng yld
1219-- | XXX we can implement it more efficienty by directly implementing instead
1220-- of combining streams using ahead.
1221{-# INLINE consMAhead #-}
1222consMAhead :: MonadAsync m => m a -> Stream m a -> Stream m a
1223consMAhead m r = once m `ahead` r
1225------------------------------------------------------------------------------
1226-- Semigroup and Monoid style compositions for parallel actions
1227------------------------------------------------------------------------------
1230async :: MonadAsync m => Stream m a -> Stream m a -> Stream m a
1231async = joinStreamVarAsync AsyncVar
1233-- | XXX we can implement it more efficienty by directly implementing instead
1234-- of combining streams using async.
1235{-# INLINE consMAsync #-}
1236consMAsync :: MonadAsync m => m a -> Stream m a -> Stream m a
1237consMAsync m r = once m `async` r
1239{-# INLINE wAsync #-}
1240wAsync :: MonadAsync m => Stream m a -> Stream m a -> Stream m a
1241wAsync = joinStreamVarAsync WAsyncVar
1243-- | XXX we can implement it more efficienty by directly implementing instead
1244-- of combining streams using wAsync.
1245{-# INLINE consMWAsync #-}
1246consMWAsync :: MonadAsync m => m a -> Stream m a -> Stream m a
1247consMWAsync m r = once m `wAsync` r
1249{-# INLINE parallel #-}
1250parallel :: MonadAsync m => Stream m a -> Stream m a -> Stream m a
1251parallel = joinStreamVarPar ParallelVar
1253-- | XXX we can implement it more efficienty by directly implementing instead
1254-- of combining streams using parallel.
1255{-# INLINE consMParallel #-}
1256consMParallel :: MonadAsync m => m a -> Stream m a -> Stream m a
1257consMParallel m r = once m `parallel` r
1259-------------------------------------------------------------------------------
1260-- Functor instace is the same for all types
1261-------------------------------------------------------------------------------
1263instance Monad m => Functor (Stream m) where
1264 fmap f m = Stream $ \_ stp sng yld ->
1265 let single = sng . f
1266 yield a r = yld (f a) (fmap f r)
1267 in (runStream m) Nothing stp single yield
1269------------------------------------------------------------------------------
1270-- Alternative & MonadPlus
1271------------------------------------------------------------------------------
1273_alt :: Stream m a -> Stream m a -> Stream m a
1274_alt m1 m2 = Stream $ \_ stp sng yld ->
1275 let stop = runStream m2 Nothing stp sng yld
1276 in runStream m1 Nothing stop sng yld
1278------------------------------------------------------------------------------
1279-- Stream to stream function application
1280------------------------------------------------------------------------------
1282applyWith :: MonadAsync m
1283 => SVarStyle -> (Stream m a -> Stream m b) -> Stream m a -> Stream m b
1284applyWith style f m = Stream $ \svr stp sng yld -> do
1285 sv <- newStreamVar1 style m
1286 runStream (f $ fromStreamVar sv) svr stp sng yld
1288------------------------------------------------------------------------------
1289-- Stream runner function application
1290------------------------------------------------------------------------------
1292runWith :: MonadAsync m
1293 => SVarStyle -> (Stream m a -> m b) -> Stream m a -> m b
1294runWith style f m = do
1295 sv <- newStreamVar1 style m
1296 f $ fromStreamVar sv
1298------------------------------------------------------------------------------
1300------------------------------------------------------------------------------
1302zipWith :: (a -> b -> c) -> Stream m a -> Stream m b -> Stream m c
1303zipWith f m1 m2 = go m1 m2
1305 go mx my = Stream $ \_ stp sng yld -> do
1307 let single2 b = sng (f a b)
1308 yield2 b rb = yld (f a b) (go ra rb)
1309 in (runStream my) Nothing stp single2 yield2
1310 let single1 a = merge a nil
1311 yield1 a ra = merge a ra
1312 (runStream mx) Nothing stp single1 yield1
1314zipAsyncWith :: MonadAsync m
1315 => (a -> b -> c) -> Stream m a -> Stream m b -> Stream m c
1316zipAsyncWith f m1 m2 = Stream $ \_ stp sng yld -> do
1319 (runStream (zipWith f ma mb)) Nothing stp sng yld
1323 mkAsync :: MonadAsync m => Stream m a -> m (Stream m a)
1324 mkAsync m = newStreamVar1 AsyncVar m
1325 >>= return . fromStreamVar
1327-------------------------------------------------------------------------------
1329-------------------------------------------------------------------------------
1331instance MonadTrans Stream where
keysesc to close
- j k
- move down, up
- enter
- open
- esc
- back, or up a level
- 1 2 3 4
- readme, files, commits, refs
- /
- filter
- y
- copy the clone command
- c
- compare the chosen commit with another (or shift-click both)
- ?
- show or hide this list