0514f3c · distributed-process

git.iko.soy/distributed-process

on master help sign in

Fix MxSend trace messages for unsafe primitives

0514f3cTim Watson8 years ago

distributed-process-tests/src/Control/Distributed/Process/Tests/Internal/Utils.hs+2−1

compare with master

unifiedsplit

@@ -74,12 +74,13 @@ import Control.Concurrent

74import Control.Concurrent.MVar

75 ( putMVar

76 )

77−import Control.Distributed.Process

77+import Control.Distributed.Process hiding (finally, catch)

78import Control.Distributed.Process.Node

79import Control.Distributed.Process.Serializable()

81import Control.Exception (AsyncException(ThreadKilled), SomeException)

82import Control.Monad (forever, void)

83+import Control.Monad.Catch (finally, catch)

83import Control.Monad.STM (atomically)

84import Control.Monad.STM (atomically)

84import Control.Rematch hiding (match)

85import Control.Rematch hiding (match)

85import Control.Rematch.Run

86import Control.Rematch.Run

distributed-process-tests/src/Control/Distributed/Process/Tests/Mx.hs+125−61

@@ -3,8 +3,9 @@ module Control.Distributed.Process.Tests.Mx (tests) where

4import Control.Distributed.Process.Tests.Internal.Utils

5import Network.Transport.Test (TestTransport(..))

6−

6+import Control.Exception (SomeException)

7−import Control.Distributed.Process

7+import Control.Distributed.Process hiding (bracket, finally, try)

8+import Control.Distributed.Process.Internal.Types (ProcessExitException(..))

8import Control.Distributed.Process.Node

9import Control.Distributed.Process.Node

9import qualified Control.Distributed.Process.UnsafePrimitives as Unsafe

10import qualified Control.Distributed.Process.UnsafePrimitives as Unsafe

10 ( send

11 ( send

@@ -28,12 +29,13 @@ import Control.Distributed.Process.Management

28 , mxBroadcast

29 , mxBroadcast

29 )

30 )

30import Control.Monad (void, unless)

31import Control.Monad (void, unless)

32+import Control.Monad.Catch(finally, bracket, try)

31import Control.Rematch (equalTo)

33import Control.Rematch (equalTo)

32import Data.Binary

34import Data.Binary

33import Data.List (find, sort)

35import Data.List (find, sort)

34import Data.Maybe (isJust, isNothing)

36import Data.Maybe (isJust, isNothing)

35import Data.Typeable

37import Data.Typeable

36−import GHC.Generics

38+import GHC.Generics hiding (from)

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

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

38import Prelude hiding (catch, log)

40import Prelude hiding (catch, log)

39#endif

41#endif

@@ -57,7 +59,7 @@ awaitExit pid =

57 (\_ -> return ())

59 (\_ -> return ())

58 ]

60 ]

59 where

61 where

60− withMonitorRef pid = bracket (monitor pid) unmonitor

62+ withMonitorRef p = bracket (monitor p) unmonitor

62testAgentBroadcast :: TestResult () -> Process ()

64testAgentBroadcast :: TestResult () -> Process ()

63testAgentBroadcast result = do

65testAgentBroadcast result = do

@@ -319,101 +321,163 @@ testMxRegMon remoteNode result = do

319ensure :: Process () -> Process () -> Process ()

321ensure :: Process () -> Process () -> Process ()

320ensure = flip finally

322ensure = flip finally

321

323

322−

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

323−testMxUnsafeSendEvents :: LocalNode -> Process ()

325+

324−testMxUnsafeSendEvents remoteNode = do

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

325−

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

326− -- ensure that when a registered process dies, we get a notification that

328+ let delay = 5000000

327− -- it has been unregistered as well as seeing the name get removed

329+ let label = "testMxSend"

328−

330+

329− let label1 = "aaaaa"

331+ register label p1

330− let label2 = "bbbbb"

332+ reg1 <- receiveChanTimeout delay sink

331− let isValid l = l ==label1 || l == label2

333+ case reg1 of

332− let agentLabel = "listener-agent"

334+ Just (MxRegistered pd lb)

333− let delay = 1000000

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

334− (regChan, regSink) <- newChan

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

335− (unRegChan, unRegSink) <- newChan

337+

338+ op label ()

339+

340+ us <- getSelfPid

341+ sent <- receiveChanTimeout delay sink

342+ case sent of

343+ Just (MxSentToName lb by _)

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

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

346+

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

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

349+ -- initiate a send

350+ op p1 ()

351+

352+ -- verify the management event

353+ us <- getSelfPid

354+ sent <- receiveChanTimeout 5000000 sink

355+ case sent of

356+ Just (MxSent pidTo pidFrom _)

357+ | pidTo == p1 && pidFrom == us -> return True

358+ _ -> return False

359+

360+testMxSend :: Maybe LocalNode -> TestResult Bool -> SendTest -> Process ()

361+testMxSend mNode result test = do

362+ us <- getSelfPid

363+ (chan, sink) <- newChan

336− agent <- mxAgent (MxAgentId agentLabel) () [

364+ agent <- mxAgent (MxAgentId $ agentLabel us) () [

337 mxSink $ \ev -> do

365 mxSink $ \ev -> do

338 case ev of

366 case ev of

367+ m@(MxSent _ fromPid _)

368+ | fromPid == us -> liftMX $ sendChan chan m

369+ m@(MxSentToName _ fromPid _)

370+ | fromPid == us -> liftMX $ sendChan chan m

339− MxRegistered pid label

371+ m@(MxRegistered _ name)

340− | isValid label -> liftMX $ sendChan regChan (label, pid)

341− MxUnRegistered pid label

342− | isValid label -> liftMX $ sendChan unRegChan (label, pid)

372+ | name == label -> liftMX $ sendChan chan m

343 _ -> return ()

373 _ -> return ()

344 mxReady

374 mxReady

345 ]

375 ]

346

376

347 (sp, rp) <- newChan

377 (sp, rp) <- newChan

378+ case mNode of

379+ Nothing -> void $ spawnLocal (proc sp)

348− liftIO $ forkProcess remoteNode $ do

380+ Just remoteNode -> void $ liftIO $ forkProcess remoteNode $ proc sp

349− getSelfPid >>= sendChan sp

350− expect :: Process ()

351

381

352 p1 <- receiveChan rp

382 p1 <- receiveChan rp

383+ res <- try (test p1 sink)

384+ case res of

385+ Left (ProcessExitException _ m) -> (liftIO $ putStrLn $ "SomeException-" ++ show m) >> stash result False

386+ Right tr -> stash result tr

387+ kill agent "bye"

388+ kill p1 "bye"

353

389

354− register label1 p1

390+ where

355− reg1 <- receiveChanTimeout delay regSink

391+ label = "testMxSend"

356− reg1 `shouldBe` equalTo (Just (label1, p1))

392+ agentLabel s = "mx-unsafe-check-agent-" ++ show s

357−

393+ proc sp' = getSelfPid >>= sendChan sp' >> expect :: Process ()

358− register label2 p1

359− reg2 <- receiveChanTimeout delay regSink

360− reg2 `shouldBe` equalTo (Just (label2, p1))

361−

362− n1 <- whereis label1

363− n1 `shouldBe` equalTo (Just p1)

364−

365− n2 <- whereis label2

366− n2 `shouldBe` equalTo (Just p1)

367−

368− kill p1 "goodbye"

369−

370− unreg1 <- receiveChanTimeout delay unRegSink

371− unreg2 <- receiveChanTimeout delay unRegSink

372−

373− sort [unreg1, unreg2]

374− `shouldBe` equalTo [Just (label1, p1), Just (label2, p1)]

375−

376− kill agent "test-complete"

377

394

378tests :: TestTransport -> IO [Test]

395tests :: TestTransport -> IO [Test]

379tests TestTransport{..} = do

396tests TestTransport{..} = do

380 node1 <- newLocalNode testTransport initRemoteTable

397 node1 <- newLocalNode testTransport initRemoteTable

381 node2 <- newLocalNode testTransport initRemoteTable

398 node2 <- newLocalNode testTransport initRemoteTable

399+ let nid = localNodeId node2

382 return [

400 return [

383− testGroup "Mx Agents" [

401+ testGroup "MxAgents" [

384− testCase "Event Handling"

402+ testCase "EventHandling"

385 (delayedAssertion

403 (delayedAssertion

386 "expected True, but events where not as expected"

404 "expected True, but events where not as expected"

387 node1 True testAgentEventHandling)

405 node1 True testAgentEventHandling)

388− , testCase "Inter-Agent Broadcast"

406+ , testCase "InterAgentBroadcast"

389 (delayedAssertion

407 (delayedAssertion

390 "expected (), but no broadcast was received"

408 "expected (), but no broadcast was received"

391 node1 () testAgentBroadcast)

409 node1 () testAgentBroadcast)

392− , testCase "Agent Mailbox Handling"

410+ , testCase "AgentMailboxHandling"

393 (delayedAssertion

411 (delayedAssertion

394 "expected (Just ()), but no regular (mailbox) input was handled"

412 "expected (Just ()), but no regular (mailbox) input was handled"

395 node1 (Just ()) testAgentMailboxHandling)

413 node1 (Just ()) testAgentMailboxHandling)

396− , testCase "Agent Dual Input Handling"

414+ , testCase "AgentDualInputHandling"

397 (delayedAssertion

415 (delayedAssertion

398 "expected sum = 15, but the result was Nothing"

416 "expected sum = 15, but the result was Nothing"

399 node1 (Just 15 :: Maybe Int) testAgentDualInput)

417 node1 (Just 15 :: Maybe Int) testAgentDualInput)

400− , testCase "Agent Input Prioritisation"

418+ , testCase "AgentInputPrioritisation"

401 (delayedAssertion

419 (delayedAssertion

402 "expected [first, second, third, fourth, fifth], but result diverged"

420 "expected [first, second, third, fourth, fifth], but result diverged"

403 node1 (sort ["first", "second",

421 node1 (sort ["first", "second",

404 "third", "fourth",

422 "third", "fourth",

405 "fifth"]) testAgentPrioritisation)

423 "fifth"]) testAgentPrioritisation)

406 ]

424 ]

407− , testGroup "Mx Events" [

425+ , testGroup "MxEvents" [

408− testCase "Name Registration Events"

426+ testCase "NameRegistrationEvents"

409 (delayedAssertion

427 (delayedAssertion

410 "expected registration events to map to the correct ProcessId"

428 "expected registration events to map to the correct ProcessId"

411 node1 () testMxRegEvents)

429 node1 () testMxRegEvents)

412− , testCase "Post Death Name UnRegistration Events"

430+ , testCase "PostDeathNameUnRegistrationEvents"

413 (delayedAssertion

431 (delayedAssertion

414 "expected process deaths to result in unregistration events"

432 "expected process deaths to result in unregistration events"

415 node1 () (testMxRegMon node2))

433 node1 () (testMxRegMon node2))

434+ , testGroup "SentEvents" [

435+ testGroup "RemoteTargets" [

416− , testCase "Monitor Events"

436+ testCase "Unsafe.nsend"

417− (runProcess node1 (testMxUnsafeSendEvents node2))

437+ (delayedAssertion "expected mx events failed"

438+ node1 True (testNSend Unsafe.nsend $ Just node2))

439+ , testCase "Unsafe.nsendRemote"

440+ (delayedAssertion "expected mx events failed"

441+ node1 True (testNSend (Unsafe.nsendRemote nid) $ Just node2))

442+ , testCase "Unsafe.send"

443+ (delayedAssertion "expected mx events failed"

444+ node1 True (testSend Unsafe.send $ Just node2))

445+ , testCase "Unsafe.usend"

446+ (delayedAssertion "expected mx events failed"

447+ node1 True (testSend Unsafe.usend $ Just node2))

448+ , testCase "nsend"

449+ (delayedAssertion "expected mx events failed"

450+ node1 True (testNSend nsend $ Just node2))

451+ , testCase "nsendRemote"

452+ (delayedAssertion "expected mx events failed"

453+ node1 True (testNSend (nsendRemote nid) $ Just node2))

454+ , testCase "send"

455+ (delayedAssertion "expected mx events failed"

456+ node1 True (testSend send $ Just node2))

457+ , testCase "usend"

458+ (delayedAssertion "expected mx events failed"

459+ node1 True (testSend usend $ Just node2))

460+ ]

461+ , testGroup "LocalTargets" [

462+ testCase "Unsafe.nsend"

463+ (delayedAssertion "expected mx events failed"

464+ node1 True (testNSend Unsafe.nsend Nothing))

465+ , testCase "Unsafe.send"

466+ (delayedAssertion "expected mx events failed"

467+ node1 True (testSend Unsafe.send Nothing))

468+ , testCase "Unsafe.usend"

469+ (delayedAssertion "expected mx events failed"

470+ node1 True (testSend Unsafe.usend Nothing))

471+ , testCase "nsend"

472+ (delayedAssertion "expected mx events failed"

473+ node1 True (testNSend nsend Nothing))

474+ , testCase "send"

475+ (delayedAssertion "expected mx events failed"

476+ node1 True (testSend send Nothing))

477+ , testCase "usend"

478+ (delayedAssertion "expected mx events failed"

479+ node1 True (testSend usend Nothing))

480+ ]

481+ ]

482+ ]

418 ]

483 ]

419− ]

src/Control/Distributed/Process/UnsafePrimitives.hs+28−30

@@ -113,10 +113,12 @@ nsend label msg = do

113

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.

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.

117--

119--

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

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

119−-- thread for local sends, which also covers local nsend, and local usend.

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

120--

122--

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

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

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

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

@@ -133,10 +135,12 @@ nsendRemote nid label msg = do

133 let node = processNode proc

135 let node = processNode proc

134 if localNodeId (processNode proc) == nid

136 if localNodeId (processNode proc) == nid

135 then nsend label msg

137 then nsend label msg

138+ else

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

136− else do -- see [note: tracing] NB: this is a remote call to another NC...

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

137− sendCtrlMsg (Just nid) (NamedSend label (createMessage msg))

138− liftIO $ traceEvent (localEventBus node)

141+ liftIO $ traceEvent (localEventBus node)

139− (MxSentToName label us (wrapMessage msg))

142+ (MxSentToName label us (wrapMessage msg))

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

140

144

141-- | Send a message

145-- | Send a message

142send :: Serializable a => ProcessId -> a -> Process ()

146send :: Serializable a => ProcessId -> a -> Process ()

@@ -144,17 +148,18 @@ send them msg = do

144 proc <- ask

148 proc <- ask

145 let node = localNodeId (processNode proc)

149 let node = localNodeId (processNode proc)

146 destNode = (processNodeId them)

150 destNode = (processNodeId them)

147− us = (processId proc) in do

151+ us = (processId proc)

148− case destNode == node of

152+ msg' = wrapMessage msg in do

149− True -> unsafeSendLocal them msg

150− False -> liftIO $ do sendMessage (processNode proc)

151− (ProcessIdentifier (processId proc))

152− (ProcessIdentifier them)

153− NoImplicitReconnect

154− msg

155− -- see [note: tracing]

153+ -- see [note: tracing]

156− liftIO $ traceEvent (localEventBus (processNode proc))

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

157− (MxSent them us (wrapMessage msg))

155+ (MxSent them us msg')

156+ if destNode == node

157+ then sendCtrlMsg Nothing $ LocalSend them msg'

158+ else liftIO $ sendMessage (processNode proc)

159+ (ProcessIdentifier (processId proc))

160+ (ProcessIdentifier them)

161+ NoImplicitReconnect

162+ msg

158

163

159-- | Send a message unreliably.

164-- | Send a message unreliably.

160--

165--

@@ -170,19 +175,12 @@ usend them msg = do

170 proc <- ask

175 proc <- ask

171 let there = processNodeId them

176 let there = processNodeId them

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

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

178+ let msg' = wrapMessage msg

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

173 if localNodeId (processNode proc) == there

180 if localNodeId (processNode proc) == there

174− then unsafeSendLocal them msg

181+ then sendCtrlMsg Nothing $ LocalSend them msg'

175− else do sendCtrlMsg (Just there) $ UnreliableSend (processLocalId them)

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

176− (createMessage msg)

183+ (createMessage msg)

177− -- I would assert that is it *still* cheaper to not encode here...

178− liftIO $ traceEvent (localEventBus node)

179− (MxSent them us (wrapMessage msg))

180−

181−unsafeSendLocal :: (Serializable a) => ProcessId -> a -> Process ()

182−unsafeSendLocal pid msg =

183− let msg' = wrapMessage msg in do

184− sendCtrlMsg Nothing $ LocalSend pid msg'

185− -- see [note: tracing]

186

184

187-- | Send a message on a typed channel

185-- | Send a message on a typed channel

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

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

@@ -192,7 +190,7 @@ sendChan (SendPort cid) msg = do

192 destNode = processNodeId (sendPortProcessId cid) in do

190 destNode = processNodeId (sendPortProcessId cid) in do

193 case destNode == node of

191 case destNode == node of

194 True -> unsafeSendChanLocal cid msg

192 True -> unsafeSendChanLocal cid msg

195− False -> do

193+ False ->

196 liftIO $ sendBinary (processNode proc)

194 liftIO $ sendBinary (processNode proc)

197 (ProcessIdentifier (processId proc))

195 (ProcessIdentifier (processId proc))

198 (SendPortIdentifier cid)

196 (SendPortIdentifier cid)