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,
97-- but keep it on for now to debug problems quickly if and when they happen.
101import Control.Exception (catches, throwIO, Handler(..),
102 BlockedIndefinitelyOnMVar(..),
103 BlockedIndefinitelyOnSTM(..))
104import System.IO (hPutStrLn, stderr)
107------------------------------------------------------------------------------
108-- Parent child thread communication type
109------------------------------------------------------------------------------
111-- | Events that a child thread may send to a parent thread.
114 | ChildStop ThreadId (Maybe SomeException)
116-- | Sorting out of turn outputs in a heap for Ahead style streams
117data AheadHeapEntry a =
119 | AheadYieldMany a (MVar ()) -- value, MVar to wakeup for more values
121------------------------------------------------------------------------------
122-- State threaded around the monad for thread management
123------------------------------------------------------------------------------
125-- XXX use a separate data structure for each type of SVar
126-- | Identify the type of the SVar. Two computations using the same style can
127-- be scheduled on the same SVar.
129 AsyncVar -- depth first concurrent
130 | WAsyncVar -- breadth first concurrent
131 | ParallelVar -- all parallel
132 | AheadVar -- Concurrent look ahead
135-- | An SVar or a Stream Var is a conduit to the output from multiple streams
136-- running concurrently and asynchronously. An SVar can be thought of as an
137-- asynchronous IO handle. We can write any number of streams to an SVar in a
138-- non-blocking manner and then read them back at any time at any pace. The
139-- SVar would run the streams asynchronously and accumulate results. An SVar
140-- may not really execute the stream completely and accumulate all the results.
141-- However, it ensures that the reader can read the results at whatever paces
142-- it wants to read. The SVar monitors and adapts to the consumer's pace.
144-- An SVar is a mini scheduler, it has an associated runqueue that holds the
145-- stream tasks to be picked and run by a pool of worker threads. It has an
146-- associated output queue where the output stream elements are placed by the
147-- worker threads. A doorBell is used by the worker threads to intimate the
148-- consumer thread about availability of new results in the output queue. More
149-- workers are added to the SVar by 'fromStreamVar' on demand if the output
150-- produced is not keeping pace with the consumer. On bounded SVars, workers
151-- block on the output queue to provide throttling of the producer when the
152-- consumer is not pulling fast enough. The number of workers may even get
153-- reduced depending on the consuming pace.
155-- New work is enqueued either at the time of creation of the SVar or as a
156-- result of executing the parallel combinators i.e. '<|' and '<|>' when the
157-- already enqueued computations get evaluated. See 'joinStreamVarAsync'.
162 svarStyle :: SVarStyle
164 -- Shared output queue (events, length)
165 , outputQueue :: IORef ([ChildEvent a], Int)
166 , doorBell :: MVar () -- signal the consumer about output
167 , siren :: MVar () -- hooter for workers to begin work
169 -- Output synchronization mechanism for Ahead streams (Ahead and
170 -- wAhead). We maintain a heap of out of sequence ahead of time
171 -- generated outputs and the sequence number of the task that is
172 -- currently at the head of the stream. Concurrent execute ahead
173 -- tasks that have a sequence number greater than the task at the
174 -- head should add their output to the heap.
175 , outputHeap :: IORef (Heap (Entry Int (AheadHeapEntry a)), Int)
176 , heapSiren :: MVar () -- hooter for workers to continue
177 -- adding to the heap
180 , workQueue :: IORef ([Stream m a], Int)
181 , enqueue :: Stream m a -> IO ()
182 , queueEmpty :: m Bool
183 , waitingForWork :: IORef Bool
186 -- Shared, thread tracking
187 , runningThreads :: IORef (Set ThreadId)
188 , activeWorkers :: IORef Int
192{-# NOINLINE dumpSVar #-}
193dumpSVar :: SVar m a -> IO String
196 (oqList, oqLen) <- readIORef $ outputQueue sv
197 db <- tryTakeMVar $ doorBell sv
198 srn <- tryTakeMVar $ siren sv
200 if svarStyle sv == AheadVar
202 (oheap, oheapSize) <- readIORef $ outputHeap sv
203 hsrn <- tryTakeMVar $ heapSiren sv
204 (wq, wqSize) <- readIORef $ workQueue sv
206 [ "heap length computed = " ++ show (H.size oheap)
207 , "heap length maintained = " ++ show oheapSize
208 , "heap siren = " ++ show hsrn
209 , "work queue length computed = " ++ show (length wq)
210 , "work queue length maintained = " ++ show wqSize
214 waiting <- readIORef $ waitingForWork sv
215 rthread <- readIORef $ runningThreads sv
216 workers <- readIORef $ activeWorkers sv
217 -- XXX queueEmpty should be made IO return type
220 [ "tid = " ++ show tid
221 , "style = " ++ show (svarStyle sv)
222 , "outputQueue length computed = " ++ show (length oqList)
223 , "outputQueue length maintained = " ++ show oqLen
224 , "output doorBell = " ++ show db
225 , "worker siren = " ++ show srn
227 ++ aheadDump ++ unlines
228 [ "waitingForWork = " ++ show waiting
229 , "running threads = " ++ show rthread
230 , "running thread count = " ++ show workers
233{-# NOINLINE mvarExcHandler #-}
234mvarExcHandler :: SVar m a -> String -> BlockedIndefinitelyOnMVar -> IO ()
235mvarExcHandler sv label e@BlockedIndefinitelyOnMVar = do
236 svInfo <- dumpSVar sv
237 hPutStrLn stderr $ label ++ " " ++ "BlockedIndefinitelyOnMVar\n" ++ svInfo
240{-# NOINLINE stmExcHandler #-}
241stmExcHandler :: SVar m a -> String -> BlockedIndefinitelyOnSTM -> IO ()
242stmExcHandler sv label e@BlockedIndefinitelyOnSTM = do
243 svInfo <- dumpSVar sv
244 hPutStrLn stderr $ label ++ " " ++ "BlockedIndefinitelyOnSTM\n" ++ svInfo
247withDBGMVar :: SVar m a -> String -> IO () -> IO ()
248withDBGMVar sv label action =
249 action `catches` [ Handler (mvarExcHandler sv label)
250 , Handler (stmExcHandler sv label)
253withDBGMVar :: SVar m a -> String -> IO () -> IO ()
254withDBGMVar _ _ action = action
257-- Slightly faster version of CAS. Gained some improvement by avoiding the use
258-- of "evaluate" because we know we do not have exceptions in fn.
259{-# INLINE atomicModifyIORefCAS #-}
260atomicModifyIORefCAS :: IORef a -> (a -> (a,b)) -> IO b
261atomicModifyIORefCAS ref fn = do
262 tkt <- readForCAS ref
268 loop _ 0 = atomicModifyIORef ref fn
270 let (new, result) = fn $ peekTicket old
271 (success, tkt) <- casIORef ref old new
274 else loop tkt (tries - 1)
276------------------------------------------------------------------------------
278------------------------------------------------------------------------------
280-- | The type 'Stream m a' represents a monadic stream of values of type 'a'
281-- constructed using actions in monad 'm'. It uses stop, singleton and yield
282-- continuations equivalent to the following direct style type:
284-- data Stream m a = Stop | Singleton a | Yield a (Stream m a)
286-- To facilitate parallel composition we maintain a local state in an SVar that
287-- is shared across and is used for synchronization of the streams being
290-- The singleton case can be expressed in terms of stop and yield but we have
291-- it as a separate case to optimize composition operations for streams with
292-- single element. We build singleton streams in the implementation of 'pure'
293-- for Applicative and Monad, and in 'lift' for MonadTrans.
297 runStream :: forall r.
298 Maybe (SVar m a) -- local state
300 -> (a -> m r) -- singleton
301 -> (a -> Stream m a -> m r) -- yield
306nil = Stream $ \_ stp _ _ -> stp
308-- | faster than consM because there is no bind.
309cons :: a -> Stream m a -> Stream m a
310cons a r = Stream $ \_ _ _ yld -> yld a r
312-- | Same as @once . return@ but may be faster because there is no bind
313singleton :: a -> Stream m a
314singleton a = Stream $ \_ _ single _ -> single a
317once :: Monad m => m a -> Stream m a
318once m = Stream $ \_ _ single _ -> m >>= single
321consM :: Monad m => m a -> Stream m a -> Stream m a
322consM m r = Stream $ \_ _ _ yld -> m >>= \a -> yld a r
324repeat :: a -> Stream m a
325repeat a = let x = cons a x in x
327------------------------------------------------------------------------------
329------------------------------------------------------------------------------
331-- | Concatenates two streams sequentially i.e. the first stream is
332-- exhausted completely before yielding any element from the second stream.
334serial :: Stream m a -> Stream m a -> Stream m a
337 go (Stream m) = Stream $ \_ stp sng yld ->
338 let stop = (runStream m2) Nothing stp sng yld
340 yield a r = yld a (go r)
341 in m Nothing stop single yield
343instance Semigroup (Stream m a) where
346------------------------------------------------------------------------------
348------------------------------------------------------------------------------
350instance Monoid (Stream m a) where
354------------------------------------------------------------------------------
356------------------------------------------------------------------------------
358{-# INLINE wSerial #-}
359wSerial :: Stream m a -> Stream m a -> Stream m a
360wSerial m1 m2 = Stream $ \_ stp sng yld -> do
361 let stop = (runStream m2) Nothing stp sng yld
363 yield a r = yld a (wSerial m2 r)
364 (runStream m1) Nothing stop single yield
366------------------------------------------------------------------------------
367-- Spawning threads and collecting result in streamed fashion
368------------------------------------------------------------------------------
370-- | A monad that can perform concurrent or parallel IO operations. Streams
371-- that can be composed concurrently require the underlying monad to be
375type MonadAsync m = (MonadIO m, MonadBaseControl IO m, MonadThrow m)
377-- Stolen from the async package. The perf improvement is modest, 2% on a
378-- thread heavy benchmark (parallel composition using noop computations).
379-- A version of forkIO that does not include the outer exception
380-- handler: saves a bit of time when we will be installing our own
382{-# INLINE rawForkIO #-}
383rawForkIO :: IO () -> IO ThreadId
384rawForkIO action = IO $ \ s ->
385 case (fork# action s) of (# s1, tid #) -> (# s1, ThreadId tid #)
388doFork :: MonadBaseControl IO m
390 -> (SomeException -> IO ())
392doFork action exHandler =
393 control $ \runInIO ->
394 mask $ \restore -> do
395 tid <- rawForkIO $ catch (restore $ void $ runInIO action)
399-- XXX exception safety of all atomic/MVar operations
401-- TBD Each worker can have their own queue and the consumer can empty one
402-- queue at a time, that way contention can be reduced.
404-- | This function is used by the producer threads to queue output for the
405-- consumer thread to consume.
407send :: SVar m a -> ChildEvent a -> IO ()
409 len <- atomicModifyIORefCAS (outputQueue sv) $ \(es, n) ->
410 ((msg : es, n + 1), n)
411 if (len <= 0) then do
412 -- XXX need a memory barrier? The wake up must happen only after the
413 -- store has finished otherwise we can have lost wakeup problems.
415 -- Since multiple workers can try this at the same time, it is possible
416 -- that we may put a spurious MVar after the consumer has already seen
417 -- the output. But that's harmless, at worst it may cause the consumer
418 -- to read the queue again and find it empty.
419 -- The important point is that the consumer is guaranteed to receive a
420 -- doorbell if something was added to the queue after it empties it.
421 void $ tryPutMVar (doorBell sv) ()
422 -- those who were waiting because of buffer limit should now go ahead
423 void $ tryPutMVar (siren sv) ()
424 -- The only thread that reduces the length of the queue is the consumer
425 -- thread. It is possible that the queue may have been emptied in the
426 -- window when we acquired "len" and now when we are using it. In that case
427 -- we will be woken up by the consumer blowing the siren.
428 -- Note: current length after adding our output is len + 1
429 else when (len + 1 >= 1500) $ do
430 withDBGMVar sv "send: wait for siren" $ takeMVar (siren sv)
431 (_, n) <- readIORef (outputQueue sv)
432 when (n <= 1500) $ void $ tryPutMVar (siren sv) ()
434{-# INLINE sendStop #-}
435sendStop :: SVar m a -> IO ()
436sendStop sv = myThreadId >>= \tid -> send sv (ChildStop tid Nothing)
438-------------------------------------------------------------------------------
440-------------------------------------------------------------------------------
442-- Note: For purely right associated expressions this queue should have at most
443-- one element. It grows to more than one when we have left associcated
444-- expressions. Large left associated compositions can grow this to a
446{-# INLINE enqueueLIFO #-}
447enqueueLIFO :: SVar m a -> IORef [Stream m a] -> Stream m a -> IO ()
448enqueueLIFO sv q m = do
449 w <- readIORef $ waitingForWork sv
452 else atomicModifyIORefCAS_ q $ \ms -> m : ms
457 v <- atomicModifyIORefCAS q $ \ ms ->
461 when v $ void $ tryPutMVar (doorBell sv) ()
462 atomicModifyIORefCAS_ (waitingForWork sv) (const False)
464runqueueLIFO :: MonadIO m => SVar m a -> IORef [Stream m a] -> m ()
465runqueueLIFO sv q = run
472 Nothing -> liftIO $ sendStop sv
473 Just m -> (runStream m) (Just sv) run single yield
475 sendit a = liftIO $ send sv (ChildYield a)
476 single a = sendit a >> run
477 yield a r = sendit a >> (runStream r) (Just sv) run single yield
479 dequeue = liftIO $ atomicModifyIORefCAS q $ \case
481 x : xs -> (xs, Just x)
483-------------------------------------------------------------------------------
485-------------------------------------------------------------------------------
487{-# INLINE enqueueFIFO #-}
488enqueueFIFO :: SVar m a -> LinkedQueue (Stream m a) -> Stream m a -> IO ()
489enqueueFIFO sv q m = do
490 w <- readIORef $ waitingForWork sv
500 when emp $ void $ tryPutMVar (doorBell sv) ()
502runqueueFIFO :: MonadIO m => SVar m a -> LinkedQueue (Stream m a) -> m ()
503runqueueFIFO sv q = run
510 Nothing -> liftIO $ sendStop sv
511 Just m -> (runStream m) (Just sv) run single yield
513 dequeue = liftIO $ tryPopR q
514 sendit a = liftIO $ send sv (ChildYield a)
515 single a = sendit a >> run
516 yield a r = sendit a >> liftIO (enqueueFIFO sv q r) >> run
518-------------------------------------------------------------------------------
520-------------------------------------------------------------------------------
523runOne :: MonadIO m => SVar m a -> Stream m a -> m ()
524runOne sv m = (runStream m) (Just sv) stop single yield
528 stop = liftIO $ sendStop sv
529 sendit a = liftIO $ send sv (ChildYield a)
530 single a = sendit a >> stop
531 yield a r = sendit a >> runOne sv r
533-------------------------------------------------------------------------------
535-------------------------------------------------------------------------------
537-- Lookahead streams can execute multiple tasks concurrently, ahead of time but
538-- always serves them in the same order as they appear in the stream. To
539-- implement lookahead streams efficiently we assign a sequence number to each
540-- task when the task is picked up for execution. When the task finishes the
541-- output is tagged with the same sequence number and we rearrange the outputs
542-- in sequence based on that number.
544-- To explain the mechanism imagine that the current task at the head of the
545-- stream has a "token" to yield to the output. The ownership of the token is
546-- determined by the current sequence number maintained in outputHeap. Sequence
547-- number is assigned when a task is queued. When a thread dequeues a task it
548-- picks up the sequence number as well and when the output is ready it uses
549-- the sequence number to queue the output to the outputQueue.
551-- The thread with current sequence number sends the output directly to the
552-- outputQueue. Other threads push the output to the outputHeap. When the task
553-- being queued on the heap is a stream of many elements we yield the first
554-- element on the heap and an MVar to wakeup for more elements. When such a
555-- task gets the "token" for outputQueue it can directly keep yielding all the
556-- elements to the outputQueue without checking for the "token".
558-- Note that no two outputs in the heap can have the same sequence numbers and
559-- therefore we do not need a stable heap. We have also separated the buffer
560-- for the current task (outputQueue) and the pending tasks (outputHeap) so
561-- that the pending tasks cannot interfere with the current task. Note that for
562-- a single task just the outputQueue is enough and for the case of many
563-- threads just a heap is good enough. However we balance between these two
564-- cases, so that both are efficient.
566-- For bigger streams it may make sense to have separate buffers for each
567-- stream. However, for singleton streams this may become inefficient. However,
568-- if we do not have separate buffers, then the streams that come later in
569-- sequence may hog the buffer, hindering the streams that are ahead. For this
570-- reason we have a single element buffer limitation for the streams being
571-- executed in advance.
573-- This scheme works pretty efficiently with only 50% more overhead compared to
574-- the Async streams where we do not have any kind of sequencing of the
575-- outputs. It is especially devised so that we are most efficient when we have
576-- short tasks and need just a single thread. Also when a thread yields many
577-- items it can hold lockfree access to the outputQueue and do it efficiently.
579-- XXX Maybe we can start the ahead threads at a lower cpu and IO priority so
580-- that they do not hog the resources and hinder the progress of the threads in
583{-# INLINE sendToHeap #-}
584sendToHeap :: SVar m a -> AheadHeapEntry a -> Int -> IO Bool
585sendToHeap sv msg seqNo = do
586 let maxHeapSize = 1500
588 -- Maybe we can use a lock instead of using CAS
589 -- insertions in the heap may be expensive if the heap is big
590 -- Note: len is valid only when added is True
591 (len, added) <- atomicModifyIORefCAS (outputHeap sv) $
594 then ((H.insert (Entry seqNo msg) h, snum), (H.size h, True))
595 else (ref, (0, False))
597 -- this is only for the case when the consumer generated siren can get
598 -- missed due to race.
599 when (len < maxHeapSize) $ void $ tryPutMVar (heapSiren sv) ()
603 when (len + 1 >= maxHeapSize) $ do
604 withDBGMVar sv "sendToHeap: wait for heapSiren"
605 $ takeMVar (heapSiren sv)
606 (h, _) <- readIORef (outputHeap sv)
607 when (H.size h <= maxHeapSize) $ void $ tryPutMVar (siren sv) ()
611 AheadYieldSingle a -> send sv (ChildYield a)
612 AheadYieldMany a mv -> do
613 send sv (ChildYield a)
614 withDBGMVar sv "sendToHeap: wakeup stream" $ putMVar mv ()
618 :: IORef (Heap (Entry Int (AheadHeapEntry a)), Int)
620 -> IO (Maybe (AheadHeapEntry a))
621popFromHeap hp nextSeqNo =
622 atomicModifyIORefCAS hp $ \(h, snum) -> do
624 let n = max nextSeqNo snum
626 Nothing -> ((h, n), Nothing)
627 Just (Entry seqNo ev, hp') ->
629 then ((hp', seqNo), Just ev)
630 else ((h, n), Nothing)
632-- | As soon as we set the nextSeqNo in outputHeap we have transferred the
633-- output "token" to the thread executing the task with that sequence number.
634-- We have to make sure that there is no entry in the heap that has nextSeqNo
635-- before we set nextSeqNo as the expected sequence number in the heap. It is
636-- possible that when we are clearing the heap entry of the current sequence
637-- number the next sequence number gets added and then we have to clear that,
638-- this can keep going on and on.
639drainHeap :: SVar m a -> Int -> IO ()
640drainHeap sv nextSeqNo = do
641 r <- popFromHeap (outputHeap sv) nextSeqNo
646 AheadYieldSingle a -> do
647 send sv (ChildYield a)
648 atomicModifyIORefCAS_ (outputHeap sv) $ \(h, snum) ->
650 drainHeap sv nextSeqNo
651 AheadYieldMany a mv -> do
652 send sv (ChildYield a)
653 withDBGMVar sv "drainHeap: wakeup stream" $ putMVar mv ()
655-- | The thread that has the token has the responsibility to move any pending
656-- entries in the heap to the outputQueue. When the token is assigned to a
657-- sequence number we must not have any entries for that sequence number in the
659releaseToken :: SVar m a -> Int -> IO ()
660releaseToken sv nextSeqNo = do
661 drainHeap sv nextSeqNo
662 void $ tryPutMVar (heapSiren sv) ()
664-- | When we have the output "token" we are allowed to append to the
665-- outputQueue otherwise we have to insert the output in the outputHeap.
666-- Returns whether we still have the token.
667handleOutput :: SVar m a
674handleOutput sv haveToken prevSeqNo seqNo a ev = do
677 let nextSeqNo = prevSeqNo + 1
678 if seqNo == nextSeqNo
679 then send sv (ChildYield a) >> return True
681 releaseToken sv nextSeqNo
684 where toHeap = sendToHeap sv ev seqNo
686-- Left associated ahead expressions are expensive. We start a new SVar for
687-- each left associative expression. The queue is used only for right
688-- associated expression, we queue the right expression and execute the left.
689-- Thererefore the queue never has more than on item in it.
690{-# INLINE enqueueAhead #-}
691enqueueAhead :: SVar m a -> IORef ([Stream m a], Int) -> Stream m a -> IO ()
692enqueueAhead sv q m = do
693 w <- readIORef $ waitingForWork sv
694 atomicModifyIORefCAS_ q $ \ case
695 ([], n) -> ([m], n + 1) -- increment sequence
696 _ -> error "not empty"
698 void $ tryPutMVar (doorBell sv) ()
699 atomicModifyIORefCAS_ (waitingForWork sv) (const False)
701runqueueAhead :: MonadIO m => SVar m a -> IORef ([Stream m a], Int) -> m ()
702runqueueAhead sv q = run False (-1)
706 run haveToken prevSeqNo = do
709 Nothing -> liftIO $ do
710 when haveToken $ releaseToken sv (prevSeqNo + 1)
712 Just (m, seqNo) -> do
713 let tStatus = haveToken || seqNo == 0
715 let nextSeqNo = prevSeqNo + 1
716 if tStatus && seqNo /= nextSeqNo
718 liftIO $ releaseToken sv nextSeqNo
720 else run tStatus seqNo
722 tokenStatus <- liftIO $
723 handleOutput sv tStatus prevSeqNo seqNo a
725 run tokenStatus seqNo
727 -- If we are yielding more than one item and we do not have the
728 -- token then we insert the current item in the pending
729 -- outputHeap and wait for our turn.
731 stopNext = run True seqNo
733 liftIO $ send sv (ChildYield a)
736 liftIO $ send sv (ChildYield a)
737 (runStream r) (Just sv) stopNext singleNext yieldNext
740 -- XXX we can allocate the MVar when actually inserting
741 -- in the heap and then return it.
745 handleOutput sv tStatus prevSeqNo seqNo a
746 (AheadYieldMany a mv)
747 when (not gotToken) $ void $
748 withDBGMVar sv "runqueueAhead: stream waiting"
750 (runStream r) (Just sv) stopNext singleNext yieldNext
752 (runStream m) (Just sv) stop single yield
754 dequeue = liftIO $ do
755 atomicModifyIORefCAS q $ \case
756 ([], n) -> (([], n), Nothing)
757 (x : [], n) -> (([], n), Just (x, n))
758 _ -> error "more than one item on queue"
760-- Thread tracking is needed for two reasons:
762-- 1) Killing threads on exceptions. Threads may not be left to go away by
763-- themselves because they may run for significant times before going away or
764-- worse they may be stuck in IO and never go away.
766-- 2) To know when all threads are done and the stream has ended.
768{-# NOINLINE addThread #-}
769addThread :: MonadIO m => SVar m a -> ThreadId -> m ()
771 liftIO $ modifyIORef (runningThreads sv) (S.insert tid)
774{-# INLINE delThread #-}
775delThread :: MonadIO m => SVar m a -> ThreadId -> m ()
777 liftIO $ modifyIORef (runningThreads sv) $ (\s -> S.delete tid s)
780-- If present then delete else add. This takes care of out of order add and
781-- delete i.e. a delete arriving before we even added a thread.
782-- This occurs when the forked thread is done even before the 'addThread' right
783-- after the fork gets a chance to run.
784{-# INLINE modifyThread #-}
785modifyThread :: MonadIO m => SVar m a -> ThreadId -> m ()
786modifyThread sv tid = do
787 changed <- liftIO $ atomicModifyIORefCAS (runningThreads sv) $ \old ->
788 if (S.member tid old)
789 then let new = (S.delete tid old) in (new, new)
790 else let new = (S.insert tid old) in (new, old)
792 then liftIO $ void $ tryPutMVar (doorBell sv) ()
795-- | This is safe even if we are adding more threads concurrently because if
796-- a child thread is adding another thread then anyway 'runningThreads' will
798{-# INLINE allThreadsDone #-}
799allThreadsDone :: MonadIO m => SVar m a -> m Bool
800allThreadsDone sv = liftIO $ S.null <$> readIORef (runningThreads sv)
802{-# NOINLINE handleChildException #-}
803handleChildException :: SVar m a -> SomeException -> IO ()
804handleChildException sv e = do
806 send sv (ChildStop tid (Just e))
808{-# NOINLINE pushWorker #-}
809pushWorker :: MonadAsync m => SVar m a -> m ()
811 liftIO $ atomicModifyIORefCAS_ (activeWorkers sv) $ \n -> n + 1
812 doFork (runqueue sv) (handleChildException sv) >>= addThread sv
814-- | In contrast to pushWorker which always happens only from the consumer
815-- thread, a pushWorkerPar can happen concurrently from multiple threads on the
816-- producer side. So we need to use a thread safe modification of
817-- runningThreads. Alternatively, we can use a CreateThread event to avoid
818-- using a CAS based modification.
819{-# NOINLINE pushWorkerPar #-}
820pushWorkerPar :: MonadAsync m => SVar m a -> Stream m a -> m ()
821pushWorkerPar sv m = do
822 -- We do not use activeWorkers in case of ParallelVar but still there is no
823 -- harm in maintaining it correctly.
824 liftIO $ atomicModifyIORefCAS_ (activeWorkers sv) $ \n -> n + 1
825 doFork (runOne sv m) (handleChildException sv) >>= modifyThread sv
827-- XXX When the queue is LIFO we can put a limit on the number of dispatches.
828-- Also, if a worker blocks on the output queue we can decide if we want to
829-- block or make it go away entirely, depending on the number of workers and
830-- the type of the queue.
831{-# INLINE sendWorkerWait #-}
832sendWorkerWait :: MonadAsync m => SVar m a -> m ()
833sendWorkerWait sv = do
834 -- When there is no output seen we dispatch more workers to help out if
835 -- there is work pending in the work queue. But we wait a little while
836 -- and check the output again so that we are not too aggressive.
837 -- If there is no output pending to process and there is no worker to be
838 -- sent then we block, so that we do not keep looping fruitlessly.
840 -- Note that we are guaranteed to have at least one outstanding worker when
841 -- we enter this function. So if we sleep we are guaranteed to be woken up.
842 -- Note that the worker count is only decremented during event processing
843 -- in fromStreamVar and therefore it is safe to read and use it without a
846 -- XXX we should wait in such a way so that we can be immediately
847 -- interrupted by a doorBell. We can sleep on the doorbell and use a
848 -- monitor thread to send us a doorbell.
849 liftIO $ threadDelay 200
850 (_, n) <- liftIO $ readIORef (outputQueue sv)
852 -- The queue may be empty temporarily if the worker has dequeued the
853 -- work item but has not enqueued the remaining part yet. For the same
854 -- reason, a worker may come back if it tries to dequeue and finds the
855 -- queue empty, even though the whole work has not finished yet.
856 done <- queueEmpty sv
859 cnt <- liftIO $ readIORef $ activeWorkers sv
860 -- Note that we may deadlock if the previous workers (tasks in the
861 -- stream) wait/depend on the future workers (tasks in the stream)
862 -- executing. In that case we should either configure the maxWorker
863 -- count to higher or use parallel style instead of ahead or async
869 else liftIO $ withDBGMVar sv "sendWorkerWait: wait for workers"
870 $ takeMVar (doorBell sv)
872 -- We found that the queue was empty, but it may be empty
873 -- temporarily, when we checked it. If that's the case we might
874 -- sleep indefinitely unless the active workers produce some
875 -- output. We may deadlock specially if the otuput from the other
876 -- workers depends on the future workers that we may never send.
877 -- So in case the queue was temporarily empty set a flag to inform
878 -- the enqueue to send us a doorbell.
880 liftIO $ atomicModifyIORefCAS_ (waitingForWork sv) $ const True
882 -- check again, this time we have set the waitingForWork flag so we
883 -- are guaranteed to get a doorbell in case the status changed
884 -- before we could sleep.
886 -- Note that this is just a best effort mechanism to avoid a
887 -- deadlock. Deadlocks may still happen if for some weird reason
888 -- the consuming computation shares an MVar or some other resource
889 -- with the producing computation and gets blocked on that resource
890 -- and therefore cannot do any pushworker to add more threads to
891 -- the producer. In such cases the programmer should use a parallel
892 -- style so that all the producers are scheduled immediately and
893 -- unconditionally. We can also use a separate monitor thread to
894 -- push workers instead of pushing them from the consumer, but then
895 -- we are no longer using pull based concurrency rate adaptation.
897 -- XXX update this in the tutorial.
900 liftIO $ withDBGMVar sv "sendWorkerWait: nothing to do"
901 $ takeMVar (doorBell sv)
903-- | Pull a stream from an SVar.
904{-# NOINLINE fromStreamVar #-}
905fromStreamVar :: MonadAsync m => SVar m a -> Stream m a
906fromStreamVar sv = Stream $ \_ stp sng yld -> do
907 if svarStyle sv == ParallelVar
908 then liftIO $ withDBGMVar sv "fromStreamVar: doorbell"
909 $ takeMVar (doorBell sv)
911 res <- liftIO $ tryTakeMVar (doorBell sv)
912 when (isNothing res) $ sendWorkerWait sv
914 (list, _) <- liftIO $ atomicModifyIORefCAS (outputQueue sv) $ \x -> (([],0), x)
915 void $ liftIO $ tryPutMVar (siren sv) ()
916 -- Reversing the output is important to guarantee that we process the
917 -- outputs in the same order as they were generated by the constituent
919 (runStream $ processEvents (reverse list)) Nothing stp sng yld
923 handleException e tid = do
924 liftIO $ atomicModifyIORefCAS_ (activeWorkers sv) $ \n -> n - 1
926 -- XXX implement kill async exception handling
927 -- liftIO $ readIORef (runningThreads sv) >>= mapM_ killThread
930 {-# INLINE processEvents #-}
931 processEvents [] = Stream $ \_ stp sng yld -> do
932 done <- allThreadsDone sv
934 then (runStream (fromStreamVar sv)) Nothing stp sng yld
937 processEvents (ev : es) = Stream $ \_ stp sng yld -> do
938 let continue = (runStream (processEvents es)) Nothing stp sng yld
939 yield a = yld a (processEvents es)
942 ChildYield a -> yield a
946 let active = activeWorkers sv
947 liftIO $ atomicModifyIORefCAS_ active $ \n -> n - 1
948 modifyThread sv tid >> continue
949 Just ex -> handleException ex tid
951getFifoSVar :: MonadIO m => SVarStyle -> IO (SVar m a)
952getFifoSVar ctype = do
953 outQ <- newIORef ([], 0)
954 outQMv <- newEmptyMVar
955 hooter <- newEmptyMVar
957 wfw <- newIORef False
958 running <- newIORef S.empty
961 SVar { outputQueue = outQ
964 , outputHeap = undefined
965 , heapSiren = undefined
966 , runningThreads = running
967 , workQueue = undefined
968 , runqueue = runqueueFIFO sv q
969 , enqueue = enqueueFIFO sv q
970 , queueEmpty = liftIO $ nullQ q
971 , waitingForWork = wfw
973 , activeWorkers = active
977getLifoSVar :: MonadIO m => SVarStyle -> IO (SVar m a)
978getLifoSVar ctype = do
979 outQ <- newIORef ([], 0)
980 outQMv <- newEmptyMVar
981 hooter <- newEmptyMVar
983 wfw <- newIORef False
984 running <- newIORef S.empty
986 let checkEmpty = null <$> liftIO (readIORef q)
988 SVar { outputQueue = outQ
991 , outputHeap = undefined
992 , heapSiren = undefined
993 , runningThreads = running
994 , workQueue = undefined
995 , runqueue = runqueueLIFO sv q
996 , enqueue = enqueueLIFO sv q
997 , queueEmpty = checkEmpty
998 , waitingForWork = wfw
1000 , activeWorkers = active
1004getParSVar :: SVarStyle -> IO (SVar m a)
1005getParSVar style = do
1006 outQ <- newIORef ([], 0)
1007 outQMv <- newEmptyMVar
1008 hooter <- newEmptyMVar
1009 active <- newIORef 0
1010 wfw <- newIORef False
1011 running <- newIORef S.empty
1013 SVar { outputQueue = outQ
1016 , outputHeap = undefined
1017 , heapSiren = undefined
1018 , runningThreads = running
1019 , workQueue = undefined
1020 , runqueue = undefined
1021 , enqueue = undefined
1022 , queueEmpty = undefined
1023 , waitingForWork = wfw
1025 , activeWorkers = active
1029getAheadSVar :: MonadIO m => SVarStyle -> IO (SVar m a)
1030getAheadSVar style = do
1031 outQ <- newIORef ([], 0)
1032 outH <- newIORef (H.empty, 0)
1033 outQMv <- newEmptyMVar
1034 hooter <- newEmptyMVar
1035 hsiren <- newEmptyMVar
1036 active <- newIORef 0
1037 wfw <- newIORef False
1038 running <- newIORef S.empty
1039 q <- newIORef ([], -1)
1040 let checkEmpty = liftIO $ do
1041 (xs, _) <- readIORef q
1044 SVar { outputQueue = outQ
1048 , heapSiren = hsiren
1049 , runningThreads = running
1051 , runqueue = runqueueAhead sv q
1052 , enqueue = undefined
1053 , queueEmpty = checkEmpty
1054 , waitingForWork = wfw
1056 , activeWorkers = active
1060-- | Create a new empty SVar.
1061newEmptySVar :: MonadAsync m => SVarStyle -> m (SVar m a)
1062newEmptySVar style = do
1065 WAsyncVar -> getFifoSVar style
1066 AsyncVar -> getLifoSVar style
1067 ParallelVar -> getParSVar style
1068 AheadVar -> getAheadSVar style
1070-- | Create a new SVar and enqueue one stream computation on it.
1071newStreamVar1 :: MonadAsync m => SVarStyle -> Stream m a -> m (SVar m a)
1072newStreamVar1 style m = do
1073 sv <- newEmptySVar style
1074 -- Note: We must have all the work on the queue before sending the
1075 -- pushworker, otherwise the pushworker may exit before we even get a
1077 if style == ParallelVar
1078 then pushWorkerPar sv m
1080 liftIO $ (enqueue sv) m
1084-- | Create a new SVar and enqueue one stream computation on it.
1085newStreamVarAhead :: MonadAsync m => Stream m a -> m (SVar m a)
1086newStreamVarAhead m = do
1087 sv <- newEmptySVar AheadVar
1088 -- Note: We must have all the work on the queue before sending the
1089 -- pushworker, otherwise the pushworker may exit before we even get a
1091 liftIO $ enqueueAhead sv (workQueue sv) m
1095-- | Write a stream to an 'SVar' in a non-blocking manner. The stream can then
1096-- be read back from the SVar using 'fromSVar'.
1097toStreamVar :: MonadAsync m => SVar m a -> Stream m a -> m ()
1098toStreamVar sv m = do
1099 liftIO $ (enqueue sv) m
1100 done <- allThreadsDone sv
1101 -- XXX This is safe only when called from the consumer thread or when no
1102 -- consumer is present. There may be a race if we are not running in the
1104 when done $ pushWorker sv
1106------------------------------------------------------------------------------
1107-- Running streams concurrently
1108------------------------------------------------------------------------------
1110-- Concurrency rate control.
1112-- Our objective is to create more threads on demand if the consumer is running
1113-- faster than us. As soon as we encounter a concurrent composition we create a
1114-- push pull pair of threads. We use an SVar for communication between the
1115-- consumer, pulling from the SVar and the producer who is pushing to the SVar.
1116-- The producer creates more threads if the SVar drains and becomes empty, that
1117-- is the consumer is running faster.
1119-- XXX Note 1: This mechanism can be problematic if the initial production
1120-- latency is high, we may end up creating too many threads. So we need some
1121-- way to monitor and use the latency as well. Having a limit on the dispatches
1122-- (programmer controlled) may also help.
1124-- TBD Note 2: We may want to run computations at the lower level of the
1125-- composition tree serially even when they are composed using a parallel
1126-- combinator. We can use 'serial' in place of 'async' and 'wSerial' in
1127-- place of 'wAsync'. If we find that an SVar immediately above a computation
1128-- gets drained empty we can switch to parallelizing the computation. For that
1129-- we can use a state flag to fork the rest of the computation at any point of
1130-- time inside the Monad bind operation if the consumer is running at a faster
1133-- TBD Note 3: the binary operation ('parallel') composition allows us to
1134-- dispatch a chunkSize of only 1. If we have to dispatch in arbitrary
1135-- chunksizes we will need to compose the parallel actions using a data
1136-- constructor (A Free container) instead so that we can divide it in chunks of
1137-- arbitrary size before dispatching. If the stream is composed of
1138-- hierarchically composed grains of different sizes then we can always switch
1139-- to a desired granularity depending on the consumer speed.
1141-- TBD Note 4: for pure work (when we are not in the IO monad) we can divide it
1142-- into just the number of CPUs.
1144-- | Join two computations on the currently running 'SVar' queue for concurrent
1145-- execution. When we are using parallel composition, an SVar is passed around
1146-- as a state variable. We try to schedule a new parallel computation on the
1147-- SVar passed to us. The first time, when no SVar exists, a new SVar is
1148-- created. Subsequently, 'joinStreamVarAsync' may get called when a computation
1149-- already scheduled on the SVar is further evaluated. For example, when (a
1150-- `parallel` b) is evaluated it calls a 'joinStreamVarAsync' to put 'a' and 'b' on
1151-- the current scheduler queue.
1153-- The 'SVarStyle' required by the current composition context is passed as one
1154-- of the parameters. If the scheduling and composition style of the new
1155-- computation being scheduled is different than the style of the current SVar,
1156-- then we create a new SVar and schedule it on that. The newly created SVar
1157-- joins as one of the computations on the current SVar queue.
1159-- Cases when we need to switch to a new SVar:
1161-- * (x `parallel` y) `parallel` (t `parallel` u) -- all of them get scheduled on the same SVar
1162-- * (x `parallel` y) `parallel` (t `async` u) -- @t@ and @u@ get scheduled on a new child SVar
1163-- because of the scheduling policy change.
1164-- * if we 'adapt' a stream of type 'async' to a stream of type
1165-- 'Parallel', we create a new SVar at the transitioning bind.
1166-- * When the stream is switching from disjunctive composition to conjunctive
1167-- composition and vice-versa we create a new SVar to isolate the scheduling
1170{-# INLINE joinStreamVarAsync #-}
1171joinStreamVarAsync :: MonadAsync m
1172 => SVarStyle -> Stream m a -> Stream m a -> Stream m a
1173joinStreamVarAsync style m1 m2 = Stream $ \svr stp sng yld ->
1175 Just sv | svarStyle sv == style ->
1176 liftIO ((enqueue sv) m2) >> (runStream m1) svr stp sng yld
1178 sv <- newStreamVar1 style (concurrently m1 m2)
1179 (runStream (fromStreamVar sv)) Nothing stp sng yld
1181 concurrently ma mb = Stream $ \svr stp sng yld -> do
1182 liftIO $ enqueue (fromJust svr) mb
1183 (runStream ma) svr stp sng yld
1185{-# INLINE joinStreamVarPar #-}
1186joinStreamVarPar :: MonadAsync m
1187 => SVarStyle -> Stream m a -> Stream m a -> Stream m a
1188joinStreamVarPar style m1 m2 = Stream $ \svr stp sng yld ->
1190 Just sv | svarStyle sv == style -> do
1191 pushWorkerPar sv m1 >> (runStream m2) svr stp sng yld
1193 sv <- newEmptySVar style
1196 (runStream (fromStreamVar sv)) Nothing stp sng yld
1199ahead :: MonadAsync m => Stream m a -> Stream m a -> Stream m a
1200ahead m1 m2 = Stream $ \svr stp sng yld -> do
1202 Just sv | svarStyle sv == AheadVar -> do
1203 liftIO $ enqueueAhead sv (workQueue sv) m2
1204 -- Always run the left side on a new SVar to avoid complexity in
1205 -- sequencing results. This means the left side cannot further
1206 -- split into more ahead computations on the same SVar.
1207 (runStream m1) Nothing stp sng yld
1209 sv <- newStreamVarAhead (concurrently m1 m2)
1210 (runStream (fromStreamVar sv)) Nothing stp sng yld
1212 concurrently ma mb = Stream $ \svr stp sng yld -> do
1213 liftIO $ enqueueAhead (fromJust svr) (workQueue (fromJust svr)) mb
1214 (runStream ma) Nothing stp sng yld
1216-- | XXX we can implement it more efficienty by directly implementing instead
1217-- of combining streams using ahead.
1218{-# INLINE consMAhead #-}
1219consMAhead :: MonadAsync m => m a -> Stream m a -> Stream m a
1220consMAhead m r = once m `ahead` r
1222------------------------------------------------------------------------------
1223-- Semigroup and Monoid style compositions for parallel actions
1224------------------------------------------------------------------------------
1227async :: MonadAsync m => Stream m a -> Stream m a -> Stream m a
1228async = joinStreamVarAsync AsyncVar
1230-- | XXX we can implement it more efficienty by directly implementing instead
1231-- of combining streams using async.
1232{-# INLINE consMAsync #-}
1233consMAsync :: MonadAsync m => m a -> Stream m a -> Stream m a
1234consMAsync m r = once m `async` r
1236{-# INLINE wAsync #-}
1237wAsync :: MonadAsync m => Stream m a -> Stream m a -> Stream m a
1238wAsync = joinStreamVarAsync WAsyncVar
1240-- | XXX we can implement it more efficienty by directly implementing instead
1241-- of combining streams using wAsync.
1242{-# INLINE consMWAsync #-}
1243consMWAsync :: MonadAsync m => m a -> Stream m a -> Stream m a
1244consMWAsync m r = once m `wAsync` r
1246{-# INLINE parallel #-}
1247parallel :: MonadAsync m => Stream m a -> Stream m a -> Stream m a
1248parallel = joinStreamVarPar ParallelVar
1250-- | XXX we can implement it more efficienty by directly implementing instead
1251-- of combining streams using parallel.
1252{-# INLINE consMParallel #-}
1253consMParallel :: MonadAsync m => m a -> Stream m a -> Stream m a
1254consMParallel m r = once m `parallel` r
1256-------------------------------------------------------------------------------
1257-- Functor instace is the same for all types
1258-------------------------------------------------------------------------------
1260instance Monad m => Functor (Stream m) where
1261 fmap f m = Stream $ \_ stp sng yld ->
1262 let single = sng . f
1263 yield a r = yld (f a) (fmap f r)
1264 in (runStream m) Nothing stp single yield
1266------------------------------------------------------------------------------
1267-- Alternative & MonadPlus
1268------------------------------------------------------------------------------
1270_alt :: Stream m a -> Stream m a -> Stream m a
1271_alt m1 m2 = Stream $ \_ stp sng yld ->
1272 let stop = runStream m2 Nothing stp sng yld
1273 in runStream m1 Nothing stop sng yld
1275------------------------------------------------------------------------------
1276-- Stream to stream function application
1277------------------------------------------------------------------------------
1279applyWith :: MonadAsync m
1280 => SVarStyle -> (Stream m a -> Stream m b) -> Stream m a -> Stream m b
1281applyWith style f m = Stream $ \svr stp sng yld -> do
1282 sv <- newStreamVar1 style m
1283 runStream (f $ fromStreamVar sv) svr stp sng yld
1285------------------------------------------------------------------------------
1286-- Stream runner function application
1287------------------------------------------------------------------------------
1289runWith :: MonadAsync m
1290 => SVarStyle -> (Stream m a -> m b) -> Stream m a -> m b
1291runWith style f m = do
1292 sv <- newStreamVar1 style m
1293 f $ fromStreamVar sv
1295------------------------------------------------------------------------------
1297------------------------------------------------------------------------------
1299zipWith :: (a -> b -> c) -> Stream m a -> Stream m b -> Stream m c
1300zipWith f m1 m2 = go m1 m2
1302 go mx my = Stream $ \_ stp sng yld -> do
1304 let single2 b = sng (f a b)
1305 yield2 b rb = yld (f a b) (go ra rb)
1306 in (runStream my) Nothing stp single2 yield2
1307 let single1 a = merge a nil
1308 yield1 a ra = merge a ra
1309 (runStream mx) Nothing stp single1 yield1
1311zipAsyncWith :: MonadAsync m
1312 => (a -> b -> c) -> Stream m a -> Stream m b -> Stream m c
1313zipAsyncWith f m1 m2 = Stream $ \_ stp sng yld -> do
1316 (runStream (zipWith f ma mb)) Nothing stp sng yld
1320 mkAsync :: MonadAsync m => Stream m a -> m (Stream m a)
1321 mkAsync m = newStreamVar1 AsyncVar m
1322 >>= return . fromStreamVar
1324-------------------------------------------------------------------------------
1326-------------------------------------------------------------------------------
1328instance 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