1{-# LANGUAGE ConstraintKinds #-}
2{-# LANGUAGE FlexibleContexts #-}
3{-# LANGUAGE FlexibleInstances #-}
4{-# LANGUAGE LambdaCase #-}
5{-# LANGUAGE MultiParamTypeClasses #-}
6{-# LANGUAGE RankNTypes #-}
7{-# LANGUAGE UndecidableInstances #-} -- XXX
10-- Module : Streamly.Core
11-- Copyright : (c) 2017 Harendra Kumar
14-- Maintainer : harendra.kumar@gmail.com
15-- Stability : experimental
33 -- * Concurrent Stream Vars (SVars)
45 -- * Concurrent Streams
51import Control.Applicative (Alternative (..))
52import Control.Concurrent (ThreadId, forkIO,
53 myThreadId, threadDelay)
54import Control.Concurrent.MVar (MVar, newEmptyMVar, tryTakeMVar,
56import Control.Exception (SomeException (..))
57import qualified Control.Exception.Lifted as EL
58import Control.Monad (MonadPlus(..), mzero, when)
59import Control.Monad.Base (MonadBase (..), liftBaseDefault)
60import Control.Monad.Catch (MonadThrow, throwM)
61import Control.Monad.Error.Class (MonadError(..))
62import Control.Monad.IO.Class (MonadIO(..))
63import Control.Monad.Reader.Class (MonadReader(..))
64import Control.Monad.State.Class (MonadState(..))
65import Control.Monad.Trans.Class (MonadTrans (lift))
66import Control.Monad.Trans.Control (MonadBaseControl, liftBaseWith)
67import Data.Atomics (atomicModifyIORefCAS,
68 atomicModifyIORefCAS_)
69import Data.Concurrent.Queue.MichaelScott (LinkedQueue, newQ, pushL,
71import Data.Functor (void)
72import Data.IORef (IORef, modifyIORef, newIORef,
74import Data.Maybe (isNothing)
75import Data.Semigroup (Semigroup(..))
77import qualified Data.Set as S
79------------------------------------------------------------------------------
80-- Parent child thread communication type
81------------------------------------------------------------------------------
83-- | Events that a child thread may send to a parent thread.
86 | ChildStop ThreadId (Maybe SomeException)
88------------------------------------------------------------------------------
89-- State threaded around the monad for thread management
90------------------------------------------------------------------------------
92-- | Conjunction is used for monadic/product style composition. Disjunction is
93-- used for fold/sum style composition. We need to distiguish the two types of
94-- SVars so that the scheduling of the two is independent.
95data SVarTag = Conjunction | Disjunction deriving Eq
97-- | For fairly interleaved parallel composition the sched policy is FIFO
98-- whereas for left biased parallel composition it is LIFO.
99data SVarSched = LIFO | FIFO deriving Eq
101-- | Identify the type of the SVar. Two computations using the same style can
102-- be scheduled on the same SVar.
103data SVarStyle = SVarStyle SVarTag SVarSched deriving Eq
105-- | An SVar or a Stream Var is a conduit to the output from multiple streams
106-- running concurrently and asynchronously. An SVar can be thought of as an
107-- asynchronous IO handle. We can write any number of streams to an SVar in a
108-- non-blocking manner and then read them back at any time at any pace. The
109-- SVar would run the streams asynchronously and accumulate results. An SVar
110-- may not really execute the stream completely and accumulate all the results.
111-- However, it ensures that the reader can read the results at whatever paces
112-- it wants to read. The SVar monitors and adapts to the consumer's pace.
114-- An SVar is a mini scheduler, it has an associated runqueue that holds the
115-- stream tasks to be picked and run by a pool of worker threads. It has an
116-- associated output queue where the output stream elements are placed by the
117-- worker threads. A doorBell is used by the worker threads to intimate the
118-- consumer thread about availability of new results in the output queue. More
119-- workers are added to the SVar by 'fromStreamVar' on demand if the output
120-- produced is not keeping pace with the consumer. On bounded SVars, workers
121-- block on the output queue to provide throttling of the producer when the
122-- consumer is not pulling fast enough. The number of workers may even get
123-- reduced depending on the consuming pace.
125-- New work is enqueued either at the time of creation of the SVar or as a
126-- result of executing the parallel combinators i.e. '<|' and '<|>' when the
127-- already enqueued computations get evaluated. See 'joinStreamVar2'.
130 SVar { outputQueue :: IORef [ChildEvent a]
131 , doorBell :: MVar Bool -- wakeup mechanism for outQ
132 , enqueue :: Stream m a -> IO ()
134 , runningThreads :: IORef (Set ThreadId)
135 , queueEmpty :: m Bool
136 , svarStyle :: SVarStyle
139------------------------------------------------------------------------------
141------------------------------------------------------------------------------
143-- TBD use a functor instead of the bare type a?
144-- XXX remove the Maybe, use "empty" as the base case
146-- | Represents a monadic stream of values of type 'a' constructed using
147-- actions in monad 'm'. Streams can be composed sequentially or in parallel;
148-- in product style compositions (monadic bind multiplies streams in a ListT
149-- fashion) or in sum style compositions like 'Semigroup', 'Monoid',
150-- 'Alternative' or variants of these.
153 runStream :: forall r.
154 Maybe (SVar m a) -- local state
156 -> (a -> Maybe (Stream m a) -> m r) -- yield
160-- | A monad that can perform asynchronous/concurrent IO operations. Streams
161-- that can be composed concurrently require the underlying monad to be
163type MonadAsync m = (MonadIO m, MonadBaseControl IO m, MonadThrow m)
165scons :: a -> Maybe (Stream m a) -> Stream m a
166scons a r = Stream $ \_ _ yld -> yld a r
169snil = Stream $ \_ stp _ -> stp
171------------------------------------------------------------------------------
173------------------------------------------------------------------------------
175-- | '<>' concatenates two streams sequentially i.e. the first stream is
176-- exhausted completely before yielding any element from the second stream.
177instance Semigroup (Stream m a) where
180 go (Stream m) = Stream $ \_ stp yld ->
181 let stop = (runStream m2) Nothing stp yld
182 yield a Nothing = yld a (Just m2)
183 yield a (Just r) = yld a (Just (go r))
184 in m Nothing stop yield
186------------------------------------------------------------------------------
188------------------------------------------------------------------------------
190instance Monoid (Stream m a) where
191 mempty = Stream $ \_ stp _ -> stp
194------------------------------------------------------------------------------
196------------------------------------------------------------------------------
199interleave :: Stream m a -> Stream m a -> Stream m a
200interleave m1 m2 = Stream $ \_ stp yld -> do
201 let stop = (runStream m2) Nothing stp yld
202 yield a Nothing = yld a (Just m2)
203 yield a (Just r) = yld a (Just (interleave m2 r))
204 (runStream m1) Nothing stop yield
206------------------------------------------------------------------------------
207-- Spawning threads and collecting result in streamed fashion
208------------------------------------------------------------------------------
211doFork :: MonadBaseControl IO m
213 -> (SomeException -> m ())
215doFork action exHandler =
216 EL.mask $ \restore ->
217 liftBaseWith $ \runInIO -> forkIO $ do
218 -- XXX test the exception handling
219 _ <- runInIO $ EL.catch (restore action) exHandler
220 -- XXX restore state here?
223-- XXX exception safety of all atomic/MVar operations
226send :: MonadIO m => SVar m a -> ChildEvent a -> m ()
227send sv msg = liftIO $ do
228 atomicModifyIORefCAS_ (outputQueue sv) $ \es -> msg : es
229 -- XXX need a memory barrier? The wake up must happen only after the
230 -- store has finished otherwise we can have lost wakeup problems.
231 void $ tryPutMVar (doorBell sv) True
233{-# INLINE sendStop #-}
234sendStop :: MonadIO m => SVar m a -> m ()
235sendStop sv = liftIO myThreadId >>= \tid -> send sv (ChildStop tid Nothing)
237-- Note: Left associated compositions can grow this queue to a large size
238{-# INLINE enqueueLIFO #-}
239enqueueLIFO :: IORef [Stream m a] -> Stream m a -> IO ()
240enqueueLIFO q m = atomicModifyIORefCAS_ q $ \ ms -> m : ms
242runqueueLIFO :: MonadIO m => SVar m a -> IORef [Stream m a] -> m ()
243runqueueLIFO sv q = run
250 Nothing -> sendStop sv
251 Just m -> (runStream m) (Just sv) run yield
253 sendit a = send sv (ChildYield a)
254 yield a Nothing = sendit a >> run
255 yield a (Just r) = sendit a >> (runStream r) (Just sv) run yield
257 dequeue = liftIO $ atomicModifyIORefCAS q $ \case
259 x : xs -> (xs, Just x)
261{-# INLINE enqueueFIFO #-}
262enqueueFIFO :: LinkedQueue (Stream m a) -> Stream m a -> IO ()
265runqueueFIFO :: MonadIO m => SVar m a -> LinkedQueue (Stream m a) -> m ()
266runqueueFIFO sv q = run
273 Nothing -> sendStop sv
274 Just m -> (runStream m) (Just sv) run yield
276 dequeue = liftIO $ tryPopR q
277 sendit a = send sv (ChildYield a)
278 yield a Nothing = sendit a >> run
279 yield a (Just r) = sendit a >> liftIO (enqueueFIFO q r) >> run
281-- Thread tracking is needed for two reasons:
283-- 1) Killing threads on exceptions. Threads may not be allowed to go away by
284-- themselves because they may run for significant times before going away or
285-- worse they may be stuck in IO and never go away.
287-- 2) To know when all threads are done.
289{-# NOINLINE addThread #-}
290addThread :: MonadIO m => SVar m a -> ThreadId -> m ()
292 liftIO $ modifyIORef (runningThreads sv) (S.insert tid)
294{-# INLINE delThread #-}
295delThread :: MonadIO m => SVar m a -> ThreadId -> m ()
297 liftIO $ modifyIORef (runningThreads sv) $ (\s -> S.delete tid s)
299{-# INLINE allThreadsDone #-}
300allThreadsDone :: MonadIO m => SVar m a -> m Bool
301allThreadsDone sv = liftIO $ S.null <$> readIORef (runningThreads sv)
303{-# NOINLINE handleChildException #-}
304handleChildException :: MonadIO m => SVar m a -> SomeException -> m ()
305handleChildException sv e = do
306 tid <- liftIO myThreadId
307 send sv (ChildStop tid (Just e))
309{-# NOINLINE pushWorker #-}
310pushWorker :: MonadAsync m => SVar m a -> m ()
312 doFork (runqueue sv) (handleChildException sv) >>= addThread sv
314-- XXX When the queue is LIFO we can put a limit on the number of dispatches.
315-- Also, if a worker blocks on the output queue we can decide if we want to
316-- block or make it go away entirely, depending on the number of workers and
317-- the type of the queue.
318{-# INLINE sendWorkerWait #-}
319sendWorkerWait :: MonadAsync m => SVar m a -> m ()
320sendWorkerWait sv = do
322 SVarStyle _ LIFO -> liftIO $ threadDelay 200
323 SVarStyle _ FIFO -> liftIO $ threadDelay 0
325 output <- liftIO $ readIORef (outputQueue sv)
326 when (null output) $ do
327 done <- queueEmpty sv
329 then pushWorker sv >> sendWorkerWait sv
330 else void (liftIO $ takeMVar (doorBell sv))
332-- | Pull a stream from an SVar.
333{-# NOINLINE fromStreamVar #-}
334fromStreamVar :: MonadAsync m => SVar m a -> Stream m a
335fromStreamVar sv = Stream $ \_ stp yld -> do
336 -- XXX if reading the IORef is costly we can use a flag in the SVar to
337 -- indicate we are done.
338 done <- allThreadsDone sv
342 res <- liftIO $ tryTakeMVar (doorBell sv)
343 when (isNothing res) $ sendWorkerWait sv
344 list <- liftIO $ atomicModifyIORefCAS (outputQueue sv) $ \x -> ([], x)
345 -- To avoid lock overhead we read all events at once instead of reading
346 -- one at a time. We just reverse the list to process the events in the
347 -- order they arrived. Maybe we can use a queue instead?
348 (runStream $ processEvents (reverse list)) Nothing stp yld
352 handleException e tid = do
354 -- XXX implement kill async exception handling
355 -- liftIO $ readIORef (runningThreads sv) >>= mapM_ killThread
358 {-# INLINE processEvents #-}
359 processEvents [] = Stream $ \_ stp yld -> do
360 done <- allThreadsDone sv
362 then (runStream (fromStreamVar sv)) Nothing stp yld
365 processEvents (ev : es) = Stream $ \_ stp yld -> do
366 let continue = (runStream (processEvents es)) Nothing stp yld
367 yield a = yld a (Just (processEvents es))
370 ChildYield a -> yield a
373 Nothing -> delThread sv tid >> continue
374 Just ex -> handleException ex tid
376getFifoSVar :: MonadIO m => SVarStyle -> IO (SVar m a)
377getFifoSVar ctype = do
379 outQMv <- newEmptyMVar
380 running <- newIORef S.empty
383 SVar { outputQueue = outQ
385 , runningThreads = running
386 , runqueue = runqueueFIFO sv q
388 , queueEmpty = liftIO $ nullQ q
393getLifoSVar :: MonadIO m => SVarStyle -> IO (SVar m a)
394getLifoSVar ctype = do
396 outQMv <- newEmptyMVar
397 running <- newIORef S.empty
399 let checkEmpty = null <$> liftIO (readIORef q)
401 SVar { outputQueue = outQ
403 , runningThreads = running
404 , runqueue = runqueueLIFO sv q
405 , enqueue = enqueueLIFO q
406 , queueEmpty = checkEmpty
411-- | Create a new empty SVar.
412newEmptySVar :: MonadAsync m => SVarStyle -> m (SVar m a)
413newEmptySVar style = do
416 SVarStyle _ FIFO -> getFifoSVar style
417 SVarStyle _ LIFO -> getLifoSVar style
419-- | Create a new SVar and enqueue one stream computation on it.
420newStreamVar1 :: MonadAsync m => SVarStyle -> Stream m a -> m (SVar m a)
421newStreamVar1 style m = do
422 sv <- newEmptySVar style
423 -- Note: We must have all the work on the queue before sending the
424 -- pushworker, otherwise the pushworker may exit before we even get a
426 liftIO $ (enqueue sv) m
430-- | Create a new SVar and enqueue two stream computations on it.
431newStreamVar2 :: MonadAsync m
432 => SVarStyle -> Stream m a -> Stream m a -> m (SVar m a)
433newStreamVar2 style m1 m2 = do
434 -- Note: We must have all the work on the queue before sending the
435 -- pushworker, otherwise the pushworker may exit before we even get a
439 SVarStyle _ FIFO -> do
440 c <- getFifoSVar style
441 (enqueue c) m1 >> (enqueue c) m2
443 SVarStyle _ LIFO -> do
444 c <- getLifoSVar style
445 (enqueue c) m2 >> (enqueue c) m1
450-- | Write a stream to an 'SVar' in a non-blocking manner. The stream can then
451-- be read back from the SVar using 'fromSVar'.
452toStreamVar :: MonadAsync m => SVar m a -> Stream m a -> m ()
454 liftIO $ (enqueue sv) m
455 done <- allThreadsDone sv
456 -- XXX there may be a race here unless we are running in the consumer
457 -- thread. This is safe only when called from the consumer thread or when
458 -- no consumer is present.
459 when done $ pushWorker sv
461------------------------------------------------------------------------------
462-- Running streams concurrently
463------------------------------------------------------------------------------
465-- Concurrency rate control. Our objective is to create more threads on demand
466-- if the consumer is running faster than us. As soon as we encounter an
467-- Alternative composition we create a push pull pair of threads. We use a
468-- channel for communication between the consumer pulling from the channel and
469-- the producer who pushing to the channel. The producer creates more threads
470-- if no output is seen on the channel, that is the consumer is running faster.
471-- However this mechanism can be problematic if the initial production latency
472-- is high, we may end up creating too many threads. So we need some way to
473-- monitor and use the latency as well.
475-- TBD We may run computations at the lower level of the composition tree
476-- serially even if they are composed using a parallel combinator. We can use
477-- <> in place of <| and <=> in place of <|>. If we find that a parallel
478-- channel immediately above a computation becomes empty we can switch to
479-- parallelizing the computation. For that we can use a state flag to fork the
480-- rest of the computation at any point of time inside the Monad bind operation
481-- if the consumer is running at a faster speed.
483-- TBD the alternative composition allows us to dispatch a chunkSize of only 1.
484-- If we have to dispatch in arbitrary chunksizes we will need to compose the
485-- parallel actions using a data constructor (Free Alternative) instead so that
486-- we can divide it in chunks of arbitrary size before dispatch. If the stream
487-- is composed of hierarchically composed grains of different sizes then we can
488-- always switch to a desired granularity depending on the consumer speed.
490-- TBD for pure work (when we are not in the IO monad) we can divide it into
491-- just the number of CPUs.
493{-# NOINLINE withNewSVar2 #-}
494withNewSVar2 :: MonadAsync m
495 => SVarStyle -> Stream m a -> Stream m a -> Stream m a
496withNewSVar2 style m1 m2 = Stream $ \_ stp yld -> do
497 sv <- newStreamVar2 style m1 m2
498 (runStream (fromStreamVar sv)) Nothing stp yld
500-- | Join two computations on the currently running 'SVar' queue for concurrent
501-- execution. The 'SVarStyle' required by the current composition context is
502-- passed as one of the parameters. If the style does not match with the style
503-- of the current 'SVar' we create a new 'SVar' and schedule the computations
504-- on that. The newly created SVar joins as one of the computations on the
505-- current SVar queue.
507-- When we are using parallel composition, an SVar is passed around as a state
508-- variable. We try to schedule a new parallel computation on the SVar passed
509-- to us. The first time, when no SVar exists, a new SVar is created.
510-- Subsequently, 'joinStreamVar2' may get called when a computation already
511-- scheduled on the SVar is further evaluated. For example, when (a \<|> b) is
512-- evaluated it calls a 'joinStreamVar2' to put 'a' and 'b' on the current scheduler
513-- queue. However, if the scheduling and composition style of the new
514-- computation being scheduled is different than the style of the current SVar,
515-- then we create a new SVar and schedule it on that.
519-- * (x \<|> y) \<|> (t \<|> u) -- all of them get scheduled on the same SVar
520-- * (x \<|> y) \<|> (t \<| u) -- @t@ and @u@ get scheduled on a new child SVar
521-- because of the scheduling policy change.
522-- * if we 'adapt' a stream of type 'AsyncT' to a stream of type
523-- 'ParallelT', we create a new SVar at the transitioning bind.
524-- * When the stream is switching from disjunctive composition to conjunctive
525-- composition and vice-versa we create a new SVar to isolate the scheduling
528{-# INLINE joinStreamVar2 #-}
529joinStreamVar2 :: MonadAsync m
530 => SVarStyle -> Stream m a -> Stream m a -> Stream m a
531joinStreamVar2 style m1 m2 = Stream $ \st stp yld ->
533 Just sv | svarStyle sv == style ->
534 liftIO ((enqueue sv) m2) >> (runStream m1) st stp yld
535 _ -> (runStream (withNewSVar2 style m1 m2)) Nothing stp yld
537------------------------------------------------------------------------------
538-- Semigroup and Monoid style compositions for parallel actions
539------------------------------------------------------------------------------
543parAhead :: Stream m a -> Stream m a -> Stream m a
546-- | Sequential composition similar to '<>' except that it can execute the
547-- action on the right in parallel ahead of time. Returns the results in
548-- sequential order like '<>' from left to right.
549(<>|) :: Stream m a -> Stream m a -> Stream m a
553-- | Same as '<|>'. Since this schedules all the composed streams fairly you
554-- cannot fold infinite number of streams using this operation.
556parAlt :: MonadAsync m => Stream m a -> Stream m a -> Stream m a
557parAlt = joinStreamVar2 (SVarStyle Disjunction FIFO)
559-- | Same as '<|'. Since this schedules the left side computation first you can
560-- right fold an infinite container using this operator. However a left fold
561-- will not work well as it first unpeels the whole structure before scheduling
562-- a computation requiring an amount of memory proportional to the size of the
564{-# INLINE parLeft #-}
565parLeft :: MonadAsync m => Stream m a -> Stream m a -> Stream m a
566parLeft = joinStreamVar2 (SVarStyle Disjunction LIFO)
568-------------------------------------------------------------------------------
569-- Instances (only used for deriving newtype instances)
570-------------------------------------------------------------------------------
572-- Stream type is not exposed, these instances are only for deriving instances
573-- for the newtype wrappers based on Stream.
575-- Dummy Instances, defined to enable the definition of other instances that
576-- require a Monad constraint. Must be defined by the newtypes.
578instance Monad m => Functor (Stream m) where
581instance Monad m => Applicative (Stream m) where
585instance Monad m => Monad (Stream m) where
589------------------------------------------------------------------------------
590-- Alternative & MonadPlus
591------------------------------------------------------------------------------
593-- | `empty` represents an action that takes non-zero time to complete. Since
594-- all actions take non-zero time, an `Alternative` composition ('<|>') is a
595-- monoidal composition executing all actions in parallel, it is similar to
596-- '<>' except that it runs all the actions in parallel and interleaves their
598instance MonadAsync m => Alternative (Stream m) where
602instance MonadAsync m => MonadPlus (Stream m) where
606-------------------------------------------------------------------------------
608-------------------------------------------------------------------------------
610instance MonadTrans Stream where
611 lift mx = Stream $ \_ _ yld -> mx >>= (\a -> (yld a Nothing))
613instance (MonadBase b m, Monad m) => MonadBase b (Stream m) where
614 liftBase = liftBaseDefault
616------------------------------------------------------------------------------
617-- Standard transformer instances
618------------------------------------------------------------------------------
620instance MonadIO m => MonadIO (Stream m) where
621 liftIO = lift . liftIO
623instance MonadThrow m => MonadThrow (Stream m) where
624 throwM = lift . throwM
626-- XXX handle and test cross thread state transfer
627instance MonadError e m => MonadError e (Stream m) where
628 throwError = lift . throwError
629 catchError m h = Stream $ \st stp yld ->
630 let handle r = r `catchError` \e -> (runStream (h e)) st stp yld
631 yield a Nothing = yld a Nothing
632 yield a (Just r) = yld a (Just (catchError r h))
633 in handle $ (runStream m) st stp yield
635instance MonadReader r m => MonadReader r (Stream m) where
637 local f m = Stream $ \st stp yld ->
638 let yield a Nothing = local f $ yld a Nothing
639 yield a (Just r) = local f $ yld a (Just (local f r))
640 in (runStream m) st (local f stp) yield
642instance MonadState s m => MonadState s (Stream m) where
645 state k = lift (state k)
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