b0b76de · distributed-process

git.iko.soy/distributed-process

on master help sign in

Run an arbitrary process closure on a VM

b0b76deEdsko de Vries14 years ago

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

compare with master

unifiedsplit

@@ -26,11 +26,12 @@ Library

26 libssh2 >= 0.2 && < 0.3,

27 pureMD5 >= 2.1 && < 2.2,

28 bytestring >= 0.9 && < 0.11,

29− distributed-process >= 0.2 && < 0.3,

29+ distributed-process >= 0.2.3 && < 0.3,

30 binary >= 0.5 && < 0.6,

31 network-transport-tcp >= 0.2 && < 0.3,

32 optparse-applicative >= 0.2 && < 0.4,

33− transformers >= 0.3 && < 0.4

33+ transformers >= 0.3 && < 0.4,

34+ certificate == 1.2.3

34 Exposed-modules: Control.Distributed.Process.Backend.Azure,

35 Exposed-modules: Control.Distributed.Process.Backend.Azure,

35 Control.Distributed.Process.Backend.Azure.GenericMain

36 Control.Distributed.Process.Backend.Azure.GenericMain

36 Extensions: ViewPatterns,

37 Extensions: ViewPatterns,

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

@@ -5,6 +5,7 @@ module Control.Distributed.Process.Backend.Azure

5 , AzureParameters(..)

6 , defaultAzureParameters

7 , initializeBackend

8+ , remoteTable

8 -- * Re-exports from Azure Service Management

9 -- * Re-exports from Azure Service Management

9 , CloudService(..)

10 , CloudService(..)

10 , VirtualMachine(..)

11 , VirtualMachine(..)

@@ -16,7 +17,7 @@ import System.Environment.Executable (getExecutablePath)

16import System.Posix.Types (Fd)

17import System.Posix.Types (Fd)

17import Data.Binary (encode, decode)

18import Data.Binary (encode, decode)

18import Data.Digest.Pure.MD5 (md5, MD5Digest)

19import Data.Digest.Pure.MD5 (md5, MD5Digest)

19−import qualified Data.ByteString.Lazy as BSL (readFile, putStr)

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

20import Data.Typeable (Typeable)

21import Data.Typeable (Typeable)

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

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

22import Control.Monad (void)

23import Control.Monad (void)

@@ -61,15 +62,18 @@ import qualified Network.SSH.Client.LibSSH2.ByteString.Lazy as SSHBS

62-- CH

63-- CH

63import Control.Distributed.Process

64import Control.Distributed.Process

64− ( Closure

65+ ( Closure(Closure)

65 , Process

66 , Process

66 , Static

67 , Static

68+ , RemoteTable

67 )

69 )

68import Control.Distributed.Process.Closure

70import Control.Distributed.Process.Closure

69 ( remotable

71 ( remotable

70− , mkClosure

71 , cpBind

72 , cpBind

72 , SerializableDict(SerializableDict)

73 , SerializableDict(SerializableDict)

74+ , staticConst

75+ , staticApply

76+ , mkStatic

73 )

77 )

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

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

@@ -79,7 +83,17 @@ encodeToStdout = liftIO . BSL.putStr . encode

79encodeToStdoutDict :: SerializableDict a -> a -> Process ()

83encodeToStdoutDict :: SerializableDict a -> a -> Process ()

80encodeToStdoutDict SerializableDict = encodeToStdout

84encodeToStdoutDict SerializableDict = encodeToStdout

82−remotable ['encodeToStdout]

86+remotable ['encodeToStdoutDict]

87+

88+-- | Remote table necessary for the Azure backend

89+remoteTable :: RemoteTable -> RemoteTable

90+remoteTable = __remoteTable

91+

92+cpEncodeToStdout :: forall a. Typeable a => Static (SerializableDict a) -> Closure (a -> Process ())

93+cpEncodeToStdout dict = Closure decoder (encode ())

94+ where

95+ decoder :: Static (BSL.ByteString -> a -> Process ())

96+ decoder = staticConst `staticApply` ($(mkStatic 'encodeToStdoutDict) `staticApply` dict)

84-- | Azure backend

98-- | Azure backend

85data Backend = Backend {

99data Backend = Backend {

@@ -89,7 +103,7 @@ data Backend = Backend {

89 , copyToVM :: VirtualMachine -> IO ()

103 , copyToVM :: VirtualMachine -> IO ()

90 -- | Check the MD5 hash of the remote executable

104 -- | Check the MD5 hash of the remote executable

91 , checkMD5 :: VirtualMachine -> IO Bool

105 , checkMD5 :: VirtualMachine -> IO Bool

92− -- | @runOnVM vm port p bg@ starts a CH node on port 'port' and runs 'p'

106+ -- | @runOnVM dict vm port p@ starts a CH node on port 'port' and runs 'p'

93 , callOnVM :: forall a. Serializable a => Static (SerializableDict a) -> VirtualMachine -> String -> Closure (Process a) -> IO a

107 , callOnVM :: forall a. Serializable a => Static (SerializableDict a) -> VirtualMachine -> String -> Closure (Process a) -> IO a

94 }

108 }

109

@@ -158,13 +172,13 @@ apiCallOnVM params dict vm port proc =

158 ++ " 2>&1"

172 ++ " 2>&1"

159 (_, r) <- SSH.withChannel (SSH.openChannelSession s) id fd s $ \ch -> do

173 (_, r) <- SSH.withChannel (SSH.openChannelSession s) id fd s $ \ch -> do

160 SSH.retryIfNeeded fd s $ SSH.channelExecute ch exe

174 SSH.retryIfNeeded fd s $ SSH.channelExecute ch exe

161− SSHBS.writeChannel fd ch (encode proc)

175+ cnt <- SSHBS.writeChannel fd ch (encode proc')

162 SSH.channelSendEOF ch

176 SSH.channelSendEOF ch

163 SSHBS.readAllChannel fd ch

177 SSHBS.readAllChannel fd ch

164 return (decode r)

178 return (decode r)

165 where

179 where

166 proc' :: Closure (Process ())

180 proc' :: Closure (Process ())

167− proc' = proc `cpBind` undefined

181+ proc' = proc `cpBind` cpEncodeToStdout dict

168

182

169-- | Check the MD5 hash of the executable on the remote machine

183-- | Check the MD5 hash of the executable on the remote machine

170apiCheckMD5 :: AzureParameters -> VirtualMachine -> IO Bool

184apiCheckMD5 :: AzureParameters -> VirtualMachine -> IO Bool

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

@@ -9,7 +9,9 @@ import Control.Distributed.Process.Backend.Azure.GenericMain

9 )

11getPid :: () -> Process ProcessId

12+getPid () = do

13+ liftIO $ appendFile "Log" "getPid did run"

12−getPid () = getSelfPid

14+ getSelfPid

14remotable ['getPid]

16remotable ['getPid]

distributed-process-azure/src/Control/Distributed/Process/Backend/Azure/GenericMain.hs+28−14

@@ -13,23 +13,19 @@ import System.IO

13 , hSetBinaryMode

14 )

15import Data.Binary (decode)

16−import qualified Data.ByteString.Lazy as BSL (ByteString, getContents)

16+import qualified Data.ByteString.Lazy as BSL (ByteString, getContents, length)

17import Control.Monad (unless, forM, forM_, join)

18−import Control.Exception (throwIO)

18+import Control.Exception (throwIO, SomeException)

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

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

20− ( AzureParameters(azureSshUserName)

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

21− , defaultAzureParameters

21+

22− , initializeBackend

22+-- SSH

23− , cloudServices

24− , CloudService(cloudServiceName, cloudServiceVMs)

25− , VirtualMachine(vmName)

26− , Backend(copyToVM, checkMD5, callOnVM)

27− )

28import qualified Network.SSH.Client.LibSSH2.Foreign as SSH

23import qualified Network.SSH.Client.LibSSH2.Foreign as SSH

29 ( initialize

24 ( initialize

30 , exit

25 , exit

31 )

26 )

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

27+

28+-- Command line options

33import Options.Applicative

29import Options.Applicative

34 ( Parser

30 ( Parser

35 , strOption

31 , strOption

@@ -47,17 +43,30 @@ import Options.Applicative

47 , header

43 , header

48 , switch

44 , switch

49 )

45 )

46+

47+-- CH

50import Control.Distributed.Process

48import Control.Distributed.Process

51 ( RemoteTable

49 ( RemoteTable

52 , Closure

50 , Closure

53 , Process

51 , Process

54 , unClosure

52 , unClosure

55 , Static

53 , Static

54+ , catch

56 )

55 )

57import Control.Distributed.Process.Node (newLocalNode, runProcess, initRemoteTable)

56import Control.Distributed.Process.Node (newLocalNode, runProcess, initRemoteTable)

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

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

59import Control.Distributed.Process.Closure (SerializableDict)

58import Control.Distributed.Process.Closure (SerializableDict)

60import Network.Transport.TCP (createTransport, defaultTCPParameters)

59import Network.Transport.TCP (createTransport, defaultTCPParameters)

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

61+ ( AzureParameters(azureSshUserName)

62+ , defaultAzureParameters

63+ , initializeBackend

64+ , cloudServices

65+ , CloudService(cloudServiceName, cloudServiceVMs)

66+ , VirtualMachine(vmName)

67+ , Backend(copyToVM, checkMD5, callOnVM)

68+ )

69+import qualified Control.Distributed.Process.Backend.Azure as Azure (remoteTable)

62--------------------------------------------------------------------------------

71--------------------------------------------------------------------------------

63-- Main --

72-- Main --

@@ -112,7 +121,9 @@ genericMain remoteTable cmds = do

112 ProcessPair rProc lProc dict ->

121 ProcessPair rProc lProc dict ->

113 callOnVM backend dict vm (remotePort cmd) rProc >>= lProc

122 callOnVM backend dict vm (remotePort cmd) rProc >>= lProc

114 OnVmCommand (vmCmd@OnVmRun {}) -> do

123 OnVmCommand (vmCmd@OnVmRun {}) -> do

115− onVmRun (remoteTable initRemoteTable) (onVmIP vmCmd) (onVmPort vmCmd)

124+ onVmRun (remoteTable . Azure.remoteTable $ initRemoteTable)

125+ (onVmIP vmCmd)

126+ (onVmPort vmCmd)

116 SSH.exit

127 SSH.exit

117 where

128 where

118 opts = info (helper <*> commandParser)

129 opts = info (helper <*> commandParser)

@@ -144,13 +155,16 @@ azureParameters opts (Just sshOpts) = do

144onVmRun :: RemoteTable -> String -> String -> IO ()

155onVmRun :: RemoteTable -> String -> String -> IO ()

145onVmRun rtable host port = do

156onVmRun rtable host port = do

146 hSetBinaryMode stdin True

157 hSetBinaryMode stdin True

158+ hSetBinaryMode stdout True

147 proc <- BSL.getContents :: IO BSL.ByteString

159 proc <- BSL.getContents :: IO BSL.ByteString

148 mTransport <- createTransport host port defaultTCPParameters

160 mTransport <- createTransport host port defaultTCPParameters

149 case mTransport of

161 case mTransport of

150 Left err -> throwIO err

162 Left err -> throwIO err

151 Right transport -> do

163 Right transport -> do

152 node <- newLocalNode transport rtable

164 node <- newLocalNode transport rtable

165+ runProcess node $

153− runProcess node $ join . unClosure . decode $ proc

166+ catch (join . unClosure . decode $ proc)

167+ (\e -> liftIO (print (e :: SomeException) >> throwIO e))

154

168

155--------------------------------------------------------------------------------

169--------------------------------------------------------------------------------

156-- Command line options --

170-- Command line options --