@@ -58,7 +58,7 @@ import Control.Concurrent (ThreadId, forkIO,
5858 myThreadId, threadDelay)
5959import Control.Concurrent.MVar (MVar, newEmptyMVar, tryTakeMVar,
6060 tryPutMVar, takeMVar)
61−import Control.Exception (SomeException (..))
61+import Control.Exception (SomeException (..), Exception)
6262import qualified Control.Exception.Lifted as EL
6363import Control.Monad (ap, liftM, MonadPlus(..), mzero,
6464 when)
@@ -390,10 +390,21 @@ sendWorkerWait ctx = dispatch >> void (liftIO $ takeMVar (doorBell ctx))
390390 when (not done) $ (pushWorker ctx) >> dispatch
391391
392392
393+data ContextUsedAfterEOF = ContextUsedAfterEOF deriving Show
394+instance Exception ContextUsedAfterEOF
395+
393396-- | Pull an AsyncT stream from a context
394397{-# NOINLINE pullFromCtx #-}
395398pullFromCtx :: MonadAsync m => Context m a -> AsyncT m a
396399pullFromCtx ctx = AsyncT $ \_ stp yld -> do
400+ -- When using an async handle to the context, one may keep using a stale
401+ -- context even after it has been fullt drained. To detect it gracefully we
402+ -- raise an explicit exception.
403+ -- XXX if reading the IORef is costly we can use a flag in the context to
404+ -- indicate we are done.
405+ done <- allThreadsDone ctx
406+ when done $ throwM ContextUsedAfterEOF
407+
397408 res <- liftIO $ tryTakeMVar (doorBell ctx)
398409 when (isNothing res) $ sendWorkerWait ctx
399410 list <- liftIO $ atomicModifyIORefCAS (outputQueue ctx) $ \x -> ([], x)