1{-# LANGUAGE ConstraintKinds #-}
2{-# LANGUAGE ExistentialQuantification #-}
3{-# LANGUAGE FlexibleContexts #-}
4{-# LANGUAGE FlexibleInstances #-}
5{-# LANGUAGE InstanceSigs #-}
6{-# LANGUAGE MultiParamTypeClasses #-}
7{-# LANGUAGE Rank2Types #-}
8{-# LANGUAGE RecordWildCards #-}
9{-# LANGUAGE ScopedTypeVariables #-}
10{-# LANGUAGE TypeFamilies #-}
11{-# LANGUAGE UndecidableInstances #-}
14-- Module : Asyncly.AsyncT
15-- Copyright : (c) 2017 Harendra Kumar
18-- Maintainer : harendra.kumar@gmail.com
19-- Stability : experimental
52import Control.Applicative (Alternative (..))
53import Control.Concurrent (ThreadId, forkIO, killThread,
55import Control.Concurrent.STM (TChan, atomically, newTChan,
56 readTChan, tryReadTChan,
58import Control.Exception (fromException,
60import qualified Control.Exception.Lifted as EL
61import Control.Monad (ap, mzero, when)
62import Control.Monad.Base (MonadBase (..), liftBaseDefault)
63import Control.Monad.Catch (MonadThrow, throwM)
64import Control.Monad.State (MonadIO (..), MonadPlus (..),
65 StateT (..), liftM, runStateT,
67import Control.Monad.Trans.Class (MonadTrans (lift))
68import Control.Monad.Trans.Control (ComposeSt, MonadBaseControl (..),
69 MonadTransControl (..),
71 defaultRestoreM, liftBaseWith)
72import Data.IORef (IORef, atomicModifyIORef,
73 modifyIORef, newIORef, readIORef,
75import Data.List (delete)
76import Data.Maybe (isJust)
78import Unsafe.Coerce (unsafeCoerce)
80import Control.Monad.Trans.Recorder (MonadRecorder(..), Paused(..),
82--import Debug.Trace (traceM)
84newtype AsyncT m a = AsyncT { runAsyncT :: StateT Context m (Maybe a) }
86------------------------------------------------------------------------------
87-- Parent child thread communication types
88------------------------------------------------------------------------------
90data ChildEvent = ChildDone ThreadId (Maybe SomeException)
92------------------------------------------------------------------------------
93-- State threaded around the monad
94------------------------------------------------------------------------------
98 ---------------------------------------------------------------------------
100 ---------------------------------------------------------------------------
105 -- When we suspend we save the logs in this IORef and exit.
106 , logsRef :: Maybe (IORef [Recording])
107 -- XXX this functionality can be in a separate layer?
108 , location :: Location
110 ---------------------------------------------------------------------------
111 -- Thread creation, communication and cleanup
112 ---------------------------------------------------------------------------
113 -- When a new thread is created the parent records it in the
114 -- 'pendingThreads' field and 'threadCredit' is decremented. When a child
115 -- thread is done it sends a done event to the parent on an unbounded
116 -- channel and goes away. Before starting a new thread the parent always
117 -- processes the unbounded channel to clear up the pending zombies. This
118 -- strategy ensures that the zombie queue will never grow more than the
119 -- number of running threads. Before exiting, the parent thread waits on
120 -- the channel until all its children are cleaned up.
121 ---------------------------------------------------------------------------
123 -- XXX setup a cleanup computation to run rather than passing all these
126 , childChannel :: TChan ChildEvent
127 -- ^ A channel for the immediate children to communicate to us when they
128 -- die. Each thread has its own dedicated channel for its children
130 -- We always track the child threads, otherwise the programmer needs to
131 -- worry about if some threads may remain hanging because they are stuck in
132 -- an infinite loop or a forever blocked IO. Also, it is not clear whether
133 -- there is a significant gain by disabling tracking. Instead of using
134 -- threadIds we can also keep a count instead, that will allow us to wait
135 -- for all children to drain but we cannot kill them.
137 -- We need these as IORefs since we need any changes to these to be
138 -- reflected back in the calling code that waits for the children.
140 , pendingThreads :: IORef [ThreadId]
141 -- ^ Active immediate child threads spawned by this thread, modified only
142 -- by this thread, therefore atomic modification is not needed.
144 -- By default this limit is shared by the current tree of threads. But at
145 -- any point the 'threads' primitive can be used to start a new limit for
146 -- the computation under that primitive.
148 -- Shared across many threads, therefore atomic access is required.
149 -- This is incremented by the child thread itelf before it exits.
151 , threadCredit :: IORef Int
152 -- ^ How many more threads are allowed to be created?
160 -> Maybe (IORef [Recording])
162initContext childChan pending credit finalizer lref =
163 Context { continuation = unsafeCoerce finalizer
166 , childChannel = childChan
167 , pendingThreads = pending
168 , threadCredit = credit }
170------------------------------------------------------------------------------
171-- Where is the computation running?
172------------------------------------------------------------------------------
174-- RemoteNode is used in the Alternative instance. We stop the computation when
175-- the location is RemoteNode, we do not execute any alternatives.
177data Location = Worker | WaitingParent | RemoteNode
180getLocation :: Monad m => StateT Context m Location
181getLocation = gets location
183setLocation :: Monad m => Location -> StateT Context m ()
184setLocation loc = modify $ \Context { .. } -> Context { location = loc, .. }
186------------------------------------------------------------------------------
188------------------------------------------------------------------------------
190-- Toggle the dbg definition for debug traces
191--dbg :: Monad m => String -> m ()
193dbg :: Monad m => a -> m ()
196type MonadAsync m = (MonadIO m, MonadBaseControl IO m, MonadThrow m)
198------------------------------------------------------------------------------
199-- Pick up from where we left in the previous thread
200------------------------------------------------------------------------------
202-- | Run the continuation of an action
203runContext :: MonadAsync m => Context -> AsyncT m a -> StateT Context m ()
204runContext ctx action = do
205 let s = runAsyncT $ action >>= unsafeCoerce (continuation ctx)
206 _ <- lift $ runStateT s ctx
209------------------------------------------------------------------------------
210-- Thread Management (creation, reaping and killing)
211------------------------------------------------------------------------------
213-- XXX We are using unbounded channels so this will not block on writing to
214-- pchan. We can use bounded channels to throttle the creation of threads based
215-- on consumption rate.
216processOneEvent :: MonadIO m
219 -> m ([ThreadId], Maybe SomeException)
220processOneEvent (ChildDone tid e) pending = do
221 when (isJust e) $ liftIO $ mapM_ killThread pending
222 return (delete tid pending, Nothing)
224drainChildren :: MonadIO m
227 -> m ([ThreadId], Maybe SomeException)
228drainChildren cchan pending =
230 [] -> return (pending, Nothing)
232 ev <- liftIO $ atomically $ readTChan cchan
233 (p, e) <- processOneEvent ev pending
234 maybe (drainChildren cchan p) (const $ return (p, e)) e
236waitForChildren :: MonadIO m => Context -> m (Maybe SomeException)
237waitForChildren ctx = do
238 let pendingRef = pendingThreads ctx
239 pending <- liftIO $ readIORef pendingRef
240 (p, e) <- drainChildren (childChannel ctx) pending
241 liftIO $ writeIORef pendingRef p
244tryReclaimZombies :: (MonadIO m, MonadThrow m) => Context -> m ()
245tryReclaimZombies ctx = do
246 let cchan = childChannel ctx
247 pendingRef = pendingThreads ctx
249 pending <- liftIO $ readIORef pendingRef
253 mev <- liftIO $ atomically $ tryReadTChan cchan
257 (p, e) <- processOneEvent ev pending
258 liftIO $ writeIORef pendingRef p
259 maybe (return ()) throwM e
260 tryReclaimZombies ctx
262waitForOneEvent :: (MonadIO m, MonadThrow m) => Context -> m ()
263waitForOneEvent ctx = do
264 -- XXX assert pending must have at least one element
265 -- assert that the tid is found in our list
266 let cchan = childChannel ctx
267 pendingRef = pendingThreads ctx
269 ev <- liftIO $ atomically $ readTChan cchan
270 pending <- liftIO $ readIORef pendingRef
271 (p, e) <- processOneEvent ev pending
272 liftIO $ writeIORef pendingRef p
273 maybe (return ()) throwM e
275-- XXX this is not a real semaphore as it does not really block on wait,
276-- instead it returns whether the value is zero or non-zero.
278waitQSemB :: IORef Int -> IO Bool
279waitQSemB sem = atomicModifyIORef sem $ \n ->
284signalQSemB :: IORef Int -> IO ()
285signalQSemB sem = atomicModifyIORef sem $ \n -> (n + 1, ())
287-- Allocation of threads
289-- global thread limit
290-- thread fan-out i.e. per thread children limit
291-- min per thread allocation to avoid starvation
293-- dynamic adjustment based on the cost, speed of consumption, cpu utilization
294-- etc. We need to adjust the limits based on cost, throughput and latencies.
296-- The event producer thread must put the work on a work-queue and the child
297-- threads can pick it up from there. But if there is just one consumer then it
298-- may not make sense to have a separate producer unless the producing cost is
302handleResult :: MonadIO m
303 => Context -> Either SomeException a -> m (Maybe SomeException)
304handleResult ctx res =
307 case fromException e of
308 Just (Paused recording) ->
309 maybe (passException e)
310 (handleRecording recording) (logsRef ctx)
311 Nothing -> passException e
312 Right _ -> waitForChildren ctx
316 handleRecording recording ref = do
317 liftIO $ atomicModifyIORef ref $ \logs -> (recording : logs, ())
321 liftIO $ readIORef (pendingThreads ctx) >>= mapM_ killThread
324forkFinally1 :: MonadAsync m
327 -> (Either SomeException () -> IO ())
328 -> StateT Context m ThreadId
329forkFinally1 ctx action preExit =
330 EL.mask $ \restore ->
331 liftBaseWith $ \runInIO -> forkIO $ do
332 _ <- runInIO $ EL.try (restore (runContext ctx action))
334 -- XXX restore state here
337-- | Run a given context in a new thread.
339forkContext :: (MonadBaseControl IO m, MonadIO m, MonadThrow m)
340 => AsyncT m a -> StateT Context m ()
341forkContext action = do
343 childCtx <- childContext parentCtx
344 tid <- forkFinally1 childCtx action
345 (beforeExit childCtx (childChannel parentCtx))
346 updatePendingThreads parentCtx tid
350 updatePendingThreads :: (MonadIO m, MonadThrow m)
351 => Context -> ThreadId -> m ()
352 updatePendingThreads ctx tid = do
353 -- update the new thread before reclaiming zombies so that if it exited
354 -- already reclaim finds it in the list and does not panic.
355 liftIO $ modifyIORef (pendingThreads ctx) $ (\ts -> tid:ts)
356 tryReclaimZombies ctx
358 childContext ctx = do
359 pendingRef <- liftIO $ newIORef []
360 chan <- liftIO $ atomically newTChan
361 -- shares the threadCredit of the parent by default
363 { pendingThreads = pendingRef
364 , childChannel = chan
367 beforeExit ctx pchan res = do
368 r <- handleResult ctx res
369 signalQSemB (threadCredit ctx)
371 liftIO $ atomically $ writeTChan pchan (ChildDone tid r)
373-- | Decide whether to resume the context in the same thread or a new thread
375canFork :: Context -> IO Bool
377 gotCredit <- liftIO $ waitQSemB (threadCredit context)
380 pending <- liftIO $ readIORef $ pendingThreads context
384 -- XXX If we have unreclaimable child threads e.g.
385 -- infinite loop, this is going to deadlock us. We need
386 -- special handling for those cases. Add those to
387 -- unreclaimable list? And always execute them in an
388 -- async thread, cannot use sync for those.
390 waitForOneEvent context
394-- Housekeeping, invoked after spawning of all child tasks is done and the
395-- parent task needs to terminate. Either the task is fully done or we handed
396-- it over to another thread, in any case the current thread is done.
398spawningParentDone :: MonadIO m => StateT Context m (Maybe a)
399spawningParentDone = do
401 when (loc /= RemoteNode) $ setLocation WaitingParent
404-- | The task may be run in the same thread or in a new thread depending on the
405-- forceAsync parameter and the current thread quota.
407runAsyncTask :: MonadAsync m
410 -> StateT Context m (Maybe a)
411runAsyncTask forceAsync action = do
413 then forkContext action
416 can <- liftIO $ canFork context
418 False -> runContext context action
419 True -> forkContext action
422-- The current model is to start a new thread for every task. The input is
423-- provided at the time of the creation and therefore no synchronization is
424-- needed compared to a pool of threads contending to get the input from a
425-- channel. However the thread creation overhead may be more than the
426-- synchronization cost?
428-- When the task is over the outputs need to be collected and that requires
429-- synchronization irrespective of a thread pool model or per task new thread
432-- XXX instead of starting a new thread every time, reuse the existing child
433-- threads and send them work via a shared channel. When there is no more work
434-- available we need a way to close the channel and wakeup all waiters so that
435-- they can go away rather than waiting indefinitely.
438-- Only those actions that are marked with 'async' are guaranteed to be
439-- asynchronous. Asyncly is free to run other actions synchronously or
440-- asynchronously and it should not matter to the semantics of the program, if
441-- it does then use async to force.
443-- Why not make async as default and ask the programmer to use a 'sync'
444-- primitive to force an action to run synchronously? But then we would not
445-- have the freedom to convert the async actions to sync dynamically. Note that
446-- it is safe to convert a sync action to async but vice-versa is not true.
447-- Converting an async to sync can cause change in semantics if the async
448-- action was an infinite loop for example.
450-- | In an 'Alternative' composition, force the action to run asynchronously.
451-- The @\<|\>@ operator implies "can be parallel", whereas 'async' implies
452-- "must be parallel". Note that outside an 'Alternative' composition 'async'
453-- is not useful and should not be used. Even in an 'Alternative' composition
454-- 'async' is not useful as the last action as the last action always runs in
455-- the current thread.
456async :: MonadAsync m => AsyncT m a -> AsyncT m a
457async action = AsyncT $ runAsyncTask True action
459-- XXX After setting the callback we should wait for the callback otherwise we
460-- will just go away. We should use the same finalization mechanism that we use
463-- | Makes an asyncly action from a callback setter function; can be used to
464-- convert existing callback style code into asyncly style code. The first
465-- parameter is a callback setter function. When we are called back,
466-- 'makeAsync' behaves as if it was an async computation that just returned a
467-- value of type 'a'. After the computation is done the result of the action
468-- in second parameter is returned to the callback.
470makeAsync :: MonadAsync m => ((a -> m ()) -> m ()) -> AsyncT m a
471makeAsync cbsetter = AsyncT $ do
472 -- XXX should we force fork a thread or keep running in the context of the
475 lift $ cbsetter $ \a -> do
476 let s = runAsyncTask False (return a)
482each :: (MonadIO m, MonadBaseControl IO m, MonadThrow m)
484each xs = foldl (<|>) empty $ map return xs
486------------------------------------------------------------------------------
487-- Controlling thread quota
488------------------------------------------------------------------------------
490-- XXX Should n be Word32 instead?
491-- | Runs a computation under a given thread limit. A limit of 0 means all new
492-- tasks start synchronously in the current thread unless overridden by
494threads :: MonadAsync m => Int -> AsyncT m a -> AsyncT m a
495threads n process = AsyncT $ do
496 oldCr <- gets threadCredit
497 newCr <- liftIO $ newIORef n
498 modify $ \s -> s { threadCredit = newCr }
499 r <- runAsyncT $ process
501 modify $ \s -> s { threadCredit = oldCr }
503 ) -- restore old credit
506------------------------------------------------------------------------------
508------------------------------------------------------------------------------
510instance MonadAsync m => Functor (AsyncT m) where
513------------------------------------------------------------------------------
515------------------------------------------------------------------------------
518instance MonadAsync m => Applicative (AsyncT m) where
519 pure a = AsyncT . return $ Just a
522------------------------------------------------------------------------------
524------------------------------------------------------------------------------
526instance MonadAsync m => Alternative (AsyncT m) where
527 empty = AsyncT $ return Nothing
528 (<|>) m1 m2 = AsyncT $ do
529 r <- runAsyncTask False m1
532 RemoteNode -> return Nothing
533 _ -> maybe (runAsyncT m2) (return . Just) r
535------------------------------------------------------------------------------
537------------------------------------------------------------------------------
539instance MonadAsync m => Monad (AsyncT m) where
541 m >>= f = AsyncT $ do
542 modify $ \Context { continuation = g, .. } ->
543 let k x = f x >>= unsafeCoerce g
544 in Context { continuation = unsafeCoerce k, .. }
547 Context { continuation = g } <- get
548 runAsyncT $ maybe mzero (unsafeCoerce g) x
550instance MonadAsync m => MonadPlus (AsyncT m) where
554instance (Monoid a, MonadAsync m) => Monoid (AsyncT m a) where
555 mappend x y = mappend <$> x <*> y
556 mempty = return mempty
558------------------------------------------------------------------------------
560------------------------------------------------------------------------------
562instance (Num a, Monad (AsyncT m)) => Num (AsyncT m a) where
563 fromInteger = return . fromInteger
564 mf + mg = (+) <$> mf <*> mg
565 mf * mg = (*) <$> mf <*> mg
566 negate f = f >>= return . negate
567 abs f = f >>= return . abs
568 signum f = f >>= return . signum
570------------------------------------------------------------------------------
571-- Special compositions
572------------------------------------------------------------------------------
574infixr 1 >>*, *>>, >>|
576-- XXX This can be moved to utility functions as it is purely app level
577-- | Run @m b@ right after the first event in @m a@ is generated but before it
578-- is yielded. The result of @m b@ is discarded.
579afterfirst :: MonadAsync m => AsyncT m a -> AsyncT m b -> AsyncT m a
581 ref <- liftIO $ newIORef False
583 done <- liftIO $ readIORef ref
584 when (not done) $ (liftIO $ writeIORef ref True) >>* mb
587-- | Same as 'afterfirst'.
588(>>|) :: MonadAsync m => AsyncT m a -> AsyncT m b -> AsyncT m a
591-- | Run @m a@ before running @m b@. The result of @m a@ is discarded.
592before :: MonadAsync m => AsyncT m a -> AsyncT m b -> AsyncT m b
593before ma mb = AsyncT $ do
594 _ <- runAsyncT (ma >> mzero)
597-- | Same as 'before'.
598(*>>) :: MonadAsync m => AsyncT m a -> AsyncT m b -> AsyncT m b
601-- | Run @m b@ after running @m a@. The result of @m b@ is discarded.
602thereafter :: MonadAsync m => AsyncT m a -> AsyncT m b -> AsyncT m a
603thereafter ma mb = AsyncT $ do
605 _ <- runAsyncT (mb >> mzero)
608-- | Same as 'thereafter'.
609(>>*) :: MonadAsync m => AsyncT m a -> AsyncT m b -> AsyncT m a
612-------------------------------------------------------------------------------
614-------------------------------------------------------------------------------
616instance MonadTrans AsyncT where
617 lift mx = AsyncT $ lift mx >>= return . Just
619-------------------------------------------------------------------------------
621-------------------------------------------------------------------------------
623instance MonadTransControl AsyncT where
624 type StT AsyncT a = (Maybe a, Context)
625 liftWith f = AsyncT $ StateT $ \s ->
626 liftM (\x -> (Just x, s))
627 (f $ \t -> runStateT (runAsyncT t) s)
628 restoreT = AsyncT . StateT . const
629 {-# INLINABLE liftWith #-}
630 {-# INLINABLE restoreT #-}
632instance (MonadBase b m, MonadAsync m) => MonadBase b (AsyncT m) where
633 liftBase = liftBaseDefault
635instance (MonadBaseControl b m, MonadAsync m) => MonadBaseControl b (AsyncT m) where
636 type StM (AsyncT m) a = ComposeSt AsyncT m a
637 liftBaseWith = defaultLiftBaseWith
638 restoreM = defaultRestoreM
639 {-# INLINABLE liftBaseWith #-}
640 {-# INLINABLE restoreM #-}
642------------------------------------------------------------------------------
643-- Standard transformer instances
644------------------------------------------------------------------------------
646instance MonadAsync m => MonadIO (AsyncT m) where
647 liftIO = lift . liftIO
649instance MonadAsync m => MonadThrow (AsyncT m) where
650 throwM = lift . throwM
652------------------------------------------------------------------------------
654------------------------------------------------------------------------------
656instance (MonadAsync m, MonadRecorder m) => MonadRecorder (AsyncT m) where
657 getJournal = lift getJournal
658 putJournal = lift . putJournal
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