c48602c · distributed-process

git.iko.soy/distributed-process

on master help sign in

Merge branch 'master' of github.com:haskell-distributed/distributed-process

c48602cEdsko de Vries14 years ago

README.md+4−4

compare with master

unifiedsplit

@@ -9,12 +9,12 @@ At present, this repository hosts

9* network-transport-tests: Test suite for Network.Transport instantiations

10* network-transport-tcp: TCP instantiation of Network.Transport

11* network-transport-inmemory: In-memory instantiation of Network.Transport (incomplete)

12−* network-transport-composed: Compose two transports (incomplete)

12+* network-transport-composed: Compose two transports (very preliminary)

13* distributed-static: Support for static values

14* distributed-process: The main Cloud Haskell package

15* distributed-process-simplelocalnet: Simple backend for local networks

16−* distributed-process-azure: Azure backend for Cloud Haskell (work in progress)

16+* distributed-process-azure: Azure backend for Cloud Haskell (proof of concept)

17−* azure-service-api: Haskell bindings for the Azure service API (work in progress)

17+* azure-service-api: Haskell bindings for the Azure service API

18* rank1dynamic: Like Data.Dynamic and Data.Typeable but with support for polymorphic values

20For more detailed information about the interfaces provided by these packages,

@@ -26,7 +26,7 @@ to the corresponding Haddock documentation ([Control.Distributed.Process][3],

26[Control.Distributed.Process.Node][5], and

27[Control.Distributed.Process.Backend.SimpleLocalnet][6]).

29−[1]: http://research.microsoft.com/en-us/um/people/simonpj/papers/parallel/remote.pdf

29+[1]: http://www.haskell.org/haskellwiki/Cloud_Haskell

30[2]: https://github.com/haskell-distributed/distributed-process/wiki

31[3]: http://hackage.haskell.org/packages/archive/distributed-process/0.2.1.4/doc/html/Control-Distributed-Process.html

32[4]: http://hackage.haskell.org/packages/archive/distributed-process/0.2.1.4/doc/html/Control-Distributed-Process-Closure.html

azure-service-api/azure-service-api.cabal+11−3

@@ -3,8 +3,16 @@

4Name: azure-service-api

5Version: 0.1.0.0

6−Synopsis: Haskell bindings for the Azure Service Management API

6+Synopsis: Haskell bindings for the Microsoft Azure Service Management API

7−-- description:

7+Description: At the moment, this module only provides minimal

8+ functionality, just enough to support the

9+ "distributed-process-azure" package (Azure backend for

10+ Cloud Haskell). However, the code is set up in such a way

11+ that adding additional functionality should be relatively

12+ straightforward; developers who wish to do will probably

13+ want to consult the Windows Azure Service Management REST

14+ API Reference

15+ (<http://msdn.microsoft.com/en-us/library/windowsazure/ee460799.aspx>).

8Homepage: github.com/haskell-distributed/azure-service-api

16Homepage: github.com/haskell-distributed/azure-service-api

9License: BSD3

17License: BSD3

10License-File: LICENSE

18License-File: LICENSE

@@ -18,7 +26,7 @@ Cabal-Version: >=1.8

18Library

26Library

19 Exposed-Modules: Network.Azure.ServiceManagement

27 Exposed-Modules: Network.Azure.ServiceManagement

20 Build-Depends: base >= 4.5 && < 5,

28 Build-Depends: base >= 4.5 && < 5,

21− http-conduit >= 1.8 && < 1.9,

29+ http-conduit >= 1.8.1 && < 1.9,

22 tls >= 1.0 && < 1.1,

30 tls >= 1.0 && < 1.1,

23 tls-extra >= 0.5 && < 0.6,

31 tls-extra >= 0.5 && < 0.6,

24 bytestring >= 0.9 && < 0.11,

32 bytestring >= 0.9 && < 0.11,

azure-service-api/src/Network/Azure/ServiceManagement.hs+46−18

@@ -1,17 +1,23 @@

1{-# LANGUAGE Arrows #-}

2module Network.Azure.ServiceManagement

3 ( -- * Data types

4− HostedService(..)

5− , CloudService(..)

4+ CloudService

5+ , cloudServiceName

6+ , cloudServiceVMs

6− , VirtualMachine(..)

7+ , VirtualMachine

8+ , vmName

9+ , vmIpAddress

10+ , vmInputEndpoints

7− , Endpoint(..)

11+ , Endpoint

12+ , endpointName

13+ , endpointPort

14+ , endpointVip

8 -- * Pure functions

15 -- * Pure functions

9 , vmSshEndpoint

16 , vmSshEndpoint

10 -- * Setup

17 -- * Setup

11 , AzureSetup(..)

18 , AzureSetup(..)

12 , azureSetup

19 , azureSetup

13 -- * High-level API

20 -- * High-level API

14− , hostedServices

15 , cloudServices

21 , cloudServices

16 ) where

22 ) where

@@ -62,11 +68,7 @@ import Text.XML.HXT.Core

62 , IOSArrow

68 , IOSArrow

63 , ArrowXml

69 , ArrowXml

64 , getText

70 , getText

65−-- , writeDocumentToString

66−-- , withIndent

67−-- , yes

68 )

71 )

69−-- import Control.Arrow.ArrowIO (arrIO)

70import Text.XML.HXT.XPath (getXPathTrees)

72import Text.XML.HXT.XPath (getXPathTrees)

72--------------------------------------------------------------------------------

74--------------------------------------------------------------------------------

@@ -77,20 +79,39 @@ data HostedService = HostedService {

77 hostedServiceName :: String

79 hostedServiceName :: String

78 }

80 }

82+-- | A cloud service is a bunch of virtual machines that are part of the same

83+-- network (i.e., can talk to each other directly using standard TCP

84+-- connections).

80data CloudService = CloudService {

85data CloudService = CloudService {

86+ -- | Name of the service.

81 cloudServiceName :: String

87 cloudServiceName :: String

88+ -- | Virtual machines that are part of this cloud service.

82 , cloudServiceVMs :: [VirtualMachine]

89 , cloudServiceVMs :: [VirtualMachine]

83 }

90 }

92+-- | Virtual machine

85data VirtualMachine = VirtualMachine {

93data VirtualMachine = VirtualMachine {

94+ -- | Name of the virtual machine.

86 vmName :: String

95 vmName :: String

96+ -- | The /internal/ IP address of the virtual machine (that is, the

97+ -- IP address on the Cloud Service). For the globally accessible IP

98+ -- address see 'vmInputEndpoints'.

87 , vmIpAddress :: String

99 , vmIpAddress :: String

100+ -- | Globally accessible endpoints to the virtual machine

88 , vmInputEndpoints :: [Endpoint]

101 , vmInputEndpoints :: [Endpoint]

89 }

102 }

103

104+-- | Globally accessible endpoint for a virtual machine

91data Endpoint = Endpoint {

105data Endpoint = Endpoint {

106+ -- | Name of the endpoint (typical example: @SSH@)

92 endpointName :: String

107 endpointName :: String

108+ -- | Port number (typical examples are 22 or high numbered ports such as 53749)

93 , endpointPort :: String

109 , endpointPort :: String

110+ -- | Virtual IP address (that is, globally accessible IP address).

111+ --

112+ -- This corresponds to the IP address associated with the Cloud Service

113+ -- (i.e., that would be returned by a DNS lookup for @name.cloudapp.net@,

114+ -- where @name@ is the name of the Cloud Service).

94 , endpointVip :: String

115 , endpointVip :: String

95 }

116 }

117

@@ -148,6 +169,7 @@ hang2 d1 = hang d1 2

148-- Pure operations --

169-- Pure operations --

149--------------------------------------------------------------------------------

170--------------------------------------------------------------------------------

150

171

172+-- | Find the endpoint with name @SSH@.

151vmSshEndpoint :: VirtualMachine -> Maybe Endpoint

173vmSshEndpoint :: VirtualMachine -> Maybe Endpoint

152vmSshEndpoint vm = listToMaybe

174vmSshEndpoint vm = listToMaybe

153 [ ep

175 [ ep

@@ -160,11 +182,21 @@ vmSshEndpoint vm = listToMaybe

160--------------------------------------------------------------------------------

182--------------------------------------------------------------------------------

161

183

162-- | Azure setup

184-- | Azure setup

185+--

186+-- The documentation of "distributed-process-azure" explains in detail how

187+-- to obtain the SSL client certificate and the private key for your Azure

188+-- account.

189+--

190+-- See also 'azureSetup'.

163data AzureSetup = AzureSetup

191data AzureSetup = AzureSetup

192+ { -- | Azure subscription ID

164− { subscriptionId :: String

193+ subscriptionId :: String

194+ -- | SSL client certificate

165− , certificate :: X509

195+ , certificate :: X509

196+ -- | RSA private key

166− , privateKey :: PrivateKey

197+ , privateKey :: PrivateKey

198+ -- | Base URL (generally <https://management.core.windows.net>)

167− , baseUrl :: String

199+ , baseUrl :: String

168 }

200 }

169

201

170-- TODO: it's dubious to be transferring private keys, but we transfer them

202-- TODO: it's dubious to be transferring private keys, but we transfer them

@@ -221,7 +253,7 @@ azureSetup sid certPath pkeyPath = do

221-- High-level API --

253-- High-level API --

222--------------------------------------------------------------------------------

254--------------------------------------------------------------------------------

223

255

224−-- | Get a list of virtual machines

256+-- | Find available cloud services

225cloudServices :: AzureSetup -> IO [CloudService]

257cloudServices :: AzureSetup -> IO [CloudService]

226cloudServices setup = azureExecute setup $ \exec -> do

258cloudServices setup = azureExecute setup $ \exec -> do

227 services <- exec hostedServicesRequest

259 services <- exec hostedServicesRequest

@@ -240,10 +272,6 @@ cloudServices setup = azureExecute setup $ \exec -> do

240 }

272 }

241 return $ CloudService (hostedServiceName service) roles

273 return $ CloudService (hostedServiceName service) roles

242

274

243−-- | Get list of available hosted services

244−hostedServices :: AzureSetup -> IO [HostedService]

245−hostedServices setup = azureExecute setup (\exec -> exec hostedServicesRequest)

246−

247hostedServicesRequest :: AzureRequest HostedService

275hostedServicesRequest :: AzureRequest HostedService

248hostedServicesRequest = AzureRequest

276hostedServicesRequest = AzureRequest

249 { relativeUrl = "/services/hostedservices"

277 { relativeUrl = "/services/hostedservices"

distributed-process-azure/distributed-process-azure.cabal+10−2

@@ -11,6 +11,14 @@ Stability: experimental

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

12Bug-Reports: mailto:edsko@well-typed.com

13Synopsis: Microsoft Azure backend for Cloud Haskell

14+Description: This is a proof of concept Azure backend for Cloud Haskell. It

15+ provides just enough functionality to run Cloud Haskell

16+ applications on Azure virtual machines. You need to create your

17+ virtual machines in the Azure management portal; you can then

18+ use this backend to copy or verify your executable to the

19+ virtual machine, start or terminate Cloud Haskell nodes on those

20+ virtual machines, and communicate with those virtual machines

21+ from your local machine.

14Category: Control

22Category: Control

16Source-Repository head

24Source-Repository head

@@ -68,7 +76,7 @@ Executable cloud-haskell-azure-ping

68 distributed-process >= 0.3.2 && < 0.5,

76 distributed-process >= 0.3.2 && < 0.5,

69 transformers >= 0.3 && < 0.4,

77 transformers >= 0.3 && < 0.4,

70 bytestring >= 0.9 && < 0.11,

78 bytestring >= 0.9 && < 0.11,

71− binary >= 0.5 && < 0.6,

79+ binary >= 0.5 && < 0.7,

72 mtl,

80 mtl,

73 libssh2

81 libssh2

74 else

82 else

@@ -83,7 +91,7 @@ Executable cloud-haskell-azure-fib

83 distributed-process >= 0.3.2 && < 0.5,

91 distributed-process >= 0.3.2 && < 0.5,

84 transformers >= 0.3 && < 0.4,

92 transformers >= 0.3 && < 0.4,

85 bytestring >= 0.9 && < 0.11,

93 bytestring >= 0.9 && < 0.11,

86− binary >= 0.5 && < 0.6,

94+ binary >= 0.5 && < 0.7,

87 binary-generic >= 0.2 && < 0.3,

95 binary-generic >= 0.2 && < 0.3,

88 rank1dynamic >= 0.1 && < 0.2,

96 rank1dynamic >= 0.1 && < 0.2,

89 distributed-static >= 0.2 && < 0.3,

97 distributed-static >= 0.2 && < 0.3,

distributed-process-azure/src/Control/Distributed/Process/Backend/Azure.hs+3−3

@@ -41,7 +41,7 @@

41--

42-- We have only tested Cloud Haskell with Linux based virtual machines;

43-- Windows based virtual machines /might/ work, but you'll be entering

44−-- unchartered territory. Cloud Haskell assumes that all nodes run the same

44+-- uncharted territory. Cloud Haskell assumes that all nodes run the same

45-- binary code; hence, you must use the same OS on all virtual machines,

46-- /as well as on your local machine/. We use Ubuntu Server 12.04 LTS for our

47-- tests (running on VirtualBox on our local machine).

@@ -312,7 +312,7 @@

312-- >

313-- > import System.Environment (getArgs)

314-- > import Data.Binary (encode, decode)

315−-- > import Control.Monad (forever)

315+-- > import Control.Monad (void, forever)

316-- > import Control.Monad.IO.Class (liftIO)

317-- > import Control.Exception (try, IOException)

318-- > import Control.Distributed.Process

@@ -386,7 +386,7 @@

386-- > -- The same binary can behave as the client or the server,

387-- > -- depending on the command line arguments

388-- > case cmd of

389−-- > "server" -> spawnOnVM backend vm port ($(mkClosure 'pingServer) ())

389+-- > "server" -> void $ spawnOnVM backend vm port ($(mkClosure 'pingServer) ())

390-- > "client" -> callOnVM backend vm port $

391-- > ProcessPair ($(mkClosure 'pingClientRemote) ())

392-- > pingClientLocal

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

@@ -141,7 +141,7 @@ executable distributed-process-simplelocalnet-mapreduce

141 distributed-static >= 0.2.0 && < 0.3,

142 containers >= 0.4 && < 0.6,

143 bytestring >= 0.9 && < 0.11,

144− binary >= 0.5 && < 0.6,

144+ binary >= 0.5 && < 0.7,

145 array >= 0.4 && < 0.5,

146 random >= 1.0 && < 1.1

147 extensions: BangPatterns, CPP, TemplateHaskell, ScopedTypeVariables

distributed-process-simplelocalnet/distributed-process-simplelocalnet.cabal+2−2

@@ -33,7 +33,7 @@ Library

33 network >= 2.3 && < 2.5,

34 network-multicast >= 0.0 && < 0.1,

35 data-accessor >= 0.2 && < 0.3,

36− binary >= 0.5 && < 0.6,

36+ binary >= 0.5 && < 0.7,

37 containers >= 0.4 && < 0.6,

38 transformers >= 0.2 && < 0.4,

39 network-transport >= 0.3 && < 0.4,

@@ -55,7 +55,7 @@ Executable TestSimpleLocalnet

55 network >= 2.3 && < 2.5,

56 network-multicast >= 0.0 && < 0.1,

57 data-accessor >= 0.2 && < 0.3,

58− binary >= 0.5 && < 0.6,

58+ binary >= 0.5 && < 0.7,

59 containers >= 0.4 && < 0.6,

60 transformers >= 0.2 && < 0.4,

61 network-transport >= 0.3 && < 0.4,

distributed-process/distributed-process.cabal+14−10

@@ -39,7 +39,7 @@ flag benchmarks

40Library

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

42− binary >= 0.5 && < 0.6,

42+ binary >= 0.5 && < 0.7,

43 network-transport >= 0.3 && < 0.4,

44 stm >= 2.3 && < 2.5,

45 transformers >= 0.2 && < 0.4,

@@ -93,9 +93,11 @@ Test-Suite TestCH

93 distributed-process,

94 network-transport >= 0.3 && < 0.4,

95 network-transport-tcp >= 0.3 && < 0.4,

96− binary >= 0.5 && < 0.6,

96+ binary >= 0.5 && < 0.7,

97− network >= 2.3 && < 2.5

97+ network >= 2.3 && < 2.5,

98− Other-modules: TestAuxiliary

98+ HUnit >= 1.2 && < 1.3,

99+ test-framework >= 0.6 && < 0.7,

100+ test-framework-hunit >= 0.2 && < 0.3

99 Extensions: CPP,

101 Extensions: CPP,

100 ScopedTypeVariables,

102 ScopedTypeVariables,

101 DeriveDataTypeable,

103 DeriveDataTypeable,

@@ -114,8 +116,10 @@ Test-Suite TestClosure

114 network-transport >= 0.3 && < 0.4,

116 network-transport >= 0.3 && < 0.4,

115 network-transport-tcp >= 0.3 && < 0.4,

117 network-transport-tcp >= 0.3 && < 0.4,

116 bytestring >= 0.9 && < 0.11,

118 bytestring >= 0.9 && < 0.11,

117− network >= 2.3 && < 2.5

119+ network >= 2.3 && < 2.5,

118− Other-modules: TestAuxiliary

120+ HUnit >= 1.2 && < 1.3,

121+ test-framework >= 0.6 && < 0.7,

122+ test-framework-hunit >= 0.2 && < 0.3

119 Extensions: CPP,

123 Extensions: CPP,

120 ScopedTypeVariables

124 ScopedTypeVariables

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

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

@@ -127,7 +131,7 @@ Executable distributed-process-throughput

127 distributed-process,

131 distributed-process,

128 network-transport-tcp >= 0.3 && < 0.4,

132 network-transport-tcp >= 0.3 && < 0.4,

129 bytestring >= 0.9 && < 0.11,

133 bytestring >= 0.9 && < 0.11,

130− binary >= 0.5 && < 0.6

134+ binary >= 0.5 && < 0.7

131 else

135 else

132 buildable: False

136 buildable: False

133 Main-Is: benchmarks/Throughput.hs

137 Main-Is: benchmarks/Throughput.hs

@@ -140,7 +144,7 @@ Executable distributed-process-latency

140 distributed-process,

144 distributed-process,

141 network-transport-tcp >= 0.3 && < 0.4,

145 network-transport-tcp >= 0.3 && < 0.4,

142 bytestring >= 0.9 && < 0.11,

146 bytestring >= 0.9 && < 0.11,

143− binary >= 0.5 && < 0.6

147+ binary >= 0.5 && < 0.7

144 else

148 else

145 buildable: False

149 buildable: False

146 Main-Is: benchmarks/Latency.hs

150 Main-Is: benchmarks/Latency.hs

@@ -153,7 +157,7 @@ Executable distributed-process-channels

153 distributed-process,

157 distributed-process,

154 network-transport-tcp >= 0.3 && < 0.4,

158 network-transport-tcp >= 0.3 && < 0.4,

155 bytestring >= 0.9 && < 0.11,

159 bytestring >= 0.9 && < 0.11,

156− binary >= 0.5 && < 0.6

160+ binary >= 0.5 && < 0.7

157 else

161 else

158 buildable: False

162 buildable: False

159 Main-Is: benchmarks/Channels.hs

163 Main-Is: benchmarks/Channels.hs

@@ -166,7 +170,7 @@ Executable distributed-process-spawns

166 distributed-process,

170 distributed-process,

167 network-transport-tcp >= 0.3 && < 0.4,

171 network-transport-tcp >= 0.3 && < 0.4,

168 bytestring >= 0.9 && < 0.11,

172 bytestring >= 0.9 && < 0.11,

169− binary >= 0.5 && < 0.6

173+ binary >= 0.5 && < 0.7

170 else

174 else

171 buildable: False

175 buildable: False

172 Main-Is: benchmarks/Spawns.hs

176 Main-Is: benchmarks/Spawns.hs

distributed-process/src/Control/Distributed/Process.hs+3−1

@@ -19,6 +19,7 @@ module Control.Distributed.Process

19 -- * Basic messaging

20 , send

21 , expect

22+ , expectTimeout

22 -- * Channels

23 -- * Channels

23 , ReceivePort

24 , ReceivePort

24 , SendPort

25 , SendPort

@@ -26,6 +27,7 @@ module Control.Distributed.Process

26 , newChan

27 , newChan

27 , sendChan

28 , sendChan

28 , receiveChan

29 , receiveChan

30+ , receiveChanTimeout

29 , mergePortsBiased

31 , mergePortsBiased

30 , mergePortsRR

32 , mergePortsRR

31 -- * Advanced messaging

33 -- * Advanced messaging

@@ -91,7 +93,6 @@ module Control.Distributed.Process

91 , bracket_

93 , bracket_

92 , finally

94 , finally

93 -- * Auxiliary API

95 -- * Auxiliary API

94− , expectTimeout

95 , spawnAsync

96 , spawnAsync

96 , spawnSupervised

97 , spawnSupervised

97 , spawnLink

98 , spawnLink

@@ -215,6 +216,7 @@ import Control.Distributed.Process.Internal.Primitives

215 , finally

216 , finally

216 -- Auxiliary API

217 -- Auxiliary API

217 , expectTimeout

218 , expectTimeout

219+ , receiveChanTimeout

218 , spawnAsync

220 , spawnAsync

219 -- Reconnecting

221 -- Reconnecting

220 , reconnect

222 , reconnect

distributed-process/src/Control/Distributed/Process/Internal/Primitives.hs+58−19

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

54 , finally

55 -- * Auxiliary API

56 , expectTimeout

57+ , receiveChanTimeout

57 , spawnAsync

58 , spawnAsync

58 , linkNode

59 , linkNode

59 , linkPort

60 , linkPort

@@ -74,6 +75,7 @@ import Data.Binary (decode)

74import Data.Time.Clock (getCurrentTime)

75import Data.Time.Clock (getCurrentTime)

75import Data.Time.Format (formatTime)

76import Data.Time.Format (formatTime)

76import System.Locale (defaultTimeLocale)

77import System.Locale (defaultTimeLocale)

78+import System.Timeout (timeout)

77import Control.Monad.Reader (ask)

79import Control.Monad.Reader (ask)

78import Control.Monad.IO.Class (MonadIO, liftIO)

80import Control.Monad.IO.Class (MonadIO, liftIO)

79import Control.Applicative ((<$>))

81import Control.Applicative ((<$>))

@@ -87,6 +89,7 @@ import Control.Distributed.Process.Internal.StrictMVar

87import Control.Concurrent.Chan (writeChan)

89import Control.Concurrent.Chan (writeChan)

88import Control.Concurrent.STM

90import Control.Concurrent.STM

89 ( STM

91 ( STM

92+ , TVar

90 , atomically

93 , atomically

91 , orElse

94 , orElse

92 , newTVar

95 , newTVar

@@ -179,7 +182,7 @@ newChan = do

179 let sport = SendPort cid

182 let sport = SendPort cid

180 chan <- liftIO newTQueueIO

183 chan <- liftIO newTQueueIO

181 chan' <- mkWeakTQueue chan $ finalizer (processState proc) lcid

184 chan' <- mkWeakTQueue chan $ finalizer (processState proc) lcid

182− let rport = ReceivePortSingle chan

185+ let rport = ReceivePort $ readTQueue chan

183 let tch = TypedChannel chan'

186 let tch = TypedChannel chan'

184 return ( (channelCounter ^: (+ 1))

187 return ( (channelCounter ^: (+ 1))

185 . (typedChannelWithId lcid ^= Just tch)

188 . (typedChannelWithId lcid ^= Just tch)

@@ -188,7 +191,7 @@ newChan = do

188 )

191 )

189 where

192 where

190 finalizer :: StrictMVar LocalProcessState -> LocalSendPortId -> IO ()

193 finalizer :: StrictMVar LocalProcessState -> LocalSendPortId -> IO ()

191− finalizer processState lcid = modifyMVar_ processState $

194+ finalizer st lcid = modifyMVar_ st $

192 return . (typedChannelWithId lcid ^= Nothing)

195 return . (typedChannelWithId lcid ^= Nothing)

193

196

194-- | Send a message on a typed channel

197-- | Send a message on a typed channel

@@ -204,33 +207,39 @@ sendChan (SendPort cid) msg = do

204-- | Wait for a message on a typed channel

207-- | Wait for a message on a typed channel

205receiveChan :: Serializable a => ReceivePort a -> Process a

208receiveChan :: Serializable a => ReceivePort a -> Process a

206receiveChan = liftIO . atomically . receiveSTM

209receiveChan = liftIO . atomically . receiveSTM

207− where

208− receiveSTM :: ReceivePort a -> STM a

209− receiveSTM (ReceivePortSingle c) =

210− readTQueue c

211− receiveSTM (ReceivePortBiased ps) =

212− foldr1 orElse (map receiveSTM ps)

213− receiveSTM (ReceivePortRR psVar) = do

214− ps <- readTVar psVar

215− a <- foldr1 orElse (map receiveSTM ps)

216− writeTVar psVar (rotate ps)

217− return a

218

210

219− rotate :: [a] -> [a]

211+-- | Like 'receiveChan' but with a timeout. If the timeout is 0, do a

220− rotate [] = []

212+-- non-blocking check for a message.

221− rotate (x:xs) = xs ++ [x]

213+receiveChanTimeout :: Serializable a => Int -> ReceivePort a -> Process (Maybe a)

214+receiveChanTimeout 0 ch = liftIO . atomically $

215+ (Just <$> receiveSTM ch) `orElse` return Nothing

216+receiveChanTimeout n ch = liftIO . timeout n . atomically $

217+ receiveSTM ch

222

218

223-- | Merge a list of typed channels.

219-- | Merge a list of typed channels.

224--

220--

225-- The result port is left-biased: if there are messages available on more

221-- The result port is left-biased: if there are messages available on more

226-- than one port, the first available message is returned.

222-- than one port, the first available message is returned.

227mergePortsBiased :: Serializable a => [ReceivePort a] -> Process (ReceivePort a)

223mergePortsBiased :: Serializable a => [ReceivePort a] -> Process (ReceivePort a)

228−mergePortsBiased = return . ReceivePortBiased

224+mergePortsBiased = return . ReceivePort. foldr1 orElse . map receiveSTM

229

225

230-- | Like 'mergePortsBiased', but with a round-robin scheduler (rather than

226-- | Like 'mergePortsBiased', but with a round-robin scheduler (rather than

231-- left-biased)

227-- left-biased)

232mergePortsRR :: Serializable a => [ReceivePort a] -> Process (ReceivePort a)

228mergePortsRR :: Serializable a => [ReceivePort a] -> Process (ReceivePort a)

229+mergePortsRR = \ps -> do

233−mergePortsRR ps = liftIO . atomically $ ReceivePortRR <$> newTVar ps

230+ psVar <- liftIO . atomically $ newTVar (map receiveSTM ps)

231+ return $ ReceivePort (rr psVar)

232+ where

233+ rotate :: [a] -> [a]

234+ rotate [] = []

235+ rotate (x:xs) = xs ++ [x]

236+

237+ rr :: TVar [STM a] -> STM a

238+ rr psVar = do

239+ ps <- readTVar psVar

240+ a <- foldr1 orElse ps

241+ writeTVar psVar (rotate ps)

242+ return a

234

243

235--------------------------------------------------------------------------------

244--------------------------------------------------------------------------------

236-- Advanced messaging --

245-- Advanced messaging --

@@ -328,6 +337,25 @@ getSelfNode = localNodeId . processNode <$> ask

328

337

329-- | Link to a remote process (asynchronous)

338-- | Link to a remote process (asynchronous)

330--

339--

340+-- When process A links to process B (that is, process A calls

341+-- @link pidB@) then an asynchronous exception will be thrown to process A

342+-- when process B terminates (normally or abnormally), or when process A gets

343+-- disconnected from process B. Although it is /technically/ possible to catch

344+-- these exceptions, chances are if you find yourself trying to do so you should

345+-- probably be using 'monitor' rather than 'link'. In particular, code such as

346+--

347+-- > link pidB -- Link to process B

348+-- > expect -- Wait for a message from process B

349+-- > unlink pidB -- Unlink again

350+--

351+-- doesn't quite do what one might expect: if process B sends a message to

352+-- process A, and /subsequently terminates/, then process A might or might not

353+-- be terminated too, depending on whether the exception is thrown before or

354+-- after the 'unlink' (i.e., this code has a race condition).

355+--

356+-- Linking is all-or-nothing: A is either linked to B, or it's not. A second

357+-- call to 'link' has no effect.

358+--

331-- Note that 'link' provides unidirectional linking (see 'spawnSupervised').

359-- Note that 'link' provides unidirectional linking (see 'spawnSupervised').

332-- Linking makes no distinction between normal and abnormal termination of

360-- Linking makes no distinction between normal and abnormal termination of

333-- the remote process.

361-- the remote process.

@@ -335,6 +363,17 @@ link :: ProcessId -> Process ()

335link = sendCtrlMsg Nothing . Link . ProcessIdentifier

363link = sendCtrlMsg Nothing . Link . ProcessIdentifier

336

364

337-- | Monitor another process (asynchronous)

365-- | Monitor another process (asynchronous)

366+--

367+-- When process A monitors process B (that is, process A calls

368+-- @monitor pidB@) then process A will receive a 'ProcessMonitorNotification'

369+-- when process B terminates (normally or abnormally), or when process A gets

370+-- disconnected from process B. You receive this message like any other (using

371+-- 'expect'); the notification includes a reason ('DiedNormal', 'DiedException',

372+-- 'DiedDisconnect', etc.).

373+--

374+-- Every call to 'monitor' returns a new monitor reference 'MonitorRef'; if

375+-- multiple monitors are set up, multiple notifications will be delivered

376+-- and monitors can be disabled individually using 'unmonitor'.

338monitor :: ProcessId -> Process MonitorRef

377monitor :: ProcessId -> Process MonitorRef

339monitor = monitor' . ProcessIdentifier

378monitor = monitor' . ProcessIdentifier

340

379

@@ -429,7 +468,7 @@ finally a sequel = bracket_ (return ()) sequel a

429

468

430-- | Like 'expect' but with a timeout

469-- | Like 'expect' but with a timeout

431expectTimeout :: forall a. Serializable a => Int -> Process (Maybe a)

470expectTimeout :: forall a. Serializable a => Int -> Process (Maybe a)

432−expectTimeout timeout = receiveTimeout timeout [match return]

471+expectTimeout n = receiveTimeout n [match return]

433

472

434-- | Asynchronous version of 'spawn'

473-- | Asynchronous version of 'spawn'

435--

474--

distributed-process/src/Control/Distributed/Process/Internal/Types.hs+10−2

@@ -82,9 +82,9 @@ import Control.Category ((>>>))

82import Control.Exception (Exception)

83import Control.Concurrent (ThreadId)

84import Control.Concurrent.Chan (Chan)

85−import Control.Concurrent.STM (TVar)

85+import Control.Concurrent.STM (STM)

86import qualified Network.Transport as NT (EndPoint, EndPointAddress, Connection)

87−import Control.Applicative (Applicative, (<$>), (<*>))

87+import Control.Applicative (Applicative, Alternative, (<$>), (<*>))

88import Control.Monad.Reader (MonadReader(..), ReaderT, runReaderT)

89import Control.Monad.IO.Class (MonadIO)

90import Control.Distributed.Process.Serializable

@@ -262,6 +262,13 @@ newtype SendPort a = SendPort {

262 deriving (Typeable, Binary, Show, Eq, Ord)

263

264-- | The receive end of a typed channel (not serializable)

265+--

266+-- Note that 'ReceivePort' implements 'Functor', 'Applicative', 'Alternative'

267+-- and 'Monad'. This is especially useful when merging receive ports.

268+newtype ReceivePort a = ReceivePort { receiveSTM :: STM a }

269+ deriving (Typeable, Functor, Applicative, Alternative, Monad)

270+

271+{-

265data ReceivePort a =

272data ReceivePort a =

266 -- | A single receive port

273 -- | A single receive port

267 ReceivePortSingle (TQueue a)

274 ReceivePortSingle (TQueue a)

@@ -270,6 +277,7 @@ data ReceivePort a =

270 -- | A round-robin combination of receive ports

277 -- | A round-robin combination of receive ports

271 | ReceivePortRR (TVar [ReceivePort a])

278 | ReceivePortRR (TVar [ReceivePort a])

272 deriving Typeable

279 deriving Typeable

280+-}

273

281

274--------------------------------------------------------------------------------

282--------------------------------------------------------------------------------

275-- Messages --

283-- Messages --

distributed-process/tests/TestAuxiliary.hsdeleted−75

@@ -1,75 +0,0 @@

1−{-# OPTIONS_GHC -fno-warn-orphans #-}

2−module TestAuxiliary ( -- Running tests

3− runTest

4− , runTests

5− -- Writing tests

6− , forkTry

7− , trySome

8− , randomThreadDelay

9− ) where

10−

11−

12−#if ! MIN_VERSION_base(4,6,0)

13−import Prelude hiding (catch)

14−#endif

15−

16−import Control.Concurrent (myThreadId, forkIO, ThreadId, throwTo, threadDelay)

17−import Control.Monad (liftM2, unless)

18−import Control.Exception (SomeException, try, catch)

19−import System.Timeout (timeout)

20−import System.IO (stdout, hFlush)

21−import System.Console.ANSI ( SGR(SetColor, Reset)

22− , Color(Red, Green)

23− , ConsoleLayer(Foreground)

24− , ColorIntensity(Vivid)

25− , setSGR

26− )

27−import System.Random (randomIO)

28−

29−-- | Like fork, but throw exceptions in the child thread to the parent

30−forkTry :: IO () -> IO ThreadId

31−forkTry p = do

32− tid <- myThreadId

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

34−

35−-- | Like try, but specialized to SomeException

36−trySome :: IO a -> IO (Either SomeException a)

37−trySome = try

38−

39−-- | Run the given test, catching timeouts and exceptions

40−runTest :: String -> IO () -> IO Bool

41−runTest description test = do

42− putStr $ "Running " ++ show description ++ ": "

43− hFlush stdout

44− done <- try . timeout 60000000 $ test -- 60 seconds

45− case done of

46− Left err -> failed $ "(exception: " ++ show (err :: SomeException) ++ ")"

47− Right Nothing -> failed "(timeout)"

48− Right (Just ()) -> ok

49− where

50− failed :: String -> IO Bool

51− failed err = do

52− setSGR [SetColor Foreground Vivid Red]

53− putStr "failed "

54− setSGR [Reset]

55− putStrLn err

56− return False

57−

58− ok :: IO Bool

59− ok = do

60− setSGR [SetColor Foreground Vivid Green]

61− putStrLn "ok"

62− setSGR [Reset]

63− return True

64−

65−-- | Run a bunch of tests and throw an exception if any fails

66−runTests :: [(String, IO ())] -> IO ()

67−runTests tests = do

68− success <- foldr (liftM2 (&&) . uncurry runTest) (return True) tests

69− unless success $ fail "Some tests failed"

70−

71−-- | Random thread delay between 0 and the specified max

72−randomThreadDelay :: Int -> IO ()

73−randomThreadDelay maxDelay = do

74− delay <- randomIO :: IO Int

75− threadDelay (delay `mod` maxDelay)

distributed-process/tests/TestCH.hs+255−120

@@ -4,10 +4,11 @@ module Main where

4import Prelude hiding (catch)

5#endif

7import Data.Binary (Binary(..))

8import Data.Binary (Binary(..))

8import Data.Typeable (Typeable)

9import Data.Typeable (Typeable)

9import Data.Foldable (forM_)

10import Data.Foldable (forM_)

10−import Control.Concurrent (forkIO, threadDelay)

11+import Control.Concurrent (forkIO, threadDelay, myThreadId, throwTo, ThreadId)

11import Control.Concurrent.MVar

12import Control.Concurrent.MVar

12 ( MVar

13 ( MVar

13 , newEmptyMVar

14 , newEmptyMVar

@@ -16,8 +17,9 @@ import Control.Concurrent.MVar

16 , readMVar

17 , readMVar

17 )

18 )

18import Control.Monad (replicateM_, replicateM, forever)

19import Control.Monad (replicateM_, replicateM, forever)

19−import Control.Exception (throwIO)

20+import Control.Exception (SomeException, throwIO)

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

20−import Control.Applicative ((<$>), (<*>))

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

21import qualified Network.Transport as NT (Transport, closeEndPoint)

23import qualified Network.Transport as NT (Transport, closeEndPoint)

22import Network.Socket (sClose)

24import Network.Socket (sClose)

23import Network.Transport.TCP

25import Network.Transport.TCP

@@ -32,7 +34,10 @@ import Control.Distributed.Process.Internal.Types

32 )

34 )

33import Control.Distributed.Process.Node

35import Control.Distributed.Process.Node

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

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

35−import TestAuxiliary

37+

38+import Test.HUnit (Assertion)

39+import Test.Framework (Test, defaultMain, testGroup)

40+import Test.Framework.Providers.HUnit (testCase)

37newtype Ping = Ping ProcessId

42newtype Ping = Ping ProcessId

38 deriving (Typeable, Binary, Show)

43 deriving (Typeable, Binary, Show)

@@ -40,6 +45,16 @@ newtype Ping = Ping ProcessId

40newtype Pong = Pong ProcessId

45newtype Pong = Pong ProcessId

41 deriving (Typeable, Binary, Show)

46 deriving (Typeable, Binary, Show)

48+--------------------------------------------------------------------------------

49+-- Supporting definitions --

50+--------------------------------------------------------------------------------

51+

52+-- | Like fork, but throw exceptions in the child thread to the parent

53+forkTry :: IO () -> IO ThreadId

54+forkTry p = do

55+ tid <- myThreadId

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

57+

43-- | The ping server from the paper

58-- | The ping server from the paper

44ping :: Process ()

59ping :: Process ()

45ping = do

60ping = do

@@ -55,35 +70,32 @@ whereisRemote nid string = do

55 WhereIsReply _ mPid <- expect

70 WhereIsReply _ mPid <- expect

56 return mPid

71 return mPid

58−-- | Basic ping test

73+data Add = Add ProcessId Double Double deriving (Typeable)

59−testPing :: NT.Transport -> IO ()

74+data Divide = Divide ProcessId Double Double deriving (Typeable)

60−testPing transport = do

75+data DivByZero = DivByZero deriving (Typeable)

61− serverAddr <- newEmptyMVar

62− clientDone <- newEmptyMVar

63−

64− -- Server

65− forkIO $ do

66− localNode <- newLocalNode transport initRemoteTable

67− addr <- forkProcess localNode ping

68− putMVar serverAddr addr

69−

70− -- Client

71− forkIO $ do

72− localNode <- newLocalNode transport initRemoteTable

73− pingServer <- readMVar serverAddr

75− let numPings = 10000

77+instance Binary Add where

78+ put (Add pid x y) = put pid >> put x >> put y

79+ get = Add <$> get <*> get <*> get

77− runProcess localNode $ do

81+instance Binary Divide where

78− pid <- getSelfPid

82+ put (Divide pid x y) = put pid >> put x >> put y

79− replicateM_ numPings $ do

83+ get = Divide <$> get <*> get <*> get

80− send pingServer (Pong pid)

81− Ping _ <- expect

82− return ()

84− putMVar clientDone ()

85+instance Binary DivByZero where

86+ put DivByZero = return ()

87+ get = return DivByZero

86− takeMVar clientDone

89+-- The math server from the paper

90+math :: Process ()

91+math = do

92+ receiveWait

93+ [ match (\(Add pid x y) -> send pid (x + y))

94+ , matchIf (\(Divide _ _ y) -> y /= 0)

95+ (\(Divide pid x y) -> send pid (x / y))

96+ , match (\(Divide pid _ _) -> send pid DivByZero)

97+ ]

98+ math

88-- | Monitor or link to a remote node

100-- | Monitor or link to a remote node

89monitorOrLink :: Bool -- ^ 'True' for monitor, 'False' for link

101monitorOrLink :: Bool -- ^ 'True' for monitor, 'False' for link

@@ -125,9 +137,42 @@ monitorTestProcess theirAddr mOrL un reason monitorSetup done =

125 liftIO $ putMVar done ()

137 liftIO $ putMVar done ()

126 )

138 )

127

139

140+--------------------------------------------------------------------------------

141+-- The tests proper --

142+--------------------------------------------------------------------------------

143+

144+-- | Basic ping test

145+testPing :: NT.Transport -> Assertion

146+testPing transport = do

147+ serverAddr <- newEmptyMVar

148+ clientDone <- newEmptyMVar

149+

150+ -- Server

151+ forkIO $ do

152+ localNode <- newLocalNode transport initRemoteTable

153+ addr <- forkProcess localNode ping

154+ putMVar serverAddr addr

155+

156+ -- Client

157+ forkIO $ do

158+ localNode <- newLocalNode transport initRemoteTable

159+ pingServer <- readMVar serverAddr

160+

161+ let numPings = 10000

162+

163+ runProcess localNode $ do

164+ pid <- getSelfPid

165+ replicateM_ numPings $ do

166+ send pingServer (Pong pid)

167+ Ping _ <- expect

168+ return ()

169+

170+ putMVar clientDone ()

171+

172+ takeMVar clientDone

128

173

129-- | Monitor a process on an unreachable node

174-- | Monitor a process on an unreachable node

130−testMonitorUnreachable :: NT.Transport -> Bool -> Bool -> IO ()

175+testMonitorUnreachable :: NT.Transport -> Bool -> Bool -> Assertion

131testMonitorUnreachable transport mOrL un = do

176testMonitorUnreachable transport mOrL un = do

132 deadProcess <- newEmptyMVar

177 deadProcess <- newEmptyMVar

133 done <- newEmptyMVar

178 done <- newEmptyMVar

@@ -147,7 +192,7 @@ testMonitorUnreachable transport mOrL un = do

147 takeMVar done

192 takeMVar done

148

193

149-- | Monitor a process which terminates normally

194-- | Monitor a process which terminates normally

150−testMonitorNormalTermination :: NT.Transport -> Bool -> Bool -> IO ()

195+testMonitorNormalTermination :: NT.Transport -> Bool -> Bool -> Assertion

151testMonitorNormalTermination transport mOrL un = do

196testMonitorNormalTermination transport mOrL un = do

152 monitorSetup <- newEmptyMVar

197 monitorSetup <- newEmptyMVar

153 monitoredProcess <- newEmptyMVar

198 monitoredProcess <- newEmptyMVar

@@ -168,7 +213,7 @@ testMonitorNormalTermination transport mOrL un = do

168 takeMVar done

213 takeMVar done

169

214

170-- | Monitor a process which terminates abnormally

215-- | Monitor a process which terminates abnormally

171−testMonitorAbnormalTermination :: NT.Transport -> Bool -> Bool -> IO ()

216+testMonitorAbnormalTermination :: NT.Transport -> Bool -> Bool -> Assertion

172testMonitorAbnormalTermination transport mOrL un = do

217testMonitorAbnormalTermination transport mOrL un = do

173 monitorSetup <- newEmptyMVar

218 monitorSetup <- newEmptyMVar

174 monitoredProcess <- newEmptyMVar

219 monitoredProcess <- newEmptyMVar

@@ -192,7 +237,7 @@ testMonitorAbnormalTermination transport mOrL un = do

192 takeMVar done

237 takeMVar done

193

238

194-- | Monitor a local process that is already dead

239-- | Monitor a local process that is already dead

195−testMonitorLocalDeadProcess :: NT.Transport -> Bool -> Bool -> IO ()

240+testMonitorLocalDeadProcess :: NT.Transport -> Bool -> Bool -> Assertion

196testMonitorLocalDeadProcess transport mOrL un = do

241testMonitorLocalDeadProcess transport mOrL un = do

197 processDead <- newEmptyMVar

242 processDead <- newEmptyMVar

198 processAddr <- newEmptyMVar

243 processAddr <- newEmptyMVar

@@ -212,7 +257,7 @@ testMonitorLocalDeadProcess transport mOrL un = do

212 takeMVar done

257 takeMVar done

213

258

214-- | Monitor a remote process that is already dead

259-- | Monitor a remote process that is already dead

215−testMonitorRemoteDeadProcess :: NT.Transport -> Bool -> Bool -> IO ()

260+testMonitorRemoteDeadProcess :: NT.Transport -> Bool -> Bool -> Assertion

216testMonitorRemoteDeadProcess transport mOrL un = do

261testMonitorRemoteDeadProcess transport mOrL un = do

217 processDead <- newEmptyMVar

262 processDead <- newEmptyMVar

218 processAddr <- newEmptyMVar

263 processAddr <- newEmptyMVar

@@ -233,7 +278,7 @@ testMonitorRemoteDeadProcess transport mOrL un = do

233 takeMVar done

278 takeMVar done

234

279

235-- | Monitor a process that becomes disconnected

280-- | Monitor a process that becomes disconnected

236−testMonitorDisconnect :: NT.Transport -> Bool -> Bool -> IO ()

281+testMonitorDisconnect :: NT.Transport -> Bool -> Bool -> Assertion

237testMonitorDisconnect transport mOrL un = do

282testMonitorDisconnect transport mOrL un = do

238 processAddr <- newEmptyMVar

283 processAddr <- newEmptyMVar

239 monitorSetup <- newEmptyMVar

284 monitorSetup <- newEmptyMVar

@@ -254,34 +299,8 @@ testMonitorDisconnect transport mOrL un = do

254

299

255 takeMVar done

300 takeMVar done

256

301

257−data Add = Add ProcessId Double Double deriving (Typeable)

258−data Divide = Divide ProcessId Double Double deriving (Typeable)

259−data DivByZero = DivByZero deriving (Typeable)

260−

261−instance Binary Add where

262− put (Add pid x y) = put pid >> put x >> put y

263− get = Add <$> get <*> get <*> get

264−

265−instance Binary Divide where

266− put (Divide pid x y) = put pid >> put x >> put y

267− get = Divide <$> get <*> get <*> get

268−

269−instance Binary DivByZero where

270− put DivByZero = return ()

271− get = return DivByZero

272−

273−math :: Process ()

274−math = do

275− receiveWait

276− [ match (\(Add pid x y) -> send pid (x + y))

277− , matchIf (\(Divide _ _ y) -> y /= 0)

278− (\(Divide pid x y) -> send pid (x / y))

279− , match (\(Divide pid _ _) -> send pid DivByZero)

280− ]

281− math

282−

283-- | Test the math server (i.e., receiveWait)

302-- | Test the math server (i.e., receiveWait)

284−testMath :: NT.Transport -> IO ()

303+testMath :: NT.Transport -> Assertion

285testMath transport = do

304testMath transport = do

286 serverAddr <- newEmptyMVar

305 serverAddr <- newEmptyMVar

287 clientDone <- newEmptyMVar

306 clientDone <- newEmptyMVar

@@ -312,7 +331,7 @@ testMath transport = do

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

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

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

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

314-- the same remote node (we're checking that the remote node did not die)

333-- the same remote node (we're checking that the remote node did not die)

315−testSendToTerminated :: NT.Transport -> IO ()

334+testSendToTerminated :: NT.Transport -> Assertion

316testSendToTerminated transport = do

335testSendToTerminated transport = do

317 serverAddr1 <- newEmptyMVar

336 serverAddr1 <- newEmptyMVar

318 serverAddr2 <- newEmptyMVar

337 serverAddr2 <- newEmptyMVar

@@ -342,7 +361,7 @@ testSendToTerminated transport = do

342 takeMVar clientDone

361 takeMVar clientDone

343

362

344-- | Test (non-zero) timeout

363-- | Test (non-zero) timeout

345−testTimeout :: NT.Transport -> IO ()

364+testTimeout :: NT.Transport -> Assertion

346testTimeout transport = do

365testTimeout transport = do

347 localNode <- newLocalNode transport initRemoteTable

366 localNode <- newLocalNode transport initRemoteTable

348 runProcess localNode $ do

367 runProcess localNode $ do

@@ -350,7 +369,7 @@ testTimeout transport = do

350 return ()

369 return ()

351

370

352-- | Test zero timeout

371-- | Test zero timeout

353−testTimeout0 :: NT.Transport -> IO ()

372+testTimeout0 :: NT.Transport -> Assertion

354testTimeout0 transport = do

373testTimeout0 transport = do

355 serverAddr <- newEmptyMVar

374 serverAddr <- newEmptyMVar

356 clientDone <- newEmptyMVar

375 clientDone <- newEmptyMVar

@@ -384,7 +403,7 @@ testTimeout0 transport = do

384 takeMVar clientDone

403 takeMVar clientDone

385

404

386-- | Test typed channels

405-- | Test typed channels

387−testTypedChannels :: NT.Transport -> IO ()

406+testTypedChannels :: NT.Transport -> Assertion

388testTypedChannels transport = do

407testTypedChannels transport = do

389 serverChannel <- newEmptyMVar :: IO (MVar (SendPort (SendPort Bool, Int)))

408 serverChannel <- newEmptyMVar :: IO (MVar (SendPort (SendPort Bool, Int)))

390 clientDone <- newEmptyMVar

409 clientDone <- newEmptyMVar

@@ -410,7 +429,7 @@ testTypedChannels transport = do

410 takeMVar clientDone

429 takeMVar clientDone

411

430

412-- | Test merging receive ports

431-- | Test merging receive ports

413−testMergeChannels :: NT.Transport -> IO ()

432+testMergeChannels :: NT.Transport -> Assertion

414testMergeChannels transport = do

433testMergeChannels transport = do

415 localNode <- newLocalNode transport initRemoteTable

434 localNode <- newLocalNode transport initRemoteTable

416 testFlat localNode True "aaabbbccc"

435 testFlat localNode True "aaabbbccc"

@@ -424,29 +443,34 @@ testMergeChannels transport = do

424 where

443 where

425 -- Single layer of merging

444 -- Single layer of merging

426 testFlat :: LocalNode -> Bool -> String -> IO ()

445 testFlat :: LocalNode -> Bool -> String -> IO ()

427− testFlat localNode biased expected =

446+ testFlat localNode biased expected = do

447+ done <- newEmptyMVar

428− runProcess localNode $ do

448+ forkProcess localNode $ do

429 rs <- mapM charChannel "abc"

449 rs <- mapM charChannel "abc"

430 m <- mergePorts biased rs

450 m <- mergePorts biased rs

431 xs <- replicateM 9 $ receiveChan m

451 xs <- replicateM 9 $ receiveChan m

432 True <- return $ xs == expected

452 True <- return $ xs == expected

433− return ()

453+ liftIO $ putMVar done ()

454+ takeMVar done

434

455

435 -- Two layers of merging

456 -- Two layers of merging

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

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

437− testNested localNode biasedInner biasedOuter expected =

458+ testNested localNode biasedInner biasedOuter expected = do

459+ done <- newEmptyMVar

438− runProcess localNode $ do

460+ forkProcess localNode $ do

439 rss <- mapM (mapM charChannel) ["abc", "def", "ghi"]

461 rss <- mapM (mapM charChannel) ["abc", "def", "ghi"]

440 ms <- mapM (mergePorts biasedInner) rss

462 ms <- mapM (mergePorts biasedInner) rss

441 m <- mergePorts biasedOuter ms

463 m <- mergePorts biasedOuter ms

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

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

443 True <- return $ xs == expected

465 True <- return $ xs == expected

444− return ()

466+ liftIO $ putMVar done ()

467+ takeMVar done

445

468

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

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

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

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

448 testBlocked localNode biased = do

471 testBlocked localNode biased = do

449 vs <- replicateM 3 newEmptyMVar

472 vs <- replicateM 3 newEmptyMVar

473+ done <- newEmptyMVar

450

474

451 forkProcess localNode $ do

475 forkProcess localNode $ do

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

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

@@ -477,13 +501,15 @@ testMergeChannels transport = do

477 , (sa, 'a')

501 , (sa, 'a')

478 ]

502 ]

479

503

480− runProcess localNode $ do

504+ forkProcess localNode $ do

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

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

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

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

483 m <- mergePorts biased rs

507 m <- mergePorts biased rs

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

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

485 True <- return $ xs == "abcacbbacbcacabcba"

509 True <- return $ xs == "abcacbbacbcacabcba"

486− return ()

510+ liftIO $ putMVar done ()

511+

512+ takeMVar done

487

513

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

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

489 mergePorts True = mergePortsBiased

515 mergePorts True = mergePortsBiased

@@ -496,7 +522,7 @@ testMergeChannels transport = do

496 liftIO $ threadDelay 10000 -- Make sure messages have been sent

522 liftIO $ threadDelay 10000 -- Make sure messages have been sent

497 return rport

523 return rport

498

524

499−testTerminate :: NT.Transport -> IO ()

525+testTerminate :: NT.Transport -> Assertion

500testTerminate transport = do

526testTerminate transport = do

501 localNode <- newLocalNode transport initRemoteTable

527 localNode <- newLocalNode transport initRemoteTable

502

528

@@ -510,7 +536,7 @@ testTerminate transport = do

510 True <- return $ ref == ref' && pid == pid' && ex == show ProcessTerminationException

536 True <- return $ ref == ref' && pid == pid' && ex == show ProcessTerminationException

511 return ()

537 return ()

512

538

513−testMonitorNode :: NT.Transport -> IO ()

539+testMonitorNode :: NT.Transport -> Assertion

514testMonitorNode transport = do

540testMonitorNode transport = do

515 [node1, node2] <- replicateM 2 $ newLocalNode transport initRemoteTable

541 [node1, node2] <- replicateM 2 $ newLocalNode transport initRemoteTable

516

542

@@ -522,7 +548,7 @@ testMonitorNode transport = do

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

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

523 return ()

549 return ()

524

550

525−testMonitorChannel :: NT.Transport -> IO ()

551+testMonitorChannel :: NT.Transport -> Assertion

526testMonitorChannel transport = do

552testMonitorChannel transport = do

527 [node1, node2] <- replicateM 2 $ newLocalNode transport initRemoteTable

553 [node1, node2] <- replicateM 2 $ newLocalNode transport initRemoteTable

528 gotNotification <- newEmptyMVar

554 gotNotification <- newEmptyMVar

@@ -543,7 +569,7 @@ testMonitorChannel transport = do

543

569

544 takeMVar gotNotification

570 takeMVar gotNotification

545

571

546−testRegistry :: NT.Transport -> IO ()

572+testRegistry :: NT.Transport -> Assertion

547testRegistry transport = do

573testRegistry transport = do

548 node <- newLocalNode transport initRemoteTable

574 node <- newLocalNode transport initRemoteTable

549

575

@@ -559,7 +585,7 @@ testRegistry transport = do

559 True <- return $ pingServer == pid'

585 True <- return $ pingServer == pid'

560 return ()

586 return ()

561

587

562−testRemoteRegistry :: NT.Transport -> IO ()

588+testRemoteRegistry :: NT.Transport -> Assertion

563testRemoteRegistry transport = do

589testRemoteRegistry transport = do

564 node1 <- newLocalNode transport initRemoteTable

590 node1 <- newLocalNode transport initRemoteTable

565 node2 <- newLocalNode transport initRemoteTable

591 node2 <- newLocalNode transport initRemoteTable

@@ -577,7 +603,7 @@ testRemoteRegistry transport = do

577 True <- return $ pingServer == pid'

603 True <- return $ pingServer == pid'

578 return ()

604 return ()

579

605

580−testSpawnLocal :: NT.Transport -> IO ()

606+testSpawnLocal :: NT.Transport -> Assertion

581testSpawnLocal transport = do

607testSpawnLocal transport = do

582 node <- newLocalNode transport initRemoteTable

608 node <- newLocalNode transport initRemoteTable

583

609

@@ -595,7 +621,7 @@ testSpawnLocal transport = do

595 send pid sport

621 send pid sport

596 expect

622 expect

597

623

598−testReconnect :: NT.Transport -> TransportInternals -> IO ()

624+testReconnect :: NT.Transport -> TransportInternals -> Assertion

599testReconnect transport transportInternals = do

625testReconnect transport transportInternals = do

600 [node1, node2] <- replicateM 2 $ newLocalNode transport initRemoteTable

626 [node1, node2] <- replicateM 2 $ newLocalNode transport initRemoteTable

601 let nid1 = localNodeId node1

627 let nid1 = localNodeId node1

@@ -664,7 +690,7 @@ testReconnect transport transportInternals = do

664

690

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

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

666-- in between

692-- in between

667−testMatchAny :: NT.Transport -> IO ()

693+testMatchAny :: NT.Transport -> Assertion

668testMatchAny transport = do

694testMatchAny transport = do

669 proxyAddr <- newEmptyMVar

695 proxyAddr <- newEmptyMVar

670 clientDone <- newEmptyMVar

696 clientDone <- newEmptyMVar

@@ -695,48 +721,157 @@ testMatchAny transport = do

695

721

696 takeMVar clientDone

722 takeMVar clientDone

697

723

698−main :: IO ()

724+-- Test 'receiveChanTimeout'

699−main = do

725+testReceiveChanTimeout :: NT.Transport -> Assertion

726+testReceiveChanTimeout transport = do

727+ done <- newEmptyMVar

728+ sendPort <- newEmptyMVar

729+

730+ forkTry $ do

731+ localNode <- newLocalNode transport initRemoteTable

732+ runProcess localNode $ do

733+ -- Create a typed channel

734+ (sp, rp) <- newChan :: Process (SendPort Bool, ReceivePort Bool)

735+ liftIO $ putMVar sendPort sp

736+

737+ -- Wait for a message with a delay. No message arrives, we should get Nothing after 1 second

738+ Nothing <- receiveChanTimeout 1000000 rp

739+

740+ -- Wait for a message with a delay again. Now a message arrives after 0.5 seconds

741+ Just True <- receiveChanTimeout 1000000 rp

742+

743+ -- Wait for a message with zero timeout: non-blocking check. No message is available, we get Nothing

744+ Nothing <- receiveChanTimeout 0 rp

745+

746+ -- Again, but now there is a message available

747+ liftIO $ threadDelay 1000000

748+ Just False <- receiveChanTimeout 0 rp

749+

750+ liftIO $ putMVar done ()

751+

752+ forkTry $ do

753+ localNode <- newLocalNode transport initRemoteTable

754+ runProcess localNode $ do

755+ sp <- liftIO $ readMVar sendPort

756+

757+ liftIO $ threadDelay 1500000

758+ sendChan sp True

759+

760+ liftIO $ threadDelay 500000

761+ sendChan sp False

762+

763+ takeMVar done

764+

765+-- | Test Functor, Applicative, Alternative and Monad instances for ReceiveChan

766+testReceiveChanFeatures :: NT.Transport -> Assertion

767+testReceiveChanFeatures transport = do

768+ done <- newEmptyMVar

769+

770+ forkTry $ do

771+ localNode <- newLocalNode transport initRemoteTable

772+ runProcess localNode $ do

773+ (spInt, rpInt) <- newChan :: Process (SendPort Int, ReceivePort Int)

774+ (spBool, rpBool) <- newChan :: Process (SendPort Bool, ReceivePort Bool)

775+

776+ -- Test Functor instance

777+

778+ sendChan spInt 2

779+ sendChan spBool False

780+

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

782+

783+ True <- receiveChan rp1

784+ False <- receiveChan rp1

785+

786+ -- Test Applicative instance

787+

788+ sendChan spInt 3

789+ sendChan spInt 4

790+

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

792+

793+ 7 <- receiveChan rp2

794+

795+ -- Test Alternative instance

796+

797+ sendChan spInt 3

798+ sendChan spBool True

799+

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

801+

802+ False <- receiveChan rp3

803+ True <- receiveChan rp3

804+

805+ -- Test Monad instance

806+

807+ sendChan spBool True

808+ sendChan spBool False

809+ sendChan spInt 5

810+

811+ let rp4 :: ReceivePort Int

812+ rp4 = do b <- rpBool

813+ if b

814+ then rpInt

815+ else return 7

816+

817+ 5 <- receiveChan rp4

818+ 7 <- receiveChan rp4

819+

820+ liftIO $ putMVar done ()

821+

822+ takeMVar done

823+

824+tests :: (NT.Transport, TransportInternals) -> [Test]

700− Right (transport, transportInternals) <- createTransportExposeInternals "127.0.0.1" "8080" defaultTCPParameters

825+tests (transport, transportInternals) = [

701− runTests

826+ testGroup "Basic features" [

702− [ ("Ping", testPing transport)

827+ testCase "Ping" (testPing transport)

703− , ("Math", testMath transport)

828+ , testCase "Math" (testMath transport)

704− , ("Timeout", testTimeout transport)

829+ , testCase "Timeout" (testTimeout transport)

705− , ("Timeout0", testTimeout0 transport)

830+ , testCase "Timeout0" (testTimeout0 transport)

706− , ("SendToTerminated", testSendToTerminated transport)

831+ , testCase "SendToTerminated" (testSendToTerminated transport)

707− , ("TypedChannnels", testTypedChannels transport)

832+ , testCase "TypedChannnels" (testTypedChannels transport)

708− , ("MergeChannels", testMergeChannels transport)

833+ , testCase "MergeChannels" (testMergeChannels transport)

709− , ("Terminate", testTerminate transport)

834+ , testCase "Terminate" (testTerminate transport)

710− , ("Registry", testRegistry transport)

835+ , testCase "Registry" (testRegistry transport)

711− , ("RemoteRegistry", testRemoteRegistry transport)

836+ , testCase "RemoteRegistry" (testRemoteRegistry transport)

712− , ("SpawnLocal", testSpawnLocal transport)

837+ , testCase "SpawnLocal" (testSpawnLocal transport)

713− , ("MatchAny", testMatchAny transport)

838+ , testCase "MatchAny" (testMatchAny transport)

839+ , testCase "ReceiveChanTimeout" (testReceiveChanTimeout transport)

840+ , testCase "ReceiveChanFeatures" (testReceiveChanFeatures transport)

841+ ]

842+ , testGroup "Monitoring and Linking" [

714 -- Monitoring processes

843 -- Monitoring processes

715 --

844 --

716 -- The "missing" combinations in the list below don't make much sense, as

845 -- The "missing" combinations in the list below don't make much sense, as

717 -- we cannot guarantee that the monitor reply or link exception will not

846 -- we cannot guarantee that the monitor reply or link exception will not

718 -- happen before the unmonitor or unlink

847 -- happen before the unmonitor or unlink

719− , ("MonitorUnreachable", testMonitorUnreachable transport True False)

848+ testCase "MonitorUnreachable" (testMonitorUnreachable transport True False)

720− , ("MonitorNormalTermination", testMonitorNormalTermination transport True False)

849+ , testCase "MonitorNormalTermination" (testMonitorNormalTermination transport True False)

721− , ("MonitorAbnormalTermination", testMonitorAbnormalTermination transport True False)

850+ , testCase "MonitorAbnormalTermination" (testMonitorAbnormalTermination transport True False)

722− , ("MonitorLocalDeadProcess", testMonitorLocalDeadProcess transport True False)

851+ , testCase "MonitorLocalDeadProcess" (testMonitorLocalDeadProcess transport True False)

723− , ("MonitorRemoteDeadProcess", testMonitorRemoteDeadProcess transport True False)

852+ , testCase "MonitorRemoteDeadProcess" (testMonitorRemoteDeadProcess transport True False)

724− , ("MonitorDisconnect", testMonitorDisconnect transport True False)

853+ , testCase "MonitorDisconnect" (testMonitorDisconnect transport True False)

725− , ("LinkUnreachable", testMonitorUnreachable transport False False)

854+ , testCase "LinkUnreachable" (testMonitorUnreachable transport False False)

726− , ("LinkNormalTermination", testMonitorNormalTermination transport False False)

855+ , testCase "LinkNormalTermination" (testMonitorNormalTermination transport False False)

727− , ("LinkAbnormalTermination", testMonitorAbnormalTermination transport False False)

856+ , testCase "LinkAbnormalTermination" (testMonitorAbnormalTermination transport False False)

728− , ("LinkLocalDeadProcess", testMonitorLocalDeadProcess transport False False)

857+ , testCase "LinkLocalDeadProcess" (testMonitorLocalDeadProcess transport False False)

729− , ("LinkRemoteDeadProcess", testMonitorRemoteDeadProcess transport False False)

858+ , testCase "LinkRemoteDeadProcess" (testMonitorRemoteDeadProcess transport False False)

730− , ("LinkDisconnect", testMonitorDisconnect transport False False)

859+ , testCase "LinkDisconnect" (testMonitorDisconnect transport False False)

731− , ("UnmonitorNormalTermination", testMonitorNormalTermination transport True True)

860+ , testCase "UnmonitorNormalTermination" (testMonitorNormalTermination transport True True)

732− , ("UnmonitorAbnormalTermination", testMonitorAbnormalTermination transport True True)

861+ , testCase "UnmonitorAbnormalTermination" (testMonitorAbnormalTermination transport True True)

733− , ("UnmonitorDisconnect", testMonitorDisconnect transport True True)

862+ , testCase "UnmonitorDisconnect" (testMonitorDisconnect transport True True)

734− , ("UnlinkNormalTermination", testMonitorNormalTermination transport False True)

863+ , testCase "UnlinkNormalTermination" (testMonitorNormalTermination transport False True)

735− , ("UnlinkAbnormalTermination", testMonitorAbnormalTermination transport False True)

864+ , testCase "UnlinkAbnormalTermination" (testMonitorAbnormalTermination transport False True)

736− , ("UnlinkDisconnect", testMonitorDisconnect transport False True)

865+ , testCase "UnlinkDisconnect" (testMonitorDisconnect transport False True)

737 -- Monitoring nodes and channels

866 -- Monitoring nodes and channels

738− , ("MonitorNode", testMonitorNode transport)

867+ , testCase "MonitorNode" (testMonitorNode transport)

739− , ("MonitorChannel", testMonitorChannel transport)

868+ , testCase "MonitorChannel" (testMonitorChannel transport)

740 -- Reconnect

869 -- Reconnect

741− , ("Reconnect", testReconnect transport transportInternals)

870+ , testCase "Reconnect" (testReconnect transport transportInternals)

742 ]

871 ]

872+ ]

873+

874+main :: IO ()

875+main = do

876+ Right transport <- createTransportExposeInternals "127.0.0.1" "8080" defaultTCPParameters

877+ defaultMain (tests transport)

distributed-process/tests/TestClosure.hs+60−45

@@ -29,7 +29,14 @@ import Control.Distributed.Process.Closure

29import Control.Distributed.Process.Node

30import Control.Distributed.Process.Internal.Types (NodeId(nodeAddress))

31import Control.Distributed.Static (staticLabel, staticClosure)

32−import TestAuxiliary

32+

33+import Test.HUnit (Assertion)

34+import Test.Framework (Test, defaultMain)

35+import Test.Framework.Providers.HUnit (testCase)

36+

37+--------------------------------------------------------------------------------

38+-- Supporting definitions --

39+--------------------------------------------------------------------------------

34quintuple :: a -> b -> c -> d -> e -> (a, b, c, d, e)

41quintuple :: a -> b -> c -> d -> e -> (a, b, c, d, e)

35quintuple a b c d e = (a, b, c, d, e)

42quintuple a b c d e = (a, b, c, d, e)

@@ -133,7 +140,18 @@ factorial' n = returnCP $(mkStatic 'sdictInt) n `bindCP` factorialOf

133waitClosure :: Int -> Closure (Process ())

140waitClosure :: Int -> Closure (Process ())

134waitClosure = $(mkClosure 'wait)

141waitClosure = $(mkClosure 'wait)

135

142

143+simulateNetworkFailure :: TransportInternals -> NodeId -> NodeId -> Process ()

144+simulateNetworkFailure transportInternals fr to = liftIO $ do

145+ threadDelay 10000

146+ sock <- socketBetween transportInternals (nodeAddress fr) (nodeAddress to)

147+ sClose sock

148+ threadDelay 10000

149+

150+--------------------------------------------------------------------------------

151+-- The tests proper --

152+--------------------------------------------------------------------------------

153+

136−testUnclosure :: Transport -> RemoteTable -> IO ()

154+testUnclosure :: Transport -> RemoteTable -> Assertion

137testUnclosure transport rtable = do

155testUnclosure transport rtable = do

138 node <- newLocalNode transport rtable

156 node <- newLocalNode transport rtable

139 done <- newEmptyMVar

157 done <- newEmptyMVar

@@ -142,7 +160,7 @@ testUnclosure transport rtable = do

142 liftIO $ putMVar done ()

160 liftIO $ putMVar done ()

143 takeMVar done

161 takeMVar done

144

162

145−testBind :: Transport -> RemoteTable -> IO ()

163+testBind :: Transport -> RemoteTable -> Assertion

146testBind transport rtable = do

164testBind transport rtable = do

147 node <- newLocalNode transport rtable

165 node <- newLocalNode transport rtable

148 done <- newEmptyMVar

166 done <- newEmptyMVar

@@ -153,7 +171,7 @@ testBind transport rtable = do

153 liftIO $ putMVar done ()

171 liftIO $ putMVar done ()

154 takeMVar done

172 takeMVar done

155

173

156−testSendPureClosure :: Transport -> RemoteTable -> IO ()

174+testSendPureClosure :: Transport -> RemoteTable -> Assertion

157testSendPureClosure transport rtable = do

175testSendPureClosure transport rtable = do

158 serverAddr <- newEmptyMVar

176 serverAddr <- newEmptyMVar

159 serverDone <- newEmptyMVar

177 serverDone <- newEmptyMVar

@@ -174,7 +192,7 @@ testSendPureClosure transport rtable = do

174

192

175 takeMVar serverDone

193 takeMVar serverDone

176

194

177−testSendIOClosure :: Transport -> RemoteTable -> IO ()

195+testSendIOClosure :: Transport -> RemoteTable -> Assertion

178testSendIOClosure transport rtable = do

196testSendIOClosure transport rtable = do

179 serverAddr <- newEmptyMVar

197 serverAddr <- newEmptyMVar

180 serverDone <- newEmptyMVar

198 serverDone <- newEmptyMVar

@@ -198,7 +216,7 @@ testSendIOClosure transport rtable = do

198

216

199 takeMVar serverDone

217 takeMVar serverDone

200

218

201−testSendProcClosure :: Transport -> RemoteTable -> IO ()

219+testSendProcClosure :: Transport -> RemoteTable -> Assertion

202testSendProcClosure transport rtable = do

220testSendProcClosure transport rtable = do

203 serverAddr <- newEmptyMVar

221 serverAddr <- newEmptyMVar

204 clientDone <- newEmptyMVar

222 clientDone <- newEmptyMVar

@@ -222,7 +240,7 @@ testSendProcClosure transport rtable = do

222

240

223 takeMVar clientDone

241 takeMVar clientDone

224

242

225−testSpawn :: Transport -> RemoteTable -> IO ()

243+testSpawn :: Transport -> RemoteTable -> Assertion

226testSpawn transport rtable = do

244testSpawn transport rtable = do

227 serverNodeAddr <- newEmptyMVar

245 serverNodeAddr <- newEmptyMVar

228 clientDone <- newEmptyMVar

246 clientDone <- newEmptyMVar

@@ -243,7 +261,7 @@ testSpawn transport rtable = do

243

261

244 takeMVar clientDone

262 takeMVar clientDone

245

263

246−testCall :: Transport -> RemoteTable -> IO ()

264+testCall :: Transport -> RemoteTable -> Assertion

247testCall transport rtable = do

265testCall transport rtable = do

248 serverNodeAddr <- newEmptyMVar

266 serverNodeAddr <- newEmptyMVar

249 clientDone <- newEmptyMVar

267 clientDone <- newEmptyMVar

@@ -261,7 +279,7 @@ testCall transport rtable = do

261

279

262 takeMVar clientDone

280 takeMVar clientDone

263

281

264−testCallBind :: Transport -> RemoteTable -> IO ()

282+testCallBind :: Transport -> RemoteTable -> Assertion

265testCallBind transport rtable = do

283testCallBind transport rtable = do

266 serverNodeAddr <- newEmptyMVar

284 serverNodeAddr <- newEmptyMVar

267 clientDone <- newEmptyMVar

285 clientDone <- newEmptyMVar

@@ -279,7 +297,7 @@ testCallBind transport rtable = do

279

297

280 takeMVar clientDone

298 takeMVar clientDone

281

299

282−testSeq :: Transport -> RemoteTable -> IO ()

300+testSeq :: Transport -> RemoteTable -> Assertion

283testSeq transport rtable = do

301testSeq transport rtable = do

284 node <- newLocalNode transport rtable

302 node <- newLocalNode transport rtable

285 done <- newEmptyMVar

303 done <- newEmptyMVar

@@ -297,7 +315,7 @@ testSeq transport rtable = do

297-- child. The supervisor then throws an exception, the child dies because it

315-- child. The supervisor then throws an exception, the child dies because it

298-- was linked to the supervisor, and the third process notices that the child

316-- was linked to the supervisor, and the third process notices that the child

299-- dies.

317-- dies.

300−testSpawnSupervised :: Transport -> RemoteTable -> IO ()

318+testSpawnSupervised :: Transport -> RemoteTable -> Assertion

301testSpawnSupervised transport rtable = do

319testSpawnSupervised transport rtable = do

302 [node1, node2] <- replicateM 2 $ newLocalNode transport rtable

320 [node1, node2] <- replicateM 2 $ newLocalNode transport rtable

303 [superPid, childPid] <- replicateM 2 $ newEmptyMVar

321 [superPid, childPid] <- replicateM 2 $ newEmptyMVar

@@ -326,7 +344,7 @@ testSpawnSupervised transport rtable = do

326 supervisorDeath :: IOException

344 supervisorDeath :: IOException

327 supervisorDeath = userError "Supervisor died"

345 supervisorDeath = userError "Supervisor died"

328

346

329−testSpawnInvalid :: Transport -> RemoteTable -> IO ()

347+testSpawnInvalid :: Transport -> RemoteTable -> Assertion

330testSpawnInvalid transport rtable = do

348testSpawnInvalid transport rtable = do

331 node <- newLocalNode transport rtable

349 node <- newLocalNode transport rtable

332 done <- newEmptyMVar

350 done <- newEmptyMVar

@@ -338,7 +356,7 @@ testSpawnInvalid transport rtable = do

338 liftIO $ putMVar done ()

356 liftIO $ putMVar done ()

339 takeMVar done

357 takeMVar done

340

358

341−testClosureExpect :: Transport -> RemoteTable -> IO ()

359+testClosureExpect :: Transport -> RemoteTable -> Assertion

342testClosureExpect transport rtable = do

360testClosureExpect transport rtable = do

343 node <- newLocalNode transport rtable

361 node <- newLocalNode transport rtable

344 done <- newEmptyMVar

362 done <- newEmptyMVar

@@ -351,7 +369,7 @@ testClosureExpect transport rtable = do

351 liftIO $ putMVar done ()

369 liftIO $ putMVar done ()

352 takeMVar done

370 takeMVar done

353

371

354−testSpawnChannel :: Transport -> RemoteTable -> IO ()

372+testSpawnChannel :: Transport -> RemoteTable -> Assertion

355testSpawnChannel transport rtable = do

373testSpawnChannel transport rtable = do

356 done <- newEmptyMVar

374 done <- newEmptyMVar

357 [node1, node2] <- replicateM 2 $ newLocalNode transport rtable

375 [node1, node2] <- replicateM 2 $ newLocalNode transport rtable

@@ -368,7 +386,7 @@ testSpawnChannel transport rtable = do

368

386

369 takeMVar done

387 takeMVar done

370

388

371−testTDict :: Transport -> RemoteTable -> IO ()

389+testTDict :: Transport -> RemoteTable -> Assertion

372testTDict transport rtable = do

390testTDict transport rtable = do

373 done <- newEmptyMVar

391 done <- newEmptyMVar

374 [node1, node2] <- replicateM 2 $ newLocalNode transport rtable

392 [node1, node2] <- replicateM 2 $ newLocalNode transport rtable

@@ -377,14 +395,7 @@ testTDict transport rtable = do

377 liftIO $ putMVar done ()

395 liftIO $ putMVar done ()

378 takeMVar done

396 takeMVar done

379

397

380−simulateNetworkFailure :: TransportInternals -> NodeId -> NodeId -> Process ()

381−simulateNetworkFailure transportInternals fr to = liftIO $ do

382− threadDelay 10000

383− sock <- socketBetween transportInternals (nodeAddress fr) (nodeAddress to)

384− sClose sock

385− threadDelay 10000

386−

387−testFib :: Transport -> RemoteTable -> IO ()

398+testFib :: Transport -> RemoteTable -> Assertion

388testFib transport rtable = do

399testFib transport rtable = do

389 nodes <- replicateM 4 $ newLocalNode transport rtable

400 nodes <- replicateM 4 $ newLocalNode transport rtable

390 done <- newEmptyMVar

401 done <- newEmptyMVar

@@ -397,7 +408,7 @@ testFib transport rtable = do

397

408

398 takeMVar done

409 takeMVar done

399

410

400−testSpawnReconnect :: Transport -> RemoteTable -> TransportInternals -> IO ()

411+testSpawnReconnect :: Transport -> RemoteTable -> TransportInternals -> Assertion

401testSpawnReconnect transport rtable transportInternals = do

412testSpawnReconnect transport rtable transportInternals = do

402 [node1, node2] <- replicateM 2 $ newLocalNode transport rtable

413 [node1, node2] <- replicateM 2 $ newLocalNode transport rtable

403 let nid1 = localNodeId node1

414 let nid1 = localNodeId node1

@@ -426,7 +437,7 @@ testSpawnReconnect transport rtable transportInternals = do

426

437

427-- | 'spawn' used to ave a race condition which would be triggered if the

438-- | 'spawn' used to ave a race condition which would be triggered if the

428-- spawning process terminates immediately after spawning

439-- spawning process terminates immediately after spawning

429−testSpawnTerminate :: Transport -> RemoteTable -> IO ()

440+testSpawnTerminate :: Transport -> RemoteTable -> Assertion

430testSpawnTerminate transport rtable = do

441testSpawnTerminate transport rtable = do

431 slave <- newLocalNode transport rtable

442 slave <- newLocalNode transport rtable

432 master <- newLocalNode transport rtable

443 master <- newLocalNode transport rtable

@@ -440,26 +451,30 @@ testSpawnTerminate transport rtable = do

440

451

441 takeMVar masterDone

452 takeMVar masterDone

442

453

454+tests :: (Transport, TransportInternals) -> RemoteTable -> [Test]

455+tests (transport, transportInternals) rtable = [

456+ testCase "Unclosure" (testUnclosure transport rtable)

457+ , testCase "Bind" (testBind transport rtable)

458+ , testCase "SendPureClosure" (testSendPureClosure transport rtable)

459+ , testCase "SendIOClosure" (testSendIOClosure transport rtable)

460+ , testCase "SendProcClosure" (testSendProcClosure transport rtable)

461+ , testCase "Spawn" (testSpawn transport rtable)

462+ , testCase "Call" (testCall transport rtable)

463+ , testCase "CallBind" (testCallBind transport rtable)

464+ , testCase "Seq" (testSeq transport rtable)

465+ , testCase "SpawnSupervised" (testSpawnSupervised transport rtable)

466+ , testCase "SpawnInvalid" (testSpawnInvalid transport rtable)

467+ , testCase "ClosureExpect" (testClosureExpect transport rtable)

468+ , testCase "SpawnChannel" (testSpawnChannel transport rtable)

469+ , testCase "TDict" (testTDict transport rtable)

470+ , testCase "Fib" (testFib transport rtable)

471+ , testCase "SpawnTerminate" (testSpawnTerminate transport rtable)

472+ , testCase "SpawnReconnect" (testSpawnReconnect transport rtable transportInternals)

473+ ]

474+

475+

443main :: IO ()

476main :: IO ()

444main = do

477main = do

445− Right (transport, transportInternals) <- createTransportExposeInternals "127.0.0.1" "8080" defaultTCPParameters

478+ Right transport <- createTransportExposeInternals "127.0.0.1" "8080" defaultTCPParameters

446 let rtable = __remoteTable . __remoteTableDecl $ initRemoteTable

479 let rtable = __remoteTable . __remoteTableDecl $ initRemoteTable

447− runTests

480+ defaultMain (tests transport rtable)

448− [ ("Unclosure", testUnclosure transport rtable)

449− , ("Bind", testBind transport rtable)

450− , ("SendPureClosure", testSendPureClosure transport rtable)

451− , ("SendIOClosure", testSendIOClosure transport rtable)

452− , ("SendProcClosure", testSendProcClosure transport rtable)

453− , ("Spawn", testSpawn transport rtable)

454− , ("Call", testCall transport rtable)

455− , ("CallBind", testCallBind transport rtable)

456− , ("Seq", testSeq transport rtable)

457− , ("SpawnSupervised", testSpawnSupervised transport rtable)

458− , ("SpawnInvalid", testSpawnInvalid transport rtable)

459− , ("ClosureExpect", testClosureExpect transport rtable)

460− , ("SpawnChannel", testSpawnChannel transport rtable)

461− , ("TDict", testTDict transport rtable)

462− , ("Fib", testFib transport rtable)

463− , ("SpawnTerminate", testSpawnTerminate transport rtable)

464− , ("SpawnReconnect", testSpawnReconnect transport rtable transportInternals)

465− ]

distributed-static/distributed-static.cabal+1−1

@@ -30,7 +30,7 @@ Library

30 rank1dynamic >= 0.1 && < 0.2,

31 containers >= 0.4 && < 0.6,

32 bytestring >= 0.9 && < 0.11,

33− binary >= 0.5 && < 0.6

33+ binary >= 0.5 && < 0.7

34 HS-Source-Dirs: src

35 Extensions: DeriveDataTypeable,

36 ScopedTypeVariables

doc/semantics/CloudHaskellSemantics.pdf

Binary file not shown.

doc/semantics/CloudHaskellSemantics.tex+54−2

@@ -3,6 +3,7 @@

3\usepackage{amsmath, amssymb, stmaryrd}

4\usepackage{mathpazo, times}

5\usepackage{float}

6+\usepackage{listings}

7\newcommand{\sLam}[2]{\lambda {#1} \cdot {#2}}

8\newcommand{\sLam}[2]{\lambda {#1} \cdot {#2}}

8\newcommand{\sApp}[2]{{#1} {#2}}

9\newcommand{\sApp}[2]{{#1} {#2}}

@@ -60,6 +61,8 @@

60\floatstyle{boxed}

61\floatstyle{boxed}

61\restylefloat{figure}

62\restylefloat{figure}

64+\lstset{basicstyle=\ttfamily\small}

65+

63\begin{document}

66\begin{document}

65\title{Cloud Haskell Semantics (DRAFT)}

68\title{Cloud Haskell Semantics (DRAFT)}

@@ -256,8 +259,8 @@ Figure~\ref{fig:disconnect} gives the semantics for disconnecting and

256reconnecting. Rule \textsc{Disconnect} models random network disconnect

259reconnecting. Rule \textsc{Disconnect} models random network disconnect

257between nodes $\sNid_1$ and $\sNid_2$. We assume that entire \emph{nodes} get

260between nodes $\sNid_1$ and $\sNid_2$. We assume that entire \emph{nodes} get

258disconnected from each other, not individual processes. \emph{Reconnecting}

261disconnected from each other, not individual processes. \emph{Reconnecting}

259−however is on a per process basis (\textsc{Recon-Ex}). Connections to and from

262+however is on a per connection basis (\textsc{Recon-Ex}). Connections to and

260−node controllers can be implicitly reconnected ($\textsc{Recon-Im}_1$ and

263+from node controllers can be implicitly reconnected ($\textsc{Recon-Im}_1$ and

261$\textsc{Recon-Im}_2$).

264$\textsc{Recon-Im}_2$).

262

265

263\begin{figure}

266\begin{figure}

@@ -648,6 +651,55 @@ silently, unless monitors are setup.

648\caption{\label{fig:termination}Process Termination}

651\caption{\label{fig:termination}Process Termination}

649\end{figure}

652\end{figure}

650

653

654+\section{Open Issues}

655+

656+\subsection{Ordering of Monitor/Link Notifications}

657+

658+The semantics as described above, like the original Unified semantics, does not

659+guarantee that messages send from process $P$ to process $Q$ (\textsc{Send})

660+are ordered with respect to the link or monitor notification sent when process

661+$P$ terminates normally or abnormally (\textsc{Mon-Ret}, \textsc{Mon-Throw}). This means that if process $P$ does

662+

663+\begin{lstlisting}

664+receiveWait [

665+ match $ \(Reply reply) -> ...

666+ , match $ \(ProcessMonitorNotification ..) -> ...

667+ ]

668+\end{lstlisting}

669+

670+and process $Q$ does

671+

672+\begin{lstlisting}

673+send pidA (Reply reply)

674+// terminate or indeed throw an exception

675+\end{lstlisting}

676+

677+then the semantics does not guarantee that the reply from process $Q$ will

678+arrive at process $P$ before the monitor notification; hence, this results in

679+an (artificial) race condition in process $P$.

680+

681+One possible solution is to regard such a link notification as a message from

682+process $Q$ to process $P$, which should be ordered along with the other

683+messages.

684+

685+\subsection{Ordering and Typed Channels}

686+

687+The situation is more tricky still than sketched above because we of typed

688+channels. The semantics does not provide ordering guarantees between messages

689+sent directly to the process and messages sent on typed channels. That means

690+that even if we consider a link or monitor notification as a message sent to

691+the process, to be ordered with other messages sent to that process, it is

692+still unordered with respect to messages sent on typed channels. This means

693+that we have similar race conditions\footnote{Even ignoring the fact that we

694+currently don't even provide a way to wait for a message on a typed channel

695+\emph{or} a direct message.}

696+

697+A possible solution is to insist that \emph{all} messages from process $P$ to

698+process $Q$ are ordered, no matter how they are sent. From a implementation

699+point of view, this would entail the use of a single ordered network connection

700+for all messages from $P$ to $Q$, rather than using an ordered connection per

701+typed channel plus one for direct messages.

702+

651\bibliographystyle{apalike}

703\bibliographystyle{apalike}

652\bibliography{references}

704\bibliography{references}

653

705

network-transport/network-transport.cabal+1−1

@@ -65,7 +65,7 @@ Source-Repository head

66Library

67 Build-Depends: base >= 4.3 && < 5,

68− binary >= 0.5 && < 0.6,

68+ binary >= 0.5 && < 0.7,

69 bytestring >= 0.9 && < 0.11,

70 transformers >= 0.2 && < 0.4

71 Exposed-Modules: Network.Transport,

rank1dynamic/rank1dynamic.cabal+1−1

@@ -19,7 +19,7 @@ Library

19 Data.Rank1Typeable

20 Build-Depends: base >= 4.4 && < 4.7,

21 ghc-prim >= 0.2 && < 0.4,

22− binary >= 0.5 && < 0.6

22+ binary >= 0.5 && < 0.7

23 HS-Source-Dirs: src

24 GHC-Options: -Wall

25 Extensions: EmptyDataDecls,