03b2197...master · distributed-process

git.iko.soy/distributed-process

10 files changed+536−162

since splitsnapshots

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+227−40

@@ -1,11 +1,23 @@

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

14+import qualified Control.Distributed.Process.UnsafePrimitives as Unsafe

15+ ( send

16+ , nsend

17+ , nsendRemote

18+ , usend

19+ , sendChan

20+ )

921import Control.Distributed.Process.Management

1022 ( MxEvent(..)

1123 , MxAgentId(..)

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

2234 , mxBroadcast

2335 )

2436import Control.Monad (void, unless)

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

2538import Control.Rematch (equalTo)

2639import Data.Binary

27−import Data.List (find, sort)

28−import Data.Maybe (isJust, isNothing)

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

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

2942import Data.Typeable

30−import GHC.Generics

43+import GHC.Generics hiding (from)

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

3245import Prelude hiding (catch, log)

3346#endif

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

5164 (\_ -> return ())

5265 ]

5366 where

54− withMonitorRef pid = bracket (monitor pid) unmonitor

67+ withMonitorRef p = bracket (monitor p) unmonitor

5568

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

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

5770testAgentBroadcast result = do

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

5972

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

6881 mxNotify ()

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

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

71− stash result =<< receiveChan resultRP

84+ stash result =<< receiveChanTimeout 10000000 resultRP

7285

73− kill publisher "finished" >> awaitExit publisher

74− kill consumer "finished" >> awaitExit consumer

86+ kill publisher "finished"

87+ kill consumer "finished"

7588

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

7790testAgentDualInput result = do

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

8699 else mxReady)

87100 ]

88101

102+ mRef <- monitor s

103+

89104 mxNotify (1 :: Int)

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

91106 mxNotify (2 :: Int)

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

93108 mxNotify (5 :: Int)

94109

95110 stash result =<< receiveChanTimeout 10000000 rp

96− awaitExit s

111+ died <- receiveTimeout 10000000 [

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

113+ ]

114+ died `shouldBe` equalTo (Just True)

97115

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

99117testAgentPrioritisation result = do

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

107125 (sp, rp) <- newChan

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

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

110− mxUpdateLocal ((s:))

128+ mxUpdateLocal (s:)

111129 st <- mxGetLocal

112130 case length st of

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

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

114132 mxDeactivate "finished"

115133 _ -> mxReceive -- go to the mailbox

116134 )

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

121139 mxNotify "fourth"

122140 nsend name "fifth"

123141

124− stash result . sort =<< receiveChan rp

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

125143 awaitExit s

126144

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

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

134152 nsend "mailbox-agent" ()

135153

136154 stash result =<< receiveChanTimeout 1000000 rp

137− kill agent "finished" >> awaitExit agent

155+ kill agent "finished"

138156

139157testAgentEventHandling :: TestResult Bool -> Process ()

140158testAgentEventHandling result = do

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

142166

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

143168 (rc, rs) <- newChan

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

145170 (mxSink $ \ev -> do

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

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

173198 faff 2000000 rs

174199

175− _ <- monitor agentPid

176− (sp, rp) <- newChan

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

177201

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

179203

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

182206 monitor pid

183207

184208 send pid ()

185− () <- receiveChan rp

209+ rct <- receiveChanTimeout 10000000 rp

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

186211

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

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

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

188214

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

190216 faff 2000000 rs

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

193219 mxNotify (MxSpawned pid, replyTo)

194220 mxNotify (MxProcessDied pid DiedNormal, replyTo)

195221

196− seenAlive <- receiveChan reply

197− seenDead <- receiveChan reply

222+ seenAlive <- receiveChanTimeout 10000000 reply

223+ seenDead <- receiveChanTimeout 10000000 reply

198224

199− stash result $ seenAlive && seenDead

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

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

226+ kill timer "test-complete"

227+ kill agentPid "test-complete"

201228 where

202229 faff delay port = do

203230 res <- receiveChanTimeout delay port

204− unless (isNothing res) $ do

205− pause delay

206− faff delay port

231+ unless (isNothing res) $ pause delay

207232

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

209234testMxRegEvents result = do

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

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

251276

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

253− awaitExit agent

254278

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

256280testMxRegMon remoteNode result = do

@@ -308,48 +332,211 @@ testMxRegMon remoteNode result = do

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

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

310334

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

335+ kill agent "test-complete"

312336

313337ensure :: Process () -> Process () -> Process ()

314338ensure = flip finally

315339

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

356+

357+ op label ()

358+

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

364+

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

423+ mxSink $ \ev -> do

424+ case ev of

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

434+ _ -> return ()

435+ mxReady

436+ ]

437+

438+ case mNode of

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

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

441+

442+ p1 <- receiveChanTimeout 2000000 rp

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

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

445+

446+ kill agent "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+

453+

454+ where

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

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

466+

316467tests :: TestTransport -> IO [Test]

317468tests TestTransport{..} = do

318469 node1 <- newLocalNode testTransport initRemoteTable

319470 node2 <- newLocalNode testTransport initRemoteTable

320471 return [

321− testGroup "Mx Agents" [

322− testCase "Event Handling"

472+ testGroup "MxAgents" [

473+ testCase "EventHandling"

323474 (delayedAssertion

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

325476 node1 True testAgentEventHandling)

326− , testCase "Inter-Agent Broadcast"

477+ , testCase "InterAgentBroadcast"

327478 (delayedAssertion

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

329− node1 () testAgentBroadcast)

330− , testCase "Agent Mailbox Handling"

480+ node1 (Just ()) testAgentBroadcast)

481+ , testCase "AgentMailboxHandling"

331482 (delayedAssertion

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

333484 node1 (Just ()) testAgentMailboxHandling)

334− , testCase "Agent Dual Input Handling"

485+ , testCase "AgentDualInputHandling"

335486 (delayedAssertion

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

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

338− , testCase "Agent Input Prioritisation"

489+ , testCase "AgentInputPrioritisation"

339490 (delayedAssertion

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

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

342493 "third", "fourth",

343494 "fifth"]) testAgentPrioritisation)

344495 ]

345− , testGroup "Mx Events" [

346− testCase "Name Registration Events"

496+ , testGroup "MxEvents" [

497+ testCase "NameRegistrationEvents"

347498 (delayedAssertion

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

349500 node1 () testMxRegEvents)

350− , testCase "Post Death Name UnRegistration Events"

501+ , testCase "PostDeathNameUnRegistrationEvents"

351502 (delayedAssertion

352503 "expected process deaths to result in unregistration events"

353504 node1 () (testMxRegMon node2))

505+ , testGroup "SendEvents" $ buildTestCases node1 node2

354506 ]

355507 ]

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

@@ -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,17 +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 -> liftIO $ sendPayload (processNode proc)

490− (ProcessIdentifier (processId proc))

491− (ProcessIdentifier them)

492− NoImplicitReconnect

493− (messageToPayload msg)

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

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

496− liftIO $ traceEvent (localEventBus node)

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

498505

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

500507--

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

508515 us = processId proc

509516 nid = localNodeId node

510517 destNode = (processNodeId them) in do

511− case destNode == nid of

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

513− False -> sendCtrlMsg (Just destNode) $ UnreliableSend (processLocalId them)

514− msg

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

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

517− liftIO $ traceEvent (localEventBus node)

518− (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

519522

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

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

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

12241227

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

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

1227−nsend label msg =

1228− 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')

12291236

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

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

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

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

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

12391246nsendRemote nid label msg = do

1240− here <- getSelfNode

1241− if here == nid then nsend label msg

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

12431256

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

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

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

@@ -57,8 +58,14 @@ data MxEvent =

5758 -- ^ fired whenever a node /dies/ (i.e., the connection is broken/disconnected)

5859 | MxSent ProcessId ProcessId Message

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

61+ | MxSentToName String ProcessId Message

62+ -- ^ fired whenever a named send occurs

63+ | MxSentToPort ProcessId SendPortId Message

64+ -- ^ fired whenever a sendChan occurs

6065 | MxReceived ProcessId Message

6166 -- ^ 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

6269 | MxConnected ConnectionId EndPointAddress

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

6471 | 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+82−33

@@ -32,7 +32,24 @@

3232-- the /normal/ strategy).

3333--

3434-- Use of the functions in this module can potentially change the runtime

35−-- behaviour of your application. You have been warned!

35+-- behaviour of your application. In addition, messages passed between Cloud

36+-- Haskell processes are written to a tracing infrastructure on the local node,

37+-- to provide improved introspection and debugging facilities for complex actor

38+-- based systems. This module makes no attempt to force evaluation in these

39+-- cases either, thus evaluation problems in passed data structures could not

40+-- only crash your processes, but could also bring down critical internal

41+-- services on which the node relies to function correctly.

42+--

43+-- If you wish to repudiate such issues, you are advised to consider the use

44+-- of NFSerialisable in the distributed-process-extras package, which type

45+-- class brings NFData into scope along with Serializable, such that we can

46+-- force evaluation. Intended for use with modules such as this one, this

47+-- approach guarantees correct evaluatedness in terms of @NFData@. Please note

48+-- however, that we /cannot/ guarantee that an @NFData@ instance will behave the

49+-- same way as a @Binary@ one with regards evaluation, so it is still possible

50+-- to introduce unexpected behaviour by using /unsafe/ primitives in this way.

51+--

52+-- You have been warned!

3653--

3754-- This module is exported so that you can replace the use of Cloud Haskell's

3855-- /safe/ messaging primitives. If you want to use both variants, then you can

@@ -55,7 +72,12 @@ import Control.Distributed.Process.Internal.Messaging

5572 , sendBinary

5673 , sendCtrlMsg

5774 )

58−

75+import Control.Distributed.Process.Management.Internal.Types

76+ ( MxEvent(..)

77+ )

78+import Control.Distributed.Process.Management.Internal.Trace.Types

79+ ( traceEvent

80+ )

5981import Control.Distributed.Process.Internal.Types

6082 ( ProcessId(..)

6183 , NodeId(..)

@@ -79,34 +101,48 @@ import Control.Monad.Reader (ask)

79101

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

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

82−nsend label msg =

83− sendCtrlMsg Nothing (NamedSend label (unsafeCreateUnencodedMessage msg))

104+nsend label msg = do

105+ proc <- ask

106+ let us = processId proc

107+ let msg' = wrapMessage msg

108+ -- see [note: tracing]

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

110+ (MxSentToName label us msg')

111+ sendCtrlMsg Nothing (NamedSend label msg')

84112

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

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

87115nsendRemote nid label msg = do

88116 proc <- ask

89− if localNodeId (processNode proc) == nid

117+ let us = processId proc

118+ let node = processNode proc

119+ if localNodeId node == nid

90120 then nsend label msg

91− else sendCtrlMsg (Just nid) (NamedSend label (createMessage 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))

92127

93128-- | Send a message

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

95130send them msg = do

96131 proc <- ask

97132 let node = localNodeId (processNode proc)

98− destNode = (processNodeId them) in do

99− case destNode == node of

100− True -> unsafeSendLocal them msg

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

102− (ProcessIdentifier (processId proc))

103− (ProcessIdentifier them)

104− NoImplicitReconnect

105− msg

106− where

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

108− unsafeSendLocal pid msg' =

109− sendCtrlMsg Nothing $ LocalSend pid (unsafeCreateUnencodedMessage msg')

133+ destNode = (processNodeId them)

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

110146

111147-- | Send a message unreliably.

112148--

@@ -121,30 +157,43 @@ usend :: Serializable a => ProcessId -> a -> Process ()

121157usend them msg = do

122158 proc <- ask

123159 let there = processNodeId them

160+ 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')

124164 if localNodeId (processNode proc) == there

125− then sendCtrlMsg Nothing $

126− LocalSend them (unsafeCreateUnencodedMessage msg)

165+ then sendCtrlMsg Nothing $ LocalSend them msg'

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

128167 (createMessage msg)

129168

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+

130177-- | Send a message on a typed channel

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

132179sendChan (SendPort cid) msg = do

133180 proc <- ask

134− let node = localNodeId (processNode proc)

135− destNode = processNodeId (sendPortProcessId cid) in do

136− case destNode == node of

137− True -> unsafeSendChanLocal cid msg

138− False -> do

139− liftIO $ sendBinary (processNode proc)

140− (ProcessIdentifier (processId proc))

141− (SendPortIdentifier cid)

142− NoImplicitReconnect

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

144194 where

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

146− unsafeSendChanLocal spId msg' =

147− sendCtrlMsg Nothing $ LocalPortSend spId (unsafeCreateUnencodedMessage msg')

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

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

148197

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

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