1{-# LANGUAGE ConstraintKinds #-}
2{-# LANGUAGE FlexibleContexts #-}
3{-# LANGUAGE FlexibleInstances #-}
4{-# LANGUAGE MultiParamTypeClasses #-}
5{-# LANGUAGE RankNTypes #-}
6{-# LANGUAGE UndecidableInstances #-} -- XXX
9-- Module : Asyncly.Core
10-- Copyright : (c) 2017 Harendra Kumar
13-- Maintainer : harendra.kumar@gmail.com
14-- Stability : experimental
27 -- * Concurrent Stream Vars (SVars)
39 -- * Concurrent Streams
45import Control.Applicative (Alternative (..))
46import Control.Concurrent (ThreadId, forkIO,
47 myThreadId, threadDelay)
48import Control.Concurrent.MVar (MVar, newEmptyMVar, tryTakeMVar,
50import Control.Exception (SomeException (..))
51import qualified Control.Exception.Lifted as EL
52import Control.Monad (MonadPlus(..), mzero, when)
53import Control.Monad.Base (MonadBase (..), liftBaseDefault)
54import Control.Monad.Catch (MonadThrow, throwM)
55import Control.Monad.Error.Class (MonadError(..))
56import Control.Monad.IO.Class (MonadIO(..))
57import Control.Monad.Reader.Class (MonadReader(..))
58import Control.Monad.State.Class (MonadState(..))
59import Control.Monad.Trans.Class (MonadTrans (lift))
60import Control.Monad.Trans.Control (MonadBaseControl, liftBaseWith)
61import Data.Atomics (atomicModifyIORefCAS,
62 atomicModifyIORefCAS_)
63import Data.Concurrent.Queue.MichaelScott (LinkedQueue, newQ, pushL,
65import Data.Functor (void)
66import Data.IORef (IORef, modifyIORef, newIORef,
68import Data.Maybe (isNothing)
69import Data.Semigroup (Semigroup(..))
71import qualified Data.Set as S
73------------------------------------------------------------------------------
74-- Parent child thread communication type
75------------------------------------------------------------------------------
77-- | Events that a child thread may send to a parent thread.
80 | ChildStop ThreadId (Maybe SomeException)
82------------------------------------------------------------------------------
83-- State threaded around the monad for thread management
84------------------------------------------------------------------------------
86-- | Conjunction is used for monadic/product style composition. Disjunction is
87-- used for fold/sum style composition. We need to distiguish the two types of
88-- SVars so that the scheduling of the two is independent.
89data SVarTag = Conjunction | Disjunction deriving Eq
91-- | For fairly interleaved parallel composition the sched policy is FIFO
92-- whereas for left biased parallel composition it is LIFO.
93data SVarSched = LIFO | FIFO deriving Eq
95-- | Identify the type of the SVar. Two computations using the same style can
96-- be scheduled on the same SVar.
97data SVarStyle = SVarStyle SVarTag SVarSched deriving Eq
99-- | An SVar or a Stream Var is a conduit to the output from multiple streams
100-- running concurrently and asynchronously. An SVar can be thought of as an
101-- asynchronous IO handle. We can write any number of streams to an SVar in a
102-- non-blocking manner and then read them back at any time at any pace. The
103-- SVar would run the streams asynchronously and accumulate results. An SVar
104-- may not really execute the stream completely and accumulate all the results.
105-- However, it ensures that the reader can read the results at whatever paces
106-- it wants to read. The SVar monitors and adapts to the consumer's pace.
108-- An SVar is a mini scheduler, it has an associated runqueue that holds the
109-- stream tasks to be picked and run by a pool of worker threads. It has an
110-- associated output queue where the output stream elements are placed by the
111-- worker threads. A doorBell is used by the worker threads to intimate the
112-- consumer thread about availability of new results in the output queue. More
113-- workers are added to the SVar by 'fromStreamVar' on demand if the output
114-- produced is not keeping pace with the consumer. On bounded SVars, workers
115-- block on the output queue to provide throttling of the producer when the
116-- consumer is not pulling fast enough. The number of workers may even get
117-- reduced depending on the consuming pace.
119-- New work is enqueued either at the time of creation of the SVar or as a
120-- result of executing the parallel combinators i.e. '<|' and '<|>' when the
121-- already enqueued computations get evaluated. See 'joinStreamVar2'.
124 SVar { outputQueue :: IORef [ChildEvent a]
125 , doorBell :: MVar Bool -- wakeup mechanism for outQ
126 , enqueue :: Stream m a -> IO ()
128 , runningThreads :: IORef (Set ThreadId)
129 , queueEmpty :: m Bool
130 , svarStyle :: SVarStyle
133------------------------------------------------------------------------------
135------------------------------------------------------------------------------
137-- TBD use a functor instead of the bare type a?
139-- | Represents a monadic stream of values of type 'a' constructed using
140-- actions in monad 'm'. Streams can be composed sequentially or in parallel;
141-- in product style compositions (monadic bind multiplies streams in a ListT
142-- fashion) or in sum style compositions like 'Semigroup', 'Monoid',
143-- 'Alternative' or variants of these.
146 runStream :: forall r.
147 Maybe (SVar m a) -- local state
149 -> (a -> Maybe (Stream m a) -> m r) -- yield
153-- | A monad that can perform asynchronous/concurrent IO operations.
154type MonadAsync m = (MonadIO m, MonadBaseControl IO m, MonadThrow m)
156-- | Yield a singleton value as a stream.
157yields :: a -> Stream m a
158yields a = Stream $ \_ _ yld -> yld a Nothing
160------------------------------------------------------------------------------
162------------------------------------------------------------------------------
164-- | '<>' concatenates two streams sequentially i.e. the first stream is
165-- exhausted completely before yielding any element from the second stream.
166instance Semigroup (Stream m a) where
169 go (Stream m) = Stream $ \_ stp yld ->
170 let stop = (runStream m2) Nothing stp yld
171 yield a Nothing = yld a (Just m2)
172 yield a (Just r) = yld a (Just (go r))
173 in m Nothing stop yield
175------------------------------------------------------------------------------
177------------------------------------------------------------------------------
179instance Monoid (Stream m a) where
180 mempty = Stream $ \_ stp _ -> stp
183------------------------------------------------------------------------------
185------------------------------------------------------------------------------
188interleave :: Stream m a -> Stream m a -> Stream m a
189interleave m1 m2 = Stream $ \_ stp yld -> do
190 let stop = (runStream m2) Nothing stp yld
191 yield a Nothing = yld a (Just m2)
192 yield a (Just r) = yld a (Just (interleave m2 r))
193 (runStream m1) Nothing stop yield
195------------------------------------------------------------------------------
196-- Spawning threads and collecting result in streamed fashion
197------------------------------------------------------------------------------
200doFork :: MonadBaseControl IO m
202 -> (SomeException -> m ())
204doFork action exHandler =
205 EL.mask $ \restore ->
206 liftBaseWith $ \runInIO -> forkIO $ do
207 -- XXX test the exception handling
208 _ <- runInIO $ EL.catch (restore action) exHandler
209 -- XXX restore state here?
212-- XXX exception safety of all atomic/MVar operations
215send :: MonadIO m => SVar m a -> ChildEvent a -> m ()
216send sv msg = liftIO $ do
217 atomicModifyIORefCAS_ (outputQueue sv) $ \es -> msg : es
218 -- XXX need a memory barrier? The wake up must happen only after the
219 -- store has finished otherwise we can have lost wakeup problems.
220 void $ tryPutMVar (doorBell sv) True
222{-# INLINE sendStop #-}
223sendStop :: MonadIO m => SVar m a -> m ()
224sendStop sv = liftIO myThreadId >>= \tid -> send sv (ChildStop tid Nothing)
226-- Note: Left associated compositions can grow this queue to a large size
227{-# INLINE enqueueLIFO #-}
228enqueueLIFO :: IORef [Stream m a] -> Stream m a -> IO ()
229enqueueLIFO q m = atomicModifyIORefCAS_ q $ \ ms -> m : ms
231runqueueLIFO :: MonadIO m => SVar m a -> IORef [Stream m a] -> m ()
232runqueueLIFO sv q = run
239 Nothing -> sendStop sv
240 Just m -> (runStream m) (Just sv) run yield
242 sendit a = send sv (ChildYield a)
243 yield a Nothing = sendit a >> run
244 yield a (Just r) = sendit a >> (runStream r) (Just sv) run yield
246 dequeue = liftIO $ atomicModifyIORefCAS q $ \ ms ->
249 x : xs -> (xs, Just x)
251{-# INLINE enqueueFIFO #-}
252enqueueFIFO :: LinkedQueue (Stream m a) -> Stream m a -> IO ()
255runqueueFIFO :: MonadIO m => SVar m a -> LinkedQueue (Stream m a) -> m ()
256runqueueFIFO sv q = run
263 Nothing -> sendStop sv
264 Just m -> (runStream m) (Just sv) run yield
266 dequeue = liftIO $ tryPopR q
267 sendit a = send sv (ChildYield a)
268 yield a Nothing = sendit a >> run
269 yield a (Just r) = sendit a >> liftIO (enqueueFIFO q r) >> run
271-- Thread tracking is needed for two reasons:
273-- 1) Killing threads on exceptions. Threads may not be allowed to go away by
274-- themselves because they may run for significant times before going away or
275-- worse they may be stuck in IO and never go away.
277-- 2) To know when all threads are done.
279{-# NOINLINE addThread #-}
280addThread :: MonadIO m => SVar m a -> ThreadId -> m ()
282 liftIO $ modifyIORef (runningThreads sv) $ (\s -> S.insert tid s)
284{-# INLINE delThread #-}
285delThread :: MonadIO m => SVar m a -> ThreadId -> m ()
287 liftIO $ modifyIORef (runningThreads sv) $ (\s -> S.delete tid s)
289{-# INLINE allThreadsDone #-}
290allThreadsDone :: MonadIO m => SVar m a -> m Bool
291allThreadsDone sv = liftIO $ do
292 readIORef (runningThreads sv) >>= return . S.null
294{-# NOINLINE handleChildException #-}
295handleChildException :: MonadIO m => SVar m a -> SomeException -> m ()
296handleChildException sv e = do
297 tid <- liftIO myThreadId
298 send sv (ChildStop tid (Just e))
300{-# NOINLINE pushWorker #-}
301pushWorker :: MonadAsync m => SVar m a -> m ()
303 doFork (runqueue sv) (handleChildException sv) >>= addThread sv
305-- XXX When the queue is LIFO we can put a limit on the number of dispatches.
306-- Also, if a worker blocks on the output queue we can decide if we want to
307-- block or make it go away entirely, depending on the number of workers and
308-- the type of the queue.
309{-# INLINE sendWorkerWait #-}
310sendWorkerWait :: MonadAsync m => SVar m a -> m ()
311sendWorkerWait sv = do
313 SVarStyle _ LIFO -> liftIO $ threadDelay 200
314 SVarStyle _ FIFO -> liftIO $ threadDelay 0
316 output <- liftIO $ readIORef (outputQueue sv)
317 when (null output) $ do
318 done <- queueEmpty sv
320 then (pushWorker sv) >> sendWorkerWait sv
321 else void (liftIO $ takeMVar (doorBell sv))
323-- | Pull a stream from an SVar.
324{-# NOINLINE fromStreamVar #-}
325fromStreamVar :: MonadAsync m => SVar m a -> Stream m a
326fromStreamVar sv = Stream $ \_ stp yld -> do
327 -- XXX if reading the IORef is costly we can use a flag in the SVar to
328 -- indicate we are done.
329 done <- allThreadsDone sv
333 res <- liftIO $ tryTakeMVar (doorBell sv)
334 when (isNothing res) $ sendWorkerWait sv
335 list <- liftIO $ atomicModifyIORefCAS (outputQueue sv) $ \x -> ([], x)
336 -- To avoid lock overhead we read all events at once instead of reading
337 -- one at a time. We just reverse the list to process the events in the
338 -- order they arrived. Maybe we can use a queue instead?
339 (runStream $ processEvents (reverse list)) Nothing stp yld
343 handleException e tid = do
345 -- XXX implement kill async exception handling
346 -- liftIO $ readIORef (runningThreads sv) >>= mapM_ killThread
349 {-# INLINE processEvents #-}
350 processEvents [] = Stream $ \_ stp yld -> do
351 done <- allThreadsDone sv
353 then (runStream (fromStreamVar sv)) Nothing stp yld
356 processEvents (ev : es) = Stream $ \_ stp yld -> do
357 let continue = (runStream (processEvents es)) Nothing stp yld
358 yield a = yld a (Just (processEvents es))
361 ChildYield a -> yield a
364 Nothing -> delThread sv tid >> continue
365 Just ex -> handleException ex tid
367getFifoSVar :: MonadIO m => SVarStyle -> IO (SVar m a)
368getFifoSVar ctype = do
370 outQMv <- newEmptyMVar
371 running <- newIORef S.empty
374 SVar { outputQueue = outQ
376 , runningThreads = running
377 , runqueue = runqueueFIFO sv q
379 , queueEmpty = liftIO $ nullQ q
384getLifoSVar :: MonadIO m => SVarStyle -> IO (SVar m a)
385getLifoSVar ctype = do
387 outQMv <- newEmptyMVar
388 running <- newIORef S.empty
390 let checkEmpty = liftIO (readIORef q) >>= return . null
392 SVar { outputQueue = outQ
394 , runningThreads = running
395 , runqueue = runqueueLIFO sv q
396 , enqueue = enqueueLIFO q
397 , queueEmpty = checkEmpty
402-- | Create a new empty SVar.
403newEmptySVar :: MonadAsync m => SVarStyle -> m (SVar m a)
404newEmptySVar style = do
407 SVarStyle _ FIFO -> do
408 c <- getFifoSVar style
410 SVarStyle _ LIFO -> do
411 c <- getLifoSVar style
415-- | Create a new SVar and enqueue one stream computation on it.
416newStreamVar1 :: MonadAsync m => SVarStyle -> Stream m a -> m (SVar m a)
417newStreamVar1 style m = do
418 sv <- newEmptySVar style
419 -- Note: We must have all the work on the queue before sending the
420 -- pushworker, otherwise the pushworker may exit before we even get a
422 liftIO $ (enqueue sv) m
426-- | Create a new SVar and enqueue two stream computations on it.
427newStreamVar2 :: MonadAsync m
428 => SVarStyle -> Stream m a -> Stream m a -> m (SVar m a)
429newStreamVar2 style m1 m2 = do
430 -- Note: We must have all the work on the queue before sending the
431 -- pushworker, otherwise the pushworker may exit before we even get a
435 SVarStyle _ FIFO -> do
436 c <- getFifoSVar style
437 (enqueue c) m1 >> (enqueue c) m2
439 SVarStyle _ LIFO -> do
440 c <- getLifoSVar style
441 (enqueue c) m2 >> (enqueue c) m1
446-- | Write a stream to an 'SVar' in a non-blocking manner. The stream can then
447-- be read back from the SVar using 'fromSVar'.
448toStreamVar :: MonadAsync m => SVar m a -> Stream m a -> m ()
450 liftIO $ (enqueue sv) m
451 done <- allThreadsDone sv
452 -- XXX there may be a race here unless we are running in the consumer
453 -- thread. This is safe only when called from the consumer thread or when
454 -- no consumer is present.
455 when done $ pushWorker sv
457------------------------------------------------------------------------------
458-- Running streams concurrently
459------------------------------------------------------------------------------
461-- Concurrency rate control. Our objective is to create more threads on demand
462-- if the consumer is running faster than us. As soon as we encounter an
463-- Alternative composition we create a push pull pair of threads. We use a
464-- channel for communication between the consumer pulling from the channel and
465-- the producer who pushing to the channel. The producer creates more threads
466-- if no output is seen on the channel, that is the consumer is running faster.
467-- However this mechanism can be problematic if the initial production latency
468-- is high, we may end up creating too many threads. So we need some way to
469-- monitor and use the latency as well.
471-- TBD We may run computations at the lower level of the composition tree
472-- serially even if they are composed using a parallel combinator. We can use
473-- <> in place of <| and <=> in place of <|>. If we find that a parallel
474-- channel immediately above a computation becomes empty we can switch to
475-- parallelizing the computation. For that we can use a state flag to fork the
476-- rest of the computation at any point of time inside the Monad bind operation
477-- if the consumer is running at a faster speed.
479-- TBD the alternative composition allows us to dispatch a chunkSize of only 1.
480-- If we have to dispatch in arbitrary chunksizes we will need to compose the
481-- parallel actions using a data constructor (Free Alternative) instead so that
482-- we can divide it in chunks of arbitrary size before dispatch. If the stream
483-- is composed of hierarchically composed grains of different sizes then we can
484-- always switch to a desired granularity depending on the consumer speed.
486-- TBD for pure work (when we are not in the IO monad) we can divide it into
487-- just the number of CPUs.
489{-# NOINLINE withNewSVar2 #-}
490withNewSVar2 :: MonadAsync m
491 => SVarStyle -> Stream m a -> Stream m a -> Stream m a
492withNewSVar2 style m1 m2 = Stream $ \_ stp yld -> do
493 sv <- newStreamVar2 style m1 m2
494 (runStream (fromStreamVar sv)) Nothing stp yld
496-- | Join two computations on the currently running 'SVar' queue for concurrent
497-- execution. The 'SVarStyle' required by the current composition context is
498-- passed as one of the parameters. If the style does not match with the style
499-- of the current 'SVar' we create a new 'SVar' and schedule the computations
500-- on that. The newly created SVar joins as one of the computations on the
501-- current SVar queue.
503-- When we are using parallel composition, an SVar is passed around as a state
504-- variable. We try to schedule a new parallel computation on the SVar passed
505-- to us. The first time, when no SVar exists, a new SVar is created.
506-- Subsequently, 'joinStreamVar2' may get called when a computation already
507-- scheduled on the SVar is further evaluated. For example, when (a \<|> b) is
508-- evaluated it calls a 'joinStreamVar2' to put 'a' and 'b' on the current scheduler
509-- queue. However, if the scheduling and composition style of the new
510-- computation being scheduled is different than the style of the current SVar,
511-- then we create a new SVar and schedule it on that.
515-- * (x \<|> y) \<|> (t \<|> u) -- all of them get scheduled on the same SVar
516-- * (x \<|> y) \<|> (t \<| u) -- @t@ and @u@ get scheduled on a new child SVar
517-- because of the scheduling policy change.
518-- * if we 'adapt' a stream of type 'AsyncT' to a stream of type
519-- 'ParallelT', we create a new SVar at the transitioning bind.
520-- * When the stream is switching from disjunctive composition to conjunctive
521-- composition and vice-versa we create a new SVar to isolate the scheduling
524{-# INLINE joinStreamVar2 #-}
525joinStreamVar2 :: MonadAsync m
526 => SVarStyle -> Stream m a -> Stream m a -> Stream m a
527joinStreamVar2 style m1 m2 = Stream $ \st stp yld -> do
529 Just sv | svarStyle sv == style ->
530 liftIO ((enqueue sv) m2) >> (runStream m1) st stp yld
531 _ -> (runStream (withNewSVar2 style m1 m2)) Nothing stp yld
533------------------------------------------------------------------------------
534-- Semigroup and Monoid style compositions for parallel actions
535------------------------------------------------------------------------------
539parAhead :: Stream m a -> Stream m a -> Stream m a
542-- | Sequential composition similar to '<>' except that it can execute the
543-- action on the right in parallel ahead of time. Returns the results in
544-- sequential order like '<>' from left to right.
545(<>|) :: Stream m a -> Stream m a -> Stream m a
549-- | Same as '<|>'. Since this schedules all the composed streams fairly you
550-- cannot fold infinite number of streams using this operation.
552parAlt :: MonadAsync m => Stream m a -> Stream m a -> Stream m a
553parAlt = joinStreamVar2 (SVarStyle Disjunction FIFO)
555-- | Same as '<|'. Since this schedules the left side computation first you can
556-- right fold an infinite container using this operator. However a left fold
557-- will not work well as it first unpeels the whole structure before scheduling
558-- a computation requiring an amount of memory proportional to the size of the
560{-# INLINE parLeft #-}
561parLeft :: MonadAsync m => Stream m a -> Stream m a -> Stream m a
562parLeft = joinStreamVar2 (SVarStyle Disjunction LIFO)
564-------------------------------------------------------------------------------
565-- Instances (only used for deriving newtype instances)
566-------------------------------------------------------------------------------
568-- Stream type is not exposed, these instances are only for deriving instances
569-- for the newtype wrappers based on Stream.
571-- Dummy Instances, defined to enable the definition of other instances that
572-- require a Monad constraint. Must be defined by the newtypes.
574instance Monad m => Functor (Stream m) where
577instance Monad m => Applicative (Stream m) where
581instance Monad m => Monad (Stream m) where
585------------------------------------------------------------------------------
586-- Alternative & MonadPlus
587------------------------------------------------------------------------------
589-- | `empty` represents an action that takes non-zero time to complete. Since
590-- all actions take non-zero time, an `Alternative` composition ('<|>') is a
591-- monoidal composition executing all actions in parallel, it is similar to
592-- '<>' except that it runs all the actions in parallel and interleaves their
594instance MonadAsync m => Alternative (Stream m) where
598instance MonadAsync m => MonadPlus (Stream m) where
602-------------------------------------------------------------------------------
604-------------------------------------------------------------------------------
606instance MonadTrans Stream where
607 lift mx = Stream $ \_ _ yld -> mx >>= (\a -> (yld a Nothing))
609instance (MonadBase b m, Monad m) => MonadBase b (Stream m) where
610 liftBase = liftBaseDefault
612------------------------------------------------------------------------------
613-- Standard transformer instances
614------------------------------------------------------------------------------
616instance MonadIO m => MonadIO (Stream m) where
617 liftIO = lift . liftIO
619instance MonadThrow m => MonadThrow (Stream m) where
620 throwM = lift . throwM
622-- XXX handle and test cross thread state transfer
623instance MonadError e m => MonadError e (Stream m) where
624 throwError = lift . throwError
625 catchError m h = Stream $ \st stp yld ->
626 let handle r = r `catchError` \e -> (runStream (h e)) st stp yld
627 yield a Nothing = yld a Nothing
628 yield a (Just r) = yld a (Just (catchError r h))
629 in handle $ (runStream m) st stp yield
631instance MonadReader r m => MonadReader r (Stream m) where
633 local f m = Stream $ \st stp yld ->
634 let yield a Nothing = local f $ yld a Nothing
635 yield a (Just r) = local f $ yld a (Just (local f r))
636 in (runStream m) st (local f stp) yield
638instance MonadState s m => MonadState s (Stream m) where
641 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