caf93de · distributed-process

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

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

168import Control.Monad.Reader (ask)

169−import Control.Concurrent (forkIO)

169+import Control.Concurrent (killThread)

170import Control.Concurrent.MVar

171 ( MVar

172 , newEmptyMVar

200 , WhereIsReply(..)

201 , RegisterReply(..)

202 , LocalProcess(processNode)

203− , runLocalProcess

204 , Message

203 , Message

204+ , localProcessWithId

205 )

206import Control.Distributed.Process.Serializable (Serializable)

207import Control.Distributed.Process.Internal.Primitives

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

312import Control.Distributed.Process.Internal.Types

313 ( processThread

314− , localState

315− , LocalNodeState(..)

316 , withValidLocalState

314 , withValidLocalState

317 )

315 )

318import Control.Distributed.Process.Internal.Spawn

316import Control.Distributed.Process.Internal.Spawn

324 , spawnSupervised

322 , spawnSupervised

325 , call

323 , call

326 )

324 )

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)

329import Control.Monad (forM_)

327import Control.Monad (forM_)

328+import Data.Accessor ((^.))

330

329

331

330

332-- INTERNAL NOTES

331-- INTERNAL NOTES

426callLocal :: Process a -> Process a

425callLocal :: Process a -> Process a

427callLocal proc = mask $ \release -> do

426callLocal proc = mask $ \release -> do

428 mv <- liftIO newEmptyMVar :: Process (MVar (Either SomeException a))

427 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

428+ child <- spawnLocal $ try (release proc) >>= liftIO . putMVar mv

433− rs <- liftIO (takeMVar mv)

429+ lproc <- ask

434− `onException`

430+ liftIO $ do

431+ rs <- Exception.onException (takeMVar mv) $ uninterruptibleMask_ $

435 -- Exceptions need to be prevented from interrupting the clean up or

432 -- Exceptions need to be prevented from interrupting the clean up or

436 -- the original exception which caused entering the handler could be

433 -- the original exception which caused entering the handler could be

437 -- forgotten. For instance, this could have a problematic effect

434 -- forgotten. For instance, this could have a problematic effect

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

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

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

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

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

437 -- 'System.Timeout.timeout').

441− --

438+ do mchildThreadId <- withValidLocalState (processNode lproc) $

442− -- Using 'uninterruptibleMask_' would be problematic as there is no

439+ \vst -> return $ fmap processThread $

443− -- way to be sure the node isn't closing when the cleanup executes.

440+ vst ^. localProcessWithId (processLocalId child)

444− -- Successfully running the cleanup is necessary to ensure

441+ forM_ mchildThreadId killThread

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)

442+ 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

443+ either throwIO 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