@@ -1,16 +1,9 @@
1−{-# LANGUAGE ConstraintKinds #-}
21{-# LANGUAGE FlexibleContexts #-}
32{-# LANGUAGE ScopedTypeVariables #-}
43
54module Asyncly.Threads
6− ( MonadAsync
7− , wait
5+ ( wait
86 , wait_
9− , threads
10−
11− , async
12− , makeAsync
13− , each
147 , gather
158
169 , waitLogged
@@ -22,349 +15,25 @@ module Asyncly.Threads
2215 )
2316where
2417
25−import Control.Applicative ((<|>), empty)
26−import Control.Concurrent (ThreadId, forkIO, killThread,
27− myThreadId)
28−import Control.Concurrent.STM (TChan, atomically, newTChan,
29− readTChan, tryReadTChan,
30− writeTChan)
18+import Control.Concurrent (killThread)
19+import Control.Concurrent.STM (atomically, newTChan)
3120import Control.Exception (SomeException (..))
32−import qualified Control.Exception.Lifted as EL
33−import Control.Monad.Catch (MonadCatch, MonadThrow, throwM,
34− try)
21+import Control.Monad.Catch (MonadCatch, throwM, try)
3522import Control.Monad.IO.Class (MonadIO (..))
36−import Control.Monad.State (get, gets, modify, mzero, put,
37− runStateT, when, StateT)
23+import Control.Monad.State (get, mzero, put, runStateT)
3824import Control.Monad.Trans.Class (MonadTrans (lift))
39−import Control.Monad.Trans.Control (MonadBaseControl, liftBaseWith)
4025import Data.IORef (IORef, atomicModifyIORef,
41− modifyIORef, newIORef, readIORef,
42− writeIORef)
43−import Data.List (delete)
44−import Data.Maybe (isJust)
45−import Unsafe.Coerce (unsafeCoerce)
26+ newIORef, readIORef)
4627
4728import Asyncly.AsyncT
4829
49−type MonadAsync m = (MonadIO m, MonadBaseControl IO m, MonadThrow m)
50−
51−------------------------------------------------------------------------------
52−-- Model of computation
53−------------------------------------------------------------------------------
54−
55−-- A computation starts in a top level thread. If no "forking primitives" are
56−-- used then the thread finishes in a straight line flow just like the IO
57−-- monad. However, if a "forking primitive" is used it "forks" the computation
58−-- into multiple flows. The "forked" computations may run concurrently (and
59−-- therefore in parallel when possible) or serially. When multiple forking
60−-- primitives are composed it results into a tree of computations where each
61−-- branch of the tree can run concurrently.
62−--
63−-- A forking primitive may create multiple forks at that point, each fork
64−-- provides a specific input value to be used in the forked computation, this
65−-- value defines the fork.
66−--
67−-- The final result of the computation is the collection of all the values
68−-- generated by all the leaf level forks. These values are then propagated up
69−-- the tree and collected at the root of the tree.
70−--
71−-- Since AsyncT is a transformer we can use things like pipe, conduit or any
72−-- other transformer monads inside the computations to utilize single threaded
73−-- composition or data flow techniques.
74−--
75−------------------------------------------------------------------------------
76−-- Pick up from where we left in the previous thread
77−------------------------------------------------------------------------------
78−
79−-- | Continue execution of the closure that we were executing when we migrated
80−-- to a new thread.
81−
82−runContext :: Monad m => Context -> AsyncT m a -> StateT Context m ()
83−runContext ctx action = do
84− let s = runAsyncT $ action >>= unsafeCoerce (continuation ctx)
85− _ <- lift $ runStateT s ctx
86− return ()
87−
88−------------------------------------------------------------------------------
89−-- Thread Management (creation, reaping and killing)
90−------------------------------------------------------------------------------
91−
92−-- XXX We are using unbounded channels so this will not block on writing to
93−-- pchan. We can use bounded channels to throttle the creation of threads based
94−-- on consumption rate.
95−processOneEvent :: MonadIO m
96− => ChildEvent
97− -> [ThreadId]
98− -> m ([ThreadId], Maybe SomeException)
99−processOneEvent (ChildDone tid e) pending = do
100− when (isJust e) $ liftIO $ mapM_ killThread pending
101− return (delete tid pending, Nothing)
102−
103−drainChildren :: MonadIO m
104− => TChan ChildEvent
105− -> [ThreadId]
106− -> m ([ThreadId], Maybe SomeException)
107−drainChildren cchan pending =
108− case pending of
109− [] -> return (pending, Nothing)
110− _ -> do
111− ev <- liftIO $ atomically $ readTChan cchan
112− (p, e) <- processOneEvent ev pending
113− maybe (drainChildren cchan p) (const $ return (p, e)) e
114−
115−waitForChildren :: MonadIO m => Context -> m (Maybe SomeException)
116−waitForChildren ctx = do
117− let pendingRef = pendingThreads ctx
118− pending <- liftIO $ readIORef pendingRef
119− (p, e) <- drainChildren (childChannel ctx) pending
120− liftIO $ writeIORef pendingRef p
121− return e
122−
123−tryReclaimZombies :: (MonadIO m, MonadThrow m) => Context -> m ()
124−tryReclaimZombies ctx = do
125− let cchan = childChannel ctx
126− pendingRef = pendingThreads ctx
127−
128− pending <- liftIO $ readIORef pendingRef
129− case pending of
130− [] -> return ()
131− _ -> do
132− mev <- liftIO $ atomically $ tryReadTChan cchan
133− case mev of
134− Nothing -> return ()
135− Just ev -> do
136− (p, e) <- processOneEvent ev pending
137− liftIO $ writeIORef pendingRef p
138− maybe (return ()) throwM e
139− tryReclaimZombies ctx
140−
141−waitForOneEvent :: (MonadIO m, MonadThrow m) => Context -> m ()
142−waitForOneEvent ctx = do
143− -- XXX assert pending must have at least one element
144− -- assert that the tid is found in our list
145− let cchan = childChannel ctx
146− pendingRef = pendingThreads ctx
147−
148− ev <- liftIO $ atomically $ readTChan cchan
149− pending <- liftIO $ readIORef pendingRef
150− (p, e) <- processOneEvent ev pending
151− liftIO $ writeIORef pendingRef p
152− maybe (return ()) throwM e
153−
154−-- XXX this is not a real semaphore as it does not really block on wait,
155−-- instead it returns whether the value is zero or non-zero.
156−--
157−waitQSemB :: IORef Int -> IO Bool
158−waitQSemB sem = atomicModifyIORef sem $ \n ->
159− if n > 0
160− then (n - 1, True)
161− else (n, False)
162−
163−signalQSemB :: IORef Int -> IO ()
164−signalQSemB sem = atomicModifyIORef sem $ \n -> (n + 1, ())
165−
166−-- Allocation of threads
167−--
168−-- global thread limit
169−-- thread fan-out i.e. per thread children limit
170−-- min per thread allocation to avoid starvation
171−--
172−-- dynamic adjustment based on the cost, speed of consumption, cpu utilization
173−-- etc. We need to adjust the limits based on cost, throughput and latencies.
174−--
175−-- The event producer thread must put the work on a work-queue and the child
176−-- threads can pick it up from there. But if there is just one consumer then it
177−-- may not make sense to have a separate producer unless the producing cost is
178−-- high.
179−--
180−
181−forkFinally1 :: (MonadIO m, MonadBaseControl IO m)
182− => Context
183− -> AsyncT m a
184− -> (Either SomeException () -> IO ())
185− -> StateT Context m ThreadId
186−forkFinally1 ctx action preExit =
187− EL.mask $ \restore ->
188− liftBaseWith $ \runInIO -> forkIO $ do
189− _ <- runInIO $ EL.try (restore (runContext ctx action))
190− >>= liftIO . preExit
191− -- XXX restore state here
192− return ()
193−
194−-- | Run a given context in a new thread.
195−--
196−forkContext :: (MonadBaseControl IO m, MonadIO m, MonadThrow m)
197− => AsyncT m a -> StateT Context m ()
198−forkContext action = do
199− parentCtx <- get
200− childCtx <- childContext parentCtx
201− tid <- forkFinally1 childCtx action
202− (beforeExit childCtx (childChannel parentCtx))
203− updatePendingThreads parentCtx tid
204−
205− where
206−
207− updatePendingThreads :: (MonadIO m, MonadThrow m)
208− => Context -> ThreadId -> m ()
209− updatePendingThreads ctx tid = do
210− -- update the new thread before reclaiming zombies so that if it exited
211− -- already reclaim finds it in the list and does not panic.
212− liftIO $ modifyIORef (pendingThreads ctx) $ (\ts -> tid:ts)
213− tryReclaimZombies ctx
214−
215− childContext ctx = do
216− pendingRef <- liftIO $ newIORef []
217− chan <- liftIO $ atomically newTChan
218− -- shares the threadCredit of the parent by default
219− return $ ctx
220− { pendingThreads = pendingRef
221− , childChannel = chan
222− }
223−
224− beforeExit ctx pchan res = do
225− tid <- myThreadId
226− r <- case res of
227− Left e -> do
228− dbg $ "beforeExit: " ++ show tid ++ " caught exception"
229− liftIO $ readIORef (pendingThreads ctx) >>= mapM_ killThread
230− return (Just e)
231− Right _ -> waitForChildren ctx
232−
233− signalQSemB (threadCredit ctx)
234− liftIO $ atomically $ writeTChan pchan (ChildDone tid r)
235−
236−-- | Decide whether to resume the context in the same thread or a new thread
237−--
238−canFork :: Context -> IO Bool
239−canFork context = do
240− gotCredit <- liftIO $ waitQSemB (threadCredit context)
241− case gotCredit of
242− False -> do
243− pending <- liftIO $ readIORef $ pendingThreads context
244− case pending of
245− [] -> return False
246− _ -> do
247− -- XXX If we have unreclaimable child threads e.g.
248− -- infinite loop, this is going to deadlock us. We need
249− -- special handling for those cases. Add those to
250− -- unreclaimable list? And always execute them in an
251− -- async thread, cannot use sync for those.
252− --
253− waitForOneEvent context
254− canFork context
255− True -> return True
256−
257−-- | Resume a captured context with a given action. The context may be resumed
258−-- in the same thread or in a new thread depending on the synch parameter and
259−-- the current thread quota.
260−--
261−resumeContextWith :: MonadAsync m
262− => Bool -- force synchronous
263− -> AsyncT m a
264− -> StateT Context m ()
265−resumeContextWith synch action = do
266− context <- get
267− can <- liftIO $ canFork context
268− case can && (not synch) of
269− False -> runContext context action
270− True -> forkContext action
271−
272−-- The current model is to start a new thread for every task. The input is
273−-- provided at the time of the creation and therefore no synchronization is
274−-- needed compared to a pool of threads contending to get the input from a
275−-- channel. However the thread creation overhead may be more than the
276−-- synchronization cost?
277−--
278−-- When the task is over the outputs need to be collected and that requires
279−-- synchronization irrespective of a thread pool model or per task new thread
280−-- model.
281−--
282−-- XXX instead of starting a new thread every time, reuse the existing child
283−-- threads and send them work via a shared channel. When there is no more work
284−-- available we need a way to close the channel and wakeup all waiters so that
285−-- they can go away rather than waiting indefinitely.
286−--
287−
288−-- Housekeeping, invoked after spawning of all child tasks is done and the
289−-- parent task needs to terminate. Either the task is fully done or we handed
290−-- it over to another thread, in any case the current thread is done.
291−
292−spawningParentDone :: MonadIO m => StateT Context m (Maybe a)
293−spawningParentDone = do
294− loc <- getLocation
295− when (loc /= RemoteNode) $ setLocation WaitingParent
296− return Nothing
297−
298−-- If a new thread cannot be created then the computation is run in the same
299−-- thread, but the functional behavior remains the same.
300−
301−async :: MonadAsync m => AsyncT m a -> AsyncT m a
302−async action = AsyncT $ do
303− resumeContextWith False action
304− spawningParentDone
305−
306−-- | Makes an asyncly action from a callback setter function; can be used to
307−-- convert existing callback style code into asyncly style code. The first
308−-- parameter is a callback setter function. When we are called back,
309−-- 'makeAsync' behaves as if it was an async computation that just returned a
310−-- value of type 'a'. After the computation is done the result of the action
311−-- in second parameter is returned to the callback.
312−--
313−makeAsync :: MonadAsync m => ((a -> m ()) -> m ()) -> AsyncT m a
314−makeAsync cbsetter = AsyncT $ do
315− -- XXX should we force fork a thread or keep running the context of the
316− -- callback?
317− ctx <- get
318− lift $ cbsetter $ \a -> do
319− let s = resumeContextWith False (return a)
320− _ <- runStateT s ctx
321− return ()
322− spawningParentDone
323−
324−-- scatter
325−each :: (MonadIO m, MonadBaseControl IO m, MonadThrow m)
326− => [a] -> AsyncT m a
327−each xs = foldl (<|>) empty $ map (async . return) xs
328−
329−------------------------------------------------------------------------------
330−-- Controlling thread quota
331−------------------------------------------------------------------------------
332−
333−-- XXX Should n be Word32 instead?
334−-- | Runs a computation under a given thread limit. A limit of 0 means new
335−-- tasks start synchronously in the current thread. New threads are created by
336−-- 'parallel', and APIs that use parallel.
337−threads :: MonadIO m => Int -> AsyncT m a -> AsyncT m a
338−threads n process = AsyncT $ do
339− oldCr <- gets threadCredit
340− newCr <- liftIO $ newIORef n
341− modify $ \s -> s { threadCredit = newCr }
342− r <- runAsyncT $ process
343− <** (AsyncT $ do
344− modify $ \s -> s { threadCredit = oldCr }
345− return (Just ())
346− ) -- restore old credit
347− return r
348−
349−{-
350−-- This can be used to set, increase or decrease the existing limit. The limit
351−-- is shared by multiple threads and therefore needs to modified atomically.
352−-- Note that when there is no limit the limit is set to maxBound it can
353−-- overflow with an increment and get reduced instead of increasing.
354−-- XXX should we use a Maybe instead? Or use separate inc/dec/set APIs to
355−-- handle overflow properly?
356−--
357−-- modifyThreads :: MonadIO m => (Int -> Int) -> AsyncT m ()
358−-- modifyThreads f =
359−-}
360−
36130------------------------------------------------------------------------------
36231-- Running the monad
36332------------------------------------------------------------------------------
36433
36534-- | Run an 'AsyncT m' computation and collect the results generated by each
36635-- thread of the computation in a list.
367−waitAsync :: (MonadIO m, MonadCatch m)
36+waitAsync :: (MonadAsync m, MonadCatch m)
36837 => (a -> AsyncT m a) -> Maybe (IORef [Log]) -> AsyncT m a -> m ()
36938waitAsync finalizer lref m = do
37039 childChan <- liftIO $ atomically newTChan
@@ -389,12 +58,12 @@ waitAsync finalizer lref m = do
38958
39059-- | Invoked to store the result of the computation in the context and finish
39160-- the computation when the computation is done
392−gatherResult :: MonadIO m => IORef [a] -> a -> AsyncT m a
61+gatherResult :: MonadAsync m => IORef [a] -> a -> AsyncT m a
39362gatherResult ref r = do
39463 liftIO $ atomicModifyIORef ref $ \rs -> (r : rs, ())
39564 mzero
39665
397−gather :: (MonadIO m, MonadCatch m) => AsyncT m a -> AsyncT m [a]
66+gather :: (MonadAsync m, MonadCatch m) => AsyncT m a -> AsyncT m [a]
39867gather m = AsyncT $ do
39968 resultsRef <- liftIO $ newIORef []
40069 lift $ waitAsync (gatherResult resultsRef) Nothing m
@@ -403,7 +72,7 @@ gather m = AsyncT $ do
40372
40473-- | Run an 'AsyncT m' computation and collect the results generated by each
40574-- thread of the computation in a list.
406−wait :: (MonadIO m, MonadCatch m) => AsyncT m a -> m [a]
75+wait :: (MonadAsync m, MonadCatch m) => AsyncT m a -> m [a]
40776wait m = do
40877 resultsRef <- liftIO $ newIORef []
40978 waitAsync (gatherResult resultsRef) Nothing m
@@ -411,14 +80,14 @@ wait m = do
41180
41281-- | Run an 'AsyncT m' computation, wait for it to finish and discard the
41382-- results.
414−wait_ :: (MonadIO m, MonadCatch m) => AsyncT m a -> m ()
83+wait_ :: (MonadAsync m, MonadCatch m) => AsyncT m a -> m ()
41584wait_ m = waitAsync (const mzero) Nothing m
41685
41786------------------------------------------------------------------------------
41887-- Logging
41988------------------------------------------------------------------------------
42089
421−logged :: (Loggable a, MonadIO m) => AsyncT m a -> AsyncT m a
90+logged :: (Loggable a, MonadAsync m) => AsyncT m a -> AsyncT m a
42291logged m = AsyncT $ do
42392 ctx <- get
42493 case journal ctx of
@@ -467,7 +136,7 @@ logged m = AsyncT $ do
467136-- completed threads output is returned and for suspended threads the logs to
468137-- resume them are returned.
469138
470−suspend :: MonadIO m => AsyncT m ()
139+suspend :: MonadAsync m => AsyncT m ()
471140suspend = logged $ AsyncT $ do
472141 ctx <- get
473142 case logsRef ctx of
@@ -492,13 +161,12 @@ withLog m (Log entries) = AsyncT $ do
492161
493162-- | Compose a computation using previously captured list of logs. Composes a
494163-- group of computations, one for each log.
495−eachWithLog :: (MonadIO m, MonadBaseControl IO m, MonadThrow m)
496− => AsyncT m a -> [Log] -> AsyncT m a
164+eachWithLog :: MonadAsync m => AsyncT m a -> [Log] -> AsyncT m a
497165eachWithLog m logs = each logs >>= withLog m
498166
499167-- | Run an 'AsyncT m' computation with logging enabled and collect the results
500168-- or logs generated by each thread of the computation.
501−waitLogged :: (MonadIO m, MonadCatch m) => AsyncT m a -> m ([a], [Log])
169+waitLogged :: (MonadAsync m, MonadCatch m) => AsyncT m a -> m ([a], [Log])
502170waitLogged m = do
503171 resultsRef <- liftIO $ newIORef []
504172 lref <- liftIO $ newIORef []
@@ -510,7 +178,7 @@ waitLogged m = do
510178-- | Run an 'AsyncT m' computation with logging enabled, wait for it to finish
511179-- and discard the results. If the computation suspends collect the logs to
512180-- replay later.
513−waitLogged_ :: (MonadIO m, MonadCatch m) => AsyncT m a -> m [Log]
181+waitLogged_ :: (MonadAsync m, MonadCatch m) => AsyncT m a -> m [Log]
514182waitLogged_ m = do
515183 lref <- liftIO $ newIORef []
516184 waitAsync (const mzero) (Just lref) m