@@ -49,7 +49,13 @@ import Control.Monad.IO.Class (MonadIO, liftIO)
4949import Control.Monad.State.Strict (MonadState, StateT, evalStateT, gets)
5050import qualified Control.Monad.State.Strict as StateT (get, put)
5151import 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+ )
5359import qualified Control.Exception as Exception (catch)
5460import Control.Concurrent (forkIO)
5561import Control.Distributed.Process.Internal.StrictMVar
@@ -195,25 +201,41 @@ newLocalNode transport rtable = do
195201-- | Create a new local node (without any service processes running)
196202createBareLocalNode :: NT.EndPoint -> RemoteTable -> IO LocalNode
197203createBareLocalNode endPoint rtable = do
198− unq <- randomIO
199− state <- newMVar LocalNodeState
200− { _localProcesses = Map.empty
201− , _localPidCounter = firstNonReservedProcessId
202− , _localPidUnique = unq
203− , _localConnections = Map.empty
204− }
205− ctrlChan <- newChan
206− let node = LocalNode { localNodeId = NodeId $ NT.address endPoint
207− , localEndPoint = endPoint
208− , localState = state
209− , localCtrlChan = ctrlChan
210− , localTracer = InactiveTracer
211− , remoteTable = rtable
212− }
213− tracedNode <- startTracing node
214− void . forkIO $ runNodeController tracedNode
215− void . forkIO $ handleIncomingMessages tracedNode
216− return tracedNode
204+ unq <- randomIO
205+ state <- newMVar LocalNodeState
206+ { _localProcesses = Map.empty
207+ , _localPidCounter = firstNonReservedProcessId
208+ , _localPidUnique = unq
209+ , _localConnections = Map.empty
210+ }
211+ ctrlChan <- newChan
212+ let node = LocalNode { localNodeId = NodeId $ NT.address endPoint
213+ , localEndPoint = endPoint
214+ , localState = state
215+ , localCtrlChan = ctrlChan
216+ , localTracer = InactiveTracer
217+ , remoteTable = rtable
218+ }
219+ tracedNode <- startTracing node
220+
221+ ncGo <- newEmptyMVar
222+ ncTid <- forkIO $ do
223+ tid <- takeMVar ncGo
224+ (runNodeController tracedNode
225+ `Exception.catch` \(e :: SomeException) -> throwTo tid e)
226+
227+ evTid <- forkIO $ do
228+ (handleIncomingMessages tracedNode >> (stopNC node))
229+ `Exception.catch` \(e :: SomeException) -> throwTo ncTid e
230+
231+ putMVar ncGo evTid
232+ return tracedNode
233+ where
234+ stopNC node =
235+ writeChan (localCtrlChan node) NCMsg
236+ { ctrlMsgSender = NodeIdentifier (localNodeId node)
237+ , ctrlMsgSignal = SigShutdown
238+ }
217239
218240-- | Start and register the service processes on a node
219241-- (for now, this is only the logger)
@@ -589,6 +611,8 @@ nodeController = do
589611 ncEffectExit from to reason
590612 NCMsg (ProcessIdentifier from) (GetInfo pid) ->
591613 ncEffectGetInfo from pid
614+ NCMsg _ SigShutdown ->
615+ liftIO $ throwIO ThreadKilled -- seems to make more sense than fail/error
592616 unexpected ->
593617 error $ "nodeController: unexpected message " ++ show unexpected
594618
@@ -938,6 +962,7 @@ destNid (Exit pid _) = Just $ processNodeId pid
938962destNid (GetInfo pid) = Just $ processNodeId pid
939963destNid (LocalSend pid _) = Just $ processNodeId pid
940964destNid (LocalPortSend cid _) = Just $ processNodeId (sendPortProcessId cid)
965+destNid (SigShutdown) = Nothing
941966
942967-- | Check if a process is local to our own node
943968isLocal :: LocalNode -> Identifier -> Bool