@@ -166,6 +166,7 @@ import Prelude hiding (catch)
166166import Control.Monad.IO.Class (liftIO)
167167import Control.Applicative ((<$>))
168168import Control.Monad.Reader (ask)
169+import Control.Concurrent (forkIO)
169170import Control.Concurrent.MVar
170171 ( MVar
171172 , newEmptyMVar
@@ -308,6 +309,12 @@ import Control.Distributed.Process.Internal.Primitives
308309 , reconnectPort
309310 )
310311import Control.Distributed.Process.Node (forkProcess)
312+import Control.Distributed.Process.Internal.Types
313+ ( processThread
314+ , localState
315+ , LocalNodeState(..)
316+ , withValidLocalState
317+ )
311318import Control.Distributed.Process.Internal.Spawn
312319 ( -- Spawning Processes/Channels
313320 spawn
@@ -317,7 +324,10 @@ import Control.Distributed.Process.Internal.Spawn
317324 , spawnSupervised
318325 , call
319326 )
320−import Control.Exception (SomeException, throw, uninterruptibleMask_)
327+import Control.Distributed.Process.Internal.StrictMVar (readMVar)
328+import Control.Exception (SomeException, throw, throwTo, throwIO)
329+import Control.Monad (forM_)
330+
321331
322332-- INTERNAL NOTES
323333--
@@ -428,9 +438,42 @@ callLocal proc = mask $ \release -> do
428438 -- when the original exception was meant to kill the thread and the
429439 -- second exception doesn't (like the exception thrown by
430440 -- 'System.Timeout.timeout').
431− do lproc <- ask
432− liftIO $ uninterruptibleMask_ $ runLocalProcess lproc $ do
433− receiveChan rpInit
434− kill child "exception in parent process"
435− liftIO (takeMVar mv)
441+ --
442+ -- Using 'uninterruptibleMask_' would be problematic as there is no
443+ -- way to be sure the node isn't closing when the cleanup executes.
444+ -- Successfully running the cleanup is necessary to ensure
445+ -- termination. If exceptions are masked uninterruptibly,
446+ -- 'receiveChan' might block indefinitely and 'kill' might have no
447+ -- effect.
448+ --
449+ do (exs0, _) <- collectExceptions [] $
450+ whenNotClosed $ receiveChan rpInit
451+ (exs1, _) <- collectExceptions exs0 $
452+ whenNotClosed $ kill child "callLocal was interrupted"
453+ (exs, mr) <- collectExceptions exs1 $
454+ whenNotClosed $ liftIO (takeMVar mv)
455+ -- Forward exceptions asynchronously.
456+ lproc <- ask
457+ liftIO $ forkIO $ forM_ (reverse exs) $
458+ throwTo (processThread lproc)
459+ here <- getSelfNode
460+ liftIO $
461+ maybe (throwIO $ userError $ "Node closed " ++ show here)
462+ return mr
436463 either throw return rs
464+ where
465+ -- Retries a given action until it succeeds.
466+ collectExceptions :: [SomeException]
467+ -> Process a
468+ -> Process ([SomeException], a)
469+ collectExceptions exs p =
470+ try p >>= either (\e -> collectExceptions (e : exs) p) (return . (,) exs)
471+
472+ -- Runs the given closure if the node is not closed.
473+ whenNotClosed :: Process a -> Process (Maybe a)
474+ whenNotClosed p = do
475+ lproc <- ask
476+ st <- liftIO $ readMVar (localState $ processNode lproc)
477+ case st of
478+ LocalNodeValid _ -> fmap Just p
479+ LocalNodeClosed -> return Nothing