@@ -309,12 +309,20 @@ allThreadsDone ctx = liftIO $ do
309 s2 <- readIORef (doneThreads ctx)
309 s2 <- readIORef (doneThreads ctx)
310 return $ S.null s1 && S.null s2
310 return $ S.null s1 && S.null s2
312−handleException :: (MonadIO m, MonadThrow m)
312+-- If an exception occurs we push it to the channel so that it can handled by
313− => SomeException -> Context m a -> ThreadId -> m r
313+-- the parent. 'Paused' exceptions are to be collected at the top level.
314−handleException e ctx tid = do
314+-- XXX Paused exceptions should only bubble up to the runRecorder computation
315− void (delThread ctx tid)
315+{-# NOINLINE handleChildException #-}
316+handleChildException :: MonadIO m => Context m a -> SomeException -> m ()
317+handleChildException ctx e = do
318+ tid <- liftIO myThreadId
319+ send ctx (ChildStop tid (Just e))
321+{-# NOINLINE pushWorker #-}
322+pushWorker :: MonadAsync m => Context m a -> m ()
324+ tid <- doFork (push ctx) (handleChildException ctx)
316− liftIO $ readIORef (runningThreads ctx) >>= mapM_ killThread
325+ liftIO $ modifyIORef (runningThreads ctx) $ (\s -> S.insert tid s)
319{-# INLINE sendWorkerWait #-}
327{-# INLINE sendWorkerWait #-}
320sendWorkerWait :: MonadAsync m => Context m a -> m ()
328sendWorkerWait :: MonadAsync m => Context m a -> m ()
@@ -334,8 +342,14 @@ pullWorker ctx = AsyncT $ \pctx stp yld -> do
334 when (isNothing res) $ sendWorkerWait ctx
342 when (isNothing res) $ sendWorkerWait ctx
335 list <- liftIO $ atomicModifyIORefCAS (outputQueue ctx) $ \x -> ([], x)
343 list <- liftIO $ atomicModifyIORefCAS (outputQueue ctx) $ \x -> ([], x)
336 (runAsyncT $ processEvents list) pctx stp yld
344 (runAsyncT $ processEvents list) pctx stp yld
348+ handleException e tid = do
349+ void (delThread ctx tid)
350+ liftIO $ readIORef (runningThreads ctx) >>= mapM_ killThread
339 {-# INLINE processEvents #-}
353 {-# INLINE processEvents #-}
340 processEvents [] = AsyncT $ \pctx stp yld -> do
354 processEvents [] = AsyncT $ \pctx stp yld -> do
341 done <- allThreadsDone ctx
355 done <- allThreadsDone ctx
@@ -353,29 +367,9 @@ pullWorker ctx = AsyncT $ \pctx stp yld -> do
353 ChildStop tid e ->
367 ChildStop tid e ->
354 case e of
368 case e of
355 Nothing -> delThread ctx tid >> continue
369 Nothing -> delThread ctx tid >> continue
356− Just ex -> handleException ex ctx tid
370+ Just ex -> handleException ex tid
357 ChildCreate tid -> addThread ctx tid >> continue
371 ChildCreate tid -> addThread ctx tid >> continue
359−-- If an exception occurs we push it to the channel so that it can handled by
360−-- the parent. 'Paused' exceptions are to be collected at the top level.
361−-- XXX Paused exceptions should only bubble up to the runRecorder computation
362−{-# NOINLINE handleChildException #-}
363−handleChildException :: MonadIO m => Context m a -> SomeException -> m ()
364−handleChildException ctx e = do
365− tid <- liftIO myThreadId
366− send ctx (ChildStop tid (Just e))
368−-- This function is different than "forkWorker" because we have to directly
369−-- insert the threadIds here and cannot use the channel to send ChildCreate
370−-- unlike on the push side. If we do that, the first thread's done message
371−-- may arrive even before the second thread is forked, in that case
372−-- pullWorker will falsely detect that all threads are over.
373−{-# NOINLINE pushWorker #-}
374−pushWorker :: MonadAsync m => Context m a -> m ()
376− tid <- doFork (push ctx) (handleChildException ctx)
377− liftIO $ modifyIORef (runningThreads ctx) $ (\s -> S.insert tid s)
379-- | Split the original computation in a pull-push pair. The original
373-- | Split the original computation in a pull-push pair. The original
380-- computation pulls from a Channel while m1 and m2 push to the channel.
374-- computation pulls from a Channel while m1 and m2 push to the channel.
381{-# NOINLINE pullFork #-}
375{-# NOINLINE pullFork #-}