master...7cb5ca7 · distributed-process

git.iko.soy/distributed-process

on 7cb5ca7 help sign in

.travis.yml+1−1

since splitsnapshots

unifiedsplit

@@ -1,6 +1,6 @@

1language: c

3−sudo: false

3+sudo: required

5matrix:

6 include:

distributed-process-tests/src/Control/Distributed/Process/Tests/CH.hs+5−6

@@ -87,10 +87,9 @@ verifyClient :: String -> MVar Bool -> IO ()

87verifyClient s b = takeMVar b >>= assertBool s

89expectPing :: MVar Bool -> Process ()

90−expectPing mv = expect >>= liftIO . putMVar mv . checkPing

90+expectPing mv = expect >>= liftIO . putMVar mv . checkPing

91 where

92 checkPing (Ping _) = True

93− checkPing _ = False

95-- | Quick and dirty synchronous version of whereisRemoteAsync

94-- | Quick and dirty synchronous version of whereisRemoteAsync

96whereisRemote :: NodeId -> String -> Process (Maybe ProcessId)

95whereisRemote :: NodeId -> String -> Process (Maybe ProcessId)

@@ -240,10 +239,10 @@ testPing TestTransport{..} = do

240 pid <- getSelfPid

239 pid <- getSelfPid

241 replicateM_ numPings $ do

240 replicateM_ numPings $ do

242 send pingServer (Pong pid)

241 send pingServer (Pong pid)

243− p <- expect

242+ p <- expectTimeout 3000000

244 case p of

243 case p of

245− Ping _ -> return ()

244+ Just (Ping _) -> return ()

246− _ -> die "Unexpected message"

245+ Nothing -> die "Failed to receive Ping"

247

246

248 putMVar clientDone ()

247 putMVar clientDone ()

249

248

@@ -560,7 +559,7 @@ testMergeChannels TestTransport{..} = do

560 ss <- liftIO $ mapM readMVar vs

559 ss <- liftIO $ mapM readMVar vs

561 case ss of

560 case ss of

562 [sa, sb, sc] ->

561 [sa, sb, sc] ->

563− mapM_ ((>> liftIO (threadDelay 10000)) . uncurry sendChan)

562+ mapM_ ((>> pause 10000) . uncurry sendChan)

564 [ -- a, b, c

563 [ -- a, b, c

565 (sa, 'a')

564 (sa, 'a')

566 , (sb, 'b')

565 , (sb, 'b')

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

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

@@ -90,6 +91,7 @@ import Test.HUnit (Assertion, assertFailure)

90import Test.HUnit.Base (assertBool)

91import Test.HUnit.Base (assertBool)

92import GHC.Generics

93import GHC.Generics

94+import System.Timeout (timeout)

94-- | A mutable cell containing a test result.

96-- | A mutable cell containing a test result.

95type TestResult a = MVar a

97type TestResult a = MVar a

@@ -232,7 +234,8 @@ testProcessReport pid = do

232tryRunProcess :: LocalNode -> Process () -> IO ()

234tryRunProcess :: LocalNode -> Process () -> IO ()

233tryRunProcess node p = do

235tryRunProcess node p = do

234 tid <- liftIO myThreadId

236 tid <- liftIO myThreadId

237+ void $ timeout (1000000 * 60 * 5 :: Int) $

235− runProcess node $ catch p (\e -> liftIO $ throwTo tid (e::SomeException))

238+ runProcess node $ catch p (\e -> liftIO $ throwTo tid (e::SomeException))

236

239

237tryForkProcess :: LocalNode -> Process () -> IO ProcessId

240tryForkProcess :: LocalNode -> Process () -> IO ProcessId

238tryForkProcess node p = do

241tryForkProcess 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 #-}

2module Control.Distributed.Process.Tests.Mx (tests) where

4module Control.Distributed.Process.Tests.Mx (tests) where

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

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

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

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

6−

8+import Control.Exception (SomeException)

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

7−import Control.Distributed.Process

10+import Control.Distributed.Process.Internal.Types

11+ ( ProcessExitException(..)

12+ , unsafeCreateUnencodedMessage

13+ )

8import Control.Distributed.Process.Node

14import Control.Distributed.Process.Node

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

16+ ( send

17+ , nsend

18+ , nsendRemote

19+ , usend

20+ , sendChan

21+ )

9import Control.Distributed.Process.Management

22import Control.Distributed.Process.Management

10 ( MxEvent(..)

23 ( MxEvent(..)

11 , MxAgentId(..)

24 , MxAgentId(..)

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

22 , mxBroadcast

35 , mxBroadcast

23 )

36 )

24import Control.Monad (void, unless)

37import Control.Monad (void, unless)

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

25import Control.Rematch (equalTo)

39import Control.Rematch (equalTo)

26import Data.Binary

40import Data.Binary

27−import Data.List (find, sort)

41+import Data.List (find, sort, intercalate)

28−import Data.Maybe (isJust, isNothing)

42+import Data.Maybe (isJust, fromJust, isNothing, fromMaybe)

29import Data.Typeable

43import Data.Typeable

30−import GHC.Generics

44+import GHC.Generics hiding (from)

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

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

32import Prelude hiding (catch, log)

46import Prelude hiding (catch, log)

33#endif

47#endif

@@ -36,6 +50,7 @@ import Test.Framework

36 ( Test

50 ( Test

37 , testGroup

51 , testGroup

38 )

52 )

53+import Test.HUnit (Assertion)

39import Test.Framework.Providers.HUnit (testCase)

54import Test.Framework.Providers.HUnit (testCase)

41data Publish = Publish

56data Publish = Publish

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

51 (\_ -> return ())

66 (\_ -> return ())

52 ]

67 ]

53 where

68 where

54− withMonitorRef pid = bracket (monitor pid) unmonitor

69+ withMonitorRef p = bracket (monitor p) unmonitor

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

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

57testAgentBroadcast result = do

72testAgentBroadcast result = do

@@ -313,43 +328,204 @@ testMxRegMon remoteNode result = do

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

328ensure :: Process () -> Process () -> Process ()

314ensure = flip finally

329ensure = flip finally

315

330

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+

316tests :: TestTransport -> IO [Test]

455tests :: TestTransport -> IO [Test]

317−tests TestTransport{..} = do

456+tests tt@TestTransport{..} = do

318 node1 <- newLocalNode testTransport initRemoteTable

457 node1 <- newLocalNode testTransport initRemoteTable

319 node2 <- newLocalNode testTransport initRemoteTable

458 node2 <- newLocalNode testTransport initRemoteTable

320 return [

459 return [

321− testGroup "Mx Agents" [

460+ testGroup "MxAgents" [

322− testCase "Event Handling"

461+ testCase "EventHandling"

323 (delayedAssertion

462 (delayedAssertion

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

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

325 node1 True testAgentEventHandling)

464 node1 True testAgentEventHandling)

326− , testCase "Inter-Agent Broadcast"

465+ , testCase "InterAgentBroadcast"

327 (delayedAssertion

466 (delayedAssertion

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

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

329 node1 () testAgentBroadcast)

468 node1 () testAgentBroadcast)

330− , testCase "Agent Mailbox Handling"

469+ , testCase "AgentMailboxHandling"

331 (delayedAssertion

470 (delayedAssertion

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

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

333 node1 (Just ()) testAgentMailboxHandling)

472 node1 (Just ()) testAgentMailboxHandling)

334− , testCase "Agent Dual Input Handling"

473+ , testCase "AgentDualInputHandling"

335 (delayedAssertion

474 (delayedAssertion

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

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

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

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

338− , testCase "Agent Input Prioritisation"

477+ , testCase "AgentInputPrioritisation"

339 (delayedAssertion

478 (delayedAssertion

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

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

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

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

342 "third", "fourth",

481 "third", "fourth",

343 "fifth"]) testAgentPrioritisation)

482 "fifth"]) testAgentPrioritisation)

344 ]

483 ]

345− , testGroup "Mx Events" [

484+ , testGroup "MxEvents" [

346− testCase "Name Registration Events"

485+ testCase "NameRegistrationEvents"

347 (delayedAssertion

486 (delayedAssertion

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

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

349 node1 () testMxRegEvents)

488 node1 () testMxRegEvents)

350− , testCase "Post Death Name UnRegistration Events"

489+ , testCase "PostDeathNameUnRegistrationEvents"

351 (delayedAssertion

490 (delayedAssertion

352 "expected process deaths to result in unregistration events"

491 "expected process deaths to result in unregistration events"

353 node1 () (testMxRegMon node2))

492 node1 () (testMxRegMon node2))

493+ , testGroup "SendEvents" $ buildTestCases tt node2

354 ]

494 ]

355 ]

495 ]

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

254 node = processNode proc

255 nodeId = localNodeId node

256 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.

266 liftIO $ traceEvent (localEventBus node)

257 liftIO $ traceEvent (localEventBus node)

267 (MxSent them us (createUnencodedMessage msg))

258 (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

268

266

269-- | /Unsafe/ variant of 'send'. This function makes /no/ attempt to serialize

267-- | /Unsafe/ variant of 'send'. This function makes /no/ attempt to serialize

270-- and (in the case when the destination process resides on the same local

268-- and (in the case when the destination process resides on the same local

@@ -284,9 +282,12 @@ unsafeSend = Unsafe.send

284--

282--

285usend :: Serializable a => ProcessId -> a -> Process ()

283usend :: Serializable a => ProcessId -> a -> Process ()

286usend them msg = do

284usend them msg = do

287− here <- getSelfNode

285+ proc <- ask

288 let there = processNodeId them

286 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

290 then sendLocal them msg

291 then sendLocal them msg

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

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

292 (createMessage msg)

293 (createMessage msg)

@@ -340,13 +341,16 @@ newChan = do

340sendChan :: Serializable a => SendPort a -> a -> Process ()

341sendChan :: Serializable a => SendPort a -> a -> Process ()

341sendChan (SendPort cid) msg = do

342sendChan (SendPort cid) msg = do

342 proc <- ask

343 proc <- ask

343− let node = localNodeId (processNode proc)

344+ let node = processNode proc

345+ pid = processId proc

346+ us = localNodeId node

344− destNode = processNodeId (sendPortProcessId cid) in do

347+ them = processNodeId (sendPortProcessId cid) in do

348+ liftIO $ traceEvent (localEventBus node) (MxSentToPort pid cid $ wrapMessage msg)

345− case destNode == node of

349+ case them == us of

346 True -> sendChanLocal cid msg

350 True -> sendChanLocal cid msg

347 False -> do

351 False -> do

348− liftIO $ sendBinary (processNode proc)

352+ liftIO $ sendBinary node

349− (ProcessIdentifier (processId proc))

353+ (ProcessIdentifier pid)

350 (SendPortIdentifier cid)

354 (SendPortIdentifier cid)

351 NoImplicitReconnect

355 NoImplicitReconnect

352 msg

356 msg

@@ -490,17 +494,14 @@ forward msg them = do

490 us = processId proc

494 us = processId proc

491 nid = localNodeId node

495 nid = localNodeId node

492 destNode = (processNodeId them) in do

496 destNode = (processNodeId them) in do

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

493− case destNode == nid of

498+ if destNode == nid

494− True -> sendCtrlMsg Nothing (LocalSend them msg)

499+ then sendCtrlMsg Nothing (LocalSend them msg)

495− False -> liftIO $ sendPayload (processNode proc)

500+ else liftIO $ sendPayload (processNode proc)

496− (ProcessIdentifier (processId proc))

501+ (ProcessIdentifier (processId proc))

497− (ProcessIdentifier them)

502+ (ProcessIdentifier them)

498− NoImplicitReconnect

503+ NoImplicitReconnect

499− (messageToPayload msg)

504+ (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)

504

505

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

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

506--

507--

@@ -514,14 +515,10 @@ uforward msg them = do

514 us = processId proc

515 us = processId proc

515 nid = localNodeId node

516 nid = localNodeId node

516 destNode = (processNodeId them) in do

517 destNode = (processNodeId them) in do

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

517− case destNode == nid of

519+ if destNode == nid

518− True -> sendCtrlMsg Nothing (LocalSend them msg)

520+ then sendCtrlMsg Nothing (LocalSend them msg)

519− False -> sendCtrlMsg (Just destNode) $ UnreliableSend (processLocalId them)

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

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)

525

522

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

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

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

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

@@ -1230,8 +1227,12 @@ whereisRemoteAsync nid label = do

1230

1227

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

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

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

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

1233−nsend label 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')

1234− sendCtrlMsg Nothing (NamedSend label (createUnencodedMessage msg))

1235+ sendCtrlMsg Nothing (NamedSend label msg')

1235

1236

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

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

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

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

@@ -1243,9 +1244,15 @@ unsafeNSend = Unsafe.nsend

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

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

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

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

1245nsendRemote nid label msg = do

1246nsendRemote nid label msg = do

1246− here <- getSelfNode

1247+ proc <- ask

1248+ let us = processId proc

1249+ let node = processNode proc

1250+ if localNodeId node == nid

1247− if here == nid then nsend label msg

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

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

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

1249

1256

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

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

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

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

21import Control.Distributed.Process.Internal.Types

22 ( Process

23 , ProcessId

24+ , SendPortId

24 , Message

25 , Message

25 , DiedReason

26 , DiedReason

26 , NodeId

27 , NodeId

@@ -57,8 +58,14 @@ data MxEvent =

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

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

58 | MxSent ProcessId ProcessId Message

59 | MxSent ProcessId ProcessId Message

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

60 -- ^ 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

60 | MxReceived ProcessId Message

65 | MxReceived ProcessId Message

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

66 -- ^ 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

62 | MxConnected ConnectionId EndPointAddress

69 | MxConnected ConnectionId EndPointAddress

63 -- ^ fired when a network-transport connection is first established

70 -- ^ fired when a network-transport connection is first established

64 | MxDisconnected ConnectionId EndPointAddress

71 | MxDisconnected ConnectionId EndPointAddress

src/Control/Distributed/Process/Node.hs+15−10

@@ -492,7 +492,7 @@ type IncomingConnection = (NT.EndPointAddress, IncomingTarget)

492data IncomingTarget =

493 Uninit

494 | ToProc ProcessId (Weak (CQueue Message))

495− | ToChan TypedChannel

495+ | ToChan SendPortId TypedChannel

496 | ToNode

497

498data ConnectionState = ConnectionState {

@@ -547,13 +547,17 @@ handleIncomingMessages node = go initConnectionState

547 enqueue queue msg -- 'enqueue' is strict

548 trace node (MxReceived pid msg)

549 go st

550− Just (_, ToChan (TypedChannel chan')) -> do

550+ Just (_, ToChan chId (TypedChannel chan')) -> do

551 mChan <- deRefWeak chan'

552 -- If mChan is Nothing, the process has given up the read end of

553 -- the channel and we simply ignore the incoming message

554− forM_ mChan $ \chan -> atomically $

554+ forM_ mChan $ \chan -> do

555+ msg' <- atomically $ do

556+ msg <- return $! decode (BSL.fromChunks payload)

555− -- We make sure the message is fully decoded when it is enqueued

557+ -- We make sure the message is fully decoded when it is enqueued

556− writeTQueue chan $! decode (BSL.fromChunks payload)

558+ writeTQueue chan msg

559+ return msg

560+ trace node $ MxReceivedPort chId $ unsafeCreateUnencodedMessage msg'

557 go st

561 go st

558 Just (_, ToNode) -> do

562 Just (_, ToNode) -> do

559 let ctrlMsg = decode . BSL.fromChunks $ payload

563 let ctrlMsg = decode . BSL.fromChunks $ payload

@@ -580,7 +584,7 @@ handleIncomingMessages node = go initConnectionState

580 mChannel <- withMVar (processState proc) $ return . (^. typedChannelWithId lcid)

584 mChannel <- withMVar (processState proc) $ return . (^. typedChannelWithId lcid)

581 case mChannel of

585 case mChannel of

582 Just channel ->

586 Just channel ->

583− go (incomingAt cid ^= Just (src, ToChan channel) $ st)

587+ go (incomingAt cid ^= Just (src, ToChan chId channel) $ st)

584 Nothing ->

588 Nothing ->

585 invalidRequest cid st $

589 invalidRequest cid st $

586 "incoming attempt to connect to unknown channel of"

590 "incoming attempt to connect to unknown channel of"

@@ -1050,11 +1054,12 @@ ncEffectLocalPortSend from msg = do

1050 -- If ch is Nothing, the process has given up the read end of

1054 -- If ch is Nothing, the process has given up the read end of

1051 -- the channel and we simply ignore the incoming message - this

1055 -- the channel and we simply ignore the incoming message - this

1052 ch <- deRefWeak chan'

1056 ch <- deRefWeak chan'

1053− forM_ ch $ \chan -> deliverChan msg chan

1057+ forM_ ch $ \chan -> deliverChan node from msg chan

1054− where deliverChan :: forall a . Message -> TQueue a -> IO ()

1058+ where deliverChan :: forall a . LocalNode -> SendPortId -> Message -> TQueue a -> IO ()

1055− deliverChan (UnencodedMessage _ raw) chan' =

1059+ deliverChan n p (UnencodedMessage _ raw) chan' = do

1056 atomically $ writeTQueue chan' ((unsafeCoerce raw) :: a)

1060 atomically $ writeTQueue chan' ((unsafeCoerce raw) :: a)

1061+ trace n (MxReceivedPort p $ unsafeCreateUnencodedMessage raw)

1057− deliverChan (EncodedMessage _ _) _ =

1062+ deliverChan _ _ (EncodedMessage _ _) _ =

1058 -- this will not happen unless someone screws with Primitives.hs

1063 -- this will not happen unless someone screws with Primitives.hs

1059 error "invalid local channel delivery"

1064 error "invalid local channel delivery"

1060

1065

src/Control/Distributed/Process/UnsafePrimitives.hs+82−33

@@ -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,48 @@ 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 msg' = wrapMessage msg

108+ -- see [note: tracing]

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

110+ (MxSentToName label us msg')

83− sendCtrlMsg Nothing (NamedSend label (unsafeCreateUnencodedMessage msg))

111+ sendCtrlMsg Nothing (NamedSend label msg')

112

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

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

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

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

87nsendRemote nid label msg = do

115nsendRemote nid label msg = do

88 proc <- ask

116 proc <- ask

117+ let us = processId proc

118+ let node = processNode proc

89− if localNodeId (processNode proc) == nid

119+ if localNodeId node == nid

90 then nsend label msg

120 then nsend label 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))

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

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

127

93-- | Send a message

128-- | Send a message

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

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

95send them msg = do

130send them msg = do

96 proc <- ask

131 proc <- ask

97 let node = localNodeId (processNode proc)

132 let node = localNodeId (processNode proc)

98− destNode = (processNodeId them) in do

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

99− case destNode == node of

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

140+ then sendCtrlMsg Nothing $ LocalSend them msg'

141+ else liftIO $ sendMessage (processNode proc)

142+ (ProcessIdentifier (processId proc))

143+ (ProcessIdentifier them)

144+ NoImplicitReconnect

145+ msg

110

146

111-- | Send a message unreliably.

147-- | Send a message unreliably.

112--

148--

@@ -121,30 +157,43 @@ usend :: Serializable a => ProcessId -> a -> Process ()

121usend them msg = do

157usend them msg = do

122 proc <- ask

158 proc <- ask

123 let there = processNodeId them

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

124 if localNodeId (processNode proc) == there

164 if localNodeId (processNode proc) == there

125− then sendCtrlMsg Nothing $

165+ then sendCtrlMsg Nothing $ LocalSend them msg'

126− LocalSend them (unsafeCreateUnencodedMessage msg)

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

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

128 (createMessage msg)

167 (createMessage msg)

129

168

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+

130-- | Send a message on a typed channel

177-- | Send a message on a typed channel

131sendChan :: Serializable a => SendPort a -> a -> Process ()

178sendChan :: Serializable a => SendPort a -> a -> Process ()

132sendChan (SendPort cid) msg = do

179sendChan (SendPort cid) msg = do

133 proc <- ask

180 proc <- ask

181+ let node = processNode proc

182+ pid = processId proc

134− let node = localNodeId (processNode proc)

183+ us = localNodeId node

135− destNode = processNodeId (sendPortProcessId cid) in do

184+ them = processNodeId (sendPortProcessId cid)

136− case destNode == node of

185+ msg' = wrapMessage msg in do

186+ liftIO $ traceEvent (localEventBus node) (MxSentToPort pid cid msg')

187+ if them == us

137− True -> unsafeSendChanLocal cid msg

188+ then unsafeSendChanLocal cid msg' -- NB: we wrap to P.Message !!!

138− False -> do

139− liftIO $ sendBinary (processNode proc)

189+ else liftIO $ sendBinary node

140− (ProcessIdentifier (processId proc))

190+ (ProcessIdentifier pid)

141− (SendPortIdentifier cid)

191+ (SendPortIdentifier cid)

142− NoImplicitReconnect

192+ NoImplicitReconnect

143− msg

193+ msg

144 where

194 where

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

195+ unsafeSendChanLocal :: SendPortId -> Message -> Process ()

146− unsafeSendChanLocal spId msg' =

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

196+ unsafeSendChanLocal p m = sendCtrlMsg Nothing $ LocalPortSend p m

148

197

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

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

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

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