master...d2b5d82 · distributed-process

git.iko.soy/distributed-process

on d2b5d82 help sign in

distributed-process-tests/src/Control/Distributed/Process/Tests/Mx.hs+64

since splitsnapshots

unifiedsplit

@@ -6,6 +6,12 @@ import Network.Transport.Test (TestTransport(..))

7import Control.Distributed.Process

8import Control.Distributed.Process.Node

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

10+ ( send

11+ , nsend

12+ , nsendRemote

13+ , usend

14+ )

9import Control.Distributed.Process.Management

15import Control.Distributed.Process.Management

10 ( MxEvent(..)

16 ( MxEvent(..)

11 , MxAgentId(..)

17 , MxAgentId(..)

@@ -313,6 +319,62 @@ testMxRegMon remoteNode result = do

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

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

314ensure = flip finally

320ensure = flip finally

315

321

322+

323+testMxUnsafeSendEvents :: LocalNode -> Process ()

324+testMxUnsafeSendEvents remoteNode = do

325+

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

328+

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

337+ mxSink $ \ev -> do

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

343+ _ -> return ()

344+ mxReady

345+ ]

346+

347+ (sp, rp) <- newChan

348+ liftIO $ forkProcess remoteNode $ do

349+ getSelfPid >>= sendChan sp

350+ expect :: Process ()

351+

352+ p1 <- receiveChan rp

353+

354+ register label1 p1

355+ reg1 <- receiveChanTimeout delay regSink

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

357+

358+ register label2 p1

359+ reg2 <- receiveChanTimeout delay regSink

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

361+

362+ n1 <- whereis label1

363+ n1 `shouldBe` equalTo (Just p1)

364+

365+ n2 <- whereis label2

366+ n2 `shouldBe` equalTo (Just p1)

367+

368+ kill p1 "goodbye"

369+

370+ unreg1 <- receiveChanTimeout delay unRegSink

371+ unreg2 <- receiveChanTimeout delay unRegSink

372+

373+ sort [unreg1, unreg2]

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

375+

376+ kill agent "test-complete"

377+

316tests :: TestTransport -> IO [Test]

378tests :: TestTransport -> IO [Test]

317tests TestTransport{..} = do

379tests TestTransport{..} = do

318 node1 <- newLocalNode testTransport initRemoteTable

380 node1 <- newLocalNode testTransport initRemoteTable

@@ -351,5 +413,7 @@ tests TestTransport{..} = do

351 (delayedAssertion

413 (delayedAssertion

352 "expected process deaths to result in unregistration events"

414 "expected process deaths to result in unregistration events"

353 node1 () (testMxRegMon node2))

415 node1 () (testMxRegMon node2))

416+ , testCase "Monitor Events"

417+ (runProcess node1 (testMxUnsafeSendEvents node2))

354 ]

418 ]

355 ]

419 ]

src/Control/Distributed/Process/Internal/Primitives.hs+13−15

@@ -486,15 +486,13 @@ forward msg them = do

486 destNode = (processNodeId them) in do

487 case destNode == nid of

488 True -> sendCtrlMsg Nothing (LocalSend them msg)

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

489+ False ->

490+ liftIO $ do

490− (ProcessIdentifier (processId proc))

491+ sendPayload (processNode proc) (ProcessIdentifier (processId proc)) (ProcessIdentifier them) NoImplicitReconnect (messageToPayload msg)

491− (ProcessIdentifier them)

492− NoImplicitReconnect

493− (messageToPayload msg)

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

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

493+ -- In this remote case, 'sendMessage' can block in the networking stack.

496− liftIO $ traceEvent (localEventBus node)

494+ liftIO $ traceEvent (localEventBus node)

497− (MxSent them us msg)

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

@@ -510,12 +508,12 @@ uforward msg them = do

510 destNode = (processNodeId them) in do

508 destNode = (processNodeId them) in do

511 case destNode == nid of

509 case destNode == nid of

512 True -> sendCtrlMsg Nothing (LocalSend them msg)

510 True -> sendCtrlMsg Nothing (LocalSend them msg)

511+ False -> do

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

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

514− msg

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

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

514+ -- In the remote case, 'sendCtrlMsg' can block in the networking stack.

517− liftIO $ traceEvent (localEventBus node)

515+ liftIO $ traceEvent (localEventBus node)

518− (MxSent them us msg)

516+ (MxSent them us msg)

519

517

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

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

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

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

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+77−20

@@ -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,60 @@ 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 node = localNodeId (processNode proc)

108+ let msg' = wrapMessage msg

109+ -- see [note: tracing]

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

111+ (MxSentToName label us msg')

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

112+ sendCtrlMsg Nothing (NamedSend label msg')

113+

114+-- [note: tracing]

115+-- In the remote case, we do not fire a trace event until after sending is

116+-- complete, since 'sendMessage' can block in the networking stack.

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

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

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

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

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

87nsendRemote nid label msg = do

130nsendRemote nid label msg = do

88 proc <- ask

131 proc <- ask

132+ let us = processId proc

133+ let node = processNode proc

89 if localNodeId (processNode proc) == nid

134 if localNodeId (processNode proc) == nid

90 then nsend label msg

135 then nsend label msg

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

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

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

138+ liftIO $ traceEvent (localEventBus node)

139+ (MxSentToName label us (wrapMessage msg))

140

93-- | Send a message

141-- | Send a message

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

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

95send them msg = do

143send them msg = do

96 proc <- ask

144 proc <- ask

97 let node = localNodeId (processNode proc)

145 let node = localNodeId (processNode proc)

98− destNode = (processNodeId them) in do

146+ destNode = (processNodeId them)

147+ us = (processId proc) in do

99 case destNode == node of

148 case destNode == node of

100 True -> unsafeSendLocal them msg

149 True -> unsafeSendLocal them msg

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

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

102− (ProcessIdentifier (processId proc))

151+ (ProcessIdentifier (processId proc))

103− (ProcessIdentifier them)

152+ (ProcessIdentifier them)

104− NoImplicitReconnect

153+ NoImplicitReconnect

105− msg

154+ msg

106− where

155+ -- see [note: tracing]

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

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

108− unsafeSendLocal pid msg' =

157+ (MxSent them us (wrapMessage msg))

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

110

158

111-- | Send a message unreliably.

159-- | Send a message unreliably.

112--

160--

@@ -121,11 +169,20 @@ usend :: Serializable a => ProcessId -> a -> Process ()

121usend them msg = do

169usend them msg = do

122 proc <- ask

170 proc <- ask

123 let there = processNodeId them

171 let there = processNodeId them

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

124 if localNodeId (processNode proc) == there

173 if localNodeId (processNode proc) == there

125− then sendCtrlMsg Nothing $

174+ then unsafeSendLocal them msg

126− LocalSend them (unsafeCreateUnencodedMessage msg)

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

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

128− (createMessage msg)

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

180+

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

182+unsafeSendLocal pid msg =

183+ let msg' = wrapMessage msg in do

184+ sendCtrlMsg Nothing $ LocalSend pid msg'

185+ -- see [note: tracing]

129

186

130-- | Send a message on a typed channel

187-- | Send a message on a typed channel

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

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

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

144 where

201 where

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

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

146 unsafeSendChanLocal spId msg' =

203 unsafeSendChanLocal spId msg' =

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

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

148

205

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

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

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

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