master...dc65e6d · distributed-process

git.iko.soy/distributed-process

on dc65e6d help sign in

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

since splitsnapshots

unifiedsplit

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

74import Control.Concurrent.MVar

75 ( putMVar

76 )

77−import Control.Distributed.Process

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

78import Control.Distributed.Process.Node

79import Control.Distributed.Process.Serializable()

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

82import Control.Monad (forever, void)

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

83import Control.Monad.STM (atomically)

84import Control.Monad.STM (atomically)

84import Control.Rematch hiding (match)

85import Control.Rematch hiding (match)

85import Control.Rematch.Run

86import Control.Rematch.Run

distributed-process-tests/src/Control/Distributed/Process/Tests/Mx.hs+171−15

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

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

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

6−

6+import Control.Exception (SomeException)

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

7−import Control.Distributed.Process

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

9+ ( ProcessExitException(..)

10+ , unsafeCreateUnencodedMessage

11+ )

8import Control.Distributed.Process.Node

12import Control.Distributed.Process.Node

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

14+ ( send

15+ , nsend

16+ , nsendRemote

17+ , usend

18+ )

9import Control.Distributed.Process.Management

19import Control.Distributed.Process.Management

10 ( MxEvent(..)

20 ( MxEvent(..)

11 , MxAgentId(..)

21 , MxAgentId(..)

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

22 , mxBroadcast

32 , mxBroadcast

23 )

33 )

24import Control.Monad (void, unless)

34import Control.Monad (void, unless)

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

25import Control.Rematch (equalTo)

36import Control.Rematch (equalTo)

26import Data.Binary

37import Data.Binary

27import Data.List (find, sort)

38import Data.List (find, sort)

28−import Data.Maybe (isJust, isNothing)

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

29import Data.Typeable

40import Data.Typeable

30−import GHC.Generics

41+import GHC.Generics hiding (from)

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

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

32import Prelude hiding (catch, log)

43import Prelude hiding (catch, log)

33#endif

44#endif

@@ -51,7 +62,7 @@ awaitExit pid =

51 (\_ -> return ())

62 (\_ -> return ())

52 ]

63 ]

53 where

64 where

54− withMonitorRef pid = bracket (monitor pid) unmonitor

65+ withMonitorRef p = bracket (monitor p) unmonitor

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

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

57testAgentBroadcast result = do

68testAgentBroadcast result = do

@@ -313,43 +324,188 @@ testMxRegMon remoteNode result = do

313ensure :: Process () -> Process () -> Process ()

324ensure :: Process () -> Process () -> Process ()

314ensure = flip finally

325ensure = flip finally

315

326

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

328+ -> Maybe LocalNode

329+ -> TestResult Bool

330+ -> Process ()

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

332+ let delay = 5000000

333+ let label = "testMxSend"

334+ let isValid = isValidLabel n label

335+

336+ register label p1

337+ reg1 <- receiveChanTimeout delay sink

338+ case reg1 of

339+ Just (MxRegistered pd lb)

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

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

342+

343+ op label ()

344+

345+ us <- getSelfPid

346+ sent <- receiveChanTimeout delay sink

347+ case sent of

348+ Just (MxSentToName lb by _)

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

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

351+

352+ where

353+ isValidLabel n l1 l2

354+ | l2 == l2 = True

355+ | isJust n = l2 == l1 ++ "@" ++ show (localNodeId $ fromJust n)

356+ | otherwise = False

357+

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

359+ -> Maybe LocalNode

360+ -> TestResult Bool

361+ -> Process ()

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

363+ -- initiate a send

364+ op p1 ()

365+

366+ -- verify the management event

367+ us <- getSelfPid

368+ sent <- receiveChanTimeout 5000000 sink

369+ case sent of

370+ Just (MxSent pidTo pidFrom _)

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

372+ _ -> return False

373+

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

375+

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

377+testMxSend mNode result test = do

378+ us <- getSelfPid

379+ (chan, sink) <- newChan

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

381+ mxSink $ \ev -> do

382+ case ev of

383+ m@(MxSent _ fromPid _)

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

385+ m@(MxSentToName _ fromPid _)

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

387+ m@(MxRegistered _ name)

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

389+ _ -> return ()

390+ mxReady

391+ ]

392+

393+ (sp, rp) <- newChan

394+ case mNode of

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

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

397+

398+ p1 <- receiveChan rp

399+ res <- try (test p1 sink)

400+ case res of

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

402+ Right tr -> stash result tr

403+ kill agent "bye"

404+ kill p1 "bye"

405+

406+ where

407+ label = "testMxSend"

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

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

410+

316tests :: TestTransport -> IO [Test]

411tests :: TestTransport -> IO [Test]

317tests TestTransport{..} = do

412tests TestTransport{..} = do

318 node1 <- newLocalNode testTransport initRemoteTable

413 node1 <- newLocalNode testTransport initRemoteTable

319 node2 <- newLocalNode testTransport initRemoteTable

414 node2 <- newLocalNode testTransport initRemoteTable

415+ let nid = localNodeId node2

320 return [

416 return [

321− testGroup "Mx Agents" [

417+ testGroup "MxAgents" [

322− testCase "Event Handling"

418+ testCase "EventHandling"

323 (delayedAssertion

419 (delayedAssertion

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

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

325 node1 True testAgentEventHandling)

421 node1 True testAgentEventHandling)

326− , testCase "Inter-Agent Broadcast"

422+ , testCase "InterAgentBroadcast"

327 (delayedAssertion

423 (delayedAssertion

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

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

329 node1 () testAgentBroadcast)

425 node1 () testAgentBroadcast)

330− , testCase "Agent Mailbox Handling"

426+ , testCase "AgentMailboxHandling"

331 (delayedAssertion

427 (delayedAssertion

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

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

333 node1 (Just ()) testAgentMailboxHandling)

429 node1 (Just ()) testAgentMailboxHandling)

334− , testCase "Agent Dual Input Handling"

430+ , testCase "AgentDualInputHandling"

335 (delayedAssertion

431 (delayedAssertion

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

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

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

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

338− , testCase "Agent Input Prioritisation"

434+ , testCase "AgentInputPrioritisation"

339 (delayedAssertion

435 (delayedAssertion

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

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

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

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

342 "third", "fourth",

438 "third", "fourth",

343 "fifth"]) testAgentPrioritisation)

439 "fifth"]) testAgentPrioritisation)

344 ]

440 ]

345− , testGroup "Mx Events" [

441+ , testGroup "MxEvents" [

346− testCase "Name Registration Events"

442+ testCase "NameRegistrationEvents"

347 (delayedAssertion

443 (delayedAssertion

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

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

349 node1 () testMxRegEvents)

445 node1 () testMxRegEvents)

350− , testCase "Post Death Name UnRegistration Events"

446+ , testCase "PostDeathNameUnRegistrationEvents"

351 (delayedAssertion

447 (delayedAssertion

352 "expected process deaths to result in unregistration events"

448 "expected process deaths to result in unregistration events"

353 node1 () (testMxRegMon node2))

449 node1 () (testMxRegMon node2))

450+ , testGroup "SentEvents" [

451+ testGroup "RemoteTargets" [

452+ testCase "Unsafe.nsend"

453+ (delayedAssertion "expected mx events failed"

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

455+ , testCase "Unsafe.nsendRemote"

456+ (delayedAssertion "expected mx events failed"

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

458+ , testCase "Unsafe.send"

459+ (delayedAssertion "expected mx events failed"

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

461+ , testCase "Unsafe.usend"

462+ (delayedAssertion "expected mx events failed"

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

464+ , testCase "nsend"

465+ (delayedAssertion "expected mx events failed"

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

467+ , testCase "nsendRemote"

468+ (delayedAssertion "expected mx events failed"

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

470+ , testCase "send"

471+ (delayedAssertion "expected mx events failed"

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

473+ , testCase "usend"

474+ (delayedAssertion "expected mx events failed"

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

476+ , testCase "forward"

477+ (delayedAssertion "expected mx events failed"

478+ node1 True (testSend (\p m -> forward (unsafeCreateUnencodedMessage m) p) $ Just node2))

479+ , testCase "uforward"

480+ (delayedAssertion "expected mx events failed"

481+ node1 True (testSend (\p m -> uforward (unsafeCreateUnencodedMessage m) p) $ Just node2))

482+ ]

483+ , testGroup "LocalTargets" [

484+ testCase "Unsafe.nsend"

485+ (delayedAssertion "expected mx events failed"

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

487+ , testCase "Unsafe.send"

488+ (delayedAssertion "expected mx events failed"

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

490+ , testCase "Unsafe.usend"

491+ (delayedAssertion "expected mx events failed"

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

493+ , testCase "nsend"

494+ (delayedAssertion "expected mx events failed"

495+ node1 True (testNSend nsend Nothing))

496+ , testCase "send"

497+ (delayedAssertion "expected mx events failed"

498+ node1 True (testSend send Nothing))

499+ , testCase "usend"

500+ (delayedAssertion "expected mx events failed"

501+ node1 True (testSend usend Nothing))

502+ , testCase "forward"

503+ (delayedAssertion "expected mx events failed"

504+ node1 True (testSend (\p m -> forward (unsafeCreateUnencodedMessage m) p) Nothing))

505+ , testCase "uforward"

506+ (delayedAssertion "expected mx events failed"

507+ node1 True (testSend (\p m -> uforward (unsafeCreateUnencodedMessage m) p) Nothing))

508+ ]

509+ ]

510+ ]

354 ]

511 ]

355− ]

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

@@ -254,17 +254,15 @@ send them msg = do

254 node = processNode proc

255 nodeId = localNodeId node

256 destNode = (processNodeId them) in do

257− case destNode == nodeId of

258− True -> sendLocal them msg

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

260− (ProcessIdentifier (processId proc))

261− (ProcessIdentifier them)

262− NoImplicitReconnect

263− msg

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

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

266 liftIO $ traceEvent (localEventBus node)

257 liftIO $ traceEvent (localEventBus node)

267 (MxSent them us (createUnencodedMessage msg))

258 (MxSent them us (createUnencodedMessage msg))

259+ if destNode == nodeId

260+ then sendLocal them msg

261+ else liftIO $ sendMessage (processNode proc)

262+ (ProcessIdentifier (processId proc))

263+ (ProcessIdentifier them)

264+ NoImplicitReconnect

265+ msg

268

266

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

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

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

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

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

284--

282--

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

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

286usend them msg = do

284usend them msg = do

287− here <- getSelfNode

285+ proc <- ask

288 let there = processNodeId them

286 let there = processNodeId them

289− if here == there

287+ let (us, node) = (processId proc, processNode proc)

288+ let msg' = wrapMessage msg

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

290+ if localNodeId (processNode proc) == there

290 then sendLocal them msg

291 then sendLocal them msg

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

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

292 (createMessage msg)

293 (createMessage msg)

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

484 us = processId proc

485 us = processId proc

485 nid = localNodeId node

486 nid = localNodeId node

486 destNode = (processNodeId them) in do

487 destNode = (processNodeId them) in do

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

487− case destNode == nid of

489+ if destNode == nid

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

490+ then sendCtrlMsg Nothing (LocalSend them msg)

489− False -> liftIO $ sendPayload (processNode proc)

491+ else liftIO $ sendPayload (processNode proc)

490− (ProcessIdentifier (processId proc))

492+ (ProcessIdentifier (processId proc))

491− (ProcessIdentifier them)

493+ (ProcessIdentifier them)

492− NoImplicitReconnect

494+ NoImplicitReconnect

493− (messageToPayload msg)

495+ (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)

498

496

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

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

500--

498--

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

508 us = processId proc

506 us = processId proc

509 nid = localNodeId node

507 nid = localNodeId node

510 destNode = (processNodeId them) in do

508 destNode = (processNodeId them) in do

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

511− case destNode == nid of

510+ if destNode == nid

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

511+ then sendCtrlMsg Nothing (LocalSend them msg)

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

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

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)

519

513

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

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

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

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

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

1224

1218

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

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

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

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

1227−nsend label msg =

1221+nsend label msg = do

1222+ proc <- ask

1223+ let msg' = createUnencodedMessage msg

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

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

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

1226+ sendCtrlMsg Nothing (NamedSend label msg')

1229

1227

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

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

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

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

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

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

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

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

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

1239nsendRemote nid label msg = do

1237nsendRemote nid label msg = do

1240− here <- getSelfNode

1238+ proc <- ask

1239+ let us = processId proc

1240+ let node = processNode proc

1241+ if localNodeId node == nid

1241− if here == nid then nsend label msg

1242+ then nsend label msg

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

1244+ liftIO $ traceEvent (localEventBus node)

1245+ (MxSentToName lbl us (wrapMessage msg))

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

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

1243

1247

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

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

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

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

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

@@ -57,6 +57,8 @@ data MxEvent =

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

58 | MxSent ProcessId ProcessId Message

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

60+ | MxSentToName String ProcessId Message

61+ -- ^ fired whenever a named send occurs

60 | MxReceived ProcessId Message

62 | MxReceived ProcessId Message

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

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

62 | MxConnected ConnectionId EndPointAddress

64 | MxConnected ConnectionId EndPointAddress

src/Control/Distributed/Process/UnsafePrimitives.hs+69−22

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

32-- the /normal/ strategy).

33--

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

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

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

52+-- You have been warned!

36--

53--

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

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

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

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

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

55 , sendBinary

72 , sendBinary

56 , sendCtrlMsg

73 , sendCtrlMsg

57 )

74 )

58−

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

76+ ( MxEvent(..)

77+ )

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

79+ ( traceEvent

80+ )

59import Control.Distributed.Process.Internal.Types

81import Control.Distributed.Process.Internal.Types

60 ( ProcessId(..)

82 ( ProcessId(..)

61 , NodeId(..)

83 , NodeId(..)

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

101

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

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

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

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

82−nsend label 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')

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

111+ sendCtrlMsg Nothing (NamedSend label msg')

112

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

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

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

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

87nsendRemote nid label msg = do

115nsendRemote nid label msg = do

88 proc <- ask

116 proc <- ask

117+ let us = processId proc

118+ let node = processNode proc

89− if localNodeId (processNode proc) == nid

119+ if localNodeId node == nid

90 then nsend label msg

120 then nsend label 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))

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

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

127

93-- | Send a message

128-- | Send a message

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

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

95send them msg = do

130send them msg = do

96 proc <- ask

131 proc <- ask

97 let node = localNodeId (processNode proc)

132 let node = localNodeId (processNode proc)

98− destNode = (processNodeId them) in do

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

99− case destNode == node of

139+ if destNode == node

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

140+ then sendCtrlMsg Nothing $ LocalSend them msg'

141+ else liftIO $ sendMessage (processNode proc)

142+ (ProcessIdentifier (processId proc))

143+ (ProcessIdentifier them)

144+ NoImplicitReconnect

145+ msg

110

146

111-- | Send a message unreliably.

147-- | Send a message unreliably.

112--

148--

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

121usend them msg = do

157usend them msg = do

122 proc <- ask

158 proc <- ask

123 let there = processNodeId them

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

124 if localNodeId (processNode proc) == there

164 if localNodeId (processNode proc) == there

125− then sendCtrlMsg Nothing $

165+ then sendCtrlMsg Nothing $ LocalSend them msg'

126− LocalSend them (unsafeCreateUnencodedMessage msg)

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

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

128 (createMessage msg)

167 (createMessage msg)

129

168

169+-- [note: tracing]

170+-- Note that tracing writes to the local node's control channel, and this

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

172+-- encoding. The same is true for the messages it puts onto the Management

173+-- event bus, however we do *not* want unevaluated thunks hitting the event

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

175+--

176+

130-- | Send a message on a typed channel

177-- | Send a message on a typed channel

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

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

132sendChan (SendPort cid) msg = do

179sendChan (SendPort cid) msg = do

@@ -135,7 +182,7 @@ sendChan (SendPort cid) msg = do

135 destNode = processNodeId (sendPortProcessId cid) in do

182 destNode = processNodeId (sendPortProcessId cid) in do

136 case destNode == node of

183 case destNode == node of

137 True -> unsafeSendChanLocal cid msg

184 True -> unsafeSendChanLocal cid msg

138− False -> do

185+ False ->

139 liftIO $ sendBinary (processNode proc)

186 liftIO $ sendBinary (processNode proc)

140 (ProcessIdentifier (processId proc))

187 (ProcessIdentifier (processId proc))

141 (SendPortIdentifier cid)

188 (SendPortIdentifier cid)

@@ -144,7 +191,7 @@ sendChan (SendPort cid) msg = do

144 where

191 where

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

192 unsafeSendChanLocal :: (Serializable a) => SendPortId -> a -> Process ()

146 unsafeSendChanLocal spId msg' =

193 unsafeSendChanLocal spId msg' =

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

194+ sendCtrlMsg Nothing $ LocalPortSend spId (wrapMessage msg')

148

195

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

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

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

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