d54d67e · distributed-process

git.iko.soy/distributed-process

on master help sign in

Initialize the backend on the server

d54d67eEdsko de Vries14 years ago

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

compare with master

unifiedsplit

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

2import Data.Data (Typeable, Data)

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

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

5+import Control.Monad (forever)

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

5import Control.Distributed.Process

7import Control.Distributed.Process

6 ( Process

8 ( Process

7 , expect

9 , expect

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

11 , mkClosure

13 , mkClosure

12 )

14 )

13import Control.Distributed.Process.Backend.Azure

15import Control.Distributed.Process.Backend.Azure

14− ( Backend

16+ ( Backend(findVMs)

15− , ProcessPair(..)

17+ , ProcessPair(ProcessPair)

16 , RemoteProcess

18 , RemoteProcess

19+ , LocalProcess

20+ , remoteSend

21+ , localExpect

17 )

22 )

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

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

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

25 get = getGeneric

30 get = getGeneric

26 put = putGeneric

31 put = putGeneric

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

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

29−conwayController () _backend = go

34+conwayStart () backend = do

30− where

35+ vms <- liftIO $ findVMs backend

31− go = do

36+ remoteSend (show vms)

32− msg <- expect

37+

33− case msg of

38+remotable ['conwayStart]

34− ControllerExit ->

35− return ()

37−remotable ['conwayController]

40+echo :: LocalProcess ()

41+echo = forever $ do

42+ msg <- localExpect

43+ liftIO $ putStrLn msg

39main :: IO ()

45main :: IO ()

40main = genericMain __remoteTable callable spawnable

46main = genericMain __remoteTable callable spawnable

41 where

47 where

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

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

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

43− callable _ = error "spawnable: unknown"

50+ callable _ = error "callable: unknown"

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

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

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

25import System.Environment.Executable (getExecutablePath)

26import System.IO (stdout, hFlush)

27−import Data.Binary (encode, decode)

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

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

29import qualified Data.ByteString as BSS

30 ( ByteString

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

43 )

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

45import Data.Typeable (Typeable)

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

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

47import Control.Monad (void, unless)

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

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

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

125 , azureSshLocalPath :: FilePath

126 }

127

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+

128-- | Create default azure parameters

141-- | Create default azure parameters

129defaultAzureParameters :: String -- ^ Azure subscription ID

142defaultAzureParameters :: String -- ^ Azure subscription ID

130 -> FilePath -- ^ Path to X509 certificate

143 -> FilePath -- ^ Path to X509 certificate

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

155 findVMs = apiFindVMs params cloudService

168 findVMs = apiFindVMs params cloudService

156 , copyToVM = apiCopyToVM params

169 , copyToVM = apiCopyToVM params

157 , checkMD5 = apiCheckMD5 params

170 , checkMD5 = apiCheckMD5 params

158− , callOnVM = apiCallOnVM params

171+ , callOnVM = apiCallOnVM params cloudService

159− , spawnOnVM = apiSpawnOnVM params

172+ , spawnOnVM = apiSpawnOnVM params cloudService

160 }

173 }

161

174

162-- | Find virtual machines

175-- | Find virtual machines

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

175

188

176-- | Call a process on a VM

189-- | Call a process on a VM

177apiCallOnVM :: AzureParameters

190apiCallOnVM :: AzureParameters

191+ -> String

178 -> VirtualMachine

192 -> VirtualMachine

179 -> String

193 -> String

180 -> ProcessPair a

194 -> ProcessPair a

181 -> IO a

195 -> IO a

182−apiCallOnVM params vm port ppair =

196+apiCallOnVM params cloudService vm port ppair =

183 withSSH2 params vm $ \s -> do

197 withSSH2 params vm $ \s -> do

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

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

185 ++ " onvm run "

199 ++ " onvm run "

186 ++ " --host " ++ vmIpAddress vm

200 ++ " --host " ++ vmIpAddress vm

187 ++ " --port " ++ port

201 ++ " --port " ++ port

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

188 ++ " 2>&1"

203 ++ " 2>&1"

204+ let paramsEnc = encode params

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

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

190 SSH.channelExecute ch exe

206 SSH.channelExecute ch exe

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

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

208+ SSH.writeAllChannel ch rprocEnc

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

192− _ <- SSH.writeAllChannel ch rprocEnc

210+ SSH.writeAllChannel ch paramsEnc

193 runLocalProcess (ppairLocal ppair) ch

211 runLocalProcess (ppairLocal ppair) ch

194 if status == 0

212 if status == 0

195 then return r

213 then return r

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

199 rprocEnc = encode (ppairRemote ppair)

217 rprocEnc = encode (ppairRemote ppair)

200

218

201apiSpawnOnVM :: AzureParameters

219apiSpawnOnVM :: AzureParameters

220+ -> String

202 -> VirtualMachine

221 -> VirtualMachine

203 -> String

222 -> String

204 -> Closure (Backend -> Process ())

223 -> Closure (Backend -> Process ())

205 -> IO ()

224 -> IO ()

206−apiSpawnOnVM params vm port proc =

225+apiSpawnOnVM params cloudService vm port proc =

207 withSSH2 params vm $ \s -> do

226 withSSH2 params vm $ \s -> do

227+ -- TODO: reduce duplication with apiCallOnVM

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

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

209 ++ " onvm run "

229 ++ " onvm run "

210 ++ " --host " ++ vmIpAddress vm

230 ++ " --host " ++ vmIpAddress vm

211 ++ " --port " ++ port

231 ++ " --port " ++ port

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

212 ++ " --background "

233 ++ " --background "

213 ++ " 2>&1"

234 ++ " 2>&1"

235+ let paramsEnc = encode params

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

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

215 SSH.channelExecute ch exe

237 SSH.channelExecute ch exe

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

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

239+ SSH.writeAllChannel ch procEnc

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

217− _ <- SSH.writeAllChannel ch procEnc

241+ SSH.writeAllChannel ch paramsEnc

218 SSH.channelSendEOF ch

242 SSH.channelSendEOF ch

219 SSH.readAllChannel ch

243 SSH.readAllChannel ch

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

244 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

149 onVmRun (remoteTable initRemoteTable)

150 (onVmIP vmCmd)

151 (onVmPort vmCmd)

152+ (onVmService vmCmd)

152 (onVmBackground vmCmd)

153 (onVmBackground vmCmd)

153 SSH.exit

154 SSH.exit

154 where

155 where

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

178-- Executing a closure on the VM --

179-- Executing a closure on the VM --

179--------------------------------------------------------------------------------

180--------------------------------------------------------------------------------

180

181

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

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

182−onVmRun rtable host port bg = do

183+onVmRun rtable host port cloudService bg = do

183 hSetBinaryMode stdin True

184 hSetBinaryMode stdin True

184 hSetBinaryMode stdout True

185 hSetBinaryMode stdout True

185− mProcEnc <- getWithLength stdin

186+ Just procEnc <- getWithLength stdin

186− forM_ mProcEnc $ \procEnc -> do

187+ Just paramsEnc <- getWithLength stdin

188+ backend <- initializeBackend (decode paramsEnc) cloudService

187− let proc = decode procEnc

189+ let proc = decode procEnc

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

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

189− if bg

191+ if bg

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

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

191− else do

193+ else do

192− startCH proc lprocMVar forkProcess (liftIO . remoteThrow)

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

193− lproc <- readMVar lprocMVar

195+ lproc <- readMVar lprocMVar

194− queueFromHandle stdin (processQueue lproc)

196+ queueFromHandle stdin (processQueue lproc)

195 where

197 where

196 startCH :: RemoteProcess ()

198 startCH :: RemoteProcess ()

197 -> MVar LocalProcess

199 -> MVar LocalProcess

200+ -> Backend

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

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

199 -> (SomeException -> Process ())

202 -> (SomeException -> Process ())

200 -> IO ()

203 -> IO ()

201− startCH rproc lprocMVar go exceptionHandler = do

204+ startCH rproc lprocMVar backend go exceptionHandler = do

202 mTransport <- createTransport host port defaultTCPParameters

205 mTransport <- createTransport host port defaultTCPParameters

203 case mTransport of

206 case mTransport of

204 Left err -> remoteThrow err

207 Left err -> remoteThrow err

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

206 node <- newLocalNode transport rtable

209 node <- newLocalNode transport rtable

207 void . go node $ do

210 void . go node $ do

208 ask >>= liftIO . putMVar lprocMVar

211 ask >>= liftIO . putMVar lprocMVar

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

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

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

211 catch (proc backend) exceptionHandler

213 catch (proc backend) exceptionHandler

212

214

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

291 OnVmRun {

293 OnVmRun {

292 onVmIP :: String

294 onVmIP :: String

293 , onVmPort :: String

295 , onVmPort :: String

296+ , onVmService :: String

294 , onVmBackground :: Bool

297 , onVmBackground :: Bool

295 }

298 }

296 deriving Show

299 deriving Show

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

387 & metavar "PORT"

390 & metavar "PORT"

388 & help "port number"

391 & help "port number"

389 )

392 )

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

394+ & metavar "CS"

395+ & help "Cloud service name"

396+ )

390 <*> switch ( long "background"

397 <*> switch ( long "background"

391 & help "Run the process in the background"

398 & help "Run the process in the background"

392 )

399 )