[reactive] Re: black hole detection and concurrency

[Sterling Clover]

You're absolutely right. The threads get killed when they hang on  
putting the full semaphore, not in the course of evaluation. I knew  
that there was going to be a fatal flaw, but I just didn't think it  
would be that obvious. :-)

Cheers,
Sterl.

On Dec 29, 2008, at 2:23 AM, Conal Elliott wrote:

> What does 'putMVar sem True' accomplish that 'putMVar v x' doesn't  
> alone?
>
> If 'evaluate a' succeeds first (for instance), 'evaluate b' will  
> keep on working, won't it?  I doubt the RTS is smart enough to  
> notice that the result of 'evaluate b' is going to be put into an  
> MVar, let alone a full one with no takeMVar, or that 'evaluate b'  
> is followed by writing to a similarly full & abandoned boolean MVar.
>
>   - Conal
>
> On Sun, Dec 28, 2008 at 11:05 PM, Sterling Clover  
> <s.clover at gmail.com> wrote:
> 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>
>
>