@@ -22,18 +22,8 @@
2222--
2323module Asyncly.AsyncT
2424 ( AsyncT (..)
25− , MonadAsync
26− , ChildEvent(..)
27− , Context(..)
28− , initContext
29− , Location(..)
30− , getLocation
31− , setLocation
3225
3326 , threads
34− , waitForChildren
35− , handleResult
36−
3727 , async
3828 , makeAsync
3929 , each
@@ -51,138 +41,26 @@ module Asyncly.AsyncT
5141where
5242
5343import Control.Applicative (Alternative (..))
54−import Control.Concurrent (ThreadId, forkIO, killThread,
55− myThreadId)
56−import Control.Concurrent.STM (TChan, atomically, newTChan,
57− readTChan, tryReadTChan,
58− writeTChan)
59−import Control.Exception (fromException,
60− SomeException (..))
61−import qualified Control.Exception.Lifted as EL
6244import Control.Monad (ap, mzero, when)
6345import Control.Monad.Base (MonadBase (..), liftBaseDefault)
6446import Control.Monad.Catch (MonadThrow, throwM)
6547import Control.Monad.State (MonadIO (..), MonadPlus (..),
6648 StateT (..), liftM, runStateT,
67− get, gets, modify)
49+ gets, modify)
6850import Control.Monad.Trans.Class (MonadTrans (lift))
6951import Control.Monad.Trans.Control (ComposeSt, MonadBaseControl (..),
7052 MonadTransControl (..),
7153 defaultLiftBaseWith,
7254 defaultRestoreM, liftBaseWith)
73−import Data.Atomics (atomicModifyIORefCAS)
74−import Data.IORef (IORef, modifyIORef, newIORef,
75− readIORef, writeIORef)
76−import Data.List (delete)
77−import Data.Maybe (isJust)
78−import GHC.Prim (Any)
7955import Unsafe.Coerce (unsafeCoerce)
8056
81−import Control.Monad.Trans.Recorder (MonadRecorder(..), Paused(..),
82− Recording)
57+import Control.Monad.Trans.Recorder (MonadRecorder(..))
58+import Data.IORef (newIORef, readIORef, writeIORef)
8359--import Debug.Trace (traceM)
8460
85−newtype AsyncT m a = AsyncT { runAsyncT :: StateT Context m (Maybe a) }
86−
87−------------------------------------------------------------------------------
88−-- Parent child thread communication types
89−------------------------------------------------------------------------------
90−
91−data ChildEvent = ChildDone ThreadId (Maybe SomeException)
92−
93−------------------------------------------------------------------------------
94−-- State threaded around the monad
95−------------------------------------------------------------------------------
96−
97−data Context = Context
98− {
99− ---------------------------------------------------------------------------
100− -- Execution state
101− ---------------------------------------------------------------------------
102−
103− -- a -> AsyncT m b
104− continuations :: [Any]
105−
106− -- When we suspend we save the logs in this IORef and exit.
107− , logsRef :: Maybe (IORef [Recording])
108− -- XXX this functionality can be in a separate layer?
109− , location :: Location
110−
111− ---------------------------------------------------------------------------
112− -- Thread creation, communication and cleanup
113− ---------------------------------------------------------------------------
114− -- When a new thread is created the parent records it in the
115− -- 'pendingThreads' field and 'threadCredit' is decremented. When a child
116− -- thread is done it sends a done event to the parent on an unbounded
117− -- channel and goes away. Before starting a new thread the parent always
118− -- processes the unbounded channel to clear up the pending zombies. This
119− -- strategy ensures that the zombie queue will never grow more than the
120− -- number of running threads. Before exiting, the parent thread waits on
121− -- the channel until all its children are cleaned up.
122− ---------------------------------------------------------------------------
123−
124− -- XXX setup a cleanup computation to run rather than passing all these
125− -- params.
126−
127− , childChannel :: TChan ChildEvent
128− -- ^ A channel for the immediate children to communicate to us when they
129− -- die. Each thread has its own dedicated channel for its children
130−
131− -- We always track the child threads, otherwise the programmer needs to
132− -- worry about if some threads may remain hanging because they are stuck in
133− -- an infinite loop or a forever blocked IO. Also, it is not clear whether
134− -- there is a significant gain by disabling tracking. Instead of using
135− -- threadIds we can also keep a count instead, that will allow us to wait
136− -- for all children to drain but we cannot kill them.
137− --
138− -- We need these as IORefs since we need any changes to these to be
139− -- reflected back in the calling code that waits for the children.
140−
141− , pendingThreads :: IORef [ThreadId]
142− -- ^ Active immediate child threads spawned by this thread, modified only
143− -- by this thread, therefore atomic modification is not needed.
144−
145− -- By default this limit is shared by the current tree of threads. But at
146− -- any point the 'threads' primitive can be used to start a new limit for
147− -- the computation under that primitive.
148− --
149− -- Shared across many threads, therefore atomic access is required.
150− -- This is incremented by the child thread itelf before it exits.
151− --
152− , threadCredit :: IORef Int
153− -- ^ How many more threads are allowed to be created?
154− }
155−
156−initContext
157− :: TChan ChildEvent
158− -> IORef [ThreadId]
159− -> IORef Int
160− -> (a -> AsyncT m a)
161− -> Maybe (IORef [Recording])
162− -> Context
163−initContext childChan pending credit finalizer lref =
164− Context { continuations = [unsafeCoerce finalizer]
165− , logsRef = lref
166− , location = Worker
167− , childChannel = childChan
168− , pendingThreads = pending
169− , threadCredit = credit }
61+import Asyncly.Threads
17062
171−------------------------------------------------------------------------------
172−-- Where is the computation running?
173−------------------------------------------------------------------------------
174−
175−-- RemoteNode is used in the Alternative instance. We stop the computation when
176−-- the location is RemoteNode, we do not execute any alternatives.
177−--
178−data Location = Worker | WaitingParent | RemoteNode
179− deriving (Eq, Show)
180−
181−getLocation :: Monad m => StateT Context m Location
182−getLocation = gets location
183−
184−setLocation :: Monad m => Location -> StateT Context m ()
185−setLocation loc = modify $ \Context { .. } -> Context { location = loc, .. }
63+newtype AsyncT m a = AsyncT { runAsyncT :: StateT Context m (Maybe a) }
18664
18765------------------------------------------------------------------------------
18866-- Utilities
@@ -194,8 +72,6 @@ setLocation loc = modify $ \Context { .. } -> Context { location = loc, .. }
19472dbg :: Monad m => a -> m ()
19573dbg _ = return ()
19674
197−type MonadAsync m = (MonadIO m, MonadBaseControl IO m, MonadThrow m)
198−
19975------------------------------------------------------------------------------
20076-- Monad
20177------------------------------------------------------------------------------
@@ -233,259 +109,6 @@ instance MonadAsync m => Applicative (AsyncT m) where
233109 pure = return
234110 (<*>) = ap
235111
236−------------------------------------------------------------------------------
237−-- Pick up from where we left in the previous thread
238−------------------------------------------------------------------------------
239−
240−{-# SPECIALIZE continue :: [Any] -> a -> StateT Context IO (Maybe a) #-}
241−continue :: Monad m => [Any] -> a -> StateT Context m (Maybe a)
242−continue [] x = do
243− return (Just x)
244−
245−continue [f] x = do
246− modify $ \ctx -> ctx {continuations = []}
247− runAsyncT $ (unsafeCoerce f) x
248−
249−continue (f:fs) x = do
250− modify $ \ctx -> ctx {continuations = fs}
251− y <- runAsyncT $ (unsafeCoerce f) x
252− case y of
253− Nothing -> return y
254− Just z -> continue fs z
255−
256−{-# SPECIALIZE runContext :: Context -> AsyncT IO a -> StateT Context IO () #-}
257−runContext :: MonadAsync m => Context -> AsyncT m a -> StateT Context m ()
258−runContext ctx action = do
259− _ <- lift $ runStateT m ctx
260− return ()
261−
262− where
263−
264− m = do
265− let conts = continuations ctx
266− x <- runAsyncT action
267− case x of
268− Just y -> continue conts y >> return ()
269− Nothing -> return ()
270−
271−------------------------------------------------------------------------------
272−-- Thread Management (creation, reaping and killing)
273−------------------------------------------------------------------------------
274−
275−-- XXX We are using unbounded channels so this will not block on writing to
276−-- pchan. We can use bounded channels to throttle the creation of threads based
277−-- on consumption rate.
278−processOneEvent :: MonadIO m
279− => ChildEvent
280− -> [ThreadId]
281− -> m ([ThreadId], Maybe SomeException)
282−processOneEvent (ChildDone tid e) pending = do
283− when (isJust e) $ liftIO $ mapM_ killThread pending
284− return (delete tid pending, Nothing)
285−
286−drainChildren :: MonadIO m
287− => TChan ChildEvent
288− -> [ThreadId]
289− -> m ([ThreadId], Maybe SomeException)
290−drainChildren cchan pending =
291− case pending of
292− [] -> return (pending, Nothing)
293− _ -> do
294− ev <- liftIO $ atomically $ readTChan cchan
295− (p, e) <- processOneEvent ev pending
296− maybe (drainChildren cchan p) (const $ return (p, e)) e
297−
298−waitForChildren :: MonadIO m => Context -> m (Maybe SomeException)
299−waitForChildren ctx = do
300− let pendingRef = pendingThreads ctx
301− pending <- liftIO $ readIORef pendingRef
302− (p, e) <- drainChildren (childChannel ctx) pending
303− liftIO $ writeIORef pendingRef p
304− return e
305−
306−tryReclaimZombies :: (MonadIO m, MonadThrow m) => Context -> m ()
307−tryReclaimZombies ctx = do
308− let cchan = childChannel ctx
309− pendingRef = pendingThreads ctx
310−
311− pending <- liftIO $ readIORef pendingRef
312− case pending of
313− [] -> return ()
314− _ -> do
315− mev <- liftIO $ atomically $ tryReadTChan cchan
316− case mev of
317− Nothing -> return ()
318− Just ev -> do
319− (p, e) <- processOneEvent ev pending
320− liftIO $ writeIORef pendingRef p
321− maybe (return ()) throwM e
322− tryReclaimZombies ctx
323−
324−{-# NOINLINE waitForOneEvent #-}
325−waitForOneEvent :: (MonadIO m, MonadThrow m) => Context -> m ()
326−waitForOneEvent ctx = do
327− -- XXX assert pending must have at least one element
328− -- assert that the tid is found in our list
329− let cchan = childChannel ctx
330− pendingRef = pendingThreads ctx
331−
332− ev <- liftIO $ atomically $ readTChan cchan
333− pending <- liftIO $ readIORef pendingRef
334− (p, e) <- processOneEvent ev pending
335− liftIO $ writeIORef pendingRef p
336− maybe (return ()) throwM e
337−
338−-- XXX this is not a real semaphore as it does not really block on wait,
339−-- instead it returns whether the value is zero or non-zero.
340−--
341−{-# INLINE waitQSemB #-}
342−waitQSemB :: IORef Int -> IO Bool
343−waitQSemB sem = atomicModifyIORefCAS sem $ \n ->
344− if n > 0
345− then (n - 1, True)
346− else (n, False)
347−
348−signalQSemB :: IORef Int -> IO ()
349−signalQSemB sem = atomicModifyIORefCAS sem $ \n -> (n + 1, ())
350−
351−-- Allocation of threads
352−--
353−-- global thread limit
354−-- thread fan-out i.e. per thread children limit
355−-- min per thread allocation to avoid starvation
356−--
357−-- dynamic adjustment based on the cost, speed of consumption, cpu utilization
358−-- etc. We need to adjust the limits based on cost, throughput and latencies.
359−--
360−-- The event producer thread must put the work on a work-queue and the child
361−-- threads can pick it up from there. But if there is just one consumer then it
362−-- may not make sense to have a separate producer unless the producing cost is
363−-- high.
364−--
365−
366−{-# INLINE handleResult #-}
367−handleResult :: MonadIO m
368− => Context -> Either SomeException a -> m (Maybe SomeException)
369−handleResult ctx res =
370− case res of
371− Left e -> do
372− case fromException e of
373− Just (Paused recording) ->
374− maybe (passException e)
375− (handleRecording recording) (logsRef ctx)
376− Nothing -> passException e
377− Right _ -> waitForChildren ctx
378−
379− where
380−
381− handleRecording recording ref = do
382− liftIO $ atomicModifyIORefCAS ref $ \logs -> (recording : logs, ())
383− waitForChildren ctx
384−
385− passException e = do
386− liftIO $ readIORef (pendingThreads ctx) >>= mapM_ killThread
387− return (Just e)
388−
389−forkFinally1 :: MonadAsync m
390− => Context
391− -> AsyncT m a
392− -> (Either SomeException () -> IO ())
393− -> StateT Context m ThreadId
394−forkFinally1 ctx action preExit =
395− EL.mask $ \restore ->
396− liftBaseWith $ \runInIO -> forkIO $ do
397− _ <- runInIO $ EL.try (restore (runContext ctx action))
398− >>= liftIO . preExit
399− -- XXX restore state here
400− return ()
401−
402−-- | Run a given context in a new thread.
403−--
404−forkContext :: (MonadBaseControl IO m, MonadIO m, MonadThrow m)
405− => AsyncT m a -> StateT Context m ()
406−forkContext action = do
407− parentCtx <- get
408− childCtx <- childContext parentCtx
409− tid <- forkFinally1 childCtx action
410− (beforeExit childCtx (childChannel parentCtx))
411− updatePendingThreads parentCtx tid
412−
413− where
414−
415− updatePendingThreads :: (MonadIO m, MonadThrow m)
416− => Context -> ThreadId -> m ()
417− updatePendingThreads ctx tid = do
418− -- update the new thread before reclaiming zombies so that if it exited
419− -- already reclaim finds it in the list and does not panic.
420− liftIO $ modifyIORef (pendingThreads ctx) $ (\ts -> tid:ts)
421− tryReclaimZombies ctx
422−
423− childContext ctx = do
424− pendingRef <- liftIO $ newIORef []
425− chan <- liftIO $ atomically newTChan
426− -- shares the threadCredit of the parent by default
427− return $ ctx
428− { pendingThreads = pendingRef
429− , childChannel = chan
430− }
431−
432− beforeExit ctx pchan res = do
433− r <- handleResult ctx res
434− signalQSemB (threadCredit ctx)
435− tid <- myThreadId
436− liftIO $ atomically $ writeTChan pchan (ChildDone tid r)
437−
438−-- | Decide whether to resume the context in the same thread or a new thread
439−--
440−canFork :: Context -> IO Bool
441−canFork context = do
442− gotCredit <- waitQSemB (threadCredit context)
443− case gotCredit of
444− False -> do
445− pending <- readIORef $ pendingThreads context
446− case pending of
447− [] -> return False
448− _ -> do
449− -- XXX If we have unreclaimable child threads e.g.
450− -- infinite loop, this is going to deadlock us. We need
451− -- special handling for those cases. Add those to
452− -- unreclaimable list? And always execute them in an
453− -- async thread, cannot use sync for those.
454− --
455− waitForOneEvent context
456− canFork context
457− True -> return True
458−
459−-- Housekeeping, invoked after spawning of all child tasks is done and the
460−-- parent task needs to terminate. Either the task is fully done or we handed
461−-- it over to another thread, in any case the current thread is done.
462−
463−spawningParentDone :: MonadIO m => StateT Context m (Maybe a)
464−spawningParentDone = do
465− loc <- getLocation
466− when (loc /= RemoteNode) $ setLocation WaitingParent
467− return Nothing
468−
469−-- | The task may be run in the same thread or in a new thread depending on the
470−-- forceAsync parameter and the current thread quota.
471−--
472−{-# SPECIALIZE runAsyncTask :: Bool -> AsyncT IO a -> StateT Context IO (Maybe a) #-}
473−{-# INLINE runAsyncTask #-}
474−runAsyncTask :: MonadAsync m
475− => Bool
476− -> AsyncT m a
477− -> StateT Context m (Maybe a)
478−runAsyncTask forceAsync action = do
479− if forceAsync
480− then forkContext action
481− else do
482− context <- get
483− can <- liftIO $ canFork context
484− case can of
485− False -> runContext context action
486− True -> forkContext action
487− return Nothing
488−
489112-- The current model is to start a new thread for every task. The input is
490113-- provided at the time of the creation and therefore no synchronization is
491114-- needed compared to a pool of threads contending to get the input from a
@@ -521,29 +144,10 @@ runAsyncTask forceAsync action = do
521144-- 'async' is not useful as the last action as the last action always runs in
522145-- the current thread.
523146async :: MonadAsync m => AsyncT m a -> AsyncT m a
524−async action = AsyncT $ runAsyncTask True action
147+async action = AsyncT $ runAsyncTask True (runAsyncT action)
525148
526−-- XXX After setting the callback we should wait for the callback otherwise we
527−-- will just go away. We should use the same finalization mechanism that we use
528−-- in fork.
529−--
530−-- | Makes an asyncly action from a callback setter function; can be used to
531−-- convert existing callback style code into asyncly style code. The first
532−-- parameter is a callback setter function. When we are called back,
533−-- 'makeAsync' behaves as if it was an async computation that just returned a
534−-- value of type 'a'. After the computation is done the result of the action
535−-- in second parameter is returned to the callback.
536−--
537149makeAsync :: MonadAsync m => ((a -> m ()) -> m ()) -> AsyncT m a
538−makeAsync cbsetter = AsyncT $ do
539− -- XXX should we force fork a thread or keep running in the context of the
540− -- callback?
541− ctx <- get
542− lift $ cbsetter $ \a -> do
543− let s = runAsyncTask False (return a)
544− _ <- runStateT s ctx
545− return ()
546− spawningParentDone
150+makeAsync = AsyncT . makeCont
547151
548152------------------------------------------------------------------------------
549153-- Alternative
@@ -553,7 +157,7 @@ instance MonadAsync m => Alternative (AsyncT m) where
553157 {-# SPECIALIZE instance Alternative (AsyncT IO) #-}
554158 empty = AsyncT $ return Nothing
555159 (<|>) m1 m2 = AsyncT $ do
556− r <- runAsyncTask False m1
160+ r <- runAsyncTask False (runAsyncT m1)
557161 maybe (runAsyncT m2) (return . Just) r
558162
559163instance MonadAsync m => MonadPlus (AsyncT m) where