Improve exception handling in callLocal.

callLocal had few flaws,

1. it could leak worker process in parent process received an
   exception but handles it. Then process was not killed, so
   linking worker to parent didn't help

2. it was not possible to handle asynchronous exception in worker
   process.

This this patch both things are solved alltogether, this patch
as a correct mechanism for exceptin handling, so if worker was
not finished then all exceptions are rethrown to that process.
In this case worker is capable for exception handling. However
if worker already finished, then exception in thrown from the
callLocal code. The resulting semantics is the same as wheb
code is executed in a single process.

Author: qnikst

src/Control/Distributed/Process.hs

@@ -160,9 +160,16 @@ import Prelude hiding (catch)
 #endif
 
 import Control.Monad.IO.Class (liftIO)
-import Control.Applicative ((<$>))
+import Control.Applicative ((<$>), (<*))
 import Control.Monad.Reader (ask)
-import Control.Concurrent.MVar (newEmptyMVar, takeMVar, putMVar)
+import Control.Concurrent (myThreadId)
+import Control.Concurrent.MVar
+  ( newMVar
+  , newEmptyMVar
+  , tryTakeMVar
+  , takeMVar
+  , putMVar
+  )
 import Control.Distributed.Static
   ( Closure
   , closure
@@ -303,7 +310,8 @@ import Control.Distributed.Process.Internal.Spawn
   , spawnSupervised
   , call
   )
-import Control.Exception (SomeException, throwIO)
+import Control.Exception (SomeException, throwIO, throwTo)
+import qualified Control.Exception as Exception
 
 -- INTERNAL NOTES
 --
@@ -402,12 +410,27 @@ spawnChannelLocal proc = do
 callLocal ::
      Process a  -- ^ Process to run
   -> Process a  -- ^ Value returned
-callLocal proc = do
-  mv <- liftIO newEmptyMVar
-  self <- getSelfPid
-  _ <- spawnLocal $ do
-     link self
-     try proc >>= liftIO . putMVar mv
-  liftIO $ takeMVar mv >>=
-     either (throwIO :: SomeException -> IO a) return
+callLocal proc = mask $ \release -> do
+  lock  <- liftIO $ newMVar ()
+  tidMV <- liftIO $ newEmptyMVar
+  mv    <- liftIO newEmptyMVar
+  _ <- spawnLocal $ mask $ \release' -> do
+         liftIO $ myThreadId >>= putMVar tidMV
+         ep <- try $ release' $ proc <* liftIO (takeMVar lock)
+         liftIO $ putMVar mv ep
+  tid <- liftIO $ takeMVar tidMV
+  let fetchResult =
+        (takeMVar mv >>= either (throwIO :: SomeException -> IO a) return)
+        `Exception.catch`
+            (\e -> Exception.mask $ \release' -> do
+                     ml <- tryTakeMVar lock
+                     case ml of
+                        Nothing -> -- lock already taked by worker so we need
+                                   -- to rethrow exception now
+                          throwIO e
+                        Just{}  -> do
+                          throwTo tid (e::SomeException)
+                          putMVar lock ()
+                     release' fetchResult)
+  release $ liftIO $ fetchResult