@@ -166,7 +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)
169+import Control.Concurrent (killThread)
170170import Control.Concurrent.MVar
171171 ( MVar
172172 , newEmptyMVar
@@ -200,8 +200,8 @@ import Control.Distributed.Process.Internal.Types
200200 , WhereIsReply(..)
201201 , RegisterReply(..)
202202 , LocalProcess(processNode)
203− , runLocalProcess
204203 , Message
204+ , localProcessWithId
205205 )
206206import Control.Distributed.Process.Serializable (Serializable)
207207import Control.Distributed.Process.Internal.Primitives
@@ -311,8 +311,6 @@ import Control.Distributed.Process.Internal.Primitives
311311import Control.Distributed.Process.Node (forkProcess)
312312import Control.Distributed.Process.Internal.Types
313313 ( processThread
314− , localState
315− , LocalNodeState(..)
316314 , withValidLocalState
317315 )
318316import Control.Distributed.Process.Internal.Spawn
@@ -324,9 +322,10 @@ import Control.Distributed.Process.Internal.Spawn
324322 , spawnSupervised
325323 , call
326324 )
327−import Control.Distributed.Process.Internal.StrictMVar (readMVar)
328−import Control.Exception (SomeException, throw, throwTo, throwIO)
325+import Control.Exception (SomeException, throwIO, uninterruptibleMask_)
326+import qualified Control.Exception as Exception (onException)
329327import Control.Monad (forM_)
328+import Data.Accessor ((^.))
330329
331330
332331-- INTERNAL NOTES
@@ -426,54 +425,19 @@ spawnChannelLocal proc = do
426425callLocal :: Process a -> Process a
427426callLocal proc = mask $ \release -> do
428427 mv <- liftIO newEmptyMVar :: Process (MVar (Either SomeException a))
429− (spInit, rpInit) <- newChan -- TODO: Remove when spawnLocal inherits the
430− -- masking state.
431− child <- spawnLocal $ mask_ $
432− try (sendChan spInit () >> release proc) >>= liftIO . putMVar mv
433− rs <- liftIO (takeMVar mv)
434− `onException`
428+ child <- spawnLocal $ try (release proc) >>= liftIO . putMVar mv
429+ lproc <- ask
430+ liftIO $ do
431+ rs <- Exception.onException (takeMVar mv) $ uninterruptibleMask_ $
435432 -- Exceptions need to be prevented from interrupting the clean up or
436433 -- the original exception which caused entering the handler could be
437434 -- forgotten. For instance, this could have a problematic effect
438435 -- when the original exception was meant to kill the thread and the
439436 -- second exception doesn't (like the exception thrown by
440437 -- 'System.Timeout.timeout').
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
463− 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
438+ do mchildThreadId <- withValidLocalState (processNode lproc) $
439+ \vst -> return $ fmap processThread $
440+ vst ^. localProcessWithId (processLocalId child)
441+ forM_ mchildThreadId killThread
442+ takeMVar mv
443+ either throwIO return rs