composewell
Forward exceptions from parent to children
I noticed when trying the examples in the tutorial in ghci, that pressing ctrl-c doesn't abort anything but the top level process.
For example, if I press ctrl-c during this program:
S.drain $ aheadly $ S.mapM (\n -> threadDelay (1000000*n) >> print n) $ S.fromList [1,2,3]
everything will keep going in the background:
> S.drain $ aheadly $ S.mapM (\n -> threadDelay (1000000*n) >> print n) $ S.fromList [1,2,3]
1
^CInterrupted.
> 2
3
This may sometimes be what we want, but most of the time it is not. If it is possible, I think the most sensible default behaviour would be to forward the exceptions in both directions.
Here is another example, using explicit exceptions instead of ctrl-c:
> S.drain $ S.mapM (\n -> print (n,n) >> when (n == 2) (error "bad!")) $ parallely $ S.mapM (\n -> threadDelay (1000000*n) >> print n >> pure n) $ S.fromList [1,2,3,4]
1
(1,1)
2
(2,2)
*** Exception: bad!
CallStack (from HasCallStack):
error, called at <interactive>:87:55 in interactive:Ghci35
> 3
4
On the other hand, if the child throws an exception, it seems to also kill its siblings, as would be expected:
> S.drain $ parallely $ S.mapM (\n -> threadDelay (1000000*n) >> print n >> when (n == 2) (error "bad!")) $ S.fromList [1,2,3,4]
1
2
*** Exception: bad!
CallStack (from HasCallStack):
error, called at <interactive>:91:91 in interactive:Ghci35
>
In this case the pending threads are cleaned up by the GC. You can try running it like this and you will get the expected results:
{-# LANGUAGE ScopedTypeVariables #-}
import qualified Streamly.Prelude as S
import Control.Concurrent (threadDelay)
import Streamly (parallely)
import Control.Monad (when)
import Control.Exception (try, SomeException)
import System.Mem (performMajorGC)
main = do
r <- try $ S.drain
$ S.mapM (\n -> print (n,n) >> when (n == 2) (error "bad!"))
$ parallely
$ S.mapM (\n -> threadDelay (1000000*n) >> print n >> pure n)
$ S.fromList [1,2,3,4]
case r of
Left (e :: SomeException) -> do
putStrLn "got exception, waiting"
performMajorGC
threadDelay 5000000
Right _ -> return ()
To make the cleanup more prompt, we can possibly provide a wrapper to catch exceptions, cleanup the threads and rethrow the exceptions. But you can write it yourself using code similar to the snippet above.
parMapM still suffers from this issue. See following example to reproduce it.
test :: IO ()
test = do
tvar <- newTVarIO (0 :: Int)
let b = bracket_ (atomically $ modifyTVar' tvar succ) (atomically $ modifyTVar' tvar pred) . (`catch` \(e :: SomeException) -> print e)
(`finally` ((performMajorGC >>) $ print =<< readTVarIO tvar)) $
S.repeat True
& S.parMapM (S.eager True . S.maxBuffer 1) (const $ b $ threadDelay 1000000)
& S.fold F.drain
You'll got 1 from tvar in the finally block while pressing Ctrl-C, even when performMajorGC was run first. The thread did receive ThreadInterrupted exception but after performMajorGC returns.
Adding one threadDelay xxx right after performMajorGC does help, but it's just a workaround in my opinion (xxx could always be too short in certain case).
I believe this is an serious issue since this behaviour make it impossible to use scoped resource (ex. allocated via bracket) safely inside body which attachs to parMapM (or combinators share same behaviour), since these threads could outlive the scope.
@TheKK thanks for following up on this. We are fixing it in 0.11.
There are two issues here:
- Cleanup leaks threads when eager option is used, due to a race
- GC based cleanup is inherently async, therefore, the time taken depends on the number of threads and in more complicated situations it may take more than one GC to clean it up.
We have made the cleanup completely robust. And two new functions are provided to create a scope for cleanup. Cleanup is guaranteed to finish within the scope. See the tests in pr #3038 .