git.iko.soy/distributed-process
on 7cb5ca7 help sign in
660d554
swap
12 ahead, 3 behind
8 files changed+355−109
since splitsnapshots
unifiedsplit
.travis.yml+1−1 distributed-process-tests/src/Control/Distributed/Process/Tests/CH.hs+5−6 distributed-process-tests/src/Control/Distributed/Process/Tests/Internal/Utils.hs+5−2 distributed-process-tests/src/Control/Distributed/Process/Tests/Mx.hs+193−17 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
.travis.yml+1−1
@@ -1,6 +1,6 @@
11language: c
22
3−sudo: false
3+sudo: required
44
55matrix:
66 include:
distributed-process-tests/src/Control/Distributed/Process/Tests/CH.hs+5−6
@@ -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+5−2
@@ -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
@@ -90,6 +91,7 @@ import Test.HUnit (Assertion, assertFailure)
9091import Test.HUnit.Base (assertBool)
9192
9293import GHC.Generics
94+import System.Timeout (timeout)
9395
9496-- | A mutable cell containing a test result.
9597type TestResult a = MVar a
@@ -232,7 +234,8 @@ testProcessReport pid = do
232234tryRunProcess :: LocalNode -> Process () -> IO ()
233235tryRunProcess node p = do
234236 tid <- liftIO myThreadId
235− runProcess node $ catch p (\e -> liftIO $ throwTo tid (e::SomeException))
237+ void $ timeout (1000000 * 60 * 5 :: Int) $
238+ runProcess node $ catch p (\e -> liftIO $ throwTo tid (e::SomeException))
236239
237240tryForkProcess :: LocalNode -> Process () -> IO ProcessId
238241tryForkProcess node p = do
distributed-process-tests/src/Control/Distributed/Process/Tests/Mx.hs+193−17
@@ -1,11 +1,24 @@
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+ , sendChan
21+ )
922import Control.Distributed.Process.Management
1023 ( MxEvent(..)
1124 , MxAgentId(..)
@@ -22,12 +35,13 @@ import Control.Distributed.Process.Management
2235 , mxBroadcast
2336 )
2437import Control.Monad (void, unless)
38+import Control.Monad.Catch(finally, bracket, try)
2539import Control.Rematch (equalTo)
2640import Data.Binary
27−import Data.List (find, sort)
28−import Data.Maybe (isJust, isNothing)
41+import Data.List (find, sort, intercalate)
42+import Data.Maybe (isJust, fromJust, isNothing, fromMaybe)
2943import Data.Typeable
30−import GHC.Generics
44+import GHC.Generics hiding (from)
3145#if ! MIN_VERSION_base(4,6,0)
3246import Prelude hiding (catch, log)
3347#endif
@@ -36,6 +50,7 @@ import Test.Framework
3650 ( Test
3751 , testGroup
3852 )
53+import Test.HUnit (Assertion)
3954import Test.Framework.Providers.HUnit (testCase)
4055
4156data Publish = Publish
@@ -51,7 +66,7 @@ awaitExit pid =
5166 (\_ -> return ())
5267 ]
5368 where
54− withMonitorRef pid = bracket (monitor pid) unmonitor
69+ withMonitorRef p = bracket (monitor p) unmonitor
5570
5671testAgentBroadcast :: TestResult () -> Process ()
5772testAgentBroadcast result = do
@@ -313,43 +328,204 @@ testMxRegMon remoteNode result = do
313328ensure :: Process () -> Process () -> Process ()
314329ensure = flip finally
315330
331+testNSend :: (String -> () -> Process ())
332+ -> Maybe LocalNode
333+ -> Process ()
334+testNSend op n = testMxSend n $ \p1 sink -> do
335+ let delay = 5000000
336+ let label = "testMxSend"
337+ let isValid = isValidLabel n label
338+
339+ register label p1
340+ reg1 <- receiveChanTimeout delay sink
341+ case reg1 of
342+ Just (MxRegistered pd lb)
343+ | pd == p1 && isValid lb -> return ()
344+ _ -> die $ "Reg-Failed: " ++ show reg1
345+
346+ op label ()
347+
348+ us <- getSelfPid
349+ sent <- receiveChanTimeout delay sink
350+ case sent of
351+ Just (MxSentToName lb by _)
352+ | by == us && isValid lb -> return True
353+ _ -> die $ "Send-Failed: " ++ show sent
354+
355+ where
356+ isValidLabel n l1 l2
357+ | l2 == l2 = True
358+ | isJust n = l2 == l1 ++ "@" ++ show (localNodeId $ fromJust n)
359+ | otherwise = False
360+
361+testSend :: (ProcessId -> () -> Process ())
362+ -> Maybe LocalNode
363+ -> Process ()
364+testSend op n = testMxSend n $ \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+testChan :: (SendPort () -> () -> Process ())
377+ -> Maybe LocalNode
378+ -> Process ()
379+testChan op n = testMxSend n $ \p1 sink -> do
380+
381+ us <- getSelfPid
382+ send p1 us
383+
384+ cleared <- receiveChan sink
385+ case cleared of
386+ (MxSent pidTo pidFrom _)
387+ | pidTo == p1 && pidFrom == us -> return ()
388+ _ -> die "Received uncleared Mx Event"
389+
390+ chan <- receiveTimeout 5000000 [ match return ]
391+ let ch' = fromMaybe (error "No reply chan received") chan
392+
393+ -- initiate a send
394+ op ch' ()
395+
396+ -- verify the management event
397+ sent <- receiveChanTimeout 5000000 sink
398+ case sent of
399+ Just (MxSentToPort sId spId _)
400+ | sId == us && spId == sendPortId ch' -> return True
401+ _ -> return False
402+
403+type SendTest = ProcessId -> ReceivePort MxEvent -> Process Bool
404+
405+testMxSend :: Maybe LocalNode -> SendTest -> Process ()
406+testMxSend mNode test = do
407+ us <- getSelfPid
408+ (sp, rp) <- newChan
409+ (chan, sink) <- newChan
410+ agent <- mxAgent (MxAgentId $ agentLabel us) () [
411+ mxSink $ \ev -> do
412+ case ev of
413+ m@(MxSentToPort _ cid _)
414+ | cid /= sendPortId chan
415+ && cid /= sendPortId sp -> liftMX $ sendChan chan m
416+ m@(MxSent _ fromPid _)
417+ | fromPid == us -> liftMX $ sendChan chan m
418+ m@(MxSentToName _ fromPid _)
419+ | fromPid == us -> liftMX $ sendChan chan m
420+ m@(MxRegistered _ name)
421+ | name == label -> liftMX $ sendChan chan m
422+ _ -> return ()
423+ mxReady
424+ ]
425+
426+ case mNode of
427+ Nothing -> void $ spawnLocal (proc sp)
428+ Just remoteNode -> void $ liftIO $ forkProcess remoteNode $ proc sp
429+
430+ p1 <- receiveChan rp
431+ res <- try (test p1 sink)
432+
433+ kill agent "bye"
434+ kill p1 "bye"
435+
436+ case res of
437+ Left (ProcessExitException _ m) -> (liftIO $ putStrLn $ "SomeException-" ++ show m) >> die m
438+ Right tr -> tr `shouldBe` equalTo True
439+
440+
441+ where
442+ label = "testMxSend"
443+ agentLabel s = "mx-unsafe-check-agent-" ++ show s
444+ proc sp' = getSelfPid >>= sendChan sp' >> go Nothing
445+
446+ go :: Maybe (ReceivePort ()) -> Process ()
447+ go Nothing = receiveTimeout 5000000 [ match replyChannel ] >>= go
448+ go c@(Just c') = receiveWait [ matchChan c' return ] >> go c
449+
450+ replyChannel p' = do
451+ (s, r) <- newChan
452+ send p' s
453+ return r
454+
316455tests :: TestTransport -> IO [Test]
317−tests TestTransport{..} = do
456+tests tt@TestTransport{..} = do
318457 node1 <- newLocalNode testTransport initRemoteTable
319458 node2 <- newLocalNode testTransport initRemoteTable
320459 return [
321− testGroup "Mx Agents" [
322− testCase "Event Handling"
460+ testGroup "MxAgents" [
461+ testCase "EventHandling"
323462 (delayedAssertion
324463 "expected True, but events where not as expected"
325464 node1 True testAgentEventHandling)
326− , testCase "Inter-Agent Broadcast"
465+ , testCase "InterAgentBroadcast"
327466 (delayedAssertion
328467 "expected (), but no broadcast was received"
329468 node1 () testAgentBroadcast)
330− , testCase "Agent Mailbox Handling"
469+ , testCase "AgentMailboxHandling"
331470 (delayedAssertion
332471 "expected (Just ()), but no regular (mailbox) input was handled"
333472 node1 (Just ()) testAgentMailboxHandling)
334− , testCase "Agent Dual Input Handling"
473+ , testCase "AgentDualInputHandling"
335474 (delayedAssertion
336475 "expected sum = 15, but the result was Nothing"
337476 node1 (Just 15 :: Maybe Int) testAgentDualInput)
338− , testCase "Agent Input Prioritisation"
477+ , testCase "AgentInputPrioritisation"
339478 (delayedAssertion
340479 "expected [first, second, third, fourth, fifth], but result diverged"
341480 node1 (sort ["first", "second",
342481 "third", "fourth",
343482 "fifth"]) testAgentPrioritisation)
344483 ]
345− , testGroup "Mx Events" [
346− testCase "Name Registration Events"
484+ , testGroup "MxEvents" [
485+ testCase "NameRegistrationEvents"
347486 (delayedAssertion
348487 "expected registration events to map to the correct ProcessId"
349488 node1 () testMxRegEvents)
350− , testCase "Post Death Name UnRegistration Events"
489+ , testCase "PostDeathNameUnRegistrationEvents"
351490 (delayedAssertion
352491 "expected process deaths to result in unregistration events"
353492 node1 () (testMxRegMon node2))
493+ , testGroup "SendEvents" $ buildTestCases tt node2
354494 ]
355495 ]
496+ where
497+ buildTestCases tt' n2 = let nid = localNodeId n2 in build tt' n2 [
498+ ("NSend", [
499+ ("nsend", testNSend nsend)
500+ , ("Unsafe.nsend", testNSend Unsafe.nsend)
501+ , ("nsendRemote", testNSend (nsendRemote nid))
502+ , ("Unsafe.nsendRemote", testNSend (Unsafe.nsendRemote nid))
503+ ])
504+ , ("Send", [
505+ ("send", testSend send)
506+ , ("Unsafe.send", testSend Unsafe.send)
507+ , ("usend", testSend usend)
508+ , ("Unsafe.usend", testSend Unsafe.usend)
509+ , ("sendChan", testChan sendChan)
510+ , ("Unsafe.sendChan", testChan Unsafe.sendChan)
511+ ])
512+ , ("Forward", [
513+ ("forward", testSend (\p m -> forward (unsafeCreateUnencodedMessage m) p))
514+ , ("uforward", testSend (\p m -> uforward (unsafeCreateUnencodedMessage m) p))
515+ ])
516+ ]
517+
518+ build :: TestTransport
519+ -> LocalNode
520+ -> [(String, [(String, (Maybe LocalNode -> Process ()))])]
521+ -> [Test]
522+ build TestTransport{..} ln specs =
523+ [ testGroup (intercalate "-" [groupName, caseSuffix]) [
524+ testCase (intercalate "-" [caseName, caseSuffix])
525+ (newLocalNode testTransport initRemoteTable >>= \n ->
526+ tryRunProcess n (caseImpl caseNode))
527+ | (caseName, caseImpl) <- groupCases
528+ ]
529+ | (groupName, groupCases) <- specs
530+ , (caseSuffix, caseNode) <- [("RemotePid", Just ln), ("LocalPid", Nothing)]
531+ ]
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)
@@ -340,13 +341,16 @@ newChan = do
340341sendChan :: Serializable a => SendPort a -> a -> Process ()
341342sendChan (SendPort cid) msg = do
342343 proc <- ask
343− let node = localNodeId (processNode proc)
344− destNode = processNodeId (sendPortProcessId cid) in do
345− case destNode == node of
344+ let node = processNode proc
345+ pid = processId proc
346+ us = localNodeId node
347+ them = processNodeId (sendPortProcessId cid) in do
348+ liftIO $ traceEvent (localEventBus node) (MxSentToPort pid cid $ wrapMessage msg)
349+ case them == us of
346350 True -> sendChanLocal cid msg
347351 False -> do
348− liftIO $ sendBinary (processNode proc)
349− (ProcessIdentifier (processId proc))
352+ liftIO $ sendBinary node
353+ (ProcessIdentifier pid)
350354 (SendPortIdentifier cid)
351355 NoImplicitReconnect
352356 msg
@@ -490,17 +494,14 @@ forward msg them = do
490494 us = processId proc
491495 nid = localNodeId node
492496 destNode = (processNodeId them) in do
493− case destNode == nid of
494− True -> sendCtrlMsg Nothing (LocalSend them msg)
495− False -> liftIO $ sendPayload (processNode proc)
496− (ProcessIdentifier (processId proc))
497− (ProcessIdentifier them)
498− NoImplicitReconnect
499− (messageToPayload msg)
500− -- We do not fire the trace event until after the sending is complete;
501− -- In the remote case, 'sendMessage' can block in the networking stack.
502− liftIO $ traceEvent (localEventBus node)
503− (MxSent them us msg)
497+ liftIO $ traceEvent (localEventBus node) (MxSent them us msg)
498+ if destNode == nid
499+ then sendCtrlMsg Nothing (LocalSend them msg)
500+ else liftIO $ sendPayload (processNode proc)
501+ (ProcessIdentifier (processId proc))
502+ (ProcessIdentifier them)
503+ NoImplicitReconnect
504+ (messageToPayload msg)
504505
505506-- | Forward a raw 'Message' to the given 'ProcessId'.
506507--
@@ -514,14 +515,10 @@ uforward msg them = do
514515 us = processId proc
515516 nid = localNodeId node
516517 destNode = (processNodeId them) in do
517− case destNode == nid of
518− True -> sendCtrlMsg Nothing (LocalSend them msg)
519− False -> sendCtrlMsg (Just destNode) $ UnreliableSend (processLocalId them)
520− msg
521− -- We do not fire the trace event until after the sending is complete;
522− -- In the remote case, 'sendCtrlMsg' can block in the networking stack.
523− liftIO $ traceEvent (localEventBus node)
524− (MxSent them us msg)
518+ liftIO $ traceEvent (localEventBus node) (MxSent them us msg)
519+ if destNode == nid
520+ then sendCtrlMsg Nothing (LocalSend them msg)
521+ else sendCtrlMsg (Just destNode) $ UnreliableSend (processLocalId them) msg
525522
526523-- | Wrap a 'Serializable' value in a 'Message'. Note that 'Message's are
527524-- 'Serializable' - like the datum they contain - but also note, deserialising
@@ -1230,8 +1227,12 @@ whereisRemoteAsync nid label = do
12301227
12311228-- | Named send to a process in the local registry (asynchronous)
12321229nsend :: Serializable a => String -> a -> Process ()
1233−nsend label msg =
1234− sendCtrlMsg Nothing (NamedSend label (createUnencodedMessage msg))
1230+nsend label msg = do
1231+ proc <- ask
1232+ let msg' = createUnencodedMessage msg
1233+ liftIO $ traceEvent (localEventBus (processNode proc))
1234+ (MxSentToName label (processId proc) msg')
1235+ sendCtrlMsg Nothing (NamedSend label msg')
12351236
12361237-- | Named send to a process in the local registry (asynchronous).
12371238-- This function makes /no/ attempt to serialize and (in the case when the
@@ -1243,9 +1244,15 @@ unsafeNSend = Unsafe.nsend
12431244-- | Named send to a process in a remote registry (asynchronous)
12441245nsendRemote :: Serializable a => NodeId -> String -> a -> Process ()
12451246nsendRemote nid label msg = do
1246− here <- getSelfNode
1247− if here == nid then nsend label msg
1248− else sendCtrlMsg (Just nid) (NamedSend label (createMessage msg))
1247+ proc <- ask
1248+ let us = processId proc
1249+ let node = processNode proc
1250+ if localNodeId node == nid
1251+ then nsend label msg
1252+ else let lbl = label ++ "@" ++ show nid in do
1253+ liftIO $ traceEvent (localEventBus node)
1254+ (MxSentToName lbl us (wrapMessage msg))
1255+ sendCtrlMsg (Just nid) (NamedSend label (createMessage msg))
12491256
12501257-- | Named send to a process in a remote registry (asynchronous)
12511258-- 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