263cb13 · distributed-process

git.iko.soy/distributed-process

on master help sign in

Ensure the NC and event loop threads hold only weak refs to one another

263cb13Tim Watson13 years ago

src/Control/Distributed/Process/Node.hs+18−12

compare with master

unifiedsplit

@@ -57,7 +57,7 @@ import Control.Exception

57 , throwTo

58 )

59import qualified Control.Exception as Exception (catch)

60−import Control.Concurrent (forkIO)

60+import Control.Concurrent (forkIO, mkWeakThreadId, ThreadId)

61import Control.Distributed.Process.Internal.StrictMVar

62 ( newMVar

63 , withMVar

@@ -142,12 +142,11 @@ import Control.Distributed.Process.Internal.Types

142 , typedChannelWithId

143 , RegisterReply(..)

144 , WhereIsReply(..)

145− , messageToPayload

146 , payloadToMessage

145 , payloadToMessage

147 , createUnencodedMessage

146 , createUnencodedMessage

148 , runLocalProcess

147 , runLocalProcess

149 , firstNonReservedProcessId

148 , firstNonReservedProcessId

150− , ImplicitReconnect(WithImplicitReconnect, NoImplicitReconnect)

149+ , ImplicitReconnect(WithImplicitReconnect)

151 )

150 )

152import Control.Distributed.Process.Internal.Trace

151import Control.Distributed.Process.Internal.Trace

153 ( TraceArg(..)

152 ( TraceArg(..)

@@ -159,7 +158,6 @@ import Control.Distributed.Process.Serializable (Serializable)

159import Control.Distributed.Process.Internal.Messaging

158import Control.Distributed.Process.Internal.Messaging

160 ( sendBinary

159 ( sendBinary

161 , sendMessage

160 , sendMessage

162− , sendPayload

163 , closeImplicitReconnections

161 , closeImplicitReconnections

164 , impliesDeathOf

162 , impliesDeathOf

165 )

163 )

@@ -222,13 +220,15 @@ createBareLocalNode endPoint rtable = do

222 ncTid <- forkIO $ do

220 ncTid <- forkIO $ do

223 tid <- takeMVar ncGo

221 tid <- takeMVar ncGo

224 (runNodeController tracedNode

222 (runNodeController tracedNode

225− `Exception.catch` \(e :: SomeException) -> throwTo tid e)

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

226

224

225+ weakNcTid <- mkWeakThreadId ncTid

227 evTid <- forkIO $ do

226 evTid <- forkIO $ do

228 (handleIncomingMessages tracedNode >> (stopNC node))

227 (handleIncomingMessages tracedNode >> (stopNC node))

229− `Exception.catch` \(e :: SomeException) -> throwTo ncTid e

228+ `Exception.catch` \(e :: SomeException) -> reThrow weakNcTid e

230

229

230+ weakEvTid <- mkWeakThreadId evTid

231− putMVar ncGo evTid

231+ putMVar ncGo weakEvTid

232 return tracedNode

233 where

234 stopNC node =

@@ -237,6 +237,13 @@ createBareLocalNode endPoint rtable = do

237 , ctrlMsgSignal = SigShutdown

238 }

239

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+

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

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

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

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

242startServiceProcesses :: LocalNode -> IO ()

249startServiceProcesses :: LocalNode -> IO ()

@@ -599,8 +606,8 @@ nodeController = do

599 ncEffectRegister from label atnode pid force

606 ncEffectRegister from label atnode pid force

600 NCMsg (ProcessIdentifier from) (WhereIs label) ->

607 NCMsg (ProcessIdentifier from) (WhereIs label) ->

601 ncEffectWhereIs from label

608 ncEffectWhereIs from label

602− NCMsg from (NamedSend label msg') ->

609+ NCMsg _ (NamedSend label msg') ->

603− ncEffectNamedSend from label msg'

610+ ncEffectNamedSend label msg'

604 NCMsg _ (LocalSend to msg') ->

611 NCMsg _ (LocalSend to msg') ->

605 ncEffectLocalSend to msg'

612 ncEffectLocalSend to msg'

606 NCMsg _ (LocalPortSend to msg') ->

613 NCMsg _ (LocalPortSend to msg') ->

@@ -807,9 +814,8 @@ ncEffectWhereIs from label = do

807 (WhereIsReply label mPid)

814 (WhereIsReply label mPid)

808

815

809-- [Unified: Table 14]

816-- [Unified: Table 14]

810−ncEffectNamedSend :: Identifier -> String -> Message -> NC ()

817+ncEffectNamedSend :: String -> Message -> NC ()

811−ncEffectNamedSend from label msg = do

818+ncEffectNamedSend label msg = do

812− node <- ask

813 mPid <- gets (^. registeredHereFor label)

819 mPid <- gets (^. registeredHereFor label)

814 -- If mPid is Nothing, we just ignore the named send (as per Table 14)

820 -- If mPid is Nothing, we just ignore the named send (as per Table 14)

815 forM_ mPid $ \pid -> postMessage pid msg

821 forM_ mPid $ \pid -> postMessage pid msg