@@ -309,12 +309,20 @@ allThreadsDone ctx = liftIO $ do
309309 s2 <- readIORef (doneThreads ctx)
310310 return $ S.null s1 && S.null s2
311311
312−handleException :: (MonadIO m, MonadThrow m)
313− => SomeException -> Context m a -> ThreadId -> m r
314−handleException e ctx tid = do
315− void (delThread ctx tid)
316− liftIO $ readIORef (runningThreads ctx) >>= mapM_ killThread
317− throwM e
312+-- If an exception occurs we push it to the channel so that it can handled by
313+-- the parent. 'Paused' exceptions are to be collected at the top level.
314+-- XXX Paused exceptions should only bubble up to the runRecorder computation
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))
320+
321+{-# NOINLINE pushWorker #-}
322+pushWorker :: MonadAsync m => Context m a -> m ()
323+pushWorker ctx = do
324+ tid <- doFork (push ctx) (handleChildException ctx)
325+ liftIO $ modifyIORef (runningThreads ctx) $ (\s -> S.insert tid s)
318326
319327{-# INLINE sendWorkerWait #-}
320328sendWorkerWait :: MonadAsync m => Context m a -> m ()
@@ -334,8 +342,14 @@ pullWorker ctx = AsyncT $ \pctx stp yld -> do
334342 when (isNothing res) $ sendWorkerWait ctx
335343 list <- liftIO $ atomicModifyIORefCAS (outputQueue ctx) $ \x -> ([], x)
336344 (runAsyncT $ processEvents list) pctx stp yld
345+
337346 where
338347
348+ handleException e tid = do
349+ void (delThread ctx tid)
350+ liftIO $ readIORef (runningThreads ctx) >>= mapM_ killThread
351+ throwM e
352+
339353 {-# INLINE processEvents #-}
340354 processEvents [] = AsyncT $ \pctx stp yld -> do
341355 done <- allThreadsDone ctx
@@ -353,29 +367,9 @@ pullWorker ctx = AsyncT $ \pctx stp yld -> do
353367 ChildStop tid e ->
354368 case e of
355369 Nothing -> delThread ctx tid >> continue
356− Just ex -> handleException ex ctx tid
370+ Just ex -> handleException ex tid
357371 ChildCreate tid -> addThread ctx tid >> continue
358372
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))
367−
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 ()
375−pushWorker ctx = do
376− tid <- doFork (push ctx) (handleChildException ctx)
377− liftIO $ modifyIORef (runningThreads ctx) $ (\s -> S.insert tid s)
378−
379373-- | Split the original computation in a pull-push pair. The original
380374-- computation pulls from a Channel while m1 and m2 push to the channel.
381375{-# NOINLINE pullFork #-}