master..a307db2 · distributed-process

.gitignore+13−3

since splitsnapshots

unifiedsplit

@@ -1,4 +1,14 @@

1+tags

2+TAGS

1dist/

3dist/

4+\#*#

5+*~

6+*.aux

7+*.hp

8+*.hi

9+*.o

10+*.prof

11+*.ps

12+*.swp

2−.cabal-sandbox

13+cabal-dev

3−cabal.sandbox.config

14+.DS_Store

4−.stack*

.travis.ymldeleted−43

@@ -1,43 +0,0 @@

1−language: c

2−

3−sudo: false

4−

5−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− addons: {apt: {packages: [libgmp-dev]}}

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

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

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

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

15−

16−cache:

17− directories:

18− - $HOME/.stack

19−

20−before_install:

21−- export PATH=$HOME/.local/bin:$PATH;

22− mkdir -p ~/.local/bin;

23− travis_retry curl -L https://www.stackage.org/stack/linux-x86_64 | tar xz --wildcards --strip-components=1 -C ~/.local/bin '*/stack';

24− stack --version

25−

26−install:

27−- stack $ARGS setup --no-terminal;

28− stack $ARGS build --no-terminal --only-snapshot --test --no-run-tests;

29−

30−script:

31−- export ARG='--test-arguments'

32− export TEST_PACKAGE="distributed-process-tests:"

33−- stack ${ARGS} test $ARG='--plain -t "!Flaky"' ${TEST_PACKAGE}TestCHInMemory

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

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

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

40−

41−notifications:

42− slack:

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

CONTRIBUTING.mddeleted−1

@@ -1 +0,0 @@

1−See https://github.com/haskell-distributed/cloud-haskell/blob/master/CONTRIBUTING.md.

ChangeLogdeleted−236

@@ -1,236 +0,0 @@

1−[Unrealeased]

2−

3−*

4−

5−2018-06-12 Facundo Domínguez <facundo.dominguez@tweag.io> 0.7.4

6−

7−* Added support for exceptions >= 0.10

8−

9−2017-08-31 Facundo Domínguez <facundo.dominguez@tweag.io> 0.7.3

10−

11−* Drop support for ghc-7.8.* and earlier.

12−

13−2017-08-31 Facundo Domínguez <facundo.dominguez@tweag.io> 0.7.2

14−

15−* Fixed build errors with ghc-8.2.1.

16−

17−2017-08-22 Facundo Domínguez <facundo.dominguez@tweag.io> 0.7.1

18−

19−* Relax upper bounds in dependencies to build with ghc-8.2.1.

20−

21−2017-08-21 Facundo Domínguez <facundo.dominguez@tweag.io> 0.7.0

22−

23−* Change type of message sent by `say` from a 3-tuple to a proper

24−type (`SayMessage`) with a proper `UTCTime`. (#291)

25−* Expose the MonitorRef in the withMonitor call.

26−* Have unmonitor remove the monitor message in the inbox. (#268)

27−* Remove Mx Data Tables. This API isn't used, is easy to replace with various

28−other packages. (#276)

29−* Add Ord instance to SpawnRef.

30−

31−2016-10-13 Facundo Domínguez <facundo.dominguez@tweag.io> 0.6.6

32−

33−* Remove monitors from remote nodes when a process dies. (#295)

34−

35−2016-10-12 Facundo Domínguez <facundo.dominguez@tweag.io> 0.6.5

36−

37−* Use only one connection to communicate between NCs. (#296, #297)

38−* Improve documentation of CQueue.

39−* Implement bidirectional multimaps for links and monitors. (#293, #294)

40−* Fix various warnings. (#292)

41−* Fix some of the intermittent failures in tests.

42−* Improve error messages when node controllers receive invalid requests.

43−(#290)

44−

45−2016-06-09 Facundo Domínguez <facundo.dominguez@tweag.io> 0.6.4

46−

47−* Fixup build errors.

48−

49−2016-06-09 Facundo Domínguez <facundo.dominguez@tweag.io> 0.6.3

50−

51−* Relax template-haskell upper bound.

52−

53−2016-06-09 Facundo Domínguez <facundo.dominguez@tweag.io> 0.6.2

54−

55−* Provide compatibility with ghc-8.0.1

56−* Remove dependency on ghc-prim.

57−* Don't throw exceptions asynchronously.

58−* Bump upper bounds of dependencies.

59−* Fix exception handling in callLocal.

60−* Have spawnLocal inherit the masking state of the caller.

61−* Have nsend send unencoded messages to local processes.

62−

63−2016-03-03 Facundo Domínguez <facundo.dominguez@tweag.io> 0.6.1

64−

65−* Implement MonadCatch, MonadThrow, MonadMask for Process.

66−

67−2016-02-18 Facundo Domínguez <facundo.dominguez@tweag.io> 0.6.0

68−

69−* Have nsendRemote skip the transport for local communication.

70−* Unsafe primitives for usend and nsendRemote.

71−* Stop using the transport for local communication.

72−* Skip the transport for whereisRemoteAsync and registerRemoteAsync.

73−* Have nsendRemote skip the transport for local communication.

74−* Have runProcess forward exceptions.

75−* Reimport distributed-process-tests. d-p and d-p-test now can be kept in

76−sync.

77−* Add a stack.yaml file for building tests and d-p all at once conveniently.

78−* Implement unreliable forward (uforward).

79−* Add Functor instance for Match data type

80−* Have `spawnAsync` not use the transport in the local case.

81−* Fix monitor race in 'call'.

82−* Add compatibility with ghc-7.10: support new typeable, loosen deps, write

83−proper NFData instances, support new TH.

84−* Kill processes on a local node upon closeLocalNode.

85−* Fix getNodeStats function, see DP-97

86−* Return size of the queue in ProcessInfo.

87−* Implement unreliable send (usend).

88−* Implement MonadFix instance for Process.

89−* Introduce callLocal primitive.

90−* Prevent message loss due to timeouts in CQueue.

91−* More informative ProcessRegistrationException. Now includes the identifier

92−of the process that owns the name, if any.

93−* Avoid message loop between threads when tracing received messages.

94−

95−2015-06-15 Facundo Domínguez <facundo.dominguez@tweag.io> 0.5.5

96−

97−* Fix dependencies.

98−* Add compatibility with GHC-7.10.

99−* Fix various race conditions (DP-99, DP-103).

100−

101−2014-12-09 Tim Watson <watson.timothy@gmail.com> 0.5.2

102−

103−* Fix docstring for `register`

104−* Added Data instance to ProcessId, LocalProcessId and NodeId

105−* Add static serialiation dictionary for 'Static', for completeness

106−* Add Closure static serialization dictionary

107−* Replacement for modifyMVarMasked for GHC <= 7.4

108−* Document the use of built-in trace flags

109−* Make forkProcess exception-safe

110−* Make -Wall clean

111−

112−2014-08-13 Tim Watson <watson.timothy@gmail.com> 0.5.1

113−

114−* Fix cabal docs (thanks Markus Barenhoff)

115−* Expose lifted version of Control.Exception.mask_ (thanks Alexander Vershilov)

116−

117−2014-05-30 Tim Watson <watson.timothy@gmail.com> 0.5.0

118−

119−* Dependency on STM implicitly changed from 1.3 to 1.4, but was not reflected in the cabal file

120−* Race condition in local monitoring when using call

121−* mask now works correctly if unmask is called by another process

122−* Improve efficiency of local message passing

123−* nsend uses local communication channels

124−* Link Node Controller and Network Listener

125−* Label spawned processes using labelThread

126−* Relaxed upper bound on syb in the cabal manifest

127−* Bump binary version to include 0.7.*

128−* Exposed process info

129−* Exposed node statistics

130−* Moved tests to https://github.com/haskell-distributed/distributed-process-tests

131−* Added "polymorphic expect"

132−* Exposed Message type and broaden scope of polymorphic expect

133−* Added Management API (for working with internal/system events)

134−* Tracing can no longer be disabled

135−* We now report node statistics for monitoring/management

136−* Node.runProcess now propagates exceptions to its caller

137−* Added simple micro benchmarks

138−

139−2013-01-27 Tim Watson <watson.timothy@gmail.com> 0.4.2

140−

141−* Improved exception handling for deferred type checked exit reasons

142−* Add matchChan primitive (thanks Simon Marlow)

143−* Expose deferred message handling/checking for AbstractMessage

144−* Add `getProcessInfo' API

145−* Add `trace' API backed by the GHC eventlog

146−

147−2012-11-22 Edsko de Vries <edsko@well-typed.com> 0.4.1

148−

149−* Make behaviour of 'register' more Erlang-like (register will now fail if the

150−name is already registered). Patch by Jeff Epstein.

151−* Functor, Applicative, Alternative and Monad instances for ReceivePort

152−* Add support for receiveChanTimeout

153−* Improved documentation

154−* Avoid name clashes in the TH generation for closures

155−* Relax package bounds to allow for Binary 0.6

156−

157−2012-10-23 Edsko de Vries <edsko@well-typed.com> 0.4.0.2

158−

159−* Fix race condition in spawn

160−

161−2012-10-04 Edsko de Vries <edsko@well-typed.com> 0.4.0.1

162−

163−* Relax package boundaries

164−

165−2012-10-03 Edsko de Vries <edsko@well-typed.com> 0.4.0

166−

167−* Improved treatment of network failure, using new failure semantics of

168−Network.Transport.

169−* Make NodeId Typeable

170−* Extend Template Haskell support with "remotableDec" so that you can refer to

171−$(mkClosure 'f) within the body of "f".

172−* Fix bug in spawnChannelLocal

173−* Numerous memory leaks plugged

174−* Relax upper bound on dependency on 'network'

175−* New primitive 'matchAny'

176−* Remove 'whereisRemote' (see comment of 'whereisRemoteAsync')

177−

178−2012-08-16 Edsko de Vries <edsko@well-typed.com> 0.3.1

179−

180−* Fix memory leaks

181−* Make Template Haskell support optional

182−* Relax dependency constraints

183−

184−2012-08-07 Edsko de Vries <edsko@well-typed.com> 0.3.0

185−

186−* Extract 'static' into a separate package (C.D.Static)

187−* Use new package rank1dynamic to proper runtime checks for polymorphic values

188−

189−2012-08-02 Edsko de Vries <edsko@well-typed.com> 0.2.3.0

190−

191−* Expose the constructors of Closure

192−* Add instance (Typeable a => Serializable (Static a)) and make sure we only

193−use the internal representation of Static where really necessary

194−* Improved docs

195−

196−2012-07-31 Edsko de Vries <edsko@well-typed.com> 0.2.2.0

197−

198−* Add exception handling primitives

199−* Fix runProcess: if the process threw an exception, a 'waiting indefinitely on

200−MVar' exception would be thrown.

201−

202−2012-07-21 Edsko de Vries <edsko@well-typed.com> 0.2.1.4

203−

204−* Bugfix in the node controller

205−(one way this bug materialized: when using the SimpleLocalnet backend,

206−slave nodes could not be reused)

207−* Improved documentation in Control.Distributed.Process.Closure

208−

209−2012-07-20 Edsko de Vries <edsko@well-typed.com> 0.2.1.3

210−

211−* Improve docs

212−* Local versions of spawn

213−

214−2012-07-16 Edsko de Vries <edsko@well-typed.com> 0.2.1.2

215−

216−* Base 4.6 compatibility

217−* Relax constraints on bytestring and containers

218−

219−2012-07-16 Edsko de Vries <edsko@well-typed.com> 0.2.1.1

220−

221−* Relax upper bound on 'time' dependency

222−

223−2012-07-11 Edsko de Vries <edsko@well-typed.com> 0.2.1

224−

225−* Complete redesign of the underlying implementation of static values and

226−closures.

227−

228−* Add support for 'spawnChannel'

229−

230−2012-07-09 Edsko de Vries <edsko@well-typed.com> 0.2.0.1

231−

232−* Bugfix: Continue processing messages when a connection breaks.

233−

234−2012-07-07 Edsko de Vries <edsko@well-typed.com> 0.2.0

235−

236−* Initial release.

README.md+29−11

@@ -1,16 +1,34 @@

1−# distributed-process

1+Haskell Distributed Project

2−[![travis](https://secure.travis-ci.org/haskell-distributed/distributed-process.png)](http://travis-ci.org/haskell-distributed/distributed-process)

2+===========================

3−[![Release](https://img.shields.io/hackage/v/distributed-process.svg)](https://hackage.haskell.org/package/distributed-process)

4

3

5−This repository is part of Cloud Haskell.

4+This repository holds an implementation of [Cloud Haskell][1].

6

5

7−See http://haskell-distributed.github.com for documentation, user guides,

6+At present, this repository hosts

8−tutorials and assistance.

9

7

10−## Getting Help / Raising Issues

8+* network-transport: Generic Network.Transport API

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)

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)

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

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

11

19

12−Please visit the [bug tracker](https://github.com/haskell-distributed/distributed-process/issues) to submit issues. You can contact the distributed-haskell@googlegroups.com mailing list for help and comments.

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

21+please refer to the [distributed-process repository wiki][2]. People who wish

22+to get started with Cloud Haskell should cabal install

23+distributed-process and possibly distributed-process-simplelocalnet and refer

24+to the corresponding Haddock documentation ([Control.Distributed.Process][3],

25+[Control.Distributed.Process.Closure][4],

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

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

13

28

14−## License

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

15−

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

16−This package is made available under a 3-clause BSD-style license.

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

33+[5]: http://hackage.haskell.org/packages/archive/distributed-process/0.2.1.4/doc/html/Control-Distributed-Process-Node.html

34+[6]: http://hackage.haskell.org/packages/archive/distributed-process-simplelocalnet/0.2.0.3/doc/html/Control-Distributed-Process-Backend-SimpleLocalnet.html

azure-service-api/LICENSEnew file+30

@@ -0,0 +1,30 @@

2+

4+

5+Redistribution and use in source and binary forms, with or without

6+modification, are permitted provided that the following conditions are met:

7+

8+ * Redistributions of source code must retain the above copyright

9+ notice, this list of conditions and the following disclaimer.

10+

11+ * Redistributions in binary form must reproduce the above

12+ copyright notice, this list of conditions and the following

13+ disclaimer in the documentation and/or other materials provided

14+ with the distribution.

15+

16+ * Neither the name of Edsko de Vries nor the names of other

17+ contributors may be used to endorse or promote products derived

18+ from this software without specific prior written permission.

19+

20+THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS

21+"AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT

22+LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR

23+A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT

24+OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,

25+SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT

26+LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,

27+DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY

28+THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT

29+(INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE

30+OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.

Setup.hs → azure-service-api/Setup.hsrenamed

The content is the same.

azure-service-api/azure-service-api.cabalnew file+36

@@ -0,0 +1,36 @@

1+-- Initial azure-service-api.cabal generated by cabal init. For further

2+-- documentation, see http://haskell.org/cabal/users-guide/

3+

4+Name: azure-service-api

5+Version: 0.1.0.0

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

7+-- description:

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

9+License: BSD3

10+License-File: LICENSE

11+Author: Edsko de Vries

12+Maintainer: edsko@well-typed.com

13+Copyright: Well-Typed LLP

14+Category: Network

15+Build-Type: Simple

16+Cabal-Version: >=1.8

17+

18+Library

19+ Exposed-Modules: Network.Azure.ServiceManagement

20+ Build-Depends: base ==4.5.*,

21+ http-conduit >= 1.6.1 && < 1.7,

22+ tls >= 1.0 && < 1.1,

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

24+ bytestring >= 0.9 && < 1.0,

25+ certificate >= 1.2 && < 1.3,

26+ case-insensitive >= 0.4 && < 0.5,

27+ hxt >= 9.2 && < 9.3,

28+ pretty >= 1.1 && < 1.2,

29+ crypto-pubkey-types >= 0.1 && < 0.2,

30+ binary >= 0.5 && < 0.6,

31+ resourcet,

32+ transformers,

33+ hxt-xpath

34+ HS-Source-Dirs: src

35+ ghc-options: -Wall

36+ Extensions: RankNTypes

azure-service-api/src/Network/Azure/ServiceManagement.hsnew file+291

@@ -0,0 +1,291 @@

1+{-# LANGUAGE Arrows #-}

2+module Network.Azure.ServiceManagement

3+ ( -- * Data types

4+ HostedService(..)

5+ , CloudService(..)

6+ , VirtualMachine(..)

7+ , Endpoint(..)

8+ -- * Pure functions

9+ , vmSshEndpoint

10+ -- * Setup

11+ , AzureSetup(..)

12+ , azureSetup

13+ -- * High-level API

14+ , hostedServices

15+ , cloudServices

16+ ) where

17+

18+import Prelude hiding (id, (.))

19+import Control.Category (id, (.))

20+import Control.Arrow (arr)

21+import Control.Monad (forM)

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

23+import Data.Maybe (listToMaybe)

24+import Data.ByteString.Lazy (ByteString)

25+import Data.ByteString.Char8 as BSC (pack)

26+import Data.ByteString.Lazy.Char8 as BSLC (unpack)

27+import Data.Binary (Binary(get,put))

28+import Data.Binary.Put (runPut)

29+import Data.Binary.Get (Get, runGet)

30+import Network.TLS (PrivateKey(PrivRSA))

31+import Network.TLS.Extra (fileReadCertificate, fileReadPrivateKey)

32+import Data.Certificate.X509 (X509, encodeCertificate, decodeCertificate)

33+import qualified Crypto.Types.PubKey.RSA as RSA (PrivateKey(..))

34+import Control.Monad.Trans.Resource (ResourceT)

35+import Control.Monad.IO.Class (liftIO)

36+import Control.Arrow.ArrowList (listA, arr2A)

37+import Text.PrettyPrint

38+ ( Doc

39+ , text

40+ , (<+>)

41+ , ($$)

42+ , vcat

43+ , hang

44+ , doubleQuotes

45+ )

46+import Network.HTTP.Conduit

47+ ( parseUrl

48+ , clientCertificates

49+ , requestHeaders

50+ , withManager

51+ , Response(Response)

52+ , httpLbs

53+ , Manager

54+ )

55+import Data.CaseInsensitive as CI (mk)

56+import Text.XML.HXT.Core

57+ ( runX

58+ , readString

59+ , withValidate

60+ , no

61+ , XmlTree

62+ , IOSArrow

63+ , ArrowXml

64+ , getText

65+-- , writeDocumentToString

66+-- , withIndent

67+-- , yes

68+ )

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

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

71+

72+--------------------------------------------------------------------------------

73+-- Data types --

74+--------------------------------------------------------------------------------

75+

76+data HostedService = HostedService {

77+ hostedServiceName :: String

78+ }

79+

80+data CloudService = CloudService {

81+ cloudServiceName :: String

82+ , cloudServiceVMs :: [VirtualMachine]

83+ }

84+

85+data VirtualMachine = VirtualMachine {

86+ vmName :: String

87+ , vmIpAddress :: String

88+ , vmInputEndpoints :: [Endpoint]

89+ }

90+

91+data Endpoint = Endpoint {

92+ endpointName :: String

93+ , endpointPort :: String

94+ , endpointVip :: String

95+ }

96+

97+--------------------------------------------------------------------------------

98+-- Pretty-printing --

99+--------------------------------------------------------------------------------

100+

101+instance Show HostedService where

102+ show = show . ppHostedService

103+

104+instance Show CloudService where

105+ show = show . ppCloudService

106+

107+instance Show VirtualMachine where

108+ show = show . ppVirtualMachine

109+

110+instance Show Endpoint where

111+ show = show . ppEndpoint

112+

113+ppHostedService :: HostedService -> Doc

114+ppHostedService = text . hostedServiceName

115+

116+ppCloudService :: CloudService -> Doc

117+ppCloudService cs =

118+ (text "Cloud Service" <+> (doubleQuotes . text . cloudServiceName $ cs))

119+ `hang2`

120+ ( text "VIRTUAL MACHINES"

121+ `hang2`

122+ (vcat . map ppVirtualMachine . cloudServiceVMs $ cs)

123+ )

124+

125+ppVirtualMachine :: VirtualMachine -> Doc

126+ppVirtualMachine vm =

127+ (text "Virtual Machine" <+> (doubleQuotes . text . vmName $ vm))

128+ `hang2`

129+ ( text "IP" <+> text (vmIpAddress vm)

130+ $$ ( text "INPUT ENDPOINTS"

131+ `hang2`

132+ (vcat . map ppEndpoint . vmInputEndpoints $ vm)

133+ )

134+ )

135+

136+ppEndpoint :: Endpoint -> Doc

137+ppEndpoint ep =

138+ (text "Input endpoint" <+> (doubleQuotes . text . endpointName $ ep))

139+ `hang2`

140+ ( text "Port" <+> text (endpointPort ep)

141+ $$ text "VIP" <+> text (endpointVip ep)

142+ )

143+

144+hang2 :: Doc -> Doc -> Doc

145+hang2 d1 = hang d1 2

146+

147+--------------------------------------------------------------------------------

148+-- Pure operations --

149+--------------------------------------------------------------------------------

150+

151+vmSshEndpoint :: VirtualMachine -> Maybe Endpoint

152+vmSshEndpoint vm = listToMaybe

153+ [ ep

154+ | ep <- vmInputEndpoints vm

155+ , endpointName ep == "SSH"

156+ ]

157+

158+--------------------------------------------------------------------------------

159+-- Setup --

160+--------------------------------------------------------------------------------

161+

162+-- | Azure setup

163+data AzureSetup = AzureSetup

164+ { subscriptionId :: String

165+ , certificate :: X509

166+ , privateKey :: PrivateKey

167+ , baseUrl :: String

168+ }

169+

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

171+-- over a secure connection and it can be argued that it's safer than actually

172+-- storing the private key on each remote server

173+

174+encodePrivateKey :: PrivateKey -> ByteString

175+encodePrivateKey (PrivRSA pkey) = runPut $ do

176+ put (RSA.private_size pkey)

177+ put (RSA.private_n pkey)

178+ put (RSA.private_d pkey)

179+ put (RSA.private_p pkey)

180+ put (RSA.private_q pkey)

181+ put (RSA.private_dP pkey)

182+ put (RSA.private_dQ pkey)

183+ put (RSA.private_qinv pkey)

184+

185+decodePrivateKey :: ByteString -> PrivateKey

186+decodePrivateKey = PrivRSA . runGet getPrivateKey

187+ where

188+ getPrivateKey :: Get RSA.PrivateKey

189+ getPrivateKey =

190+ RSA.PrivateKey <$> get <*> get <*> get <*> get <*> get <*> get <*> get <*> get

191+

192+instance Binary AzureSetup where

193+ put (AzureSetup sid cert pkey url) = do

194+ put sid

195+ put (encodeCertificate cert)

196+ put (encodePrivateKey pkey)

197+ put url

198+ get = do

199+ sid <- get

200+ Right cert <- decodeCertificate <$> get

201+ pkey <- decodePrivateKey <$> get

202+ url <- get

203+ return $ AzureSetup sid cert pkey url

204+

205+-- | Initialize Azure

206+azureSetup :: String -- ^ Subscription ID

207+ -> String -- ^ Path to certificate

208+ -> String -- ^ Path to private key

209+ -> IO AzureSetup

210+azureSetup sid certPath pkeyPath = do

211+ cert <- fileReadCertificate certPath

212+ pkey <- fileReadPrivateKey pkeyPath

213+ return AzureSetup {

214+ subscriptionId = sid

215+ , certificate = cert

216+ , privateKey = pkey

217+ , baseUrl = "https://management.core.windows.net"

218+ }

219+

220+--------------------------------------------------------------------------------

221+-- High-level API --

222+--------------------------------------------------------------------------------

223+

224+-- | Get a list of virtual machines

225+cloudServices :: AzureSetup -> IO [CloudService]

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

227+ services <- exec hostedServicesRequest

228+ forM services $ \service -> do

229+ roles <- exec AzureRequest {

230+ relativeUrl = "/services/hostedservices/" ++ hostedServiceName service

231+ ++ "?embed-detail=true"

232+ , apiVersion = "2012-03-01"

233+ , parser = proc xml -> do

234+ role <- getXPathTrees "//Role[@type='PersistentVMRole']" -< xml

235+ name <- getText . getXPathTrees "/Role/RoleName/text()" -< role

236+ roleInst <- arr2A getXPathTrees -< ("//RoleInstance[RoleName='" ++ name ++ "']", xml)

237+ ip <- getText . getXPathTrees "/RoleInstance/IpAddress/text()" -< roleInst

238+ endpoints <- listA (parseEndpoint . getXPathTrees "//InputEndpoint") -< role

239+ id -< VirtualMachine name ip endpoints

240+ }

241+ return $ CloudService (hostedServiceName service) roles

242+

243+-- | Get list of available hosted services

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

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

246+

247+hostedServicesRequest :: AzureRequest HostedService

248+hostedServicesRequest = AzureRequest

249+ { relativeUrl = "/services/hostedservices"

250+ , apiVersion = "2012-03-01"

251+ , parser = arr HostedService

252+ . getText

253+ . getXPathTrees "//ServiceName/text()"

254+ }

255+

256+parseEndpoint :: ArrowXml t => t XmlTree Endpoint

257+parseEndpoint = proc endpoint -> do

258+ name <- getText . getXPathTrees "//Name/text()" -< endpoint

259+ port <- getText . getXPathTrees "//Port/text()" -< endpoint

260+ vip <- getText . getXPathTrees "//Vip/text()" -< endpoint

261+ id -< Endpoint name port vip

262+

263+--------------------------------------------------------------------------------

264+-- Low-level API --

265+--------------------------------------------------------------------------------

266+

267+data AzureRequest c = AzureRequest {

268+ relativeUrl :: String

269+ , apiVersion :: String

270+ , parser :: IOSArrow XmlTree c

271+ }

272+

273+azureExecute :: AzureSetup -> ((forall b. AzureRequest b -> ResourceT IO [b]) -> ResourceT IO a) -> IO a

274+azureExecute setup f = withManager (\manager -> f (go manager))

275+ where

276+ go :: Manager -> forall b. AzureRequest b -> ResourceT IO [b]

277+ go manager request = do

278+ req <- parseUrl $ baseUrl setup

279+ ++ "/" ++ subscriptionId setup

280+ ++ "/" ++ relativeUrl request

281+ let req' = req {

282+ clientCertificates = [ (certificate setup, Just $ privateKey setup) ]

283+ , requestHeaders = [ (CI.mk $ BSC.pack "x-ms-version", BSC.pack $ apiVersion request)

284+ , (CI.mk $ BSC.pack "content-type", BSC.pack "application/xml")

285+ ]

286+ }

287+ Response _ _ _ lbs <- httpLbs req' manager

288+ liftIO . runX $ proc _ -> do

289+ xml <- readString [withValidate no] (BSLC.unpack lbs) -< ()

290+ -- arrIO putStrLn . writeDocumentToString [withIndent yes] -< xml

291+ parser request -< xml

benchmarks/ProcessRing.hsdeleted−117

@@ -1,117 +0,0 @@

1−{- ProcessRing benchmarks.

2−

3−To run the benchmarks, select a value for the ring size (sz) and

4−the number of times to send a message around the ring

5−

6−-}

7−

8−{-# LANGUAGE BangPatterns #-}

9−{-# LANGUAGE ScopedTypeVariables #-}

10−

11−import Control.Monad

12−import Control.Distributed.Process hiding (catch)

13−import Control.Distributed.Process.Node

14−import Control.Exception (catch, SomeException)

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

16−import System.Environment

17−import System.Console.GetOpt

18−

19−data Options = Options

20− { optRingSize :: Int

21− , optIterations :: Int

22− , optForward :: Bool

23− , optParallel :: Bool

24− , optUnsafe :: Bool

25− } deriving Show

26−

27−initialProcess :: Options -> Process ()

28−initialProcess op =

29− let ringSz = optRingSize op

30− msgCnt = optIterations op

31− fwd = optForward op

32− unsafe = optUnsafe op

33− msg = ("foobar", "baz")

34− in do

35− self <- getSelfPid

36− ring <- makeRing fwd unsafe ringSz self

37− forM_ [1..msgCnt] (\_ -> send ring msg)

38− collect msgCnt

39− where relay fsend pid = do

40− msg <- expect :: Process (String, String)

41− fsend pid msg

42− relay fsend pid

43−

44− forward' pid =

45− receiveWait [ matchAny (\m -> forward m pid) ] >> forward' pid

46−

47− makeRing :: Bool -> Bool -> Int -> ProcessId -> Process ProcessId

48− makeRing !f !u !n !pid

49− | n == 0 = go f u pid

50− | otherwise = go f u pid >>= makeRing f u (n - 1)

51−

52− go :: Bool -> Bool -> ProcessId -> Process ProcessId

53− go False False next = spawnLocal $ relay send next

54− go False True next = spawnLocal $ relay unsafeSend next

55− go True _ next = spawnLocal $ forward' next

56−

57− collect :: Int -> Process ()

58− collect !n

59− | n == 0 = return ()

60− | otherwise = do

61− receiveWait [

62− matchIf (\(a, b) -> a == "foobar" && b == "baz")

63− (\_ -> return ())

64− , matchAny (\_ -> error "unexpected input!")

65− ]

66− collect (n - 1)

67−

68−defaultOptions :: Options

69−defaultOptions = Options

70− { optRingSize = 10

71− , optIterations = 100

72− , optForward = False

73− , optParallel = False

74− , optUnsafe = False

75− }

76−

77−options :: [OptDescr (Options -> Options)]

78−options =

79− [ Option ['s'] ["ring-size"] (OptArg optSz "SIZE") "# of processes in ring"

80− , Option ['i'] ["iterations"] (OptArg optMsgCnt "ITER") "# of times to send"

81− , Option ['f'] ["forward"]

82− (NoArg (\opts -> opts { optForward = True }))

83− "use `forward' instead of send - default = False"

84− , Option ['u'] ["unsafe-send"]

85− (NoArg (\opts -> opts { optUnsafe = True }))

86− "use 'unsafeSend' (ignored with -f) - default = False"

87− , Option ['p'] ["parallel"]

88− (NoArg (\opts -> opts { optParallel = True }))

89− "send in parallel and consume sequentially - default = False"

90− ]

91−

92−optMsgCnt :: Maybe String -> Options -> Options

93−optMsgCnt Nothing opts = opts

94−optMsgCnt (Just c) opts = opts { optIterations = ((read c) :: Int) }

95−

96−optSz :: Maybe String -> Options -> Options

97−optSz Nothing opts = opts

98−optSz (Just s) opts = opts { optRingSize = ((read s) :: Int) }

99−

100−parseArgv :: [String] -> IO (Options, [String])

101−parseArgv argv = do

102− pn <- getProgName

103− case getOpt Permute options argv of

104− (o,n,[] ) -> return (foldl (flip id) defaultOptions o, n)

105− (_,_,errs) -> ioError (userError (concat errs ++ usageInfo (header pn) options))

106− where header pn' = "Usage: " ++ pn' ++ " [OPTION...]"

107−

108−main :: IO ()

109−main = do

110− argv <- getArgs

111− (opt, _) <- parseArgv argv

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

113− Right transport <- createTransport

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

115− node <- newLocalNode transport initRemoteTable

116− catch (void $ runProcess node $ initialProcess opt)

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

benchmarks/erlang/ring.erldeleted−62

@@ -1,62 +0,0 @@

1−%% can be run via escript if desired...

2−-module(ring).

3−

4−-export([main/1]).

5−

6−-record(opts, {

7− ring_size = 1 :: non_neg_integer(),

8− iterations = 1 :: non_neg_integer()

9−}).

10−

11−main(Argv) when is_list(Argv) ->

12− Opts = parse_args(Argv),

13− {Time, ok} = timer:tc(fun() -> run(Opts) end),

14− io:format("elapsed (wallclock) time: ~pms~n", [Time / 1000]).

15−

16−run(#opts{ ring_size=RingSize, iterations=SendCount }) ->

17− Self = self(),

18− Msg = {"foobar", "baz"},

19− Pid = make_ring(RingSize, Self),

20− [Pid ! Msg || _ <- lists:seq(1, SendCount)],

21− collect(SendCount).

22−

23−collect(0) ->

24− ok;

25−collect(N) ->

26− receive

27− {"foobar", "baz"} -> collect(N - 1);

28− Other -> exit({unexpected_message, Other})

29− end.

30−

31−-spec(make_ring(integer(), pid()) -> pid()).

32−make_ring(0, NextPid) ->

33− go(NextPid);

34−make_ring(NumProcs, NextPid) ->

35− make_ring(NumProcs - 1, go(NextPid)).

36−

37−-spec(go(pid()) -> pid()).

38−go(NextPid) ->

39− spawn(fun() -> relay(NextPid) end).

40−

41−-spec(relay(pid()) -> no_return()).

42−relay(NextPid) ->

43− receive

44− {_, _}=Msg -> NextPid ! Msg, relay(NextPid)

45− end.

46−

47−-spec(parse_args([string()]) -> #opts{}).

48−parse_args(Argv) ->

49− lists:foldl(fun parse_args/2, #opts{}, Argv).

50−

51−parse_args(("-r" ++ Val), Opts) ->

52− Sz = check_positive(ring_size, list_to_integer(Val)),

53− Opts#opts{ ring_size = Sz };

54−parse_args(("-i" ++ Val), Opts) ->

55− Iter = check_positive(iterations, list_to_integer(Val)),

56− Opts#opts{ iterations = Iter }.

57−

58−check_positive(K, N) when N < 1 ->

59− io:format("~p must be >= 1~n", [K]),

60− error({K, N});

61−check_positive(_, N) ->

62− N.

distributed-process-tests/LICENSE → distributed-process-azure/LICENSErenamed

The content is the same.

distributed-process-tests/Setup.hs → distributed-process-azure/Setup.hsrenamed

The content is the same.

distributed-process-azure/demos/Echo.hsnew file+49

@@ -0,0 +1,49 @@

1+{-# LANGUAGE TemplateHaskell #-}

2+

3+import System.IO (hFlush, stdout)

4+import System.Environment (getArgs)

5+import Control.Monad (unless, forever)

6+import Control.Monad.IO.Class (liftIO)

7+import Control.Distributed.Process (Process, expect)

8+import Control.Distributed.Process.Closure (remotable, mkClosure)

9+import Control.Distributed.Process.Backend.Azure

10+

11+echoRemote :: () -> Backend -> Process ()

12+echoRemote () _backend = forever $ do

13+ str <- expect

14+ remoteSend (str :: String)

15+

16+remotable ['echoRemote]

17+

18+echoLocal :: LocalProcess ()

19+echoLocal = do

20+ str <- liftIO $ putStr "# " >> hFlush stdout >> getLine

21+ unless (null str) $ do

22+ localSend str

23+ liftIO $ putStr "Echo: " >> hFlush stdout

24+ echo <- localExpect

25+ liftIO $ putStrLn echo

26+ echoLocal

27+

28+main :: IO ()

29+main = do

30+ args <- getArgs

31+ case args of

32+ "onvm":args' ->

33+ -- Pass execution to 'onVmMain' if we are running on the VM

34+ -- ('callOnVM' will provide the right arguments)

35+ onVmMain __remoteTable args'

36+

37+ sid:x509:pkey:user:cloudService:virtualMachine:port:_ -> do

38+ -- Initialize the Azure backend

39+ params <- defaultAzureParameters sid x509 pkey

40+ let params' = params { azureSshUserName = user }

41+ backend <- initializeBackend params' cloudService

42+

43+ -- Find the specified virtual machine

44+ Just vm <- findNamedVM backend virtualMachine

45+

46+ -- Run the echo client proper

47+ callOnVM backend vm port $

48+ ProcessPair ($(mkClosure 'echoRemote) ())

49+ echoLocal

distributed-process-azure/demos/Fib.hsnew file+74

@@ -0,0 +1,74 @@

1+{-# LANGUAGE TemplateHaskell #-}

2+import System.Environment (getArgs)

3+import System.Random (randomIO)

4+import Control.Monad (forM)

5+import Control.Monad.IO.Class (liftIO)

6+import Control.Distributed.Process

7+ ( Process

8+ , NodeId

9+ , SendPort

10+ , newChan

11+ , sendChan

12+ , spawn

13+ , receiveChan

14+ , spawnLocal

15+ )

16+import Control.Distributed.Process.Backend.Azure

17+import Control.Distributed.Process.Closure

18+ ( remotable

19+ , remotableDecl

20+ , mkClosure

21+ )

22+

23+randomElement :: [a] -> IO a

24+randomElement xs = do

25+ ix <- randomIO

26+ return (xs !! (ix `mod` length xs))

27+

28+remotableDecl [

29+ [d| dfib :: ([NodeId], SendPort Integer, Integer) -> Process () ;

30+ dfib (_, reply, 0) = sendChan reply 0

31+ dfib (_, reply, 1) = sendChan reply 1

32+ dfib (nids, reply, n) = do

33+ nid1 <- liftIO $ randomElement nids

34+ nid2 <- liftIO $ randomElement nids

35+ (sport, rport) <- newChan

36+ _ <- spawn nid1 $ $(mkClosure 'dfib) (nids, sport, n - 2)

37+ _ <- spawn nid2 $ $(mkClosure 'dfib) (nids, sport, n - 1)

38+ n1 <- receiveChan rport

39+ n2 <- receiveChan rport

40+ sendChan reply $ n1 + n2

41+ |]

42+ ]

43+

44+remoteFib :: ([NodeId], Integer) -> Backend -> Process ()

45+remoteFib (nids, n) _backend = do

46+ (sport, rport) <- newChan

47+ _ <- spawnLocal $ dfib (nids, sport, n)

48+ fib_n <- receiveChan rport

49+ mapM_ terminateNode nids

50+ remoteSend fib_n

51+

52+remotable ['remoteFib]

53+

54+printResult :: LocalProcess ()

55+printResult = do

56+ result <- localExpect :: LocalProcess Integer

57+ liftIO $ print result

58+

59+main :: IO ()

60+main = do

61+ args <- getArgs

62+ case args of

63+ "onvm":args' -> onVmMain (__remoteTable . __remoteTableDecl) args'

64+ [sid, x509, pkey, user, cloudService, n] -> do

65+ params <- defaultAzureParameters sid x509 pkey

66+ let params' = params { azureSshUserName = user }

67+ backend <- initializeBackend params' cloudService

68+ vms <- findVMs backend

69+ nids <- forM vms $ \vm -> spawnNodeOnVM backend vm "8080"

70+ callOnVM backend (head vms) "8081" $

71+ ProcessPair ($(mkClosure 'remoteFib) (nids, read n :: Integer))

72+ printResult

73+ _ ->

74+ error "Invalid command line arguments"

distributed-process-azure/demos/Ping.hsnew file+85

@@ -0,0 +1,85 @@

1+{-# LANGUAGE TemplateHaskell #-}

2+

3+import System.Environment (getArgs)

4+import Data.Binary (encode, decode)

5+import Control.Monad (forever)

6+import Control.Monad.IO.Class (liftIO)

7+import Control.Exception (try, IOException)

8+import Control.Distributed.Process

9+ ( Process

10+ , getSelfPid

11+ , expect

12+ , send

13+ , monitor

14+ , receiveWait

15+ , match

16+ , ProcessMonitorNotification(..)

17+ )

18+import Control.Distributed.Process.Closure (remotable, mkClosure)

19+import Control.Distributed.Process.Backend.Azure

20+import qualified Data.ByteString.Lazy as BSL (readFile, writeFile)

21+

22+pingServer :: () -> Backend -> Process ()

23+pingServer () _backend = do

24+ us <- getSelfPid

25+ liftIO $ BSL.writeFile "pingServer.pid" (encode us)

26+ forever $ do

27+ them <- expect

28+ send them ()

29+

30+pingClientRemote :: () -> Backend -> Process ()

31+pingClientRemote () _backend = do

32+ mPingServerEnc <- liftIO $ try (BSL.readFile "pingServer.pid")

33+ case mPingServerEnc of

34+ Left err ->

35+ remoteSend $ "Ping server not found: " ++ show (err :: IOException)

36+ Right pingServerEnc -> do

37+ let pingServerPid = decode pingServerEnc

38+ pid <- getSelfPid

39+ _ref <- monitor pingServerPid

40+ send pingServerPid pid

41+ gotReply <- receiveWait

42+ [ match (\() -> return True)

43+ , match (\(ProcessMonitorNotification {}) -> return False)

44+ ]

45+ if gotReply

46+ then remoteSend $ "Ping server at " ++ show pingServerPid ++ " ok"

47+ else remoteSend $ "Ping server at " ++ show pingServerPid ++ " failure"

48+

49+remotable ['pingClientRemote, 'pingServer]

50+

51+pingClientLocal :: LocalProcess ()

52+pingClientLocal = localExpect >>= liftIO . putStrLn

53+

54+main :: IO ()

55+main = do

56+ args <- getArgs

57+ case args of

58+ "onvm":args' ->

59+ -- Pass execution to 'onVmMain' if we are running on the VM

60+ onVmMain __remoteTable args'

61+

62+ "list":sid:x509:pkey:_ -> do

63+ -- List all available cloud services

64+ -- (useful, but not strictly necessary for the example)

65+ params <- defaultAzureParameters sid x509 pkey

66+ css <- cloudServices (azureSetup params)

67+ mapM_ print css

68+

69+ cmd:sid:x509:pkey:user:cloudService:virtualMachine:port:_ -> do

70+ -- Initialize the backend and find the right VM

71+ params <- defaultAzureParameters sid x509 pkey

72+ let params' = params { azureSshUserName = user }

73+ backend <- initializeBackend params' cloudService

74+ Just vm <- findNamedVM backend virtualMachine

75+

76+ -- The same binary can behave as the client or the server,

77+ -- depending on the command line arguments

78+ case cmd of

79+ "server" -> do

80+ pid <- spawnOnVM backend vm port ($(mkClosure 'pingServer) ())

81+ putStrLn $ "Ping server started at " ++ show pid

82+ "client" ->

83+ callOnVM backend vm port $

84+ ProcessPair ($(mkClosure 'pingClientRemote) ())

85+ pingClientLocal

distributed-process-azure/distributed-process-azure.cabalnew file+95

@@ -0,0 +1,95 @@

1+Name: distributed-process-azure

2+Version: 0.1.0

3+Cabal-Version: >=1.8

4+Build-Type: Simple

5+License: BSD3

6+License-File: LICENSE

7+Copyright: Well-Typed LLP

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

9+Maintainer: edsko@well-typed.com, dcoutts@well-typed.com

10+Stability: experimental

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

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

13+Synopsis: Microsoft Azure backend for Cloud Haskell

14+Category: Control

15+

16+Source-Repository head

17+ Type: git

18+ Location: https://github.com/haskell-distributed/distributed-process

19+ SubDir: distributed-process-azure

20+

21+Flag build-demos

22+ description: Build the demos

23+ default: False

24+

25+Library

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

27+ azure-service-api >= 0.1 && < 0.2,

28+ filepath >= 1.3 && < 1.4,

29+ executable-path >= 0.0.3 && < 0.1,

30+ libssh2 >= 0.2 && < 0.3,

31+ pureMD5 >= 2.1 && < 2.2,

32+ bytestring >= 0.9 && < 0.11,

33+ distributed-process >= 0.3.2 && < 0.4,

34+ binary >= 0.5 && < 0.6,

35+ network-transport-tcp >= 0.3 && < 0.4,

36+ optparse-applicative >= 0.2 && < 0.4,

37+ transformers >= 0.3 && < 0.4,

38+ certificate >= 1.2.4 && < 1.3,

39+ unix >= 2.5 && < 2.6,

40+ network-transport >= 0.3 && < 0.4,

41+ mtl >= 2.1 && < 2.2,

42+ rank1dynamic >= 0.1 && < 0.2,

43+ distributed-static >= 0.2 && < 0.3

44+ Exposed-modules: Control.Distributed.Process.Backend.Azure

45+ Extensions: ViewPatterns,

46+ RankNTypes,

47+ ExistentialQuantification,

48+ ScopedTypeVariables,

49+ DeriveDataTypeable,

50+ GeneralizedNewtypeDeriving

51+ ghc-options: -Wall

52+ HS-Source-Dirs: src

53+

54+Executable cloud-haskell-azure-echo

55+ if flag(build-demos)

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

57+ distributed-process-azure >= 0.1 && < 0.2,

58+ distributed-process >= 0.3.2 && < 0.4,

59+ transformers >= 0.3 && < 0.4

60+ else

61+ buildable: False

62+ Main-Is: demos/Echo.hs

63+ ghc-options: -Wall

64+

65+Executable cloud-haskell-azure-ping

66+ if flag(build-demos)

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

68+ distributed-process-azure >= 0.1 && < 0.2,

69+ distributed-process >= 0.3.2 && < 0.4,

70+ transformers >= 0.3 && < 0.4,

71+ bytestring >= 0.9 && < 0.11,

72+ binary >= 0.5 && < 0.6,

73+ mtl,

74+ libssh2

75+ else

76+ buildable: False

77+ Main-Is: demos/Ping.hs

78+ ghc-options: -Wall

79+

80+Executable cloud-haskell-azure-fib

81+ if flag(build-demos)

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

83+ distributed-process-azure >= 0.1 && < 0.2,

84+ distributed-process >= 0.3.2 && < 0.4,

85+ transformers >= 0.3 && < 0.4,

86+ bytestring >= 0.9 && < 0.11,

87+ binary >= 0.5 && < 0.6,

88+ binary-generic >= 0.2 && < 0.3,

89+ rank1dynamic >= 0.1 && < 0.2,

90+ distributed-static >= 0.2 && < 0.3,

91+ random >= 1.0 && < 1.1

92+ else

93+ buildable: False

94+ Main-Is: demos/Fib.hs

95+ ghc-options: -Wall

distributed-process-azure/src/Control/Distributed/Process/Backend/Azure.hsnew file+964

@@ -0,0 +1,964 @@

1+-- | This module provides the API for running Cloud Haskell on Microsoft Azure

2+-- virtual machines (<http://www.windowsazure.com>). Virtual machines within an

3+-- Azure cloud service can talk to each other directly using standard Cloud

4+-- Haskell primitives (using TCP/IP under the hood); to talk to the remote

5+-- machines from your local machine you can use the primitives provided in this

6+-- module (which use ssh under the hood). It looks something like

7+--

8+-- > _ _

9+-- > ( ` )_

10+-- > ( ) `) Azure cloud service

11+-- > (_ (_ . _) _)

12+-- >

13+-- > |

14+-- > | ssh connection

15+-- > |

16+-- >

17+-- > +---+

18+-- > | | Local machine

19+-- > +---+

20+--

21+-- /NOTE/: It is unfortunate that the local machine cannot talk to the remote

22+-- machine using the standard Cloud Haskell primitives. In an ideal world, we

23+-- could just start a Cloud Haskell node on the local machine, too.

24+-- Unfortunately, Cloud Haskell does not yet support using multiple network

25+-- transports within the same system (i.e. both TCP/IP and SSH). This is a

26+-- temporary workaround.

27+--

28+-- [Azure Setup]

29+--

30+-- In this section we describe how to set up an Azure Cloud Service for use

31+-- with Cloud Haskell, starting from a brand new Azure account. It is not

32+-- intended as an Azure tutorial, but as a guide to making the right choices to

33+-- get Cloud Haskell up and running as quickly as possible.

34+--

35+-- An Azure /Cloud Service/ is a set of virtual machines that can talk to each

36+-- other directly over TCP/IP (they are part of the same private network). You

37+-- don't create the cloud service directly; instead, after you have set up your

38+-- first virtual machine as a /stand alone/ virtual machine, you can /connect/

39+-- subsequent virtual machines to the first virtual machine, thereby implicitly

40+-- setting up a Cloud Service.

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

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

48+--

49+-- When you set up your virtual machine, you can pick an arbitrary virtual

50+-- machine name; these names are for your own use only and do not need to be

51+-- globally unique. Set a username and password; you should use the same

52+-- username on all virtual machines. You should also upload an SSH key for

53+-- authentication (see

54+-- /Converting OpenSSH keys for use on Windows Azure Linux VM's/,

55+-- <http://utlemming.azurewebsites.net/?p=91>, for

56+-- information on how to convert a standard Linux @id_rsa.pub@ public key to

57+-- X509 format suitable for Azure). For the first VM you create select

58+-- /Standalone Virtual Machine/, and pick an appropriate DNS name. The DNS name

59+-- /does/ have to be globally unique, and will also be the name of the Cloud

60+-- Service. For subsequent virtual machines, select

61+-- /Connect to Existing Virtual Machine/ instead and then select the first VM

62+-- you created.

63+--

64+-- Once your virtual machines have been set up, you have to make sure that the

65+-- user you created when you created the VM can ssh from any virtual machine to

66+-- any other using public key authentication. Moreover, you have to make sure

67+-- that @libssh2@ is installed; if you are using the Ubuntu image you can

68+-- install @libssh2@ using

69+--

70+-- > sudo apt-get install libssh2-1

71+--

72+-- (TODO: if you don't install libssh2 things will break without a clear error

73+-- message.)

74+--

75+-- In these notes, we assume three virtual machines called @CHDemo1@,

76+-- @CHDemo2@, and @CHDemo3@, all part of a @CloudHaskellDemo@ cloud service.

77+--

78+-- [Obtaining a Management Certificate]

79+--

80+-- Azure authentication is by means of an X509 certificate and corresponding

81+-- private key. /Create management certificates for Linux in Windows Azure/,

82+-- <https://www.windowsazure.com/en-us/manage/linux/common-tasks/manage-certificates/>,

83+-- describes how you can create a management certificate for Azure, download it

84+-- as a @.publishsettings@ file, and extract an @.pfx@ file from it. You cannot

85+-- use this @.pfx@ directly; instead, you will need to extract an X509

86+-- certificate from it and a private key in suitable format. You can use the

87+-- @openssl@ command line tool for both tasks; assuming that you stored the

88+-- @.pfx@ file as @credentials.pfx@, to extract the X509 certificate:

89+--

90+-- > openssl pkcs12 -in credentials.pfx -nokeys -out credentials.x509

91+--

92+-- And to extract the private key:

93+--

94+-- > openssl pkcs12 -in credentials.pfx -nocerts -nodes | openssl rsa -out credentials.private

95+--

96+-- (@openssl pkcs12@ outputs the private key in PKCS#8 format (BEGIN PRIVATE

97+-- KEY), but we need it in PKCS#1 format (BEGIN RSA PRIVATE KEY).

98+--

99+-- [Testing the Setup]

100+--

101+-- Build and install the @distributed-process-azure@ package, making sure to

102+-- pass the @build-demos@ flag to Cabal.

103+--

104+-- > cabal-dev install distributed-process-azure -f build-demos

105+--

106+-- We can the @cloud-haskell-azure-ping@ demo to test our setup:

107+--

108+-- > cloud-haskell-azure-ping list \

109+-- > <<your subscription ID>> \

110+-- > /path/to/credentials.x509 \

111+-- > /path/to/credentials.private

112+--

113+-- (you can find your subscription ID in the @.publishsettings@ file from the previous step).

114+-- If everything went well, this will output something like

115+--

116+-- > Cloud Service "CloudHaskellDemo"

117+-- > VIRTUAL MACHINES

118+-- > Virtual Machine "CHDemo3"

119+-- > IP 10.119.182.127

120+-- > INPUT ENDPOINTS

121+-- > Input endpoint "SSH"

122+-- > Port 50136

123+-- > VIP 168.63.31.38

124+-- > Virtual Machine "CHDemo2"

125+-- > IP 10.59.238.125

126+-- > INPUT ENDPOINTS

127+-- > Input endpoint "SSH"

128+-- > Port 63365

129+-- > VIP 168.63.31.38

130+-- > Virtual Machine "CHDemo1"

131+-- > IP 10.59.224.122

132+-- > INPUT ENDPOINTS

133+-- > Input endpoint "SSH"

134+-- > Port 22

135+-- > VIP 168.63.31.38

136+--

137+-- The IP addresses listed are /internal/ IP addresses; they can be used by the

138+-- virtual machines to talk to each other, but not by the outside world to talk

139+-- to the virtual machines. To do that, you will need to use the VIP (Virtual

140+-- IP) address instead, which you will notice is the same for all virtual

141+-- machines that are part of the cloud service. The corresponding DNS name

142+-- (here @CloudHaskellDemo.cloudapp.net@) will also resolve to this (V)IP

143+-- address. To login to individual machines (through SSH) you will need to use

144+-- the specific port mentioned under INPUT ENDPOINTS.

145+--

146+-- [Overview of the API]

147+--

148+-- The Azure 'Backend' provides low-level functionality for interacting with

149+-- Azure virtual machines. 'findVMs' finds all currently available virtual

150+-- machines; 'copyToVM' copies the executable to a specified VM (recall that

151+-- all VMs, as well as the local machine, are assumed to run the same OS so

152+-- that they can all run the same binary), and 'checkMD5' checks the MD5 hash

153+-- of the executable on a remote machine.

154+--

155+-- 'callOnVM' and 'spawnOnVM' deal with setting up Cloud Haskell nodes.

156+-- 'spawnOnVM' takes a virtual machine and a port number, as well as a

157+-- @RemoteProcess ()@, starts the executable on the remote node, sets up a new

158+-- Cloud Haskell node, and then runs the specified process. The Cloud Haskell

159+-- node will be shut down when the given process terminates. 'RemoteProcess' is

160+-- defined as

161+--

162+-- > type RemoteProcess a = Closure (Backend -> Process a)

163+--

164+-- (If you don't know what a 'Closure' is you should read

165+-- "Control.Distributed.Process.Closure".); the remote process will be supplied

166+-- with an Azure backend initialized with the same parameters. 'spawnOnVM'

167+-- returns once the Cloud Haskell node has been set up.

168+--

169+-- 'callOnVM' is similar to 'spawnOnVM', but it takes a /pair/ of processes:

170+-- one to run on the remote host (on a newly created Cloud Haskell node), and

171+-- one to run on the local machine. In this case, the new Cloud Haskell node

172+-- will be terminated when the /local/ process terminates. 'callOnVM' is useful

173+-- because the remote process and the local process can communicate through a

174+-- set of primitives provided in this module ('localSend', 'localExpect', and

175+-- 'remoteSend' -- there is no 'remoteExpect'; instead the remote process can

176+-- use the standard Cloud Haskell 'expect' primitive).

177+--

178+-- [First Example: Echo]

179+--

180+-- When we run the @cloud-haskell-azure-echo@ demo on our local machine, it

181+-- starts a new Cloud Haskell node on the specified remote virtual machine. It

182+-- then repeatedly waits for input from the user on the local machine, sends

183+-- this to the remote virtual machine which will echo it back, and wait for and

184+-- show the echo.

185+--

186+-- Before you can try it you will first need to copy the executable (for

187+-- example, using scp, although the Azure backend also provides this natively

188+-- in Haskell). Once that's done, you can run the demo as follows:

189+--

190+-- > cloud-haskell-azure-echo \

191+-- > <<subscription ID>> \

192+-- > /path/to/credentials.x509 \

193+-- > /path/to/credentials.private \

194+-- > <<remote username>> \

195+-- > <<cloud service name>> \

196+-- > <<virtual machine name>> \

197+-- > <<port number>>

198+-- > # Everything I type gets echoed back

199+-- > Echo: Everything I type gets echoed back

200+-- > # Until I enter a blank line

201+-- > Echo: Until I enter a blank line

202+-- > #

203+--

204+-- The full @echo@ demo is

205+--

206+-- > {-# LANGUAGE TemplateHaskell #-}

207+-- >

208+-- > import System.IO (hFlush, stdout)

209+-- > import System.Environment (getArgs)

210+-- > import Control.Monad (unless, forever)

211+-- > import Control.Monad.IO.Class (liftIO)

212+-- > import Control.Distributed.Process (Process, expect)

213+-- > import Control.Distributed.Process.Closure (remotable, mkClosure)

214+-- > import Control.Distributed.Process.Backend.Azure

215+-- >

216+-- > echoRemote :: () -> Backend -> Process ()

217+-- > echoRemote () _backend = forever $ do

218+-- > str <- expect

219+-- > remoteSend (str :: String)

220+-- >

221+-- > remotable ['echoRemote]

222+-- >

223+-- > echoLocal :: LocalProcess ()

224+-- > echoLocal = do

225+-- > str <- liftIO $ putStr "# " >> hFlush stdout >> getLine

226+-- > unless (null str) $ do

227+-- > localSend str

228+-- > liftIO $ putStr "Echo: " >> hFlush stdout

229+-- > echo <- localExpect

230+-- > liftIO $ putStrLn echo

231+-- > echoLocal

232+-- >

233+-- > main :: IO ()

234+-- > main = do

235+-- > args <- getArgs

236+-- > case args of

237+-- > "onvm":args' ->

238+-- > -- Pass execution to 'onVmMain' if we are running on the VM

239+-- > -- ('callOnVM' will provide the right arguments)

240+-- > onVmMain __remoteTable args'

241+-- >

242+-- > sid:x509:pkey:user:cloudService:virtualMachine:port:_ -> do

243+-- > -- Initialize the Azure backend

244+-- > params <- defaultAzureParameters sid x509 pkey

245+-- > let params' = params { azureSshUserName = user }

246+-- > backend <- initializeBackend params' cloudService

247+-- >

248+-- > -- Find the specified virtual machine

249+-- > Just vm <- findNamedVM backend virtualMachine

250+-- >

251+-- > -- Run the echo client proper

252+-- > callOnVM backend vm port $

253+-- > ProcessPair ($(mkClosure 'echoRemote) ())

254+-- > echoLocal

255+--

256+-- The most important part of this code is the last three lines

257+--

258+-- > callOnVM backend vm port $

259+-- > ProcessPair ($(mkClosure 'echoRemote) ())

260+-- > echoLocal

261+--

262+-- 'callOnVM' creats a new Cloud Haskell node on the specified virtual machine,

263+-- then runs @echoRemote@ on the remote machine and @echoLocal@ on the local

264+-- machine.

265+--

266+-- [Second Example: Ping]

267+--

268+-- The second example differs from the @echo@ demo in that it uses both

269+-- 'callOnVM' and 'spawnOnVM'. It uses the latter to

270+-- install a ping server which keeps running in the background; it uses the

271+-- former to run a ping client which sends a request to the ping server and

272+-- outputs the response.

273+--

274+-- As with the @echo@ demo, make sure to copy the executable to the remote server first.

275+-- Once that is done, you can start a ping server on a virtual machine using

276+--

277+-- > cloud-haskell-azure-ping server \

278+-- > <<subscription ID>> \

279+-- > /path/to/credentials.x509 \

280+-- > /path/to/credentials.private \

281+-- > <<remote username>> \

282+-- > <<cloud service name>> \

283+-- > <<virtual machine name>> \

284+-- > <<port number>>

285+--

286+-- As before, when we execute this on our local machine, it starts a new Cloud

287+-- Haskell node on the specified remote virtual machine and then executes the

288+-- ping server. Unlike with the echo example, however, this command will

289+-- terminate once the Cloud Haskell node has been set up, leaving the ping

290+-- server running in the background.

291+--

292+-- Once the ping server is running we can run the ping client:

293+--

294+-- > cloud-haskell-azure-ping client \

295+-- > <<subscription ID>> \

296+-- > /path/to/credentials.x509 \

297+-- > /path/to/credentials.private \

298+-- > <<remote username>> \

299+-- > <<cloud service name>> \

300+-- > <<virtual machine name>> \

301+-- > <<DIFFERENT port number>>

302+-- > Ping server at pid://10.59.224.122:8080:0:2 ok

303+--

304+-- Note that we must pass a different port number, because the client will run

305+-- within its own Cloud Haskell instance.

306+--

307+-- The code for the @ping@ demo is similar to the @echo@ demo, but uses both

308+-- 'callOnVM' and 'spawnOnVM' and demonstrates a way to discover processes (in

309+-- this case, through a PID file).

310+--

311+-- > {-# LANGUAGE TemplateHaskell #-}

312+-- >

313+-- > import System.Environment (getArgs)

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

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

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

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

318+-- > import Control.Distributed.Process

319+-- > ( Process

320+-- > , getSelfPid

321+-- > , expect

322+-- > , send

323+-- > , monitor

324+-- > , receiveWait

325+-- > , match

326+-- > , ProcessMonitorNotification(..)

327+-- > )

328+-- > import Control.Distributed.Process.Closure (remotable, mkClosure)

329+-- > import Control.Distributed.Process.Backend.Azure

330+-- > import qualified Data.ByteString.Lazy as BSL (readFile, writeFile)

331+-- >

332+-- > pingServer :: () -> Backend -> Process ()

333+-- > pingServer () _backend = do

334+-- > us <- getSelfPid

335+-- > liftIO $ BSL.writeFile "pingServer.pid" (encode us)

336+-- > forever $ do

337+-- > them <- expect

338+-- > send them ()

339+-- >

340+-- > pingClientRemote :: () -> Backend -> Process ()

341+-- > pingClientRemote () _backend = do

342+-- > mPingServerEnc <- liftIO $ try (BSL.readFile "pingServer.pid")

343+-- > case mPingServerEnc of

344+-- > Left err ->

345+-- > remoteSend $ "Ping server not found: " ++ show (err :: IOException)

346+-- > Right pingServerEnc -> do

347+-- > let pingServerPid = decode pingServerEnc

348+-- > pid <- getSelfPid

349+-- > _ref <- monitor pingServerPid

350+-- > send pingServerPid pid

351+-- > gotReply <- receiveWait

352+-- > [ match (\() -> return True)

353+-- > , match (\(ProcessMonitorNotification {}) -> return False)

354+-- > ]

355+-- > if gotReply

356+-- > then remoteSend $ "Ping server at " ++ show pingServerPid ++ " ok"

357+-- > else remoteSend $ "Ping server at " ++ show pingServerPid ++ " failure"

358+-- >

359+-- > remotable ['pingClientRemote, 'pingServer]

360+-- >

361+-- > pingClientLocal :: LocalProcess ()

362+-- > pingClientLocal = localExpect >>= liftIO . putStrLn

363+-- >

364+-- > main :: IO ()

365+-- > main = do

366+-- > args <- getArgs

367+-- > case args of

368+-- > "onvm":args' ->

369+-- > -- Pass execution to 'onVmMain' if we are running on the VM

370+-- > onVmMain __remoteTable args'

371+-- >

372+-- > "list":sid:x509:pkey:_ -> do

373+-- > -- List all available cloud services

374+-- > -- (useful, but not strictly necessary for the example)

375+-- > params <- defaultAzureParameters sid x509 pkey

376+-- > css <- cloudServices (azureSetup params)

377+-- > mapM_ print css

378+-- >

379+-- > cmd:sid:x509:pkey:user:cloudService:virtualMachine:port:_ -> do

380+-- > -- Initialize the backend and find the right VM

381+-- > params <- defaultAzureParameters sid x509 pkey

382+-- > let params' = params { azureSshUserName = user }

383+-- > backend <- initializeBackend params' cloudService

384+-- > Just vm <- findNamedVM backend virtualMachine

385+-- >

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

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

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

392+-- > pingClientLocal

393+module Control.Distributed.Process.Backend.Azure

394+ ( -- * Initialization

395+ Backend(..)

396+ , AzureParameters(..)

397+ , defaultAzureParameters

398+ , initializeBackend

399+ -- * Utilities

400+ , findNamedVM

401+ -- * On-VM main

402+ , onVmMain

403+ -- * Re-exports from Azure Service Management

404+ , CloudService(..)

405+ , VirtualMachine(..)

406+ , Endpoint(..)

407+ , AzureSetup

408+ , Azure.cloudServices

409+ -- * Remote and local processes

410+ , ProcessPair(..)

411+ , RemoteProcess

412+ , LocalProcess

413+ , localSend

414+ , localExpect

415+ , remoteSend

416+ -- * High-level API

417+ , spawnNodeOnVM

418+ , terminateNode

419+ ) where

420+

421+import Prelude hiding (catch)

422+import System.Environment (getEnv)

423+import System.FilePath ((</>), takeFileName)

424+import System.Environment.Executable (getExecutablePath)

425+import System.IO

426+ ( stdout

427+ , hFlush

428+ , hSetBinaryMode

429+ , stdin

430+ , stdout

431+ , stderr

432+ , Handle

433+ , hClose

434+ )

435+import qualified System.Posix.Process as Posix (forkProcess, createSession)

436+import Data.Maybe (listToMaybe)

437+import Data.Binary (Binary(get, put), encode, decode, getWord8, putWord8)

438+import Data.Digest.Pure.MD5 (md5, MD5Digest)

439+import qualified Data.ByteString as BSS

440+ ( ByteString

441+ , length

442+ , concat

443+ , hPut

444+ , hGet

445+ )

446+import qualified Data.ByteString.Char8 as BSSC (pack)

447+import qualified Data.ByteString.Lazy as BSL

448+ ( ByteString

449+ , readFile

450+ , length

451+ , fromChunks

452+ , toChunks

453+ , hPut

454+ , hGet

455+ )

456+import Data.Typeable (Typeable)

457+import Data.Foldable (forM_)

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

459+import Control.Monad (void, when)

460+import Control.Monad.Reader (MonadReader, ReaderT, runReaderT, ask)

461+import Control.Exception

462+ ( Exception

463+ , catches

464+ , Handler(Handler)

465+ , throwIO

466+ , SomeException

467+ )

468+import Control.Monad.IO.Class (MonadIO, liftIO)

469+import Control.Concurrent.MVar (MVar, newEmptyMVar, readMVar, putMVar)

470+

471+-- Azure

472+import Network.Azure.ServiceManagement

473+ ( CloudService(..)

474+ , VirtualMachine(..)

475+ , Endpoint(..)

476+ , AzureSetup

477+ )

478+import qualified Network.Azure.ServiceManagement as Azure

479+ ( cloudServices

480+ , azureSetup

481+ , vmSshEndpoint

482+ )

483+

484+-- SSH

485+import qualified Network.SSH.Client.LibSSH2 as SSH

486+ ( withSSH2

487+ , scpSendFile

488+ , withChannelBy

489+ , Session

490+ , readAllChannel

491+ , writeAllChannel

492+ , Channel

493+ )

494+import qualified Network.SSH.Client.LibSSH2.Foreign as SSH

495+ ( openChannelSession

496+ , channelExecute

497+ , writeChannel

498+ , readChannel

499+ , channelSendEOF

500+ )

501+import qualified Network.SSH.Client.LibSSH2.Errors as SSH

502+ ( ErrorCode

503+ , NULL_POINTER

504+ , getLastError

505+ )

506+

507+-- CH

508+import Control.Distributed.Process

509+ ( Process

510+ , Closure

511+ , RemoteTable

512+ , catch

513+ , unClosure

514+ , ProcessId

515+ , getSelfPid

516+ , NodeId

517+ , processNodeId

518+ , register

519+ , expect

520+ , nsendRemote

521+ )

522+import Control.Distributed.Process.Serializable (Serializable)

523+import qualified Control.Distributed.Process.Internal.Types as CH

524+ ( LocalNode

525+ , LocalProcess(processQueue)

526+ , Message

527+ , payloadToMessage

528+ , messageToPayload

529+ , createMessage

530+ )

531+import Control.Distributed.Process.Node

532+ ( runProcess

533+ , forkProcess

534+ , newLocalNode

535+ , initRemoteTable

536+ )

537+import Control.Distributed.Process.Internal.CQueue (CQueue, enqueue)

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

539+import Network.Transport.Internal (encodeInt32, decodeInt32, prependLength)

540+

541+-- Static

542+import Control.Distributed.Static

543+ ( Static

544+ , registerStatic

545+ , staticClosure

546+ , staticLabel

547+ )

548+import Data.Rank1Dynamic (toDynamic)

549+

550+-- | Azure backend

551+data Backend = Backend {

552+ -- | Find virtual machines

553+ findVMs :: IO [VirtualMachine]

554+ -- | Copy the executable to a virtual machine

555+ , copyToVM :: VirtualMachine -> IO ()

556+ -- | Check the MD5 hash of the remote executable

557+ , checkMD5 :: VirtualMachine -> IO Bool

558+ -- | @runOnVM vm port pp@ starts a new CH node on machine @vm@ and then

559+ -- runs the specified process pair. The CH node will shut down when the

560+ -- /local/ process exists. @callOnVM@ returns the returned by the local

561+ -- process on exit.

562+ , callOnVM :: forall a. VirtualMachine -> String -> ProcessPair a -> IO a

563+ -- | Create a new CH node and run the specified process.

564+ -- The CH node will shut down when the /remote/ process exists. @spawnOnVM@

565+ -- returns as soon as the process has been spawned.

566+ , spawnOnVM :: VirtualMachine -> String -> RemoteProcess () -> IO ProcessId

567+ } deriving (Typeable)

568+

569+-- | Azure connection parameters

570+data AzureParameters = AzureParameters {

571+ azureSetup :: AzureSetup

572+ , azureSshUserName :: FilePath

573+ , azureSshPublicKey :: FilePath

574+ , azureSshPrivateKey :: FilePath

575+ , azureSshPassphrase :: String

576+ , azureSshKnownHosts :: FilePath

577+ , azureSshRemotePath :: FilePath

578+ , azureSshLocalPath :: FilePath

579+ }

580+

581+instance Binary AzureParameters where

582+ put params = do

583+ put (azureSetup params)

584+ put (azureSshUserName params)

585+ put (azureSshPublicKey params)

586+ put (azureSshPrivateKey params)

587+ put (azureSshPassphrase params)

588+ put (azureSshKnownHosts params)

589+ put (azureSshRemotePath params)

590+ put (azureSshLocalPath params)

591+ get =

592+ AzureParameters <$> get <*> get <*> get <*> get <*> get <*> get <*> get <*> get

593+

594+-- | Create default azure parameters

595+defaultAzureParameters :: String -- ^ Azure subscription ID

596+ -> FilePath -- ^ Path to X509 certificate

597+ -> FilePath -- ^ Path to private key

598+ -> IO AzureParameters

599+defaultAzureParameters sid x509 pkey = do

600+ home <- getEnv "HOME"

601+ user <- getEnv "USER"

602+ self <- getExecutablePath

603+ setup <- Azure.azureSetup sid x509 pkey

604+ return AzureParameters

605+ { azureSetup = setup

606+ , azureSshUserName = user

607+ , azureSshPublicKey = home </> ".ssh" </> "id_rsa.pub"

608+ , azureSshPrivateKey = home </> ".ssh" </> "id_rsa"

609+ , azureSshPassphrase = ""

610+ , azureSshKnownHosts = home </> ".ssh" </> "known_hosts"

611+ , azureSshRemotePath = takeFileName self

612+ , azureSshLocalPath = self

613+ }

614+

615+-- | Initialize the backend

616+initializeBackend :: AzureParameters -- ^ Connection parameters

617+ -> String -- ^ Cloud service name

618+ -> IO Backend

619+initializeBackend params cloudService =

620+ return Backend {

621+ findVMs = apiFindVMs params cloudService

622+ , copyToVM = apiCopyToVM params

623+ , checkMD5 = apiCheckMD5 params

624+ , callOnVM = apiCallOnVM params cloudService

625+ , spawnOnVM = apiSpawnOnVM params cloudService

626+ }

627+

628+-- | Find virtual machines

629+apiFindVMs :: AzureParameters -> String -> IO [VirtualMachine]

630+apiFindVMs params cloudService = do

631+ css <- Azure.cloudServices (azureSetup params)

632+ case filter ((== cloudService) . cloudServiceName) css of

633+ [cs] -> return $ cloudServiceVMs cs

634+ _ -> return []

635+

636+-- | Start a CH node on the given virtual machine

637+apiCopyToVM :: AzureParameters -> VirtualMachine -> IO ()

638+apiCopyToVM params vm =

639+ void . withSSH2 params vm $ \s -> catchSshError s $

640+ SSH.scpSendFile s 0o700 (azureSshLocalPath params) (azureSshRemotePath params)

641+

642+-- | Call a process on a VM

643+apiCallOnVM :: AzureParameters

644+ -> String

645+ -> VirtualMachine

646+ -> String

647+ -> ProcessPair a

648+ -> IO a

649+apiCallOnVM = runOnVM False

650+

651+apiSpawnOnVM :: AzureParameters

652+ -> String

653+ -> VirtualMachine

654+ -> String

655+ -> Closure (Backend -> Process ())

656+ -> IO ProcessId

657+apiSpawnOnVM params cloudService vm port rproc =

658+ runOnVM True params cloudService vm port $

659+ ProcessPair rproc localExpect

660+

661+-- | Internal generalization of 'spawnOnVM' and 'callOnVM'

662+runOnVM :: Bool

663+ -> AzureParameters

664+ -> String

665+ -> VirtualMachine

666+ -> String

667+ -> ProcessPair a

668+ -> IO a

669+runOnVM bg params cloudService vm port ppair =

670+ withSSH2 params vm $ \s -> do

671+ -- TODO: reduce duplication with apiCallOnVM

672+ let exe = "PATH=. " ++ azureSshRemotePath params

673+ ++ " onvm"

674+ ++ " " ++ vmIpAddress vm

675+ ++ " " ++ port

676+ ++ " " ++ cloudService

677+ ++ " " ++ show bg

678+ ++ " 2>&1"

679+ let paramsEnc = encode params

680+ let rprocEnc = encode (ppairRemote ppair)

681+ (status, r) <- SSH.withChannelBy (SSH.openChannelSession s) id $ \ch -> do

682+ SSH.channelExecute ch exe

683+ SSH.writeChannel ch (encodeInt32 (BSL.length rprocEnc))

684+ SSH.writeAllChannel ch rprocEnc

685+ SSH.writeChannel ch (encodeInt32 (BSL.length paramsEnc))

686+ SSH.writeAllChannel ch paramsEnc

687+ runLocalProcess (ppairLocal ppair) ch

688+ if status == 0

689+ then return r

690+ else error "runOnVM: Non-zero exit status" -- This would a bug

691+

692+-- | Check the MD5 hash of the executable on the remote machine

693+apiCheckMD5 :: AzureParameters -> VirtualMachine -> IO Bool

694+apiCheckMD5 params vm = do

695+ hash <- localHash params

696+ withSSH2 params vm $ \s -> do

697+ (r, _) <- SSH.withChannelBy (SSH.openChannelSession s) id $ \ch -> do

698+ SSH.channelExecute ch "md5sum -c --status"

699+ SSH.writeChannel ch . BSSC.pack $ show hash ++ " " ++ azureSshRemotePath params

700+ SSH.channelSendEOF ch

701+ SSH.readAllChannel ch

702+ return (r == 0)

703+

704+withSSH2 :: AzureParameters -> VirtualMachine -> (SSH.Session -> IO a) -> IO a

705+withSSH2 params (Azure.vmSshEndpoint -> Just ep) =

706+ SSH.withSSH2 (azureSshKnownHosts params)

707+ (azureSshPublicKey params)

708+ (azureSshPrivateKey params)

709+ (azureSshPassphrase params)

710+ (azureSshUserName params)

711+ (endpointVip ep)

712+ (read $ endpointPort ep)

713+withSSH2 _ vm =

714+ error $ "withSSH2: No SSH endpoint for virtual machine " ++ vmName vm

715+

716+catchSshError :: SSH.Session -> IO a -> IO a

717+catchSshError s io =

718+ catches io [ Handler handleErrorCode

719+ , Handler handleNullPointer

720+ ]

721+ where

722+ handleErrorCode :: SSH.ErrorCode -> IO a

723+ handleErrorCode _ = do

724+ (_, str) <- SSH.getLastError s

725+ error str

726+

727+ handleNullPointer :: SSH.NULL_POINTER -> IO a

728+ handleNullPointer _ = do

729+ (_, str) <- SSH.getLastError s

730+ error str

731+

732+localHash :: AzureParameters -> IO MD5Digest

733+localHash params = md5 <$> BSL.readFile (azureSshLocalPath params)

734+

735+--------------------------------------------------------------------------------

736+-- Utilities --

737+--------------------------------------------------------------------------------

738+

739+-- | Find a virtual machine with a particular name

740+findNamedVM :: Backend -> String -> IO (Maybe VirtualMachine)

741+findNamedVM backend vm =

742+ listToMaybe . filter ((== vm) . vmName) <$> findVMs backend

743+

744+--------------------------------------------------------------------------------

745+-- Local and remote processes --

746+--------------------------------------------------------------------------------

747+

748+-- | A process pair consists of a remote process and a local process. The local

749+-- process can send messages to the remote process using 'localSend' and wait

750+-- for messages from the remote process using 'localExpect'. The remote process

751+-- can send messages to the local process using 'remoteSend', and wait for

752+-- messages from the local process using the standard Cloud Haskell primitives.

753+--

754+-- See also 'callOnVM'.

755+data ProcessPair a = ProcessPair {

756+ ppairRemote :: RemoteProcess ()

757+ , ppairLocal :: LocalProcess a

758+ }

759+

760+-- | The process to run on the remote node (see 'ProcessPair' and 'callOnVM').

761+type RemoteProcess a = Closure (Backend -> Process a)

762+

763+-- | The process to run on the local node (see 'ProcessPair' and 'callOnVM').

764+newtype LocalProcess a = LocalProcess { unLocalProcess :: ReaderT SSH.Channel IO a }

765+ deriving (Functor, Monad, MonadIO, MonadReader SSH.Channel)

766+

767+runLocalProcess :: LocalProcess a -> SSH.Channel -> IO a

768+runLocalProcess = runReaderT . unLocalProcess

769+

770+-- | Send a messages from the local process to the remote process

771+-- (see 'ProcessPair')

772+localSend :: Serializable a => a -> LocalProcess ()

773+localSend x = LocalProcess $ do

774+ ch <- ask

775+ liftIO $ mapM_ (SSH.writeChannel ch)

776+ . prependLength

777+ . CH.messageToPayload

778+ . CH.createMessage

779+ $ x

780+

781+-- | Wait for a message from the remote process (see 'ProcessPair').

782+-- Note that unlike for the standard Cloud Haskell 'expect' it will result in a

783+-- runtime error if the remote process sends a message of type other than @a@.

784+--

785+-- Since it is relatively easy for the remote process to mess up the

786+-- communication protocol (for instance, by doing a putStr) we ask for the

787+-- length twice, as some sort of sanity check.

788+localExpect :: Serializable a => LocalProcess a

789+localExpect = LocalProcess $ do

790+ ch <- ask

791+ liftIO $ do

792+ isE <- readIntChannel ch

793+ len <- readIntChannel ch

794+ lenAgain <- readIntChannel ch

795+ when (len /= lenAgain) $ throwIO (userError "Internal error: protocol violation (perhaps the remote binary is not installed correctly?)")

796+ msg <- readSizeChannel ch len

797+ if isE /= 0

798+ then error (decode msg)

799+ else return (decode msg)

800+

801+-- | Send a message from the remote process to the local process (see

802+-- 'ProcessPair'). Note that the remote process can use the standard Cloud

803+-- Haskell primitives to /receive/ messages from the local process.

804+remoteSend :: Serializable a => a -> Process ()

805+remoteSend = liftIO . remoteSend'

806+

807+remoteSend' :: Serializable a => a -> IO ()

808+remoteSend' = remoteSendFlagged 0

809+

810+-- | If the remote process encounters an error it can use 'remoteThrow'. This

811+-- will cause the exception to be raised (as a user-exception, not as the

812+-- original type) in the local process (as well as in the remote process).

813+remoteThrow :: Exception e => e -> IO ()

814+remoteThrow e = remoteSendFlagged 1 (show e) >> throwIO e

815+

816+remoteSendFlagged :: Serializable a => Int -> a -> IO ()

817+remoteSendFlagged flags x = do

818+ let enc = encode x

819+ BSS.hPut stdout (encodeInt32 flags)

820+ -- See 'localExpect' for why we send the length twice

821+ BSS.hPut stdout (encodeInt32 (BSL.length enc))

822+ BSS.hPut stdout (encodeInt32 (BSL.length enc))

823+ BSL.hPut stdout enc

824+ hFlush stdout

825+

826+--------------------------------------------------------------------------------

827+-- On-VM main --

828+--------------------------------------------------------------------------------

829+

830+-- | Program main when run on the VM. A typical 'main' function looks like

831+--

832+-- > main :: IO ()

833+-- > main = do

834+-- > args <- getArgs

835+-- > case args of

836+-- > "onvm":args' -> onVmMain __remoteTable args'

837+-- > _ -> -- your normal main

838+onVmMain :: (RemoteTable -> RemoteTable) -> [String] -> IO ()

839+onVmMain rtable [host, port, cloudService, bg] = do

840+ hSetBinaryMode stdin True

841+ hSetBinaryMode stdout True

842+ Just rprocEnc <- getWithLength stdin

843+ Just paramsEnc <- getWithLength stdin

844+ backend <- initializeBackend (decode paramsEnc) cloudService

845+ let rproc = decode rprocEnc

846+ lprocMVar <- newEmptyMVar :: IO (MVar CH.LocalProcess)

847+ if read bg

848+ then

849+ void . Posix.forkProcess $ do

850+ -- We inherit the file descriptors from the parent, so the SSH

851+ -- session will not be terminated until we close them

852+ void Posix.createSession

853+ startCH rproc lprocMVar backend

854+ (\node proc -> runProcess node $ do

855+ us <- getSelfPid

856+ liftIO $ do

857+ remoteSend' us

858+ mapM_ hClose [stdin, stdout, stderr]

859+ proc)

860+ else do

861+ startCH rproc lprocMVar backend forkProcess

862+ lproc <- readMVar lprocMVar

863+ queueFromHandle stdin (CH.processQueue lproc)

864+ where

865+ startCH :: RemoteProcess ()

866+ -> MVar CH.LocalProcess

867+ -> Backend

868+ -> (CH.LocalNode -> Process () -> IO a)

869+ -> IO ()

870+ startCH rproc lprocMVar backend go = do

871+ mTransport <- createTransport host port defaultTCPParameters

872+ case mTransport of

873+ Left err -> remoteThrow err

874+ Right transport -> do

875+ node <- newLocalNode transport (rtable . __remoteTable $ initRemoteTable)

876+ void . go node $ do

877+ ask >>= liftIO . putMVar lprocMVar

878+ proc <- unClosure rproc :: Process (Backend -> Process ())

879+ catch (proc backend)

880+ (liftIO . (remoteThrow :: SomeException -> IO ()))

881+onVmMain _ _

882+ = error "Invalid arguments passed on onVmMain"

883+

884+-- | Read a 4-byte length @l@ and then an @l@-byte payload

885+--

886+-- Returns Nothing on EOF

887+getWithLength :: Handle -> IO (Maybe BSL.ByteString)

888+getWithLength h = do

889+ lenEnc <- BSS.hGet h 4

890+ if BSS.length lenEnc < 4

891+ then return Nothing

892+ else do

893+ let len = decodeInt32 lenEnc

894+ bs <- BSL.hGet h len

895+ if BSL.length bs < fromIntegral len

896+ then return Nothing

897+ else return (Just bs)

898+

899+queueFromHandle :: Handle -> CQueue CH.Message -> IO ()

900+queueFromHandle h q = do

901+ mPayload <- getWithLength stdin

902+ forM_ mPayload $ \payload -> do

903+ enqueue q $ CH.payloadToMessage (BSL.toChunks payload)

904+ queueFromHandle h q

905+

906+--------------------------------------------------------------------------------

907+-- SSH utilities --

908+--------------------------------------------------------------------------------

909+

910+readSizeChannel :: SSH.Channel -> Int -> IO BSL.ByteString

911+readSizeChannel ch = go []

912+ where

913+ go :: [BSS.ByteString] -> Int -> IO BSL.ByteString

914+ go acc 0 = return (BSL.fromChunks $ reverse acc)

915+ go acc size = do

916+ bs <- SSH.readChannel ch (fromIntegral (0x400 `min` size))

917+ go (bs : acc) (size - BSS.length bs)

918+

919+readIntChannel :: SSH.Channel -> IO Int

920+readIntChannel ch =

921+ decodeInt32 . BSS.concat . BSL.toChunks <$> readSizeChannel ch 4

922+

923+--------------------------------------------------------------------------------

924+-- High-level API --

925+--------------------------------------------------------------------------------

926+

927+data ServiceProcessMsg =

928+ ServiceProcessTerminate

929+ deriving Typeable

930+

931+instance Binary ServiceProcessMsg where

932+ put ServiceProcessTerminate = putWord8 0

933+ get = do

934+ header <- getWord8

935+ case header of

936+ 0 -> return ServiceProcessTerminate

937+ _ -> fail "ServiceProcessMsg.get"

938+

939+serviceProcess :: Backend -> Process ()

940+serviceProcess _backend = do

941+ us <- getSelfPid

942+ register "$azureBackendServiceProcess" us

943+ go

944+ where

945+ go = do

946+ msg <- expect

947+ case msg of

948+ ServiceProcessTerminate ->

949+ return ()

950+

951+serviceProcessStatic :: Static (Backend -> Process ())

952+serviceProcessStatic = staticLabel "serviceProcess"

953+

954+-- | Start a new Cloud Haskell node on the given virtual machine

955+spawnNodeOnVM :: Backend -> VirtualMachine -> String -> IO NodeId

956+spawnNodeOnVM backend vm port =

957+ processNodeId <$> spawnOnVM backend vm port (staticClosure serviceProcessStatic)

958+

959+-- | Terminate a node started with 'spawnNodeOnVM'

960+terminateNode :: NodeId -> Process ()

961+terminateNode nid = nsendRemote nid "$azureBackendServiceProcess" ServiceProcessTerminate

962+

963+__remoteTable :: RemoteTable -> RemoteTable

964+__remoteTable = registerStatic "serviceProcess" (toDynamic serviceProcess)

distributed-process-demos/LICENSEnew file+30