[reactive] Re: black hole detection and concurrency

[Sterling Clover]

Not sent to glasgow-haskell, because I feel it would be getting off  
topic:

After playing around with the track we were on I got in a hopeless  
muddle, and somewhat doubt that the current ghc runtime makes it  
easy. So I started again from first principles and got the following,  
which seems to work. A casual test with the TestRace harness doesn't  
seem to show any obvious leaks. The ghci runtime *should* detect when  
sparks block indefinitely on putting into the blocked MVar (the  
semaphore, which goes out of scope), and kill them.  The messiness of  
the story of unamb so far makes me doubt this solution, or wonder if  
trusting the runtime to eventually get around to it is insufficient,  
but nonetheless, it may just work.

I've also tried a versuion with real threads instead of sparks, to  
the same effect, relying on the runtime to detect and kill indefinite  
blocking, rather than using explicit asynchronous exceptions.

Both would need more stress testing for me to feel confident that  
they don't leak.

unamb :: a -> a -> a
a `unamb` b = unsafePerformIO (a `amb` b)

amb a b = do
     v <- newEmptyMVar
     sem <- newEmptyMVar
     forkIO . evaluate $ unsafePerformIO (evaluate a >>= \x ->  
putMVar sem True >> putMVar v x) `par`
                                       unsafePerformIO (evaluate b  
 >>= \x -> putMVar sem True >> putMVar v x)
     takeMVar v

--This version tries the same thing, but with normal threads. it  
seems zippier at first, but also like it may have a leak.
amb' a b = block $ do
     v <- newEmptyMVar
     sem <- newEmptyMVar
     forkIO $ evaluate a >>= \x -> putMVar sem True >> putMVar v x
     forkIO $ evaluate b >>= \x -> putMVar sem True >> putMVar v x
     takeMVar v

Cheers,
Sterl.

On Dec 28, 2008, at 7:34 PM, Conal Elliott wrote:

> Thanks very much for these ideas.  Peter Verswyvelen suggested  
> running the example repeatedly to see if it always runs correctly.   
> He found, and I verified, that the example runs fine with Bertram's  
> last version of unamb below, unless it's compiled with -threaded  
> and run with +RTS -N2.  In the latter case, it locks up after a while.
>
> I also tried a version with brackAsync and found that it eventually  
> locks up even under ghci.  When compiled & run multi-threaded, it  
> locks up almost immediately.
>
> I've attached a module, TestRace.hs, containing these experiments.
>
>     - Conal
>
> On Sat, Dec 27, 2008 at 6:03 PM, Bertram Felgenhauer  
> <bertram.felgenhauer at googlemail.com> wrote:
> 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
> _______________________________________________
> Reactive mailing list
> Reactive at haskell.org
> http://www.haskell.org/mailman/listinfo/reactive
>
> <TestRace.hs>