ae6ef8e · distributed-process

git.iko.soy/distributed-process

on master help sign in

Link the node controller and node event listener threads

ae6ef8eTim Watson14 years ago

src/Control/Distributed/Process/Internal/Types.hs+3

compare with master

unifiedsplit

@@ -484,6 +484,7 @@ data ProcessSignal =

484 | Kill !ProcessId !String

485 | Exit !ProcessId !Message

486 | GetInfo !ProcessId

487+ | SigShutdown

487 deriving Show

488 deriving Show

488

489

489--------------------------------------------------------------------------------

490--------------------------------------------------------------------------------

@@ -535,6 +536,7 @@ instance Binary ProcessSignal where

535 put (Kill pid reason) = putWord8 9 >> put pid >> put reason

536 put (Kill pid reason) = putWord8 9 >> put pid >> put reason

536 put (Exit pid reason) = putWord8 10 >> put pid >> put (messageToPayload reason)

537 put (Exit pid reason) = putWord8 10 >> put pid >> put (messageToPayload reason)

537 put (GetInfo about) = putWord8 30 >> put about

538 put (GetInfo about) = putWord8 30 >> put about

539+ put (SigShutdown) = putWord8 31

538 get = do

540 get = do

539 header <- getWord8

541 header <- getWord8

540 case header of

542 case header of

@@ -550,6 +552,7 @@ instance Binary ProcessSignal where

550 9 -> Kill <$> get <*> get

552 9 -> Kill <$> get <*> get

551 10 -> Exit <$> get <*> (payloadToMessage <$> get)

553 10 -> Exit <$> get <*> (payloadToMessage <$> get)

552 30 -> GetInfo <$> get

554 30 -> GetInfo <$> get

555+ 31 -> return SigShutdown

553 _ -> fail "ProcessSignal.get: invalid"

556 _ -> fail "ProcessSignal.get: invalid"

554

557

555instance Binary DiedReason where

558instance Binary DiedReason where

src/Control/Distributed/Process/Node.hs+45−20

@@ -49,7 +49,13 @@ import Control.Monad.IO.Class (MonadIO, liftIO)

49import Control.Monad.State.Strict (MonadState, StateT, evalStateT, gets)

50import qualified Control.Monad.State.Strict as StateT (get, put)

51import Control.Monad.Reader (MonadReader, ReaderT, runReaderT, ask)

52−import Control.Exception (throwIO, SomeException, Exception, throwTo)

52+import Control.Exception

53+ ( throwIO

54+ , AsyncException(ThreadKilled)

55+ , SomeException

56+ , Exception

57+ , throwTo

58+ )

53import qualified Control.Exception as Exception (catch)

59import qualified Control.Exception as Exception (catch)

54import Control.Concurrent (forkIO)

60import Control.Concurrent (forkIO)

55import Control.Distributed.Process.Internal.StrictMVar

61import Control.Distributed.Process.Internal.StrictMVar

@@ -193,25 +199,41 @@ newLocalNode transport rtable = do

193-- | Create a new local node (without any service processes running)

199-- | Create a new local node (without any service processes running)

194createBareLocalNode :: NT.EndPoint -> RemoteTable -> IO LocalNode

200createBareLocalNode :: NT.EndPoint -> RemoteTable -> IO LocalNode

195createBareLocalNode endPoint rtable = do

201createBareLocalNode endPoint rtable = do

196− unq <- randomIO

202+ unq <- randomIO

197− state <- newMVar LocalNodeState

203+ state <- newMVar LocalNodeState

198− { _localProcesses = Map.empty

204+ { _localProcesses = Map.empty

199− , _localPidCounter = firstNonReservedProcessId

205+ , _localPidCounter = firstNonReservedProcessId

200− , _localPidUnique = unq

206+ , _localPidUnique = unq

201− , _localConnections = Map.empty

207+ , _localConnections = Map.empty

202− }

208+ }

203− ctrlChan <- newChan

209+ ctrlChan <- newChan

204− let node = LocalNode { localNodeId = NodeId $ NT.address endPoint

210+ let node = LocalNode { localNodeId = NodeId $ NT.address endPoint

205− , localEndPoint = endPoint

211+ , localEndPoint = endPoint

206− , localState = state

212+ , localState = state

207− , localCtrlChan = ctrlChan

213+ , localCtrlChan = ctrlChan

208− , localTracer = InactiveTracer

214+ , localTracer = InactiveTracer

209− , remoteTable = rtable

215+ , remoteTable = rtable

210− }

216+ }

211− tracedNode <- startTracing node

217+ tracedNode <- startTracing node

218+

219+ ncGo <- newEmptyMVar

220+ ncTid <- forkIO $ do

221+ tid <- takeMVar ncGo

212− void . forkIO $ runNodeController tracedNode

222+ (runNodeController tracedNode

223+ `Exception.catch` \(e :: SomeException) -> throwTo tid e)

224+

225+ evTid <- forkIO $ do

213− void . forkIO $ handleIncomingMessages tracedNode

226+ (handleIncomingMessages tracedNode >> (stopNC node))

227+ `Exception.catch` \(e :: SomeException) -> throwTo ncTid e

228+

229+ putMVar ncGo evTid

214− return tracedNode

230+ return tracedNode

231+ where

232+ stopNC node =

233+ writeChan (localCtrlChan node) NCMsg

234+ { ctrlMsgSender = NodeIdentifier (localNodeId node)

235+ , ctrlMsgSignal = SigShutdown

236+ }

215

237

216-- | Start and register the service processes on a node

238-- | Start and register the service processes on a node

217-- (for now, this is only the logger)

239-- (for now, this is only the logger)

@@ -583,6 +605,8 @@ nodeController = do

583 ncEffectExit from to reason

605 ncEffectExit from to reason

584 NCMsg (ProcessIdentifier from) (GetInfo pid) ->

606 NCMsg (ProcessIdentifier from) (GetInfo pid) ->

585 ncEffectGetInfo from pid

607 ncEffectGetInfo from pid

608+ NCMsg _ SigShutdown ->

609+ liftIO $ throwIO ThreadKilled -- seems to make more sense than fail/error

586 unexpected ->

610 unexpected ->

587 error $ "nodeController: unexpected message " ++ show unexpected

611 error $ "nodeController: unexpected message " ++ show unexpected

588

612

@@ -902,6 +926,7 @@ destNid (Died _ _) = Nothing

902destNid (Kill pid _) = Just $ processNodeId pid

926destNid (Kill pid _) = Just $ processNodeId pid

903destNid (Exit pid _) = Just $ processNodeId pid

927destNid (Exit pid _) = Just $ processNodeId pid

904destNid (GetInfo pid) = Just $ processNodeId pid

928destNid (GetInfo pid) = Just $ processNodeId pid

929+destNid (SigShutdown) = Nothing

905

930

906-- | Check if a process is local to our own node

931-- | Check if a process is local to our own node

907isLocal :: LocalNode -> Identifier -> Bool

932isLocal :: LocalNode -> Identifier -> Bool