master..f8533f0 · distributed-process

git.iko.soy/distributed-process

on f8533f0 help sign in

.travis.yml+6−11

since splitsnapshots

unifiedsplit

@@ -6,11 +6,11 @@ matrix:

6 include:

7 - env: ARGS="--stack-yaml stack-ghc-7.10.3.yaml"

8 addons: {apt: {packages: [libgmp-dev]}}

9− - env: ARGS="--stack-yaml stack-ghc-8.2.2.yaml"

10+ - env: ARGS="--stack-yaml stack-ghc-8.0.2.yaml"

10− addons: {apt: {packages: [libgmp-dev]}}

11− - env: ARGS="--stack-yaml stack-ghc-8.4.4.yaml"

12 addons: {apt: {packages: [libgmp-dev]}}

11 addons: {apt: {packages: [libgmp-dev]}}

13− - env: ARGS="--stack-yaml stack-nightly.yaml --resolver nightly"

12+

13+ - env: ARGS=

14 addons: {apt: {packages: [libgmp-dev]}}

16cache:

@@ -34,10 +34,5 @@ script:

34- stack ${ARGS} test $ARG='--plain -t "!Flaky"' ${TEST_PACKAGE}TestCHInTCP

35- stack ${ARGS} test $ARG='--plain -t "!SpawnReconnect"' ${TEST_PACKAGE}TestClosure

36- stack ${ARGS} test $ARG='--plain' ${TEST_PACKAGE}TestStats

37−- stack ${ARGS} test $ARG='--plain' ${TEST_PACKAGE}TestMxInMemory

37+- stack ${ARGS} test $ARG='--plain' ${TEST_PACKAGE}TestMx

38−- stack ${ARGS} test $ARG='--plain' ${TEST_PACKAGE}TestMxInTCP

38+- stack ${ARGS} test $ARG='--plain' ${TEST_PACKAGE}TestTracing

39−- stack ${ARGS} test $ARG='--plain' ${TEST_PACKAGE}TestTracingInMemory

40−

41−notifications:

42− slack:

43− secure: LbqbNjK1KiHNiQo/i/Za4vkMD9l9mhmN+PWYFBQkNvGFWffSRayRKFrTqt4znV2p7h4dY1XZFoC7T5RemBVQLq2ppZAUWxkITeu1OUlnmVLDfJLeVYGTR/fn90nP6Y5ITE7T3A07nKmWaRbKpIgFBAOjzgqaM0csscDx3z0WW18=

benchmarks/Channels.hs+5−6

@@ -4,7 +4,7 @@ import Control.Monad

4import Control.Applicative

5import Control.Distributed.Process

6import Control.Distributed.Process.Node

7−import Network.Transport.TCP (createTransport, defaultTCPAddr, defaultTCPParameters)

7+import Network.Transport.TCP (createTransport, defaultTCPParameters)

8import Data.Binary (encode, decode)

9import qualified Data.ByteString.Lazy as BSL

@@ -36,8 +36,7 @@ initialProcess "CLIENT" = do

36main :: IO ()

37main = do

38 [role, host, port] <- getArgs

39+ Right transport <- createTransport

39− trans <- createTransport (defaultTCPAddr host port) defaultTCPParameters

40+ host port (\sn -> (host, port)) defaultTCPParameters

40− case trans of

41− Right transport -> do node <- newLocalNode transport initRemoteTable

41+ node <- newLocalNode transport initRemoteTable

42− runProcess node $ initialProcess role

42+ runProcess node $ initialProcess role

43− Left other -> error $ show other

benchmarks/Latency.hs+5−7

@@ -3,7 +3,7 @@ import Control.Monad

3import Control.Applicative

4import Control.Distributed.Process

5import Control.Distributed.Process.Node

6−import Network.Transport.TCP (createTransport, defaultTCPAddr, defaultTCPParameters)

6+import Network.Transport.TCP (createTransport, defaultTCPParameters)

7import Data.Binary (encode, decode)

8import qualified Data.ByteString.Lazy as BSL

@@ -31,9 +31,7 @@ initialProcess "CLIENT" = do

31main :: IO ()

32main = do

33 [role, host, port] <- getArgs

34− trans <- createTransport

34+ Right transport <- createTransport

35− (defaultTCPAddr host port) defaultTCPParameters

35+ host port (\sn -> (host, sn)) defaultTCPParameters

36− case trans of

37− Right transport -> do node <- newLocalNode transport initRemoteTable

36+ node <- newLocalNode transport initRemoteTable

38− runProcess node $ initialProcess role

37+ runProcess node $ initialProcess role

39− Left other -> error $ show other

benchmarks/ProcessRing.hs+2−2

@@ -12,7 +12,7 @@ import Control.Monad

12import Control.Distributed.Process hiding (catch)

13import Control.Distributed.Process.Node

14import Control.Exception (catch, SomeException)

15−import Network.Transport.TCP (createTransport, defaultTCPAddr, defaultTCPParameters)

15+import Network.Transport.TCP (createTransport, defaultTCPParameters)

16import System.Environment

17import System.Console.GetOpt

@@ -111,7 +111,7 @@ main = do

111 (opt, _) <- parseArgv argv

112 putStrLn $ "options: " ++ (show opt)

113 Right transport <- createTransport

114− (defaultTCPAddr "127.0.0.1" "8090" ) defaultTCPParameters

114+ "127.0.0.1" "8090" (\sn -> ("127.0.0.1", sn)) defaultTCPParameters

115 node <- newLocalNode transport initRemoteTable

116 catch (void $ runProcess node $ initialProcess opt)

117 (\(e :: SomeException) -> putStrLn $ "ERROR: " ++ (show e))

benchmarks/Spawns.hs+5−6

@@ -7,7 +7,7 @@ import Control.Monad

7import Control.Applicative

8import Control.Distributed.Process

9import Control.Distributed.Process.Node

10−import Network.Transport.TCP (createTransport, defaultTCPAddr, defaultTCPParameters)

10+import Network.Transport.TCP (createTransport, defaultTCPParameters)

11import Data.Binary (encode, decode)

12import qualified Data.ByteString.Lazy as BSL

@@ -42,8 +42,7 @@ initialProcess "CLIENT" = do

42main :: IO ()

43main = do

44 [role, host, port] <- getArgs

45+ Right transport <- createTransport

45− trans <- createTransport (defaultTCPAddr host port) defaultTCPParameters

46+ host port (\sn -> (host, sn)) defaultTCPParameters

46− case trans of

47− Right transport -> do node <- newLocalNode transport initRemoteTable

47+ node <- newLocalNode transport initRemoteTable

48− runProcess node $ initialProcess role

48+ runProcess node $ initialProcess role

49− Left other -> error $ show other

benchmarks/Throughput.hs+5−6

@@ -6,7 +6,7 @@ import Control.Monad

6import Control.Applicative

7import Control.Distributed.Process

8import Control.Distributed.Process.Node

9−import Network.Transport.TCP (createTransport, defaultTCPParameters, defaultTCPAddr)

9+import Network.Transport.TCP (createTransport, defaultTCPParameters)

10import Data.Binary

11import qualified Data.ByteString.Lazy as BSL

12import Data.Typeable

@@ -67,8 +67,7 @@ initialProcess "CLIENT" = do

67main :: IO ()

68main = do

69 [role, host, port] <- getArgs

70+ Right transport <- createTransport

70− trans <- createTransport (defaultTCPAddr host port) defaultTCPParameters

71+ host port (\sn -> (host, sn)) defaultTCPParameters

71− case trans of

72− Right transport -> do node <- newLocalNode transport initRemoteTable

72+ node <- newLocalNode transport initRemoteTable

73− runProcess node $ initialProcess role

73+ runProcess node $ initialProcess role

74− Left other -> error $ show other

distributed-process-tests/distributed-process-tests.cabal+6−20

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

6license: BSD3

7license-file: LICENSE

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

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

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

10copyright: Well-Typed LLP

11category: Control, Cloud Haskell

12build-type: Simple

@@ -31,7 +31,6 @@ library

31 bytestring >= 0.9 && < 0.12,

32 distributed-process >= 0.6.0 && < 0.8,

33 distributed-static,

34− exceptions >= 0.5,

35 HUnit >= 1.2 && < 1.7,

34 HUnit >= 1.2 && < 1.7,

36 network-transport >= 0.4.1.0 && < 0.6,

35 network-transport >= 0.4.1.0 && < 0.6,

37 network >= 2.5 && < 2.9,

36 network >= 2.5 && < 2.9,

@@ -115,7 +114,8 @@ Test-Suite TestStats

115 ghc-options: -Wall -eventlog -threaded -rtsopts -with-rtsopts=-N -fno-warn-unused-do-bind

114 ghc-options: -Wall -eventlog -threaded -rtsopts -with-rtsopts=-N -fno-warn-unused-do-bind

116 HS-Source-Dirs: tests

115 HS-Source-Dirs: tests

117

116

117+

118−Test-Suite TestMxInMemory

118+Test-Suite TestMx

119 Type: exitcode-stdio-1.0

120 Main-Is: runInMemory.hs

121 CPP-Options: -DTEST_SUITE_MODULE=Control.Distributed.Process.Tests.Mx

@@ -126,10 +126,10 @@ Test-Suite TestMxInMemory

126 network-transport-inmemory >= 0.5,

127 test-framework >= 0.6 && < 0.9

128 Extensions: CPP

129− ghc-options: -threaded -rtsopts -with-rtsopts=-N -fno-warn-unused-do-bind

129+ ghc-options: -Wall -threaded -rtsopts -with-rtsopts=-N -fno-warn-unused-do-bind

130 HS-Source-Dirs: tests

131

132−Test-Suite TestTracingInMemory

132+Test-Suite TestTracing

133 Type: exitcode-stdio-1.0

134 Main-Is: runInMemory.hs

135 CPP-Options: -DTEST_SUITE_MODULE=Control.Distributed.Process.Tests.Tracing

@@ -140,19 +140,5 @@ Test-Suite TestTracingInMemory

140 network-transport-inmemory >= 0.5,

141 test-framework >= 0.6 && < 0.9

142 Extensions: CPP

143− ghc-options: -eventlog -threaded -rtsopts -with-rtsopts=-N -fno-warn-unused-do-bind

143+ ghc-options: -Wall -eventlog -threaded -rtsopts -with-rtsopts=-N -fno-warn-unused-do-bind

144− HS-Source-Dirs: tests

145−

146−Test-Suite TestMxInTCP

147− Type: exitcode-stdio-1.0

148− Main-Is: runInMemory.hs

149− CPP-Options: -DTEST_SUITE_MODULE=Control.Distributed.Process.Tests.Mx

150− Build-Depends: base >= 4.4 && < 5,

151− distributed-process-tests,

152− network >= 2.3 && < 2.9,

153− network-transport >= 0.4.1.0 && < 0.6,

154− network-transport-inmemory >= 0.5,

155− test-framework >= 0.6 && < 0.9

156− Extensions: CPP

157− ghc-options: -Wall -threaded -rtsopts -with-rtsopts=-N -fno-warn-unused-do-bind

158 HS-Source-Dirs: tests

144 HS-Source-Dirs: tests

distributed-process-tests/src/Control/Distributed/Process/Tests/CH.hs+234−466

@@ -25,32 +25,24 @@ import Control.Concurrent.MVar

25 )

26import Control.Monad (replicateM_, replicateM, forever, void, unless, join)

27import Control.Exception (SomeException, throwIO, ErrorCall(..))

28−import Control.Monad.Catch (try, catch, finally, mask, onException)

28+import qualified Control.Exception as Ex (catch)

29−import Control.Applicative ((<|>))

29+import Control.Applicative ((<$>), (<*>), pure, (<|>))

30import qualified Network.Transport as NT (closeEndPoint, EndPointAddress)

31−import Control.Distributed.Process hiding

31+import Control.Distributed.Process

32− ( try

33− , catch

34− , finally

35− , mask

36− , onException

37− )

38import Control.Distributed.Process.Internal.Types

32import Control.Distributed.Process.Internal.Types

33+ ( NodeId(nodeAddress)

39− ( LocalNode(localEndPoint)

34+ , LocalNode(localEndPoint)

40 , ProcessExitException(..)

35 , ProcessExitException(..)

41 , nullProcessId

36 , nullProcessId

42 , createUnencodedMessage

37 , createUnencodedMessage

43 )

38 )

44import Control.Distributed.Process.Node

39import Control.Distributed.Process.Node

45−import Control.Distributed.Process.Debug

46−import Control.Distributed.Process.Management.Internal.Types

47−import Control.Distributed.Process.Tests.Internal.Utils (shouldBe, pause)

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

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

49−import Data.Maybe (isNothing, isJust)

41+

50−import Test.HUnit (Assertion, assertBool, assertFailure)

42+import Test.HUnit (Assertion, assertFailure)

51import Test.Framework (Test, testGroup)

43import Test.Framework (Test, testGroup)

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

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

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

45+import Control.Rematch hiding (match)

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

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

56newtype Ping = Ping ProcessId

48newtype Ping = Ping ProcessId

@@ -73,7 +65,7 @@ expectThat a matcher = case res of

73forkTry :: IO () -> IO ThreadId

65forkTry :: IO () -> IO ThreadId

74forkTry p = do

66forkTry p = do

75 tid <- myThreadId

67 tid <- myThreadId

76− forkIO $ catch p (\e -> throwTo tid (e :: SomeException))

68+ forkIO $ Ex.catch p (\e -> throwTo tid (e :: SomeException))

78-- | The ping server from the paper

70-- | The ping server from the paper

79ping :: Process ()

71ping :: Process ()

@@ -83,46 +75,12 @@ ping = do

83 send partner (Ping self)

75 send partner (Ping self)

84 ping

76 ping

86−verifyClient :: String -> MVar Bool -> IO ()

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

88−

89−expectPing :: MVar Bool -> Process ()

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

91− where

92− checkPing (Ping _) = True

93−

94-- | Quick and dirty synchronous version of whereisRemoteAsync

78-- | Quick and dirty synchronous version of whereisRemoteAsync

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

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

96whereisRemote nid string = do

80whereisRemote nid string = do

97 whereisRemoteAsync nid string

81 whereisRemoteAsync nid string

98− receiveWait [

99− match (\(WhereIsReply _ mPid) -> return mPid)

82+ WhereIsReply _ mPid <- expect

100− ]

101−

102−verifyWhereIsRemote :: NodeId -> String -> Process ProcessId

103−verifyWhereIsRemote n s = whereisRemote n s >>= maybe (die "remote name not found") return

104−

105−waitForExit :: ProcessId -> Process DiedReason

106−waitForExit pid = monitor pid >>= waitForDown

107−

108−waitForDown :: MonitorRef -> Process DiedReason

109−waitForDown ref =

110− receiveWait [ matchIf (\(ProcessMonitorNotification ref' _ _) -> ref == ref')

111− (\(ProcessMonitorNotification _ _ dr) -> return dr) ]

112−

113−syncBreakConnection :: (NT.EndPointAddress -> NT.EndPointAddress -> IO ()) -> LocalNode -> LocalNode -> IO ()

114−syncBreakConnection breakConnection nid0 nid1 = do

115− m <- newEmptyMVar

116− _ <- forkProcess nid1 $ getSelfPid >>= liftIO . putMVar m

117− runProcess nid0 $ do

118− them <- liftIO $ takeMVar m

119− pinger <- spawnLocal $ forever $ send them ()

120− _ <- monitorNode (localNodeId nid1)

121− liftIO $ breakConnection (nodeAddress $ localNodeId nid0)

122− (nodeAddress $ localNodeId nid1)

123− NodeMonitorNotification _ _ _ <- expect

124− kill pinger "finished"

125− return ()

83+ return mPid

126

127data Add = Add ProcessId Double Double deriving (Typeable)

85data Add = Add ProcessId Double Double deriving (Typeable)

128data Divide = Divide ProcessId Double Double deriving (Typeable)

86data Divide = Divide ProcessId Double Double deriving (Typeable)

@@ -197,18 +155,12 @@ monitorTestProcess theirAddr mOrL un reason monitorSetup done =

197 unmonitor ref

155 unmonitor ref

198 liftIO $ putMVar done ()

156 liftIO $ putMVar done ()

199 (False, ref) -> do

157 (False, ref) -> do

200− receiveWait [

201− match (\(ProcessMonitorNotification ref' pid reason') -> do

158+ ProcessMonitorNotification ref' pid reason' <- expect

202− liftIO $ do

203− assertBool "Bad Monitor Signal"

204− (Just ref' == ref && pid == theirAddr &&

159+ True <- return $ Just ref' == ref && pid == theirAddr && mOrL && reason == reason'

205− mOrL && reason == reason')

206− putMVar done ())

160+ liftIO $ putMVar done ()

207− ]

208 )

161 )

209 (\(ProcessLinkException pid reason') -> do

162 (\(ProcessLinkException pid reason') -> do

210− (liftIO $ assertBool "link exception unmatched" $

211− pid == theirAddr && not mOrL && not un && reason == reason')

163+ True <- return $ pid == theirAddr && not mOrL && not un && reason == reason'

212 liftIO $ putMVar done ()

164 liftIO $ putMVar done ()

213 )

165 )

214

166

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

239 pid <- getSelfPid

191 pid <- getSelfPid

240 replicateM_ numPings $ do

192 replicateM_ numPings $ do

241 send pingServer (Pong pid)

193 send pingServer (Pong pid)

242− p <- expectTimeout 3000000

194+ Ping _ <- expect

243− case p of

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

195+ return ()

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

246

196

247 putMVar clientDone ()

197 putMVar clientDone ()

248

198

@@ -329,7 +279,7 @@ testMonitorLocalDeadProcess TestTransport{..} mOrL un = do

329 runProcess localNode $ do

279 runProcess localNode $ do

330 monitor theirAddr

280 monitor theirAddr

331 -- wait for the process to die

281 -- wait for the process to die

332− expect :: Process ProcessMonitorNotification

282+ ProcessMonitorNotification _ _ _ <- expect

333 monitorTestProcess theirAddr mOrL un DiedUnknownId Nothing done

283 monitorTestProcess theirAddr mOrL un DiedUnknownId Nothing done

334

284

335 takeMVar done

285 takeMVar done

@@ -403,17 +353,14 @@ testMath TestTransport{..} = do

403 runProcess localNode $ do

353 runProcess localNode $ do

404 pid <- getSelfPid

354 pid <- getSelfPid

405 send mathServer (Add pid 1 2)

355 send mathServer (Add pid 1 2)

406− three <- expect :: Process Double

356+ 3 <- expect :: Process Double

407 send mathServer (Divide pid 8 2)

357 send mathServer (Divide pid 8 2)

408− four <- expect :: Process Double

358+ 4 <- expect :: Process Double

409 send mathServer (Divide pid 8 0)

359 send mathServer (Divide pid 8 0)

410− divByZ <- expect

360+ DivByZero <- expect

411− liftIO $ putMVar clientDone (three, four, divByZ)

361+ liftIO $ putMVar clientDone ()

412

362

413− res <- takeMVar clientDone

363+ takeMVar clientDone

414− case res of

415− (3, 4, DivByZero) -> return ()

416− _ -> error $ "Something went horribly wrong"

417

364

418-- | Send first message (i.e. connect) to an already terminated process

365-- | Send first message (i.e. connect) to an already terminated process

419-- (without monitoring); then send another message to a second process on

366-- (without monitoring); then send another message to a second process on

@@ -441,9 +388,11 @@ testSendToTerminated TestTransport{..} = do

441 pid <- getSelfPid

388 pid <- getSelfPid

442 send server1 "Hi"

389 send server1 "Hi"

443 send server2 (Pong pid)

390 send server2 (Pong pid)

391+ Ping pid' <- expect

392+ True <- return $ pid' == server2

444− expectPing clientDone

393+ liftIO $ putMVar clientDone ()

445

394

446− verifyClient "Expected Ping from server" clientDone

395+ takeMVar clientDone

447

396

448-- | Test (non-zero) timeout

397-- | Test (non-zero) timeout

449testTimeout :: TestTransport -> Assertion

398testTimeout :: TestTransport -> Assertion

@@ -452,10 +401,10 @@ testTimeout TestTransport{..} = do

452 done <- newEmptyMVar

401 done <- newEmptyMVar

453

402

454 runProcess localNode $ do

403 runProcess localNode $ do

455− res <- receiveTimeout 1000000 [match (\Add{} -> return ())]

404+ Nothing <- receiveTimeout 1000000 [match (\(Add _ _ _) -> return ())]

456− liftIO $ putMVar done $ res == Nothing

405+ liftIO $ putMVar done ()

457

406

458− verifyClient "Expected receiveTimeout to timeout..." done

407+ takeMVar done

459

408

460-- | Test zero timeout

409-- | Test zero timeout

461testTimeout0 :: TestTransport -> Assertion

410testTimeout0 :: TestTransport -> Assertion

@@ -483,9 +432,10 @@ testTimeout0 TestTransport{..} = do

483 -- is not interested in, and then a single message that it wants

432 -- is not interested in, and then a single message that it wants

484 replicateM_ 10000 $ send server "Irrelevant message"

433 replicateM_ 10000 $ send server "Irrelevant message"

485 send server (Pong pid)

434 send server (Pong pid)

435+ Ping _ <- expect

486− expectPing clientDone

436+ liftIO $ putMVar clientDone ()

487

437

488− verifyClient "Expected Ping from server" clientDone

438+ takeMVar clientDone

489

439

490-- | Test typed channels

440-- | Test typed channels

491testTypedChannels :: TestTransport -> Assertion

441testTypedChannels :: TestTransport -> Assertion

@@ -508,10 +458,10 @@ testTypedChannels TestTransport{..} = do

508 runProcess localNode $ do

458 runProcess localNode $ do

509 (clientSendPort, rport) <- newChan

459 (clientSendPort, rport) <- newChan

510 sendChan serverSendPort (clientSendPort, 5)

460 sendChan serverSendPort (clientSendPort, 5)

511− ch <- receiveChan rport

461+ False <- receiveChan rport

512− liftIO $ putMVar clientDone $ ch == False

462+ liftIO $ putMVar clientDone ()

513

463

514− verifyClient "Expected channel to send 'False'" clientDone

464+ takeMVar clientDone

515

465

516-- | Test merging receive ports

466-- | Test merging receive ports

517testMergeChannels :: TestTransport -> Assertion

467testMergeChannels :: TestTransport -> Assertion

@@ -534,8 +484,9 @@ testMergeChannels TestTransport{..} = do

534 rs <- mapM charChannel "abc"

484 rs <- mapM charChannel "abc"

535 m <- mergePorts biased rs

485 m <- mergePorts biased rs

536 xs <- replicateM 9 $ receiveChan m

486 xs <- replicateM 9 $ receiveChan m

537− liftIO $ putMVar done $ xs == expected

487+ True <- return $ xs == expected

538− verifyClient "Expected single layer merge to match expected ordering" done

488+ liftIO $ putMVar done ()

489+ takeMVar done

539

490

540 -- Two layers of merging

491 -- Two layers of merging

541 testNested :: LocalNode -> Bool -> Bool -> String -> IO ()

492 testNested :: LocalNode -> Bool -> Bool -> String -> IO ()

@@ -546,8 +497,9 @@ testMergeChannels TestTransport{..} = do

546 ms <- mapM (mergePorts biasedInner) rss

497 ms <- mapM (mergePorts biasedInner) rss

547 m <- mergePorts biasedOuter ms

498 m <- mergePorts biasedOuter ms

548 xs <- replicateM (9 * 3) $ receiveChan m

499 xs <- replicateM (9 * 3) $ receiveChan m

549− liftIO $ putMVar done $ xs == expected

500+ True <- return $ xs == expected

550− verifyClient "Expected nested channels to match expeted ordering" done

501+ liftIO $ putMVar done ()

502+ takeMVar done

551

503

552 -- Test that if no messages are (immediately) available, the scheduler makes no difference

504 -- Test that if no messages are (immediately) available, the scheduler makes no difference

553 testBlocked :: LocalNode -> Bool -> IO ()

505 testBlocked :: LocalNode -> Bool -> IO ()

@@ -556,45 +508,43 @@ testMergeChannels TestTransport{..} = do

556 done <- newEmptyMVar

508 done <- newEmptyMVar

557

509

558 forkProcess localNode $ do

510 forkProcess localNode $ do

559− ss <- liftIO $ mapM readMVar vs

511+ [sa, sb, sc] <- liftIO $ mapM readMVar vs

560− case ss of

561− [sa, sb, sc] ->

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

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

563− [ -- a, b, c

513+ [ -- a, b, c

564− (sa, 'a')

514+ (sa, 'a')

565− , (sb, 'b')

515+ , (sb, 'b')

566− , (sc, 'c')

516+ , (sc, 'c')

567− -- a, c, b

517+ -- a, c, b

568− , (sa, 'a')

518+ , (sa, 'a')

569− , (sc, 'c')

519+ , (sc, 'c')

570− , (sb, 'b')

520+ , (sb, 'b')

571− -- b, a, c

521+ -- b, a, c

572− , (sb, 'b')

522+ , (sb, 'b')

573− , (sa, 'a')

523+ , (sa, 'a')

574− , (sc, 'c')

524+ , (sc, 'c')

575− -- b, c, a

525+ -- b, c, a

576− , (sb, 'b')

526+ , (sb, 'b')

577− , (sc, 'c')

527+ , (sc, 'c')

578− , (sa, 'a')

528+ , (sa, 'a')

579− -- c, a, b

529+ -- c, a, b

580− , (sc, 'c')

530+ , (sc, 'c')

581− , (sa, 'a')

531+ , (sa, 'a')

582− , (sb, 'b')

532+ , (sb, 'b')

583− -- c, b, a

533+ -- c, b, a

584− , (sc, 'c')

534+ , (sc, 'c')

585− , (sb, 'b')

535+ , (sb, 'b')

586− , (sa, 'a')

536+ , (sa, 'a')

587− ]

537+ ]

588− _ -> die "Something went horribly wrong"

589

538

590 forkProcess localNode $ do

539 forkProcess localNode $ do

591 (ss, rs) <- unzip <$> replicateM 3 newChan

540 (ss, rs) <- unzip <$> replicateM 3 newChan

592 liftIO $ mapM_ (uncurry putMVar) $ zip vs ss

541 liftIO $ mapM_ (uncurry putMVar) $ zip vs ss

593 m <- mergePorts biased rs

542 m <- mergePorts biased rs

594 xs <- replicateM (6 * 3) $ receiveChan m

543 xs <- replicateM (6 * 3) $ receiveChan m

595− liftIO $ putMVar done $ xs == "abcacbbacbcacabcba"

544+ True <- return $ xs == "abcacbbacbcacabcba"

545+ liftIO $ putMVar done ()

596

546

597− verifyClient "Expected merged ports to match expected ordering" done

547+ takeMVar done

598

548

599 mergePorts :: Serializable a => Bool -> [ReceivePort a] -> Process (ReceivePort a)

549 mergePorts :: Serializable a => Bool -> [ReceivePort a] -> Process (ReceivePort a)

600 mergePorts True = mergePortsBiased

550 mergePorts True = mergePortsBiased

@@ -612,9 +562,8 @@ testTerminate TestTransport{..} = do

612 localNode <- newLocalNode testTransport initRemoteTable

562 localNode <- newLocalNode testTransport initRemoteTable

613 runProcess localNode $ do

563 runProcess localNode $ do

614 e <- try terminate :: Process (Either ProcessTerminationException ())

564 e <- try terminate :: Process (Either ProcessTerminationException ())

615− if either show show e == show ProcessTerminationException

565+ True <- return $ either show show e == show ProcessTerminationException

616− then return ()

566+ return ()

617− else die "Unexpected result from terminate"

618

567

619testMonitorNode :: TestTransport -> Assertion

568testMonitorNode :: TestTransport -> Assertion

620testMonitorNode TestTransport{..} = do

569testMonitorNode TestTransport{..} = do

@@ -625,12 +574,11 @@ testMonitorNode TestTransport{..} = do

625

574

626 runProcess node2 $ do

575 runProcess node2 $ do

627 ref <- monitorNode (localNodeId node1)

576 ref <- monitorNode (localNodeId node1)

628− receiveWait [

629− match (\(NodeMonitorNotification ref' nid DiedDisconnect) ->

577+ NodeMonitorNotification ref' nid DiedDisconnect <- expect

630− return $ ref == ref' && nid == localNodeId node1)

578+ True <- return $ ref == ref' && nid == localNodeId node1

631− ] >>= liftIO . putMVar done

579+ liftIO $ putMVar done ()

632

580

633− verifyClient "Expected NodeMonitorNotification with matching ref & nodeId" done

581+ takeMVar done

634

582

635testMonitorLiveNode :: TestTransport -> Assertion

583testMonitorLiveNode :: TestTransport -> Assertion

636testMonitorLiveNode TestTransport{..} = do

584testMonitorLiveNode TestTransport{..} = do

@@ -645,16 +593,15 @@ testMonitorLiveNode TestTransport{..} = do

645 liftIO $ putMVar ready ()

593 liftIO $ putMVar ready ()

646 liftIO $ takeMVar readyr

594 liftIO $ takeMVar readyr

647 send p ()

595 send p ()

648− receiveWait [

649− match (\(NodeMonitorNotification ref' nid _) ->

596+ NodeMonitorNotification ref' nid _ <- expect

650− (return $ ref == ref' && nid == localNodeId node1))

597+ True <- return $ ref == ref' && nid == localNodeId node1

651− ] >>= liftIO . putMVar done

598+ liftIO $ putMVar done ()

652

599

653 takeMVar ready

600 takeMVar ready

654 closeLocalNode node1

601 closeLocalNode node1

655 putMVar readyr ()

602 putMVar readyr ()

656

603

657− verifyClient "Expected NodeMonitorNotification for LIVE node" done

604+ takeMVar done

658

605

659testMonitorChannel :: TestTransport -> Assertion

606testMonitorChannel :: TestTransport -> Assertion

660testMonitorChannel TestTransport{..} = do

607testMonitorChannel TestTransport{..} = do

@@ -664,20 +611,18 @@ testMonitorChannel TestTransport{..} = do

664 pid <- forkProcess node1 $ do

611 pid <- forkProcess node1 $ do

665 sport <- expect :: Process (SendPort ())

612 sport <- expect :: Process (SendPort ())

666 ref <- monitorPort sport

613 ref <- monitorPort sport

667− receiveWait [

614+ PortMonitorNotification ref' port' reason <- expect

668− -- reason might be DiedUnknownId if the receive port is GCed before the

615+ -- reason might be DiedUnknownId if the receive port is GCed before the

669− -- monitor is established (TODO: not sure that this is reasonable)

616+ -- monitor is established (TODO: not sure that this is reasonable)

670− match (\(PortMonitorNotification ref' port' reason) ->

671− return $ ref' == ref && port' == sendPortId sport &&

672− (reason == DiedNormal || reason == DiedUnknownId))

617+ return $ ref' == ref && port' == sendPortId sport && (reason == DiedNormal || reason == DiedUnknownId)

673− ] >>= liftIO . putMVar gotNotification

618+ liftIO $ putMVar gotNotification ()

674

619

675 runProcess node2 $ do

620 runProcess node2 $ do

676 (sport, _) <- newChan :: Process (SendPort (), ReceivePort ())

621 (sport, _) <- newChan :: Process (SendPort (), ReceivePort ())

677 send pid sport

622 send pid sport

678 liftIO $ threadDelay 100000

623 liftIO $ threadDelay 100000

679

624

680− verifyClient "Expected PortMonitorNotification" gotNotification

625+ takeMVar gotNotification

681

626

682testRegistry :: TestTransport -> Assertion

627testRegistry :: TestTransport -> Assertion

683testRegistry TestTransport{..} = do

628testRegistry TestTransport{..} = do

@@ -689,28 +634,20 @@ testRegistry TestTransport{..} = do

689

634

690 runProcess node $ do

635 runProcess node $ do

691 register "ping" pingServer

636 register "ping" pingServer

692− whereis "ping" >>= liftIO . assertBool "Unexpected ping" . (== Just pingServer)

637+ Just pid <- whereis "ping"

638+ True <- return $ pingServer == pid

693 us <- getSelfPid

639 us <- getSelfPid

694 nsend "ping" (Pong us)

640 nsend "ping" (Pong us)

695− receiveWait [

641+ Ping pid' <- expect

696− matchIf (\(Ping pid') -> pingServer == pid') (const $ return ())

642+ True <- return $ pingServer == pid'

697− ]

643+ Left (ProcessRegistrationException "dead" Nothing) <- try $ register "dead" deadProcess

698− checkRegException "dead" Nothing deadProcess

644+ Left (ProcessRegistrationException "ping" (Just x)) <- try $ register "ping" deadProcess

699− checkRegException "ping" (Just pingServer) deadProcess

645+ True <- return $ x == pingServer

700− try (unregister "dead") >>= checkReg "dead" Nothing

646+ Left (ProcessRegistrationException "dead" Nothing) <- try $ unregister "dead"

701 liftIO $ putMVar done ()

647 liftIO $ putMVar done ()

702

648

703 takeMVar done

649 takeMVar done

704

650

705− where

706− checkRegException name pid dead =

707− try (register name dead) >>= checkReg name pid

708−

709− checkReg name pid res =

710− case res of

711− Left (ProcessRegistrationException name pid) -> return ()

712− _ -> die $ "Unexpected Registration" ++ show res

713−

714testRegistryRemoteProcess :: TestTransport -> Assertion

651testRegistryRemoteProcess :: TestTransport -> Assertion

715testRegistryRemoteProcess TestTransport{..} = do

652testRegistryRemoteProcess TestTransport{..} = do

716 node1 <- newLocalNode testTransport initRemoteTable

653 node1 <- newLocalNode testTransport initRemoteTable

@@ -721,13 +658,13 @@ testRegistryRemoteProcess TestTransport{..} = do

721

658

722 runProcess node2 $ do

659 runProcess node2 $ do

723 register "ping" pingServer

660 register "ping" pingServer

724− whereis "ping" >>= liftIO . assertBool "Unexpected ping" . (== Just pingServer)

661+ Just pid <- whereis "ping"

662+ True <- return $ pingServer == pid

725 us <- getSelfPid

663 us <- getSelfPid

726 nsend "ping" (Pong us)

664 nsend "ping" (Pong us)

727− receiveWait [

665+ Ping pid' <- expect

728− matchIf (\(Ping pid') -> pingServer == pid')

666+ True <- return $ pingServer == pid'

729− (const $ liftIO $ putMVar done ())

667+ liftIO $ putMVar done ()

730− ]

731

668

732 takeMVar done

669 takeMVar done

733

670

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

735testRemoteRegistry TestTransport{..} = do

672testRemoteRegistry TestTransport{..} = do

736 node1 <- newLocalNode testTransport initRemoteTable

673 node1 <- newLocalNode testTransport initRemoteTable

737 node2 <- newLocalNode testTransport initRemoteTable

674 node2 <- newLocalNode testTransport initRemoteTable

675+ done <- newEmptyMVar

738

676

739 pingServer <- forkProcess node1 ping

677 pingServer <- forkProcess node1 ping

740 deadProcess <- forkProcess node1 (return ())

678 deadProcess <- forkProcess node1 (return ())

@@ -747,34 +685,32 @@ testRemoteRegistry TestTransport{..} = do

747 "ping" == label' && pid == pingServer)

685 "ping" == label' && pid == pingServer)

748 (\(RegisterReply _ _ _) -> return ()) ]

686 (\(RegisterReply _ _ _) -> return ()) ]

749

687

750− pid <- verifyWhereIsRemote nid1 "ping"

688+ Just pid <- whereisRemote nid1 "ping"

751− liftIO $ assertBool "Expected pindServer to match pid" $ pingServer == pid

689+ True <- return $ pingServer == pid

752−

753 us <- getSelfPid

690 us <- getSelfPid

754 nsendRemote nid1 "ping" (Pong us)

691 nsendRemote nid1 "ping" (Pong us)

755− receiveWait [

692+ Ping pid' <- expect

756− match (\(Ping pid') -> return $ pingServer == pid')

693+ True <- return $ pingServer == pid'

757− ] >>= liftIO . assertBool "Expected Ping with ping server's ProcessId"

758

694

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

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

760 -- in owner field.

696 -- in owner field.

761 registerRemoteAsync nid1 "dead" deadProcess

697 registerRemoteAsync nid1 "dead" deadProcess

698+ receiveWait [

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

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

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

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

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

765−

766 registerRemoteAsync nid1 "ping" deadProcess

701 registerRemoteAsync nid1 "ping" deadProcess

767 receiveWait [

702 receiveWait [

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

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

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

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

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

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

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

772−

773 unregisterRemoteAsync nid1 "dead"

706 unregisterRemoteAsync nid1 "dead"

774 receiveWait [

707 receiveWait [

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

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

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

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

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

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

711+ liftIO $ putMVar done ()

712+

713+ takeMVar done

778

714

779testRemoteRegistryRemoteProcess :: TestTransport -> Assertion

715testRemoteRegistryRemoteProcess :: TestTransport -> Assertion

780testRemoteRegistryRemoteProcess TestTransport{..} = do

716testRemoteRegistryRemoteProcess TestTransport{..} = do

@@ -790,15 +726,15 @@ testRemoteRegistryRemoteProcess TestTransport{..} = do

790 receiveWait [

726 receiveWait [

791 matchIf (\(RegisterReply label' _ _) -> "ping" == label')

727 matchIf (\(RegisterReply label' _ _) -> "ping" == label')

792 (\(RegisterReply _ _ _) -> return ()) ]

728 (\(RegisterReply _ _ _) -> return ()) ]

793− pid <- verifyWhereIsRemote nid1 "ping"

729+ Just pid <- whereisRemote nid1 "ping"

794− liftIO $ assertBool "Expected pingServer to match remote name" $ pingServer == pid

730+ True <- return $ pingServer == pid

795 us <- getSelfPid

731 us <- getSelfPid

796 nsendRemote nid1 "ping" (Pong us)

732 nsendRemote nid1 "ping" (Pong us)

797− receiveWait [

733+ Ping pid' <- expect

798− match (\(Ping pid') -> return $ pingServer == pid')

734+ True <- return $ pingServer == pid'

799− ] >>= liftIO . putMVar done

735+ liftIO $ putMVar done ()

800

736

801− verifyClient "Expected Ping with ping server's ProcessId" done

737+ takeMVar done

802

738

803testSpawnLocal :: TestTransport -> Assertion

739testSpawnLocal :: TestTransport -> Assertion

804testSpawnLocal TestTransport{..} = do

740testSpawnLocal TestTransport{..} = do

@@ -812,14 +748,15 @@ testSpawnLocal TestTransport{..} = do

812 sport <- expect

748 sport <- expect

813 sendChan sport (1234 :: Int)

749 sendChan sport (1234 :: Int)

814

750

815− sport <- spawnChannelLocal $

751+ sport <- spawnChannelLocal $ \rport -> do

816− \rport -> (receiveChan rport :: Process Int) >>= send us

752+ (1234 :: Int) <- receiveChan rport

753+ send us ()

817

754

818 send pid sport

755 send pid sport

756+ () <- expect

819− expect >>= liftIO . putMVar done

757+ liftIO $ putMVar done ()

820

758

821− res <- takeMVar done

759+ takeMVar done

822− assertBool "Expected 1234 :: Int" $ res == (1234 :: Int)

823

760

824testSpawnAsyncStrictness :: TestTransport -> Assertion

761testSpawnAsyncStrictness :: TestTransport -> Assertion

825testSpawnAsyncStrictness TestTransport{..} = do

762testSpawnAsyncStrictness TestTransport{..} = do

@@ -848,9 +785,8 @@ testReconnect TestTransport{..} = do

848 liftIO $ putMVar processA us

785 liftIO $ putMVar processA us

849 msg1 <- expect

786 msg1 <- expect

850 msg2 <- expect

787 msg2 <- expect

851− liftIO $ do

852− assertBool "messages did not match" $ msg1 == "message 1" && msg2 == "message 3"

788+ True <- return $ msg1 == "message 1" && msg2 == "message 3"

853− putMVar sendTestOk ()

789+ liftIO $ putMVar sendTestOk ()

854

790

855 forkProcess node2 $ do

791 forkProcess node2 $ do

856 {-

792 {-

@@ -883,7 +819,7 @@ testReconnect TestTransport{..} = do

883 matchIf (\(RegisterReply label' _ _) -> "a" == label')

819 matchIf (\(RegisterReply label' _ _) -> "a" == label')

884 (\(RegisterReply _ _ _) -> return ()) ]

820 (\(RegisterReply _ _ _) -> return ()) ]

885

821

886− _ <- whereisRemote nid1 "a"

822+ Just _ <- whereisRemote nid1 "a"

887

823

888

824

889 -- Simulate network failure

825 -- Simulate network failure

@@ -902,10 +838,9 @@ testReconnect TestTransport{..} = do

902 (\(RegisterReply _ _ _) -> return ()) ]

838 (\(RegisterReply _ _ _) -> return ()) ]

903

839

904 -- Check

840 -- Check

905− mPid <- whereisRemote nid1 "a" -- this will fail because the name is removed when the node is disconnected

841+ Nothing <- whereisRemote nid1 "a" -- this will fail because the name is removed when the node is disconnected

906− liftIO $ assertBool "Expected remote name to be lost" $ mPid == Nothing

907− verifyWhereIsRemote nid1 "b" -- this will suceed because the value is set after thereconnect

842+ Just _ <- whereisRemote nid1 "b" -- this will suceed because the value is set after thereconnect

908− verifyWhereIsRemote nid1 "c"

843+ Just _ <- whereisRemote nid1 "c"

909

844

910 liftIO $ putMVar registerTestOk ()

845 liftIO $ putMVar registerTestOk ()

911

846

@@ -946,19 +881,19 @@ testUSend usendPrim TestTransport{..} numMessages = do

946 isSorted _ = True

881 isSorted _ = True

947 -- The list can't be null since there are no failures after sending

882 -- The list can't be null since there are no failures after sending

948 -- the last message.

883 -- the last message.

949− liftIO $ putMVar usendTestOk $ isSorted msgs && not (null msgs)

884+ True <- return $ isSorted msgs && not (null msgs)

885+ liftIO $ putMVar usendTestOk ()

950

886

951 forkProcess node2 $ do

887 forkProcess node2 $ do

952 them <- liftIO $ readMVar processA

888 them <- liftIO $ readMVar processA

953 getSelfPid >>= send them

889 getSelfPid >>= send them

954− expect :: Process ()

890+ () <- expect

955 forM_ [1..numMessages] $ \i -> do

891 forM_ [1..numMessages] $ \i -> do

956 liftIO $ testBreakConnection (nodeAddress nid1) (nodeAddress nid2)

892 liftIO $ testBreakConnection (nodeAddress nid1) (nodeAddress nid2)

957 usendPrim them i

893 usendPrim them i

958 liftIO (threadDelay 30000)

894 liftIO (threadDelay 30000)

959

895

960− res <- takeMVar usendTestOk

896+ takeMVar usendTestOk

961− assertBool "Unexpected failure after sending last msg" res

962

897

963-- | Test 'matchAny'. This repeats the 'testMath' but with a proxy server

898-- | Test 'matchAny'. This repeats the 'testMath' but with a proxy server

964-- in between

899-- in between

@@ -984,17 +919,14 @@ testMatchAny TestTransport{..} = do

984 runProcess localNode $ do

919 runProcess localNode $ do

985 pid <- getSelfPid

920 pid <- getSelfPid

986 send mathServer (Add pid 1 2)

921 send mathServer (Add pid 1 2)

987− three <- expect :: Process Double

922+ 3 <- expect :: Process Double

988 send mathServer (Divide pid 8 2)

923 send mathServer (Divide pid 8 2)

989− four <- expect :: Process Double

924+ 4 <- expect :: Process Double

990 send mathServer (Divide pid 8 0)

925 send mathServer (Divide pid 8 0)

991− divByZ <- expect

926+ DivByZero <- expect

992− liftIO $ putMVar clientDone (three, four, divByZ)

927+ liftIO $ putMVar clientDone ()

993

928

994− res <- takeMVar clientDone

929+ takeMVar clientDone

995− case res of

996− (3, 4, DivByZero) -> return ()

997− _ -> error "Unexpected result"

998

930

999-- | Test 'matchAny'. This repeats the 'testMath' but with a proxy server

931-- | Test 'matchAny'. This repeats the 'testMath' but with a proxy server

1000-- in between, however we block 'Divide' requests ....

932-- in between, however we block 'Divide' requests ....

@@ -1021,13 +953,12 @@ testMatchAnyHandle TestTransport{..} = do

1021 runProcess localNode $ do

953 runProcess localNode $ do

1022 pid <- getSelfPid

954 pid <- getSelfPid

1023 send mathServer (Add pid 1 2)

955 send mathServer (Add pid 1 2)

1024− three <- expect :: Process Double

956+ 3 <- expect :: Process Double

1025 send mathServer (Divide pid 8 2)

957 send mathServer (Divide pid 8 2)

1026− res <- (expectTimeout 100000) :: Process (Maybe Double)

958+ Nothing <- (expectTimeout 100000) :: Process (Maybe Double)

1027− liftIO $ putMVar clientDone $ three == 3 && res == Nothing

959+ liftIO $ putMVar clientDone ()

1028−

1029− verifyClient "Expected Nothing (i.e. timeout)" clientDone

1030

960

961+ takeMVar clientDone

1031 where maybeForward :: ProcessId -> Message -> Process (Maybe ())

962 where maybeForward :: ProcessId -> Message -> Process (Maybe ())

1032 maybeForward s msg =

963 maybeForward s msg =

1033 handleMessage msg (\m@(Add _ _ _) -> send s m)

964 handleMessage msg (\m@(Add _ _ _) -> send s m)

@@ -1055,10 +986,8 @@ testMatchAnyNoHandle TestTransport{..} = do

1055 Just _ -> die "NONSENSE")

986 Just _ -> die "NONSENSE")

1056 ]

987 ]

1057 -- we *must* have removed the message from our mailbox though!!!

988 -- we *must* have removed the message from our mailbox though!!!

1058− res <- receiveTimeout 100000 [ match (\(Add _ _ _) -> return ()) ]

989+ Nothing <- receiveTimeout 100000 [ match (\(Add _ _ _) -> return ()) ]

1059− liftIO $ do

1060− assertBool "Expected timeout!" $ res == Nothing

1061− putMVar serverDone ()

990+ liftIO $ putMVar serverDone ()

1062 putMVar addr server

991 putMVar addr server

1063

992

1064 -- Client

993 -- Client

@@ -1101,21 +1030,14 @@ testMatchAnyIf TestTransport{..} = do

1101 runProcess localNode $ do

1030 runProcess localNode $ do

1102 pid <- getSelfPid

1031 pid <- getSelfPid

1103 send server (pid, "foo")

1032 send server (pid, "foo")

1104− foo <- expect

1033+ "foo" <- expect

1105− -- provoking what would be the wrong ordering is informative here...

1106−

1107− send server (pid, "bar")

1108− bar <- (expectTimeout 100000) :: Process (Maybe String) -- was Double o_O !?

1109−

1110 send server (pid, "baz")

1034 send server (pid, "baz")

1111− baz <- expect

1035+ "baz" <- expect

1112−

1036+ send server (pid, "bar")

1037+ Nothing <- (expectTimeout 100000) :: Process (Maybe Double)

1113− liftIO $ putMVar clientDone (foo, bar, baz)

1038+ liftIO $ putMVar clientDone ()

1114−

1115− res <- takeMVar clientDone

1116− let res' = res == ("foo", Nothing, "baz")

1117− assertBool "Expected timeout due to type mismatch" res'

1118

1039

1040+ takeMVar clientDone

1119 where tryHandleMessage :: Message -> Process (Maybe ())

1041 where tryHandleMessage :: Message -> Process (Maybe ())

1120 tryHandleMessage msg =

1042 tryHandleMessage msg =

1121 handleMessage msg (\(pid :: ProcessId, (m :: String))

1043 handleMessage msg (\(pid :: ProcessId, (m :: String))

@@ -1148,13 +1070,12 @@ testMatchMessageWithUnwrap TestTransport{..} = do

1148 runProcess localNode $ do

1070 runProcess localNode $ do

1149 pid <- getSelfPid

1071 pid <- getSelfPid

1150 send server (pid, wrapMessage ("foo" :: String))

1072 send server (pid, wrapMessage ("foo" :: String))

1151− foo <- expect

1073+ "foo" <- expect

1152 send server (pid, wrapMessage ("baz" :: String))

1074 send server (pid, wrapMessage ("baz" :: String))

1153− baz <- expect

1075+ "baz" <- expect

1154− liftIO $ putMVar clientDone (foo, baz)

1076+ liftIO $ putMVar clientDone ()

1155

1077

1156− res <- takeMVar clientDone

1078+ takeMVar clientDone

1157− assertBool "Unexpected unwrapped results" $ res == ("foo", "baz")

1158

1079

1159-- Test 'receiveChanTimeout'

1080-- Test 'receiveChanTimeout'

1160testReceiveChanTimeout :: TestTransport -> Assertion

1081testReceiveChanTimeout :: TestTransport -> Assertion

@@ -1172,19 +1093,18 @@ testReceiveChanTimeout TestTransport{..} = do

1172

1093

1173 -- Wait for a message with a delay. No message arrives, we should get

1094 -- Wait for a message with a delay. No message arrives, we should get

1174 -- Nothing after the delay.

1095 -- Nothing after the delay.

1175− receiveChanTimeout 100000 rp >>= maybe (return ()) (const $ die "Expected Timeout")

1096+ Nothing <- receiveChanTimeout 100000 rp

1176

1097

1177 -- Let the sender know that it can send a message.

1098 -- Let the sender know that it can send a message.

1178 liftIO $ putMVar mvSender ()

1099 liftIO $ putMVar mvSender ()

1179

1100

1180 -- Wait for a message with a delay again. Now a message arrives after

1101 -- Wait for a message with a delay again. Now a message arrives after

1181 -- 0.1 seconds

1102 -- 0.1 seconds

1182− res <- receiveChanTimeout 20000000 rp >>= maybe (die "Timeout") return

1103+ Just True <- receiveChanTimeout 20000000 rp

1183− liftIO $ assertBool "Expected result to be 'True'" res

1184

1104

1185 -- Wait for a message with zero timeout: non-blocking check. No message is

1105 -- Wait for a message with zero timeout: non-blocking check. No message is

1186 -- available, we get Nothing

1106 -- available, we get Nothing

1187− receiveChanTimeout 0 rp >>= maybe (return ()) (const $ die "Expected Timeout")

1107+ Nothing <- receiveChanTimeout 0 rp

1188

1108

1189 -- Let the sender know that it can send a message.

1109 -- Let the sender know that it can send a message.

1190 liftIO $ putMVar mvSender ()

1110 liftIO $ putMVar mvSender ()

@@ -1194,7 +1114,7 @@ testReceiveChanTimeout TestTransport{..} = do

1194 liftIO $ threadDelay 100000

1114 liftIO $ threadDelay 100000

1195 mb <- receiveChanTimeout 0 rp

1115 mb <- receiveChanTimeout 0 rp

1196 case mb of

1116 case mb of

1197− Just b -> do liftIO $ assertBool "Unexpected Message" $ not b

1117+ Just b -> do False <- return b; return ()

1198 _ -> loop

1118 _ -> loop

1199

1119

1200 liftIO $ putMVar done ()

1120 liftIO $ putMVar done ()

@@ -1231,8 +1151,8 @@ testReceiveChanFeatures TestTransport{..} = do

1231

1151

1232 rp1 <- mergePortsBiased [even <$> rpInt, rpBool]

1152 rp1 <- mergePortsBiased [even <$> rpInt, rpBool]

1233

1153

1234− receiveChan rp1 >>= liftIO . assertBool "Expected True"

1154+ True <- receiveChan rp1

1235− receiveChan rp1 >>= liftIO . assertBool "Expected False" . not

1155+ False <- receiveChan rp1

1236

1156

1237 -- Test Applicative instance

1157 -- Test Applicative instance

1238

1158

@@ -1241,7 +1161,7 @@ testReceiveChanFeatures TestTransport{..} = do

1241

1161

1242 let rp2 = pure (+) <*> rpInt <*> rpInt

1162 let rp2 = pure (+) <*> rpInt <*> rpInt

1243

1163

1244− receiveChan rp2 >>= liftIO . assertBool "Expected 7" . (== 7)

1164+ 7 <- receiveChan rp2

1245

1165

1246 -- Test Alternative instance

1166 -- Test Alternative instance

1247

1167

@@ -1250,8 +1170,8 @@ testReceiveChanFeatures TestTransport{..} = do

1250

1170

1251 let rp3 = (even <$> rpInt) <|> rpBool

1171 let rp3 = (even <$> rpInt) <|> rpBool

1252

1172

1253− receiveChan rp3 >>= liftIO . assertBool "Expected False" . not

1173+ False <- receiveChan rp3

1254− receiveChan rp3 >>= liftIO . assertBool "Expected True"

1174+ True <- receiveChan rp3

1255

1175

1256 -- Test Monad instance

1176 -- Test Monad instance

1257

1177

@@ -1265,86 +1185,17 @@ testReceiveChanFeatures TestTransport{..} = do

1265 then rpInt

1185 then rpInt

1266 else return 7

1186 else return 7

1267

1187

1268− receiveChan rp4 >>= liftIO . assertBool "Expected 5" . (== 5)

1188+ 5 <- receiveChan rp4

1269− receiveChan rp4 >>= liftIO . assertBool "Expected 7" . (== 7)

1189+ 7 <- receiveChan rp4

1270

1190

1271 liftIO $ putMVar done ()

1191 liftIO $ putMVar done ()

1272

1192

1273 takeMVar done

1193 takeMVar done

1274

1194

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−

1345testKillLocal :: TestTransport -> Assertion

1195testKillLocal :: TestTransport -> Assertion

1346testKillLocal TestTransport{..} = do

1196testKillLocal TestTransport{..} = do

1347 localNode <- newLocalNode testTransport initRemoteTable

1197 localNode <- newLocalNode testTransport initRemoteTable

1198+ done <- newEmptyMVar

1348

1199

1349 pid <- forkProcess localNode $ do

1200 pid <- forkProcess localNode $ do

1350 liftIO $ threadDelay 1000000

1201 liftIO $ threadDelay 1000000

@@ -1353,18 +1204,17 @@ testKillLocal TestTransport{..} = do

1353 ref <- monitor pid

1204 ref <- monitor pid

1354 us <- getSelfPid

1205 us <- getSelfPid

1355 kill pid "TestKill"

1206 kill pid "TestKill"

1356− mn <- expect

1207+ ProcessMonitorNotification ref' pid' (DiedException ex) <- expect

1357− case mn of

1358− ProcessMonitorNotification ref' pid' (DiedException ex) ->

1359− case ref == ref' && pid == pid' && ex == "killed-by=" ++ show us ++ ",reason=TestKill" of

1208+ True <- return $ ref == ref' && pid == pid' && ex == "killed-by=" ++ show us ++ ",reason=TestKill"

1360− True -> return ()

1209+ liftIO $ putMVar done ()

1361− False -> die "Invalid ProcessMonitorNotification received"

1210+

1362− _ -> die "Something went horribly wrong"

1211+ takeMVar done

1363

1212

1364testKillRemote :: TestTransport -> Assertion

1213testKillRemote :: TestTransport -> Assertion

1365testKillRemote TestTransport{..} = do

1214testKillRemote TestTransport{..} = do

1366 node1 <- newLocalNode testTransport initRemoteTable

1215 node1 <- newLocalNode testTransport initRemoteTable

1367 node2 <- newLocalNode testTransport initRemoteTable

1216 node2 <- newLocalNode testTransport initRemoteTable

1217+ done <- newEmptyMVar

1368

1218

1369 pid <- forkProcess node1 $ do

1219 pid <- forkProcess node1 $ do

1370 liftIO $ threadDelay 1000000

1220 liftIO $ threadDelay 1000000

@@ -1373,17 +1223,11 @@ testKillRemote TestTransport{..} = do

1373 ref <- monitor pid

1223 ref <- monitor pid

1374 us <- getSelfPid

1224 us <- getSelfPid

1375 kill pid "TestKill"

1225 kill pid "TestKill"

1376− mn <- expect

1377− case mn of

1378− ProcessMonitorNotification ref' pid' (DiedException reason) ->

1226+ ProcessMonitorNotification ref' pid' (DiedException reason) <- expect

1379− case (ref == ref', pid == pid', reason == "killed-by=" ++ show us ++ ",reason=TestKill") of

1227+ True <- return $ ref == ref' && pid == pid' && reason == "killed-by=" ++ show us ++ ",reason=TestKill"

1380− (True, True, True) -> return ()

1228+ liftIO $ putMVar done ()

1381− (a, b, c) -> do

1229+

1382− let a' = if a then "" else "Invalid ref"

1230+ takeMVar done

1383− let b' = if b then "" else "Invalid pid"

1384− let c' = if c then "" else "Invalid message"

1385− die $ unwords [a', b', c']

1386− _ -> die "Received unexpected message"

1387

1231

1388testCatchesExit :: TestTransport -> Assertion

1232testCatchesExit :: TestTransport -> Assertion

1389testCatchesExit TestTransport{..} = do

1233testCatchesExit TestTransport{..} = do

@@ -1409,10 +1253,7 @@ testHandleMessageIf TestTransport{..} = do

1409 self <- getSelfPid

1253 self <- getSelfPid

1410 send self (5 :: Integer, 10 :: Integer)

1254 send self (5 :: Integer, 10 :: Integer)

1411 msg <- receiveWait [ matchMessage return ]

1255 msg <- receiveWait [ matchMessage return ]

1412− handleMessageIf msg

1256+ Nothing <- handleMessageIf msg (\() -> True) (\() -> die $ "whoops")

1413− (\() -> True)

1414− (\() -> die "whoops") >>= maybe (return ())

1415− (const $ die "Expected Mismatch")

1416 handleMessageIf msg (\(x :: Integer, y :: Integer) -> x == 5 && y == 10)

1257 handleMessageIf msg (\(x :: Integer, y :: Integer) -> x == 5 && y == 10)

1417 (\input -> liftIO $ putMVar done input)

1258 (\input -> liftIO $ putMVar done input)

1418 return ()

1259 return ()

@@ -1457,11 +1298,8 @@ testDie TestTransport{..} = do

1457 (die ("foobar", 123 :: Int))

1298 (die ("foobar", 123 :: Int))

1458 `catchExit` \_from reason -> do

1299 `catchExit` \_from reason -> do

1459 -- TODO: should verify that 'from' has the right value

1300 -- TODO: should verify that 'from' has the right value

1460− let res = reason == ("foobar", 123 :: Int)

1301+ True <- return $ reason == ("foobar", 123 :: Int)

1461 liftIO $ putMVar done ()

1302 liftIO $ putMVar done ()

1462− if res

1463− then return ()

1464− else die "Something went horribly wrong"

1465

1303

1466 takeMVar done

1304 takeMVar done

1467

1305

@@ -1475,11 +1313,8 @@ testPrettyExit TestTransport{..} = do

1475 `catch` \ex@(ProcessExitException from _) ->

1313 `catch` \ex@(ProcessExitException from _) ->

1476 let expected = "exit-from=" ++ (show from)

1314 let expected = "exit-from=" ++ (show from)

1477 in do

1315 in do

1478− let res = (show ex) == expected

1316+ True <- return $ (show ex) == expected

1479 liftIO $ putMVar done ()

1317 liftIO $ putMVar done ()

1480− if res

1481− then return ()

1482− else die "Something went horribly wrong"

1483

1318

1484 takeMVar done

1319 takeMVar done

1485

1320

@@ -1495,26 +1330,17 @@ testExitLocal TestTransport{..} = do

1495 pid <- forkProcess localNode $ do

1330 pid <- forkProcess localNode $ do

1496 (liftIO (putMVar handlerSetUp ()) >> expect) `catchExit` \_from reason -> do

1331 (liftIO (putMVar handlerSetUp ()) >> expect) `catchExit` \_from reason -> do

1497 -- TODO: should verify that 'from' has the right value

1332 -- TODO: should verify that 'from' has the right value

1498− res <- return $ reason == "TestExit"

1333+ True <- return $ reason == "TestExit"

1499 liftIO $ putMVar supervisedDone ()

1334 liftIO $ putMVar supervisedDone ()

1500− if res

1501− then return ()

1502− else die "Something went horribly wrong"

1503

1335

1504 runProcess localNode $ do

1336 runProcess localNode $ do

1505 liftIO $ takeMVar handlerSetUp

1337 liftIO $ takeMVar handlerSetUp

1506 ref <- monitor pid

1338 ref <- monitor pid

1507 exit pid "TestExit"

1339 exit pid "TestExit"

1508 -- This time the client catches the exception, so it dies normally

1340 -- This time the client catches the exception, so it dies normally

1509− mn <- expect

1510− case mn of

1511− ProcessMonitorNotification ref' pid' DiedNormal -> do

1341+ ProcessMonitorNotification ref' pid' DiedNormal <- expect

1512− let res = ref == ref' && pid == pid'

1342+ True <- return $ ref == ref' && pid == pid'

1513− liftIO $ putMVar supervisorDone ()

1343+ liftIO $ putMVar supervisorDone ()

1514− if res

1515− then return ()

1516− else die "Something went horribly wrong"

1517− _ -> die "Something went horribly wrong"

1518

1344

1519 takeMVar supervisedDone

1345 takeMVar supervisedDone

1520 takeMVar supervisorDone

1346 takeMVar supervisorDone

@@ -1530,98 +1356,19 @@ testExitRemote TestTransport{..} = do

1530 (receiveWait [] :: Process ()) -- block forever

1356 (receiveWait [] :: Process ()) -- block forever

1531 `catchExit` \_from reason -> do

1357 `catchExit` \_from reason -> do

1532 -- TODO: should verify that 'from' has the right value

1358 -- TODO: should verify that 'from' has the right value

1533− res <- return $ reason == "TestExit"

1359+ True <- return $ reason == "TestExit"

1534 liftIO $ putMVar supervisedDone ()

1360 liftIO $ putMVar supervisedDone ()

1535− if res

1536− then return ()

1537− else die "Something went horribly wrong"

1538

1361

1539 runProcess node2 $ do

1362 runProcess node2 $ do

1540 ref <- monitor pid

1363 ref <- monitor pid

1541 exit pid "TestExit"

1364 exit pid "TestExit"

1542− mn <- expect

1543− case mn of

1544− ProcessMonitorNotification ref' pid' DiedNormal -> do

1365+ ProcessMonitorNotification ref' pid' DiedNormal <- expect

1545− res' <- return $ ref == ref' && pid == pid'

1366+ True <- return $ ref == ref' && pid == pid'

1546− liftIO $ putMVar supervisorDone ()

1367+ liftIO $ putMVar supervisorDone ()

1547− if res'

1548− then return ()

1549− else die "Something went horribly wrong"

1550− _ -> die "Something went horribly wrong"

1551

1368

1552 takeMVar supervisedDone

1369 takeMVar supervisedDone

1553 takeMVar supervisorDone

1370 takeMVar supervisorDone

1554

1371

1555−testRegistryMonitoring :: TestTransport -> Assertion

1556−testRegistryMonitoring TestTransport{..} = do

1557− node1 <- newLocalNode testTransport initRemoteTable

1558− node2 <- newLocalNode testTransport initRemoteTable

1559−

1560− let nid = localNodeId node2

1561− 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"

1567−

1568− runProcess node2 $ do

1569− register regName pid

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

1571− res <- whereis regName

1572− send pid ()

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

1574− us <- getSelfPid

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

1576−

1577−

1578− -- This delay isn't essential!

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

1580− -- and see), however waiting a few seconds here, makes it much more likely

1581− -- that in delayUntilMaybeUnregistered we will hit the match case right

1582− -- away, and thus not be subjected to a 20 second delay. The value of 4

1583− -- seconds appears to work optimally on osx and across several linux distros

1584− -- running in virtual machines (which is essentially what we do in CI)

1585− void $ receiveTimeout 4000000 [ matchAny return ]

1586−

1587− -- This delay doesn't serve much purpose in the happy path, however if some

1588− -- future patch breaks the cooperative behaviour of node controllers viz

1589− -- remote process registration and notification taking place via ncEffectDied,

1590− -- there would be the possibility of a race in the test case should we attempt

1591− -- to evaluate `whereis regName` on node2 right away. In case the name is still

1592− -- erroneously registered, observing the 20 second delay (or lack of), could at

1593− -- least give a hint that something is wrong, and we give up our time slice

1594− -- so that there's a higher change the registrations have been cleaned up

1595− -- in either case.

1596− runProcess node2 $ delayUntilMaybeUnregistered nid pid

1597−

1598− regHere <- newEmptyMVar

1599− runProcess node2 $ whereis regName >>= liftIO . putMVar regHere

1600− res <- takeMVar regHere

1601− case res of

1602− Nothing -> return ()

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

1604−

1605− where

1606− runUntilRegistered nid us = do

1607− whereisRemoteAsync nid regName

1608− receiveWait [

1609− matchIf (\(WhereIsReply n (Just p)) -> n == regName && p == us)

1610− (const $ return ())

1611− ]

1612−

1613− delayUntilMaybeUnregistered nid p = do

1614− whereisRemoteAsync nid regName

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

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

1617− (const $ return ())

1618− ]

1619− case res of

1620− Just () -> return ()

1621− Nothing -> delayUntilMaybeUnregistered nid p

1622−

1623− regName = "testRegisterRemote"

1624−

1625testUnsafeSend :: TestTransport -> Assertion

1372testUnsafeSend :: TestTransport -> Assertion

1626testUnsafeSend TestTransport{..} = do

1373testUnsafeSend TestTransport{..} = do

1627 serverAddr <- newEmptyMVar

1374 serverAddr <- newEmptyMVar

@@ -1637,7 +1384,8 @@ testUnsafeSend TestTransport{..} = do

1637 void $ forkProcess localNode $ do

1384 void $ forkProcess localNode $ do

1638 serverPid <- liftIO $ takeMVar serverAddr

1385 serverPid <- liftIO $ takeMVar serverAddr

1639 getSelfPid >>= unsafeSend serverPid

1386 getSelfPid >>= unsafeSend serverPid

1387+ () <- expect

1640− expect >>= liftIO . putMVar clientDone

1388+ liftIO $ putMVar clientDone ()

1641

1389

1642 takeMVar clientDone

1390 takeMVar clientDone

1643

1391

@@ -1656,7 +1404,8 @@ testUnsafeUSend TestTransport{..} = do

1656 void $ forkProcess localNode $ do

1404 void $ forkProcess localNode $ do

1657 serverPid <- liftIO $ takeMVar serverAddr

1405 serverPid <- liftIO $ takeMVar serverAddr

1658 getSelfPid >>= unsafeUSend serverPid

1406 getSelfPid >>= unsafeUSend serverPid

1407+ () <- expect

1659− expect >>= liftIO . putMVar clientDone

1408+ liftIO $ putMVar clientDone ()

1660

1409

1661 takeMVar clientDone

1410 takeMVar clientDone

1662

1411

@@ -1667,7 +1416,8 @@ testUnsafeNSend TestTransport{..} = do

1667 localNode <- newLocalNode testTransport initRemoteTable

1416 localNode <- newLocalNode testTransport initRemoteTable

1668

1417

1669 pid <- forkProcess localNode $ do

1418 pid <- forkProcess localNode $ do

1419+ () <- expect

1670− expect >>= liftIO . putMVar clientDone

1420+ liftIO $ putMVar clientDone ()

1671

1421

1672 void $ runProcess localNode $ do

1422 void $ runProcess localNode $ do

1673 register "foobar" pid

1423 register "foobar" pid

@@ -1685,7 +1435,8 @@ testUnsafeNSendRemote TestTransport{..} = do

1685 _ <- forkProcess localNode1 $ do

1435 _ <- forkProcess localNode1 $ do

1686 getSelfPid >>= register "foobar"

1436 getSelfPid >>= register "foobar"

1687 liftIO $ putMVar clientDone ()

1437 liftIO $ putMVar clientDone ()

1438+ () <- expect

1688− expect >>= liftIO . putMVar clientDone

1439+ liftIO $ putMVar clientDone ()

1689

1440

1690 takeMVar clientDone

1441 takeMVar clientDone

1691 void $ runProcess localNode2 $ do

1442 void $ runProcess localNode2 $ do

@@ -1709,7 +1460,7 @@ testUnsafeSendChan TestTransport{..} = do

1709 serverPid <- liftIO $ takeMVar serverAddr

1460 serverPid <- liftIO $ takeMVar serverAddr

1710 (sp, rp) <- newChan

1461 (sp, rp) <- newChan

1711 unsafeSend serverPid sp

1462 unsafeSend serverPid sp

1712− receiveChan rp :: Process ()

1463+ () <- receiveChan rp

1713 liftIO $ putMVar clientDone ()

1464 liftIO $ putMVar clientDone ()

1714

1465

1715 takeMVar clientDone

1466 takeMVar clientDone

@@ -1723,8 +1474,8 @@ testCallLocal TestTransport{..} = do

1723 runProcess node $ do

1474 runProcess node $ do

1724 r <- (/=) <$> getSelfPid <*> callLocal getSelfPid

1475 r <- (/=) <$> getSelfPid <*> callLocal getSelfPid

1725 liftIO $ putMVar result r

1476 liftIO $ putMVar result r

1726−

1727− takeMVar result >>= assertBool "Expected 'True'"

1477+ True <- takeMVar result

1478+ return ()

1728

1479

1729 -- Testing that when callLocal is interrupted, the worker is interrupted.

1480 -- Testing that when callLocal is interrupted, the worker is interrupted.

1730 ibox <- newIORef False

1481 ibox <- newIORef False

@@ -1741,9 +1492,10 @@ testCallLocal TestTransport{..} = do

1741 (send keeper ())

1492 (send keeper ())

1742 caller <- expect

1493 caller <- expect

1743 exit caller "test"

1494 exit caller "test"

1744− expect :: Process ()

1495+ () <- expect

1745−

1496+ return ()

1746− readIORef ibox >>= assertBool "Expected 'True'"

1497+ True <- readIORef ibox

1498+ return ()

1747

1499

1748 -- Testing that when the worker raises an exception, the exception is propagated to the parent.

1500 -- Testing that when the worker raises an exception, the exception is propagated to the parent.

1749 ibox2 <- newIORef False

1501 ibox2 <- newIORef False

@@ -1752,8 +1504,8 @@ testCallLocal TestTransport{..} = do

1752 liftIO $ writeIORef ibox2 $ case r of

1504 liftIO $ writeIORef ibox2 $ case r of

1753 Left (ErrorCall "e") -> True

1505 Left (ErrorCall "e") -> True

1754 _ -> False

1506 _ -> False

1755−

1756− readIORef ibox >>= assertBool "Expected 'True'"

1507+ True <- readIORef ibox

1508+ return ()

1757

1509

1758 -- Test that caller waits for the worker in correct situation

1510 -- Test that caller waits for the worker in correct situation

1759 ibox3 <- newIORef False

1511 ibox3 <- newIORef False

@@ -1770,8 +1522,8 @@ testCallLocal TestTransport{..} = do

1770 liftIO $ putMVar result3 =<< readIORef ibox3

1522 liftIO $ putMVar result3 =<< readIORef ibox3

1771 worker <- expect

1523 worker <- expect

1772 send worker ()

1524 send worker ()

1773−

1774− takeMVar result3 >>= assertBool "Expected 'True'"

1525+ True <- takeMVar result3

1526+ return ()

1775

1527

1776 -- Test that caller waits for the worker in case when caller gets an exception

1528 -- Test that caller waits for the worker in case when caller gets an exception

1777 ibox4 <- newIORef False

1529 ibox4 <- newIORef False

@@ -1787,10 +1539,14 @@ testCallLocal TestTransport{..} = do

1787 `finally` (liftIO $ putMVar result4 =<< readIORef ibox4)

1539 `finally` (liftIO $ putMVar result4 =<< readIORef ibox4)

1788 caller <- expect

1540 caller <- expect

1789 exit caller "hi!"

1541 exit caller "hi!"

1790−

1791− takeMVar result4 >>= assertBool "Expected 'True'"

1542+ True <- takeMVar result4

1543+ return ()

1792 -- XXX: Testing that when mask_ $ callLocal p runs p in masked state.

1544 -- XXX: Testing that when mask_ $ callLocal p runs p in masked state.

1793

1545

1546+

1547+

1548+

1549+

1794tests :: TestTransport -> IO [Test]

1550tests :: TestTransport -> IO [Test]

1795tests testtrans = return [

1551tests testtrans = return [

1796 testGroup "Basic features" [

1552 testGroup "Basic features" [

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

1814 , testCase "MatchMessageUnwrap" (testMatchMessageWithUnwrap testtrans)

1570 , testCase "MatchMessageUnwrap" (testMatchMessageWithUnwrap testtrans)

1815 , testCase "ReceiveChanTimeout" (testReceiveChanTimeout testtrans)

1571 , testCase "ReceiveChanTimeout" (testReceiveChanTimeout testtrans)

1816 , testCase "ReceiveChanFeatures" (testReceiveChanFeatures testtrans)

1572 , testCase "ReceiveChanFeatures" (testReceiveChanFeatures testtrans)

1817− , testCase "ChanLifecycle" (testChanLifecycle testtrans)

1818 , testCase "KillLocal" (testKillLocal testtrans)

1573 , testCase "KillLocal" (testKillLocal testtrans)

1819 , testCase "KillRemote" (testKillRemote testtrans)

1574 , testCase "KillRemote" (testKillRemote testtrans)

1820 , testCase "Die" (testDie testtrans)

1575 , testCase "Die" (testDie testtrans)

@@ -1824,7 +1579,6 @@ tests testtrans = return [

1824 , testCase "MaskRestoreScope" (testMaskRestoreScope testtrans)

1579 , testCase "MaskRestoreScope" (testMaskRestoreScope testtrans)

1825 , testCase "ExitLocal" (testExitLocal testtrans)

1580 , testCase "ExitLocal" (testExitLocal testtrans)

1826 , testCase "ExitRemote" (testExitRemote testtrans)

1581 , testCase "ExitRemote" (testExitRemote testtrans)

1827− , testCase "RegistryMonitoring" (testRegistryMonitoring testtrans)

1828 , testCase "TextCallLocal" (testCallLocal testtrans)

1582 , testCase "TextCallLocal" (testCallLocal testtrans)

1829 -- Unsafe Primitives

1583 -- Unsafe Primitives

1830 , testCase "TestUnsafeSend" (testUnsafeSend testtrans)

1584 , testCase "TestUnsafeSend" (testUnsafeSend testtrans)

@@ -1877,3 +1631,17 @@ tests testtrans = return [

1877 , testCase "MonitorUnreachable" (testMonitorUnreachable testtrans True False)

1631 , testCase "MonitorUnreachable" (testMonitorUnreachable testtrans True False)

1878 ]

1632 ]

1879 ]

1633 ]

1634+

1635+syncBreakConnection :: (NT.EndPointAddress -> NT.EndPointAddress -> IO ()) -> LocalNode -> LocalNode -> IO ()

1636+syncBreakConnection breakConnection nid0 nid1 = do

1637+ m <- newEmptyMVar

1638+ _ <- forkProcess nid1 $ getSelfPid >>= liftIO . putMVar m

1639+ runProcess nid0 $ do

1640+ them <- liftIO $ takeMVar m

1641+ pinger <- spawnLocal $ forever $ send them ()

1642+ _ <- monitorNode (localNodeId nid1)

1643+ liftIO $ breakConnection (nodeAddress $ localNodeId nid0)

1644+ (nodeAddress $ localNodeId nid1)

1645+ NodeMonitorNotification _ _ _ <- expect

1646+ kill pinger "finished"

1647+ return ()

distributed-process-tests/src/Control/Distributed/Process/Tests/Closure.hs+24−32

@@ -412,34 +412,29 @@ testSpawnSupervised TestTransport{..} rtable = do

412 throw supervisorDeath

413

414 forkProcess node2 $ do

415− res <- liftIO $ mapM readMVar [superPid, childPid]

415+ [super, child] <- liftIO $ mapM readMVar [superPid, childPid]

416+ ref <- monitor child

417+ self <- getSelfPid

418+ let waitForMOrL = do

419+ liftIO $ threadDelay 10000

420+ mpinfo <- getProcessInfo child

421+ case mpinfo of

422+ Nothing -> waitForMOrL

423+ Just pinfo ->

424+ unless (isJust $ lookup self (infoMonitors pinfo)) waitForMOrL

425+ waitForMOrL

426+ liftIO $ putMVar linkUp ()

427+ -- because monitor message was sent before message to process

428+ -- we hope that it will be processed before

429+ res <- expect

416 case res of

430 case res of

417− [super, child] -> do

418− ref <- monitor child

419− self <- getSelfPid

420− let waitForMOrL = do

421− liftIO $ threadDelay 10000

422− mpinfo <- getProcessInfo child

423− case mpinfo of

424− Nothing -> waitForMOrL

425− Just pinfo ->

426− unless (isJust $ lookup self (infoMonitors pinfo)) waitForMOrL

427− waitForMOrL

428− liftIO $ putMVar linkUp ()

429− -- because monitor message was sent before message to process

430− -- we hope that it will be processed before

431− res' <- expect

432− case res' of

433− (ProcessMonitorNotification ref' pid' (DiedException e)) ->

431+ (ProcessMonitorNotification ref' pid' (DiedException e)) ->

434− if (ref' == ref && pid' == child &&

432+ if (ref' == ref && pid' == child &&

435− e == show (ProcessLinkException super

433+ e == show (ProcessLinkException super

436− (DiedException (show supervisorDeath))))

434+ (DiedException (show supervisorDeath))))

437− then liftIO $ putMVar thirdProcessDone ()

435+ then liftIO $ putMVar thirdProcessDone ()

438− else error "Something went horribly wrong"

436+ else error "Something went horribly wrong"

439− _ -> error "Something went horribly wrong"

437+ _ -> error "Something went horribly wrong"

440−

441− _ -> die $ "Something went horribly wrong"

442−

443 takeMVar thirdProcessDone

438 takeMVar thirdProcessDone

444 where

439 where

445 supervisorDeath :: IOException

440 supervisorDeath :: IOException

@@ -510,11 +505,8 @@ testFib TestTransport{..} rtable = do

510 forkProcess (head nodes) $ do

505 forkProcess (head nodes) $ do

511 (sport, rport) <- newChan

506 (sport, rport) <- newChan

512 spawnLocal $ dfib (map localNodeId nodes, sport, 10)

507 spawnLocal $ dfib (map localNodeId nodes, sport, 10)

513− ff <- receiveChan rport :: Process Integer

508+ 55 <- receiveChan rport :: Process Integer

514 liftIO $ putMVar done ()

509 liftIO $ putMVar done ()

515− if ff /= 55

516− then die $ "Something went horribly wrong"

517− else return ()

518

510

519 takeMVar done

511 takeMVar done

520

512

@@ -522,7 +514,7 @@ testSpawnReconnect :: TestTransport -> RemoteTable -> Assertion

522testSpawnReconnect testtrans@TestTransport{..} rtable = do

514testSpawnReconnect testtrans@TestTransport{..} rtable = do

523 [node1, node2] <- replicateM 2 $ newLocalNode testTransport rtable

515 [node1, node2] <- replicateM 2 $ newLocalNode testTransport rtable

524 let nid1 = localNodeId node1

516 let nid1 = localNodeId node1

525− -- nid2 = localNodeId node2

517+ nid2 = localNodeId node2

526 done <- newEmptyMVar

518 done <- newEmptyMVar

527 iv <- newMVar (0 :: Int)

519 iv <- newMVar (0 :: Int)

528

520

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

@@ -20,7 +20,6 @@ module Control.Distributed.Process.Tests.Internal.Utils

20 -- ping !

21 , Ping(Ping)

22 , ping

23− , pause

24 , shouldBe

23 , shouldBe

25 , shouldMatch

24 , shouldMatch

26 , shouldContain

25 , shouldContain

@@ -74,16 +73,15 @@ import Control.Concurrent

74import Control.Concurrent.MVar

73import Control.Concurrent.MVar

75 ( putMVar

74 ( putMVar

76 )

75 )

77−import Control.Distributed.Process hiding (finally, catch)

76+import Control.Distributed.Process

78import Control.Distributed.Process.Node

77import Control.Distributed.Process.Node

79import Control.Distributed.Process.Serializable()

78import Control.Distributed.Process.Serializable()

81import Control.Exception (AsyncException(ThreadKilled), SomeException)

80import Control.Exception (AsyncException(ThreadKilled), SomeException)

82−import Control.Monad (forever, void)

81+import Control.Monad (forever)

83−import Control.Monad.Catch (finally, catch)

84import Control.Monad.STM (atomically)

82import Control.Monad.STM (atomically)

85import Control.Rematch hiding (match)

83import Control.Rematch hiding (match)

86−import Control.Rematch.Run

84+import Control.Rematch.Run

87import Data.Binary

85import Data.Binary

88import Data.Typeable (Typeable)

86import Data.Typeable (Typeable)

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

91import Test.HUnit.Base (assertBool)

89import Test.HUnit.Base (assertBool)

93import GHC.Generics

91import GHC.Generics

94−import System.Timeout (timeout)

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

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

97type TestResult a = MVar a

94type TestResult a = MVar a

@@ -110,18 +107,10 @@ data TestProcessControl = Stop | Go | Report ProcessId

110

107

111instance Binary TestProcessControl where

108instance Binary TestProcessControl where

112

109

113−data Private = Private

114− deriving (Typeable, Generic)

115−instance Binary Private where

116−

117-- | Does exactly what it says on the tin, doing so in the @Process@ monad.

110-- | Does exactly what it says on the tin, doing so in the @Process@ monad.

118noop :: Process ()

111noop :: Process ()

119noop = return ()

112noop = return ()

120

113

121−pause :: Int -> Process ()

122−pause delay =

123− void $ receiveTimeout delay [ match (\Private -> return ()) ]

124−

125synchronisedAssertion :: Eq a

114synchronisedAssertion :: Eq a

126 => String

115 => String

127 -> LocalNode

116 -> LocalNode

@@ -234,10 +223,10 @@ testProcessReport pid = do

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

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

235tryRunProcess node p = do

224tryRunProcess node p = do

236 tid <- liftIO myThreadId

225 tid <- liftIO myThreadId

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

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

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

239

227

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

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

241tryForkProcess node p = do

229tryForkProcess node p = do

242 tid <- liftIO myThreadId

230 tid <- liftIO myThreadId

243 forkProcess node $ catch p (\e -> liftIO $ throwTo tid (e::SomeException))

231 forkProcess node $ catch p (\e -> liftIO $ throwTo tid (e::SomeException))

232+

distributed-process-tests/src/Control/Distributed/Process/Tests/Mx.hs+53−390

@@ -1,23 +1,12 @@

1−{-# LANGUAGE DeriveGeneric #-}

1+{-# LANGUAGE DeriveGeneric #-}

2−{-# LANGUAGE ParallelListComp #-}

3−

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

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

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

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

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

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

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

7+import Control.Concurrent (threadDelay)

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

8+import Control.Distributed.Process

10− ( ProcessExitException(..)

11− , unsafeCreateUnencodedMessage

12− )

13import Control.Distributed.Process.Node

9import Control.Distributed.Process.Node

14−import qualified Control.Distributed.Process.UnsafePrimitives as Unsafe

15− ( send

16− , nsend

17− , nsendRemote

18− , usend

19− , sendChan

20− )

21import Control.Distributed.Process.Management

10import Control.Distributed.Process.Management

22 ( MxEvent(..)

11 ( MxEvent(..)

23 , MxAgentId(..)

12 , MxAgentId(..)

@@ -32,15 +21,14 @@ import Control.Distributed.Process.Management

32 , mxUpdateLocal

21 , mxUpdateLocal

33 , mxNotify

22 , mxNotify

34 , mxBroadcast

23 , mxBroadcast

24+ , mxGetId

35 )

25 )

36−import Control.Monad (void, unless)

26+import Control.Monad (void)

37−import Control.Monad.Catch(finally, bracket, try)

38−import Control.Rematch (equalTo)

39import Data.Binary

27import Data.Binary

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

28+import Data.List (find, sort)

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

29+import Data.Maybe (isJust)

42import Data.Typeable

30import Data.Typeable

43−import GHC.Generics hiding (from)

31+import GHC.Generics

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

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

45import Prelude hiding (catch, log)

33import Prelude hiding (catch, log)

46#endif

34#endif

@@ -56,17 +44,7 @@ data Publish = Publish

57instance Binary Publish where

45instance Binary Publish where

59−awaitExit :: ProcessId -> Process ()

60−awaitExit pid =

61− withMonitorRef pid $ \ref -> do

62− receiveWait

63− [ matchIf (\(ProcessMonitorNotification r _ _) -> r == ref)

64− (\_ -> return ())

65− ]

66− where

67− withMonitorRef p = bracket (monitor p) unmonitor

68−

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

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

70testAgentBroadcast result = do

48testAgentBroadcast result = do

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

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

@@ -81,7 +59,7 @@ testAgentBroadcast result = do

81 mxNotify ()

59 mxNotify ()

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

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

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

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

84− stash result =<< receiveChanTimeout 10000000 resultRP

62+ stash result =<< receiveChan resultRP

86 kill publisher "finished"

64 kill publisher "finished"

87 kill consumer "finished"

65 kill consumer "finished"

@@ -89,7 +67,7 @@ testAgentBroadcast result = do

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

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

90testAgentDualInput result = do

68testAgentDualInput result = do

91 (sp, rp) <- newChan

69 (sp, rp) <- newChan

92− s <- mxAgent (MxAgentId "sum-agent") (0 :: Int) [

70+ _ <- mxAgent (MxAgentId "sum-agent") (0 :: Int) [

93 mxSink $ (\(i :: Int) -> do

71 mxSink $ (\(i :: Int) -> do

94 mxSetLocal . (+i) =<< mxGetLocal

72 mxSetLocal . (+i) =<< mxGetLocal

95 i' <- mxGetLocal

73 i' <- mxGetLocal

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

99 else mxReady)

77 else mxReady)

100 ]

78 ]

101

102− mRef <- monitor s

103−

104 mxNotify (1 :: Int)

80 mxNotify (1 :: Int)

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

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

106 mxNotify (2 :: Int)

82 mxNotify (2 :: Int)

@@ -108,10 +84,6 @@ testAgentDualInput result = do

108 mxNotify (5 :: Int)

84 mxNotify (5 :: Int)

109

110 stash result =<< receiveChanTimeout 10000000 rp

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

115

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

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

117testAgentPrioritisation result = do

89testAgentPrioritisation result = do

@@ -123,12 +95,12 @@ testAgentPrioritisation result = do

123

124 let name = "prioritising-agent"

96 let name = "prioritising-agent"

125 (sp, rp) <- newChan

97 (sp, rp) <- newChan

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

98+ void $ mxAgent (MxAgentId name) ["first"] [

127− mxSink (\(s :: String) -> do

99+ mxSink (\(s :: String) -> do

128− mxUpdateLocal (s:)

100+ mxUpdateLocal ((s:))

129 st <- mxGetLocal

101 st <- mxGetLocal

130 case length st of

102 case length st of

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

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

132 mxDeactivate "finished"

104 mxDeactivate "finished"

133 _ -> mxReceive -- go to the mailbox

105 _ -> mxReceive -- go to the mailbox

134 )

106 )

@@ -139,33 +111,23 @@ testAgentPrioritisation result = do

139 mxNotify "fourth"

111 mxNotify "fourth"

140 nsend name "fifth"

112 nsend name "fifth"

141

113

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

114+ stash result . sort =<< receiveChan rp

143− awaitExit s

144

115

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

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

146testAgentMailboxHandling result = do

117testAgentMailboxHandling result = do

147 (sp, rp) <- newChan

118 (sp, rp) <- newChan

148− agent <- mxAgent (MxAgentId "mailbox-agent") () [

119+ agent <- mxAgent (MxAgentId "listener-agent") () [

149 mxSink $ \() -> (liftMX $ sendChan sp ()) >> mxReady

120 mxSink $ \() -> (liftMX $ sendChan sp ()) >> mxReady

150 ]

121 ]

151

122

152− nsend "mailbox-agent" ()

123+ nsend "listener-agent" ()

153

124

154 stash result =<< receiveChanTimeout 1000000 rp

125 stash result =<< receiveChanTimeout 1000000 rp

155 kill agent "finished"

126 kill agent "finished"

156

127

157testAgentEventHandling :: TestResult Bool -> Process ()

128testAgentEventHandling :: TestResult Bool -> Process ()

158testAgentEventHandling result = do

129testAgentEventHandling 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−

167 let initState = [] :: [MxEvent]

130 let initState = [] :: [MxEvent]

168− (rc, rs) <- newChan

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

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

170 (mxSink $ \ev -> do

132 (mxSink $ \ev -> do

171 st <- mxGetLocal

133 st <- mxGetLocal

@@ -174,7 +136,7 @@ testAgentEventHandling result = do

174 (MxSpawned _) -> mxSetLocal (ev:st)

136 (MxSpawned _) -> mxSetLocal (ev:st)

175 (MxProcessDied _ _) -> mxSetLocal (ev:st)

137 (MxProcessDied _ _) -> mxSetLocal (ev:st)

176 _ -> return ()

138 _ -> return ()

177− act >> (liftMX $ sendChan rc ()) >> mxReady),

139+ act >> mxReady),

178 (mxSink $ \(ev, sp :: SendPort Bool) -> do

140 (mxSink $ \(ev, sp :: SendPort Bool) -> do

179 st <- mxGetLocal

141 st <- mxGetLocal

180 let found =

142 let found =

@@ -194,349 +156,50 @@ testAgentEventHandling result = do

194 mxReady)

156 mxReady)

195 ]

157 ]

196

158

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

159+ _ <- monitor agentPid

198− faff 2000000 rs

199−

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

160+ (sp, rp) <- newChan

201−

202− pid <- spawnLocal $ expect >>= sendChan sp

161+ pid <- spawnLocal $ sendChan sp ()

162+ () <- receiveChan rp

203

163

204 -- By waiting for a monitor notification, we have a

164 -- By waiting for a monitor notification, we have a

205− -- higher probably that the agent has seen the spawn and died events

165+ -- higher probably that the agent has seen

206 monitor pid

166 monitor pid

207−

208− send pid ()

209− rct <- receiveChanTimeout 10000000 rp

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

211−

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

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

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

214−

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

216− faff 2000000 rs

217

168

218 (replyTo, reply) <- newChan :: Process (SendPort Bool, ReceivePort Bool)

169 (replyTo, reply) <- newChan :: Process (SendPort Bool, ReceivePort Bool)

219 mxNotify (MxSpawned pid, replyTo)

170 mxNotify (MxSpawned pid, replyTo)

220 mxNotify (MxProcessDied pid DiedNormal, replyTo)

171 mxNotify (MxProcessDied pid DiedNormal, replyTo)

221

172

222− seenAlive <- receiveChanTimeout 10000000 reply

173+ seenAlive <- receiveChan reply

223− seenDead <- receiveChanTimeout 10000000 reply

174+ seenDead <- receiveChan reply

224−

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

226− kill timer "test-complete"

227− kill agentPid "test-complete"

228− where

229− faff delay port = do

230− res <- receiveChanTimeout delay port

231− unless (isNothing res) $ pause delay

232−

233−testMxRegEvents :: TestResult () -> Process ()

234−testMxRegEvents result = do

235− {- This test only deals with the local case, to ensure that we are being

236− notified in the expected order - the remote cases related to the

237− behaviour of the node controller are contained in the CH test suite. -}

238− ensure (stash result ()) $ do

239− let label = "testMxRegEvents"

240− let agentLabel = "mxRegEvents-agent"

241− let delay = 1000000

242− (regChan, regSink) <- newChan

243− (unRegChan, unRegSink) <- newChan

244− agent <- mxAgent (MxAgentId agentLabel) () [

245− mxSink $ \ev -> do

246− case ev of

247− MxRegistered pid label'

248− | label' == label -> liftMX $ sendChan regChan (label', pid)

249− MxUnRegistered pid label'

250− | label' == label -> liftMX $ sendChan unRegChan (label', pid)

251− _ -> return ()

252− mxReady

253− ]

254−

255− p1 <- spawnLocal expect

256− p2 <- spawnLocal expect

257−

258− register label p1

259− reg1 <- receiveChanTimeout delay regSink

260− reg1 `shouldBe` equalTo (Just (label, p1))

261−

262− unregister label

263− unreg1 <- receiveChanTimeout delay unRegSink

264− unreg1 `shouldBe` equalTo (Just (label, p1))

265−

266− register label p2

267− reg2 <- receiveChanTimeout delay regSink

268− reg2 `shouldBe` equalTo (Just (label, p2))

269−

270− reregister label p1

271− unreg2 <- receiveChanTimeout delay unRegSink

272− unreg2 `shouldBe` equalTo (Just (label, p2))

273−

274− reg3 <- receiveChanTimeout delay regSink

275− reg3 `shouldBe` equalTo (Just (label, p1))

276−

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

278−

279−testMxRegMon :: LocalNode -> TestResult () -> Process ()

280−testMxRegMon remoteNode result = do

281− ensure (stash result ()) $ do

282− -- ensure that when a registered process dies, we get a notification that

283− -- it has been unregistered as well as seeing the name get removed

284− let label1 = "aaaaa"

285− let label2 = "bbbbb"

286− let isValid l = l == label1 || l == label2

287− let agentLabel = "mxRegMon-agent"

288− let delay = 1000000

289− (regChan, regSink) <- newChan

290− (unRegChan, unRegSink) <- newChan

291− agent <- mxAgent (MxAgentId agentLabel) () [

292− mxSink $ \ev -> do

293− case ev of

294− MxRegistered pid label

295− | isValid label -> liftMX $ sendChan regChan (label, pid)

296− MxUnRegistered pid label

297− | isValid label -> liftMX $ sendChan unRegChan (label, pid)

298− _ -> return ()

299− mxReady

300− ]

301−

302− (sp, rp) <- newChan

303− liftIO $ forkProcess remoteNode $ do

304− getSelfPid >>= sendChan sp

305− expect :: Process ()

306−

307− p1 <- receiveChan rp

308−

309− register label1 p1

310− reg1 <- receiveChanTimeout delay regSink

311− reg1 `shouldBe` equalTo (Just (label1, p1))

312−

313− register label2 p1

314− reg2 <- receiveChanTimeout delay regSink

315− reg2 `shouldBe` equalTo (Just (label2, p1))

316−

317− n1 <- whereis label1

318− n1 `shouldBe` equalTo (Just p1)

319−

320− n2 <- whereis label2

321− n2 `shouldBe` equalTo (Just p1)

322−

323− kill p1 "goodbye"

324

175

325− unreg1 <- receiveChanTimeout delay unRegSink

176+ stash result $ seenAlive && seenDead

326− unreg2 <- receiveChanTimeout delay unRegSink

327−

328− let evts = [unreg1, unreg2]

329− -- we can't rely on the order of the values in the node controller's

330− -- map (it's either racy to do so, or no such guarantee exists for Data.Map),

331− -- so we simply verify that we received the un-registration events we expect

332− evts `shouldContain` (Just (label1, p1))

333− evts `shouldContain` (Just (label2, p1))

334−

335− kill agent "test-complete"

336−

337−ensure :: Process () -> Process () -> Process ()

338−ensure = flip finally

339−

340−testNSend :: (String -> () -> Process ())

341− -> Maybe LocalNode

342− -> Process ()

343−testNSend op n = do

344− us <- getSelfPid

345− let delay = 5000000

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

347− let isValid = isValidLabel n label

348−

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

356−

357− op label ()

358−

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

364−

365− where

366− isValidLabel nd l1 l2

367− | l2 == l2 = True

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

369− | otherwise = False

370−

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−

405− -- initiate a send

406− op ch' ()

407−

408− -- verify the management event

409− sent <- receiveChanTimeout 5000000 sink

410− 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

416−

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

418−testMxSend mNode label test = do

419− us <- getSelfPid

420− (sp, rp) <- newChan

421− (chan, sink) <- newChan

422− agent <- mxAgent (MxAgentId $ agentLabel us) () [

423− mxSink $ \ev -> do

424− case ev of

425− m@(MxSentToPort _ cid _)

426− | cid /= sendPortId chan

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

428− m@(MxSent _ fromPid _)

429− | fromPid == us -> liftMX $ sendChan chan m

430− m@(MxSentToName _ fromPid _)

431− | fromPid == us -> liftMX $ sendChan chan m

432− m@(MxRegistered _ name)

433− | name == label -> liftMX $ sendChan chan m

434− _ -> return ()

435− mxReady

436− ]

437−

438− case mNode of

439− Nothing -> void $ spawnLocal (proc sp)

440− Just remoteNode -> void $ liftIO $ forkProcess remoteNode $ proc sp

441−

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−

449− case res of

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

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

452−

453−

454− where

455− 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

466

177

467tests :: TestTransport -> IO [Test]

178tests :: TestTransport -> IO [Test]

468tests TestTransport{..} = do

179tests TestTransport{..} = do

469 node1 <- newLocalNode testTransport initRemoteTable

180 node1 <- newLocalNode testTransport initRemoteTable

470− node2 <- newLocalNode testTransport initRemoteTable

471 return [

181 return [

472− testGroup "MxAgents" [

182+ testGroup "Mx Agents" [

473− testCase "EventHandling"

183+ testCase "Event Handling"

474− (delayedAssertion

184+ (delayedAssertion

475− "expected True, but events where not as expected"

185+ "expected True, but events where not as expected"

476− node1 True testAgentEventHandling)

186+ node1 True testAgentEventHandling)

477− , testCase "InterAgentBroadcast"

187+ , testCase "Inter-Agent Broadcast"

478− (delayedAssertion

188+ (delayedAssertion

479− "expected (), but no broadcast was received"

189+ "expected (), but no broadcast was received"

480− node1 (Just ()) testAgentBroadcast)

190+ node1 () testAgentBroadcast)

481− , testCase "AgentMailboxHandling"

191+ , testCase "Agent Mailbox Handling"

482− (delayedAssertion

192+ (delayedAssertion

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

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

484− node1 (Just ()) testAgentMailboxHandling)

194+ node1 (Just ()) testAgentMailboxHandling)

485− , testCase "AgentDualInputHandling"

195+ , testCase "Agent Dual Input Handling"

486− (delayedAssertion

196+ (delayedAssertion

487− "expected sum = 15, but the result was Nothing"

197+ "expected sum = 15, but the result was Nothing"

488− node1 (Just 15 :: Maybe Int) testAgentDualInput)

198+ node1 (Just 15 :: Maybe Int) testAgentDualInput)

489− , testCase "AgentInputPrioritisation"

199+ , testCase "Agent Input Prioritisation"

490− (delayedAssertion

200+ (delayedAssertion

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

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

492− node1 (sort ["first", "second",

202+ node1 (sort ["first", "second",

493− "third", "fourth",

203+ "third", "fourth",

494− "fifth"]) testAgentPrioritisation)

204+ "fifth"]) testAgentPrioritisation)

495− ]

496− , testGroup "MxEvents" [

497− testCase "NameRegistrationEvents"

498− (delayedAssertion

499− "expected registration events to map to the correct ProcessId"

500− node1 () testMxRegEvents)

501− , testCase "PostDeathNameUnRegistrationEvents"

502− (delayedAssertion

503− "expected process deaths to result in unregistration events"

504− node1 () (testMxRegMon node2))

505− , testGroup "SendEvents" $ buildTestCases node1 node2

506− ]

205+ ]]

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

528− ]

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

542− ]

distributed-process-tests/src/Control/Distributed/Process/Tests/Receive.hs+8−7

@@ -10,12 +10,13 @@ import Network.Transport.Test (TestTransport(..))

11import Network.Transport (Transport)

12import Control.Distributed.Process

13+import Control.Distributed.Process.Closure

13import Control.Distributed.Process.Node

14import Control.Distributed.Process.Node

15import Control.Monad

16import Control.Monad

17import Test.HUnit (Assertion, (@?=))

18import Test.HUnit (Assertion, (@?=))

18−import Test.Framework (Test)

19+import Test.Framework (Test, defaultMain)

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

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

21-- Tests:

22-- Tests:

@@ -42,7 +43,7 @@ recTest2 :: ReceivePort ()

42 -> SendPort String

43 -> SendPort String

43 -> ReceivePort String -> ReceivePort String

44 -> ReceivePort String -> ReceivePort String

44 -> Process ()

45 -> Process ()

45−recTest2 wait sync r1 _ = do

46+recTest2 wait sync r1 r2 = do

46 forever $ do

47 forever $ do

47 receiveChan wait

48 receiveChan wait

48 r <- receiveWait

49 r <- receiveWait

@@ -55,7 +56,7 @@ recTest3 :: ReceivePort ()

55 -> SendPort String

56 -> SendPort String

56 -> ReceivePort String -> ReceivePort String

57 -> ReceivePort String -> ReceivePort String

57 -> Process ()

58 -> Process ()

58−recTest3 wait sync r1 _ = do

59+recTest3 wait sync r1 r2 = do

59 forever $ do

60 forever $ do

60 receiveChan wait

61 receiveChan wait

61 r <- receiveWait

62 r <- receiveWait

@@ -68,7 +69,7 @@ recTest4 :: ReceivePort ()

68 -> SendPort String

69 -> SendPort String

69 -> ReceivePort String -> ReceivePort String

70 -> ReceivePort String -> ReceivePort String

70 -> Process ()

71 -> Process ()

71−recTest4 wait sync r1 _ = do

72+recTest4 wait sync r1 r2 = do

72 forever $ do

73 forever $ do

73 receiveChan wait

74 receiveChan wait

74 r <- receiveWait

75 r <- receiveWait

@@ -82,11 +83,11 @@ master :: Process ()

82master = do

83master = do

83 (waits,waitr) <- newChan

84 (waits,waitr) <- newChan

84 (syncs,syncr) <- newChan

85 (syncs,syncr) <- newChan

85− let go expected = do

86+ let go expect = do

86 sendChan waits ()

87 sendChan waits ()

87 r <- receiveChan syncr

88 r <- receiveChan syncr

88− liftIO $ print (r, expected, r == expected)

89+ liftIO $ print (r,expect, r == expect)

89− liftIO $ r @?= expected

90+ liftIO $ r @?= expect

91 liftIO $ putStrLn "---- Test 1 ----"

92 liftIO $ putStrLn "---- Test 1 ----"

92 (s1,r1) <- newChan

93 (s1,r1) <- newChan

distributed-process-tests/tests/runInMemory.hs+1

@@ -8,6 +8,7 @@ import Network.Transport.Test (TestTransport(..))

8import Network.Transport.InMemory

9import Test.Framework (defaultMainWithArgs)

11+import Control.Concurrent (threadDelay)

11import System.Environment (getArgs)

12import System.Environment (getArgs)

13main :: IO ()

14main :: IO ()

distributed-process-tests/tests/runTCP.hs+2−2

@@ -10,7 +10,6 @@ import Network.Transport.TCP

10 ( createTransportExposeInternals

11 , TransportInternals(socketBetween)

12 , defaultTCPParameters

13− , defaultTCPAddr

14 , TCPParameters(..)

13 , TCPParameters(..)

15 )

14 )

16import Test.Framework (defaultMainWithArgs)

15import Test.Framework (defaultMainWithArgs)

@@ -25,7 +24,8 @@ main = do

25 hSetBuffering stdout LineBuffering

24 hSetBuffering stdout LineBuffering

26 hSetBuffering stderr LineBuffering

25 hSetBuffering stderr LineBuffering

27 Right (transport, internals) <-

26 Right (transport, internals) <-

28− createTransportExposeInternals (defaultTCPAddr "127.0.0.1" "8080")

27+ createTransportExposeInternals "127.0.0.1" "8080"

28+ (\sn -> ("127.0.0.1", sn))

29 defaultTCPParameters { transportConnectTimeout = Just 3000000 }

30 ts <- tests TestTransport

31 { testTransport = transport

distributed-process.cabal+2−2

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

6License-File: LICENSE

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

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

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

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

10Stability: experimental

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

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

@@ -21,7 +21,7 @@ Description: This is an implementation of Cloud Haskell, as described in

22 You will probably also want to install a Cloud Haskell backend such

23 as distributed-process-simplelocalnet.

24−Tested-With: GHC==7.10.3 GHC==8.0.2 GHC==8.2.2 GHC==8.4.4

24+Tested-With: GHC==7.6.3 GHC==7.8.4 GHC==7.10.3 GHC==8.0.1

25Category: Control

26extra-source-files: ChangeLog

src/Control/Distributed/Process/Internal/Primitives.hs+42−57

@@ -76,7 +76,6 @@ module Control.Distributed.Process.Internal.Primitives

76 , unlink

77 , monitor

78 , unmonitor

79− , unmonitorAsync

80 , withMonitor

79 , withMonitor

81 , withMonitor_

80 , withMonitor_

82 -- * Logging

81 -- * Logging

@@ -254,15 +253,17 @@ send them msg = do

254 node = processNode proc

253 node = processNode proc

255 nodeId = localNodeId node

254 nodeId = localNodeId node

256 destNode = (processNodeId them) in do

255 destNode = (processNodeId them) in do

256+ case destNode == nodeId of

257+ True -> sendLocal them msg

258+ False -> liftIO $ sendMessage (processNode proc)

259+ (ProcessIdentifier (processId proc))

260+ (ProcessIdentifier them)

261+ NoImplicitReconnect

262+ msg

263+ -- We do not fire the trace event until after the sending is complete;

264+ -- In the remote case, 'sendMessage' can block in the networking stack.

257 liftIO $ traceEvent (localEventBus node)

265 liftIO $ traceEvent (localEventBus node)

258 (MxSent them us (createUnencodedMessage msg))

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

266

267

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

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

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

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

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

282--

283--

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

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

284usend them msg = do

285usend them msg = do

285− proc <- ask

286+ here <- getSelfNode

286 let there = processNodeId them

287 let there = processNodeId them

287− let (us, node) = (processId proc, processNode proc)

288+ if here == there

288− let msg' = wrapMessage msg

289− liftIO $ traceEvent (localEventBus node) (MxSent them us msg')

290− if localNodeId (processNode proc) == there

291 then sendLocal them msg

289 then sendLocal them msg

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

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

293 (createMessage msg)

291 (createMessage msg)

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

307-- Channels --

305-- Channels --

308--------------------------------------------------------------------------------

306--------------------------------------------------------------------------------

309

307

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

308+-- | Create a new typed channel

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

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

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

318newChan = do

310newChan = do

319 proc <- ask

311 proc <- ask

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

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

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

342sendChan (SendPort cid) msg = do

334sendChan (SendPort cid) msg = do

343 proc <- ask

335 proc <- ask

344− let node = processNode proc

345− pid = processId proc

346− us = localNodeId node

336+ let node = localNodeId (processNode proc)

347− them = processNodeId (sendPortProcessId cid) in do

337+ destNode = processNodeId (sendPortProcessId cid) in do

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

349− case them == us of

338+ case destNode == node of

350 True -> sendChanLocal cid msg

339 True -> sendChanLocal cid msg

351 False -> do

340 False -> do

352− liftIO $ sendBinary node

341+ liftIO $ sendBinary (processNode proc)

353− (ProcessIdentifier pid)

342+ (ProcessIdentifier (processId proc))

354 (SendPortIdentifier cid)

343 (SendPortIdentifier cid)

355 NoImplicitReconnect

344 NoImplicitReconnect

356 msg

345 msg

@@ -414,8 +403,7 @@ receiveWait ms = do

414 mProc <- liftIO $ dequeue queue Blocking (map unMatch ms)

403 mProc <- liftIO $ dequeue queue Blocking (map unMatch ms)

415 case mProc of

404 case mProc of

416 Just proc' -> proc'

405 Just proc' -> proc'

417− Nothing -> die $ "System Invariant Violation: CQueue.hs returned `Nothing` "

406+ Nothing -> error "Well... That wasn't supposed to happen! o_O"

418− ++ "in the absence of a timeout value."

419

407

420-- | Like 'receiveWait' but with a timeout.

408-- | Like 'receiveWait' but with a timeout.

421--

409--

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

494 us = processId proc

482 us = processId proc

495 nid = localNodeId node

483 nid = localNodeId node

496 destNode = (processNodeId them) in do

484 destNode = (processNodeId them) in do

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

498− if destNode == nid

485+ case destNode == nid of

499− then sendCtrlMsg Nothing (LocalSend them msg)

486+ True -> sendCtrlMsg Nothing (LocalSend them msg)

500− else liftIO $ sendPayload (processNode proc)

487+ False -> liftIO $ sendPayload (processNode proc)

501− (ProcessIdentifier (processId proc))

488+ (ProcessIdentifier (processId proc))

502− (ProcessIdentifier them)

489+ (ProcessIdentifier them)

503− NoImplicitReconnect

490+ NoImplicitReconnect

504− (messageToPayload msg)

491+ (messageToPayload msg)

492+ -- We do not fire the trace event until after the sending is complete;

493+ -- In the remote case, 'sendMessage' can block in the networking stack.

494+ liftIO $ traceEvent (localEventBus node)

495+ (MxSent them us msg)

505

496

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

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

507--

498--

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

515 us = processId proc

506 us = processId proc

516 nid = localNodeId node

507 nid = localNodeId node

517 destNode = (processNodeId them) in do

508 destNode = (processNodeId them) in do

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

519− if destNode == nid

509+ case destNode == nid of

520− then sendCtrlMsg Nothing (LocalSend them msg)

510+ True -> sendCtrlMsg Nothing (LocalSend them msg)

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

511+ False -> sendCtrlMsg (Just destNode) $ UnreliableSend (processLocalId them)

512+ msg

513+ -- We do not fire the trace event until after the sending is complete;

514+ -- In the remote case, 'sendCtrlMsg' can block in the networking stack.

515+ liftIO $ traceEvent (localEventBus node)

516+ (MxSent them us msg)

522

517

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

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

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

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

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

1227

1222

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

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

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

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

1230−nsend label msg = do

1225+nsend label msg =

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

1226+ sendCtrlMsg Nothing (NamedSend label (createUnencodedMessage msg))

1236

1227

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

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

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

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

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

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

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

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

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

1246nsendRemote nid label msg = do

1237nsendRemote nid label msg = do

1247− proc <- ask

1238+ here <- getSelfNode

1248− let us = processId proc

1249− let node = processNode proc

1250− if localNodeId node == nid

1251− then nsend label msg

1239+ if here == nid 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))

1240+ else sendCtrlMsg (Just nid) (NamedSend label (createMessage msg))

1256

1241

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

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

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

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

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

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

76--

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

78--

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

113-- [Management Data API]

79-- [Management Data API]

114--

80--

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

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

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

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

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

179-- > monitorNames2 = getSelfPid >>= mxNotify

145-- > monitorNames2 = getSelfPid >>= mxNotify

180--

146--

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

148+--

181-- [Performance, Stablity and Scalability]

149-- [Performance, Stablity and Scalability]

182--

150--

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

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

src/Control/Distributed/Process/Management/Internal/Trace/Primitives.hs+1−2

@@ -174,6 +174,5 @@ withRegisteredTracer act = do

174 -- is not considered cool in later versions of MonadFail

175 case mTP of

176 Just p' -> act p'

177− Nothing -> die $ "System Invariant Violation: Tracer Process "

177+ Nothing -> die "Initial Tracer Name Not Registered"

178− ++ "Name Not Found (whereis tracer.initial)"

179 (Just p) -> act p

178 (Just p) -> act p

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

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

21import Control.Distributed.Process.Internal.Types

22 ( Process

23 , ProcessId

24− , SendPortId

25 , Message

24 , Message

26 , DiedReason

25 , DiedReason

27 , NodeId

26 , NodeId

@@ -31,7 +30,6 @@ import qualified Control.Monad.State as ST

31 ( MonadState

30 ( MonadState

32 , StateT

31 , StateT

33 )

32 )

34−import Control.Monad.Fix (MonadFix)

35import Data.Binary

33import Data.Binary

36import Data.Typeable (Typeable)

34import Data.Typeable (Typeable)

37import GHC.Generics

35import GHC.Generics

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

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

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

59 | MxSent ProcessId ProcessId Message

57 | MxSent ProcessId ProcessId Message

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

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

65 | MxReceived ProcessId Message

59 | MxReceived ProcessId Message

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

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

69 | MxConnected ConnectionId EndPointAddress

61 | MxConnected ConnectionId EndPointAddress

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

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

71 | MxDisconnected ConnectionId EndPointAddress

63 | MxDisconnected ConnectionId EndPointAddress

@@ -120,7 +112,6 @@ newtype MxAgent s a =

120 } deriving ( Functor

112 } deriving ( Functor

121 , Monad

113 , Monad

122 , MonadIO

114 , MonadIO

123− , MonadFix

124 , ST.MonadState (MxAgentState s)

115 , ST.MonadState (MxAgentState s)

125 , Typeable

116 , Typeable

126 , Applicative

117 , Applicative

src/Control/Distributed/Process/Node.hs+19−38

@@ -28,12 +28,9 @@ import Data.Binary (decode)

28import Data.Map (Map)

29import qualified Data.Map as Map

30 ( empty

31− , delete

32 , toList

31 , toList

33 , fromList

32 , fromList

34− , insert

33+ , filter

35− , lookup

36− , partition

37 , partitionWithKey

34 , partitionWithKey

38 , elems

35 , elems

39 , size

36 , size

@@ -186,9 +183,6 @@ import Control.Distributed.Process.Internal.Types

186import Control.Distributed.Process.Management.Internal.Agent

183import Control.Distributed.Process.Management.Internal.Agent

187 ( mxAgentController

184 ( mxAgentController

188 )

185 )

189−import Control.Distributed.Process.Management.Internal.Types

190− ( MxEvent(..)

191− )

192import qualified Control.Distributed.Process.Management.Internal.Trace.Remote as Trace

186import qualified Control.Distributed.Process.Management.Internal.Trace.Remote as Trace

193 ( remoteTable

187 ( remoteTable

194 )

188 )

@@ -201,6 +195,9 @@ import Control.Distributed.Process.Management.Internal.Trace.Types

201 , traceLogFmt

195 , traceLogFmt

202 , enableTrace

196 , enableTrace

203 )

197 )

198+import Control.Distributed.Process.Management.Internal.Types

199+ ( MxEvent(..)

200+ )

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

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

205import Control.Distributed.Process.Internal.Messaging

202import Control.Distributed.Process.Internal.Messaging

206 ( sendBinary

203 ( sendBinary

@@ -213,9 +210,6 @@ import Control.Distributed.Process.Internal.Primitives

213 , match

210 , match

214 , sendChan

211 , sendChan

215 , unwrapMessage

212 , unwrapMessage

216− , monitor

217− , unmonitorAsync

218− , getSelfNode

219 , SayMessage(..)

213 , SayMessage(..)

220 )

214 )

221import Control.Distributed.Process.Internal.Types (SendPort, Tracer(..))

215import Control.Distributed.Process.Internal.Types (SendPort, Tracer(..))

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

492data IncomingTarget =

486data IncomingTarget =

493 Uninit

487 Uninit

494 | ToProc ProcessId (Weak (CQueue Message))

488 | ToProc ProcessId (Weak (CQueue Message))

495− | ToChan SendPortId TypedChannel

489+ | ToChan TypedChannel

496 | ToNode

490 | ToNode

497

491

498data ConnectionState = ConnectionState {

492data ConnectionState = ConnectionState {

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

547 enqueue queue msg -- 'enqueue' is strict

541 enqueue queue msg -- 'enqueue' is strict

548 trace node (MxReceived pid msg)

542 trace node (MxReceived pid msg)

549 go st

543 go st

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

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

551 mChan <- deRefWeak chan'

545 mChan <- deRefWeak chan'

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

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

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

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

554− forM_ mChan $ \chan -> do

548+ forM_ mChan $ \chan -> atomically $

555− msg' <- atomically $ do

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

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

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

558− writeTQueue chan msg

550+ writeTQueue chan $! decode (BSL.fromChunks payload)

559− return msg

560− trace node $ MxReceivedPort chId $ unsafeCreateUnencodedMessage msg'

561 go st

551 go st

562 Just (_, ToNode) -> do

552 Just (_, ToNode) -> do

563 let ctrlMsg = decode . BSL.fromChunks $ payload

553 let ctrlMsg = decode . BSL.fromChunks $ payload

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

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

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

585 case mChannel of

575 case mChannel of

586 Just channel ->

576 Just channel ->

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

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

588 Nothing ->

578 Nothing ->

589 invalidRequest cid st $

579 invalidRequest cid st $

590 "incoming attempt to connect to unknown channel of"

580 "incoming attempt to connect to unknown channel of"

@@ -913,11 +903,7 @@ ncEffectDied ident reason = do

913

903

914 modify' $ (links ^= unaffectedLinks') . (monitors ^= unaffectedMons')

904 modify' $ (links ^= unaffectedLinks') . (monitors ^= unaffectedMons')

915

905

916− -- we now consider all labels for this identifier unregistered

917− let toDrop pid = not $ ident `impliesDeathOf` ProcessIdentifier pid

906+ modify' $ registeredHere ^: Map.filter (\pid -> not $ ident `impliesDeathOf` ProcessIdentifier pid)

918− (keepNames, dropNames) <- Map.partition toDrop <$> gets (^. registeredHere)

919− mapM_ (\(p, l) -> liftIO $ trace node (MxUnRegistered l p)) (Map.toList dropNames)

920− modify' $ registeredHere ^= keepNames

921

907

922 remaining <- fmap Map.toList (gets (^. registeredOnNodes)) >>=

908 remaining <- fmap Map.toList (gets (^. registeredOnNodes)) >>=

923 mapM (\(pid,nidlist) ->

909 mapM (\(pid,nidlist) ->

@@ -969,11 +955,7 @@ ncEffectRegister from label atnode mPid reregistration = do

969 do modify' $ registeredHereFor label ^= mPid

955 do modify' $ registeredHereFor label ^= mPid

970 updateRemote node currentVal mPid

956 updateRemote node currentVal mPid

971 case mPid of

957 case mPid of

972− (Just p) -> do

973− if reregistration

974− then liftIO $ trace node (MxUnRegistered (fromJust currentVal) label)

975− else return ()

976− liftIO $ trace node (MxRegistered p label)

958+ (Just p) -> liftIO $ trace node (MxRegistered p label)

977 Nothing -> liftIO $ trace node (MxUnRegistered (fromJust currentVal) label)

959 Nothing -> liftIO $ trace node (MxUnRegistered (fromJust currentVal) label)

978 newVal <- gets (^. registeredHereFor label)

960 newVal <- gets (^. registeredHereFor label)

979 ncSendToProcess from $ unsafeCreateUnencodedMessage $

961 ncSendToProcess from $ unsafeCreateUnencodedMessage $

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

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

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

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

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

1056 ch <- deRefWeak chan'

1038 ch <- deRefWeak chan'

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

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

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

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

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

1041+ deliverChan (UnencodedMessage _ raw) chan' =

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

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

1061− trace n (MxReceivedPort p $ unsafeCreateUnencodedMessage raw)

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

1043+ deliverChan (EncodedMessage _ _) _ =

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

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

1064 error "invalid local channel delivery"

1045 error "invalid local channel delivery"

1065

1046

@@ -1089,7 +1070,7 @@ ncEffectGetInfo from pid =

1089 case mProc of

1070 case mProc of

1090 Nothing -> dispatch (isLocal node (ProcessIdentifier from))

1071 Nothing -> dispatch (isLocal node (ProcessIdentifier from))

1091 from (ProcessInfoNone DiedUnknownId)

1072 from (ProcessInfoNone DiedUnknownId)

1092− Just proc -> do

1073+ Just proc -> do

1093 itsLinks <- Set.map fst . BiMultiMap.lookupBy1st them <$>

1074 itsLinks <- Set.map fst . BiMultiMap.lookupBy1st them <$>

1094 gets (^. links)

1075 gets (^. links)

1095 itsMons <- BiMultiMap.lookupBy1st them <$> gets (^. monitors)

1076 itsMons <- BiMultiMap.lookupBy1st them <$> gets (^. monitors)

@@ -1103,8 +1084,8 @@ ncEffectGetInfo from pid =

1103 infoNode = (processNodeId pid)

1084 infoNode = (processNodeId pid)

1104 , infoRegisteredNames = reg

1085 , infoRegisteredNames = reg

1105 , infoMessageQueueLength = size

1086 , infoMessageQueueLength = size

1106− , infoMonitors = Set.toList itsMons

1087+ , infoMonitors = Set.toList itsMons

1107− , infoLinks = Set.toList itsLinks

1088+ , infoLinks = Set.toList itsLinks

1108 }

1089 }

1109 where dispatch :: (Serializable a)

1090 where dispatch :: (Serializable a)

1110 => Bool

1091 => Bool

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

@@ -32,24 +32,7 @@

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

52−-- You have been warned!

35+-- behaviour of your application. You have been warned!

53--

36--

54-- This module is exported so that you can replace the use of Cloud Haskell's

37-- This module is exported so that you can replace the use of Cloud Haskell's

55-- /safe/ messaging primitives. If you want to use both variants, then you can

38-- /safe/ messaging primitives. If you want to use both variants, then you can

@@ -72,12 +55,7 @@ import Control.Distributed.Process.Internal.Messaging

72 , sendBinary

55 , sendBinary

73 , sendCtrlMsg

56 , sendCtrlMsg

74 )

57 )

75−import Control.Distributed.Process.Management.Internal.Types

58+

76− ( MxEvent(..)

77− )

78−import Control.Distributed.Process.Management.Internal.Trace.Types

79− ( traceEvent

80− )

81import Control.Distributed.Process.Internal.Types

59import Control.Distributed.Process.Internal.Types

82 ( ProcessId(..)

60 ( ProcessId(..)

83 , NodeId(..)

61 , NodeId(..)

@@ -101,48 +79,34 @@ import Control.Monad.Reader (ask)

101

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

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

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

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

104−nsend label msg = do

82+nsend label msg =

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

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

112

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

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

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

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

115nsendRemote nid label msg = do

87nsendRemote nid label msg = do

116 proc <- ask

88 proc <- ask

117− let us = processId proc

118− let node = processNode proc

119− if localNodeId node == nid

89+ if localNodeId (processNode proc) == nid

120 then nsend label msg

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

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

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

127

128-- | Send a message

93-- | Send a message

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

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

130send them msg = do

95send them msg = do

131 proc <- ask

96 proc <- ask

132 let node = localNodeId (processNode proc)

97 let node = localNodeId (processNode proc)

133− destNode = (processNodeId them)

98+ destNode = (processNodeId them) in do

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

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

140− then sendCtrlMsg Nothing $ LocalSend them msg'

109+ sendCtrlMsg Nothing $ LocalSend pid (unsafeCreateUnencodedMessage msg')

141− else liftIO $ sendMessage (processNode proc)

142− (ProcessIdentifier (processId proc))

143− (ProcessIdentifier them)

144− NoImplicitReconnect

145− msg

146

110

147-- | Send a message unreliably.

111-- | Send a message unreliably.

148--

112--

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

157usend them msg = do

121usend them msg = do

158 proc <- ask

122 proc <- ask

159 let there = processNodeId them

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

164 if localNodeId (processNode proc) == there

124 if localNodeId (processNode proc) == there

165− then sendCtrlMsg Nothing $ LocalSend them msg'

125+ then sendCtrlMsg Nothing $

126+ LocalSend them (unsafeCreateUnencodedMessage msg)

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

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

167 (createMessage msg)

128 (createMessage msg)

168

129

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−

177-- | Send a message on a typed channel

130-- | Send a message on a typed channel

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

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

179sendChan (SendPort cid) msg = do

132sendChan (SendPort cid) msg = do

180 proc <- ask

133 proc <- ask

181− let node = processNode proc

182− pid = processId proc

183− us = localNodeId node

134+ let node = localNodeId (processNode proc)

184− them = processNodeId (sendPortProcessId cid)

135+ destNode = processNodeId (sendPortProcessId cid) in do

185− msg' = wrapMessage msg in do

136+ case destNode == node of

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

187− if them == us

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

137+ True -> unsafeSendChanLocal cid msg

138+ False -> do

189− else liftIO $ sendBinary node

139+ liftIO $ sendBinary (processNode proc)

190− (ProcessIdentifier pid)

140+ (ProcessIdentifier (processId proc))

191− (SendPortIdentifier cid)

141+ (SendPortIdentifier cid)

192− NoImplicitReconnect

142+ NoImplicitReconnect

193− msg

143+ msg

194 where

144 where

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

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

146+ unsafeSendChanLocal spId msg' =

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

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

197

148

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

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

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

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

stack-ghc-7.10.3.yaml+1−5

@@ -3,18 +3,14 @@ resolver: nightly-2016-03-08

3packages:

4- '.'

5- distributed-process-tests/

6−- location:

7− git: https://github.com/haskell-distributed/network-transport-tcp.git

8− commit: d87d6f55697a94a1fbf211ff9c1bb769a1e129cd

9− extra-dep: true

11extra-deps:

7extra-deps:

12- rank1dynamic-0.4.0

8- rank1dynamic-0.4.0

13- distributed-static-0.3.8

9- distributed-static-0.3.8

14- network-transport-0.5.2

10- network-transport-0.5.2

11+- network-transport-tcp-0.6.0

15- network-transport-inmemory-0.5.2

12- network-transport-inmemory-0.5.2

16- rematch-0.2.0.0

13- rematch-0.2.0.0

17−- exceptions-0.8.2.1 # test dependency

19flags:

15flags:

20 distributed-process-tests:

16 distributed-process-tests:

stack-ghc-8.0.2.yaml+2−8

@@ -1,22 +1,16 @@

1−resolver: lts-9.21

1+resolver: lts-9.0

2−# added -location for network-transport-tcp-0.6.0

3−# since c7fd79 broke the client facing API for createTransport :/

5packages:

3packages:

6- '.'

4- '.'

7- distributed-process-tests/

5- distributed-process-tests/

8−- location:

9− git: https://github.com/haskell-distributed/network-transport-tcp.git

10− commit: d87d6f55697a94a1fbf211ff9c1bb769a1e129cd

11− extra-dep: true

13extra-deps:

7extra-deps:

14- rank1dynamic-0.4.0

8- rank1dynamic-0.4.0

15- distributed-static-0.3.8

9- distributed-static-0.3.8

16- network-transport-0.5.2

10- network-transport-0.5.2

11+- network-transport-tcp-0.6.0

17- network-transport-inmemory-0.5.2

12- network-transport-inmemory-0.5.2

18- rematch-0.2.0.0

13- rematch-0.2.0.0

19−# - exceptions-0.8.2.1 # test dependency

21flags:

15flags:

22 distributed-process-tests:

16 distributed-process-tests:

stack-ghc-8.2.2.yamldeleted−21

@@ -1,21 +0,0 @@

1−resolver: lts-11.22

2−

3−packages:

4−- '.'

5−- distributed-process-tests/

6−- location:

7− git: https://github.com/haskell-distributed/network-transport-tcp.git

8− commit: d87d6f55697a94a1fbf211ff9c1bb769a1e129cd

9− extra-dep: true

10−

11−extra-deps:

12−- rank1dynamic-0.4.0

13−- distributed-static-0.3.8

14−- network-transport-0.5.2

15−- network-transport-inmemory-0.5.2

16−- rematch-0.2.0.0

17−# - exceptions-0.8.2.1 # test dependency

18−

19−flags:

20− distributed-process-tests:

21− tcp: true

stack-ghc-8.4.4.yamldeleted−20

@@ -1,20 +0,0 @@

1−resolver: lts-12.18

2−

3−packages:

4−- '.'

5−- distributed-process-tests/

6−- location:

7− git: https://github.com/haskell-distributed/network-transport-tcp.git

8− commit: d87d6f55697a94a1fbf211ff9c1bb769a1e129cd

9− extra-dep: true

10−

11−extra-deps:

12−- rank1dynamic-0.4.0

13−- distributed-static-0.3.8

14−- network-transport-0.5.2

15−- network-transport-inmemory-0.5.2

16−- rematch-0.2.0.0

17−

18−flags:

19− distributed-process-tests:

20− tcp: true

stack-nightly.yaml+3−3

@@ -1,4 +1,4 @@

1−resolver: nightly

1+resolver: nightly-2018-11-10

3packages:

4- '.'

@@ -9,11 +9,11 @@ packages:

9 extra-dep: true

10- location:

11 git: https://github.com/haskell-distributed/network-transport-tcp.git

12− commit: d58250e2388b86a1865ac4139373cbdc106e72e1

12+ commit: d87d6f55697a94a1fbf211ff9c1bb769a1e129cd

13 extra-dep: true

14- location:

15 git: https://github.com/haskell-distributed/network-transport-inmemory.git

16− commit: 0ce97924f15a8c1570b2355151834305c9bd2a28

16+ commit: b7a10ffc07a77f4fe6affbf8abe535b11933a01a

17 extra-dep: true

19extra-deps:

stack.yaml+2−7

@@ -1,19 +1,14 @@

1−resolver: lts-12.18 # Use GHC 8.4.4

1+resolver: nightly-2017-08-20

2−

3−# TODO: we can enable nightly-2018-11-11 once PR #319 is merged

5packages:

3packages:

6- '.'

4- '.'

7- distributed-process-tests/

5- distributed-process-tests/

8−- location:

9− git: https://github.com/haskell-distributed/network-transport-tcp.git

10− commit: d87d6f55697a94a1fbf211ff9c1bb769a1e129cd

11− extra-dep: true

13extra-deps:

7extra-deps:

14- rank1dynamic-0.4.0

8- rank1dynamic-0.4.0

15- distributed-static-0.3.8

9- distributed-static-0.3.8

16- network-transport-0.5.2

10- network-transport-0.5.2

11+- network-transport-tcp-0.6.0

17- network-transport-inmemory-0.5.2

12- network-transport-inmemory-0.5.2

18- rematch-0.2.0.0

13- rematch-0.2.0.0