32d3bf4 · distributed-process

git.iko.soy/distributed-process

on master help sign in

Support spawn (as well as call)

32d3bf4Edsko de Vries14 years ago

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

compare with master

unifiedsplit

@@ -31,7 +31,8 @@ Library

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

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

33 transformers >= 0.3 && < 0.4,

34− certificate == 1.2.3

34+ certificate == 1.2.3,

35+ unix >= 2.5 && < 2.6

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

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

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

37 Control.Distributed.Process.Backend.Azure.GenericMain

37 Extensions: ViewPatterns,

38 Extensions: ViewPatterns,

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

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

17import Data.Binary (encode, decode)

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

19import qualified Data.ByteString.Char8 as BSSC (pack)

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

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

21import Data.Typeable (Typeable)

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

23import Control.Monad (void)

@@ -102,6 +102,9 @@ data Backend = Backend {

102 , checkMD5 :: VirtualMachine -> IO Bool

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

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

105+ -- | Create a new CH node and run the specified process in the background.

106+ -- The CH node will exit when the process exists.

107+ , spawnOnVM :: VirtualMachine -> String -> Closure (Process ()) -> IO ()

105 }

108 }

106

109

107data AzureParameters = AzureParameters {

110data AzureParameters = AzureParameters {

@@ -150,6 +153,7 @@ initializeBackend params = do

150 , copyToVM = apiCopyToVM params

153 , copyToVM = apiCopyToVM params

151 , checkMD5 = apiCheckMD5 params

154 , checkMD5 = apiCheckMD5 params

152 , callOnVM = apiCallOnVM params

155 , callOnVM = apiCallOnVM params

156+ , spawnOnVM = apiSpawnOnVM params

153 }

157 }

154

158

155-- | Start a CH node on the given virtual machine

159-- | Start a CH node on the given virtual machine

@@ -161,21 +165,35 @@ apiCopyToVM params vm =

161-- | Call a process on a VM

165-- | Call a process on a VM

162apiCallOnVM :: Serializable a => AzureParameters -> Static (SerializableDict a) -> VirtualMachine -> String -> Closure (Process a) -> IO a

166apiCallOnVM :: Serializable a => AzureParameters -> Static (SerializableDict a) -> VirtualMachine -> String -> Closure (Process a) -> IO a

163apiCallOnVM params dict vm port proc =

167apiCallOnVM params dict vm port proc =

164− withSSH2 params vm $ \s -> do

168+ withSSH2 params vm $ \s -> do

165− let exe = "PATH=. " ++ azureSshRemotePath params

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

166− ++ " onvm run "

170+ ++ " onvm run "

167− ++ " --host " ++ vmIpAddress vm

171+ ++ " --host " ++ vmIpAddress vm

168− ++ " --port " ++ port

172+ ++ " --port " ++ port

169− ++ " 2>&1"

173+ ++ " 2>&1"

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

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

171− SSH.channelExecute ch exe

175+ SSH.channelExecute ch exe

172− _cnt <- SSH.writeAllChannel ch (encode proc')

176+ _cnt <- SSH.writeAllChannel ch (encode $ proc `cpBind` cpEncodeToStdout dict)

173− SSH.channelSendEOF ch

177+ SSH.channelSendEOF ch

174− SSH.readAllChannel ch

178+ SSH.readAllChannel ch

175− return (decode r)

179+ return (decode r)

176− where

180+

177− proc' :: Closure (Process ())

181+apiSpawnOnVM :: AzureParameters -> VirtualMachine -> String -> Closure (Process ()) -> IO ()

178− proc' = proc `cpBind` cpEncodeToStdout dict

182+apiSpawnOnVM params vm port proc =

183+ withSSH2 params vm $ \s -> do

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

185+ ++ " onvm run "

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

187+ ++ " --port " ++ port

188+ ++ " --background "

189+ ++ " 2>&1"

190+ BSL.writeFile "closure" (encode proc)

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

192+ SSH.channelExecute ch exe

193+ _cnt <- SSH.writeAllChannel ch (encode proc)

194+ SSH.channelSendEOF ch

195+ SSH.readAllChannel ch

196+ print r

179

197

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

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

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

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

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

@@ -1,7 +1,8 @@

1{-# LANGUAGE TemplateHaskell #-}

3import Control.Monad.IO.Class (liftIO)

4+import Control.Concurrent (threadDelay)

4−import Control.Distributed.Process (Process, ProcessId, getSelfPid)

5+import Control.Distributed.Process (Process, ProcessId, getSelfPid, Closure)

5import Control.Distributed.Process.Closure (remotable, mkClosure, sdictProcessId)

6import Control.Distributed.Process.Closure (remotable, mkClosure, sdictProcessId)

6import Control.Distributed.Process.Backend.Azure.GenericMain

7import Control.Distributed.Process.Backend.Azure.GenericMain

7 ( genericMain

8 ( genericMain

@@ -13,9 +14,24 @@ getPid () = do

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

14 liftIO $ appendFile "Log" "getPid did run"

14 getSelfPid

15 getSelfPid

16−remotable ['getPid]

17+logN :: Int -> Process ()

18+logN 0 =

19+ liftIO $ appendFile "Log" "logN done\n"

20+logN n = do

21+ liftIO $ do

22+ appendFile "Log" $ "logN " ++ show n ++ "\n"

23+ threadDelay 1000000

24+ logN (n - 1)

26+remotable ['getPid, 'logN]

27+

28+main :: IO ()

18−main = genericMain __remoteTable $ \cmd ->

29+main = genericMain __remoteTable callable spawnable

19− case cmd of

30+ where

31+ callable :: String -> IO (ProcessPair ())

20− "hello" -> return $ ProcessPair ($(mkClosure 'getPid) ()) print sdictProcessId

32+ callable "getPid" = return $ ProcessPair ($(mkClosure 'getPid) ()) print sdictProcessId

21− _ -> error "unknown command"

33+ callable _ = error "spawnable: unknown"

34+

35+ spawnable :: String -> IO (Closure (Process ()))

36+ spawnable "logN" = return $ $(mkClosure 'logN) (10 :: Int)

37+ spawnable _ = error "callable: unknown"

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

@@ -10,15 +10,20 @@ import System.IO

10 ( hFlush

11 , stdout

12 , stdin

13+ , stderr

13 , hSetBinaryMode

14 , hSetBinaryMode

15+ , hClose

14 )

16 )

15import Data.Binary (decode)

17import Data.Binary (decode)

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

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

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

19+import Control.Monad (unless, forM, forM_, join, void)

18−import Control.Exception (throwIO, SomeException)

20+import Control.Exception (throwIO, SomeException, evaluate)

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

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

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

22import Control.Monad.IO.Class (liftIO)

24+-- Posix

25+import System.Posix.Process (forkProcess, createSession)

26+

22-- SSH

27-- SSH

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

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

24 ( initialize

29 ( initialize

@@ -64,7 +69,7 @@ import Control.Distributed.Process.Backend.Azure

64 , cloudServices

69 , cloudServices

65 , CloudService(cloudServiceName, cloudServiceVMs)

70 , CloudService(cloudServiceName, cloudServiceVMs)

66 , VirtualMachine(vmName)

71 , VirtualMachine(vmName)

67− , Backend(copyToVM, checkMD5, callOnVM)

72+ , Backend(copyToVM, checkMD5, callOnVM, spawnOnVM)

68 )

73 )

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

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

@@ -78,10 +83,11 @@ data ProcessPair b = forall a. Serializable a => ProcessPair {

78 , ppairDict :: Static (SerializableDict a)

83 , ppairDict :: Static (SerializableDict a)

79 }

84 }

81−genericMain :: (RemoteTable -> RemoteTable) -- ^ Standard CH remote table

86+genericMain :: (RemoteTable -> RemoteTable) -- ^ Standard CH remote table

82− -> (String -> IO (ProcessPair ())) -- ^ Closures to support in 'run'

87+ -> (String -> IO (ProcessPair ())) -- ^ Closures to support in 'run'

88+ -> (String -> IO (Closure (Process ()))) -- ^ Closures to support in @run --background@

83 -> IO ()

89 -> IO ()

84−genericMain remoteTable cmds = do

90+genericMain remoteTable callable spawnable = do

85 _ <- SSH.initialize True

91 _ <- SSH.initialize True

86 cmd <- execParser opts

92 cmd <- execParser opts

87 case cmd of

93 case cmd of

@@ -110,20 +116,31 @@ genericMain remoteTable cmds = do

110 if and matches

116 if and matches

111 then exitSuccess

117 then exitSuccess

112 else exitFailure

118 else exitFailure

113− RunOn {} -> do

119+ RunOn {} | background cmd -> do

120+ rProc <- spawnable (closureId cmd)

121+ params <- azureParameters (azureOptions cmd) (Just (sshOptions cmd))

122+ backend <- initializeBackend params

123+ css <- cloudServices backend

124+ forM_ (findTarget (target cmd) css) $ \vm -> do

125+ putStr (vmName vm ++ ": ") >> hFlush stdout

126+ spawnOnVM backend vm (remotePort cmd) rProc

127+ putStrLn "OK"

128+ RunOn {} {- not (background cmd) -} -> do

114− procPair <- cmds (closureId cmd)

129+ procPair <- callable (closureId cmd)

115 params <- azureParameters (azureOptions cmd) (Just (sshOptions cmd))

130 params <- azureParameters (azureOptions cmd) (Just (sshOptions cmd))

116 backend <- initializeBackend params

131 backend <- initializeBackend params

117 css <- cloudServices backend

132 css <- cloudServices backend

118 forM_ (findTarget (target cmd) css) $ \vm -> do

133 forM_ (findTarget (target cmd) css) $ \vm -> do

119 putStr (vmName vm ++ ": ") >> hFlush stdout

134 putStr (vmName vm ++ ": ") >> hFlush stdout

120 case procPair of

135 case procPair of

121− ProcessPair rProc lProc dict ->

136+ ProcessPair rProc lProc dict -> do

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

137+ result <- callOnVM backend dict vm (remotePort cmd) rProc

138+ lProc result

123 OnVmCommand (vmCmd@OnVmRun {}) ->

139 OnVmCommand (vmCmd@OnVmRun {}) ->

124 onVmRun (remoteTable . Azure.remoteTable $ initRemoteTable)

140 onVmRun (remoteTable . Azure.remoteTable $ initRemoteTable)

125 (onVmIP vmCmd)

141 (onVmIP vmCmd)

126 (onVmPort vmCmd)

142 (onVmPort vmCmd)

143+ (onVmBackground vmCmd)

127 SSH.exit

144 SSH.exit

128 where

145 where

129 opts = info (helper <*> commandParser)

146 opts = info (helper <*> commandParser)

@@ -152,19 +169,35 @@ azureParameters opts (Just sshOpts) = do

152 azureSshUserName = remoteUser sshOpts

169 azureSshUserName = remoteUser sshOpts

153 }

170 }

154

171

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

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

156−onVmRun rtable host port = do

173+onVmRun rtable host port bg = do

157− hSetBinaryMode stdin True

174+ hSetBinaryMode stdin True

158− hSetBinaryMode stdout True

175+ hSetBinaryMode stdout True

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

176+ procEnc <- BSL.getContents

177+ -- Force evaluation (so that we can safely close stdin)

178+ _length <- evaluate (BSL.length procEnc)

179+ let proc = decode procEnc

180+ if bg

181+ then do

182+ hClose stdin

183+ hClose stdout

184+ hClose stderr

185+ void . forkProcess $ do

186+ void createSession

187+ startCH proc

188+ else

189+ startCH proc

190+ where

191+ startCH :: Closure (Process ()) -> IO ()

192+ startCH proc = do

160− mTransport <- createTransport host port defaultTCPParameters

193+ mTransport <- createTransport host port defaultTCPParameters

161− case mTransport of

194+ case mTransport of

162− Left err -> throwIO err

195+ Left err -> throwIO err

163− Right transport -> do

196+ Right transport -> do

164− node <- newLocalNode transport rtable

197+ node <- newLocalNode transport rtable

165− runProcess node $

198+ runProcess node $

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

199+ catch (join . unClosure $ proc)

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

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

168

201

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

202--------------------------------------------------------------------------------

170-- Command line options --

203-- Command line options --

@@ -217,8 +250,9 @@ data Command =

217

250

218data OnVmCommand =

251data OnVmCommand =

219 OnVmRun {

252 OnVmRun {

220− onVmIP :: String

253+ onVmIP :: String

221− , onVmPort :: String

254+ , onVmPort :: String

255+ , onVmBackground :: Bool

222 }

256 }

223 deriving Show

257 deriving Show

224

258

@@ -317,6 +351,9 @@ onVmRunParser = OnVmRun

317 & metavar "PORT"

351 & metavar "PORT"

318 & help "port number"

352 & help "port number"

319 )

353 )

354+ <*> switch ( long "background"

355+ & help "Run the process in the background"

356+ )

320

357

321onVmCommandParser :: Parser Command

358onVmCommandParser :: Parser Command

322onVmCommandParser = OnVmCommand <$> subparser

359onVmCommandParser = OnVmCommand <$> subparser