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
47import Control.Applicative (Alternative (..))
48import Control.Concurrent (ThreadId, forkIO,
49 myThreadId, threadDelay)
50import Control.Concurrent.MVar (MVar, newEmptyMVar, tryTakeMVar,
52import Control.Exception (SomeException (..), Exception)
53import qualified Control.Exception.Lifted as EL
54import Control.Monad (liftM, MonadPlus(..), mzero,
56import Control.Monad.Base (MonadBase (..), liftBaseDefault)
57import Control.Monad.Catch (MonadThrow, throwM)
58import Control.Monad.Error.Class (MonadError(..))
59import Control.Monad.IO.Class (MonadIO(..))
60import Control.Monad.Reader.Class (MonadReader(..))
61import Control.Monad.State.Class (MonadState(..))
62import Control.Monad.Trans.Class (MonadTrans (lift))
63import Control.Monad.Trans.Control (MonadBaseControl, liftBaseWith)
64import Data.Concurrent.Queue.MichaelScott (LinkedQueue, newQ, pushL,
66import Data.Functor (void)
67import Data.IORef (IORef, modifyIORef, newIORef,
69import Data.Atomics (atomicModifyIORefCAS,
70 atomicModifyIORefCAS_)
71import Data.Maybe (isNothing)
72import Data.Semigroup (Semigroup(..))
74import qualified Data.Set as S
76------------------------------------------------------------------------------
77-- Concurrency Semantics
78------------------------------------------------------------------------------
80-- Asyncly is essentially a concurrent list transformer. To fold the lists
81-- (AsyncT m a) it provides many different ways of composing them. The Monoid
82-- instance folds lists in a sequential fashion. The Alternative instance folds
83-- a list in a concurrent manner running each action in parallel and therefore
84-- the order of results in the fold is not deterministic, but the results are
85-- fairly interleaved. There are other ways to bind and compose. The
86-- combinations of different types of binds and monoidal compositions can be
87-- used to achieve the desired results.
89------------------------------------------------------------------------------
90-- Parent child thread communication types
91------------------------------------------------------------------------------
95 | ChildStop ThreadId (Maybe SomeException)
97------------------------------------------------------------------------------
98-- State threaded around the monad for thread management
99------------------------------------------------------------------------------
101data Dimension = Conjunction | Disjunction deriving Eq
102data SchedPolicy = LIFO | FIFO deriving Eq
103data CtxType = CtxType Dimension SchedPolicy deriving Eq
106 Context { outputQueue :: IORef [ChildEvent a]
107 , doorBell :: MVar Bool -- wakeup mechanism for outQ
108 , enqueue :: Stream m a -> IO ()
110 , runningThreads :: IORef (Set ThreadId)
111 , queueEmpty :: m Bool
115-- | Represents a monadic stream of values of type 'a' resulting from actions
116-- in monad 'm'. 'AsyncT' streams can be composed sequentially or in parallel
117-- in various ways using monadic bind or variants of it and using monoidal
118-- compositions like 'Monoid', 'Alternative' or variants of these.
121 runStream :: forall r.
122 Maybe (Context m a) -- local state
124 -> (a -> Maybe (Stream m a) -> m r) -- yield
128-- | A monad that can do asynchronous (or parallel) IO operations.
129type MonadAsync m = (MonadIO m, MonadBaseControl IO m, MonadThrow m)
131yielding :: a -> Stream m a
132yielding a = Stream $ \_ _ yld -> yld a Nothing
134------------------------------------------------------------------------------
136------------------------------------------------------------------------------
138-- | '<>' concatenates two 'AsyncT' streams sequentially i.e. the first stream
139-- is exhausted completely before yielding any element from the second stream.
140instance Semigroup (Stream m a) where
143 go (Stream m) = Stream $ \_ stp yld ->
144 let stop = (runStream m2) Nothing stp yld
145 yield a Nothing = yld a (Just m2)
146 yield a (Just r) = yld a (Just (go r))
147 in m Nothing stop yield
149instance Monoid (Stream m a) where
150 mempty = Stream $ \_ stp _ -> stp
153interleave :: Stream m a -> Stream m a -> Stream m a
154interleave m1 m2 = Stream $ \_ stp yld -> do
155 let stop = (runStream m2) Nothing stp yld
156 yield a Nothing = yld a (Just m2)
157 yield a (Just r) = yld a (Just (interleave m2 r))
158 (runStream m1) Nothing stop yield
160------------------------------------------------------------------------------
161-- Streaming threads for parallel composition
162------------------------------------------------------------------------------
165doFork :: MonadBaseControl IO m
167 -> (SomeException -> m ())
169doFork action exHandler =
170 EL.mask $ \restore ->
171 liftBaseWith $ \runInIO -> forkIO $ do
172 -- XXX test the exception handling
173 _ <- runInIO $ EL.catch (restore action) exHandler
174 -- XXX restore state here?
177-- XXX exception safety of all atomic/MVar operations
180send :: MonadIO m => Context m a -> ChildEvent a -> m ()
181send ctx msg = liftIO $ do
182 atomicModifyIORefCAS_ (outputQueue ctx) $ \es -> msg : es
183 -- XXX need a memory barrier? The wake up must happen only after the
184 -- store has finished otherwise we can have lost wakeup problems.
185 void $ tryPutMVar (doorBell ctx) True
187{-# INLINE sendStop #-}
188sendStop :: MonadIO m => Context m a -> m ()
189sendStop ctx = liftIO myThreadId >>= \tid -> send ctx (ChildStop tid Nothing)
191-- Note: Left associated operations can grow this queue to a large size
192{-# INLINE enqueueLIFO #-}
193enqueueLIFO :: IORef [Stream m a] -> Stream m a -> IO ()
194enqueueLIFO q m = atomicModifyIORefCAS_ q $ \ ms -> m : ms
196runqueueLIFO :: MonadIO m => Context m a -> IORef [Stream m a] -> m ()
197runqueueLIFO ctx q = run
204 Nothing -> sendStop ctx
205 Just m -> (runStream m) (Just ctx) run yield
207 sendit a = send ctx (ChildYield a)
208 yield a Nothing = sendit a >> run
209 yield a (Just r) = sendit a >> (runStream r) (Just ctx) run yield
211 dequeue = liftIO $ atomicModifyIORefCAS q $ \ ms ->
214 x : xs -> (xs, Just x)
216{-# INLINE enqueueFIFO #-}
217enqueueFIFO :: LinkedQueue (Stream m a) -> Stream m a -> IO ()
220runqueueFIFO :: MonadIO m => Context m a -> LinkedQueue (Stream m a) -> m ()
221runqueueFIFO ctx q = run
228 Nothing -> sendStop ctx
229 Just m -> (runStream m) (Just ctx) run yield
231 dequeue = liftIO $ tryPopR q
232 sendit a = send ctx (ChildYield a)
233 yield a Nothing = sendit a >> run
234 yield a (Just r) = sendit a >> liftIO (enqueueFIFO q r) >> run
236-- Thread tracking is needed for two reasons:
238-- 1) Killing threads on exceptions. Threads may not be allowed to go away by
239-- themselves because they may run for significant times before going away or
240-- worse they may be stuck in IO and never go away.
242-- 2) To know when all threads are done.
244{-# NOINLINE addThread #-}
245addThread :: MonadIO m => Context m a -> ThreadId -> m ()
247 liftIO $ modifyIORef (runningThreads ctx) $ (\s -> S.insert tid s)
249{-# INLINE delThread #-}
250delThread :: MonadIO m => Context m a -> ThreadId -> m ()
252 liftIO $ modifyIORef (runningThreads ctx) $ (\s -> S.delete tid s)
254{-# INLINE allThreadsDone #-}
255allThreadsDone :: MonadIO m => Context m a -> m Bool
256allThreadsDone ctx = liftIO $ do
257 readIORef (runningThreads ctx) >>= return . S.null
259{-# NOINLINE handleChildException #-}
260handleChildException :: MonadIO m => Context m a -> SomeException -> m ()
261handleChildException ctx e = do
262 tid <- liftIO myThreadId
263 send ctx (ChildStop tid (Just e))
265{-# NOINLINE pushWorker #-}
266pushWorker :: MonadAsync m => Context m a -> m ()
268 doFork (runqueue ctx) (handleChildException ctx) >>= addThread ctx
270-- XXX When the queue is LIFO we can put a limit on the number of dispatches.
271-- Also, if a worker blocks on the output queue we can decide if we want to
272-- block or make it go away entirely, depending on the number of workers and
273-- the type of the queue.
274{-# INLINE sendWorkerWait #-}
275sendWorkerWait :: MonadAsync m => Context m a -> m ()
276sendWorkerWait ctx = do
278 CtxType _ LIFO -> liftIO $ threadDelay 200
279 CtxType _ FIFO -> liftIO $ threadDelay 0
281 output <- liftIO $ readIORef (outputQueue ctx)
282 when (null output) $ do
283 done <- queueEmpty ctx
285 then (pushWorker ctx) >> sendWorkerWait ctx
286 else void (liftIO $ takeMVar (doorBell ctx))
288data ContextUsedAfterEOF = ContextUsedAfterEOF deriving Show
289instance Exception ContextUsedAfterEOF
291-- | Pull an AsyncT stream from a context
292{-# NOINLINE pullFromCtx #-}
293pullFromCtx :: MonadAsync m => Context m a -> Stream m a
294pullFromCtx ctx = Stream $ \_ stp yld -> do
295 -- When using an async handle to the context, one may keep using a stale
296 -- context even after it has been fully drained. To detect it gracefully we
297 -- raise an explicit exception.
298 -- XXX if reading the IORef is costly we can use a flag in the context to
299 -- indicate we are done.
300 done <- allThreadsDone ctx
301 when done $ throwM ContextUsedAfterEOF
303 res <- liftIO $ tryTakeMVar (doorBell ctx)
304 when (isNothing res) $ sendWorkerWait ctx
305 list <- liftIO $ atomicModifyIORefCAS (outputQueue ctx) $ \x -> ([], x)
306 -- To avoid lock overhead we read all events at once instead of reading one
307 -- at a time. We just reverse the list to process the events in the order
308 -- they arrived. Maybe we can use a queue instead?
309 (runStream $ processEvents (reverse list)) Nothing stp yld
313 handleException e tid = do
315 -- XXX implement kill async exception handling
316 -- liftIO $ readIORef (runningThreads ctx) >>= mapM_ killThread
319 {-# INLINE processEvents #-}
320 processEvents [] = Stream $ \_ stp yld -> do
321 done <- allThreadsDone ctx
323 then (runStream (pullFromCtx ctx)) Nothing stp yld
326 processEvents (ev : es) = Stream $ \_ stp yld -> do
327 let continue = (runStream (processEvents es)) Nothing stp yld
328 yield a = yld a (Just (processEvents es))
331 ChildYield a -> yield a
334 Nothing -> delThread ctx tid >> continue
335 Just ex -> handleException ex tid
337getFifoCtx :: MonadIO m => CtxType -> IO (Context m a)
340 outQMv <- newEmptyMVar
341 running <- newIORef S.empty
344 Context { outputQueue = outQ
346 , runningThreads = running
347 , runqueue = runqueueFIFO ctx q
349 , queueEmpty = liftIO $ nullQ q
354getLifoCtx :: MonadIO m => CtxType -> IO (Context m a)
357 outQMv <- newEmptyMVar
358 running <- newIORef S.empty
360 let checkEmpty = liftIO (readIORef q) >>= return . null
362 Context { outputQueue = outQ
364 , runningThreads = running
365 , runqueue = runqueueLIFO ctx q
366 , enqueue = enqueueLIFO q
367 , queueEmpty = checkEmpty
372-- | Split the original computation in a pull-push pair. The original
373-- computation pulls from a Channel while m1 and m2 push to the channel.
374pushPairToCtx :: MonadAsync m
375 => CtxType -> Stream m a -> Stream m a -> m (Context m a)
376pushPairToCtx ctype m1 m2 = do
377 -- Note: We must have all the work on the queue before sending the
378 -- pushworker, otherwise the pushworker may exit before we even get a
383 c <- getFifoCtx ctype
384 (enqueue c) m1 >> (enqueue c) m2
387 c <- getLifoCtx ctype
388 (enqueue c) m2 >> (enqueue c) m1
393{-# NOINLINE pushPullFork #-}
394pushPullFork :: MonadAsync m
395 => CtxType -> Stream m a -> Stream m a -> Stream m a
396pushPullFork ct m1 m2 = Stream $ \_ stp yld -> do
397 ctx <- pushPairToCtx ct m1 m2
398 (runStream (pullFromCtx ctx)) Nothing stp yld
400-- Concurrency rate control. Our objective is to create more threads on
401-- demand if the consumer is running faster than us. As soon as we
402-- encounter an Alternative composition we create a push pull pair of
403-- threads. We use a channel for communication between the consumer that
404-- pulls from the channel and the producer that pushes to the channel. The
405-- producer creates more threads if the channel becomes empty at times,
406-- that is the consumer is running faster. However this mechanism can be
407-- problematic if the initial production latency is high, we may end up
408-- creating too many threads. So we need some way to monitor and use the
411-- TBD For quick response we may have to increase the rate in the middle of
412-- a serially running computation. For that we can use a state flag to fork
413-- the rest of the computation at any point of time inside the Monad bind
414-- operation if the consumer is running at a faster speed.
416-- TBD the alternative composition allows us to dispatch a chunkSize of only 1.
417-- If we have to dispatch in arbitrary chunksizes we will need to compose the
418-- parallel actions using a data constructor instead so that we can divide it
419-- in chunks of arbitrary size before dispatch. When batching we can convert
420-- the structure into Alternative batches of Monoid composition. That will also
421-- allow us to dispatch more work to existing threads rather than creating new
424-- TBD for pure work (when we are not in the IO monad) we can divide it into
425-- just the number of CPUs.
427-- XXX to rate control left folded structrues we will have to return the
428-- residual work back to the dispatcher. It will also consume a lot of
429-- memory due to queueing of all the work before execution starts.
431-- Note: This is designed to scale for right associated compositions,
432-- therefore always use a right fold for folding bigger structures.
433{-# INLINE parallel #-}
434parallel :: MonadAsync m => CtxType -> Stream m a -> Stream m a -> Stream m a
435parallel ct m1 m2 = Stream $ \ctx stp yld -> do
437 Nothing -> (runStream (pushPullFork ct m1 m2)) Nothing stp yld
438 Just c | ctxType c /= ct ->
439 (runStream (pushPullFork ct m1 m2)) Nothing stp yld
440 Just c -> liftIO ((enqueue c) m2) >> (runStream m1) ctx stp yld
442-- | The computation specified in the argument is pushed to a new thread and
443-- the context is returned. The returned context can be used to pull the output
444-- of the computation.
445pushOneToCtx :: MonadAsync m => CtxType -> Stream m a -> m (Context m a)
446pushOneToCtx ctype m = do
450 c <- getFifoCtx ctype
453 c <- getLifoCtx ctype
455 -- Note: We must have all the work on the queue before sending the
456 -- pushworker, otherwise the pushworker may exit before we even get a
458 liftIO $ (enqueue ctx) m
462------------------------------------------------------------------------------
463-- Semigroup and Monoid style compositions for parallel actions
464------------------------------------------------------------------------------
468parAhead :: Stream m a -> Stream m a -> Stream m a
471-- | Sequential composition similar to '<>' except that it can execute the
472-- action on the right in parallel ahead of time. Returns the results in
473-- sequential order like '<>' from left to right.
474(<>|) :: Stream m a -> Stream m a -> Stream m a
480parAlt :: MonadAsync m => Stream m a -> Stream m a -> Stream m a
481parAlt = parallel (CtxType Disjunction FIFO)
484{-# INLINE parLeft #-}
485parLeft :: MonadAsync m => Stream m a -> Stream m a -> Stream m a
486parLeft = parallel (CtxType Disjunction LIFO)
488-------------------------------------------------------------------------------
489-- Stream type is not exposed, these instances are only for deriving instances
490-- for the newtype wrappers based on Stream.
491-------------------------------------------------------------------------------
493-- Dummy Instances, defined to enable the definition of other instances that
494-- require a Monad constraint. Must be defined by the newtypes.
496instance Monad m => Functor (Stream m) where
499instance Monad m => Applicative (Stream m) where
503instance Monad m => Monad (Stream m) where
507------------------------------------------------------------------------------
508-- Alternative & MonadPlus
509------------------------------------------------------------------------------
511-- | `empty` represents an action that takes non-zero time to complete. Since
512-- all actions take non-zero time, an `Alternative` composition ('<|>') is a
513-- monoidal composition executing all actions in parallel, it is similar to
514-- '<>' except that it runs all the actions in parallel and interleaves their
516instance MonadAsync m => Alternative (Stream m) where
520instance MonadAsync m => MonadPlus (Stream m) where
524-------------------------------------------------------------------------------
526-------------------------------------------------------------------------------
528instance MonadTrans Stream where
529 lift mx = Stream $ \_ _ yld -> mx >>= (\a -> (yld a Nothing))
531instance (MonadBase b m, Monad m) => MonadBase b (Stream m) where
532 liftBase = liftBaseDefault
534------------------------------------------------------------------------------
535-- Standard transformer instances
536------------------------------------------------------------------------------
538instance MonadIO m => MonadIO (Stream m) where
539 liftIO = lift . liftIO
541instance MonadThrow m => MonadThrow (Stream m) where
542 throwM = lift . throwM
544-- XXX handle and test cross thread state transfer
545instance MonadError e m => MonadError e (Stream m) where
546 throwError = lift . throwError
547 catchError m h = Stream $ \ctx stp yld ->
548 let handle r = r `catchError` \e -> (runStream (h e)) ctx stp yld
549 yield a Nothing = yld a Nothing
550 yield a (Just r) = yld a (Just (catchError r h))
551 in handle $ (runStream m) ctx stp yield
553instance MonadReader r m => MonadReader r (Stream m) where
555 local f m = Stream $ \ctx stp yld ->
556 let yield a Nothing = local f $ yld a Nothing
557 yield a (Just r) = local f $ yld a (Just (local f r))
558 in (runStream m) ctx (local f stp) yield
560instance MonadState s m => MonadState s (Stream m) where
563 state k = lift (state k)
565------------------------------------------------------------------------------
566-- Types that can behave as a Stream
567------------------------------------------------------------------------------
569class Streaming t where
570 toStream :: t m a -> Stream m a
571 fromStream :: Stream m a -> t m a
573-- XXX The async API is useful for exploring each stream arbitrarily when
574-- zipping or merging two streams. We can use a newtype wrapper with a monad
575-- instance that composes like regular streaming libraries to facilitate linear
576-- composition. We will also need a yield API for that.
578-- | Run a computation asynchronously, triggers the computation and returns
579-- another computation (i.e. a promise) that when executed produces the output
580-- from the original computation. Note that the returned action must be
581-- executed exactly once and drained completely. If not executed or not drained
582-- fully we will may have a thread blocked forever and if executed more than
583-- once a ContextUsedAfterEOF exception will be raised.
585async :: (Streaming t, MonadAsync m) => t m a -> m (t m a)
587 ctx <- pushOneToCtx (CtxType Disjunction LIFO) (toStream m)
588 return $ fromStream $ pullFromCtx ctx
592-- | Sequential interleaved composition, similar to '<>' except that it fairly
593-- interleaves the two 'AsyncT' streams, yielding an element from each stream
594-- alternately. Should be used only on finite streams.
596(<=>) :: Streaming t => t m a -> t m a -> t m a
597m1 <=> m2 = fromStream $ interleave (toStream m1) (toStream m2)
599-- | Parallel composition similar to '<|>' except that it is left-biased
600-- instead of being fair. Action on the left is likely to be given a chance to
601-- execute before the action on the right. If the left action keeps yielding
602-- results without blocking it continues running until it finishes and only
603-- then the right action runs. Unlike '<|>' it can be used on infinite streams.
605(<|) :: (Streaming t, MonadAsync m) => t m a -> t m a -> t m a
606m1 <| m2 = fromStream $ parLeft (toStream m1) (toStream m2)
608------------------------------------------------------------------------------
610------------------------------------------------------------------------------
612-- | Run an 'AsyncT' computation, wait for it to finish and discard the
614runStreaming :: (Monad m, Streaming t) => t m a -> m ()
615runStreaming m = go (toStream m)
619 go m1 = (runStream m1) Nothing stop yield
624 yield _ Nothing = stop
625 yield _ (Just x) = go x
627-- | Collect the results of an 'AsyncT' stream into a list.
628{-# INLINABLE toList #-}
629toList :: (Monad m, Streaming t) => t m a -> m [a]
630toList m = go (toStream m)
634 go m1 = (runStream m1) Nothing stop yield
639 yield a Nothing = return [a]
640 yield a (Just x) = liftM (a :) (go x)
642------------------------------------------------------------------------------
644------------------------------------------------------------------------------
646-- | Fold a 'Foldable' container using the given function.
647{-# INLINABLE foldWith #-}
648foldWith :: (Monoid b, Foldable t) => (a -> b -> b) -> t a -> b
649foldWith f = foldr f mempty
651-- | Fold a 'Foldable' container using a function that is a composition of the
653{-# INLINABLE foldMapWith #-}
654foldMapWith :: (Monoid b, Foldable t) =>
655 (b1 -> b -> b) -> (a -> b1) -> t a -> b
656foldMapWith f g = foldr (f . g) mempty
658-- | Fold a 'Foldable' container using a function that is a composition of the
659-- first and the third argument.
660{-# INLINABLE forEachWith #-}
661forEachWith :: (Monoid b, Foldable t) =>
662 (b1 -> b -> b) -> t a -> (a -> b1) -> b
663forEachWith f xs g = foldr (f . g) mempty xs
665------------------------------------------------------------------------------
666-- Convert a callback into an 'AsyncT' computation
667------------------------------------------------------------------------------
669-- | Convert a callback into an 'AsyncT' stream.
670fromCallback :: (forall r. (a -> m r) -> m r) -> Stream m a
671fromCallback k = Stream $ \_ _ yld -> k (\a -> yld a Nothing)
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