@@ -35,7 +35,7 @@ module Asyncly.AsyncT
3535where
3636
3737import Control.Applicative (Alternative (..), liftA2)
38−import Control.Concurrent (ThreadId, forkIO, killThread,
38+import Control.Concurrent (ThreadId, forkIO,
3939 myThreadId, threadDelay)
4040import Control.Concurrent.MVar (MVar, newEmptyMVar, tryTakeMVar,
4141 tryPutMVar, takeMVar)
@@ -55,7 +55,7 @@ import Data.Concurrent.Queue.MichaelScott (LinkedQueue, newQ, pushL,
5555 tryPopR)
5656import Data.Functor (void)
5757import Data.IORef (IORef, modifyIORef, newIORef,
58− writeIORef, readIORef)
58+ readIORef)
5959import Data.Atomics (atomicModifyIORefCAS,
6060 atomicModifyIORefCAS_)
6161import Data.Maybe (isNothing)
@@ -87,9 +87,7 @@ import Control.Monad.Trans.Recorder (MonadRecorder(..))
8787
8888data ChildEvent a =
8989 ChildYield a
90− | ChildDone ThreadId a
9190 | ChildStop ThreadId (Maybe SomeException)
92− | ChildCreate ThreadId
9391
9492------------------------------------------------------------------------------
9593-- State threaded around the monad for thread management
@@ -103,7 +101,6 @@ data Context m a =
103101 , enqueue :: AsyncT m a -> IO ()
104102 , runqueue :: m ()
105103 , runningThreads :: IORef (Set ThreadId)
106− , doneThreads :: IORef (Set ThreadId)
107104 }
108105
109106-- The 'Maybe (AsyncT m a)' is redundant as we can use 'stop' value for the
@@ -272,51 +269,30 @@ runqueueFIFO ctx q = run
272269 yield a _ Nothing = sendit a >> run
273270 -- XXX do we need to save the context as well?
274271 -- we can enqueue "(runAsyncT m) ctx run yield" instead
275− yield a c (Just r) = sendit a >> liftIO (enqueueFIFO q r) >> run
272+ yield a _ (Just r) = sendit a >> liftIO (enqueueFIFO q r) >> run
276273
277−-- Thread tracking has a significant performance overhead (~20% on empty
278−-- threads, it will be lower for heavy threads). It is needed for two reasons:
274+-- Thread tracking is needed for two reasons:
279275--
280276-- 1) Killing threads on exceptions. Threads may not be allowed to go away by
281277-- themselves because they may run for significant times before going away or
282278-- worse they may be stuck in IO and never go away.
283279--
284−-- 2) To know when all threads are done. This can be acheived by detecting a
285−-- BlockedIndefinitelyOnSTM exception too. But we will have to trigger a GC to
286−-- make sure that we detect it promptly.
287−--
288−-- This is a bit messy because ChildCreate and ChildDone events can arrive out
289−-- of order in case of pushSideDispatch. Returns whether we are done draining
290−-- threads.
291−{-# INLINE accountThread #-}
292−accountThread :: MonadIO m
293− => ThreadId -> IORef (Set ThreadId) -> IORef (Set ThreadId) -> m Bool
294−accountThread tid ref1 ref2 = liftIO $ do
295− s1 <- readIORef ref1
296− s2 <- readIORef ref2
297−
298− if (S.member tid s1) then do
299− let r = S.delete tid s1
300− writeIORef ref1 r
301− return $ S.null r && S.null s2
302− else do
303− liftIO $ writeIORef ref2 (S.insert tid s2)
304− return False
280+-- 2) To know when all threads are done.
305281
306282{-# NOINLINE addThread #-}
307−addThread :: MonadIO m => Context m a -> ThreadId -> m Bool
308−addThread ctx tid = accountThread tid (doneThreads ctx) (runningThreads ctx)
283+addThread :: MonadIO m => Context m a -> ThreadId -> m ()
284+addThread ctx tid =
285+ liftIO $ modifyIORef (runningThreads ctx) $ (\s -> S.insert tid s)
309286
310287{-# INLINE delThread #-}
311−delThread :: MonadIO m => Context m a -> ThreadId -> m Bool
312−delThread ctx tid = accountThread tid (runningThreads ctx) (doneThreads ctx)
288+delThread :: MonadIO m => Context m a -> ThreadId -> m ()
289+delThread ctx tid =
290+ liftIO $ modifyIORef (runningThreads ctx) $ (\s -> S.delete tid s)
313291
314292{-# INLINE allThreadsDone #-}
315293allThreadsDone :: MonadIO m => Context m a -> m Bool
316294allThreadsDone ctx = liftIO $ do
317− s1 <- readIORef (runningThreads ctx)
318− s2 <- readIORef (doneThreads ctx)
319− return $ S.null s1 && S.null s2
295+ readIORef (runningThreads ctx) >>= return . S.null
320296
321297-- If an exception occurs we push it to the channel so that it can handled by
322298-- the parent. 'Paused' exceptions are to be collected at the top level.
@@ -329,9 +305,8 @@ handleChildException ctx e = do
329305
330306{-# NOINLINE pushWorker #-}
331307pushWorker :: MonadAsync m => Context m a -> m ()
332−pushWorker ctx = do
333− tid <- doFork (runqueue ctx) (handleChildException ctx)
334− liftIO $ modifyIORef (runningThreads ctx) $ (\s -> S.insert tid s)
308+pushWorker ctx =
309+ doFork (runqueue ctx) (handleChildException ctx) >>= addThread ctx
335310
336311{-# INLINE sendWorkerWait #-}
337312sendWorkerWait :: MonadAsync m => Context m a -> m ()
@@ -355,8 +330,9 @@ pullWorker ctx = AsyncT $ \pctx stp yld -> do
355330 where
356331
357332 handleException e tid = do
358− void (delThread ctx tid)
359− liftIO $ readIORef (runningThreads ctx) >>= mapM_ killThread
333+ delThread ctx tid
334+ -- XXX implement kill async exception handling
335+ -- liftIO $ readIORef (runningThreads ctx) >>= mapM_ killThread
360336 throwM e
361337
362338 {-# INLINE processEvents #-}
@@ -372,12 +348,10 @@ pullWorker ctx = AsyncT $ \pctx stp yld -> do
372348
373349 case ev of
374350 ChildYield a -> yield a
375− ChildDone tid a -> delThread ctx tid >> yield a
376351 ChildStop tid e ->
377352 case e of
378353 Nothing -> delThread ctx tid >> continue
379354 Just ex -> handleException ex tid
380− ChildCreate tid -> addThread ctx tid >> continue
381355
382356-- | Split the original computation in a pull-push pair. The original
383357-- computation pulls from a Channel while m1 and m2 push to the channel.
@@ -393,7 +367,6 @@ pullFork m1 m2 fifo = AsyncT $ \_ stp yld -> do
393367 outQ <- newIORef []
394368 outQMv <- newEmptyMVar
395369 running <- newIORef S.empty
396− done <- newIORef S.empty
397370
398371 case fifo of
399372 True -> do
@@ -403,7 +376,6 @@ pullFork m1 m2 fifo = AsyncT $ \_ stp yld -> do
403376 Context { outputQueue = outQ
404377 , synchOutQ = outQMv
405378 , runningThreads = running
406− , doneThreads = done
407379 , runqueue = runqueueFIFO ctx q
408380 , enqueue = pushL q
409381 }
@@ -415,7 +387,6 @@ pullFork m1 m2 fifo = AsyncT $ \_ stp yld -> do
415387 Context { outputQueue = outQ
416388 , synchOutQ = outQMv
417389 , runningThreads = running
418− , doneThreads = done
419390 , runqueue = runqueueLIFO ctx q
420391 , enqueue = enqueueLIFO q
421392 }