[reactive] Re: black hole detection and concurrency

[Bertram Felgenhauer]

Sterling Clover wrote:
> On Dec 27, 2008, at 9:02 AM, Bertram Felgenhauer wrote:
>> In the above code, there is a small window between catching the
>> ThreadKilled exception and throwing it again though, where other
>> exceptions may creep in. The only way I see of fixing that is to use
>> 'block' and 'unblock' directly.
>
> That certainly seems to do the trick for the simple example at least. One 
> way to reason about it better would be, instead of folding everything into 
> the race function, to simply modify ghc's bracket function to give us the 
> behavior we'd prefer (speaking of which, I recall there's something in the 
> works for 6.12 or so to improve rethrowing of asynchronous exceptions?)
>
> brackAsync before after thing =
>   block (do
>     a <- before
>     r <- catch
>            (unblock (thing a))
>            (\_ -> after a >> myThreadId >>= killThread >>
>                   brackAsync before after thing )
>     after a
>     return r
>  )
>     where threadKilled ThreadKilled = Just ()
>           threadKilled _            = Nothing

This code turns any exception into ThreadKilled further down the stack.

  (\e -> do
       after a
       myThreadId >>= flip throwTo (e :: SomeException)
       ...

might do the trick.

My assumption was that anything but 'ThreadKilled' would be a
real error. This isn't really true, I guess - thanks to throwTo,
any exception could be asynchronous.

If an exception is thrown, 'after a' is run again after the computation
has resumed.

That's why I did the cleanup within the 'catch'.

But there's no reason why you couldn't do that as well:

  brackAsync before after thing =
    block $ do
      a <- before
      catch  (unblock (thing a) >>= \r -> after a >> return r) $
             \e -> do
                    after a
                    myThreadId >>= flip throwTo (e :: SomeException)
                    brackAsync before after thing )

> This brackAsync just drops in to the previous code where bracket was and 
> appears to perform correctly.

Right. 'race' should also unblock exceptions in the worker threads,

    withThread u v = brackAsync (forkIO (unblock u)) killThread (const v)

but that's an independent change.

> Further, if we place a trace after the 
> killThread, we se it gets executed once when the example is read (i.e. a 
> resumption) but it does not get executed if the (`seq` v) is removed from 
> the example So this gives me some hope that this is actually doing what 
> we'd like. I don't doubt it may have further kinks however.

At least the GHC RTS has support for the hard part - unwinding the stack
so that computations can be resumed seamlessly.

I'm not sure which of the approaches I like better - it seems that we
have a choice between turning async exceptions into sync ones or vice
versa, and neither choice is strictly superior to the other.

Enjoy,

Bertram

'race' update:
- Bugfix: Previously, only AsyncException-s would be caught.
  Use 'fromException' to select the ThreadKilled exception.
- I tried using a custom 'SuspendException' type, but this resulted in
  'test: SuspendException' messages on the console, while ThreadKilled
  is silently ignored... as documented:
     http://www.haskell.org/ghc/docs/latest/html/libraries/base/Control-Concurrent.html#v%3AforkIO
     (http://tinyurl.com/9t5pxs)
- Tweak: Block exceptions while running 'cleanup' to avoid killing
  threads twice.
- Trick: takeMVar is a blocking operation, so exceptions can be
  delivered while it's waiting - there's no need to use 'unblock' for
  this. In other words,  unblock (takeMVar v)  and  takeMVar v  are
  essentially equivalent for our purposes.

race :: IO a -> IO a -> IO a
race a b = block $ do
    v <- newEmptyMVar
    let t x = unblock (x >>= putMVar v)
    ta <- forkIO (t a)
    tb <- forkIO (t b)
    let cleanup = killThread ta >> killThread tb
    (do r <- takeMVar v; cleanup; return r) `catch`
        \e -> cleanup >>
            case fromException e of
                Just ThreadKilled -> do
                    myThreadId >>= killThread
                    unblock (race a b)
                _ -> throwIO e