@@ -1,16 +1,9 @@
1−{-# LANGUAGE ConstraintKinds #-}
2{-# LANGUAGE FlexibleContexts #-}
1{-# LANGUAGE FlexibleContexts #-}
3{-# LANGUAGE ScopedTypeVariables #-}
2{-# LANGUAGE ScopedTypeVariables #-}
5module Asyncly.Threads
4module Asyncly.Threads
16 , waitLogged
9 , waitLogged
@@ -22,349 +15,25 @@ module Asyncly.Threads
25−import Control.Applicative ((<|>), empty)
26−import Control.Concurrent (ThreadId, forkIO, killThread,
18+import Control.Concurrent (killThread)
28−import Control.Concurrent.STM (TChan, atomically, newTChan,
19+import Control.Concurrent.STM (atomically, newTChan)
29− readTChan, tryReadTChan,
31import Control.Exception (SomeException (..))
20import Control.Exception (SomeException (..))
32−import qualified Control.Exception.Lifted as EL
33−import Control.Monad.Catch (MonadCatch, MonadThrow, throwM,
21+import Control.Monad.Catch (MonadCatch, throwM, try)
35import Control.Monad.IO.Class (MonadIO (..))
22import Control.Monad.IO.Class (MonadIO (..))
36−import Control.Monad.State (get, gets, modify, mzero, put,
23+import Control.Monad.State (get, mzero, put, runStateT)
37− runStateT, when, StateT)
38import Control.Monad.Trans.Class (MonadTrans (lift))
24import Control.Monad.Trans.Class (MonadTrans (lift))
39−import Control.Monad.Trans.Control (MonadBaseControl, liftBaseWith)
40import Data.IORef (IORef, atomicModifyIORef,
25import Data.IORef (IORef, atomicModifyIORef,
41− modifyIORef, newIORef, readIORef,
26+ newIORef, readIORef)
43−import Data.List (delete)
44−import Data.Maybe (isJust)
45−import Unsafe.Coerce (unsafeCoerce)
47import Asyncly.AsyncT
28import Asyncly.AsyncT
49−type MonadAsync m = (MonadIO m, MonadBaseControl IO m, MonadThrow m)
51−------------------------------------------------------------------------------
52−-- Model of computation
53−------------------------------------------------------------------------------
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.
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.
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.
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.
75−------------------------------------------------------------------------------
76−-- Pick up from where we left in the previous thread
77−------------------------------------------------------------------------------
79−-- | Continue execution of the closure that we were executing when we migrated
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
88−------------------------------------------------------------------------------
89−-- Thread Management (creation, reaping and killing)
90−------------------------------------------------------------------------------
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
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)
103−drainChildren :: MonadIO m
106− -> m ([ThreadId], Maybe SomeException)
107−drainChildren cchan pending =
109− [] -> return (pending, Nothing)
111− ev <- liftIO $ atomically $ readTChan cchan
112− (p, e) <- processOneEvent ev pending
113− maybe (drainChildren cchan p) (const $ return (p, e)) e
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
123−tryReclaimZombies :: (MonadIO m, MonadThrow m) => Context -> m ()
124−tryReclaimZombies ctx = do
125− let cchan = childChannel ctx
126− pendingRef = pendingThreads ctx
128− pending <- liftIO $ readIORef pendingRef
132− mev <- liftIO $ atomically $ tryReadTChan cchan
134− Nothing -> return ()
136− (p, e) <- processOneEvent ev pending
137− liftIO $ writeIORef pendingRef p
138− maybe (return ()) throwM e
139− tryReclaimZombies ctx
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
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
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.
157−waitQSemB :: IORef Int -> IO Bool
158−waitQSemB sem = atomicModifyIORef sem $ \n ->
163−signalQSemB :: IORef Int -> IO ()
164−signalQSemB sem = atomicModifyIORef sem $ \n -> (n + 1, ())
166−-- Allocation of threads
168−-- global thread limit
169−-- thread fan-out i.e. per thread children limit
170−-- min per thread allocation to avoid starvation
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.
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
181−forkFinally1 :: (MonadIO m, MonadBaseControl IO m)
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
194−-- | Run a given context in a new thread.
196−forkContext :: (MonadBaseControl IO m, MonadIO m, MonadThrow m)
197− => AsyncT m a -> StateT Context m ()
198−forkContext action = do
200− childCtx <- childContext parentCtx
201− tid <- forkFinally1 childCtx action
202− (beforeExit childCtx (childChannel parentCtx))
203− updatePendingThreads parentCtx tid
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
215− childContext ctx = do
216− pendingRef <- liftIO $ newIORef []
217− chan <- liftIO $ atomically newTChan
218− -- shares the threadCredit of the parent by default
220− { pendingThreads = pendingRef
221− , childChannel = chan
224− beforeExit ctx pchan res = do
228− dbg $ "beforeExit: " ++ show tid ++ " caught exception"
229− liftIO $ readIORef (pendingThreads ctx) >>= mapM_ killThread
231− Right _ -> waitForChildren ctx
233− signalQSemB (threadCredit ctx)
234− liftIO $ atomically $ writeTChan pchan (ChildDone tid r)
236−-- | Decide whether to resume the context in the same thread or a new thread
238−canFork :: Context -> IO Bool
240− gotCredit <- liftIO $ waitQSemB (threadCredit context)
243− pending <- liftIO $ readIORef $ pendingThreads context
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.
253− waitForOneEvent context
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.
261−resumeContextWith :: MonadAsync m
262− => Bool -- force synchronous
264− -> StateT Context m ()
265−resumeContextWith synch action = do
267− can <- liftIO $ canFork context
268− case can && (not synch) of
269− False -> runContext context action
270− True -> forkContext action
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?
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
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.
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.
292−spawningParentDone :: MonadIO m => StateT Context m (Maybe a)
293−spawningParentDone = do
295− when (loc /= RemoteNode) $ setLocation WaitingParent
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.
301−async :: MonadAsync m => AsyncT m a -> AsyncT m a
302−async action = AsyncT $ do
303− resumeContextWith False action
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.
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
318− lift $ cbsetter $ \a -> do
319− let s = resumeContextWith False (return a)
320− _ <- runStateT s ctx
325−each :: (MonadIO m, MonadBaseControl IO m, MonadThrow m)
326− => [a] -> AsyncT m a
327−each xs = foldl (<|>) empty $ map (async . return) xs
329−------------------------------------------------------------------------------
330−-- Controlling thread quota
331−------------------------------------------------------------------------------
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
344− modify $ \s -> s { threadCredit = oldCr }
346− ) -- restore old credit
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?
357−-- modifyThreads :: MonadIO m => (Int -> Int) -> AsyncT m ()
361------------------------------------------------------------------------------
30------------------------------------------------------------------------------
362-- Running the monad
31-- Running the monad
363------------------------------------------------------------------------------
32------------------------------------------------------------------------------
365-- | Run an 'AsyncT m' computation and collect the results generated by each
34-- | Run an 'AsyncT m' computation and collect the results generated by each
366-- thread of the computation in a list.
35-- thread of the computation in a list.
367−waitAsync :: (MonadIO m, MonadCatch m)
36+waitAsync :: (MonadAsync m, MonadCatch m)
368 => (a -> AsyncT m a) -> Maybe (IORef [Log]) -> AsyncT m a -> m ()
37 => (a -> AsyncT m a) -> Maybe (IORef [Log]) -> AsyncT m a -> m ()
369waitAsync finalizer lref m = do
38waitAsync finalizer lref m = do
370 childChan <- liftIO $ atomically newTChan
39 childChan <- liftIO $ atomically newTChan
@@ -389,12 +58,12 @@ waitAsync finalizer lref m = do
390-- | Invoked to store the result of the computation in the context and finish
59-- | Invoked to store the result of the computation in the context and finish
391-- the computation when the computation is done
60-- 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
393gatherResult ref r = do
62gatherResult ref r = do
394 liftIO $ atomicModifyIORef ref $ \rs -> (r : rs, ())
63 liftIO $ atomicModifyIORef ref $ \rs -> (r : rs, ())
397−gather :: (MonadIO m, MonadCatch m) => AsyncT m a -> AsyncT m [a]
66+gather :: (MonadAsync m, MonadCatch m) => AsyncT m a -> AsyncT m [a]
398gather m = AsyncT $ do
67gather m = AsyncT $ do
399 resultsRef <- liftIO $ newIORef []
68 resultsRef <- liftIO $ newIORef []
400 lift $ waitAsync (gatherResult resultsRef) Nothing m
69 lift $ waitAsync (gatherResult resultsRef) Nothing m
@@ -403,7 +72,7 @@ gather m = AsyncT $ do
404-- | Run an 'AsyncT m' computation and collect the results generated by each
73-- | Run an 'AsyncT m' computation and collect the results generated by each
405-- thread of the computation in a list.
74-- 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]
407wait m = do
76wait m = do
408 resultsRef <- liftIO $ newIORef []
77 resultsRef <- liftIO $ newIORef []
409 waitAsync (gatherResult resultsRef) Nothing m
78 waitAsync (gatherResult resultsRef) Nothing m
@@ -411,14 +80,14 @@ wait m = do
412-- | Run an 'AsyncT m' computation, wait for it to finish and discard the
81-- | Run an 'AsyncT m' computation, wait for it to finish and discard the
413-- results.
82-- results.
414−wait_ :: (MonadIO m, MonadCatch m) => AsyncT m a -> m ()
83+wait_ :: (MonadAsync m, MonadCatch m) => AsyncT m a -> m ()
415wait_ m = waitAsync (const mzero) Nothing m
84wait_ m = waitAsync (const mzero) Nothing m
417------------------------------------------------------------------------------
86------------------------------------------------------------------------------
418-- Logging
87-- Logging
419------------------------------------------------------------------------------
88------------------------------------------------------------------------------
421−logged :: (Loggable a, MonadIO m) => AsyncT m a -> AsyncT m a
90+logged :: (Loggable a, MonadAsync m) => AsyncT m a -> AsyncT m a
422logged m = AsyncT $ do
91logged m = AsyncT $ do
423 ctx <- get
92 ctx <- get
424 case journal ctx of
93 case journal ctx of
@@ -467,7 +136,7 @@ logged m = AsyncT $ do
467-- completed threads output is returned and for suspended threads the logs to
136-- completed threads output is returned and for suspended threads the logs to
468-- resume them are returned.
137-- resume them are returned.
470−suspend :: MonadIO m => AsyncT m ()
139+suspend :: MonadAsync m => AsyncT m ()
471suspend = logged $ AsyncT $ do
140suspend = logged $ AsyncT $ do
472 ctx <- get
141 ctx <- get
473 case logsRef ctx of
142 case logsRef ctx of
@@ -492,13 +161,12 @@ withLog m (Log entries) = AsyncT $ do
493-- | Compose a computation using previously captured list of logs. Composes a
162-- | Compose a computation using previously captured list of logs. Composes a
494-- group of computations, one for each log.
163-- 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
497eachWithLog m logs = each logs >>= withLog m
165eachWithLog m logs = each logs >>= withLog m
499-- | Run an 'AsyncT m' computation with logging enabled and collect the results
167-- | Run an 'AsyncT m' computation with logging enabled and collect the results
500-- or logs generated by each thread of the computation.
168-- 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])
502waitLogged m = do
170waitLogged m = do
503 resultsRef <- liftIO $ newIORef []
171 resultsRef <- liftIO $ newIORef []
504 lref <- liftIO $ newIORef []
172 lref <- liftIO $ newIORef []
@@ -510,7 +178,7 @@ waitLogged m = do
510-- | Run an 'AsyncT m' computation with logging enabled, wait for it to finish
178-- | Run an 'AsyncT m' computation with logging enabled, wait for it to finish
511-- and discard the results. If the computation suspends collect the logs to
179-- and discard the results. If the computation suspends collect the logs to
512-- replay later.
180-- replay later.
513−waitLogged_ :: (MonadIO m, MonadCatch m) => AsyncT m a -> m [Log]
181+waitLogged_ :: (MonadAsync m, MonadCatch m) => AsyncT m a -> m [Log]
514waitLogged_ m = do
182waitLogged_ m = do
515 lref <- liftIO $ newIORef []
183 lref <- liftIO $ newIORef []
516 waitAsync (const mzero) (Just lref) m
184 waitAsync (const mzero) (Just lref) m