e648b03 · distributed-process

git.iko.soy/distributed-process

on master help sign in

Have callLocal terminate when the node is closing.

e648b03Facundo Domínguez11 years ago

src/Control/Distributed/Process.hs+49−6

compare with master

unifiedsplit

@@ -166,6 +166,7 @@ import Prelude hiding (catch)

166import Control.Monad.IO.Class (liftIO)

167import Control.Applicative ((<$>))

168import Control.Monad.Reader (ask)

169+import Control.Concurrent (forkIO)

169import Control.Concurrent.MVar

170import Control.Concurrent.MVar

170 ( MVar

171 ( MVar

171 , newEmptyMVar

172 , newEmptyMVar

@@ -308,6 +309,12 @@ import Control.Distributed.Process.Internal.Primitives

308 , reconnectPort

309 , reconnectPort

309 )

310 )

310import Control.Distributed.Process.Node (forkProcess)

311import Control.Distributed.Process.Node (forkProcess)

312+import Control.Distributed.Process.Internal.Types

313+ ( processThread

314+ , localState

315+ , LocalNodeState(..)

316+ , withValidLocalState

317+ )

311import Control.Distributed.Process.Internal.Spawn

318import Control.Distributed.Process.Internal.Spawn

312 ( -- Spawning Processes/Channels

319 ( -- Spawning Processes/Channels

313 spawn

320 spawn

@@ -317,7 +324,10 @@ import Control.Distributed.Process.Internal.Spawn

317 , spawnSupervised

324 , spawnSupervised

318 , call

325 , call

319 )

326 )

327+import Control.Distributed.Process.Internal.StrictMVar (readMVar)

320−import Control.Exception (SomeException, throw, uninterruptibleMask_)

328+import Control.Exception (SomeException, throw, throwTo, throwIO)

329+import Control.Monad (forM_)

330+

321

331

322-- INTERNAL NOTES

332-- INTERNAL NOTES

323--

333--

@@ -428,9 +438,42 @@ callLocal proc = mask $ \release -> do

428 -- when the original exception was meant to kill the thread and the

438 -- when the original exception was meant to kill the thread and the

429 -- second exception doesn't (like the exception thrown by

439 -- second exception doesn't (like the exception thrown by

430 -- 'System.Timeout.timeout').

440 -- 'System.Timeout.timeout').

431− do lproc <- ask

441+ --

432− liftIO $ uninterruptibleMask_ $ runLocalProcess lproc $ do

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 [] $

433− receiveChan rpInit

450+ whenNotClosed $ receiveChan rpInit

434− kill child "exception in parent process"

451+ (exs1, _) <- collectExceptions exs0 $

452+ whenNotClosed $ kill child "callLocal was interrupted"

453+ (exs, mr) <- collectExceptions exs1 $

435− liftIO (takeMVar mv)

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

436 either throw return rs

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