d2b5d82..master · distributed-process

git.iko.soy/distributed-process

10 files changed+476−223

unifiedsplit

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+5−2

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

7474import Control.Concurrent.MVar

7575 ( putMVar

7676 )

77−import Control.Distributed.Process

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

7878import Control.Distributed.Process.Node

7979import Control.Distributed.Process.Serializable()

8080

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

8282import Control.Monad (forever, void)

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

8384import Control.Monad.STM (atomically)

8485import Control.Rematch hiding (match)

8586import Control.Rematch.Run

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

9091import Test.HUnit.Base (assertBool)

9192

9293import GHC.Generics

94+import System.Timeout (timeout)

9395

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

9597type TestResult a = MVar a

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

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

233235tryRunProcess node p = do

234236 tid <- liftIO myThreadId

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

236239

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

238241tryForkProcess node p = do

distributed-process-tests/src/Control/Distributed/Process/Tests/Mx.hs+204−81

@@ -1,16 +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−

7−import Control.Distributed.Process

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

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

10+ ( ProcessExitException(..)

11+ , unsafeCreateUnencodedMessage

12+ )

813import Control.Distributed.Process.Node

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

1015 ( send

1116 , nsend

1217 , nsendRemote

1318 , usend

19+ , sendChan

1420 )

1521import Control.Distributed.Process.Management

1622 ( MxEvent(..)

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

2834 , mxBroadcast

2935 )

3036import Control.Monad (void, unless)

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

3138import Control.Rematch (equalTo)

3239import Data.Binary

33−import Data.List (find, sort)

34−import Data.Maybe (isJust, isNothing)

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

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

3542import Data.Typeable

36−import GHC.Generics

43+import GHC.Generics hiding (from)

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

3845import Prelude hiding (catch, log)

3946#endif

@@ -57,9 +64,9 @@ awaitExit pid =

5764 (\_ -> return ())

5865 ]

5966 where

60− withMonitorRef pid = bracket (monitor pid) unmonitor

67+ withMonitorRef p = bracket (monitor p) unmonitor

6168

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

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

6370testAgentBroadcast result = do

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

6572

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

7481 mxNotify ()

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

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

77− stash result =<< receiveChan resultRP

84+ stash result =<< receiveChanTimeout 10000000 resultRP

7885

79− kill publisher "finished" >> awaitExit publisher

80− kill consumer "finished" >> awaitExit consumer

86+ kill publisher "finished"

87+ kill consumer "finished"

8188

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

8390testAgentDualInput result = do

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

9299 else mxReady)

93100 ]

94101

102+ mRef <- monitor s

103+

95104 mxNotify (1 :: Int)

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

97106 mxNotify (2 :: Int)

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

99108 mxNotify (5 :: Int)

100109

101110 stash result =<< receiveChanTimeout 10000000 rp

102− awaitExit s

111+ died <- receiveTimeout 10000000 [

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

113+ ]

114+ died `shouldBe` equalTo (Just True)

103115

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

105117testAgentPrioritisation result = do

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

113125 (sp, rp) <- newChan

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

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

116− mxUpdateLocal ((s:))

128+ mxUpdateLocal (s:)

117129 st <- mxGetLocal

118130 case length st of

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

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

120132 mxDeactivate "finished"

121133 _ -> mxReceive -- go to the mailbox

122134 )

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

127139 mxNotify "fourth"

128140 nsend name "fifth"

129141

130− stash result . sort =<< receiveChan rp

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

131143 awaitExit s

132144

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

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

140152 nsend "mailbox-agent" ()

141153

142154 stash result =<< receiveChanTimeout 1000000 rp

143− kill agent "finished" >> awaitExit agent

155+ kill agent "finished"

144156

145157testAgentEventHandling :: TestResult Bool -> Process ()

146158testAgentEventHandling result = do

147− 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"

148166

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

149168 (rc, rs) <- newChan

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

151170 (mxSink $ \ev -> do

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

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

179198 faff 2000000 rs

180199

181− _ <- monitor agentPid

182− (sp, rp) <- newChan

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

183201

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

185203

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

188206 monitor pid

189207

190208 send pid ()

191− () <- receiveChan rp

209+ rct <- receiveChanTimeout 10000000 rp

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

192211

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

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

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

194214

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

196216 faff 2000000 rs

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

199219 mxNotify (MxSpawned pid, replyTo)

200220 mxNotify (MxProcessDied pid DiedNormal, replyTo)

201221

202− seenAlive <- receiveChan reply

203− seenDead <- receiveChan reply

222+ seenAlive <- receiveChanTimeout 10000000 reply

223+ seenDead <- receiveChanTimeout 10000000 reply

204224

205− stash result $ seenAlive && seenDead

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

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

226+ kill timer "test-complete"

227+ kill agentPid "test-complete"

207228 where

208229 faff delay port = do

209230 res <- receiveChanTimeout delay port

210− unless (isNothing res) $ do

211− pause delay

212− faff delay port

231+ unless (isNothing res) $ pause delay

213232

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

215234testMxRegEvents result = do

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

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

257276

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

259− awaitExit agent

260278

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

262280testMxRegMon remoteNode result = do

@@ -314,106 +332,211 @@ testMxRegMon remoteNode result = do

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

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

316334

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

335+ kill agent "test-complete"

318336

319337ensure :: Process () -> Process () -> Process ()

320338ensure = flip finally

321339

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

341+ -> Maybe LocalNode

342+ -> Process ()

343+testNSend op n = do

344+ us <- getSelfPid

345+ let delay = 5000000

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

322356

323−testMxUnsafeSendEvents :: LocalNode -> Process ()

324−testMxUnsafeSendEvents remoteNode = do

357+ op label ()

325358

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

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

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

328364

329− let label1 = "aaaaa"

330− let label2 = "bbbbb"

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

332− let agentLabel = "listener-agent"

333− let delay = 1000000

334− (regChan, regSink) <- newChan

335− (unRegChan, unRegSink) <- newChan

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

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

375+ us <- getSelfPid

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

404+

405+ -- initiate a send

406+ op ch' ()

407+

408+ -- verify the management event

409+ sent <- receiveChanTimeout 5000000 sink

410+ case sent of

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

416+

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

418+testMxSend mNode label test = do

419+ us <- getSelfPid

420+ (sp, rp) <- newChan

421+ (chan, sink) <- newChan

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

337423 mxSink $ \ev -> do

338424 case ev of

339− MxRegistered pid label

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

341− MxUnRegistered pid label

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

425+ m@(MxSentToPort _ cid _)

426+ | cid /= sendPortId chan

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

428+ m@(MxSent _ fromPid _)

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

430+ m@(MxSentToName _ fromPid _)

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

432+ m@(MxRegistered _ name)

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

343434 _ -> return ()

344435 mxReady

345436 ]

346437

347− (sp, rp) <- newChan

348− liftIO $ forkProcess remoteNode $ do

349− getSelfPid >>= sendChan sp

350− expect :: Process ()

351−

352− p1 <- receiveChan rp

438+ case mNode of

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

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

353441

354− register label1 p1

355− reg1 <- receiveChanTimeout delay regSink

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

442+ p1 <- receiveChanTimeout 2000000 rp

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

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

357445

358− register label2 p1

359− reg2 <- receiveChanTimeout delay regSink

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

446+ kill agent "bye"

447+ kill (fromJust p1) "bye"

361448

362− n1 <- whereis label1

363− n1 `shouldBe` equalTo (Just p1)

449+ case res of

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

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

364452

365− n2 <- whereis label2

366− n2 `shouldBe` equalTo (Just p1)

367453

368− kill p1 "goodbye"

369−

370− unreg1 <- receiveChanTimeout delay unRegSink

371− unreg2 <- receiveChanTimeout delay unRegSink

454+ where

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

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

372457

373− sort [unreg1, unreg2]

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

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

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

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

375461

376− kill agent "test-complete"

462+ replyChannel p' = do

463+ (s, r) <- newChan

464+ send p' s

465+ return r

377466

378467tests :: TestTransport -> IO [Test]

379468tests TestTransport{..} = do

380469 node1 <- newLocalNode testTransport initRemoteTable

381470 node2 <- newLocalNode testTransport initRemoteTable

382471 return [

383− testGroup "Mx Agents" [

384− testCase "Event Handling"

472+ testGroup "MxAgents" [

473+ testCase "EventHandling"

385474 (delayedAssertion

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

387476 node1 True testAgentEventHandling)

388− , testCase "Inter-Agent Broadcast"

477+ , testCase "InterAgentBroadcast"

389478 (delayedAssertion

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

391− node1 () testAgentBroadcast)

392− , testCase "Agent Mailbox Handling"

480+ node1 (Just ()) testAgentBroadcast)

481+ , testCase "AgentMailboxHandling"

393482 (delayedAssertion

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

395484 node1 (Just ()) testAgentMailboxHandling)

396− , testCase "Agent Dual Input Handling"

485+ , testCase "AgentDualInputHandling"

397486 (delayedAssertion

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

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

400− , testCase "Agent Input Prioritisation"

489+ , testCase "AgentInputPrioritisation"

401490 (delayedAssertion

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

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

404493 "third", "fourth",

405494 "fifth"]) testAgentPrioritisation)

406495 ]

407− , testGroup "Mx Events" [

408− testCase "Name Registration Events"

496+ , testGroup "MxEvents" [

497+ testCase "NameRegistrationEvents"

409498 (delayedAssertion

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

411500 node1 () testMxRegEvents)

412− , testCase "Post Death Name UnRegistration Events"

501+ , testCase "PostDeathNameUnRegistrationEvents"

413502 (delayedAssertion

414503 "expected process deaths to result in unregistration events"

415504 node1 () (testMxRegMon node2))

416− , testCase "Monitor Events"

417− (runProcess node1 (testMxUnsafeSendEvents node2))

505+ , testGroup "SendEvents" $ buildTestCases node1 node2

418506 ]

419507 ]

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

528+ ]

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

542+ ]

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+47−55

@@ -104,39 +104,26 @@ 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.

117−--

118−-- 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.

120−--

121−-- 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

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

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

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

126−--

127−

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

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

130115nsendRemote nid label msg = do

131116 proc <- ask

132117 let us = processId proc

133118 let node = processNode proc

134− if localNodeId (processNode proc) == nid

119+ if localNodeId node == nid

135120 then nsend label msg

136− else do -- 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)

139− (MxSentToName label us (wrapMessage msg))

121+ else

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

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

124+ liftIO $ traceEvent (localEventBus node)

125+ (MxSentToName lbl us (wrapMessage msg))

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

140127

141128-- | Send a message

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

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

144131 proc <- ask

145132 let node = localNodeId (processNode proc)

146133 destNode = (processNodeId them)

147− us = (processId proc) in do

148− case destNode == node of

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]

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

157− (MxSent them us (wrapMessage msg))

134+ us = (processId proc)

135+ msg' = wrapMessage msg in do

136+ -- see [note: tracing]

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

138+ (MxSent them us msg')

139+ if destNode == node

140+ then sendCtrlMsg Nothing $ LocalSend them msg'

141+ else liftIO $ sendMessage (processNode proc)

142+ (ProcessIdentifier (processId proc))

143+ (ProcessIdentifier them)

144+ NoImplicitReconnect

145+ msg

158146

159147-- | Send a message unreliably.

160148--

@@ -170,38 +158,42 @@ usend them msg = do

170158 proc <- ask

171159 let there = processNodeId them

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

161+ let msg' = wrapMessage msg

162+ -- see [note: tracing]

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

173164 if localNodeId (processNode proc) == there

174− then unsafeSendLocal them msg

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

176− (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))

165+ then sendCtrlMsg Nothing $ LocalSend them msg'

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

167+ (createMessage msg)

180168

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]

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

186176

187177-- | Send a message on a typed channel

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

189179sendChan (SendPort cid) msg = do

190180 proc <- ask

191− let node = localNodeId (processNode proc)

192− destNode = processNodeId (sendPortProcessId cid) in do

193− case destNode == node of

194− True -> unsafeSendChanLocal cid msg

195− False -> do

196− liftIO $ sendBinary (processNode proc)

197− (ProcessIdentifier (processId proc))

198− (SendPortIdentifier cid)

199− NoImplicitReconnect

200− 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

201194 where

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

203− unsafeSendChanLocal spId msg' =

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

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

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

205197

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

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