master...03b2197 · distributed-process

git.iko.soy/distributed-process

5 files changed+265−73

since splitsnapshots

unifiedsplit

distributed-process-tests/src/Control/Distributed/Process/Tests/Internal/Utils.hs+2−1 distributed-process-tests/src/Control/Distributed/Process/Tests/Mx.hs+153−15 src/Control/Distributed/Process/Internal/Primitives.hs+39−35 src/Control/Distributed/Process/Management/Internal/Types.hs+2 src/Control/Distributed/Process/UnsafePrimitives.hs+69−22

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

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

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

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

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

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

6−

7−import Control.Distributed.Process

6+import Control.Exception (SomeException)

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

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

89import Control.Distributed.Process.Node

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

11+ ( send

12+ , nsend

13+ , nsendRemote

14+ , usend

15+ )

916import Control.Distributed.Process.Management

1017 ( MxEvent(..)

1118 , MxAgentId(..)

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

2229 , mxBroadcast

2330 )

2431import Control.Monad (void, unless)

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

2533import Control.Rematch (equalTo)

2634import Data.Binary

2735import Data.List (find, sort)

28−import Data.Maybe (isJust, isNothing)

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

2937import Data.Typeable

30−import GHC.Generics

38+import GHC.Generics hiding (from)

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

3240import Prelude hiding (catch, log)

3341#endif

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

5159 (\_ -> return ())

5260 ]

5361 where

54− withMonitorRef pid = bracket (monitor pid) unmonitor

62+ withMonitorRef p = bracket (monitor p) unmonitor

5563

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

5765testAgentBroadcast result = do

@@ -313,43 +321,173 @@ testMxRegMon remoteNode result = do

313321ensure :: Process () -> Process () -> Process ()

314322ensure = flip finally

315323

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

325+

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

327+ -> Maybe LocalNode

328+ -> TestResult Bool

329+ -> Process ()

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

331+ let delay = 5000000

332+ let label = "testMxSend"

333+ let isValid = isValidLabel n label

334+

335+ register label p1

336+ reg1 <- receiveChanTimeout delay sink

337+ case reg1 of

338+ Just (MxRegistered pd lb)

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

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

341+

342+ op label ()

343+

344+ us <- getSelfPid

345+ sent <- receiveChanTimeout delay sink

346+ case sent of

347+ Just (MxSentToName lb by _)

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

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

350+

351+ where

352+ isValidLabel n l1 l2

353+ | l2 == l2 = True

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

355+ | otherwise = False

356+

357+testSend :: (ProcessId -> () -> Process ()) -> Maybe LocalNode -> TestResult Bool -> Process ()

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

359+ -- initiate a send

360+ op p1 ()

361+

362+ -- verify the management event

363+ us <- getSelfPid

364+ sent <- receiveChanTimeout 5000000 sink

365+ case sent of

366+ Just (MxSent pidTo pidFrom _)

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

368+ _ -> return False

369+

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

371+testMxSend mNode result test = do

372+ us <- getSelfPid

373+ (chan, sink) <- newChan

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

375+ mxSink $ \ev -> do

376+ case ev of

377+ m@(MxSent _ fromPid _)

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

379+ m@(MxSentToName _ fromPid _)

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

381+ m@(MxRegistered _ name)

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

383+ _ -> return ()

384+ mxReady

385+ ]

386+

387+ (sp, rp) <- newChan

388+ case mNode of

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

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

391+

392+ p1 <- receiveChan rp

393+ res <- try (test p1 sink)

394+ case res of

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

396+ Right tr -> stash result tr

397+ kill agent "bye"

398+ kill p1 "bye"

399+

400+ where

401+ label = "testMxSend"

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

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

404+

316405tests :: TestTransport -> IO [Test]

317406tests TestTransport{..} = do

318407 node1 <- newLocalNode testTransport initRemoteTable

319408 node2 <- newLocalNode testTransport initRemoteTable

409+ let nid = localNodeId node2

320410 return [

321− testGroup "Mx Agents" [

322− testCase "Event Handling"

411+ testGroup "MxAgents" [

412+ testCase "EventHandling"

323413 (delayedAssertion

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

325415 node1 True testAgentEventHandling)

326− , testCase "Inter-Agent Broadcast"

416+ , testCase "InterAgentBroadcast"

327417 (delayedAssertion

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

329419 node1 () testAgentBroadcast)

330− , testCase "Agent Mailbox Handling"

420+ , testCase "AgentMailboxHandling"

331421 (delayedAssertion

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

333423 node1 (Just ()) testAgentMailboxHandling)

334− , testCase "Agent Dual Input Handling"

424+ , testCase "AgentDualInputHandling"

335425 (delayedAssertion

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

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

338− , testCase "Agent Input Prioritisation"

428+ , testCase "AgentInputPrioritisation"

339429 (delayedAssertion

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

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

342432 "third", "fourth",

343433 "fifth"]) testAgentPrioritisation)

344434 ]

345− , testGroup "Mx Events" [

346− testCase "Name Registration Events"

435+ , testGroup "MxEvents" [

436+ testCase "NameRegistrationEvents"

347437 (delayedAssertion

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

349439 node1 () testMxRegEvents)

350− , testCase "Post Death Name UnRegistration Events"

440+ , testCase "PostDeathNameUnRegistrationEvents"

351441 (delayedAssertion

352442 "expected process deaths to result in unregistration events"

353443 node1 () (testMxRegMon node2))

444+ , testGroup "SentEvents" [

445+ testGroup "RemoteTargets" [

446+ testCase "Unsafe.nsend"

447+ (delayedAssertion "expected mx events failed"

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

449+ , testCase "Unsafe.nsendRemote"

450+ (delayedAssertion "expected mx events failed"

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

452+ , testCase "Unsafe.send"

453+ (delayedAssertion "expected mx events failed"

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

455+ , testCase "Unsafe.usend"

456+ (delayedAssertion "expected mx events failed"

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

458+ , testCase "nsend"

459+ (delayedAssertion "expected mx events failed"

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

461+ , testCase "nsendRemote"

462+ (delayedAssertion "expected mx events failed"

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

464+ , testCase "send"

465+ (delayedAssertion "expected mx events failed"

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

467+ , testCase "usend"

468+ (delayedAssertion "expected mx events failed"

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

470+ ]

471+ , testGroup "LocalTargets" [

472+ testCase "Unsafe.nsend"

473+ (delayedAssertion "expected mx events failed"

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

475+ , testCase "Unsafe.send"

476+ (delayedAssertion "expected mx events failed"

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

478+ , testCase "Unsafe.usend"

479+ (delayedAssertion "expected mx events failed"

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

481+ , testCase "nsend"

482+ (delayedAssertion "expected mx events failed"

483+ node1 True (testNSend nsend Nothing))

484+ , testCase "send"

485+ (delayedAssertion "expected mx events failed"

486+ node1 True (testSend send Nothing))

487+ , testCase "usend"

488+ (delayedAssertion "expected mx events failed"

489+ node1 True (testSend usend Nothing))

490+ ]

491+ ]

492+ ]

354493 ]

355− ]

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

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

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

484485 us = processId proc

485486 nid = localNodeId node

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

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

489+ if destNode == nid

490+ then sendCtrlMsg Nothing (LocalSend them msg)

491+ else liftIO $ sendPayload (processNode proc)

492+ (ProcessIdentifier (processId proc))

493+ (ProcessIdentifier them)

494+ NoImplicitReconnect

495+ (messageToPayload msg)

498496

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

500498--

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

508506 us = processId proc

509507 nid = localNodeId node

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

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

510+ if destNode == nid

511+ then sendCtrlMsg Nothing (LocalSend them msg)

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

519513

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

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

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

12241218

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

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

1227−nsend label msg =

1228− sendCtrlMsg Nothing (NamedSend label (createUnencodedMessage 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')

1226+ sendCtrlMsg Nothing (NamedSend label msg')

12291227

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

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

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

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

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

12391237nsendRemote nid label msg = do

1240− here <- getSelfNode

1241− if here == nid then nsend label msg

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

1238+ proc <- ask

1239+ let us = processId proc

1240+ let node = processNode proc

1241+ if localNodeId node == nid

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

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

12431247

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

12451249-- 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 =

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

5858 | MxSent ProcessId ProcessId Message

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

60+ | MxSentToName String ProcessId Message

61+ -- ^ fired whenever a named send occurs

6062 | MxReceived ProcessId Message

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

6264 | MxConnected ConnectionId EndPointAddress

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

@@ -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,12 +157,23 @@ 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

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

135182 destNode = processNodeId (sendPortProcessId cid) in do

136183 case destNode == node of

137184 True -> unsafeSendChanLocal cid msg

138− False -> do

185+ False ->

139186 liftIO $ sendBinary (processNode proc)

140187 (ProcessIdentifier (processId proc))

141188 (SendPortIdentifier cid)

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

144191 where

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

146193 unsafeSendChanLocal spId msg' =

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

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

148195

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

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