Link the node controller and node event listener threads

Tim Watson · Tim Watson · commit ae6ef8e9f563 · 2013-02-15T12:52:49.000Z
If either fails, the exception will be rethrown to the other thread. In
addition, when the NT listening loop ends normally (i.e., the endpoint
has been shut down) and control channel signal notifies the NC thread to
terminate as well - no more potential zombies.

DP-39 #resolve
diff --git a/src/Control/Distributed/Process/Internal/Types.hs b/src/Control/Distributed/Process/Internal/Types.hs
@@ -484,6 +484,7 @@ data ProcessSignal =
   | Kill !ProcessId !String
   | Exit !ProcessId !Message
   | GetInfo !ProcessId
+  | SigShutdown
   deriving Show
 
 --------------------------------------------------------------------------------
@@ -535,6 +536,7 @@ instance Binary ProcessSignal where
   put (Kill pid reason)     = putWord8 9 >> put pid >> put reason
   put (Exit pid reason)     = putWord8 10 >> put pid >> put (messageToPayload reason)
   put (GetInfo about)       = putWord8 30 >> put about
+  put (SigShutdown)         = putWord8 31
   get = do
     header <- getWord8
     case header of
@@ -550,6 +552,7 @@ instance Binary ProcessSignal where
       9  -> Kill <$> get <*> get
       10 -> Exit <$> get <*> (payloadToMessage <$> get)
       30 -> GetInfo <$> get
+      31 -> return SigShutdown
       _ -> fail "ProcessSignal.get: invalid"
 
 instance Binary DiedReason where
diff --git a/src/Control/Distributed/Process/Node.hs b/src/Control/Distributed/Process/Node.hs
@@ -49,7 +49,13 @@ import Control.Monad.IO.Class (MonadIO, liftIO)
 import Control.Monad.State.Strict (MonadState, StateT, evalStateT, gets)
 import qualified Control.Monad.State.Strict as StateT (get, put)
 import Control.Monad.Reader (MonadReader, ReaderT, runReaderT, ask)
-import Control.Exception (throwIO, SomeException, Exception, throwTo)
+import Control.Exception
+  ( throwIO
+  , AsyncException(ThreadKilled)
+  , SomeException
+  , Exception
+  , throwTo
+  )
 import qualified Control.Exception as Exception (catch)
 import Control.Concurrent (forkIO)
 import Control.Distributed.Process.Internal.StrictMVar
@@ -193,25 +199,41 @@ newLocalNode transport rtable = do
 -- | Create a new local node (without any service processes running)
 createBareLocalNode :: NT.EndPoint -> RemoteTable -> IO LocalNode
 createBareLocalNode endPoint rtable = do
-  unq <- randomIO
-  state <- newMVar LocalNodeState
-    { _localProcesses   = Map.empty
-    , _localPidCounter  = firstNonReservedProcessId
-    , _localPidUnique   = unq
-    , _localConnections = Map.empty
-    }
-  ctrlChan <- newChan
-  let node = LocalNode { localNodeId   = NodeId $ NT.address endPoint
-                       , localEndPoint = endPoint
-                       , localState    = state
-                       , localCtrlChan = ctrlChan
-                       , localTracer   = InactiveTracer
-                       , remoteTable   = rtable
-                       }
-  tracedNode <- startTracing node
-  void . forkIO $ runNodeController tracedNode
-  void . forkIO $ handleIncomingMessages tracedNode
-  return tracedNode
+    unq <- randomIO
+    state <- newMVar LocalNodeState
+      { _localProcesses   = Map.empty
+      , _localPidCounter  = firstNonReservedProcessId
+      , _localPidUnique   = unq
+      , _localConnections = Map.empty
+      }
+    ctrlChan <- newChan
+    let node = LocalNode { localNodeId   = NodeId $ NT.address endPoint
+                         , localEndPoint = endPoint
+                         , localState    = state
+                         , localCtrlChan = ctrlChan
+                         , localTracer   = InactiveTracer
+                         , remoteTable   = rtable
+                         }
+    tracedNode <- startTracing node
+
+    ncGo <- newEmptyMVar
+    ncTid <- forkIO $ do
+      tid <- takeMVar ncGo
+      (runNodeController tracedNode
+       `Exception.catch` \(e :: SomeException) -> throwTo tid e)
+
+    evTid <- forkIO $ do
+      (handleIncomingMessages tracedNode >> (stopNC node))
+        `Exception.catch` \(e :: SomeException) -> throwTo ncTid e
+
+    putMVar ncGo evTid
+    return tracedNode
+  where
+    stopNC node =
+       writeChan (localCtrlChan node) NCMsg
+            { ctrlMsgSender = NodeIdentifier (localNodeId node)
+            , ctrlMsgSignal = SigShutdown
+            }
 
 -- | Start and register the service processes on a node
 -- (for now, this is only the logger)
@@ -583,6 +605,8 @@ nodeController = do
         ncEffectExit from to reason
       NCMsg (ProcessIdentifier from) (GetInfo pid) ->
         ncEffectGetInfo from pid
+      NCMsg _ SigShutdown ->
+        liftIO $ throwIO ThreadKilled -- seems to make more sense than fail/error
       unexpected ->
         error $ "nodeController: unexpected message " ++ show unexpected
 
@@ -902,6 +926,7 @@ destNid (Died _ _)          = Nothing
 destNid (Kill pid _)        = Just $ processNodeId pid
 destNid (Exit pid _)        = Just $ processNodeId pid
 destNid (GetInfo pid)       = Just $ processNodeId pid
+destNid (SigShutdown)       = Nothing
 
 -- | Check if a process is local to our own node
 isLocal :: LocalNode -> Identifier -> Bool
