baebc4f · distributed-process

git.iko.soy/distributed-process

Pass (as yet uninit) backend to remote processes

commitbaebc4f9e0e4a009d094f3c467408e6862c19008 authorEdsko de Vries <edsko@well-typed.com> dateMon 6 Aug 2012, 17:49 (14 years ago) parentd1004d5

5 files changed+75−43

compare with master

unifiedsplit

distributed-process-azure/demos/Conway.hs+5−3 distributed-process-azure/demos/Ping.hs+7−5 distributed-process-azure/distributed-process-azure.cabal+2−1 distributed-process-azure/src/Control/Distributed/Process/Backend/Azure.hs+42−22 distributed-process-azure/src/Control/Distributed/Process/Backend/Azure/GenericMain.hs+19−12

distributed-process-azure/demos/Conway.hs+5−3

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

1111 ( remotable

1212 , mkClosure

1313 )

14+import Control.Distributed.Process.Backend.Azure (Backend)

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

1516 ( genericMain

1617 , ProcessPair(..)

18+ , RemoteProcess

1719 )

1820

1921data ControllerMsg =

@@ -24,8 +26,8 @@ instance Binary ControllerMsg where

2426 get = getGeneric

2527 put = putGeneric

2628

27−conwayController :: () -> Process ()

28−conwayController () = go

29+conwayController :: () -> Backend -> Process ()

30+conwayController () _backend = go

2931 where

3032 go = do

3133 msg <- expect

@@ -41,6 +43,6 @@ main = genericMain __remoteTable callable spawnable

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

4244 callable _ = error "spawnable: unknown"

4345

44− spawnable :: String -> IO (Closure (Process ()))

46+ spawnable :: String -> IO (RemoteProcess ())

4547 spawnable "controller" = return $ $(mkClosure 'conwayController) ()

4648 spawnable _ = error "callable: unknown"

distributed-process-azure/demos/Ping.hs+7−5

@@ -24,17 +24,19 @@ import Control.Distributed.Process.Closure

2424 , SerializableDict(..)

2525 , mkStatic

2626 )

27+import Control.Distributed.Process.Backend.Azure (Backend)

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

2829 ( genericMain

2930 , ProcessPair(..)

31+ , RemoteProcess

3032 )

3133import qualified Data.ByteString.Lazy as BSL (readFile, writeFile)

3234

3335sdictString :: SerializableDict String

3436sdictString = SerializableDict

3537

36−ping :: () -> Process String

37−ping () = do

38+ping :: () -> Backend -> Process String

39+ping () _backend = do

3840 mPingServerEnc <- liftIO $ try (BSL.readFile "pingServer.pid")

3941 case mPingServerEnc of

4042 Left err ->

@@ -52,8 +54,8 @@ ping () = do

5254 then return $ "Ping server at " ++ show pingServer ++ " ok"

5355 else return $ "Ping server at " ++ show pingServer ++ " failure"

5456

55−pingServer :: () -> Process ()

56−pingServer () = do

57+pingServer :: () -> Backend -> Process ()

58+pingServer () _backend = do

5759 pid <- getSelfPid

5860 liftIO $ BSL.writeFile "pingServer.pid" (encode pid)

5961 forever $ do

@@ -69,6 +71,6 @@ main = genericMain __remoteTable callable spawnable

6971 callable "ping" = return $ ProcessPair ($(mkClosure 'ping) ()) putStrLn $(mkStatic 'sdictString)

7072 callable _ = error "spawnable: unknown"

7173

72− spawnable :: String -> IO (Closure (Process ()))

74+ spawnable :: String -> IO (RemoteProcess ())

7375 spawnable "pingServer" = return $ $(mkClosure 'pingServer) ()

7476 spawnable _ = error "callable: unknown"

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

@@ -42,7 +42,8 @@ Library

4242 Extensions: ViewPatterns,

4343 RankNTypes,

4444 ExistentialQuantification,

45− ScopedTypeVariables

45+ ScopedTypeVariables,

46+ DeriveDataTypeable

4647 ghc-options: -Wall

4748 HS-Source-Dirs: src

4849

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

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

1818import Data.Binary (encode, decode)

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

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

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

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

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

2323import Data.Typeable (Typeable)

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

@@ -69,7 +69,7 @@ import Control.Distributed.Process

6969 )

7070import Control.Distributed.Process.Closure

7171 ( remotable

72− , cpBind

72+ , cpComp

7373 , SerializableDict(SerializableDict)

7474 , staticConst

7575 , staticApply

@@ -104,11 +104,19 @@ data Backend = Backend {

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

105105 , checkMD5 :: VirtualMachine -> IO Bool

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

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

107+ , callOnVM :: forall a. Serializable a

108+ => Static (SerializableDict a)

109+ -> VirtualMachine

110+ -> String

111+ -> Closure (Backend -> Process a)

112+ -> IO a

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

109114 -- The CH node will exit when the process exists.

110− , spawnOnVM :: VirtualMachine -> String -> Closure (Process ()) -> IO ()

111− }

115+ , spawnOnVM :: VirtualMachine

116+ -> String

117+ -> Closure (Backend -> Process ())

118+ -> IO ()

119+ } deriving (Typeable)

112120

113121data AzureParameters = AzureParameters {

114122 azureSetup :: AzureSetup

@@ -170,24 +178,37 @@ apiCopyToVM params vm =

170178 SSH.scpSendFile s 0o700 (azureSshLocalPath params) (azureSshRemotePath params)

171179

172180-- | Call a process on a VM

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

181+apiCallOnVM :: Serializable a

182+ => AzureParameters

183+ -> Static (SerializableDict a)

184+ -> VirtualMachine

185+ -> String

186+ -> Closure (Backend -> Process a)

187+ -> IO a

174188apiCallOnVM params dict vm port proc =

175− withSSH2 params vm $ \s -> do

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

177− ++ " onvm run "

178− ++ " --host " ++ vmIpAddress vm

179− ++ " --port " ++ port

180− ++ " 2>&1"

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

182− SSH.channelExecute ch exe

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

184− SSH.channelSendEOF ch

185− SSH.readAllChannel ch

186− if status == 0

187− then return $ decode r

188− else error (BSLC.unpack r)

189+ withSSH2 params vm $ \s -> do

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

191+ ++ " onvm run "

192+ ++ " --host " ++ vmIpAddress vm

193+ ++ " --port " ++ port

194+ ++ " 2>&1"

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

196+ SSH.channelExecute ch exe

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

198+ SSH.channelSendEOF ch

199+ SSH.readAllChannel ch

200+ if status == 0

201+ then return $ decode r

202+ else error (BSLC.unpack r)

203+ where

204+ proc' :: Closure (Backend -> Process ())

205+ proc' = proc `cpComp` cpEncodeToStdout dict

189206

190−apiSpawnOnVM :: AzureParameters -> VirtualMachine -> String -> Closure (Process ()) -> IO ()

207+apiSpawnOnVM :: AzureParameters

208+ -> VirtualMachine

209+ -> String

210+ -> Closure (Backend -> Process ())

211+ -> IO ()

191212apiSpawnOnVM params vm port proc =

192213 withSSH2 params vm $ \s -> do

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

@@ -196,7 +217,6 @@ apiSpawnOnVM params vm port proc =

196217 ++ " --port " ++ port

197218 ++ " --background "

198219 ++ " 2>&1"

199− BSL.writeFile "closure" (encode proc)

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

201221 SSH.channelExecute ch exe

202222 _cnt <- SSH.writeAllChannel ch (encode proc)

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

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

22module Control.Distributed.Process.Backend.Azure.GenericMain

33 ( genericMain

44 , ProcessPair(..)

5+ , RemoteProcess

6+ , LocalProcess

57 ) where

79import Prelude hiding (catch)

@@ -16,7 +18,7 @@ import System.IO

1618 )

1719import Data.Binary (decode)

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

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

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

2022import Control.Exception (throwIO, SomeException, evaluate)

2123import Control.Applicative ((<$>), (<*>), optional)

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

@@ -76,15 +78,18 @@ import qualified Control.Distributed.Process.Backend.Azure as Azure (remoteTable

7678-- Main --

7779--------------------------------------------------------------------------------

7880

81+type RemoteProcess a = Closure (Backend -> Process a)

82+type LocalProcess a = IO a

83+

7984data ProcessPair b = forall a. Serializable a => ProcessPair {

80− ppairRemote :: Closure (Process a)

81− , ppairLocal :: a -> IO b

85+ ppairRemote :: RemoteProcess a

86+ , ppairLocal :: a -> LocalProcess b

8287 , ppairDict :: Static (SerializableDict a)

8388 }

8489

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

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

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

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

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

92+ -> (String -> IO (RemoteProcess ())) -- ^ Closures to support in @run --background@

8893 -> IO ()

8994genericMain remoteTable callable spawnable = do

9095 _ <- SSH.initialize True

@@ -170,7 +175,7 @@ onVmRun rtable host port bg = do

170175 procEnc <- BSL.getContents

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

172177 _length <- evaluate (BSL.length procEnc)

173− let proc = decode procEnc

178+ let proc = decode procEnc :: RemoteProcess ()

174179 if bg

175180 then do

176181 hClose stdin

@@ -182,15 +187,17 @@ onVmRun rtable host port bg = do

182187 else

183188 startCH proc

184189 where

185− startCH :: Closure (Process ()) -> IO ()

186− startCH proc = do

190+ startCH :: RemoteProcess () -> IO ()

191+ startCH rproc = do

187192 mTransport <- createTransport host port defaultTCPParameters

188193 case mTransport of

189194 Left err -> throwIO err

190195 Right transport -> do

191196 node <- newLocalNode transport rtable

192− runProcess node $

193− catch (join . unClosure $ proc)

197+ runProcess node $ do

198+ let backend = error "TODO: backend not initialized in onVmRun"

199+ proc <- unClosure rproc :: Process (Backend -> Process ())

200+ catch (proc backend)

194201 (\e -> liftIO (print (e :: SomeException) >> throwIO e))

195202

196203--------------------------------------------------------------------------------

@@ -290,7 +297,7 @@ targetParser = Target

290297 )

291298 <*> optional (strOption ( long "virtual-machine"

292299 & metavar "VM"

293− & help "Virtual machine name"

300+ & help "Virtual machine name (all VMs if unspecified)"

294301 ))

295302

296303checkMD5Parser :: Parser Command