baebc4f · distributed-process

git.iko.soy/distributed-process

on master help sign in

Pass (as yet uninit) backend to remote processes

baebc4fEdsko de Vries14 years ago

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

compare with master

unifiedsplit

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

11 ( remotable

12 , mkClosure

13 )

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

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

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

15 ( genericMain

16 ( genericMain

16 , ProcessPair(..)

17 , ProcessPair(..)

18+ , RemoteProcess

17 )

19 )

19data ControllerMsg =

21data ControllerMsg =

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

24 get = getGeneric

26 get = getGeneric

25 put = putGeneric

27 put = putGeneric

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

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

28−conwayController () = go

30+conwayController () _backend = go

29 where

31 where

30 go = do

32 go = do

31 msg <- expect

33 msg <- expect

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

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

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

42 callable _ = error "spawnable: unknown"

44 callable _ = error "spawnable: unknown"

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

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

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

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

46 spawnable _ = error "callable: unknown"

48 spawnable _ = error "callable: unknown"

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

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

24 , SerializableDict(..)

25 , mkStatic

26 )

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

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

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

28 ( genericMain

29 ( genericMain

29 , ProcessPair(..)

30 , ProcessPair(..)

31+ , RemoteProcess

30 )

32 )

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

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

33sdictString :: SerializableDict String

35sdictString :: SerializableDict String

34sdictString = SerializableDict

36sdictString = SerializableDict

36−ping :: () -> Process String

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

37−ping () = do

39+ping () _backend = do

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

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

39 case mPingServerEnc of

41 case mPingServerEnc of

40 Left err ->

42 Left err ->

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

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

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

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

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

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

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

56−pingServer () = do

58+pingServer () _backend = do

57 pid <- getSelfPid

59 pid <- getSelfPid

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

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

59 forever $ do

61 forever $ do

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

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

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

70 callable _ = error "spawnable: unknown"

72 callable _ = error "spawnable: unknown"

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

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

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

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

74 spawnable _ = error "callable: unknown"

76 spawnable _ = error "callable: unknown"

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

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

42 Extensions: ViewPatterns,

43 RankNTypes,

44 ExistentialQuantification,

45− ScopedTypeVariables

45+ ScopedTypeVariables,

46+ DeriveDataTypeable

46 ghc-options: -Wall

47 ghc-options: -Wall

47 HS-Source-Dirs: src

48 HS-Source-Dirs: src

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

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

18import Data.Binary (encode, decode)

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

20import 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)

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

23import Data.Typeable (Typeable)

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

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

69 )

70import Control.Distributed.Process.Closure

71 ( remotable

72− , cpBind

72+ , cpComp

73 , SerializableDict(SerializableDict)

74 , staticConst

75 , staticApply

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

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

105 , checkMD5 :: VirtualMachine -> IO Bool

106 -- | @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

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

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

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

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

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

115+ , spawnOnVM :: VirtualMachine

111− }

116+ -> String

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

118+ -> IO ()

119+ } deriving (Typeable)

112

120

113data AzureParameters = AzureParameters {

121data AzureParameters = AzureParameters {

114 azureSetup :: AzureSetup

122 azureSetup :: AzureSetup

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

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

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

171

179

172-- | Call a process on a VM

180-- | 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

174apiCallOnVM params dict vm port proc =

188apiCallOnVM params dict vm port proc =

175− withSSH2 params vm $ \s -> do

189+ withSSH2 params vm $ \s -> do

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

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

177− ++ " onvm run "

191+ ++ " onvm run "

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

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

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

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

180− ++ " 2>&1"

194+ ++ " 2>&1"

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

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

182− SSH.channelExecute ch exe

196+ SSH.channelExecute ch exe

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

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

184− SSH.channelSendEOF ch

198+ SSH.channelSendEOF ch

185− SSH.readAllChannel ch

199+ SSH.readAllChannel ch

186− if status == 0

200+ if status == 0

187− then return $ decode r

201+ then return $ decode r

188− else error (BSLC.unpack r)

202+ else error (BSLC.unpack r)

203+ where

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

205+ proc' = proc `cpComp` cpEncodeToStdout dict

189

206

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

207+apiSpawnOnVM :: AzureParameters

208+ -> VirtualMachine

209+ -> String

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

211+ -> IO ()

191apiSpawnOnVM params vm port proc =

212apiSpawnOnVM params vm port proc =

192 withSSH2 params vm $ \s -> do

213 withSSH2 params vm $ \s -> do

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

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

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

196 ++ " --port " ++ port

217 ++ " --port " ++ port

197 ++ " --background "

218 ++ " --background "

198 ++ " 2>&1"

219 ++ " 2>&1"

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

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

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

201 SSH.channelExecute ch exe

221 SSH.channelExecute ch exe

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

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

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

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

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

3 ( genericMain

4 , ProcessPair(..)

5+ , RemoteProcess

6+ , LocalProcess

5 ) where

7 ) where

7import Prelude hiding (catch)

9import Prelude hiding (catch)

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

16 )

18 )

17import Data.Binary (decode)

19import Data.Binary (decode)

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

20import 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)

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

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

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

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

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

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

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

76-- Main --

78-- Main --

77--------------------------------------------------------------------------------

79--------------------------------------------------------------------------------

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

82+type LocalProcess a = IO a

83+

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

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

80− ppairRemote :: Closure (Process a)

85+ ppairRemote :: RemoteProcess a

81− , ppairLocal :: a -> IO b

86+ , ppairLocal :: a -> LocalProcess b

82 , ppairDict :: Static (SerializableDict a)

87 , ppairDict :: Static (SerializableDict a)

83 }

88 }

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

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

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

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

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

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

88 -> IO ()

93 -> IO ()

89genericMain remoteTable callable spawnable = do

94genericMain remoteTable callable spawnable = do

90 _ <- SSH.initialize True

95 _ <- SSH.initialize True

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

170 procEnc <- BSL.getContents

175 procEnc <- BSL.getContents

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

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

172 _length <- evaluate (BSL.length procEnc)

177 _length <- evaluate (BSL.length procEnc)

173− let proc = decode procEnc

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

174 if bg

179 if bg

175 then do

180 then do

176 hClose stdin

181 hClose stdin

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

182 else

187 else

183 startCH proc

188 startCH proc

184 where

189 where

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

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

186− startCH proc = do

191+ startCH rproc = do

187 mTransport <- createTransport host port defaultTCPParameters

192 mTransport <- createTransport host port defaultTCPParameters

188 case mTransport of

193 case mTransport of

189 Left err -> throwIO err

194 Left err -> throwIO err

190 Right transport -> do

195 Right transport -> do

191 node <- newLocalNode transport rtable

196 node <- newLocalNode transport rtable

192− runProcess node $

197+ runProcess node $ do

193− catch (join . unClosure $ proc)

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

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

200+ catch (proc backend)

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

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

195

202

196--------------------------------------------------------------------------------

203--------------------------------------------------------------------------------

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

290 )

297 )

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

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

292 & metavar "VM"

299 & metavar "VM"

293− & help "Virtual machine name"

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

294 ))

301 ))

295

302

296checkMD5Parser :: Parser Command

303checkMD5Parser :: Parser Command