03b2197 · distributed-process

git.iko.soy/distributed-process

Fix MxSend tracing for ordinary primitives

commit03b21974e9159d38242015f682424dd45c52e612 authorTim Watson <watson.timothy@gmail.com> dateSun 18 Nov 2018, 01:32 (8 years ago) parent0514f3c

3 files changed+66−58

compare with master

unifiedsplit

distributed-process-tests/src/Control/Distributed/Process/Tests/Mx.hs+16−6 src/Control/Distributed/Process/Internal/Primitives.hs+39−33 src/Control/Distributed/Process/UnsafePrimitives.hs+11−19

distributed-process-tests/src/Control/Distributed/Process/Tests/Mx.hs+16−6

@@ -33,7 +33,7 @@ import Control.Monad.Catch(finally, bracket, try)

3333import Control.Rematch (equalTo)

3434import Data.Binary

3535import Data.List (find, sort)

36−import Data.Maybe (isJust, isNothing)

36+import Data.Maybe (isJust, fromJust, isNothing)

3737import Data.Typeable

3838import GHC.Generics hiding (from)

3939#if ! MIN_VERSION_base(4,6,0)

@@ -323,17 +323,21 @@ ensure = flip finally

323323

324324type SendTest = ProcessId -> ReceivePort MxEvent -> Process Bool

325325

326−testNSend :: (String -> () -> Process ()) -> Maybe LocalNode -> TestResult Bool -> Process ()

326+testNSend :: (String -> () -> Process ())

327+ -> Maybe LocalNode

328+ -> TestResult Bool

329+ -> Process ()

327330testNSend op n r = testMxSend n r $ \p1 sink -> do

328331 let delay = 5000000

329332 let label = "testMxSend"

333+ let isValid = isValidLabel n label

330334

331335 register label p1

332336 reg1 <- receiveChanTimeout delay sink

333337 case reg1 of

334338 Just (MxRegistered pd lb)

335− | pd == p1 && lb == label -> return ()

336− _ -> die $ "Reg-Failed: " ++ show reg1

339+ | pd == p1 && isValid lb -> return ()

340+ _ -> die $ "Reg-Failed: " ++ show reg1

337341

338342 op label ()

339343

@@ -341,8 +345,14 @@ testNSend op n r = testMxSend n r $ \p1 sink -> do

341345 sent <- receiveChanTimeout delay sink

342346 case sent of

343347 Just (MxSentToName lb by _)

344− | by == us && lb == label -> return True

345− _ -> die $ "Send-Failed: " ++ show sent

348+ | by == us && isValid lb -> return True

349+ _ -> die $ "Send-Failed: " ++ show sent

350+

351+ where

352+ isValidLabel n l1 l2

353+ | l2 == l2 = True

354+ | isJust n = l2 == l1 ++ "@" ++ show (localNodeId $ fromJust n)

355+ | otherwise = False

346356

347357testSend :: (ProcessId -> () -> Process ()) -> Maybe LocalNode -> TestResult Bool -> Process ()

348358testSend op n r = testMxSend n r $ \p1 sink -> do

src/Control/Distributed/Process/Internal/Primitives.hs+39−33

@@ -254,17 +254,15 @@ send them msg = do

254254 node = processNode proc

255255 nodeId = localNodeId node

256256 destNode = (processNodeId them) in do

257− case destNode == nodeId of

258− True -> sendLocal them msg

259− False -> liftIO $ sendMessage (processNode proc)

260− (ProcessIdentifier (processId proc))

261− (ProcessIdentifier them)

262− NoImplicitReconnect

263− msg

264− -- We do not fire the trace event until after the sending is complete;

265− -- In the remote case, 'sendMessage' can block in the networking stack.

266257 liftIO $ traceEvent (localEventBus node)

267258 (MxSent them us (createUnencodedMessage msg))

259+ if destNode == nodeId

260+ then sendLocal them msg

261+ else liftIO $ sendMessage (processNode proc)

262+ (ProcessIdentifier (processId proc))

263+ (ProcessIdentifier them)

264+ NoImplicitReconnect

265+ msg

268266

269267-- | /Unsafe/ variant of 'send'. This function makes /no/ attempt to serialize

270268-- and (in the case when the destination process resides on the same local

@@ -284,9 +282,12 @@ unsafeSend = Unsafe.send

284282--

285283usend :: Serializable a => ProcessId -> a -> Process ()

286284usend them msg = do

287− here <- getSelfNode

285+ proc <- ask

288286 let there = processNodeId them

289− if here == there

287+ let (us, node) = (processId proc, processNode proc)

288+ let msg' = wrapMessage msg

289+ liftIO $ traceEvent (localEventBus node) (MxSent them us msg')

290+ if localNodeId (processNode proc) == there

290291 then sendLocal them msg

291292 else sendCtrlMsg (Just there) $ UnreliableSend (processLocalId them)

292293 (createMessage msg)

@@ -484,15 +485,14 @@ forward msg them = do

484485 us = processId proc

485486 nid = localNodeId node

486487 destNode = (processNodeId them) in do

487− case destNode == nid of

488− True -> sendCtrlMsg Nothing (LocalSend them msg)

489− False ->

490− liftIO $ do

491− sendPayload (processNode proc) (ProcessIdentifier (processId proc)) (ProcessIdentifier them) NoImplicitReconnect (messageToPayload msg)

492− -- We do not fire the trace event until after the sending is complete;

493− -- In this remote case, 'sendMessage' can block in the networking stack.

494− liftIO $ traceEvent (localEventBus node)

495− (MxSent them us msg)

488+ liftIO $ traceEvent (localEventBus node) (MxSent them us msg)

489+ if destNode == nid

490+ then sendCtrlMsg Nothing (LocalSend them msg)

491+ else liftIO $ sendPayload (processNode proc)

492+ (ProcessIdentifier (processId proc))

493+ (ProcessIdentifier them)

494+ NoImplicitReconnect

495+ (messageToPayload msg)

496496

497497-- | Forward a raw 'Message' to the given 'ProcessId'.

498498--

@@ -506,14 +506,10 @@ uforward msg them = do

506506 us = processId proc

507507 nid = localNodeId node

508508 destNode = (processNodeId them) in do

509− case destNode == nid of

510− True -> sendCtrlMsg Nothing (LocalSend them msg)

511− False -> do

512− sendCtrlMsg (Just destNode) $ UnreliableSend (processLocalId them) msg

513− -- We do not fire the trace event until after the sending is complete;

514− -- In the remote case, 'sendCtrlMsg' can block in the networking stack.

515− liftIO $ traceEvent (localEventBus node)

516− (MxSent them us msg)

509+ liftIO $ traceEvent (localEventBus node) (MxSent them us msg)

510+ if destNode == nid

511+ then sendCtrlMsg Nothing (LocalSend them msg)

512+ else sendCtrlMsg (Just destNode) $ UnreliableSend (processLocalId them) msg

517513

518514-- | Wrap a 'Serializable' value in a 'Message'. Note that 'Message's are

519515-- 'Serializable' - like the datum they contain - but also note, deserialising

@@ -1222,8 +1218,12 @@ whereisRemoteAsync nid label = do

12221218

12231219-- | Named send to a process in the local registry (asynchronous)

12241220nsend :: Serializable a => String -> a -> Process ()

1225−nsend label msg =

1226− sendCtrlMsg Nothing (NamedSend label (createUnencodedMessage msg))

1221+nsend label msg = do

1222+ proc <- ask

1223+ let msg' = createUnencodedMessage msg

1224+ liftIO $ traceEvent (localEventBus (processNode proc))

1225+ (MxSentToName label (processId proc) msg')

1226+ sendCtrlMsg Nothing (NamedSend label msg')

12271227

12281228-- | Named send to a process in the local registry (asynchronous).

12291229-- This function makes /no/ attempt to serialize and (in the case when the

@@ -1235,9 +1235,15 @@ unsafeNSend = Unsafe.nsend

12351235-- | Named send to a process in a remote registry (asynchronous)

12361236nsendRemote :: Serializable a => NodeId -> String -> a -> Process ()

12371237nsendRemote nid label msg = do

1238− here <- getSelfNode

1239− if here == nid then nsend label msg

1240− else sendCtrlMsg (Just nid) (NamedSend label (createMessage msg))

1238+ proc <- ask

1239+ let us = processId proc

1240+ let node = processNode proc

1241+ if localNodeId node == nid

1242+ then nsend label msg

1243+ else let lbl = label ++ "@" ++ show nid in do

1244+ liftIO $ traceEvent (localEventBus node)

1245+ (MxSentToName lbl us (wrapMessage msg))

1246+ sendCtrlMsg (Just nid) (NamedSend label (createMessage msg))

12411247

12421248-- | Named send to a process in a remote registry (asynchronous)

12431249-- This function makes /no/ attempt to serialize and (in the case when the

src/Control/Distributed/Process/UnsafePrimitives.hs+11−19

@@ -104,42 +104,25 @@ nsend :: Serializable a => String -> a -> Process ()

104104nsend label msg = do

105105 proc <- ask

106106 let us = processId proc

107− let node = localNodeId (processNode proc)

108107 let msg' = wrapMessage msg

109108 -- see [note: tracing]

110109 liftIO $ traceEvent (localEventBus (processNode proc))

111110 (MxSentToName label us msg')

112111 sendCtrlMsg Nothing (NamedSend label msg')

113112

114−-- [note: tracing]

115−-- In the remote case, we do not fire a trace event until after sending is

116−-- complete, since 'sendMessage' can block in the networking stack. For locally

117−-- registered processes, we trace before sending the NC a control message, since

118−-- we cannot guarantee ordering otherwise.

119−--

120−-- In addition, MxSent trace messages are dispatched by the node controller's

121−-- thread for local sends, which also covers usend.

122−--

123−-- Also note that tracing writes to the local node's control channel, and this

124−-- module explicitly specifies to its clients that it does unsafe message

125−-- encoding. The same is true for the messages it puts onto the Management

126−-- event bus, however we do *not* want unevaluated thunks hitting the event

127−-- bus control thread. Hence the word /Unsafe/ in this module's name!

128−--

129−

130113-- | Named send to a process in a remote registry (asynchronous)

131114nsendRemote :: Serializable a => NodeId -> String -> a -> Process ()

132115nsendRemote nid label msg = do

133116 proc <- ask

134117 let us = processId proc

135118 let node = processNode proc

136− if localNodeId (processNode proc) == nid

119+ if localNodeId node == nid

137120 then nsend label msg

138121 else

139122 let lbl = label ++ "@" ++ show nid in do

140123 -- see [note: tracing] NB: this is a remote call to another NC...

141124 liftIO $ traceEvent (localEventBus node)

142− (MxSentToName label us (wrapMessage msg))

125+ (MxSentToName lbl us (wrapMessage msg))

143126 sendCtrlMsg (Just nid) (NamedSend label (createMessage msg))

144127

145128-- | Send a message

@@ -176,12 +159,21 @@ usend them msg = do

176159 let there = processNodeId them

177160 let (us, node) = (processId proc, processNode proc)

178161 let msg' = wrapMessage msg

162+ -- see [note: tracing]

179163 liftIO $ traceEvent (localEventBus node) (MxSent them us msg')

180164 if localNodeId (processNode proc) == there

181165 then sendCtrlMsg Nothing $ LocalSend them msg'

182166 else sendCtrlMsg (Just there) $ UnreliableSend (processLocalId them)

183167 (createMessage msg)

184168

169+-- [note: tracing]

170+-- Note that tracing writes to the local node's control channel, and this

171+-- module explicitly specifies to its clients that it does unsafe message

172+-- encoding. The same is true for the messages it puts onto the Management

173+-- event bus, however we do *not* want unevaluated thunks hitting the event

174+-- bus control thread. Hence the word /Unsafe/ in this module's name!

175+--

176+

185177-- | Send a message on a typed channel

186178sendChan :: Serializable a => SendPort a -> a -> Process ()

187179sendChan (SendPort cid) msg = do