git.iko.soy/distributed-process
on 0514f3c help sign in
660d554
swap
3 ahead, 7 behind
5 files changed+235−51
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+142−14 src/Control/Distributed/Process/Internal/Primitives.hs+13−15 src/Control/Distributed/Process/Management/Internal/Types.hs+2 src/Control/Distributed/Process/UnsafePrimitives.hs+76−21
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+142−14
@@ -3,9 +3,16 @@ module Control.Distributed.Process.Tests.Mx (tests) where
33
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)
2836import Data.Maybe (isJust, 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,163 @@ 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 ()) -> 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+
316395tests :: TestTransport -> IO [Test]
317396tests TestTransport{..} = do
318397 node1 <- newLocalNode testTransport initRemoteTable
319398 node2 <- newLocalNode testTransport initRemoteTable
399+ let nid = localNodeId node2
320400 return [
321− testGroup "Mx Agents" [
322− testCase "Event Handling"
401+ testGroup "MxAgents" [
402+ testCase "EventHandling"
323403 (delayedAssertion
324404 "expected True, but events where not as expected"
325405 node1 True testAgentEventHandling)
326− , testCase "Inter-Agent Broadcast"
406+ , testCase "InterAgentBroadcast"
327407 (delayedAssertion
328408 "expected (), but no broadcast was received"
329409 node1 () testAgentBroadcast)
330− , testCase "Agent Mailbox Handling"
410+ , testCase "AgentMailboxHandling"
331411 (delayedAssertion
332412 "expected (Just ()), but no regular (mailbox) input was handled"
333413 node1 (Just ()) testAgentMailboxHandling)
334− , testCase "Agent Dual Input Handling"
414+ , testCase "AgentDualInputHandling"
335415 (delayedAssertion
336416 "expected sum = 15, but the result was Nothing"
337417 node1 (Just 15 :: Maybe Int) testAgentDualInput)
338− , testCase "Agent Input Prioritisation"
418+ , testCase "AgentInputPrioritisation"
339419 (delayedAssertion
340420 "expected [first, second, third, fourth, fifth], but result diverged"
341421 node1 (sort ["first", "second",
342422 "third", "fourth",
343423 "fifth"]) testAgentPrioritisation)
344424 ]
345− , testGroup "Mx Events" [
346− testCase "Name Registration Events"
425+ , testGroup "MxEvents" [
426+ testCase "NameRegistrationEvents"
347427 (delayedAssertion
348428 "expected registration events to map to the correct ProcessId"
349429 node1 () testMxRegEvents)
350− , testCase "Post Death Name UnRegistration Events"
430+ , testCase "PostDeathNameUnRegistrationEvents"
351431 (delayedAssertion
352432 "expected process deaths to result in unregistration events"
353433 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+ ]
354483 ]
355− ]
src/Control/Distributed/Process/Internal/Primitives.hs+13−15
@@ -486,15 +486,13 @@ forward msg them = do
486486 destNode = (processNodeId them) in do
487487 case destNode == nid of
488488 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)
489+ False ->
490+ liftIO $ do
491+ sendPayload (processNode proc) (ProcessIdentifier (processId proc)) (ProcessIdentifier them) NoImplicitReconnect (messageToPayload msg)
492+ -- We do not fire the trace event until after the sending is complete;
493+ -- In this remote case, 'sendMessage' can block in the networking stack.
494+ liftIO $ traceEvent (localEventBus node)
495+ (MxSent them us msg)
498496
499497-- | Forward a raw 'Message' to the given 'ProcessId'.
500498--
@@ -510,12 +508,12 @@ uforward msg them = do
510508 destNode = (processNodeId them) in do
511509 case destNode == nid of
512510 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)
511+ False -> do
512+ sendCtrlMsg (Just destNode) $ UnreliableSend (processLocalId them) msg
513+ -- We do not fire the trace event until after the sending is complete;
514+ -- In the remote case, 'sendCtrlMsg' can block in the networking stack.
515+ liftIO $ traceEvent (localEventBus node)
516+ (MxSent them us msg)
519517
520518-- | Wrap a 'Serializable' value in a 'Message'. Note that 'Message's are
521519-- '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 =
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+76−21
@@ -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,65 @@ 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 node = localNodeId (processNode proc)
108+ let msg' = wrapMessage msg
109+ -- see [note: tracing]
110+ liftIO $ traceEvent (localEventBus (processNode proc))
111+ (MxSentToName label us 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+--
84129
85130-- | Named send to a process in a remote registry (asynchronous)
86131nsendRemote :: Serializable a => NodeId -> String -> a -> Process ()
87132nsendRemote nid label msg = do
88133 proc <- ask
134+ let us = processId proc
135+ let node = processNode proc
89136 if localNodeId (processNode proc) == nid
90137 then nsend label msg
91− else sendCtrlMsg (Just nid) (NamedSend label (createMessage 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))
143+ sendCtrlMsg (Just nid) (NamedSend label (createMessage msg))
92144
93145-- | Send a message
94146send :: Serializable a => ProcessId -> a -> Process ()
95147send them msg = do
96148 proc <- ask
97149 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')
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')
156+ if destNode == node
157+ then sendCtrlMsg Nothing $ LocalSend them msg'
158+ else liftIO $ sendMessage (processNode proc)
159+ (ProcessIdentifier (processId proc))
160+ (ProcessIdentifier them)
161+ NoImplicitReconnect
162+ msg
110163
111164-- | Send a message unreliably.
112165--
@@ -121,9 +174,11 @@ usend :: Serializable a => ProcessId -> a -> Process ()
121174usend them msg = do
122175 proc <- ask
123176 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')
124180 if localNodeId (processNode proc) == there
125− then sendCtrlMsg Nothing $
126− LocalSend them (unsafeCreateUnencodedMessage msg)
181+ then sendCtrlMsg Nothing $ LocalSend them msg'
127182 else sendCtrlMsg (Just there) $ UnreliableSend (processLocalId them)
128183 (createMessage msg)
129184
@@ -135,7 +190,7 @@ sendChan (SendPort cid) msg = do
135190 destNode = processNodeId (sendPortProcessId cid) in do
136191 case destNode == node of
137192 True -> unsafeSendChanLocal cid msg
138− False -> do
193+ False ->
139194 liftIO $ sendBinary (processNode proc)
140195 (ProcessIdentifier (processId proc))
141196 (SendPortIdentifier cid)
@@ -144,7 +199,7 @@ sendChan (SendPort cid) msg = do
144199 where
145200 unsafeSendChanLocal :: (Serializable a) => SendPortId -> a -> Process ()
146201 unsafeSendChanLocal spId msg' =
147− sendCtrlMsg Nothing $ LocalPortSend spId (unsafeCreateUnencodedMessage msg')
202+ sendCtrlMsg Nothing $ LocalPortSend spId (wrapMessage msg')
148203
149204-- | Create an unencoded @Message@ for any @Serializable@ type.
150205wrapMessage :: Serializable a => a -> Message