0514f3c..master · distributed-process

git.iko.soy/distributed-process

10 files changed+421−231

unifiedsplit

distributed-process-tests/distributed-process-tests.cabal+1−1 distributed-process-tests/src/Control/Distributed/Process/Tests/CH.hs+110−32 distributed-process-tests/src/Control/Distributed/Process/Tests/Internal/Utils.hs+3−1 distributed-process-tests/src/Control/Distributed/Process/Tests/Mx.hs+172−113 distributed-process.cabal+1−1 src/Control/Distributed/Process/Internal/Primitives.hs+54−39 src/Control/Distributed/Process/Management.hs+34−2 src/Control/Distributed/Process/Management/Internal/Types.hs+5 src/Control/Distributed/Process/Node.hs+15−10 src/Control/Distributed/Process/UnsafePrimitives.hs+26−32

distributed-process-tests/distributed-process-tests.cabal+1−1

@@ -6,7 +6,7 @@ description: Tests and test suite for Cloud Haskell libraries.

66license: BSD3

77license-file: LICENSE

88Author: Duncan Coutts, Nicolas Wu, Edsko de Vries

9−Maintainer: Facundo Domínguez <facundo.dominguez@tweag.io>

9+Maintainer: Tim Watson <watson.timothy@gmail.com>

1010copyright: Well-Typed LLP

1111category: Control, Cloud Haskell

1212build-type: Simple

distributed-process-tests/src/Control/Distributed/Process/Tests/CH.hs+110−32

@@ -44,13 +44,13 @@ import Control.Distributed.Process.Internal.Types

4444import Control.Distributed.Process.Node

4545import Control.Distributed.Process.Debug

4646import Control.Distributed.Process.Management.Internal.Types

47−import Control.Distributed.Process.Tests.Internal.Utils (shouldBe)

47+import Control.Distributed.Process.Tests.Internal.Utils (shouldBe, pause)

4848import Control.Distributed.Process.Serializable (Serializable)

49−

49+import Data.Maybe (isNothing, isJust)

5050import Test.HUnit (Assertion, assertBool, assertFailure)

5151import Test.Framework (Test, testGroup)

5252import Test.Framework.Providers.HUnit (testCase)

53−import Control.Rematch hiding (match)

53+import Control.Rematch hiding (match, isNothing, isJust)

5454import Control.Rematch.Run (Match(..))

5555

5656newtype Ping = Ping ProcessId

@@ -87,10 +87,9 @@ verifyClient :: String -> MVar Bool -> IO ()

8787verifyClient s b = takeMVar b >>= assertBool s

8888

8989expectPing :: MVar Bool -> Process ()

90−expectPing mv = expect >>= liftIO . putMVar mv . checkPing

90+expectPing mv = expect >>= liftIO . putMVar mv . checkPing

9191 where

9292 checkPing (Ping _) = True

93− checkPing _ = False

9493

9594-- | Quick and dirty synchronous version of whereisRemoteAsync

9695whereisRemote :: NodeId -> String -> Process (Maybe ProcessId)

@@ -240,10 +239,10 @@ testPing TestTransport{..} = do

240239 pid <- getSelfPid

241240 replicateM_ numPings $ do

242241 send pingServer (Pong pid)

243− p <- expect

242+ p <- expectTimeout 3000000

244243 case p of

245− Ping _ -> return ()

246− _ -> die "Unexpected message"

244+ Just (Ping _) -> return ()

245+ Nothing -> die "Failed to receive Ping"

247246

248247 putMVar clientDone ()

249248

@@ -560,7 +559,7 @@ testMergeChannels TestTransport{..} = do

560559 ss <- liftIO $ mapM readMVar vs

561560 case ss of

562561 [sa, sb, sc] ->

563− mapM_ ((>> liftIO (threadDelay 10000)) . uncurry sendChan)

562+ mapM_ ((>> pause 10000) . uncurry sendChan)

564563 [ -- a, b, c

565564 (sa, 'a')

566565 , (sb, 'b')

@@ -736,7 +735,6 @@ testRemoteRegistry :: TestTransport -> Assertion

736735testRemoteRegistry TestTransport{..} = do

737736 node1 <- newLocalNode testTransport initRemoteTable

738737 node2 <- newLocalNode testTransport initRemoteTable

739− done <- newEmptyMVar

740738

741739 pingServer <- forkProcess node1 ping

742740 deadProcess <- forkProcess node1 (return ())

@@ -751,6 +749,7 @@ testRemoteRegistry TestTransport{..} = do

751749

752750 pid <- verifyWhereIsRemote nid1 "ping"

753751 liftIO $ assertBool "Expected pindServer to match pid" $ pingServer == pid

752+

754753 us <- getSelfPid

755754 nsendRemote nid1 "ping" (Pong us)

756755 receiveWait [

@@ -760,22 +759,22 @@ testRemoteRegistry TestTransport{..} = do

760759 -- test that if process was not registered Nothing is returned

761760 -- in owner field.

762761 registerRemoteAsync nid1 "dead" deadProcess

763− receiveWait [

764− matchIf (\(RegisterReply label' False Nothing) -> "dead" == label')

765− (\(RegisterReply _ _ _) -> return ()) ]

762+ receiveWait [ matchIf (\(RegisterReply label' f mPid) -> "dead" == label')

763+ (\(RegisterReply _ f mPid) -> return (not f && isNothing mPid))

764+ ] >>= liftIO . assertBool "Expected False Nothing in RegisterReply"

765+

766766 registerRemoteAsync nid1 "ping" deadProcess

767767 receiveWait [

768− matchIf (\(RegisterReply label' False (Just pid'')) ->

769− "ping" == label' && pid'' == pingServer)

770− (\(RegisterReply _ _ _) -> return ()) ]

768+ matchIf (\(RegisterReply label' False mPid) ->

769+ "ping" == label' && isJust mPid)

770+ (\(RegisterReply _ f (Just pid'')) -> return (not f && pid'' == pingServer))

771+ ] >>= liftIO . assertBool "Expected False and (Just alreadyRegisteredPid) in RegisterReply"

772+

771773 unregisterRemoteAsync nid1 "dead"

772774 receiveWait [

773− matchIf (\(RegisterReply label' False Nothing) ->

774− "dead" == label' && pid == pingServer)

775− (\(RegisterReply _ _ _) -> return ()) ]

776− liftIO $ putMVar done ()

777−

778− takeMVar done

775+ matchIf (\(RegisterReply label' _ _) -> "dead" == label')

776+ (\(RegisterReply _ f mPid) -> return (not f && isNothing mPid))

777+ ] >>= liftIO . assertBool "Expected False and Nothing in RegisterReply"

779778

780779testRemoteRegistryRemoteProcess :: TestTransport -> Assertion

781780testRemoteRegistryRemoteProcess TestTransport{..} = do

@@ -1273,6 +1272,76 @@ testReceiveChanFeatures TestTransport{..} = do

12731272

12741273 takeMVar done

12751274

1275+testChanLifecycle :: TestTransport -> Assertion

1276+testChanLifecycle TestTransport{..} = let delay = 3000000 in do

1277+ result <- newEmptyMVar

1278+ tchMV <- newEmptyMVar

1279+ localNode <- newLocalNode testTransport initRemoteTable

1280+ runProcess localNode $ do

1281+

1282+ pid <- spawnLocal $ do tCh <- newChan :: Process (SendPort (), ReceivePort ())

1283+ liftIO $ putMVar tchMV tCh

1284+ expect :: Process ()

1285+ let (sp, _) = tCh

1286+ sendChan sp ()

1287+ expect :: Process ()

1288+

1289+ mRefPid <- monitor pid

1290+

1291+ cPid <- spawnLocal $ do

1292+ (sp', rp) <- liftIO $ takeMVar tchMV

1293+ -- say "picked up our test channel"

1294+ send pid ()

1295+ -- say "told pid to continue"

1296+ res <- receiveChanTimeout delay rp

1297+ case res of

1298+ Nothing -> say "initial chan () missing!" >> (liftIO $ putMVar result False)

1299+ Just () -> do mr <- monitor pid

1300+ pause 10000

1301+ -- say "sending pid a second () will cause it to exit"

1302+ send pid ()

1303+

1304+ -- say "make sure we see a DOWN notification for pid having stopped"

1305+ receiveWait [ match (\(_ :: ProcessMonitorNotification) -> return ()) ]

1306+

1307+ -- now that pid has died, the send port should be useless...

1308+ liftIO $ putMVar tchMV (sp', rp)

1309+

1310+ -- let's verify that we do not see the message from our

1311+ -- parent process on the channel, once pid has died...

1312+ recv <- receiveChanTimeout delay rp

1313+ -- say $ "finished waiting for second (), writing result" ++ (show recv)

1314+ liftIO $ putMVar result $ isNothing recv

1315+

1316+ mRefCPid <- monitor cPid

1317+

1318+ receiveWait

1319+ [ matchIf (\(ProcessMonitorNotification r _ _) -> r == mRefPid)

1320+ (\_ -> return ())

1321+ ]

1322+

1323+ -- say "seen first pid die..."

1324+

1325+ (sendPort, _) <- liftIO $ takeMVar tchMV

1326+ sendChan sendPort ()

1327+ -- say "sent () after owning pid died"

1328+

1329+ -- let cPid know we've written to the channel...

1330+ send cPid ()

1331+

1332+ receiveWait

1333+ [ matchIf (\(ProcessMonitorNotification r _ _) -> r == mRefCPid)

1334+ (\_ -> return ())

1335+ ]

1336+

1337+ -- say "seen both pids die now..."

1338+

1339+ -- and wait on the result back in IO land...

1340+ testRes <- takeMVar result

1341+ -- runProcess localNode $ say "got result..."

1342+ assertBool "Expected sending on the channel to fail, but received data!" testRes

1343+

1344+

12761345testKillLocal :: TestTransport -> Assertion

12771346testKillLocal TestTransport{..} = do

12781347 localNode <- newLocalNode testTransport initRemoteTable

@@ -1487,20 +1556,24 @@ testRegistryMonitoring :: TestTransport -> Assertion

14871556testRegistryMonitoring TestTransport{..} = do

14881557 node1 <- newLocalNode testTransport initRemoteTable

14891558 node2 <- newLocalNode testTransport initRemoteTable

1490− waitH <- newEmptyMVar

14911559

14921560 let nid = localNodeId node2

14931561 pid <- forkProcess node1 $ do

1494− getSelfPid >>= runUntilRegistered nid

1495− liftIO $ takeMVar waitH

1562+ self <- getSelfPid

1563+ runUntilRegistered nid self

1564+ say $ (show self) ++ " registered as " ++ regName

1565+ expect :: Process ()

1566+ say $ (show self) ++ " exiting normally"

14961567

14971568 runProcess node2 $ do

14981569 register regName pid

1570+ say $ regName ++ " registered to " ++ show pid

14991571 res <- whereis regName

1572+ send pid ()

1573+ say $ " sent finish signal to " ++ show pid

15001574 us <- getSelfPid

1501− liftIO $ do

1502− putMVar waitH ()

1503− assertBool "expected (Just pid)" $ res == (Just pid)

1575+ liftIO $ assertBool "expected (Just pid)" $ res == (Just pid)

1576+

15041577

15051578 -- This delay isn't essential!

15061579 -- The test case passes perfectly fine without it (feel free to comment out

@@ -1525,7 +1598,9 @@ testRegistryMonitoring TestTransport{..} = do

15251598 regHere <- newEmptyMVar

15261599 runProcess node2 $ whereis regName >>= liftIO . putMVar regHere

15271600 res <- takeMVar regHere

1528− assertBool "expected Nothing, but process still registered" (res == Nothing)

1601+ case res of

1602+ Nothing -> return ()

1603+ Just pid -> assertBool ("expected Nothing, but got " ++ show pid) False

15291604

15301605 where

15311606 runUntilRegistered nid us = do

@@ -1537,11 +1612,13 @@ testRegistryMonitoring TestTransport{..} = do

15371612

15381613 delayUntilMaybeUnregistered nid p = do

15391614 whereisRemoteAsync nid regName

1540− receiveTimeout 20000000 {- 20 sec delay -} [

1541− matchIf (\(WhereIsReply n p) -> n == regName && p == Nothing)

1615+ res <- receiveTimeout 20000000 {- 20 sec delay -} [

1616+ matchIf (\(WhereIsReply n p) -> n == regName && isNothing p)

15421617 (const $ return ())

15431618 ]

1544− return ()

1619+ case res of

1620+ Just () -> return ()

1621+ Nothing -> delayUntilMaybeUnregistered nid p

15451622

15461623 regName = "testRegisterRemote"

15471624

@@ -1737,6 +1814,7 @@ tests testtrans = return [

17371814 , testCase "MatchMessageUnwrap" (testMatchMessageWithUnwrap testtrans)

17381815 , testCase "ReceiveChanTimeout" (testReceiveChanTimeout testtrans)

17391816 , testCase "ReceiveChanFeatures" (testReceiveChanFeatures testtrans)

1817+ , testCase "ChanLifecycle" (testChanLifecycle testtrans)

17401818 , testCase "KillLocal" (testKillLocal testtrans)

17411819 , testCase "KillRemote" (testKillRemote testtrans)

17421820 , testCase "Die" (testDie testtrans)

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

@@ -91,6 +91,7 @@ import Test.HUnit (Assertion, assertFailure)

9191import Test.HUnit.Base (assertBool)

9292

9393import GHC.Generics

94+import System.Timeout (timeout)

9495

9596-- | A mutable cell containing a test result.

9697type TestResult a = MVar a

@@ -233,7 +234,8 @@ testProcessReport pid = do

233234tryRunProcess :: LocalNode -> Process () -> IO ()

234235tryRunProcess node p = do

235236 tid <- liftIO myThreadId

236− runProcess node $ catch p (\e -> liftIO $ throwTo tid (e::SomeException))

237+ void $ timeout (1000000 * 60 * 5 :: Int) $

238+ runProcess node $ catch p (\e -> liftIO $ throwTo tid (e::SomeException))

237239

238240tryForkProcess :: LocalNode -> Process () -> IO ProcessId

239241tryForkProcess node p = do

distributed-process-tests/src/Control/Distributed/Process/Tests/Mx.hs+172−113

@@ -1,17 +1,22 @@

1−{-# LANGUAGE DeriveGeneric #-}

1+{-# LANGUAGE DeriveGeneric #-}

2+{-# LANGUAGE ParallelListComp #-}

24module Control.Distributed.Process.Tests.Mx (tests) where

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

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

6−import Control.Exception (SomeException)

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

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

9+import Control.Distributed.Process.Internal.Types

10+ ( ProcessExitException(..)

11+ , unsafeCreateUnencodedMessage

12+ )

913import Control.Distributed.Process.Node

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

1115 ( send

1216 , nsend

1317 , nsendRemote

1418 , usend

19+ , sendChan

1520 )

1621import Control.Distributed.Process.Management

1722 ( MxEvent(..)

@@ -32,8 +37,8 @@ import Control.Monad (void, unless)

3237import Control.Monad.Catch(finally, bracket, try)

3338import Control.Rematch (equalTo)

3439import Data.Binary

35−import Data.List (find, sort)

36−import Data.Maybe (isJust, isNothing)

40+import Data.List (find, sort, intercalate)

41+import Data.Maybe (isJust, fromJust, isNothing, fromMaybe, catMaybes)

3742import Data.Typeable

3843import GHC.Generics hiding (from)

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

@@ -61,7 +66,7 @@ awaitExit pid =

6166 where

6267 withMonitorRef p = bracket (monitor p) unmonitor

6368

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

69+testAgentBroadcast :: TestResult (Maybe ()) -> Process ()

6570testAgentBroadcast result = do

6671 (resultSP, resultRP) <- newChan :: Process (SendPort (), ReceivePort ())

6772

@@ -76,10 +81,10 @@ testAgentBroadcast result = do

7681 mxNotify ()

7782 -- Once the publisher has seen our message, it will broadcast the Publish

7883 -- and the consumer will see that and send the result to our typed channel.

79− stash result =<< receiveChan resultRP

84+ stash result =<< receiveChanTimeout 10000000 resultRP

8085

81− kill publisher "finished" >> awaitExit publisher

82− kill consumer "finished" >> awaitExit consumer

86+ kill publisher "finished"

87+ kill consumer "finished"

8388

8489testAgentDualInput :: TestResult (Maybe Int) -> Process ()

8590testAgentDualInput result = do

@@ -94,6 +99,8 @@ testAgentDualInput result = do

9499 else mxReady)

95100 ]

96101

102+ mRef <- monitor s

103+

97104 mxNotify (1 :: Int)

98105 nsend "sum-agent" (3 :: Int)

99106 mxNotify (2 :: Int)

@@ -101,7 +108,10 @@ testAgentDualInput result = do

101108 mxNotify (5 :: Int)

102109

103110 stash result =<< receiveChanTimeout 10000000 rp

104− awaitExit s

111+ died <- receiveTimeout 10000000 [

112+ matchIf (\(ProcessMonitorNotification r _ _) -> r == mRef) (const $ return True)

113+ ]

114+ died `shouldBe` equalTo (Just True)

105115

106116testAgentPrioritisation :: TestResult [String] -> Process ()

107117testAgentPrioritisation result = do

@@ -115,10 +125,10 @@ testAgentPrioritisation result = do

115125 (sp, rp) <- newChan

116126 s <- mxAgent (MxAgentId name) ["first"] [

117127 mxSink (\(s :: String) -> do

118− mxUpdateLocal ((s:))

128+ mxUpdateLocal (s:)

119129 st <- mxGetLocal

120130 case length st of

121− n | n == 5 -> do liftMX $ sendChan sp st

131+ n | n == 5 -> do liftMX $ sendChan sp (sort st)

122132 mxDeactivate "finished"

123133 _ -> mxReceive -- go to the mailbox

124134 )

@@ -129,7 +139,7 @@ testAgentPrioritisation result = do

129139 mxNotify "fourth"

130140 nsend name "fifth"

131141

132− stash result . sort =<< receiveChan rp

142+ stash result . fromMaybe [] =<< receiveChanTimeout 10000000 rp

133143 awaitExit s

134144

135145testAgentMailboxHandling :: TestResult (Maybe ()) -> Process ()

@@ -142,12 +152,19 @@ testAgentMailboxHandling result = do

142152 nsend "mailbox-agent" ()

143153

144154 stash result =<< receiveChanTimeout 1000000 rp

145− kill agent "finished" >> awaitExit agent

155+ kill agent "finished"

146156

147157testAgentEventHandling :: TestResult Bool -> Process ()

148158testAgentEventHandling result = do

149− let initState = [] :: [MxEvent]

159+ us <- getSelfPid

160+ -- because this test is a bit racy, let's ensure it can't run indefinitely

161+ timer <- spawnLocal $ do

162+ pause (1000000 * 60 * 5)

163+ -- okay we've waited 5 mins, let's kill the test off if it's stuck...

164+ stash result False

165+ kill us "Test Timed Out"

150166

167+ let initState = [] :: [MxEvent]

151168 (rc, rs) <- newChan

152169 agentPid <- mxAgent (MxAgentId "lifecycle-listener-agent") initState [

153170 (mxSink $ \ev -> do

@@ -180,8 +197,7 @@ testAgentEventHandling result = do

180197 -- TODO: yes, this is racy, but we're at the mercy of the scheduler here...

181198 faff 2000000 rs

182199

183− _ <- monitor agentPid

184− (sp, rp) <- newChan

200+ (sp, rp) <- newChan :: Process (SendPort (), ReceivePort ())

185201

186202 pid <- spawnLocal $ expect >>= sendChan sp

187203

@@ -190,9 +206,11 @@ testAgentEventHandling result = do

190206 monitor pid

191207

192208 send pid ()

193− () <- receiveChan rp

209+ rct <- receiveChanTimeout 10000000 rp

210+ unless (isJust rct) $ die "No response on channel"

194211

195− receiveWait [ match (\(ProcessMonitorNotification _ _ _) -> return ()) ]

212+ pmn <- receiveTimeout 2000000 [ match (\ProcessMonitorNotification{} -> return ()) ]

213+ unless (isJust pmn) $ die "No monitor notification arrived"

196214

197215 -- TODO: yes, this is racy, but we're at the mercy of the scheduler here...

198216 faff 2000000 rs

@@ -201,17 +219,16 @@ testAgentEventHandling result = do

201219 mxNotify (MxSpawned pid, replyTo)

202220 mxNotify (MxProcessDied pid DiedNormal, replyTo)

203221

204− seenAlive <- receiveChan reply

205− seenDead <- receiveChan reply

222+ seenAlive <- receiveChanTimeout 10000000 reply

223+ seenDead <- receiveChanTimeout 10000000 reply

206224

207− stash result $ seenAlive && seenDead

208− kill agentPid "test-complete" >> awaitExit agentPid

225+ stash result $ foldr (&&) True $ catMaybes [seenAlive, seenDead]

226+ kill timer "test-complete"

227+ kill agentPid "test-complete"

209228 where

210229 faff delay port = do

211230 res <- receiveChanTimeout delay port

212− unless (isNothing res) $ do

213− pause delay

214− faff delay port

231+ unless (isNothing res) $ pause delay

215232

216233testMxRegEvents :: TestResult () -> Process ()

217234testMxRegEvents result = do

@@ -258,7 +275,6 @@ testMxRegEvents result = do

258275 reg3 `shouldBe` equalTo (Just (label, p1))

259276

260277 mapM_ (flip kill $ "test-complete") [agent, p1, p2]

261− awaitExit agent

262278

263279testMxRegMon :: LocalNode -> TestResult () -> Process ()

264280testMxRegMon remoteNode result = do

@@ -316,54 +332,99 @@ testMxRegMon remoteNode result = do

316332 evts `shouldContain` (Just (label1, p1))

317333 evts `shouldContain` (Just (label2, p1))

318334

319− kill agent "test-complete" >> awaitExit agent

335+ kill agent "test-complete"

320336

321337ensure :: Process () -> Process () -> Process ()

322338ensure = flip finally

323339

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

325−

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

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

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

341+ -> Maybe LocalNode

342+ -> Process ()

343+testNSend op n = do

344+ us <- getSelfPid

328345 let delay = 5000000

329− let label = "testMxSend"

346+ let label = "testMxSend" ++ (show us)

347+ let isValid = isValidLabel n label

348+

349+ testMxSend n label $ \p1 sink -> do

350+ register label p1

351+ reg1 <- receiveChanTimeout delay sink

352+ case reg1 of

353+ Just (MxRegistered pd lb)

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

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

330356

331− register label p1

332− reg1 <- receiveChanTimeout delay sink

333− case reg1 of

334− Just (MxRegistered pd lb)

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

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

357+ op label ()

337358

338− op label ()

359+ sent <- receiveChanTimeout delay sink

360+ case sent of

361+ Just (MxSentToName lb by _)

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

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

339364

365+ where

366+ isValidLabel nd l1 l2

367+ | l2 == l2 = True

368+ | isJust nd = l2 == l1 ++ "@" ++ show (localNodeId $ fromJust nd)

369+ | otherwise = False

370+

371+testSend :: (ProcessId -> () -> Process ())

372+ -> Maybe LocalNode

373+ -> Process ()

374+testSend op n = do

340375 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

376+ let label = "testMxSend" ++ (show us)

377+ testMxSend n label $ \p1 sink -> do

378+ -- initiate a send

379+ op p1 ()

380+

381+ -- verify the management event

382+ sent <- receiveChanTimeout 5000000 sink

383+ case sent of

384+ Just (MxSent pidTo pidFrom _)

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

386+ _ -> return False

387+

388+testChan :: (SendPort () -> () -> Process ())

389+ -> Maybe LocalNode

390+ -> Process ()

391+testChan op n = testMxSend n "" $ \p1 sink -> do

392+

393+ us <- getSelfPid

394+ send p1 us

395+

396+ cleared <- receiveChanTimeout 2000000 sink

397+ case cleared of

398+ Just (MxSent pidTo pidFrom _)

399+ | pidTo == p1 && pidFrom == us -> return ()

400+ _ -> die "Received uncleared Mx Event"

401+

402+ chan <- receiveTimeout 5000000 [ match return ]

403+ let ch' = fromMaybe (error "No reply chan received") chan

346404

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

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

349405 -- initiate a send

350− op p1 ()

406+ op ch' ()

351407

352408 -- verify the management event

353− us <- getSelfPid

354409 sent <- receiveChanTimeout 5000000 sink

355410 case sent of

356− Just (MxSent pidTo pidFrom _)

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

358− _ -> return False

411+ Just (MxSentToPort sId spId _)

412+ | sId == us && spId == sendPortId ch' -> return True

413+ _ -> return False

414+

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

359416

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

361−testMxSend mNode result test = do

417+testMxSend :: Maybe LocalNode -> String -> SendTest -> Process ()

418+testMxSend mNode label test = do

362419 us <- getSelfPid

420+ (sp, rp) <- newChan

363421 (chan, sink) <- newChan

364422 agent <- mxAgent (MxAgentId $ agentLabel us) () [

365423 mxSink $ \ev -> do

366424 case ev of

425+ m@(MxSentToPort _ cid _)

426+ | cid /= sendPortId chan

427+ && cid /= sendPortId sp -> liftMX $ sendChan chan m

367428 m@(MxSent _ fromPid _)

368429 | fromPid == us -> liftMX $ sendChan chan m

369430 m@(MxSentToName _ fromPid _)

@@ -374,29 +435,39 @@ testMxSend mNode result test = do

374435 mxReady

375436 ]

376437

377− (sp, rp) <- newChan

378438 case mNode of

379439 Nothing -> void $ spawnLocal (proc sp)

380440 Just remoteNode -> void $ liftIO $ forkProcess remoteNode $ proc sp

381441

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

442+ p1 <- receiveChanTimeout 2000000 rp

443+ unless (isJust p1) $ die "Timed out waiting for ProcessId"

444+ res <- try $ test (fromJust p1) sink

445+

387446 kill agent "bye"

388− kill p1 "bye"

447+ kill (fromJust p1) "bye"

448+

449+ case res of

450+ Left (ProcessExitException _ m) -> (liftIO $ putStrLn $ "SomeException-" ++ show m) >> die m

451+ Right tr -> tr `shouldBe` equalTo True

452+

389453

390454 where

391− label = "testMxSend"

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

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

456+ proc sp' = getSelfPid >>= sendChan sp' >> go Nothing

457+

458+ go :: Maybe (ReceivePort ()) -> Process ()

459+ go Nothing = receiveTimeout 5000000 [ match replyChannel ] >>= go

460+ go c@(Just c') = receiveWait [ matchChan c' return ] >> go c

461+

462+ replyChannel p' = do

463+ (s, r) <- newChan

464+ send p' s

465+ return r

394466

395467tests :: TestTransport -> IO [Test]

396468tests TestTransport{..} = do

397469 node1 <- newLocalNode testTransport initRemoteTable

398470 node2 <- newLocalNode testTransport initRemoteTable

399− let nid = localNodeId node2

400471 return [

401472 testGroup "MxAgents" [

402473 testCase "EventHandling"

@@ -406,7 +477,7 @@ tests TestTransport{..} = do

406477 , testCase "InterAgentBroadcast"

407478 (delayedAssertion

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

409− node1 () testAgentBroadcast)

480+ node1 (Just ()) testAgentBroadcast)

410481 , testCase "AgentMailboxHandling"

411482 (delayedAssertion

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

@@ -431,53 +502,41 @@ tests TestTransport{..} = do

431502 (delayedAssertion

432503 "expected process deaths to result in unregistration events"

433504 node1 () (testMxRegMon node2))

434− , testGroup "SentEvents" [

435− testGroup "RemoteTargets" [

436− testCase "Unsafe.nsend"

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

505+ , testGroup "SendEvents" $ buildTestCases node1 node2

506+ ]

507+ ]

508+ where

509+ buildTestCases n1 n2 = let nid = localNodeId n2 in build n1 n2 [

510+ ("NSend", [

511+ ("nsend", testNSend nsend)

512+ , ("Unsafe.nsend", testNSend Unsafe.nsend)

513+ , ("nsendRemote", testNSend (nsendRemote nid))

514+ , ("Unsafe.nsendRemote", testNSend (Unsafe.nsendRemote nid))

515+ ])

516+ , ("Send", [

517+ ("send", testSend send)

518+ , ("Unsafe.send", testSend Unsafe.send)

519+ , ("usend", testSend usend)

520+ , ("Unsafe.usend", testSend Unsafe.usend)

521+ , ("sendChan", testChan sendChan)

522+ , ("Unsafe.sendChan", testChan Unsafe.sendChan)

523+ ])

524+ , ("Forward", [

525+ ("forward", testSend (\p m -> forward (unsafeCreateUnencodedMessage m) p))

526+ , ("uforward", testSend (\p m -> uforward (unsafeCreateUnencodedMessage m) p))

527+ ])

480528 ]

481− ]

482− ]

529+

530+ build :: LocalNode

531+ -> LocalNode

532+ -> [(String, [(String, (Maybe LocalNode -> Process ()))])]

533+ -> [Test]

534+ build n ln specs =

535+ [ testGroup (intercalate "-" [groupName, caseSuffix]) [

536+ testCase (intercalate "-" [caseName, caseSuffix])

537+ (runProcess n (caseImpl caseNode))

538+ | (caseName, caseImpl) <- groupCases

539+ ]

540+ | (groupName, groupCases) <- specs

541+ , (caseSuffix, caseNode) <- [("RemotePid", Just ln), ("LocalPid", Nothing)]

483542 ]

distributed-process.cabal+1−1

@@ -6,7 +6,7 @@ License: BSD3

66License-File: LICENSE

77Copyright: Well-Typed LLP, Tweag I/O Limited

88Author: Duncan Coutts, Nicolas Wu, Edsko de Vries

9−Maintainer: Facundo Domínguez <facundo.dominguez@tweag.io>

9+Maintainer: Tim Watson <watson.timothy@gmail.com>

1010Stability: experimental

1111Homepage: http://haskell-distributed.github.com/

1212Bug-Reports: https://github.com/haskell-distributed/distributed-process/issues

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

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

@@ -306,7 +307,13 @@ expect = receiveWait [match return]

306307-- Channels --

307308--------------------------------------------------------------------------------

308309

309−-- | Create a new typed channel

310+-- | Create a new typed channel, bound to the calling Process

311+--

312+-- Note that the channel is bound to the lifecycle of the process that evaluates

313+-- this function, such that when it dies/exits, the channel will no longer

314+-- function, but will remain accessible. Thus reading from the ReceivePort will

315+-- fail silently thereafter, blocking indefinitely (unless a timeout is used).

316+--

310317newChan :: Serializable a => Process (SendPort a, ReceivePort a)

311318newChan = do

312319 proc <- ask

@@ -334,13 +341,16 @@ newChan = do

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

335342sendChan (SendPort cid) msg = do

336343 proc <- ask

337− let node = localNodeId (processNode proc)

338− destNode = processNodeId (sendPortProcessId cid) in do

339− case destNode == node of

344+ let node = processNode proc

345+ pid = processId proc

346+ us = localNodeId node

347+ them = processNodeId (sendPortProcessId cid) in do

348+ liftIO $ traceEvent (localEventBus node) (MxSentToPort pid cid $ wrapMessage msg)

349+ case them == us of

340350 True -> sendChanLocal cid msg

341351 False -> do

342− liftIO $ sendBinary (processNode proc)

343− (ProcessIdentifier (processId proc))

352+ liftIO $ sendBinary node

353+ (ProcessIdentifier pid)

344354 (SendPortIdentifier cid)

345355 NoImplicitReconnect

346356 msg

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

484494 us = processId proc

485495 nid = localNodeId node

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

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

498+ if destNode == nid

499+ then sendCtrlMsg Nothing (LocalSend them msg)

500+ else liftIO $ sendPayload (processNode proc)

501+ (ProcessIdentifier (processId proc))

502+ (ProcessIdentifier them)

503+ NoImplicitReconnect

504+ (messageToPayload msg)

496505

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

498507--

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

506515 us = processId proc

507516 nid = localNodeId node

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

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

519+ if destNode == nid

520+ then sendCtrlMsg Nothing (LocalSend them msg)

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

517522

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

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

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

12221227

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

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

1225−nsend label msg =

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

1230+nsend label msg = do

1231+ proc <- ask

1232+ let msg' = createUnencodedMessage msg

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

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

1235+ sendCtrlMsg Nothing (NamedSend label msg')

12271236

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

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

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

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

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

12371246nsendRemote nid label msg = do

1238− here <- getSelfNode

1239− if here == nid then nsend label msg

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

1247+ proc <- ask

1248+ let us = processId proc

1249+ let node = processNode proc

1250+ if localNodeId node == nid

1251+ then nsend label msg

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

1253+ liftIO $ traceEvent (localEventBus node)

1254+ (MxSentToName lbl us (wrapMessage msg))

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

12411256

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

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

src/Control/Distributed/Process/Management.hs+34−2

@@ -76,6 +76,40 @@

7676--

7777-- * Whether messages will be taken from the mailbox first, or the event bus.

7878--

79+-- Since the event bus uses STM broadcast channels to communicate with agents,

80+-- no message written to the bus successfully can be lost.

81+--

82+-- Agents can also receive messages via their mailboxes - these are subject to

83+-- the same guarantees as all inter-process message sending.

84+--

85+-- Messages dispatched on an STM broadcast channel (i.e., management event bus)

86+-- are guaranteed to be delivered with the same FIFO ordering guarantees that

87+-- exist between two communicating processes, such that communication from the

88+-- node controller's threads (i.e., MxEvent's) will never be re-ordered, but

89+-- messages dispatched to the event bus by other processes (including, but not

90+-- limited to agents) are only guaranteed to be ordered between one sender and

91+-- one receiver.

92+--

93+-- No guarantee exists for the ordering in which messages sent to an agent's

94+-- mailbox will be delivered, vs messages dispatched via the event bus.

95+--

96+-- Because of the above, there are no ordering guarantees for messages sent

97+-- between agents, or for processes to agents, except for those that apply to

98+-- messages sent between regular processes, since agents are

99+-- implemented as such.

100+--

101+-- The event bus is serial and single threaded. Anything that is published by

102+-- the node controller will be seen in FIFO order. There are no ordering

103+-- guarantees pertaining to entries published to the event bus by other

104+-- processes or agents.

105+--

106+-- It should not be possible to see, for example, an @MxReceived@ before the

107+-- corresponding @MxSent@ event, since the places where we issue the @MxSent@

108+-- write directly to the event bus (using STM) in the calling (green) thread,

109+-- before dispatching instructions to the node controller to perform the

110+-- necessary routing to deliver the message to a process (or registered name,

111+-- or typed channel) locally or remotely.

112+--

79113-- [Management Data API]

80114--

81115-- Both management agents and clients of the API have access to a variety of

@@ -144,8 +178,6 @@

144178-- > monitorNames = getSelfPid >>= nsend "name-monitor"

145179-- > monitorNames2 = getSelfPid >>= mxNotify

146180--

147−-- For some real-world examples, see the distributed-process-platform package.

148−--

149181-- [Performance, Stablity and Scalability]

150182--

151183-- /Management Agents/ offer numerous advantages over regular processes:

src/Control/Distributed/Process/Management/Internal/Types.hs+5

@@ -21,6 +21,7 @@ import Control.Concurrent.STM

2121import Control.Distributed.Process.Internal.Types

2222 ( Process

2323 , ProcessId

24+ , SendPortId

2425 , Message

2526 , DiedReason

2627 , NodeId

@@ -59,8 +60,12 @@ data MxEvent =

5960 -- ^ fired whenever a message is sent from a local process

6061 | MxSentToName String ProcessId Message

6162 -- ^ fired whenever a named send occurs

63+ | MxSentToPort ProcessId SendPortId Message

64+ -- ^ fired whenever a sendChan occurs

6265 | MxReceived ProcessId Message

6366 -- ^ fired whenever a message is received by a local process

67+ | MxReceivedPort SendPortId Message

68+ -- ^ fired whenever a message is received via a typed channel

6469 | MxConnected ConnectionId EndPointAddress

6570 -- ^ fired when a network-transport connection is first established

6671 | MxDisconnected ConnectionId EndPointAddress

src/Control/Distributed/Process/Node.hs+15−10

@@ -492,7 +492,7 @@ type IncomingConnection = (NT.EndPointAddress, IncomingTarget)

492492data IncomingTarget =

493493 Uninit

494494 | ToProc ProcessId (Weak (CQueue Message))

495− | ToChan TypedChannel

495+ | ToChan SendPortId TypedChannel

496496 | ToNode

497497

498498data ConnectionState = ConnectionState {

@@ -547,13 +547,17 @@ handleIncomingMessages node = go initConnectionState

547547 enqueue queue msg -- 'enqueue' is strict

548548 trace node (MxReceived pid msg)

549549 go st

550− Just (_, ToChan (TypedChannel chan')) -> do

550+ Just (_, ToChan chId (TypedChannel chan')) -> do

551551 mChan <- deRefWeak chan'

552552 -- If mChan is Nothing, the process has given up the read end of

553553 -- the channel and we simply ignore the incoming message

554− forM_ mChan $ \chan -> atomically $

555− -- We make sure the message is fully decoded when it is enqueued

556− writeTQueue chan $! decode (BSL.fromChunks payload)

554+ forM_ mChan $ \chan -> do

555+ msg' <- atomically $ do

556+ msg <- return $! decode (BSL.fromChunks payload)

557+ -- We make sure the message is fully decoded when it is enqueued

558+ writeTQueue chan msg

559+ return msg

560+ trace node $ MxReceivedPort chId $ unsafeCreateUnencodedMessage msg'

557561 go st

558562 Just (_, ToNode) -> do

559563 let ctrlMsg = decode . BSL.fromChunks $ payload

@@ -580,7 +584,7 @@ handleIncomingMessages node = go initConnectionState

580584 mChannel <- withMVar (processState proc) $ return . (^. typedChannelWithId lcid)

581585 case mChannel of

582586 Just channel ->

583− go (incomingAt cid ^= Just (src, ToChan channel) $ st)

587+ go (incomingAt cid ^= Just (src, ToChan chId channel) $ st)

584588 Nothing ->

585589 invalidRequest cid st $

586590 "incoming attempt to connect to unknown channel of"

@@ -1050,11 +1054,12 @@ ncEffectLocalPortSend from msg = do

10501054 -- If ch is Nothing, the process has given up the read end of

10511055 -- the channel and we simply ignore the incoming message - this

10521056 ch <- deRefWeak chan'

1053− forM_ ch $ \chan -> deliverChan msg chan

1054− where deliverChan :: forall a . Message -> TQueue a -> IO ()

1055− deliverChan (UnencodedMessage _ raw) chan' =

1057+ forM_ ch $ \chan -> deliverChan node from msg chan

1058+ where deliverChan :: forall a . LocalNode -> SendPortId -> Message -> TQueue a -> IO ()

1059+ deliverChan n p (UnencodedMessage _ raw) chan' = do

10561060 atomically $ writeTQueue chan' ((unsafeCoerce raw) :: a)

1057− deliverChan (EncodedMessage _ _) _ =

1061+ trace n (MxReceivedPort p $ unsafeCreateUnencodedMessage raw)

1062+ deliverChan _ _ (EncodedMessage _ _) _ =

10581063 -- this will not happen unless someone screws with Primitives.hs

10591064 error "invalid local channel delivery"

10601065

src/Control/Distributed/Process/UnsafePrimitives.hs+26−32

@@ -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,30 +159,41 @@ 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

188180 proc <- ask

189− let node = localNodeId (processNode proc)

190− destNode = processNodeId (sendPortProcessId cid) in do

191− case destNode == node of

192− True -> unsafeSendChanLocal cid msg

193− False ->

194− liftIO $ sendBinary (processNode proc)

195− (ProcessIdentifier (processId proc))

196− (SendPortIdentifier cid)

197− NoImplicitReconnect

198− msg

181+ let node = processNode proc

182+ pid = processId proc

183+ us = localNodeId node

184+ them = processNodeId (sendPortProcessId cid)

185+ msg' = wrapMessage msg in do

186+ liftIO $ traceEvent (localEventBus node) (MxSentToPort pid cid msg')

187+ if them == us

188+ then unsafeSendChanLocal cid msg' -- NB: we wrap to P.Message !!!

189+ else liftIO $ sendBinary node

190+ (ProcessIdentifier pid)

191+ (SendPortIdentifier cid)

192+ NoImplicitReconnect

193+ msg

199194 where

200− unsafeSendChanLocal :: (Serializable a) => SendPortId -> a -> Process ()

201− unsafeSendChanLocal spId msg' =

202− sendCtrlMsg Nothing $ LocalPortSend spId (wrapMessage msg')

195+ unsafeSendChanLocal :: SendPortId -> Message -> Process ()

196+ unsafeSendChanLocal p m = sendCtrlMsg Nothing $ LocalPortSend p m

203197

204198-- | Create an unencoded @Message@ for any @Serializable@ type.

205199wrapMessage :: Serializable a => a -> Message