1{-# LANGUAGE ConstraintKinds #-}
2{-# LANGUAGE FlexibleContexts #-}
3{-# LANGUAGE FlexibleInstances #-}
4{-# LANGUAGE MultiParamTypeClasses #-}
5{-# LANGUAGE RankNTypes #-}
6{-# LANGUAGE UndecidableInstances #-} -- XXX
9-- Module : Streamly.Core
10-- Copyright : (c) 2017 Harendra Kumar
13-- Maintainer : harendra.kumar@gmail.com
14-- Stability : experimental
32 -- * Concurrent Stream Vars (SVars)
44 -- * Concurrent Streams
50import Control.Applicative (Alternative (..))
51import Control.Concurrent (ThreadId, forkIO,
52 myThreadId, threadDelay)
53import Control.Concurrent.MVar (MVar, newEmptyMVar, tryTakeMVar,
55import Control.Exception (SomeException (..))
56import qualified Control.Exception.Lifted as EL
57import Control.Monad (MonadPlus(..), mzero, when)
58import Control.Monad.Base (MonadBase (..), liftBaseDefault)
59import Control.Monad.Catch (MonadThrow, throwM)
60import Control.Monad.Error.Class (MonadError(..))
61import Control.Monad.IO.Class (MonadIO(..))
62import Control.Monad.Reader.Class (MonadReader(..))
63import Control.Monad.State.Class (MonadState(..))
64import Control.Monad.Trans.Class (MonadTrans (lift))
65import Control.Monad.Trans.Control (MonadBaseControl, liftBaseWith)
66import Data.Atomics (atomicModifyIORefCAS,
67 atomicModifyIORefCAS_)
68import Data.Concurrent.Queue.MichaelScott (LinkedQueue, newQ, pushL,
70import Data.Functor (void)
71import Data.IORef (IORef, modifyIORef, newIORef,
73import Data.Maybe (isNothing)
74import Data.Semigroup (Semigroup(..))
76import qualified Data.Set as S
78------------------------------------------------------------------------------
79-- Parent child thread communication type
80------------------------------------------------------------------------------
82-- | Events that a child thread may send to a parent thread.
85 | ChildStop ThreadId (Maybe SomeException)
87------------------------------------------------------------------------------
88-- State threaded around the monad for thread management
89------------------------------------------------------------------------------
91-- | Conjunction is used for monadic/product style composition. Disjunction is
92-- used for fold/sum style composition. We need to distiguish the two types of
93-- SVars so that the scheduling of the two is independent.
94data SVarTag = Conjunction | Disjunction deriving Eq
96-- | For fairly interleaved parallel composition the sched policy is FIFO
97-- whereas for left biased parallel composition it is LIFO.
98data SVarSched = LIFO | FIFO deriving Eq
100-- | Identify the type of the SVar. Two computations using the same style can
101-- be scheduled on the same SVar.
102data SVarStyle = SVarStyle SVarTag SVarSched deriving Eq
104-- | An SVar or a Stream Var is a conduit to the output from multiple streams
105-- running concurrently and asynchronously. An SVar can be thought of as an
106-- asynchronous IO handle. We can write any number of streams to an SVar in a
107-- non-blocking manner and then read them back at any time at any pace. The
108-- SVar would run the streams asynchronously and accumulate results. An SVar
109-- may not really execute the stream completely and accumulate all the results.
110-- However, it ensures that the reader can read the results at whatever paces
111-- it wants to read. The SVar monitors and adapts to the consumer's pace.
113-- An SVar is a mini scheduler, it has an associated runqueue that holds the
114-- stream tasks to be picked and run by a pool of worker threads. It has an
115-- associated output queue where the output stream elements are placed by the
116-- worker threads. A doorBell is used by the worker threads to intimate the
117-- consumer thread about availability of new results in the output queue. More
118-- workers are added to the SVar by 'fromStreamVar' on demand if the output
119-- produced is not keeping pace with the consumer. On bounded SVars, workers
120-- block on the output queue to provide throttling of the producer when the
121-- consumer is not pulling fast enough. The number of workers may even get
122-- reduced depending on the consuming pace.
124-- New work is enqueued either at the time of creation of the SVar or as a
125-- result of executing the parallel combinators i.e. '<|' and '<|>' when the
126-- already enqueued computations get evaluated. See 'joinStreamVar2'.
129 SVar { outputQueue :: IORef [ChildEvent a]
130 , doorBell :: MVar Bool -- wakeup mechanism for outQ
131 , enqueue :: Stream m a -> IO ()
133 , runningThreads :: IORef (Set ThreadId)
134 , queueEmpty :: m Bool
135 , svarStyle :: SVarStyle
138------------------------------------------------------------------------------
140------------------------------------------------------------------------------
142-- TBD use a functor instead of the bare type a?
143-- XXX remove the Maybe, use "empty" as the base case
145-- | Represents a monadic stream of values of type 'a' constructed using
146-- actions in monad 'm'. Streams can be composed sequentially or in parallel;
147-- in product style compositions (monadic bind multiplies streams in a ListT
148-- fashion) or in sum style compositions like 'Semigroup', 'Monoid',
149-- 'Alternative' or variants of these.
152 runStream :: forall r.
153 Maybe (SVar m a) -- local state
155 -> (a -> Maybe (Stream m a) -> m r) -- yield
159-- | A monad that can perform asynchronous/concurrent IO operations. Streams
160-- that can be composed concurrently require the underlying monad to be
162type MonadAsync m = (MonadIO m, MonadBaseControl IO m, MonadThrow m)
164scons :: a -> Maybe (Stream m a) -> Stream m a
165scons a r = Stream $ \_ _ yld -> yld a r
168snil = Stream $ \_ stp _ -> stp
170------------------------------------------------------------------------------
172------------------------------------------------------------------------------
174-- | '<>' concatenates two streams sequentially i.e. the first stream is
175-- exhausted completely before yielding any element from the second stream.
176instance Semigroup (Stream m a) where
179 go (Stream m) = Stream $ \_ stp yld ->
180 let stop = (runStream m2) Nothing stp yld
181 yield a Nothing = yld a (Just m2)
182 yield a (Just r) = yld a (Just (go r))
183 in m Nothing stop yield
185------------------------------------------------------------------------------
187------------------------------------------------------------------------------
189instance Monoid (Stream m a) where
190 mempty = Stream $ \_ stp _ -> stp
193------------------------------------------------------------------------------
195------------------------------------------------------------------------------
198interleave :: Stream m a -> Stream m a -> Stream m a
199interleave m1 m2 = Stream $ \_ stp yld -> do
200 let stop = (runStream m2) Nothing stp yld
201 yield a Nothing = yld a (Just m2)
202 yield a (Just r) = yld a (Just (interleave m2 r))
203 (runStream m1) Nothing stop yield
205------------------------------------------------------------------------------
206-- Spawning threads and collecting result in streamed fashion
207------------------------------------------------------------------------------
210doFork :: MonadBaseControl IO m
212 -> (SomeException -> m ())
214doFork action exHandler =
215 EL.mask $ \restore ->
216 liftBaseWith $ \runInIO -> forkIO $ do
217 -- XXX test the exception handling
218 _ <- runInIO $ EL.catch (restore action) exHandler
219 -- XXX restore state here?
222-- XXX exception safety of all atomic/MVar operations
225send :: MonadIO m => SVar m a -> ChildEvent a -> m ()
226send sv msg = liftIO $ do
227 atomicModifyIORefCAS_ (outputQueue sv) $ \es -> msg : es
228 -- XXX need a memory barrier? The wake up must happen only after the
229 -- store has finished otherwise we can have lost wakeup problems.
230 void $ tryPutMVar (doorBell sv) True
232{-# INLINE sendStop #-}
233sendStop :: MonadIO m => SVar m a -> m ()
234sendStop sv = liftIO myThreadId >>= \tid -> send sv (ChildStop tid Nothing)
236-- Note: Left associated compositions can grow this queue to a large size
237{-# INLINE enqueueLIFO #-}
238enqueueLIFO :: IORef [Stream m a] -> Stream m a -> IO ()
239enqueueLIFO q m = atomicModifyIORefCAS_ q $ \ ms -> m : ms
241runqueueLIFO :: MonadIO m => SVar m a -> IORef [Stream m a] -> m ()
242runqueueLIFO sv q = run
249 Nothing -> sendStop sv
250 Just m -> (runStream m) (Just sv) run yield
252 sendit a = send sv (ChildYield a)
253 yield a Nothing = sendit a >> run
254 yield a (Just r) = sendit a >> (runStream r) (Just sv) run yield
256 dequeue = liftIO $ atomicModifyIORefCAS q $ \ ms ->
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 -> S.insert tid s)
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 $ do
302 readIORef (runningThreads sv) >>= return . S.null
304{-# NOINLINE handleChildException #-}
305handleChildException :: MonadIO m => SVar m a -> SomeException -> m ()
306handleChildException sv e = do
307 tid <- liftIO myThreadId
308 send sv (ChildStop tid (Just e))
310{-# NOINLINE pushWorker #-}
311pushWorker :: MonadAsync m => SVar m a -> m ()
313 doFork (runqueue sv) (handleChildException sv) >>= addThread sv
315-- XXX When the queue is LIFO we can put a limit on the number of dispatches.
316-- Also, if a worker blocks on the output queue we can decide if we want to
317-- block or make it go away entirely, depending on the number of workers and
318-- the type of the queue.
319{-# INLINE sendWorkerWait #-}
320sendWorkerWait :: MonadAsync m => SVar m a -> m ()
321sendWorkerWait sv = do
323 SVarStyle _ LIFO -> liftIO $ threadDelay 200
324 SVarStyle _ FIFO -> liftIO $ threadDelay 0
326 output <- liftIO $ readIORef (outputQueue sv)
327 when (null output) $ do
328 done <- queueEmpty sv
330 then (pushWorker sv) >> sendWorkerWait sv
331 else void (liftIO $ takeMVar (doorBell sv))
333-- | Pull a stream from an SVar.
334{-# NOINLINE fromStreamVar #-}
335fromStreamVar :: MonadAsync m => SVar m a -> Stream m a
336fromStreamVar sv = Stream $ \_ stp yld -> do
337 -- XXX if reading the IORef is costly we can use a flag in the SVar to
338 -- indicate we are done.
339 done <- allThreadsDone sv
343 res <- liftIO $ tryTakeMVar (doorBell sv)
344 when (isNothing res) $ sendWorkerWait sv
345 list <- liftIO $ atomicModifyIORefCAS (outputQueue sv) $ \x -> ([], x)
346 -- To avoid lock overhead we read all events at once instead of reading
347 -- one at a time. We just reverse the list to process the events in the
348 -- order they arrived. Maybe we can use a queue instead?
349 (runStream $ processEvents (reverse list)) Nothing stp yld
353 handleException e tid = do
355 -- XXX implement kill async exception handling
356 -- liftIO $ readIORef (runningThreads sv) >>= mapM_ killThread
359 {-# INLINE processEvents #-}
360 processEvents [] = Stream $ \_ stp yld -> do
361 done <- allThreadsDone sv
363 then (runStream (fromStreamVar sv)) Nothing stp yld
366 processEvents (ev : es) = Stream $ \_ stp yld -> do
367 let continue = (runStream (processEvents es)) Nothing stp yld
368 yield a = yld a (Just (processEvents es))
371 ChildYield a -> yield a
374 Nothing -> delThread sv tid >> continue
375 Just ex -> handleException ex tid
377getFifoSVar :: MonadIO m => SVarStyle -> IO (SVar m a)
378getFifoSVar ctype = do
380 outQMv <- newEmptyMVar
381 running <- newIORef S.empty
384 SVar { outputQueue = outQ
386 , runningThreads = running
387 , runqueue = runqueueFIFO sv q
389 , queueEmpty = liftIO $ nullQ q
394getLifoSVar :: MonadIO m => SVarStyle -> IO (SVar m a)
395getLifoSVar ctype = do
397 outQMv <- newEmptyMVar
398 running <- newIORef S.empty
400 let checkEmpty = liftIO (readIORef q) >>= return . null
402 SVar { outputQueue = outQ
404 , runningThreads = running
405 , runqueue = runqueueLIFO sv q
406 , enqueue = enqueueLIFO q
407 , queueEmpty = checkEmpty
412-- | Create a new empty SVar.
413newEmptySVar :: MonadAsync m => SVarStyle -> m (SVar m a)
414newEmptySVar style = do
417 SVarStyle _ FIFO -> do
418 c <- getFifoSVar style
420 SVarStyle _ LIFO -> do
421 c <- getLifoSVar style
425-- | Create a new SVar and enqueue one stream computation on it.
426newStreamVar1 :: MonadAsync m => SVarStyle -> Stream m a -> m (SVar m a)
427newStreamVar1 style m = do
428 sv <- newEmptySVar style
429 -- Note: We must have all the work on the queue before sending the
430 -- pushworker, otherwise the pushworker may exit before we even get a
432 liftIO $ (enqueue sv) m
436-- | Create a new SVar and enqueue two stream computations on it.
437newStreamVar2 :: MonadAsync m
438 => SVarStyle -> Stream m a -> Stream m a -> m (SVar m a)
439newStreamVar2 style m1 m2 = do
440 -- Note: We must have all the work on the queue before sending the
441 -- pushworker, otherwise the pushworker may exit before we even get a
445 SVarStyle _ FIFO -> do
446 c <- getFifoSVar style
447 (enqueue c) m1 >> (enqueue c) m2
449 SVarStyle _ LIFO -> do
450 c <- getLifoSVar style
451 (enqueue c) m2 >> (enqueue c) m1
456-- | Write a stream to an 'SVar' in a non-blocking manner. The stream can then
457-- be read back from the SVar using 'fromSVar'.
458toStreamVar :: MonadAsync m => SVar m a -> Stream m a -> m ()
460 liftIO $ (enqueue sv) m
461 done <- allThreadsDone sv
462 -- XXX there may be a race here unless we are running in the consumer
463 -- thread. This is safe only when called from the consumer thread or when
464 -- no consumer is present.
465 when done $ pushWorker sv
467------------------------------------------------------------------------------
468-- Running streams concurrently
469------------------------------------------------------------------------------
471-- Concurrency rate control. Our objective is to create more threads on demand
472-- if the consumer is running faster than us. As soon as we encounter an
473-- Alternative composition we create a push pull pair of threads. We use a
474-- channel for communication between the consumer pulling from the channel and
475-- the producer who pushing to the channel. The producer creates more threads
476-- if no output is seen on the channel, that is the consumer is running faster.
477-- However this mechanism can be problematic if the initial production latency
478-- is high, we may end up creating too many threads. So we need some way to
479-- monitor and use the latency as well.
481-- TBD We may run computations at the lower level of the composition tree
482-- serially even if they are composed using a parallel combinator. We can use
483-- <> in place of <| and <=> in place of <|>. If we find that a parallel
484-- channel immediately above a computation becomes empty we can switch to
485-- parallelizing the computation. For that we can use a state flag to fork the
486-- rest of the computation at any point of time inside the Monad bind operation
487-- if the consumer is running at a faster speed.
489-- TBD the alternative composition allows us to dispatch a chunkSize of only 1.
490-- If we have to dispatch in arbitrary chunksizes we will need to compose the
491-- parallel actions using a data constructor (Free Alternative) instead so that
492-- we can divide it in chunks of arbitrary size before dispatch. If the stream
493-- is composed of hierarchically composed grains of different sizes then we can
494-- always switch to a desired granularity depending on the consumer speed.
496-- TBD for pure work (when we are not in the IO monad) we can divide it into
497-- just the number of CPUs.
499{-# NOINLINE withNewSVar2 #-}
500withNewSVar2 :: MonadAsync m
501 => SVarStyle -> Stream m a -> Stream m a -> Stream m a
502withNewSVar2 style m1 m2 = Stream $ \_ stp yld -> do
503 sv <- newStreamVar2 style m1 m2
504 (runStream (fromStreamVar sv)) Nothing stp yld
506-- | Join two computations on the currently running 'SVar' queue for concurrent
507-- execution. The 'SVarStyle' required by the current composition context is
508-- passed as one of the parameters. If the style does not match with the style
509-- of the current 'SVar' we create a new 'SVar' and schedule the computations
510-- on that. The newly created SVar joins as one of the computations on the
511-- current SVar queue.
513-- When we are using parallel composition, an SVar is passed around as a state
514-- variable. We try to schedule a new parallel computation on the SVar passed
515-- to us. The first time, when no SVar exists, a new SVar is created.
516-- Subsequently, 'joinStreamVar2' may get called when a computation already
517-- scheduled on the SVar is further evaluated. For example, when (a \<|> b) is
518-- evaluated it calls a 'joinStreamVar2' to put 'a' and 'b' on the current scheduler
519-- queue. However, if the scheduling and composition style of the new
520-- computation being scheduled is different than the style of the current SVar,
521-- then we create a new SVar and schedule it on that.
525-- * (x \<|> y) \<|> (t \<|> u) -- all of them get scheduled on the same SVar
526-- * (x \<|> y) \<|> (t \<| u) -- @t@ and @u@ get scheduled on a new child SVar
527-- because of the scheduling policy change.
528-- * if we 'adapt' a stream of type 'AsyncT' to a stream of type
529-- 'ParallelT', we create a new SVar at the transitioning bind.
530-- * When the stream is switching from disjunctive composition to conjunctive
531-- composition and vice-versa we create a new SVar to isolate the scheduling
534{-# INLINE joinStreamVar2 #-}
535joinStreamVar2 :: MonadAsync m
536 => SVarStyle -> Stream m a -> Stream m a -> Stream m a
537joinStreamVar2 style m1 m2 = Stream $ \st stp yld -> do
539 Just sv | svarStyle sv == style ->
540 liftIO ((enqueue sv) m2) >> (runStream m1) st stp yld
541 _ -> (runStream (withNewSVar2 style m1 m2)) Nothing stp yld
543------------------------------------------------------------------------------
544-- Semigroup and Monoid style compositions for parallel actions
545------------------------------------------------------------------------------
549parAhead :: Stream m a -> Stream m a -> Stream m a
552-- | Sequential composition similar to '<>' except that it can execute the
553-- action on the right in parallel ahead of time. Returns the results in
554-- sequential order like '<>' from left to right.
555(<>|) :: Stream m a -> Stream m a -> Stream m a
559-- | Same as '<|>'. Since this schedules all the composed streams fairly you
560-- cannot fold infinite number of streams using this operation.
562parAlt :: MonadAsync m => Stream m a -> Stream m a -> Stream m a
563parAlt = joinStreamVar2 (SVarStyle Disjunction FIFO)
565-- | Same as '<|'. Since this schedules the left side computation first you can
566-- right fold an infinite container using this operator. However a left fold
567-- will not work well as it first unpeels the whole structure before scheduling
568-- a computation requiring an amount of memory proportional to the size of the
570{-# INLINE parLeft #-}
571parLeft :: MonadAsync m => Stream m a -> Stream m a -> Stream m a
572parLeft = joinStreamVar2 (SVarStyle Disjunction LIFO)
574-------------------------------------------------------------------------------
575-- Instances (only used for deriving newtype instances)
576-------------------------------------------------------------------------------
578-- Stream type is not exposed, these instances are only for deriving instances
579-- for the newtype wrappers based on Stream.
581-- Dummy Instances, defined to enable the definition of other instances that
582-- require a Monad constraint. Must be defined by the newtypes.
584instance Monad m => Functor (Stream m) where
587instance Monad m => Applicative (Stream m) where
591instance Monad m => Monad (Stream m) where
595------------------------------------------------------------------------------
596-- Alternative & MonadPlus
597------------------------------------------------------------------------------
599-- | `empty` represents an action that takes non-zero time to complete. Since
600-- all actions take non-zero time, an `Alternative` composition ('<|>') is a
601-- monoidal composition executing all actions in parallel, it is similar to
602-- '<>' except that it runs all the actions in parallel and interleaves their
604instance MonadAsync m => Alternative (Stream m) where
608instance MonadAsync m => MonadPlus (Stream m) where
612-------------------------------------------------------------------------------
614-------------------------------------------------------------------------------
616instance MonadTrans Stream where
617 lift mx = Stream $ \_ _ yld -> mx >>= (\a -> (yld a Nothing))
619instance (MonadBase b m, Monad m) => MonadBase b (Stream m) where
620 liftBase = liftBaseDefault
622------------------------------------------------------------------------------
623-- Standard transformer instances
624------------------------------------------------------------------------------
626instance MonadIO m => MonadIO (Stream m) where
627 liftIO = lift . liftIO
629instance MonadThrow m => MonadThrow (Stream m) where
630 throwM = lift . throwM
632-- XXX handle and test cross thread state transfer
633instance MonadError e m => MonadError e (Stream m) where
634 throwError = lift . throwError
635 catchError m h = Stream $ \st stp yld ->
636 let handle r = r `catchError` \e -> (runStream (h e)) st stp yld
637 yield a Nothing = yld a Nothing
638 yield a (Just r) = yld a (Just (catchError r h))
639 in handle $ (runStream m) st stp yield
641instance MonadReader r m => MonadReader r (Stream m) where
643 local f m = Stream $ \st stp yld ->
644 let yield a Nothing = local f $ yld a Nothing
645 yield a (Just r) = local f $ yld a (Just (local f r))
646 in (runStream m) st (local f stp) yield
648instance MonadState s m => MonadState s (Stream m) where
651 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