d54d67e · distributed-process

git.iko.soy/distributed-process

Initialize the backend on the server

This completes the low-level Azure API.

commitd54d67e27072877fdec3b1f8b3c82d94d86c7b70 authorEdsko de Vries <edsko@well-typed.com> dateTue 21 Aug 2012, 09:12 (14 years ago) parentfa28db9

3 files changed+75−38

compare with master

unifiedsplit

distributed-process-azure/demos/Conway.hs+20−14 distributed-process-azure/src/Control/Distributed/Process/Backend/Azure.hs+34−10 distributed-process-azure/src/Control/Distributed/Process/Backend/Azure/GenericMain.hs+21−14

distributed-process-azure/demos/Conway.hs+20−14

@@ -2,6 +2,8 @@

22import Data.Data (Typeable, Data)

33import Data.Binary (Binary(get, put))

44import Data.Binary.Generic (getGeneric, putGeneric)

5+import Control.Monad (forever)

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

57import Control.Distributed.Process

68 ( Process

79 , expect

@@ -11,9 +13,12 @@ import Control.Distributed.Process.Closure

1113 , mkClosure

1214 )

1315import Control.Distributed.Process.Backend.Azure

14− ( Backend

15− , ProcessPair(..)

16+ ( Backend(findVMs)

17+ , ProcessPair(ProcessPair)

1618 , RemoteProcess

19+ , LocalProcess

20+ , remoteSend

21+ , localExpect

1722 )

1823import Control.Distributed.Process.Backend.Azure.GenericMain (genericMain)

1924

@@ -25,23 +30,24 @@ instance Binary ControllerMsg where

2530 get = getGeneric

2631 put = putGeneric

2732

28−conwayController :: () -> Backend -> Process ()

29−conwayController () _backend = go

30− where

31− go = do

32− msg <- expect

33− case msg of

34− ControllerExit ->

35− return ()

33+conwayStart :: () -> Backend -> Process ()

34+conwayStart () backend = do

35+ vms <- liftIO $ findVMs backend

36+ remoteSend (show vms)

37+

38+remotable ['conwayStart]

3639

37−remotable ['conwayController]

40+echo :: LocalProcess ()

41+echo = forever $ do

42+ msg <- localExpect

43+ liftIO $ putStrLn msg

3844

3945main :: IO ()

4046main = genericMain __remoteTable callable spawnable

4147 where

4248 callable :: String -> IO (ProcessPair ())

43− callable _ = error "spawnable: unknown"

49+ callable "start" = return $ ProcessPair ($(mkClosure 'conwayStart) ()) echo

50+ callable _ = error "callable: unknown"

4451

4552 spawnable :: String -> IO (RemoteProcess ())

46− spawnable "controller" = return $ $(mkClosure 'conwayController) ()

47− spawnable _ = error "callable: unknown"

53+ spawnable _ = error "spawnable: unknown"

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

@@ -24,7 +24,7 @@ import System.Environment (getEnv)

2424import System.FilePath ((</>), takeFileName)

2525import System.Environment.Executable (getExecutablePath)

2626import System.IO (stdout, hFlush)

27−import Data.Binary (encode, decode)

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

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

2929import qualified Data.ByteString as BSS

3030 ( ByteString

@@ -43,7 +43,7 @@ import qualified Data.ByteString.Lazy as BSL

4343 )

4444import qualified Data.ByteString.Lazy.Char8 as BSLC (unpack)

4545import Data.Typeable (Typeable)

46−import Control.Applicative ((<$>))

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

4747import Control.Monad (void, unless)

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

4949import Control.Exception (Exception, catches, Handler(Handler), throwIO)

@@ -125,6 +125,19 @@ data AzureParameters = AzureParameters {

125125 , azureSshLocalPath :: FilePath

126126 }

127127

128+instance Binary AzureParameters where

129+ put params = do

130+ put (azureSetup params)

131+ put (azureSshUserName params)

132+ put (azureSshPublicKey params)

133+ put (azureSshPrivateKey params)

134+ put (azureSshPassphrase params)

135+ put (azureSshKnownHosts params)

136+ put (azureSshRemotePath params)

137+ put (azureSshLocalPath params)

138+ get =

139+ AzureParameters <$> get <*> get <*> get <*> get <*> get <*> get <*> get <*> get

140+

128141-- | Create default azure parameters

129142defaultAzureParameters :: String -- ^ Azure subscription ID

130143 -> FilePath -- ^ Path to X509 certificate

@@ -155,8 +168,8 @@ initializeBackend params cloudService =

155168 findVMs = apiFindVMs params cloudService

156169 , copyToVM = apiCopyToVM params

157170 , checkMD5 = apiCheckMD5 params

158− , callOnVM = apiCallOnVM params

159− , spawnOnVM = apiSpawnOnVM params

171+ , callOnVM = apiCallOnVM params cloudService

172+ , spawnOnVM = apiSpawnOnVM params cloudService

160173 }

161174

162175-- | Find virtual machines

@@ -175,21 +188,26 @@ apiCopyToVM params vm =

175188

176189-- | Call a process on a VM

177190apiCallOnVM :: AzureParameters

191+ -> String

178192 -> VirtualMachine

179193 -> String

180194 -> ProcessPair a

181195 -> IO a

182−apiCallOnVM params vm port ppair =

196+apiCallOnVM params cloudService vm port ppair =

183197 withSSH2 params vm $ \s -> do

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

185199 ++ " onvm run "

186200 ++ " --host " ++ vmIpAddress vm

187201 ++ " --port " ++ port

202+ ++ " --cloud-service " ++ cloudService

188203 ++ " 2>&1"

204+ let paramsEnc = encode params

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

190206 SSH.channelExecute ch exe

191− _ <- SSH.writeChannel ch (encodeInt32 (BSL.length rprocEnc))

192− _ <- SSH.writeAllChannel ch rprocEnc

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

208+ SSH.writeAllChannel ch rprocEnc

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

210+ SSH.writeAllChannel ch paramsEnc

193211 runLocalProcess (ppairLocal ppair) ch

194212 if status == 0

195213 then return r

@@ -199,22 +217,28 @@ apiCallOnVM params vm port ppair =

199217 rprocEnc = encode (ppairRemote ppair)

200218

201219apiSpawnOnVM :: AzureParameters

220+ -> String

202221 -> VirtualMachine

203222 -> String

204223 -> Closure (Backend -> Process ())

205224 -> IO ()

206−apiSpawnOnVM params vm port proc =

225+apiSpawnOnVM params cloudService vm port proc =

207226 withSSH2 params vm $ \s -> do

227+ -- TODO: reduce duplication with apiCallOnVM

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

209229 ++ " onvm run "

210230 ++ " --host " ++ vmIpAddress vm

211231 ++ " --port " ++ port

232+ ++ " --cloud-service " ++ cloudService

212233 ++ " --background "

213234 ++ " 2>&1"

235+ let paramsEnc = encode params

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

215237 SSH.channelExecute ch exe

216− _ <- SSH.writeChannel ch (encodeInt32 (BSL.length procEnc))

217− _ <- SSH.writeAllChannel ch procEnc

238+ SSH.writeChannel ch (encodeInt32 (BSL.length procEnc))

239+ SSH.writeAllChannel ch procEnc

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

241+ SSH.writeAllChannel ch paramsEnc

218242 SSH.channelSendEOF ch

219243 SSH.readAllChannel ch

220244 unless (status == 0) $ error (BSLC.unpack r)

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

@@ -149,6 +149,7 @@ genericMain remoteTable callable spawnable = do

149149 onVmRun (remoteTable initRemoteTable)

150150 (onVmIP vmCmd)

151151 (onVmPort vmCmd)

152+ (onVmService vmCmd)

152153 (onVmBackground vmCmd)

153154 SSH.exit

154155 where

@@ -178,27 +179,29 @@ azureParameters opts (Just sshOpts) = do

178179-- Executing a closure on the VM --

179180--------------------------------------------------------------------------------

180181

181−onVmRun :: RemoteTable -> String -> String -> Bool -> IO ()

182−onVmRun rtable host port bg = do

182+onVmRun :: RemoteTable -> String -> String -> String -> Bool -> IO ()

183+onVmRun rtable host port cloudService bg = do

183184 hSetBinaryMode stdin True

184185 hSetBinaryMode stdout True

185− mProcEnc <- getWithLength stdin

186− forM_ mProcEnc $ \procEnc -> do

187− let proc = decode procEnc

188− lprocMVar <- newEmptyMVar :: IO (MVar LocalProcess)

189− if bg

190− then detach $ startCH proc lprocMVar runProcess (\_ -> return ())

191− else do

192− startCH proc lprocMVar forkProcess (liftIO . remoteThrow)

193− lproc <- readMVar lprocMVar

194− queueFromHandle stdin (processQueue lproc)

186+ Just procEnc <- getWithLength stdin

187+ Just paramsEnc <- getWithLength stdin

188+ backend <- initializeBackend (decode paramsEnc) cloudService

189+ let proc = decode procEnc

190+ lprocMVar <- newEmptyMVar :: IO (MVar LocalProcess)

191+ if bg

192+ then detach $ startCH proc lprocMVar backend runProcess (\_ -> return ())

193+ else do

194+ startCH proc lprocMVar backend forkProcess (liftIO . remoteThrow)

195+ lproc <- readMVar lprocMVar

196+ queueFromHandle stdin (processQueue lproc)

195197 where

196198 startCH :: RemoteProcess ()

197199 -> MVar LocalProcess

200+ -> Backend

198201 -> (LocalNode -> Process () -> IO a)

199202 -> (SomeException -> Process ())

200203 -> IO ()

201− startCH rproc lprocMVar go exceptionHandler = do

204+ startCH rproc lprocMVar backend go exceptionHandler = do

202205 mTransport <- createTransport host port defaultTCPParameters

203206 case mTransport of

204207 Left err -> remoteThrow err

@@ -206,7 +209,6 @@ onVmRun rtable host port bg = do

206209 node <- newLocalNode transport rtable

207210 void . go node $ do

208211 ask >>= liftIO . putMVar lprocMVar

209− let backend = error "TODO: backend not initialized in onVmRun"

210212 proc <- unClosure rproc :: Process (Backend -> Process ())

211213 catch (proc backend) exceptionHandler

212214

@@ -291,6 +293,7 @@ data OnVmCommand =

291293 OnVmRun {

292294 onVmIP :: String

293295 , onVmPort :: String

296+ , onVmService :: String

294297 , onVmBackground :: Bool

295298 }

296299 deriving Show

@@ -387,6 +390,10 @@ onVmRunParser = OnVmRun

387390 & metavar "PORT"

388391 & help "port number"

389392 )

393+ <*> strOption ( long "cloud-service"

394+ & metavar "CS"

395+ & help "Cloud service name"

396+ )

390397 <*> switch ( long "background"

391398 & help "Run the process in the background"

392399 )