master...0514f3c · distributed-process

git.iko.soy/distributed-process

on 0514f3c 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+142−14

@@ -3,9 +3,16 @@ 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

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

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

8import Control.Distributed.Process.Node

9import Control.Distributed.Process.Node

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

11+ ( send

12+ , nsend

13+ , nsendRemote

14+ , usend

15+ )

9import Control.Distributed.Process.Management

16import Control.Distributed.Process.Management

10 ( MxEvent(..)

17 ( MxEvent(..)

11 , MxAgentId(..)

18 , MxAgentId(..)

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

22 , mxBroadcast

29 , mxBroadcast

23 )

30 )

24import Control.Monad (void, unless)

31import Control.Monad (void, unless)

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

25import Control.Rematch (equalTo)

33import Control.Rematch (equalTo)

26import Data.Binary

34import Data.Binary

27import Data.List (find, sort)

35import Data.List (find, sort)

28import Data.Maybe (isJust, isNothing)

36import Data.Maybe (isJust, isNothing)

29import Data.Typeable

37import Data.Typeable

30−import GHC.Generics

38+import GHC.Generics hiding (from)

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

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

32import Prelude hiding (catch, log)

40import Prelude hiding (catch, log)

33#endif

41#endif

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

51 (\_ -> return ())

59 (\_ -> return ())

52 ]

60 ]

53 where

61 where

54− withMonitorRef pid = bracket (monitor pid) unmonitor

62+ withMonitorRef p = bracket (monitor p) unmonitor

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

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

57testAgentBroadcast result = do

65testAgentBroadcast result = do

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

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

321ensure :: Process () -> Process () -> Process ()

314ensure = flip finally

322ensure = flip finally

315

323

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

325+

326+testNSend :: (String -> () -> Process ()) -> Maybe LocalNode -> TestResult Bool -> Process ()

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

328+ let delay = 5000000

329+ let label = "testMxSend"

330+

331+ register label p1

332+ reg1 <- receiveChanTimeout delay sink

333+ case reg1 of

334+ Just (MxRegistered pd lb)

335+ | pd == p1 && lb == label -> return ()

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

337+

338+ op label ()

339+

340+ us <- getSelfPid

341+ sent <- receiveChanTimeout delay sink

342+ case sent of

343+ Just (MxSentToName lb by _)

344+ | by == us && lb == label -> return True

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

346+

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

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

349+ -- initiate a send

350+ op p1 ()

351+

352+ -- verify the management event

353+ us <- getSelfPid

354+ sent <- receiveChanTimeout 5000000 sink

355+ case sent of

356+ Just (MxSent pidTo pidFrom _)

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

358+ _ -> return False

359+

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

361+testMxSend mNode result test = do

362+ us <- getSelfPid

363+ (chan, sink) <- newChan

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

365+ mxSink $ \ev -> do

366+ case ev of

367+ m@(MxSent _ fromPid _)

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

369+ m@(MxSentToName _ fromPid _)

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

371+ m@(MxRegistered _ name)

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

373+ _ -> return ()

374+ mxReady

375+ ]

376+

377+ (sp, rp) <- newChan

378+ case mNode of

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

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

381+

382+ p1 <- receiveChan rp

383+ res <- try (test p1 sink)

384+ case res of

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

386+ Right tr -> stash result tr

387+ kill agent "bye"

388+ kill p1 "bye"

389+

390+ where

391+ label = "testMxSend"

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

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

394+

316tests :: TestTransport -> IO [Test]

395tests :: TestTransport -> IO [Test]

317tests TestTransport{..} = do

396tests TestTransport{..} = do

318 node1 <- newLocalNode testTransport initRemoteTable

397 node1 <- newLocalNode testTransport initRemoteTable

319 node2 <- newLocalNode testTransport initRemoteTable

398 node2 <- newLocalNode testTransport initRemoteTable

399+ let nid = localNodeId node2

320 return [

400 return [

321− testGroup "Mx Agents" [

401+ testGroup "MxAgents" [

322− testCase "Event Handling"

402+ testCase "EventHandling"

323 (delayedAssertion

403 (delayedAssertion

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

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

325 node1 True testAgentEventHandling)

405 node1 True testAgentEventHandling)

326− , testCase "Inter-Agent Broadcast"

406+ , testCase "InterAgentBroadcast"

327 (delayedAssertion

407 (delayedAssertion

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

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

329 node1 () testAgentBroadcast)

409 node1 () testAgentBroadcast)

330− , testCase "Agent Mailbox Handling"

410+ , testCase "AgentMailboxHandling"

331 (delayedAssertion

411 (delayedAssertion

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

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

333 node1 (Just ()) testAgentMailboxHandling)

413 node1 (Just ()) testAgentMailboxHandling)

334− , testCase "Agent Dual Input Handling"

414+ , testCase "AgentDualInputHandling"

335 (delayedAssertion

415 (delayedAssertion

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

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

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

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

338− , testCase "Agent Input Prioritisation"

418+ , testCase "AgentInputPrioritisation"

339 (delayedAssertion

419 (delayedAssertion

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

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

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

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

342 "third", "fourth",

422 "third", "fourth",

343 "fifth"]) testAgentPrioritisation)

423 "fifth"]) testAgentPrioritisation)

344 ]

424 ]

345− , testGroup "Mx Events" [

425+ , testGroup "MxEvents" [

346− testCase "Name Registration Events"

426+ testCase "NameRegistrationEvents"

347 (delayedAssertion

427 (delayedAssertion

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

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

349 node1 () testMxRegEvents)

429 node1 () testMxRegEvents)

350− , testCase "Post Death Name UnRegistration Events"

430+ , testCase "PostDeathNameUnRegistrationEvents"

351 (delayedAssertion

431 (delayedAssertion

352 "expected process deaths to result in unregistration events"

432 "expected process deaths to result in unregistration events"

353 node1 () (testMxRegMon node2))

433 node1 () (testMxRegMon node2))

434+ , testGroup "SentEvents" [

435+ testGroup "RemoteTargets" [

436+ testCase "Unsafe.nsend"

437+ (delayedAssertion "expected mx events failed"

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

439+ , testCase "Unsafe.nsendRemote"

440+ (delayedAssertion "expected mx events failed"

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

442+ , testCase "Unsafe.send"

443+ (delayedAssertion "expected mx events failed"

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

445+ , testCase "Unsafe.usend"

446+ (delayedAssertion "expected mx events failed"

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

448+ , testCase "nsend"

449+ (delayedAssertion "expected mx events failed"

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

451+ , testCase "nsendRemote"

452+ (delayedAssertion "expected mx events failed"

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

454+ , testCase "send"

455+ (delayedAssertion "expected mx events failed"

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

457+ , testCase "usend"

458+ (delayedAssertion "expected mx events failed"

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

460+ ]

461+ , testGroup "LocalTargets" [

462+ testCase "Unsafe.nsend"

463+ (delayedAssertion "expected mx events failed"

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

465+ , testCase "Unsafe.send"

466+ (delayedAssertion "expected mx events failed"

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

468+ , testCase "Unsafe.usend"

469+ (delayedAssertion "expected mx events failed"

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

471+ , testCase "nsend"

472+ (delayedAssertion "expected mx events failed"

473+ node1 True (testNSend nsend Nothing))

474+ , testCase "send"

475+ (delayedAssertion "expected mx events failed"

476+ node1 True (testSend send Nothing))

477+ , testCase "usend"

478+ (delayedAssertion "expected mx events failed"

479+ node1 True (testSend usend Nothing))

480+ ]

481+ ]

482+ ]

354 ]

483 ]

355− ]

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+76−21

@@ -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,65 @@ 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. For locally

117+-- registered processes, we trace before sending the NC a control message, since

118+-- we cannot guarantee ordering otherwise.

119+--

120+-- In addition, MxSent trace messages are dispatched by the node controller's

121+-- thread for local sends, which also covers usend.

122+--

123+-- Also note that tracing writes to the local node's control channel, and this

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

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

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

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

128+--

129

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

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

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

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

87nsendRemote nid label msg = do

132nsendRemote nid label msg = do

88 proc <- ask

133 proc <- ask

134+ let us = processId proc

135+ let node = processNode proc

89 if localNodeId (processNode proc) == nid

136 if localNodeId (processNode proc) == nid

90 then nsend label msg

137 then nsend label msg

138+ else

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

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

141+ liftIO $ traceEvent (localEventBus node)

142+ (MxSentToName label us (wrapMessage msg))

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

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

144

93-- | Send a message

145-- | Send a message

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

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

95send them msg = do

147send them msg = do

96 proc <- ask

148 proc <- ask

97 let node = localNodeId (processNode proc)

149 let node = localNodeId (processNode proc)

98− destNode = (processNodeId them) in do

150+ destNode = (processNodeId them)

151+ us = (processId proc)

152+ msg' = wrapMessage msg in do

153+ -- see [note: tracing]

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

155+ (MxSent them us msg')

99− case destNode == node of

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

157+ then sendCtrlMsg Nothing $ LocalSend them msg'

158+ else liftIO $ sendMessage (processNode proc)

159+ (ProcessIdentifier (processId proc))

160+ (ProcessIdentifier them)

161+ NoImplicitReconnect

162+ msg

110

163

111-- | Send a message unreliably.

164-- | Send a message unreliably.

112--

165--

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

121usend them msg = do

174usend them msg = do

122 proc <- ask

175 proc <- ask

123 let there = processNodeId them

176 let there = processNodeId them

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

178+ let msg' = wrapMessage msg

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

124 if localNodeId (processNode proc) == there

180 if localNodeId (processNode proc) == there

125− then sendCtrlMsg Nothing $

181+ then sendCtrlMsg Nothing $ LocalSend them msg'

126− LocalSend them (unsafeCreateUnencodedMessage msg)

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

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

128 (createMessage msg)

183 (createMessage msg)

129

184

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

135 destNode = processNodeId (sendPortProcessId cid) in do

190 destNode = processNodeId (sendPortProcessId cid) in do

136 case destNode == node of

191 case destNode == node of

137 True -> unsafeSendChanLocal cid msg

192 True -> unsafeSendChanLocal cid msg

138− False -> do

193+ False ->

139 liftIO $ sendBinary (processNode proc)

194 liftIO $ sendBinary (processNode proc)

140 (ProcessIdentifier (processId proc))

195 (ProcessIdentifier (processId proc))

141 (SendPortIdentifier cid)

196 (SendPortIdentifier cid)

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

144 where

199 where

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

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

146 unsafeSendChanLocal spId msg' =

201 unsafeSendChanLocal spId msg' =

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

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

148

203

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

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

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

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