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
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.
160type MonadAsync m = (MonadIO m, MonadBaseControl IO m, MonadThrow m)
162scons :: a -> Maybe (Stream m a) -> Stream m a
163scons a r = Stream $ \_ _ yld -> yld a r
166snil = Stream $ \_ stp _ -> stp
168------------------------------------------------------------------------------
170------------------------------------------------------------------------------
172-- | '<>' concatenates two streams sequentially i.e. the first stream is
173-- exhausted completely before yielding any element from the second stream.
174instance Semigroup (Stream m a) where
177 go (Stream m) = Stream $ \_ stp yld ->
178 let stop = (runStream m2) Nothing stp yld
179 yield a Nothing = yld a (Just m2)
180 yield a (Just r) = yld a (Just (go r))
181 in m Nothing stop yield
183------------------------------------------------------------------------------
185------------------------------------------------------------------------------
187instance Monoid (Stream m a) where
188 mempty = Stream $ \_ stp _ -> stp
191------------------------------------------------------------------------------
193------------------------------------------------------------------------------
196interleave :: Stream m a -> Stream m a -> Stream m a
197interleave m1 m2 = Stream $ \_ stp yld -> do
198 let stop = (runStream m2) Nothing stp yld
199 yield a Nothing = yld a (Just m2)
200 yield a (Just r) = yld a (Just (interleave m2 r))
201 (runStream m1) Nothing stop yield
203------------------------------------------------------------------------------
204-- Spawning threads and collecting result in streamed fashion
205------------------------------------------------------------------------------
208doFork :: MonadBaseControl IO m
210 -> (SomeException -> m ())
212doFork action exHandler =
213 EL.mask $ \restore ->
214 liftBaseWith $ \runInIO -> forkIO $ do
215 -- XXX test the exception handling
216 _ <- runInIO $ EL.catch (restore action) exHandler
217 -- XXX restore state here?
220-- XXX exception safety of all atomic/MVar operations
223send :: MonadIO m => SVar m a -> ChildEvent a -> m ()
224send sv msg = liftIO $ do
225 atomicModifyIORefCAS_ (outputQueue sv) $ \es -> msg : es
226 -- XXX need a memory barrier? The wake up must happen only after the
227 -- store has finished otherwise we can have lost wakeup problems.
228 void $ tryPutMVar (doorBell sv) True
230{-# INLINE sendStop #-}
231sendStop :: MonadIO m => SVar m a -> m ()
232sendStop sv = liftIO myThreadId >>= \tid -> send sv (ChildStop tid Nothing)
234-- Note: Left associated compositions can grow this queue to a large size
235{-# INLINE enqueueLIFO #-}
236enqueueLIFO :: IORef [Stream m a] -> Stream m a -> IO ()
237enqueueLIFO q m = atomicModifyIORefCAS_ q $ \ ms -> m : ms
239runqueueLIFO :: MonadIO m => SVar m a -> IORef [Stream m a] -> m ()
240runqueueLIFO sv q = run
247 Nothing -> sendStop sv
248 Just m -> (runStream m) (Just sv) run yield
250 sendit a = send sv (ChildYield a)
251 yield a Nothing = sendit a >> run
252 yield a (Just r) = sendit a >> (runStream r) (Just sv) run yield
254 dequeue = liftIO $ atomicModifyIORefCAS q $ \ ms ->
257 x : xs -> (xs, Just x)
259{-# INLINE enqueueFIFO #-}
260enqueueFIFO :: LinkedQueue (Stream m a) -> Stream m a -> IO ()
263runqueueFIFO :: MonadIO m => SVar m a -> LinkedQueue (Stream m a) -> m ()
264runqueueFIFO sv q = run
271 Nothing -> sendStop sv
272 Just m -> (runStream m) (Just sv) run yield
274 dequeue = liftIO $ tryPopR q
275 sendit a = send sv (ChildYield a)
276 yield a Nothing = sendit a >> run
277 yield a (Just r) = sendit a >> liftIO (enqueueFIFO q r) >> run
279-- Thread tracking is needed for two reasons:
281-- 1) Killing threads on exceptions. Threads may not be allowed to go away by
282-- themselves because they may run for significant times before going away or
283-- worse they may be stuck in IO and never go away.
285-- 2) To know when all threads are done.
287{-# NOINLINE addThread #-}
288addThread :: MonadIO m => SVar m a -> ThreadId -> m ()
290 liftIO $ modifyIORef (runningThreads sv) $ (\s -> S.insert tid s)
292{-# INLINE delThread #-}
293delThread :: MonadIO m => SVar m a -> ThreadId -> m ()
295 liftIO $ modifyIORef (runningThreads sv) $ (\s -> S.delete tid s)
297{-# INLINE allThreadsDone #-}
298allThreadsDone :: MonadIO m => SVar m a -> m Bool
299allThreadsDone sv = liftIO $ do
300 readIORef (runningThreads sv) >>= return . S.null
302{-# NOINLINE handleChildException #-}
303handleChildException :: MonadIO m => SVar m a -> SomeException -> m ()
304handleChildException sv e = do
305 tid <- liftIO myThreadId
306 send sv (ChildStop tid (Just e))
308{-# NOINLINE pushWorker #-}
309pushWorker :: MonadAsync m => SVar m a -> m ()
311 doFork (runqueue sv) (handleChildException sv) >>= addThread sv
313-- XXX When the queue is LIFO we can put a limit on the number of dispatches.
314-- Also, if a worker blocks on the output queue we can decide if we want to
315-- block or make it go away entirely, depending on the number of workers and
316-- the type of the queue.
317{-# INLINE sendWorkerWait #-}
318sendWorkerWait :: MonadAsync m => SVar m a -> m ()
319sendWorkerWait sv = do
321 SVarStyle _ LIFO -> liftIO $ threadDelay 200
322 SVarStyle _ FIFO -> liftIO $ threadDelay 0
324 output <- liftIO $ readIORef (outputQueue sv)
325 when (null output) $ do
326 done <- queueEmpty sv
328 then (pushWorker sv) >> sendWorkerWait sv
329 else void (liftIO $ takeMVar (doorBell sv))
331-- | Pull a stream from an SVar.
332{-# NOINLINE fromStreamVar #-}
333fromStreamVar :: MonadAsync m => SVar m a -> Stream m a
334fromStreamVar sv = Stream $ \_ stp yld -> do
335 -- XXX if reading the IORef is costly we can use a flag in the SVar to
336 -- indicate we are done.
337 done <- allThreadsDone sv
341 res <- liftIO $ tryTakeMVar (doorBell sv)
342 when (isNothing res) $ sendWorkerWait sv
343 list <- liftIO $ atomicModifyIORefCAS (outputQueue sv) $ \x -> ([], x)
344 -- To avoid lock overhead we read all events at once instead of reading
345 -- one at a time. We just reverse the list to process the events in the
346 -- order they arrived. Maybe we can use a queue instead?
347 (runStream $ processEvents (reverse list)) Nothing stp yld
351 handleException e tid = do
353 -- XXX implement kill async exception handling
354 -- liftIO $ readIORef (runningThreads sv) >>= mapM_ killThread
357 {-# INLINE processEvents #-}
358 processEvents [] = Stream $ \_ stp yld -> do
359 done <- allThreadsDone sv
361 then (runStream (fromStreamVar sv)) Nothing stp yld
364 processEvents (ev : es) = Stream $ \_ stp yld -> do
365 let continue = (runStream (processEvents es)) Nothing stp yld
366 yield a = yld a (Just (processEvents es))
369 ChildYield a -> yield a
372 Nothing -> delThread sv tid >> continue
373 Just ex -> handleException ex tid
375getFifoSVar :: MonadIO m => SVarStyle -> IO (SVar m a)
376getFifoSVar ctype = do
378 outQMv <- newEmptyMVar
379 running <- newIORef S.empty
382 SVar { outputQueue = outQ
384 , runningThreads = running
385 , runqueue = runqueueFIFO sv q
387 , queueEmpty = liftIO $ nullQ q
392getLifoSVar :: MonadIO m => SVarStyle -> IO (SVar m a)
393getLifoSVar ctype = do
395 outQMv <- newEmptyMVar
396 running <- newIORef S.empty
398 let checkEmpty = liftIO (readIORef q) >>= return . null
400 SVar { outputQueue = outQ
402 , runningThreads = running
403 , runqueue = runqueueLIFO sv q
404 , enqueue = enqueueLIFO q
405 , queueEmpty = checkEmpty
410-- | Create a new empty SVar.
411newEmptySVar :: MonadAsync m => SVarStyle -> m (SVar m a)
412newEmptySVar style = do
415 SVarStyle _ FIFO -> do
416 c <- getFifoSVar style
418 SVarStyle _ LIFO -> do
419 c <- getLifoSVar style
423-- | Create a new SVar and enqueue one stream computation on it.
424newStreamVar1 :: MonadAsync m => SVarStyle -> Stream m a -> m (SVar m a)
425newStreamVar1 style m = do
426 sv <- newEmptySVar style
427 -- Note: We must have all the work on the queue before sending the
428 -- pushworker, otherwise the pushworker may exit before we even get a
430 liftIO $ (enqueue sv) m
434-- | Create a new SVar and enqueue two stream computations on it.
435newStreamVar2 :: MonadAsync m
436 => SVarStyle -> Stream m a -> Stream m a -> m (SVar m a)
437newStreamVar2 style m1 m2 = do
438 -- Note: We must have all the work on the queue before sending the
439 -- pushworker, otherwise the pushworker may exit before we even get a
443 SVarStyle _ FIFO -> do
444 c <- getFifoSVar style
445 (enqueue c) m1 >> (enqueue c) m2
447 SVarStyle _ LIFO -> do
448 c <- getLifoSVar style
449 (enqueue c) m2 >> (enqueue c) m1
454-- | Write a stream to an 'SVar' in a non-blocking manner. The stream can then
455-- be read back from the SVar using 'fromSVar'.
456toStreamVar :: MonadAsync m => SVar m a -> Stream m a -> m ()
458 liftIO $ (enqueue sv) m
459 done <- allThreadsDone sv
460 -- XXX there may be a race here unless we are running in the consumer
461 -- thread. This is safe only when called from the consumer thread or when
462 -- no consumer is present.
463 when done $ pushWorker sv
465------------------------------------------------------------------------------
466-- Running streams concurrently
467------------------------------------------------------------------------------
469-- Concurrency rate control. Our objective is to create more threads on demand
470-- if the consumer is running faster than us. As soon as we encounter an
471-- Alternative composition we create a push pull pair of threads. We use a
472-- channel for communication between the consumer pulling from the channel and
473-- the producer who pushing to the channel. The producer creates more threads
474-- if no output is seen on the channel, that is the consumer is running faster.
475-- However this mechanism can be problematic if the initial production latency
476-- is high, we may end up creating too many threads. So we need some way to
477-- monitor and use the latency as well.
479-- TBD We may run computations at the lower level of the composition tree
480-- serially even if they are composed using a parallel combinator. We can use
481-- <> in place of <| and <=> in place of <|>. If we find that a parallel
482-- channel immediately above a computation becomes empty we can switch to
483-- parallelizing the computation. For that we can use a state flag to fork the
484-- rest of the computation at any point of time inside the Monad bind operation
485-- if the consumer is running at a faster speed.
487-- TBD the alternative composition allows us to dispatch a chunkSize of only 1.
488-- If we have to dispatch in arbitrary chunksizes we will need to compose the
489-- parallel actions using a data constructor (Free Alternative) instead so that
490-- we can divide it in chunks of arbitrary size before dispatch. If the stream
491-- is composed of hierarchically composed grains of different sizes then we can
492-- always switch to a desired granularity depending on the consumer speed.
494-- TBD for pure work (when we are not in the IO monad) we can divide it into
495-- just the number of CPUs.
497{-# NOINLINE withNewSVar2 #-}
498withNewSVar2 :: MonadAsync m
499 => SVarStyle -> Stream m a -> Stream m a -> Stream m a
500withNewSVar2 style m1 m2 = Stream $ \_ stp yld -> do
501 sv <- newStreamVar2 style m1 m2
502 (runStream (fromStreamVar sv)) Nothing stp yld
504-- | Join two computations on the currently running 'SVar' queue for concurrent
505-- execution. The 'SVarStyle' required by the current composition context is
506-- passed as one of the parameters. If the style does not match with the style
507-- of the current 'SVar' we create a new 'SVar' and schedule the computations
508-- on that. The newly created SVar joins as one of the computations on the
509-- current SVar queue.
511-- When we are using parallel composition, an SVar is passed around as a state
512-- variable. We try to schedule a new parallel computation on the SVar passed
513-- to us. The first time, when no SVar exists, a new SVar is created.
514-- Subsequently, 'joinStreamVar2' may get called when a computation already
515-- scheduled on the SVar is further evaluated. For example, when (a \<|> b) is
516-- evaluated it calls a 'joinStreamVar2' to put 'a' and 'b' on the current scheduler
517-- queue. However, if the scheduling and composition style of the new
518-- computation being scheduled is different than the style of the current SVar,
519-- then we create a new SVar and schedule it on that.
523-- * (x \<|> y) \<|> (t \<|> u) -- all of them get scheduled on the same SVar
524-- * (x \<|> y) \<|> (t \<| u) -- @t@ and @u@ get scheduled on a new child SVar
525-- because of the scheduling policy change.
526-- * if we 'adapt' a stream of type 'AsyncT' to a stream of type
527-- 'ParallelT', we create a new SVar at the transitioning bind.
528-- * When the stream is switching from disjunctive composition to conjunctive
529-- composition and vice-versa we create a new SVar to isolate the scheduling
532{-# INLINE joinStreamVar2 #-}
533joinStreamVar2 :: MonadAsync m
534 => SVarStyle -> Stream m a -> Stream m a -> Stream m a
535joinStreamVar2 style m1 m2 = Stream $ \st stp yld -> do
537 Just sv | svarStyle sv == style ->
538 liftIO ((enqueue sv) m2) >> (runStream m1) st stp yld
539 _ -> (runStream (withNewSVar2 style m1 m2)) Nothing stp yld
541------------------------------------------------------------------------------
542-- Semigroup and Monoid style compositions for parallel actions
543------------------------------------------------------------------------------
547parAhead :: Stream m a -> Stream m a -> Stream m a
550-- | Sequential composition similar to '<>' except that it can execute the
551-- action on the right in parallel ahead of time. Returns the results in
552-- sequential order like '<>' from left to right.
553(<>|) :: Stream m a -> Stream m a -> Stream m a
557-- | Same as '<|>'. Since this schedules all the composed streams fairly you
558-- cannot fold infinite number of streams using this operation.
560parAlt :: MonadAsync m => Stream m a -> Stream m a -> Stream m a
561parAlt = joinStreamVar2 (SVarStyle Disjunction FIFO)
563-- | Same as '<|'. Since this schedules the left side computation first you can
564-- right fold an infinite container using this operator. However a left fold
565-- will not work well as it first unpeels the whole structure before scheduling
566-- a computation requiring an amount of memory proportional to the size of the
568{-# INLINE parLeft #-}
569parLeft :: MonadAsync m => Stream m a -> Stream m a -> Stream m a
570parLeft = joinStreamVar2 (SVarStyle Disjunction LIFO)
572-------------------------------------------------------------------------------
573-- Instances (only used for deriving newtype instances)
574-------------------------------------------------------------------------------
576-- Stream type is not exposed, these instances are only for deriving instances
577-- for the newtype wrappers based on Stream.
579-- Dummy Instances, defined to enable the definition of other instances that
580-- require a Monad constraint. Must be defined by the newtypes.
582instance Monad m => Functor (Stream m) where
585instance Monad m => Applicative (Stream m) where
589instance Monad m => Monad (Stream m) where
593------------------------------------------------------------------------------
594-- Alternative & MonadPlus
595------------------------------------------------------------------------------
597-- | `empty` represents an action that takes non-zero time to complete. Since
598-- all actions take non-zero time, an `Alternative` composition ('<|>') is a
599-- monoidal composition executing all actions in parallel, it is similar to
600-- '<>' except that it runs all the actions in parallel and interleaves their
602instance MonadAsync m => Alternative (Stream m) where
606instance MonadAsync m => MonadPlus (Stream m) where
610-------------------------------------------------------------------------------
612-------------------------------------------------------------------------------
614instance MonadTrans Stream where
615 lift mx = Stream $ \_ _ yld -> mx >>= (\a -> (yld a Nothing))
617instance (MonadBase b m, Monad m) => MonadBase b (Stream m) where
618 liftBase = liftBaseDefault
620------------------------------------------------------------------------------
621-- Standard transformer instances
622------------------------------------------------------------------------------
624instance MonadIO m => MonadIO (Stream m) where
625 liftIO = lift . liftIO
627instance MonadThrow m => MonadThrow (Stream m) where
628 throwM = lift . throwM
630-- XXX handle and test cross thread state transfer
631instance MonadError e m => MonadError e (Stream m) where
632 throwError = lift . throwError
633 catchError m h = Stream $ \st stp yld ->
634 let handle r = r `catchError` \e -> (runStream (h e)) st stp yld
635 yield a Nothing = yld a Nothing
636 yield a (Just r) = yld a (Just (catchError r h))
637 in handle $ (runStream m) st stp yield
639instance MonadReader r m => MonadReader r (Stream m) where
641 local f m = Stream $ \st stp yld ->
642 let yield a Nothing = local f $ yld a Nothing
643 yield a (Just r) = local f $ yld a (Just (local f r))
644 in (runStream m) st (local f stp) yield
646instance MonadState s m => MonadState s (Stream m) where
649 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