master..03b2197 · distributed-process

git.iko.soy/distributed-process

10 files changed+182−364

unifiedsplit

distributed-process-tests/distributed-process-tests.cabal+1−1

@@ -6,7 +6,7 @@ description: Tests and test suite for Cloud Haskell libraries.

66license: BSD3

77license-file: LICENSE

88Author: Duncan Coutts, Nicolas Wu, Edsko de Vries

9−Maintainer: Tim Watson <watson.timothy@gmail.com>

9+Maintainer: Facundo Domínguez <facundo.dominguez@tweag.io>

1010copyright: Well-Typed LLP

1111category: Control, Cloud Haskell

1212build-type: Simple

distributed-process-tests/src/Control/Distributed/Process/Tests/CH.hs+32−110

@@ -44,13 +44,13 @@ 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, pause)

47+import Control.Distributed.Process.Tests.Internal.Utils (shouldBe)

4848import Control.Distributed.Process.Serializable (Serializable)

49−import Data.Maybe (isNothing, isJust)

49+

5050import Test.HUnit (Assertion, assertBool, assertFailure)

5151import Test.Framework (Test, testGroup)

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

53−import Control.Rematch hiding (match, isNothing, isJust)

53+import Control.Rematch hiding (match)

5454import Control.Rematch.Run (Match(..))

5555

5656newtype Ping = Ping ProcessId

@@ -87,9 +87,10 @@ 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

9394

9495-- | Quick and dirty synchronous version of whereisRemoteAsync

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

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

239240 pid <- getSelfPid

240241 replicateM_ numPings $ do

241242 send pingServer (Pong pid)

242− p <- expectTimeout 3000000

243+ p <- expect

243244 case p of

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

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

245+ Ping _ -> return ()

246+ _ -> die "Unexpected message"

246247

247248 putMVar clientDone ()

248249

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

559560 ss <- liftIO $ mapM readMVar vs

560561 case ss of

561562 [sa, sb, sc] ->

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

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

563564 [ -- a, b, c

564565 (sa, 'a')

565566 , (sb, 'b')

@@ -735,6 +736,7 @@ testRemoteRegistry :: TestTransport -> Assertion

735736testRemoteRegistry TestTransport{..} = do

736737 node1 <- newLocalNode testTransport initRemoteTable

737738 node2 <- newLocalNode testTransport initRemoteTable

739+ done <- newEmptyMVar

738740

739741 pingServer <- forkProcess node1 ping

740742 deadProcess <- forkProcess node1 (return ())

@@ -749,7 +751,6 @@ testRemoteRegistry TestTransport{..} = do

749751

750752 pid <- verifyWhereIsRemote nid1 "ping"

751753 liftIO $ assertBool "Expected pindServer to match pid" $ pingServer == pid

752−

753754 us <- getSelfPid

754755 nsendRemote nid1 "ping" (Pong us)

755756 receiveWait [

@@ -759,22 +760,22 @@ testRemoteRegistry TestTransport{..} = do

759760 -- test that if process was not registered Nothing is returned

760761 -- in owner field.

761762 registerRemoteAsync nid1 "dead" deadProcess

762− receiveWait [ matchIf (\(RegisterReply label' f mPid) -> "dead" == label')

763− (\(RegisterReply _ f mPid) -> return (not f && isNothing mPid))

764− ] >>= liftIO . assertBool "Expected False Nothing in RegisterReply"

765−

763+ receiveWait [

764+ matchIf (\(RegisterReply label' False Nothing) -> "dead" == label')

765+ (\(RegisterReply _ _ _) -> return ()) ]

766766 registerRemoteAsync nid1 "ping" deadProcess

767767 receiveWait [

768− matchIf (\(RegisterReply label' False mPid) ->

769− "ping" == label' && isJust mPid)

770− (\(RegisterReply _ f (Just pid'')) -> return (not f && pid'' == pingServer))

771− ] >>= liftIO . assertBool "Expected False and (Just alreadyRegisteredPid) in RegisterReply"

772−

768+ matchIf (\(RegisterReply label' False (Just pid'')) ->

769+ "ping" == label' && pid'' == pingServer)

770+ (\(RegisterReply _ _ _) -> return ()) ]

773771 unregisterRemoteAsync nid1 "dead"

774772 receiveWait [

775− matchIf (\(RegisterReply label' _ _) -> "dead" == label')

776− (\(RegisterReply _ f mPid) -> return (not f && isNothing mPid))

777− ] >>= liftIO . assertBool "Expected False and Nothing in RegisterReply"

773+ matchIf (\(RegisterReply label' False Nothing) ->

774+ "dead" == label' && pid == pingServer)

775+ (\(RegisterReply _ _ _) -> return ()) ]

776+ liftIO $ putMVar done ()

777+

778+ takeMVar done

778779

779780testRemoteRegistryRemoteProcess :: TestTransport -> Assertion

780781testRemoteRegistryRemoteProcess TestTransport{..} = do

@@ -1272,76 +1273,6 @@ testReceiveChanFeatures TestTransport{..} = do

12721273

12731274 takeMVar done

12741275

1275−testChanLifecycle :: TestTransport -> Assertion

1276−testChanLifecycle TestTransport{..} = let delay = 3000000 in do

1277− result <- newEmptyMVar

1278− tchMV <- newEmptyMVar

1279− localNode <- newLocalNode testTransport initRemoteTable

1280− runProcess localNode $ do

1281−

1282− pid <- spawnLocal $ do tCh <- newChan :: Process (SendPort (), ReceivePort ())

1283− liftIO $ putMVar tchMV tCh

1284− expect :: Process ()

1285− let (sp, _) = tCh

1286− sendChan sp ()

1287− expect :: Process ()

1288−

1289− mRefPid <- monitor pid

1290−

1291− cPid <- spawnLocal $ do

1292− (sp', rp) <- liftIO $ takeMVar tchMV

1293− -- say "picked up our test channel"

1294− send pid ()

1295− -- say "told pid to continue"

1296− res <- receiveChanTimeout delay rp

1297− case res of

1298− Nothing -> say "initial chan () missing!" >> (liftIO $ putMVar result False)

1299− Just () -> do mr <- monitor pid

1300− pause 10000

1301− -- say "sending pid a second () will cause it to exit"

1302− send pid ()

1303−

1304− -- say "make sure we see a DOWN notification for pid having stopped"

1305− receiveWait [ match (\(_ :: ProcessMonitorNotification) -> return ()) ]

1306−

1307− -- now that pid has died, the send port should be useless...

1308− liftIO $ putMVar tchMV (sp', rp)

1309−

1310− -- let's verify that we do not see the message from our

1311− -- parent process on the channel, once pid has died...

1312− recv <- receiveChanTimeout delay rp

1313− -- say $ "finished waiting for second (), writing result" ++ (show recv)

1314− liftIO $ putMVar result $ isNothing recv

1315−

1316− mRefCPid <- monitor cPid

1317−

1318− receiveWait

1319− [ matchIf (\(ProcessMonitorNotification r _ _) -> r == mRefPid)

1320− (\_ -> return ())

1321− ]

1322−

1323− -- say "seen first pid die..."

1324−

1325− (sendPort, _) <- liftIO $ takeMVar tchMV

1326− sendChan sendPort ()

1327− -- say "sent () after owning pid died"

1328−

1329− -- let cPid know we've written to the channel...

1330− send cPid ()

1331−

1332− receiveWait

1333− [ matchIf (\(ProcessMonitorNotification r _ _) -> r == mRefCPid)

1334− (\_ -> return ())

1335− ]

1336−

1337− -- say "seen both pids die now..."

1338−

1339− -- and wait on the result back in IO land...

1340− testRes <- takeMVar result

1341− -- runProcess localNode $ say "got result..."

1342− assertBool "Expected sending on the channel to fail, but received data!" testRes

1343−

1344−

13451276testKillLocal :: TestTransport -> Assertion

13461277testKillLocal TestTransport{..} = do

13471278 localNode <- newLocalNode testTransport initRemoteTable

@@ -1556,24 +1487,20 @@ testRegistryMonitoring :: TestTransport -> Assertion

15561487testRegistryMonitoring TestTransport{..} = do

15571488 node1 <- newLocalNode testTransport initRemoteTable

15581489 node2 <- newLocalNode testTransport initRemoteTable

1490+ waitH <- newEmptyMVar

15591491

15601492 let nid = localNodeId node2

15611493 pid <- forkProcess node1 $ do

1562− self <- getSelfPid

1563− runUntilRegistered nid self

1564− say $ (show self) ++ " registered as " ++ regName

1565− expect :: Process ()

1566− say $ (show self) ++ " exiting normally"

1494+ getSelfPid >>= runUntilRegistered nid

1495+ liftIO $ takeMVar waitH

15671496

15681497 runProcess node2 $ do

15691498 register regName pid

1570− say $ regName ++ " registered to " ++ show pid

15711499 res <- whereis regName

1572− send pid ()

1573− say $ " sent finish signal to " ++ show pid

15741500 us <- getSelfPid

1575− liftIO $ assertBool "expected (Just pid)" $ res == (Just pid)

1576−

1501+ liftIO $ do

1502+ putMVar waitH ()

1503+ assertBool "expected (Just pid)" $ res == (Just pid)

15771504

15781505 -- This delay isn't essential!

15791506 -- The test case passes perfectly fine without it (feel free to comment out

@@ -1598,9 +1525,7 @@ testRegistryMonitoring TestTransport{..} = do

15981525 regHere <- newEmptyMVar

15991526 runProcess node2 $ whereis regName >>= liftIO . putMVar regHere

16001527 res <- takeMVar regHere

1601− case res of

1602− Nothing -> return ()

1603− Just pid -> assertBool ("expected Nothing, but got " ++ show pid) False

1528+ assertBool "expected Nothing, but process still registered" (res == Nothing)

16041529

16051530 where

16061531 runUntilRegistered nid us = do

@@ -1612,13 +1537,11 @@ testRegistryMonitoring TestTransport{..} = do

16121537

16131538 delayUntilMaybeUnregistered nid p = do

16141539 whereisRemoteAsync nid regName

1615− res <- receiveTimeout 20000000 {- 20 sec delay -} [

1616− matchIf (\(WhereIsReply n p) -> n == regName && isNothing p)

1540+ receiveTimeout 20000000 {- 20 sec delay -} [

1541+ matchIf (\(WhereIsReply n p) -> n == regName && p == Nothing)

16171542 (const $ return ())

16181543 ]

1619− case res of

1620− Just () -> return ()

1621− Nothing -> delayUntilMaybeUnregistered nid p

1544+ return ()

16221545

16231546 regName = "testRegisterRemote"

16241547

@@ -1814,7 +1737,6 @@ tests testtrans = return [

18141737 , testCase "MatchMessageUnwrap" (testMatchMessageWithUnwrap testtrans)

18151738 , testCase "ReceiveChanTimeout" (testReceiveChanTimeout testtrans)

18161739 , testCase "ReceiveChanFeatures" (testReceiveChanFeatures testtrans)

1817− , testCase "ChanLifecycle" (testChanLifecycle testtrans)

18181740 , testCase "KillLocal" (testKillLocal testtrans)

18191741 , testCase "KillRemote" (testKillRemote testtrans)

18201742 , testCase "Die" (testDie testtrans)

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

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

9191import Test.HUnit.Base (assertBool)

9292

9393import GHC.Generics

94−import System.Timeout (timeout)

9594

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

9796type TestResult a = MVar a

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

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

235234tryRunProcess node p = do

236235 tid <- liftIO myThreadId

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

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

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

239237

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

241239tryForkProcess node p = do

distributed-process-tests/src/Control/Distributed/Process/Tests/Mx.hs+116−165

@@ -1,22 +1,17 @@

1−{-# LANGUAGE DeriveGeneric #-}

2−{-# LANGUAGE ParallelListComp #-}

3−

1+{-# LANGUAGE DeriveGeneric #-}

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

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

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

6+import Control.Exception (SomeException)

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

9−import Control.Distributed.Process.Internal.Types

10− ( ProcessExitException(..)

11− , unsafeCreateUnencodedMessage

12− )

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

139import Control.Distributed.Process.Node

1410import qualified Control.Distributed.Process.UnsafePrimitives as Unsafe

1511 ( send

1612 , nsend

1713 , nsendRemote

1814 , usend

19− , sendChan

2015 )

2116import Control.Distributed.Process.Management

2217 ( MxEvent(..)

@@ -37,8 +32,8 @@ import Control.Monad (void, unless)

3732import Control.Monad.Catch(finally, bracket, try)

3833import Control.Rematch (equalTo)

3934import Data.Binary

40−import Data.List (find, sort, intercalate)

41−import Data.Maybe (isJust, fromJust, isNothing, fromMaybe, catMaybes)

35+import Data.List (find, sort)

36+import Data.Maybe (isJust, fromJust, isNothing)

4237import Data.Typeable

4338import GHC.Generics hiding (from)

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

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

6661 where

6762 withMonitorRef p = bracket (monitor p) unmonitor

6863

69−testAgentBroadcast :: TestResult (Maybe ()) -> Process ()

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

7065testAgentBroadcast result = do

7166 (resultSP, resultRP) <- newChan :: Process (SendPort (), ReceivePort ())

7267

@@ -81,10 +76,10 @@ testAgentBroadcast result = do

8176 mxNotify ()

8277 -- Once the publisher has seen our message, it will broadcast the Publish

8378 -- and the consumer will see that and send the result to our typed channel.

84− stash result =<< receiveChanTimeout 10000000 resultRP

79+ stash result =<< receiveChan resultRP

8580

86− kill publisher "finished"

87− kill consumer "finished"

81+ kill publisher "finished" >> awaitExit publisher

82+ kill consumer "finished" >> awaitExit consumer

8883

8984testAgentDualInput :: TestResult (Maybe Int) -> Process ()

9085testAgentDualInput result = do

@@ -99,8 +94,6 @@ testAgentDualInput result = do

9994 else mxReady)

10095 ]

10196

102− mRef <- monitor s

103−

10497 mxNotify (1 :: Int)

10598 nsend "sum-agent" (3 :: Int)

10699 mxNotify (2 :: Int)

@@ -108,10 +101,7 @@ testAgentDualInput result = do

108101 mxNotify (5 :: Int)

109102

110103 stash result =<< receiveChanTimeout 10000000 rp

111− died <- receiveTimeout 10000000 [

112− matchIf (\(ProcessMonitorNotification r _ _) -> r == mRef) (const $ return True)

113− ]

114− died `shouldBe` equalTo (Just True)

104+ awaitExit s

115105

116106testAgentPrioritisation :: TestResult [String] -> Process ()

117107testAgentPrioritisation result = do

@@ -125,10 +115,10 @@ testAgentPrioritisation result = do

125115 (sp, rp) <- newChan

126116 s <- mxAgent (MxAgentId name) ["first"] [

127117 mxSink (\(s :: String) -> do

128− mxUpdateLocal (s:)

118+ mxUpdateLocal ((s:))

129119 st <- mxGetLocal

130120 case length st of

131− n | n == 5 -> do liftMX $ sendChan sp (sort st)

121+ n | n == 5 -> do liftMX $ sendChan sp st

132122 mxDeactivate "finished"

133123 _ -> mxReceive -- go to the mailbox

134124 )

@@ -139,7 +129,7 @@ testAgentPrioritisation result = do

139129 mxNotify "fourth"

140130 nsend name "fifth"

141131

142− stash result . fromMaybe [] =<< receiveChanTimeout 10000000 rp

132+ stash result . sort =<< receiveChan rp

143133 awaitExit s

144134

145135testAgentMailboxHandling :: TestResult (Maybe ()) -> Process ()

@@ -152,19 +142,12 @@ testAgentMailboxHandling result = do

152142 nsend "mailbox-agent" ()

153143

154144 stash result =<< receiveChanTimeout 1000000 rp

155− kill agent "finished"

145+ kill agent "finished" >> awaitExit agent

156146

157147testAgentEventHandling :: TestResult Bool -> Process ()

158148testAgentEventHandling result = do

159− us <- getSelfPid

160− -- because this test is a bit racy, let's ensure it can't run indefinitely

161− timer <- spawnLocal $ do

162− pause (1000000 * 60 * 5)

163− -- okay we've waited 5 mins, let's kill the test off if it's stuck...

164− stash result False

165− kill us "Test Timed Out"

166−

167149 let initState = [] :: [MxEvent]

150+

168151 (rc, rs) <- newChan

169152 agentPid <- mxAgent (MxAgentId "lifecycle-listener-agent") initState [

170153 (mxSink $ \ev -> do

@@ -197,7 +180,8 @@ testAgentEventHandling result = do

197180 -- TODO: yes, this is racy, but we're at the mercy of the scheduler here...

198181 faff 2000000 rs

199182

200− (sp, rp) <- newChan :: Process (SendPort (), ReceivePort ())

183+ _ <- monitor agentPid

184+ (sp, rp) <- newChan

201185

202186 pid <- spawnLocal $ expect >>= sendChan sp

203187

@@ -206,11 +190,9 @@ testAgentEventHandling result = do

206190 monitor pid

207191

208192 send pid ()

209− rct <- receiveChanTimeout 10000000 rp

210− unless (isJust rct) $ die "No response on channel"

193+ () <- receiveChan rp

211194

212− pmn <- receiveTimeout 2000000 [ match (\ProcessMonitorNotification{} -> return ()) ]

213− unless (isJust pmn) $ die "No monitor notification arrived"

195+ receiveWait [ match (\(ProcessMonitorNotification _ _ _) -> return ()) ]

214196

215197 -- TODO: yes, this is racy, but we're at the mercy of the scheduler here...

216198 faff 2000000 rs

@@ -219,16 +201,17 @@ testAgentEventHandling result = do

219201 mxNotify (MxSpawned pid, replyTo)

220202 mxNotify (MxProcessDied pid DiedNormal, replyTo)

221203

222− seenAlive <- receiveChanTimeout 10000000 reply

223− seenDead <- receiveChanTimeout 10000000 reply

204+ seenAlive <- receiveChan reply

205+ seenDead <- receiveChan reply

224206

225− stash result $ foldr (&&) True $ catMaybes [seenAlive, seenDead]

226− kill timer "test-complete"

227− kill agentPid "test-complete"

207+ stash result $ seenAlive && seenDead

208+ kill agentPid "test-complete" >> awaitExit agentPid

228209 where

229210 faff delay port = do

230211 res <- receiveChanTimeout delay port

231− unless (isNothing res) $ pause delay

212+ unless (isNothing res) $ do

213+ pause delay

214+ faff delay port

232215

233216testMxRegEvents :: TestResult () -> Process ()

234217testMxRegEvents result = do

@@ -275,6 +258,7 @@ testMxRegEvents result = do

275258 reg3 `shouldBe` equalTo (Just (label, p1))

276259

277260 mapM_ (flip kill $ "test-complete") [agent, p1, p2]

261+ awaitExit agent

278262

279263testMxRegMon :: LocalNode -> TestResult () -> Process ()

280264testMxRegMon remoteNode result = do

@@ -332,99 +316,64 @@ testMxRegMon remoteNode result = do

332316 evts `shouldContain` (Just (label1, p1))

333317 evts `shouldContain` (Just (label2, p1))

334318

335− kill agent "test-complete"

319+ kill agent "test-complete" >> awaitExit agent

336320

337321ensure :: Process () -> Process () -> Process ()

338322ensure = flip finally

339323

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

325+

340326testNSend :: (String -> () -> Process ())

341327 -> Maybe LocalNode

328+ -> TestResult Bool

342329 -> Process ()

343−testNSend op n = do

344− us <- getSelfPid

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

345331 let delay = 5000000

346− let label = "testMxSend" ++ (show us)

332+ let label = "testMxSend"

347333 let isValid = isValidLabel n label

348334

349− testMxSend n label $ \p1 sink -> do

350− register label p1

351− reg1 <- receiveChanTimeout delay sink

352− case reg1 of

353− Just (MxRegistered pd lb)

354− | pd == p1 && isValid lb -> return ()

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

335+ register label p1

336+ reg1 <- receiveChanTimeout delay sink

337+ case reg1 of

338+ Just (MxRegistered pd lb)

339+ | pd == p1 && isValid lb -> return ()

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

356341

357− op label ()

342+ op label ()

358343

359− sent <- receiveChanTimeout delay sink

360− case sent of

361− Just (MxSentToName lb by _)

362− | by == us && isValid lb -> return True

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

344+ us <- getSelfPid

345+ sent <- receiveChanTimeout delay sink

346+ case sent of

347+ Just (MxSentToName lb by _)

348+ | by == us && isValid lb -> return True

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

364350

365351 where

366− isValidLabel nd l1 l2

352+ isValidLabel n l1 l2

367353 | l2 == l2 = True

368− | isJust nd = l2 == l1 ++ "@" ++ show (localNodeId $ fromJust nd)

354+ | isJust n = l2 == l1 ++ "@" ++ show (localNodeId $ fromJust n)

369355 | otherwise = False

370356

371−testSend :: (ProcessId -> () -> Process ())

372− -> Maybe LocalNode

373− -> Process ()

374−testSend op n = do

375− us <- getSelfPid

376− let label = "testMxSend" ++ (show us)

377− testMxSend n label $ \p1 sink -> do

378− -- initiate a send

379− op p1 ()

380−

381− -- verify the management event

382− sent <- receiveChanTimeout 5000000 sink

383− case sent of

384− Just (MxSent pidTo pidFrom _)

385− | pidTo == p1 && pidFrom == us -> return True

386− _ -> return False

387−

388−testChan :: (SendPort () -> () -> Process ())

389− -> Maybe LocalNode

390− -> Process ()

391−testChan op n = testMxSend n "" $ \p1 sink -> do

392−

393− us <- getSelfPid

394− send p1 us

395−

396− cleared <- receiveChanTimeout 2000000 sink

397− case cleared of

398− Just (MxSent pidTo pidFrom _)

399− | pidTo == p1 && pidFrom == us -> return ()

400− _ -> die "Received uncleared Mx Event"

401−

402− chan <- receiveTimeout 5000000 [ match return ]

403− let ch' = fromMaybe (error "No reply chan received") chan

404−

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

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

405359 -- initiate a send

406− op ch' ()

360+ op p1 ()

407361

408362 -- verify the management event

363+ us <- getSelfPid

409364 sent <- receiveChanTimeout 5000000 sink

410365 case sent of

411− Just (MxSentToPort sId spId _)

412− | sId == us && spId == sendPortId ch' -> return True

413− _ -> return False

414−

415−type SendTest = ProcessId -> ReceivePort MxEvent -> Process Bool

366+ Just (MxSent pidTo pidFrom _)

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

368+ _ -> return False

416369

417−testMxSend :: Maybe LocalNode -> String -> SendTest -> Process ()

418−testMxSend mNode label test = do

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

371+testMxSend mNode result test = do

419372 us <- getSelfPid

420− (sp, rp) <- newChan

421373 (chan, sink) <- newChan

422374 agent <- mxAgent (MxAgentId $ agentLabel us) () [

423375 mxSink $ \ev -> do

424376 case ev of

425− m@(MxSentToPort _ cid _)

426− | cid /= sendPortId chan

427− && cid /= sendPortId sp -> liftMX $ sendChan chan m

428377 m@(MxSent _ fromPid _)

429378 | fromPid == us -> liftMX $ sendChan chan m

430379 m@(MxSentToName _ fromPid _)

@@ -435,39 +384,29 @@ testMxSend mNode label test = do

435384 mxReady

436385 ]

437386

387+ (sp, rp) <- newChan

438388 case mNode of

439389 Nothing -> void $ spawnLocal (proc sp)

440390 Just remoteNode -> void $ liftIO $ forkProcess remoteNode $ proc sp

441391

442− p1 <- receiveChanTimeout 2000000 rp

443− unless (isJust p1) $ die "Timed out waiting for ProcessId"

444− res <- try $ test (fromJust p1) sink

445−

446− kill agent "bye"

447− kill (fromJust p1) "bye"

448−

392+ p1 <- receiveChan rp

393+ res <- try (test p1 sink)

449394 case res of

450− Left (ProcessExitException _ m) -> (liftIO $ putStrLn $ "SomeException-" ++ show m) >> die m

451− Right tr -> tr `shouldBe` equalTo True

452−

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

396+ Right tr -> stash result tr

397+ kill agent "bye"

398+ kill p1 "bye"

453399

454400 where

401+ label = "testMxSend"

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

456− proc sp' = getSelfPid >>= sendChan sp' >> go Nothing

457−

458− go :: Maybe (ReceivePort ()) -> Process ()

459− go Nothing = receiveTimeout 5000000 [ match replyChannel ] >>= go

460− go c@(Just c') = receiveWait [ matchChan c' return ] >> go c

461−

462− replyChannel p' = do

463− (s, r) <- newChan

464− send p' s

465− return r

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

466404

467405tests :: TestTransport -> IO [Test]

468406tests TestTransport{..} = do

469407 node1 <- newLocalNode testTransport initRemoteTable

470408 node2 <- newLocalNode testTransport initRemoteTable

409+ let nid = localNodeId node2

471410 return [

472411 testGroup "MxAgents" [

473412 testCase "EventHandling"

@@ -477,7 +416,7 @@ tests TestTransport{..} = do

477416 , testCase "InterAgentBroadcast"

478417 (delayedAssertion

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

480− node1 (Just ()) testAgentBroadcast)

419+ node1 () testAgentBroadcast)

481420 , testCase "AgentMailboxHandling"

482421 (delayedAssertion

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

@@ -502,41 +441,53 @@ tests TestTransport{..} = do

502441 (delayedAssertion

503442 "expected process deaths to result in unregistration events"

504443 node1 () (testMxRegMon node2))

505− , testGroup "SendEvents" $ buildTestCases node1 node2

506− ]

507− ]

508− where

509− buildTestCases n1 n2 = let nid = localNodeId n2 in build n1 n2 [

510− ("NSend", [

511− ("nsend", testNSend nsend)

512− , ("Unsafe.nsend", testNSend Unsafe.nsend)

513− , ("nsendRemote", testNSend (nsendRemote nid))

514− , ("Unsafe.nsendRemote", testNSend (Unsafe.nsendRemote nid))

515− ])

516− , ("Send", [

517− ("send", testSend send)

518− , ("Unsafe.send", testSend Unsafe.send)

519− , ("usend", testSend usend)

520− , ("Unsafe.usend", testSend Unsafe.usend)

521− , ("sendChan", testChan sendChan)

522− , ("Unsafe.sendChan", testChan Unsafe.sendChan)

523− ])

524− , ("Forward", [

525− ("forward", testSend (\p m -> forward (unsafeCreateUnencodedMessage m) p))

526− , ("uforward", testSend (\p m -> uforward (unsafeCreateUnencodedMessage m) p))

527− ])

444+ , testGroup "SentEvents" [

445+ testGroup "RemoteTargets" [

446+ testCase "Unsafe.nsend"

447+ (delayedAssertion "expected mx events failed"

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

449+ , testCase "Unsafe.nsendRemote"

450+ (delayedAssertion "expected mx events failed"

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

452+ , testCase "Unsafe.send"

453+ (delayedAssertion "expected mx events failed"

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

455+ , testCase "Unsafe.usend"

456+ (delayedAssertion "expected mx events failed"

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

458+ , testCase "nsend"

459+ (delayedAssertion "expected mx events failed"

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

461+ , testCase "nsendRemote"

462+ (delayedAssertion "expected mx events failed"

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

464+ , testCase "send"

465+ (delayedAssertion "expected mx events failed"

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

467+ , testCase "usend"

468+ (delayedAssertion "expected mx events failed"

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

470+ ]

471+ , testGroup "LocalTargets" [

472+ testCase "Unsafe.nsend"

473+ (delayedAssertion "expected mx events failed"

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

475+ , testCase "Unsafe.send"

476+ (delayedAssertion "expected mx events failed"

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

478+ , testCase "Unsafe.usend"

479+ (delayedAssertion "expected mx events failed"

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

481+ , testCase "nsend"

482+ (delayedAssertion "expected mx events failed"

483+ node1 True (testNSend nsend Nothing))

484+ , testCase "send"

485+ (delayedAssertion "expected mx events failed"

486+ node1 True (testSend send Nothing))

487+ , testCase "usend"

488+ (delayedAssertion "expected mx events failed"

489+ node1 True (testSend usend Nothing))

528490 ]

529−

530− build :: LocalNode

531− -> LocalNode

532− -> [(String, [(String, (Maybe LocalNode -> Process ()))])]

533− -> [Test]

534− build n ln specs =

535− [ testGroup (intercalate "-" [groupName, caseSuffix]) [

536− testCase (intercalate "-" [caseName, caseSuffix])

537− (runProcess n (caseImpl caseNode))

538− | (caseName, caseImpl) <- groupCases

539− ]

540− | (groupName, groupCases) <- specs

541− , (caseSuffix, caseNode) <- [("RemotePid", Just ln), ("LocalPid", Nothing)]

491+ ]

492+ ]

542493 ]

distributed-process.cabal+1−1

@@ -6,7 +6,7 @@ License: BSD3

66License-File: LICENSE

77Copyright: Well-Typed LLP, Tweag I/O Limited

88Author: Duncan Coutts, Nicolas Wu, Edsko de Vries

9−Maintainer: Tim Watson <watson.timothy@gmail.com>

9+Maintainer: Facundo Domínguez <facundo.dominguez@tweag.io>

1010Stability: experimental

1111Homepage: http://haskell-distributed.github.com/

1212Bug-Reports: https://github.com/haskell-distributed/distributed-process/issues

src/Control/Distributed/Process/Internal/Primitives.hs+6−15

@@ -307,13 +307,7 @@ expect = receiveWait [match return]

307307-- Channels --

308308--------------------------------------------------------------------------------

309309

310−-- | Create a new typed channel, bound to the calling Process

311−--

312−-- Note that the channel is bound to the lifecycle of the process that evaluates

313−-- this function, such that when it dies/exits, the channel will no longer

314−-- function, but will remain accessible. Thus reading from the ReceivePort will

315−-- fail silently thereafter, blocking indefinitely (unless a timeout is used).

316−--

310+-- | Create a new typed channel

317311newChan :: Serializable a => Process (SendPort a, ReceivePort a)

318312newChan = do

319313 proc <- ask

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

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

342336sendChan (SendPort cid) msg = do

343337 proc <- ask

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

338+ let node = localNodeId (processNode proc)

339+ destNode = processNodeId (sendPortProcessId cid) in do

340+ case destNode == node of

350341 True -> sendChanLocal cid msg

351342 False -> do

352− liftIO $ sendBinary node

353− (ProcessIdentifier pid)

343+ liftIO $ sendBinary (processNode proc)

344+ (ProcessIdentifier (processId proc))

354345 (SendPortIdentifier cid)

355346 NoImplicitReconnect

356347 msg

src/Control/Distributed/Process/Management.hs+2−34

@@ -76,40 +76,6 @@

7676--

7777-- * Whether messages will be taken from the mailbox first, or the event bus.

7878--

79−-- Since the event bus uses STM broadcast channels to communicate with agents,

80−-- no message written to the bus successfully can be lost.

81−--

82−-- Agents can also receive messages via their mailboxes - these are subject to

83−-- the same guarantees as all inter-process message sending.

84−--

85−-- Messages dispatched on an STM broadcast channel (i.e., management event bus)

86−-- are guaranteed to be delivered with the same FIFO ordering guarantees that

87−-- exist between two communicating processes, such that communication from the

88−-- node controller's threads (i.e., MxEvent's) will never be re-ordered, but

89−-- messages dispatched to the event bus by other processes (including, but not

90−-- limited to agents) are only guaranteed to be ordered between one sender and

91−-- one receiver.

92−--

93−-- No guarantee exists for the ordering in which messages sent to an agent's

94−-- mailbox will be delivered, vs messages dispatched via the event bus.

95−--

96−-- Because of the above, there are no ordering guarantees for messages sent

97−-- between agents, or for processes to agents, except for those that apply to

98−-- messages sent between regular processes, since agents are

99−-- implemented as such.

100−--

101−-- The event bus is serial and single threaded. Anything that is published by

102−-- the node controller will be seen in FIFO order. There are no ordering

103−-- guarantees pertaining to entries published to the event bus by other

104−-- processes or agents.

105−--

106−-- It should not be possible to see, for example, an @MxReceived@ before the

107−-- corresponding @MxSent@ event, since the places where we issue the @MxSent@

108−-- write directly to the event bus (using STM) in the calling (green) thread,

109−-- before dispatching instructions to the node controller to perform the

110−-- necessary routing to deliver the message to a process (or registered name,

111−-- or typed channel) locally or remotely.

112−--

11379-- [Management Data API]

11480--

11581-- Both management agents and clients of the API have access to a variety of

@@ -178,6 +144,8 @@

178144-- > monitorNames = getSelfPid >>= nsend "name-monitor"

179145-- > monitorNames2 = getSelfPid >>= mxNotify

180146--

147+-- For some real-world examples, see the distributed-process-platform package.

148+--

181149-- [Performance, Stablity and Scalability]

182150--

183151-- /Management Agents/ offer numerous advantages over regular processes:

src/Control/Distributed/Process/Management/Internal/Types.hs−5

@@ -21,7 +21,6 @@ import Control.Concurrent.STM

2121import Control.Distributed.Process.Internal.Types

2222 ( Process

2323 , ProcessId

24− , SendPortId

2524 , Message

2625 , DiedReason

2726 , NodeId

@@ -60,12 +59,8 @@ data MxEvent =

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

6160 | MxSentToName String ProcessId Message

6261 -- ^ fired whenever a named send occurs

63− | MxSentToPort ProcessId SendPortId Message

64− -- ^ fired whenever a sendChan occurs

6562 | MxReceived ProcessId Message

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

6964 | MxConnected ConnectionId EndPointAddress

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

7166 | MxDisconnected ConnectionId EndPointAddress

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

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

492492data IncomingTarget =

493493 Uninit

494494 | ToProc ProcessId (Weak (CQueue Message))

495− | ToChan SendPortId TypedChannel

495+ | ToChan TypedChannel

496496 | ToNode

497497

498498data ConnectionState = ConnectionState {

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

547547 enqueue queue msg -- 'enqueue' is strict

548548 trace node (MxReceived pid msg)

549549 go st

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

550+ Just (_, ToChan (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 -> 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'

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)

561557 go st

562558 Just (_, ToNode) -> do

563559 let ctrlMsg = decode . BSL.fromChunks $ payload

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

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

585581 case mChannel of

586582 Just channel ->

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

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

588584 Nothing ->

589585 invalidRequest cid st $

590586 "incoming attempt to connect to unknown channel of"

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

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

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

10561052 ch <- deRefWeak 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

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

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

1055+ deliverChan (UnencodedMessage _ raw) chan' =

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

1061− trace n (MxReceivedPort p $ unsafeCreateUnencodedMessage raw)

1062− deliverChan _ _ (EncodedMessage _ _) _ =

1057+ deliverChan (EncodedMessage _ _) _ =

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

10641059 error "invalid local channel delivery"

10651060

src/Control/Distributed/Process/UnsafePrimitives.hs+13−15

@@ -178,22 +178,20 @@ usend them msg = do

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

179179sendChan (SendPort cid) msg = do

180180 proc <- ask

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

181+ let node = localNodeId (processNode proc)

182+ destNode = processNodeId (sendPortProcessId cid) in do

183+ case destNode == node of

184+ True -> unsafeSendChanLocal cid msg

185+ False ->

186+ liftIO $ sendBinary (processNode proc)

187+ (ProcessIdentifier (processId proc))

188+ (SendPortIdentifier cid)

189+ NoImplicitReconnect

190+ msg

194191 where

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

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

192+ unsafeSendChanLocal :: (Serializable a) => SendPortId -> a -> Process ()

193+ unsafeSendChanLocal spId msg' =

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

197195

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

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