git.iko.soy/distributed-process
on a5191ce help sign in
660d554
swap
7 ahead, 7 behind
7 files changed+307−107
since splitsnapshots
unifiedsplit
distributed-process-tests/src/Control/Distributed/Process/Tests/CH.hs+6−7 distributed-process-tests/src/Control/Distributed/Process/Tests/Internal/Utils.hs+2−1 distributed-process-tests/src/Control/Distributed/Process/Tests/Mx.hs+148−16 src/Control/Distributed/Process/Internal/Primitives.hs+47−40 src/Control/Distributed/Process/Management/Internal/Types.hs+7 src/Control/Distributed/Process/Node.hs+15−10 src/Control/Distributed/Process/UnsafePrimitives.hs+82−33
distributed-process-tests/src/Control/Distributed/Process/Tests/CH.hs+6−7
@@ -44,7 +44,7 @@ import Control.Distributed.Process.Internal.Types
4444import Control.Distributed.Process.Node
4545import Control.Distributed.Process.Debug
4646import Control.Distributed.Process.Management.Internal.Types
47−import Control.Distributed.Process.Tests.Internal.Utils (shouldBe)
47+import Control.Distributed.Process.Tests.Internal.Utils (shouldBe, pause)
4848import Control.Distributed.Process.Serializable (Serializable)
4949
5050import Test.HUnit (Assertion, assertBool, assertFailure)
@@ -87,10 +87,9 @@ verifyClient :: String -> MVar Bool -> IO ()
8787verifyClient s b = takeMVar b >>= assertBool s
8888
8989expectPing :: MVar Bool -> Process ()
90−expectPing mv = expect >>= liftIO . putMVar mv . checkPing
90+expectPing mv = expect >>= liftIO . putMVar mv . checkPing
9191 where
9292 checkPing (Ping _) = True
93− checkPing _ = False
9493
9594-- | Quick and dirty synchronous version of whereisRemoteAsync
9695whereisRemote :: NodeId -> String -> Process (Maybe ProcessId)
@@ -240,10 +239,10 @@ testPing TestTransport{..} = do
240239 pid <- getSelfPid
241240 replicateM_ numPings $ do
242241 send pingServer (Pong pid)
243− p <- expect
242+ p <- expectTimeout 3000000
244243 case p of
245− Ping _ -> return ()
246− _ -> die "Unexpected message"
244+ Just (Ping _) -> return ()
245+ Nothing -> die "Failed to receive Ping"
247246
248247 putMVar clientDone ()
249248
@@ -560,7 +559,7 @@ testMergeChannels TestTransport{..} = do
560559 ss <- liftIO $ mapM readMVar vs
561560 case ss of
562561 [sa, sb, sc] ->
563− mapM_ ((>> liftIO (threadDelay 10000)) . uncurry sendChan)
562+ mapM_ ((>> pause 10000) . uncurry sendChan)
564563 [ -- a, b, c
565564 (sa, 'a')
566565 , (sb, 'b')
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+148−16
@@ -1,11 +1,23 @@
1−{-# LANGUAGE DeriveGeneric #-}
1+{-# LANGUAGE DeriveGeneric #-}
2+{-# LANGUAGE ParallelListComp #-}
3+
24module Control.Distributed.Process.Tests.Mx (tests) where
35
46import Control.Distributed.Process.Tests.Internal.Utils
57import Network.Transport.Test (TestTransport(..))
6−
7−import Control.Distributed.Process
8+import Control.Exception (SomeException)
9+import Control.Distributed.Process hiding (bracket, finally, try)
10+import Control.Distributed.Process.Internal.Types
11+ ( ProcessExitException(..)
12+ , unsafeCreateUnencodedMessage
13+ )
814import Control.Distributed.Process.Node
15+import qualified Control.Distributed.Process.UnsafePrimitives as Unsafe
16+ ( send
17+ , nsend
18+ , nsendRemote
19+ , usend
20+ )
921import Control.Distributed.Process.Management
1022 ( MxEvent(..)
1123 , MxAgentId(..)
@@ -22,12 +34,13 @@ import Control.Distributed.Process.Management
2234 , mxBroadcast
2335 )
2436import Control.Monad (void, unless)
37+import Control.Monad.Catch(finally, bracket, try)
2538import Control.Rematch (equalTo)
2639import Data.Binary
27−import Data.List (find, sort)
28−import Data.Maybe (isJust, isNothing)
40+import Data.List (find, sort, intercalate)
41+import Data.Maybe (isJust, fromJust, isNothing)
2942import Data.Typeable
30−import GHC.Generics
43+import GHC.Generics hiding (from)
3144#if ! MIN_VERSION_base(4,6,0)
3245import Prelude hiding (catch, log)
3346#endif
@@ -51,7 +64,7 @@ awaitExit pid =
5164 (\_ -> return ())
5265 ]
5366 where
54− withMonitorRef pid = bracket (monitor pid) unmonitor
67+ withMonitorRef p = bracket (monitor p) unmonitor
5568
5669testAgentBroadcast :: TestResult () -> Process ()
5770testAgentBroadcast result = do
@@ -313,43 +326,162 @@ testMxRegMon remoteNode result = do
313326ensure :: Process () -> Process () -> Process ()
314327ensure = flip finally
315328
329+testNSend :: (String -> () -> Process ())
330+ -> Maybe LocalNode
331+ -> TestResult Bool
332+ -> Process ()
333+testNSend op n r = testMxSend n r $ \p1 sink -> do
334+ let delay = 5000000
335+ let label = "testMxSend"
336+ let isValid = isValidLabel n label
337+
338+ register label p1
339+ reg1 <- receiveChanTimeout delay sink
340+ case reg1 of
341+ Just (MxRegistered pd lb)
342+ | pd == p1 && isValid lb -> return ()
343+ _ -> die $ "Reg-Failed: " ++ show reg1
344+
345+ op label ()
346+
347+ us <- getSelfPid
348+ sent <- receiveChanTimeout delay sink
349+ case sent of
350+ Just (MxSentToName lb by _)
351+ | by == us && isValid lb -> return True
352+ _ -> die $ "Send-Failed: " ++ show sent
353+
354+ where
355+ isValidLabel n l1 l2
356+ | l2 == l2 = True
357+ | isJust n = l2 == l1 ++ "@" ++ show (localNodeId $ fromJust n)
358+ | otherwise = False
359+
360+testSend :: (ProcessId -> () -> Process ())
361+ -> Maybe LocalNode
362+ -> TestResult Bool
363+ -> Process ()
364+testSend op n r = testMxSend n r $ \p1 sink -> do
365+ -- initiate a send
366+ op p1 ()
367+
368+ -- verify the management event
369+ us <- getSelfPid
370+ sent <- receiveChanTimeout 5000000 sink
371+ case sent of
372+ Just (MxSent pidTo pidFrom _)
373+ | pidTo == p1 && pidFrom == us -> return True
374+ _ -> return False
375+
376+type SendTest = ProcessId -> ReceivePort MxEvent -> Process Bool
377+
378+testMxSend :: Maybe LocalNode -> TestResult Bool -> SendTest -> Process ()
379+testMxSend mNode result test = do
380+ us <- getSelfPid
381+ (chan, sink) <- newChan
382+ agent <- mxAgent (MxAgentId $ agentLabel us) () [
383+ mxSink $ \ev -> do
384+ case ev of
385+ m@(MxSent _ fromPid _)
386+ | fromPid == us -> liftMX $ sendChan chan m
387+ m@(MxSentToName _ fromPid _)
388+ | fromPid == us -> liftMX $ sendChan chan m
389+ m@(MxRegistered _ name)
390+ | name == label -> liftMX $ sendChan chan m
391+ _ -> return ()
392+ mxReady
393+ ]
394+
395+ (sp, rp) <- newChan
396+ case mNode of
397+ Nothing -> void $ spawnLocal (proc sp)
398+ Just remoteNode -> void $ liftIO $ forkProcess remoteNode $ proc sp
399+
400+ p1 <- receiveChan rp
401+ res <- try (test p1 sink)
402+ case res of
403+ Left (ProcessExitException _ m) -> (liftIO $ putStrLn $ "SomeException-" ++ show m) >> stash result False
404+ Right tr -> stash result tr
405+ kill agent "bye"
406+ kill p1 "bye"
407+
408+ where
409+ label = "testMxSend"
410+ agentLabel s = "mx-unsafe-check-agent-" ++ show s
411+ proc sp' = getSelfPid >>= sendChan sp' >> expect :: Process ()
412+
316413tests :: TestTransport -> IO [Test]
317414tests TestTransport{..} = do
318415 node1 <- newLocalNode testTransport initRemoteTable
319416 node2 <- newLocalNode testTransport initRemoteTable
320417 return [
321− testGroup "Mx Agents" [
322− testCase "Event Handling"
418+ testGroup "MxAgents" [
419+ testCase "EventHandling"
323420 (delayedAssertion
324421 "expected True, but events where not as expected"
325422 node1 True testAgentEventHandling)
326− , testCase "Inter-Agent Broadcast"
423+ , testCase "InterAgentBroadcast"
327424 (delayedAssertion
328425 "expected (), but no broadcast was received"
329426 node1 () testAgentBroadcast)
330− , testCase "Agent Mailbox Handling"
427+ , testCase "AgentMailboxHandling"
331428 (delayedAssertion
332429 "expected (Just ()), but no regular (mailbox) input was handled"
333430 node1 (Just ()) testAgentMailboxHandling)
334− , testCase "Agent Dual Input Handling"
431+ , testCase "AgentDualInputHandling"
335432 (delayedAssertion
336433 "expected sum = 15, but the result was Nothing"
337434 node1 (Just 15 :: Maybe Int) testAgentDualInput)
338− , testCase "Agent Input Prioritisation"
435+ , testCase "AgentInputPrioritisation"
339436 (delayedAssertion
340437 "expected [first, second, third, fourth, fifth], but result diverged"
341438 node1 (sort ["first", "second",
342439 "third", "fourth",
343440 "fifth"]) testAgentPrioritisation)
344441 ]
345− , testGroup "Mx Events" [
346− testCase "Name Registration Events"
442+ , testGroup "MxEvents" [
443+ testCase "NameRegistrationEvents"
347444 (delayedAssertion
348445 "expected registration events to map to the correct ProcessId"
349446 node1 () testMxRegEvents)
350− , testCase "Post Death Name UnRegistration Events"
447+ , testCase "PostDeathNameUnRegistrationEvents"
351448 (delayedAssertion
352449 "expected process deaths to result in unregistration events"
353450 node1 () (testMxRegMon node2))
451+ , testGroup "SendEvents" $ buildTestCases node1 node2
354452 ]
355453 ]
454+ where
455+ buildTestCases n1 n2 =
456+ let nid = localNodeId n2 in build n2 [
457+ ("NSend", [
458+ ("Unsafe.nsend", testNSend Unsafe.nsend)
459+ , ("Unsafe.nsendRemote", testNSend (Unsafe.nsendRemote nid))
460+ , ("nsend", testNSend nsend)
461+ , ("nsendRemote", testNSend (nsendRemote nid))
462+ ])
463+ , ("Send", [
464+ ("Unsafe.send", testSend Unsafe.send)
465+ , ("Unsafe.usend", testSend Unsafe.usend)
466+ , ("send", testSend send)
467+ , ("usend", testSend usend)
468+ ])
469+ , ("Forward", [
470+ ("forward", testSend (\p m -> forward (unsafeCreateUnencodedMessage m) p))
471+ , ("uforward", testSend (\p m -> uforward (unsafeCreateUnencodedMessage m) p))
472+ ])
473+ ]
474+
475+ build :: LocalNode
476+ -> [(String, [(String, (Maybe LocalNode -> TestResult Bool -> Process ()))])]
477+ -> [Test]
478+ build ln specs =
479+ [ testGroup (intercalate "-" [groupName, caseSuffix]) [
480+ testCase (intercalate "-" [caseName, caseSuffix])
481+ (delayedAssertion "expected mx events failed"
482+ ln True (caseImpl caseNode))
483+ | (caseName, caseImpl) <- groupCases
484+ ]
485+ | (groupName, groupCases) <- specs
486+ , (caseSuffix, caseNode) <- [("RemotePid", Just ln), ("LocalPid", Nothing)]
487+ ]
src/Control/Distributed/Process/Internal/Primitives.hs+47−40
@@ -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)
@@ -334,13 +335,16 @@ newChan = do
334335sendChan :: Serializable a => SendPort a -> a -> Process ()
335336sendChan (SendPort cid) msg = do
336337 proc <- ask
337− let node = localNodeId (processNode proc)
338− destNode = processNodeId (sendPortProcessId cid) in do
339− case destNode == node of
338+ let node = processNode proc
339+ pid = processId proc
340+ us = localNodeId node
341+ them = processNodeId (sendPortProcessId cid) in do
342+ liftIO $ traceEvent (localEventBus node) (MxSentToPort pid cid $ wrapMessage msg)
343+ case them == us of
340344 True -> sendChanLocal cid msg
341345 False -> do
342− liftIO $ sendBinary (processNode proc)
343− (ProcessIdentifier (processId proc))
346+ liftIO $ sendBinary node
347+ (ProcessIdentifier pid)
344348 (SendPortIdentifier cid)
345349 NoImplicitReconnect
346350 msg
@@ -484,17 +488,14 @@ forward msg them = do
484488 us = processId proc
485489 nid = localNodeId node
486490 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)
491+ liftIO $ traceEvent (localEventBus node) (MxSent them us msg)
492+ if destNode == nid
493+ then sendCtrlMsg Nothing (LocalSend them msg)
494+ else liftIO $ sendPayload (processNode proc)
495+ (ProcessIdentifier (processId proc))
496+ (ProcessIdentifier them)
497+ NoImplicitReconnect
498+ (messageToPayload msg)
498499
499500-- | Forward a raw 'Message' to the given 'ProcessId'.
500501--
@@ -508,14 +509,10 @@ uforward msg them = do
508509 us = processId proc
509510 nid = localNodeId node
510511 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)
512+ liftIO $ traceEvent (localEventBus node) (MxSent them us msg)
513+ if destNode == nid
514+ then sendCtrlMsg Nothing (LocalSend them msg)
515+ else sendCtrlMsg (Just destNode) $ UnreliableSend (processLocalId them) msg
519516
520517-- | Wrap a 'Serializable' value in a 'Message'. Note that 'Message's are
521518-- 'Serializable' - like the datum they contain - but also note, deserialising
@@ -1224,8 +1221,12 @@ whereisRemoteAsync nid label = do
12241221
12251222-- | Named send to a process in the local registry (asynchronous)
12261223nsend :: Serializable a => String -> a -> Process ()
1227−nsend label msg =
1228− sendCtrlMsg Nothing (NamedSend label (createUnencodedMessage msg))
1224+nsend label msg = do
1225+ proc <- ask
1226+ let msg' = createUnencodedMessage msg
1227+ liftIO $ traceEvent (localEventBus (processNode proc))
1228+ (MxSentToName label (processId proc) msg')
1229+ sendCtrlMsg Nothing (NamedSend label msg')
12291230
12301231-- | Named send to a process in the local registry (asynchronous).
12311232-- This function makes /no/ attempt to serialize and (in the case when the
@@ -1237,9 +1238,15 @@ unsafeNSend = Unsafe.nsend
12371238-- | Named send to a process in a remote registry (asynchronous)
12381239nsendRemote :: Serializable a => NodeId -> String -> a -> Process ()
12391240nsendRemote nid label msg = do
1240− here <- getSelfNode
1241− if here == nid then nsend label msg
1242− else sendCtrlMsg (Just nid) (NamedSend label (createMessage msg))
1241+ proc <- ask
1242+ let us = processId proc
1243+ let node = processNode proc
1244+ if localNodeId node == nid
1245+ then nsend label msg
1246+ else let lbl = label ++ "@" ++ show nid in do
1247+ liftIO $ traceEvent (localEventBus node)
1248+ (MxSentToName lbl us (wrapMessage msg))
1249+ sendCtrlMsg (Just nid) (NamedSend label (createMessage msg))
12431250
12441251-- | Named send to a process in a remote registry (asynchronous)
12451252-- This function makes /no/ attempt to serialize and (in the case when the
src/Control/Distributed/Process/Management/Internal/Types.hs+7
@@ -21,6 +21,7 @@ import Control.Concurrent.STM
2121import Control.Distributed.Process.Internal.Types
2222 ( Process
2323 , ProcessId
24+ , SendPortId
2425 , Message
2526 , DiedReason
2627 , NodeId
@@ -57,8 +58,14 @@ data MxEvent =
5758 -- ^ fired whenever a node /dies/ (i.e., the connection is broken/disconnected)
5859 | MxSent ProcessId ProcessId Message
5960 -- ^ fired whenever a message is sent from a local process
61+ | MxSentToName String ProcessId Message
62+ -- ^ fired whenever a named send occurs
63+ | MxSentToPort ProcessId SendPortId Message
64+ -- ^ fired whenever a sendChan occurs
6065 | MxReceived ProcessId Message
6166 -- ^ fired whenever a message is received by a local process
67+ | MxReceivedPort SendPortId Message
68+ -- ^ fired whenever a message is received via a typed channel
6269 | MxConnected ConnectionId EndPointAddress
6370 -- ^ fired when a network-transport connection is first established
6471 | MxDisconnected ConnectionId EndPointAddress
src/Control/Distributed/Process/Node.hs+15−10
@@ -492,7 +492,7 @@ type IncomingConnection = (NT.EndPointAddress, IncomingTarget)
492492data IncomingTarget =
493493 Uninit
494494 | ToProc ProcessId (Weak (CQueue Message))
495− | ToChan TypedChannel
495+ | ToChan SendPortId TypedChannel
496496 | ToNode
497497
498498data ConnectionState = ConnectionState {
@@ -547,13 +547,17 @@ handleIncomingMessages node = go initConnectionState
547547 enqueue queue msg -- 'enqueue' is strict
548548 trace node (MxReceived pid msg)
549549 go st
550− Just (_, ToChan (TypedChannel chan')) -> do
550+ Just (_, ToChan chId (TypedChannel chan')) -> do
551551 mChan <- deRefWeak chan'
552552 -- If mChan is Nothing, the process has given up the read end of
553553 -- the channel and we simply ignore the incoming message
554− forM_ mChan $ \chan -> atomically $
555− -- We make sure the message is fully decoded when it is enqueued
556− writeTQueue chan $! decode (BSL.fromChunks payload)
554+ forM_ mChan $ \chan -> do
555+ msg' <- atomically $ do
556+ msg <- return $! decode (BSL.fromChunks payload)
557+ -- We make sure the message is fully decoded when it is enqueued
558+ writeTQueue chan msg
559+ return msg
560+ trace node $ MxReceivedPort chId $ unsafeCreateUnencodedMessage msg'
557561 go st
558562 Just (_, ToNode) -> do
559563 let ctrlMsg = decode . BSL.fromChunks $ payload
@@ -580,7 +584,7 @@ handleIncomingMessages node = go initConnectionState
580584 mChannel <- withMVar (processState proc) $ return . (^. typedChannelWithId lcid)
581585 case mChannel of
582586 Just channel ->
583− go (incomingAt cid ^= Just (src, ToChan channel) $ st)
587+ go (incomingAt cid ^= Just (src, ToChan chId channel) $ st)
584588 Nothing ->
585589 invalidRequest cid st $
586590 "incoming attempt to connect to unknown channel of"
@@ -1050,11 +1054,12 @@ ncEffectLocalPortSend from msg = do
10501054 -- If ch is Nothing, the process has given up the read end of
10511055 -- the channel and we simply ignore the incoming message - this
10521056 ch <- deRefWeak chan'
1053− forM_ ch $ \chan -> deliverChan msg chan
1054− where deliverChan :: forall a . Message -> TQueue a -> IO ()
1055− deliverChan (UnencodedMessage _ raw) chan' =
1057+ forM_ ch $ \chan -> deliverChan node from msg chan
1058+ where deliverChan :: forall a . LocalNode -> SendPortId -> Message -> TQueue a -> IO ()
1059+ deliverChan n p (UnencodedMessage _ raw) chan' = do
10561060 atomically $ writeTQueue chan' ((unsafeCoerce raw) :: a)
1057− deliverChan (EncodedMessage _ _) _ =
1061+ trace n (MxReceivedPort p $ unsafeCreateUnencodedMessage raw)
1062+ deliverChan _ _ (EncodedMessage _ _) _ =
10581063 -- this will not happen unless someone screws with Primitives.hs
10591064 error "invalid local channel delivery"
10601065
src/Control/Distributed/Process/UnsafePrimitives.hs+82−33
@@ -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,30 +157,43 @@ 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
133180 proc <- ask
134− let node = localNodeId (processNode proc)
135− destNode = processNodeId (sendPortProcessId cid) in do
136− case destNode == node of
137− True -> unsafeSendChanLocal cid msg
138− False -> do
139− liftIO $ sendBinary (processNode proc)
140− (ProcessIdentifier (processId proc))
141− (SendPortIdentifier cid)
142− NoImplicitReconnect
143− msg
181+ let node = processNode proc
182+ pid = processId proc
183+ us = localNodeId node
184+ them = processNodeId (sendPortProcessId cid)
185+ msg' = wrapMessage msg in do
186+ liftIO $ traceEvent (localEventBus node) (MxSentToPort pid cid msg')
187+ if them == us
188+ then unsafeSendChanLocal cid msg' -- NB: we wrap to P.Message !!!
189+ else liftIO $ sendBinary node
190+ (ProcessIdentifier pid)
191+ (SendPortIdentifier cid)
192+ NoImplicitReconnect
193+ msg
144194 where
145− unsafeSendChanLocal :: (Serializable a) => SendPortId -> a -> Process ()
146− unsafeSendChanLocal spId msg' =
147− sendCtrlMsg Nothing $ LocalPortSend spId (unsafeCreateUnencodedMessage msg')
195+ unsafeSendChanLocal :: SendPortId -> Message -> Process ()
196+ unsafeSendChanLocal p m = sendCtrlMsg Nothing $ LocalPortSend p m
148197
149198-- | Create an unencoded @Message@ for any @Serializable@ type.
150199wrapMessage :: Serializable a => a -> Message