git.iko.soy/distributed-process
on master help sign in
Support spawn (as well as call)
commit32d3bf424da5aec81f8eb9488099b04e44fec535 authorEdsko de Vries <edsko@well-typed.com> dateMon 6 Aug 2012, 11:56 (14 years ago) parentb0fb243
4 files changed+121−49
compare with master
unifiedsplit
distributed-process-azure/distributed-process-azure.cabal+2−1 distributed-process-azure/src/Control/Distributed/Process/Backend/Azure.hs+34−16 distributed-process-azure/src/Control/Distributed/Process/Backend/Azure/CommandLine.hs+22−6 distributed-process-azure/src/Control/Distributed/Process/Backend/Azure/GenericMain.hs+63−26
distributed-process-azure/distributed-process-azure.cabal+2−1
@@ -31,7 +31,8 @@ Library
3131 network-transport-tcp >= 0.2 && < 0.3,
3232 optparse-applicative >= 0.2 && < 0.4,
3333 transformers >= 0.3 && < 0.4,
34− certificate == 1.2.3
34+ certificate == 1.2.3,
35+ unix >= 2.5 && < 2.6
3536 Exposed-modules: Control.Distributed.Process.Backend.Azure,
3637 Control.Distributed.Process.Backend.Azure.GenericMain
3738 Extensions: ViewPatterns,
distributed-process-azure/src/Control/Distributed/Process/Backend/Azure.hs+34−16
@@ -17,7 +17,7 @@ import System.Environment.Executable (getExecutablePath)
1717import Data.Binary (encode, decode)
1818import Data.Digest.Pure.MD5 (md5, MD5Digest)
1919import 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)
2121import Data.Typeable (Typeable)
2222import Control.Applicative ((<$>))
2323import Control.Monad (void)
@@ -102,6 +102,9 @@ data Backend = Backend {
102102 , checkMD5 :: VirtualMachine -> IO Bool
103103 -- | @runOnVM dict vm port p@ starts a CH node on port 'port' and runs 'p'
104104 , 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 ()
105108 }
106109
107110data AzureParameters = AzureParameters {
@@ -150,6 +153,7 @@ initializeBackend params = do
150153 , copyToVM = apiCopyToVM params
151154 , checkMD5 = apiCheckMD5 params
152155 , callOnVM = apiCallOnVM params
156+ , spawnOnVM = apiSpawnOnVM params
153157 }
154158
155159-- | Start a CH node on the given virtual machine
@@ -161,21 +165,35 @@ apiCopyToVM params vm =
161165-- | Call a process on a VM
162166apiCallOnVM :: Serializable a => AzureParameters -> Static (SerializableDict a) -> VirtualMachine -> String -> Closure (Process a) -> IO a
163167apiCallOnVM params dict vm port proc =
164− withSSH2 params vm $ \s -> do
165− let exe = "PATH=. " ++ azureSshRemotePath params
166− ++ " onvm run "
167− ++ " --host " ++ vmIpAddress vm
168− ++ " --port " ++ port
169− ++ " 2>&1"
170− (_, r) <- SSH.withChannelBy (SSH.openChannelSession s) id $ \ch -> do
171− SSH.channelExecute ch exe
172− _cnt <- SSH.writeAllChannel ch (encode proc')
173− SSH.channelSendEOF ch
174− SSH.readAllChannel ch
175− return (decode r)
176− where
177− proc' :: Closure (Process ())
178− proc' = proc `cpBind` cpEncodeToStdout dict
168+ withSSH2 params vm $ \s -> do
169+ let exe = "PATH=. " ++ azureSshRemotePath params
170+ ++ " onvm run "
171+ ++ " --host " ++ vmIpAddress vm
172+ ++ " --port " ++ port
173+ ++ " 2>&1"
174+ (_, r) <- SSH.withChannelBy (SSH.openChannelSession s) id $ \ch -> do
175+ SSH.channelExecute ch exe
176+ _cnt <- SSH.writeAllChannel ch (encode $ proc `cpBind` cpEncodeToStdout dict)
177+ SSH.channelSendEOF ch
178+ SSH.readAllChannel ch
179+ return (decode r)
180+
181+apiSpawnOnVM :: AzureParameters -> VirtualMachine -> String -> Closure (Process ()) -> IO ()
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
179197
180198-- | Check the MD5 hash of the executable on the remote machine
181199apiCheckMD5 :: AzureParameters -> VirtualMachine -> IO Bool
distributed-process-azure/src/Control/Distributed/Process/Backend/Azure/CommandLine.hs+22−6
@@ -1,7 +1,8 @@
11{-# LANGUAGE TemplateHaskell #-}
22
33import Control.Monad.IO.Class (liftIO)
4−import Control.Distributed.Process (Process, ProcessId, getSelfPid)
4+import Control.Concurrent (threadDelay)
5+import Control.Distributed.Process (Process, ProcessId, getSelfPid, Closure)
56import Control.Distributed.Process.Closure (remotable, mkClosure, sdictProcessId)
67import Control.Distributed.Process.Backend.Azure.GenericMain
78 ( genericMain
@@ -13,9 +14,24 @@ getPid () = do
1314 liftIO $ appendFile "Log" "getPid did run"
1415 getSelfPid
1516
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)
1725
18−main = genericMain __remoteTable $ \cmd ->
19− case cmd of
20− "hello" -> return $ ProcessPair ($(mkClosure 'getPid) ()) print sdictProcessId
21− _ -> error "unknown command"
26+remotable ['getPid, 'logN]
27+
28+main :: IO ()
29+main = genericMain __remoteTable callable spawnable
30+ where
31+ callable :: String -> IO (ProcessPair ())
32+ callable "getPid" = return $ ProcessPair ($(mkClosure 'getPid) ()) print sdictProcessId
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
1010 ( hFlush
1111 , stdout
1212 , stdin
13+ , stderr
1314 , hSetBinaryMode
15+ , hClose
1416 )
1517import Data.Binary (decode)
16−import qualified Data.ByteString.Lazy as BSL (ByteString, getContents)
17−import Control.Monad (unless, forM, forM_, join)
18−import Control.Exception (throwIO, SomeException)
18+import qualified Data.ByteString.Lazy as BSL (getContents, length)
19+import Control.Monad (unless, forM, forM_, join, void)
20+import Control.Exception (throwIO, SomeException, evaluate)
1921import Control.Applicative ((<$>), (<*>), (<|>))
2022import Control.Monad.IO.Class (liftIO)
2123
24+-- Posix
25+import System.Posix.Process (forkProcess, createSession)
26+
2227-- SSH
2328import qualified Network.SSH.Client.LibSSH2.Foreign as SSH
2429 ( initialize
@@ -64,7 +69,7 @@ import Control.Distributed.Process.Backend.Azure
6469 , cloudServices
6570 , CloudService(cloudServiceName, cloudServiceVMs)
6671 , VirtualMachine(vmName)
67− , Backend(copyToVM, checkMD5, callOnVM)
72+ , Backend(copyToVM, checkMD5, callOnVM, spawnOnVM)
6873 )
6974import qualified Control.Distributed.Process.Backend.Azure as Azure (remoteTable)
7075
@@ -78,10 +83,11 @@ data ProcessPair b = forall a. Serializable a => ProcessPair {
7883 , ppairDict :: Static (SerializableDict a)
7984 }
8085
81−genericMain :: (RemoteTable -> RemoteTable) -- ^ Standard CH remote table
82− -> (String -> IO (ProcessPair ())) -- ^ Closures to support in 'run'
86+genericMain :: (RemoteTable -> RemoteTable) -- ^ Standard CH remote table
87+ -> (String -> IO (ProcessPair ())) -- ^ Closures to support in 'run'
88+ -> (String -> IO (Closure (Process ()))) -- ^ Closures to support in @run --background@
8389 -> IO ()
84−genericMain remoteTable cmds = do
90+genericMain remoteTable callable spawnable = do
8591 _ <- SSH.initialize True
8692 cmd <- execParser opts
8793 case cmd of
@@ -110,20 +116,31 @@ genericMain remoteTable cmds = do
110116 if and matches
111117 then exitSuccess
112118 else exitFailure
113− RunOn {} -> do
114− procPair <- cmds (closureId cmd)
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
129+ procPair <- callable (closureId cmd)
115130 params <- azureParameters (azureOptions cmd) (Just (sshOptions cmd))
116131 backend <- initializeBackend params
117132 css <- cloudServices backend
118133 forM_ (findTarget (target cmd) css) $ \vm -> do
119134 putStr (vmName vm ++ ": ") >> hFlush stdout
120135 case procPair of
121− ProcessPair rProc lProc dict ->
122− callOnVM backend dict vm (remotePort cmd) rProc >>= lProc
136+ ProcessPair rProc lProc dict -> do
137+ result <- callOnVM backend dict vm (remotePort cmd) rProc
138+ lProc result
123139 OnVmCommand (vmCmd@OnVmRun {}) ->
124140 onVmRun (remoteTable . Azure.remoteTable $ initRemoteTable)
125141 (onVmIP vmCmd)
126142 (onVmPort vmCmd)
143+ (onVmBackground vmCmd)
127144 SSH.exit
128145 where
129146 opts = info (helper <*> commandParser)
@@ -152,19 +169,35 @@ azureParameters opts (Just sshOpts) = do
152169 azureSshUserName = remoteUser sshOpts
153170 }
154171
155−onVmRun :: RemoteTable -> String -> String -> IO ()
156−onVmRun rtable host port = do
157− hSetBinaryMode stdin True
158− hSetBinaryMode stdout True
159− proc <- BSL.getContents :: IO BSL.ByteString
160− mTransport <- createTransport host port defaultTCPParameters
161− case mTransport of
162− Left err -> throwIO err
163− Right transport -> do
164− node <- newLocalNode transport rtable
165− runProcess node $
166− catch (join . unClosure . decode $ proc)
167− (\e -> liftIO (print (e :: SomeException) >> throwIO e))
172+onVmRun :: RemoteTable -> String -> String -> Bool -> IO ()
173+onVmRun rtable host port bg = do
174+ hSetBinaryMode stdin True
175+ hSetBinaryMode stdout True
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
193+ mTransport <- createTransport host port defaultTCPParameters
194+ case mTransport of
195+ Left err -> throwIO err
196+ Right transport -> do
197+ node <- newLocalNode transport rtable
198+ runProcess node $
199+ catch (join . unClosure $ proc)
200+ (\e -> liftIO (print (e :: SomeException) >> throwIO e))
168201
169202--------------------------------------------------------------------------------
170203-- Command line options --
@@ -217,8 +250,9 @@ data Command =
217250
218251data OnVmCommand =
219252 OnVmRun {
220− onVmIP :: String
221− , onVmPort :: String
253+ onVmIP :: String
254+ , onVmPort :: String
255+ , onVmBackground :: Bool
222256 }
223257 deriving Show
224258
@@ -317,6 +351,9 @@ onVmRunParser = OnVmRun
317351 & metavar "PORT"
318352 & help "port number"
319353 )
354+ <*> switch ( long "background"
355+ & help "Run the process in the background"
356+ )
320357
321358onVmCommandParser :: Parser Command
322359onVmCommandParser = OnVmCommand <$> subparser