@@ -57,7 +57,7 @@ import Control.Exception
5757 , throwTo
5858 )
5959import qualified Control.Exception as Exception (catch)
60−import Control.Concurrent (forkIO)
60+import Control.Concurrent (forkIO, mkWeakThreadId, ThreadId)
6161import Control.Distributed.Process.Internal.StrictMVar
6262 ( newMVar
6363 , withMVar
@@ -142,12 +142,11 @@ import Control.Distributed.Process.Internal.Types
142142 , typedChannelWithId
143143 , RegisterReply(..)
144144 , WhereIsReply(..)
145− , messageToPayload
146145 , payloadToMessage
147146 , createUnencodedMessage
148147 , runLocalProcess
149148 , firstNonReservedProcessId
150− , ImplicitReconnect(WithImplicitReconnect, NoImplicitReconnect)
149+ , ImplicitReconnect(WithImplicitReconnect)
151150 )
152151import Control.Distributed.Process.Internal.Trace
153152 ( TraceArg(..)
@@ -159,7 +158,6 @@ import Control.Distributed.Process.Serializable (Serializable)
159158import Control.Distributed.Process.Internal.Messaging
160159 ( sendBinary
161160 , sendMessage
162− , sendPayload
163161 , closeImplicitReconnections
164162 , impliesDeathOf
165163 )
@@ -222,13 +220,15 @@ createBareLocalNode endPoint rtable = do
222220 ncTid <- forkIO $ do
223221 tid <- takeMVar ncGo
224222 (runNodeController tracedNode
225− `Exception.catch` \(e :: SomeException) -> throwTo tid e)
223+ `Exception.catch` \(e :: SomeException) -> reThrow tid e)
226224
225+ weakNcTid <- mkWeakThreadId ncTid
227226 evTid <- forkIO $ do
228227 (handleIncomingMessages tracedNode >> (stopNC node))
229− `Exception.catch` \(e :: SomeException) -> throwTo ncTid e
228+ `Exception.catch` \(e :: SomeException) -> reThrow weakNcTid e
230229
231− putMVar ncGo evTid
230+ weakEvTid <- mkWeakThreadId evTid
231+ putMVar ncGo weakEvTid
232232 return tracedNode
233233 where
234234 stopNC node =
@@ -237,6 +237,13 @@ createBareLocalNode endPoint rtable = do
237237 , ctrlMsgSignal = SigShutdown
238238 }
239239
240+ reThrow :: (Exception e) => Weak ThreadId -> e -> IO ()
241+ reThrow wTid ex = do
242+ tid <- deRefWeak wTid
243+ case tid of
244+ Nothing -> return ()
245+ Just t -> throwTo t ex
246+
240247-- | Start and register the service processes on a node
241248-- (for now, this is only the logger)
242249startServiceProcesses :: LocalNode -> IO ()
@@ -599,8 +606,8 @@ nodeController = do
599606 ncEffectRegister from label atnode pid force
600607 NCMsg (ProcessIdentifier from) (WhereIs label) ->
601608 ncEffectWhereIs from label
602− NCMsg from (NamedSend label msg') ->
603− ncEffectNamedSend from label msg'
609+ NCMsg _ (NamedSend label msg') ->
610+ ncEffectNamedSend label msg'
604611 NCMsg _ (LocalSend to msg') ->
605612 ncEffectLocalSend to msg'
606613 NCMsg _ (LocalPortSend to msg') ->
@@ -807,9 +814,8 @@ ncEffectWhereIs from label = do
807814 (WhereIsReply label mPid)
808815
809816-- [Unified: Table 14]
810−ncEffectNamedSend :: Identifier -> String -> Message -> NC ()
811−ncEffectNamedSend from label msg = do
812− node <- ask
817+ncEffectNamedSend :: String -> Message -> NC ()
818+ncEffectNamedSend label msg = do
813819 mPid <- gets (^. registeredHereFor label)
814820 -- If mPid is Nothing, we just ignore the named send (as per Table 14)
815821 forM_ mPid $ \pid -> postMessage pid msg