03b2197 · distributed-process

git.iko.soy/distributed-process

on master help sign in

Fix MxSend tracing for ordinary primitives

03b2197Tim Watson8 years ago

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

compare with master

unifiedsplit

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

33import Control.Rematch (equalTo)

34import Data.Binary

35import Data.List (find, sort)

36−import Data.Maybe (isJust, isNothing)

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

37import Data.Typeable

38import GHC.Generics hiding (from)

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

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

323

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

325

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

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

327+ -> Maybe LocalNode

328+ -> TestResult Bool

329+ -> Process ()

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

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

328 let delay = 5000000

331 let delay = 5000000

329 let label = "testMxSend"

332 let label = "testMxSend"

333+ let isValid = isValidLabel n label

330

334

331 register label p1

335 register label p1

332 reg1 <- receiveChanTimeout delay sink

336 reg1 <- receiveChanTimeout delay sink

333 case reg1 of

337 case reg1 of

334 Just (MxRegistered pd lb)

338 Just (MxRegistered pd lb)

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

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

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

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

337

341

338 op label ()

342 op label ()

339

343

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

341 sent <- receiveChanTimeout delay sink

345 sent <- receiveChanTimeout delay sink

342 case sent of

346 case sent of

343 Just (MxSentToName lb by _)

347 Just (MxSentToName lb by _)

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

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

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

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

346

356

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

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

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

358testSend 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

254 node = processNode proc

255 nodeId = localNodeId node

256 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.

266 liftIO $ traceEvent (localEventBus node)

257 liftIO $ traceEvent (localEventBus node)

267 (MxSent them us (createUnencodedMessage msg))

258 (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

268

266

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

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

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

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

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

284--

282--

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

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

286usend them msg = do

284usend them msg = do

287− here <- getSelfNode

285+ proc <- ask

288 let there = processNodeId them

286 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

290 then sendLocal them msg

291 then sendLocal them msg

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

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

292 (createMessage msg)

293 (createMessage msg)

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

484 us = processId proc

485 us = processId proc

485 nid = localNodeId node

486 nid = localNodeId node

486 destNode = (processNodeId them) in do

487 destNode = (processNodeId them) in do

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

487− case destNode == nid of

489+ if destNode == nid

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

490+ then sendCtrlMsg Nothing (LocalSend them msg)

489− False ->

491+ else liftIO $ sendPayload (processNode proc)

490− liftIO $ do

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

492+ (ProcessIdentifier (processId proc))

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

493+ (ProcessIdentifier them)

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

494+ NoImplicitReconnect

494− liftIO $ traceEvent (localEventBus node)

495+ (messageToPayload msg)

495− (MxSent them us msg)

496

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

498--

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

506 us = processId proc

507 nid = localNodeId node

508 destNode = (processNodeId them) in do

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

509− case destNode == nid of

510+ if destNode == nid

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

511+ then sendCtrlMsg Nothing (LocalSend them msg)

511− False -> do

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

512+ else 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)

517

513

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

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

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

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

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

1222

1218

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

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

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

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

1225−nsend label 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 (createUnencodedMessage msg))

1226+ sendCtrlMsg Nothing (NamedSend label msg')

1227

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

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

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

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

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

1237nsendRemote nid label msg = do

1238− here <- getSelfNode

1238+ proc <- ask

1239+ let us = processId proc

1240+ let node = processNode proc

1241+ if localNodeId node == nid

1239− if here == nid then nsend label msg

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))

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

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

1241

1247

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

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

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

1249-- 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 ()

104nsend label msg = do

105 proc <- ask

106 let us = processId proc

107− let node = localNodeId (processNode proc)

108 let msg' = wrapMessage msg

107 let msg' = wrapMessage msg

109 -- see [note: tracing]

108 -- see [note: tracing]

110 liftIO $ traceEvent (localEventBus (processNode proc))

109 liftIO $ traceEvent (localEventBus (processNode proc))

111 (MxSentToName label us msg')

110 (MxSentToName label us msg')

112 sendCtrlMsg Nothing (NamedSend label msg')

111 sendCtrlMsg Nothing (NamedSend label msg')

113

112

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−

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

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

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

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

132nsendRemote nid label msg = do

115nsendRemote nid label msg = do

133 proc <- ask

116 proc <- ask

134 let us = processId proc

117 let us = processId proc

135 let node = processNode proc

118 let node = processNode proc

136− if localNodeId (processNode proc) == nid

119+ if localNodeId node == nid

137 then nsend label msg

120 then nsend label msg

138 else

121 else

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

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

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

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

141 liftIO $ traceEvent (localEventBus node)

124 liftIO $ traceEvent (localEventBus node)

142− (MxSentToName label us (wrapMessage msg))

125+ (MxSentToName lbl us (wrapMessage msg))

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

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

144

127

145-- | Send a message

128-- | Send a message

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

176 let there = processNodeId them

159 let there = processNodeId them

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

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

178 let msg' = wrapMessage msg

161 let msg' = wrapMessage msg

162+ -- see [note: tracing]

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

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

180 if localNodeId (processNode proc) == there

164 if localNodeId (processNode proc) == there

181 then sendCtrlMsg Nothing $ LocalSend them msg'

165 then sendCtrlMsg Nothing $ LocalSend them msg'

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

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

183 (createMessage msg)

167 (createMessage msg)

184

168

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+

185-- | Send a message on a typed channel

177-- | Send a message on a typed channel

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

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

187sendChan (SendPort cid) msg = do

179sendChan (SendPort cid) msg = do