@@ -241,22 +241,20 @@ pullWorker :: (MonadIO m, MonadThrow m) => Context m a -> AsyncT m a
241241pullWorker ctx = AsyncT $ \_ stp yld -> do
242242 let continue = (runAsyncT (pullWorker ctx)) Nothing stp yld
243243 yield a = yld a Nothing (Just (pullWorker ctx))
244+ threadOp tid f finish cont = do
245+ done <- f ctx tid
246+ if done then finish else cont
244247
245248 ev <- liftIO $ atomically $ readTBQueue (childChannel ctx)
246249 case ev of
247250 ChildYield a -> yield a
248− ChildDone tid a -> do
249− done <- delThread ctx tid
250− if done then (yld a Nothing Nothing) else (yield a)
251− ChildStop tid e -> do
251+ ChildDone tid a ->
252+ threadOp tid delThread (yld a Nothing Nothing) (yield a)
253+ ChildStop tid e ->
252254 case e of
253− Nothing -> do
254− done <- delThread ctx tid
255− if done then stp else continue
256− Just x -> handleException x ctx tid
257− ChildCreate tid -> do
258− done <- addThread ctx tid
259− if done then stp else continue
255+ Nothing -> threadOp tid delThread stp continue
256+ Just ex -> handleException ex ctx tid
257+ ChildCreate tid -> threadOp tid addThread stp continue
260258
261259-- If an exception occurs we push it to the channel so that it can handled by
262260-- the parent. 'Paused' exceptions are to be collected at the top level.