@@ -1,449 +0,0 @@
1−{-# LANGUAGE ConstraintKinds #-}
2−{-# LANGUAGE FlexibleContexts #-}
3−{-# LANGUAGE RecordWildCards #-}
4−
5−-- |
6−-- Module : Asyncly.Threads
7−-- Copyright : (c) 2017 Harendra Kumar
8−--
9−-- License : MIT-style
10−-- Maintainer : harendra.kumar@gmail.com
11−-- Stability : experimental
12−-- Portability : GHC
13−--
14−module Asyncly.Threads
15− ( -- ChildEvent(..)
16− MonadAsync
17− , Context(..)
18− , initContext
19− --, runAsyncTask
20− --, makeCont
21− , threadCtl
22− -- , Location(..)
23− -- , getLocation
24− -- , setLocation
25−
26− -- , waitForChildren
27− --, handleResult
28− )
29−where
30−
31−import Control.Concurrent (ThreadId, forkIO, killThread,
32− myThreadId)
33−import Control.Concurrent.STM (TChan, atomically, newTChan,
34− readTChan, tryReadTChan,
35− writeTChan)
36−import Control.Exception (fromException,
37− SomeException (..))
38−import qualified Control.Exception.Lifted as EL
39−import Control.Monad (when)
40−import Control.Monad.Catch (MonadThrow, throwM)
41−import Control.Monad.State (MonadIO (..),
42− StateT (..), runStateT,
43− get, gets, modify)
44−import Control.Monad.Trans.Class (MonadTrans (lift))
45−import Control.Monad.Trans.Control (MonadBaseControl, liftBaseWith)
46−import Data.Atomics (atomicModifyIORefCAS)
47−import Data.IORef (IORef, modifyIORef, newIORef,
48− readIORef, writeIORef)
49−import Data.List (delete)
50−import Data.Maybe (isJust)
51−import GHC.Prim (Any)
52−import Unsafe.Coerce (unsafeCoerce)
53−
54−import Control.Monad.Trans.Recorder (Paused (..), Recording)
55−
56−type MonadAsync m = (MonadIO m, MonadBaseControl IO m, MonadThrow m)
57−
58−------------------------------------------------------------------------------
59−-- Parent child thread communication types
60−------------------------------------------------------------------------------
61−
62−data ChildEvent = ChildDone ThreadId (Maybe SomeException)
63−
64−------------------------------------------------------------------------------
65−-- State threaded around the monad
66−------------------------------------------------------------------------------
67−
68−data Context = Context
69− {
70− ---------------------------------------------------------------------------
71− -- Execution state
72− ---------------------------------------------------------------------------
73−
74− -- When we suspend we save the logs in this IORef and exit.
75− logsRef :: Maybe (IORef [Recording])
76− -- XXX this functionality can be in a separate layer?
77− , location :: Location
78−
79− ---------------------------------------------------------------------------
80− -- Thread creation, communication and cleanup
81− ---------------------------------------------------------------------------
82− -- When a new thread is created the parent records it in the
83− -- 'pendingThreads' field and 'threadCredit' is decremented. When a child
84− -- thread is done it sends a done event to the parent on an unbounded
85− -- channel and goes away. Before starting a new thread the parent always
86− -- processes the unbounded channel to clear up the pending zombies. This
87− -- strategy ensures that the zombie queue will never grow more than the
88− -- number of running threads. Before exiting, the parent thread waits on
89− -- the channel until all its children are cleaned up.
90− ---------------------------------------------------------------------------
91−
92− -- XXX setup a cleanup computation to run rather than passing all these
93− -- params.
94−
95− , childChannel :: TChan ChildEvent
96− -- ^ A channel for the immediate children to communicate to us when they
97− -- die. Each thread has its own dedicated channel for its children
98−
99− -- We always track the child threads, otherwise the programmer needs to
100− -- worry about if some threads may remain hanging because they are stuck in
101− -- an infinite loop or a forever blocked IO. Also, it is not clear whether
102− -- there is a significant gain by disabling tracking. Instead of using
103− -- threadIds we can also keep a count instead, that will allow us to wait
104− -- for all children to drain but we cannot kill them.
105− --
106− -- We need these as IORefs since we need any changes to these to be
107− -- reflected back in the calling code that waits for the children.
108−
109− , pendingThreads :: IORef [ThreadId]
110− -- ^ Active immediate child threads spawned by this thread, modified only
111− -- by this thread, therefore atomic modification is not needed.
112−
113− -- By default this limit is shared by the current tree of threads. But at
114− -- any point the 'threads' primitive can be used to start a new limit for
115− -- the computation under that primitive.
116− --
117− -- Shared across many threads, therefore atomic access is required.
118− -- This is incremented by the child thread itelf before it exits.
119− --
120− , threadCredit :: IORef Int
121− -- ^ How many more threads are allowed to be created?
122− }
123−
124−initContext
125− :: TChan ChildEvent
126− -> IORef [ThreadId]
127− -> IORef Int
128− -> Maybe (IORef [Recording])
129− -> Context
130−initContext childChan pending credit lref =
131− Context { logsRef = lref
132− , location = Worker
133− , childChannel = childChan
134− , pendingThreads = pending
135− , threadCredit = credit }
136−
137−------------------------------------------------------------------------------
138−-- Where is the computation running?
139−------------------------------------------------------------------------------
140−
141−-- RemoteNode is used in the Alternative instance. We stop the computation when
142−-- the location is RemoteNode, we do not execute any alternatives.
143−--
144−data Location = Worker | WaitingParent | RemoteNode
145− deriving (Eq, Show)
146−
147−getLocation :: Monad m => StateT Context m Location
148−getLocation = gets location
149−
150−setLocation :: Monad m => Location -> StateT Context m ()
151−setLocation loc = modify $ \Context { .. } -> Context { location = loc, .. }
152−
153−{-
154−------------------------------------------------------------------------------
155−-- Pick up from where we left in the previous thread
156−------------------------------------------------------------------------------
157−
158−{-# SPECIALIZE continue :: [Any] -> a -> StateT Context IO (Maybe a) #-}
159−continue :: Monad m => [Any] -> a -> StateT Context m (Maybe a)
160−continue [] x = do
161− return (Just x)
162−
163−continue [f] x = do
164− modify $ \ctx -> ctx {continuations = []}
165− (unsafeCoerce f) x
166−
167−continue (f:fs) x = do
168− modify $ \ctx -> ctx {continuations = fs}
169− y <- (unsafeCoerce f) x
170− case y of
171− Nothing -> return y
172− Just z -> continue fs z
173−
174−{-# SPECIALIZE runContext :: Context -> StateT Context IO (Maybe a) -> StateT Context IO () #-}
175−runContext :: MonadAsync m => Context -> StateT Context m (Maybe a) -> StateT Context m ()
176−runContext ctx action = do
177− _ <- lift $ runStateT m ctx
178− return ()
179−
180− where
181−
182− m = do
183− let conts = continuations ctx
184− x <- action
185− case x of
186− Just y -> continue conts y >> return ()
187− Nothing -> return ()
188−
189−------------------------------------------------------------------------------
190−-- Thread Management (creation, reaping and killing)
191−------------------------------------------------------------------------------
192−
193−-- XXX We are using unbounded channels so this will not block on writing to
194−-- pchan. We can use bounded channels to throttle the creation of threads based
195−-- on consumption rate.
196−processOneEvent :: MonadIO m
197− => ChildEvent
198− -> [ThreadId]
199− -> m ([ThreadId], Maybe SomeException)
200−processOneEvent (ChildDone tid e) pending = do
201− when (isJust e) $ liftIO $ mapM_ killThread pending
202− return (delete tid pending, Nothing)
203−
204−drainChildren :: MonadIO m
205− => TChan ChildEvent
206− -> [ThreadId]
207− -> m ([ThreadId], Maybe SomeException)
208−drainChildren cchan pending =
209− case pending of
210− [] -> return (pending, Nothing)
211− _ -> do
212− ev <- liftIO $ atomically $ readTChan cchan
213− (p, e) <- processOneEvent ev pending
214− maybe (drainChildren cchan p) (const $ return (p, e)) e
215−
216−waitForChildren :: MonadIO m => Context -> m (Maybe SomeException)
217−waitForChildren ctx = do
218− let pendingRef = pendingThreads ctx
219− pending <- liftIO $ readIORef pendingRef
220− (p, e) <- drainChildren (childChannel ctx) pending
221− liftIO $ writeIORef pendingRef p
222− return e
223−
224−tryReclaimZombies :: (MonadIO m, MonadThrow m) => Context -> m ()
225−tryReclaimZombies ctx = do
226− let cchan = childChannel ctx
227− pendingRef = pendingThreads ctx
228−
229− pending <- liftIO $ readIORef pendingRef
230− case pending of
231− [] -> return ()
232− _ -> do
233− mev <- liftIO $ atomically $ tryReadTChan cchan
234− case mev of
235− Nothing -> return ()
236− Just ev -> do
237− (p, e) <- processOneEvent ev pending
238− liftIO $ writeIORef pendingRef p
239− maybe (return ()) throwM e
240− tryReclaimZombies ctx
241−
242−{-# NOINLINE waitForOneEvent #-}
243−waitForOneEvent :: (MonadIO m, MonadThrow m) => Context -> m ()
244−waitForOneEvent ctx = do
245− -- XXX assert pending must have at least one element
246− -- assert that the tid is found in our list
247− let cchan = childChannel ctx
248− pendingRef = pendingThreads ctx
249−
250− ev <- liftIO $ atomically $ readTChan cchan
251− pending <- liftIO $ readIORef pendingRef
252− (p, e) <- processOneEvent ev pending
253− liftIO $ writeIORef pendingRef p
254− maybe (return ()) throwM e
255−
256−-- XXX this is not a real semaphore as it does not really block on wait,
257−-- instead it returns whether the value is zero or non-zero.
258−--
259−{-# INLINE waitQSemB #-}
260−waitQSemB :: IORef Int -> IO Bool
261−waitQSemB sem = atomicModifyIORefCAS sem $ \n ->
262− if n > 0
263− then (n - 1, True)
264− else (n, False)
265−
266−signalQSemB :: IORef Int -> IO ()
267−signalQSemB sem = atomicModifyIORefCAS sem $ \n -> (n + 1, ())
268−
269−-- Allocation of threads
270−--
271−-- global thread limit
272−-- thread fan-out i.e. per thread children limit
273−-- min per thread allocation to avoid starvation
274−--
275−-- dynamic adjustment based on the cost, speed of consumption, cpu utilization
276−-- etc. We need to adjust the limits based on cost, throughput and latencies.
277−--
278−-- The event producer thread must put the work on a work-queue and the child
279−-- threads can pick it up from there. But if there is just one consumer then it
280−-- may not make sense to have a separate producer unless the producing cost is
281−-- high.
282−--
283−
284−{-# INLINE handleResult #-}
285−handleResult :: MonadIO m
286− => Context -> Either SomeException a -> m (Maybe SomeException)
287−handleResult ctx res =
288− case res of
289− Left e -> do
290− case fromException e of
291− Just (Paused recording) ->
292− maybe (passException e)
293− (handleRecording recording) (logsRef ctx)
294− Nothing -> passException e
295− Right _ -> waitForChildren ctx
296−
297− where
298−
299− handleRecording recording ref = do
300− liftIO $ atomicModifyIORefCAS ref $ \logs -> (recording : logs, ())
301− waitForChildren ctx
302−
303− passException e = do
304− liftIO $ readIORef (pendingThreads ctx) >>= mapM_ killThread
305− return (Just e)
306−
307−forkFinally1 :: MonadAsync m
308− => Context
309− -> StateT Context m (Maybe a)
310− -> (Either SomeException () -> IO ())
311− -> StateT Context m ThreadId
312−forkFinally1 ctx action preExit =
313− EL.mask $ \restore ->
314− liftBaseWith $ \runInIO -> forkIO $ do
315− _ <- runInIO $ EL.try (restore (runContext ctx action))
316− >>= liftIO . preExit
317− -- XXX restore state here
318− return ()
319−
320−-- | Run a given context in a new thread.
321−--
322−forkContext :: (MonadBaseControl IO m, MonadIO m, MonadThrow m)
323− => StateT Context m (Maybe a) -> StateT Context m ()
324−forkContext action = do
325− parentCtx <- get
326− childCtx <- childContext parentCtx
327− tid <- forkFinally1 childCtx action
328− (beforeExit childCtx (childChannel parentCtx))
329− updatePendingThreads parentCtx tid
330−
331− where
332−
333− updatePendingThreads :: (MonadIO m, MonadThrow m)
334− => Context -> ThreadId -> m ()
335− updatePendingThreads ctx tid = do
336− -- update the new thread before reclaiming zombies so that if it exited
337− -- already reclaim finds it in the list and does not panic.
338− liftIO $ modifyIORef (pendingThreads ctx) $ (\ts -> tid:ts)
339− tryReclaimZombies ctx
340−
341− childContext ctx = do
342− pendingRef <- liftIO $ newIORef []
343− chan <- liftIO $ atomically newTChan
344− -- shares the threadCredit of the parent by default
345− return $ ctx
346− { pendingThreads = pendingRef
347− , childChannel = chan
348− }
349−
350− beforeExit ctx pchan res = do
351− r <- handleResult ctx res
352− signalQSemB (threadCredit ctx)
353− tid <- myThreadId
354− liftIO $ atomically $ writeTChan pchan (ChildDone tid r)
355−
356−-- | Decide whether to resume the context in the same thread or a new thread
357−--
358−canFork :: Context -> IO Bool
359−canFork context = do
360− gotCredit <- waitQSemB (threadCredit context)
361− case gotCredit of
362− False -> do
363− pending <- readIORef $ pendingThreads context
364− case pending of
365− [] -> return False
366− _ -> do
367− -- XXX If we have unreclaimable child threads e.g.
368− -- infinite loop, this is going to deadlock us. We need
369− -- special handling for those cases. Add those to
370− -- unreclaimable list? And always execute them in an
371− -- async thread, cannot use sync for those.
372− --
373− waitForOneEvent context
374− canFork context
375− True -> return True
376−
377−-- Housekeeping, invoked after spawning of all child tasks is done and the
378−-- parent task needs to terminate. Either the task is fully done or we handed
379−-- it over to another thread, in any case the current thread is done.
380−
381−spawningParentDone :: MonadIO m => StateT Context m (Maybe a)
382−spawningParentDone = do
383− loc <- getLocation
384− when (loc /= RemoteNode) $ setLocation WaitingParent
385− return Nothing
386−
387−-- | The task may be run in the same thread or in a new thread depending on the
388−-- forceAsync parameter and the current thread quota.
389−--
390−{-# SPECIALIZE runAsyncTask :: Bool -> StateT Context IO (Maybe a)
391− -> StateT Context IO (Maybe a) #-}
392−{-# INLINE runAsyncTask #-}
393−runAsyncTask :: MonadAsync m
394− => Bool
395− -> StateT Context m (Maybe a)
396− -> StateT Context m (Maybe a)
397−runAsyncTask forceAsync action = do
398− if forceAsync
399− then forkContext action
400− else do
401− context <- get
402− can <- liftIO $ canFork context
403− case can of
404− False -> runContext context action
405− True -> forkContext action
406− return Nothing
407−
408−-- XXX After setting the callback we should wait for the callback otherwise we
409−-- will just go away. We should use the same finalization mechanism that we use
410−-- in fork.
411−--
412−-- | Makes an asyncly action from a callback setter function; can be used to
413−-- convert existing callback style code into asyncly style code. The first
414−-- parameter is a callback setter function. When we are called back,
415−-- 'makeAsync' behaves as if it was an async computation that just returned a
416−-- value of type 'a'. After the computation is done the result of the action
417−-- in second parameter is returned to the callback.
418−--
419−makeCont :: MonadAsync m => ((a -> m ()) -> m ()) -> StateT Context m (Maybe a)
420−makeCont cbsetter = do
421− -- XXX should we force fork a thread or keep running in the context of the
422− -- callback?
423− ctx <- get
424− lift $ cbsetter $ \a -> do
425− let s = runAsyncTask False (return (Just a))
426− _ <- runStateT s ctx
427− return ()
428− spawningParentDone
429−
430−-}
431−------------------------------------------------------------------------------
432−-- Controlling thread quota
433−------------------------------------------------------------------------------
434−
435−-- XXX Should n be Word32 instead?
436−-- | Runs a computation under a given thread limit. A limit of 0 means all new
437−-- tasks start synchronously in the current thread unless overridden by
438−-- 'async'.
439−threadCtl :: MonadAsync m
440− => Int
441− -> StateT Context m a
442− -> StateT Context m a
443−threadCtl n action = do
444− oldCr <- gets threadCredit
445− newCr <- liftIO $ newIORef n
446− modify $ \s -> s { threadCredit = newCr }
447− r <- action
448− modify $ \s -> s { threadCredit = oldCr }
449− return r