@@ -103,13 +103,13 @@
103103--
104104-- > cabal-dev install distributed-process-azure -f build-demos
105105--
106−-- We can use any of the demos to test our setup; we will use the @ping@ demo:
106+-- We can the @cloud-haskell-azure-ping@ demo to test our setup:
107107--
108108-- > cloud-haskell-azure-ping list \
109−-- > --subscription-id <<your subscription ID>> \
110−-- > --certificate /path/to/credentials.x509 \
111−-- > --private /path/to/credentials.private
112−--
109+-- > <<your subscription ID>> \
110+-- > /path/to/credentials.x509 \
111+-- > /path/to/credentials.private
112+--
113113-- (you can find your subscription ID in the @.publishsettings@ file from the previous step).
114114-- If everything went well, this will output something like
115115--
@@ -182,63 +182,35 @@
182182-- machine will echo this back; the local machine will wait for the echo, show
183183-- the echo, and repeat.
184184--
185−-- Before you can try it you will first need to copy the executable to the remote server:
186−--
187−-- > cloud-haskell-azure-echo install \
188−-- > --subscription-id <<your subscription ID>> \
189−-- > --certificate /path/to/credentials.x509 \
190−-- > --private /path/to/credentials.private \
191−-- > --user <<remote username>> \
192−-- > --cloud-service <<Cloud Service name>> \
193−-- > --virtual-machine <<Virtual Machine name>>
194−--
195−-- (If you leave out the @--virtual-machine@ argument the binary will be copied
196−-- to every virtual machine in the specified cloud service). Once installed,
197−-- you can run it as follows:
198−--
199−-- > cloud-haskell-azure-echo run \
200−-- > --subscription-id <<your subscription ID>> \
201−-- > --certificate /path/to/credentials.x509 \
202−-- > --private /path/to/credentials.private \
203−-- > --user <<remote username>> \
204−-- > --cloud-service <<Cloud Service name>> \
205−-- > --virtual-machine <<Virtual Machine name>> \
206−-- > --port 8080 \
207−-- > --closure echo
208−-- > CHDemo1: # Everything I type will be echoed back
209−-- > Echo: Everything I type will be echoed back
185+-- Before you can try it you will first need to copy the executable (for
186+-- example, using scp, although the Azure backend also provides this natively
187+-- in Haskell). Once that's done, you can run the demo as follows:
188+--
189+-- > cloud-haskell-azure-echo \
190+-- > <<subscription ID>> \
191+-- > /path/to/credentials.x509 \
192+-- > /path/to/credentials.private \
193+-- > <<remote username>> \
194+-- > <<cloud service name>> \
195+-- > <<virtual machine name>> \
196+-- > <<port number>>
197+-- > # Everything I type gets echoed back
198+-- > Echo: Everything I type gets echoed back
210199-- > # Until I enter a blank line
211200-- > Echo: Until I enter a blank line
212201-- > #
213202--
214−-- "Control.Distributed.Process.Backend.Azure.GenericMain" provides a generic
215−-- main function that you can use to structure your code. It provides command line
216−-- arguments such as the ones we saw in section /Testing the Setup/, it
217−-- initializes the Azure backend, and it takes care of running your code on the
218−-- remote machines. You don't have to use 'genericMain' if you prefer not to,
219−-- but then it will be your own responsibility to initialize Azure and to make
220−-- sure that your executable does the right thing when it's invoked on the
221−-- remote node. In these notes we will assume that you will use 'genericMain'.
222−--
223203-- The full @echo@ demo is
224204--
225205-- > {-# LANGUAGE TemplateHaskell #-}
226206-- >
227207-- > import System.IO (hFlush, stdout)
208+-- > import System.Environment (getArgs)
228209-- > import Control.Monad (unless, forever)
229210-- > import Control.Monad.IO.Class (liftIO)
230211-- > import Control.Distributed.Process (Process, expect)
231212-- > import Control.Distributed.Process.Closure (remotable, mkClosure)
232213-- > import Control.Distributed.Process.Backend.Azure
233−-- > ( Backend
234−-- > , ProcessPair(..)
235−-- > , RemoteProcess
236−-- > , LocalProcess
237−-- > , localExpect
238−-- > , remoteSend
239−-- > , localSend
240−-- > )
241−-- > import Control.Distributed.Process.Backend.Azure.GenericMain (genericMain)
242214-- >
243215-- > echoRemote :: () -> Backend -> Process ()
244216-- > echoRemote () _backend = forever $ do
@@ -258,84 +230,61 @@
258230-- > echoLocal
259231-- >
260232-- > main :: IO ()
261−-- > main = genericMain __remoteTable callable spawnable
262−-- > where
263−-- > callable :: String -> IO (ProcessPair ())
264−-- > callable "echo" = return $ ProcessPair ($(mkClosure 'echoRemote) ()) echoLocal
265−-- > callable _ = error "callable: unknown"
266−-- >
267−-- > spawnable :: String -> IO (RemoteProcess ())
268−-- > spawnable _ = error "spawnable: unknown"
269−--
270−-- 'genericMain' expects three arguments: the first is the standard
271−-- '__remoteTable' argument familiar from
272−-- "Control.Distributed.Process.Closure"; the second and third should map
273−-- strings to process pairs (for use with 'callOnVM') or remote processes (for
274−-- use with 'spawnOnVM') respectively.
275−--
276−-- When you invoke the @echo@ demo with @--closure echo@ the 'genericMain'
277−-- function calls 'callOnVM' with the process pair consisting of 'echoRemote'
278−-- and 'echoLocal'. Hopefully the definition of these two functions is
279−-- self-explanatory.
233+-- > main = do
234+-- > args <- getArgs
235+-- > case args of
236+-- > "onvm":args' -> onVmMain __remoteTable args'
237+-- > sid:x509:pkey:user:cloudService:virtualMachine:port:_ -> do
238+-- > params <- defaultAzureParameters sid x509 pkey
239+-- > let params' = params { azureSshUserName = user }
240+-- > backend <- initializeBackend params' cloudService
241+-- > Just vm <- findNamedVM backend virtualMachine
242+-- > callOnVM backend vm port $
243+-- > ProcessPair ($(mkClosure 'echoRemote) ())
244+-- > echoLocal
280245--
281246-- [Second Example: Ping]
282247--
283248-- The second example differs from the @echo@ demo in that it uses both
284−-- 'callable' ('callOnVM') and 'spawnable' ('spawnOnVM'). It uses the latter to
249+-- 'callOnVM' and 'spawnOnVM'. It uses the latter to
285250-- install a ping server which keeps running in the background; it uses the
286251-- former to run a ping client which sends a request to the ping server and
287−-- outputs the response. As with the @echo@ server, we must first copy the
288−-- executable:
289−--
290−-- > cloud-haskell-azure-ping install \
291−-- > --subscription-id <<your subscription ID>> \
292−-- > --certificate /path/to/credentials.x509 \
293−-- > --private /path/to/credentials.private \
294−-- > --user <<remote username>> \
295−-- > --cloud-service <<Cloud Service name>> \
296−-- > CHDemo3: Done
297−-- > CHDemo2: Done
298−-- > CHDemo1: Done
299−--
300−-- Now we can start the ping server on every virtual machine in the cloud
301−-- service (to install it to a single virtual machine only, pass the
302−-- @--virtual-machine@ argument):
303−--
304−-- > cloud-haskell-azure-ping run \
305−-- > --subscription-id <<your subscription ID>> \
306−-- > --certificate /path/to/credentials.x509 \
307−-- > --private /path/to/credentials.private \
308−-- > --user <<remote username>> \
309−-- > --cloud-service <<Cloud Service name>> \
310−-- > --port 8080
311−-- > --closure server
312−-- > --background
313−-- > CHDemo3: OK
314−-- > CHDemo2: OK
315−-- > CHDemo1: OK
252+-- outputs the response.
253+--
254+-- As with the @echo@ demo, make sure to copy the executable to the remote server first.
255+-- Once that is done, you can start a ping server on a virtual machine using
256+--
257+-- > cloud-haskell-azure-ping server \
258+-- > <<subscription ID>> \
259+-- > /path/to/credentials.x509 \
260+-- > /path/to/credentials.private \
261+-- > <<remote username>> \
262+-- > <<cloud service name>> \
263+-- > <<virtual machine name>> \
264+-- > <<port number>>
316265--
317266-- Finally, we can run the ping client:
318267--
319−-- > cloud-haskell-azure-ping run
320−-- > --subscription-id <<your subscription ID>> \
321−-- > --certificate /path/to/credentials.x509 \
322−-- > --private /path/to/credentials.private \
323−-- > --user <<remote username>> \
324−-- > --cloud-service <<Cloud Service name>> \
325−-- > --port 8081
326−-- > --closure client
327−-- > CHDemo3: Ping server at pid://10.119.182.127:8080:0:2 ok
328−-- > CHDemo2: Ping server at pid://10.59.238.125:8080:0:2 ok
329−-- > CHDemo1: Ping server at pid://10.59.224.122:8080:0:2 ok
268+-- > cloud-haskell-azure-ping client \
269+-- > <<subscription ID>> \
270+-- > /path/to/credentials.x509 \
271+-- > /path/to/credentials.private \
272+-- > <<remote username>> \
273+-- > <<cloud service name>> \
274+-- > <<virtual machine name>> \
275+-- > <<DIFFERENT port number>>
276+-- > Ping server at pid://10.59.224.122:8080:0:2 ok
330277--
331278-- Note that we must pass a different port number, because the client will run
332279-- within its own Cloud Haskell instance.
333280--
334−-- The code for the ping demo is similar to the echo demo, but demonstrates
335−-- both 'callable' and 'spawnable' and shows one way to discover nodes.
281+-- The code for the @ping@ demo is similar to the @echo@ demo, but uses both
282+-- 'callOnVM' and 'spawnOnVM' and demonstrates a way to discover processes (in
283+-- this case, through a PID file).
336284--
337285-- > {-# LANGUAGE TemplateHaskell #-}
338286-- >
287+-- > import System.Environment (getArgs)
339288-- > import Data.Binary (encode, decode)
340289-- > import Control.Monad (forever)
341290-- > import Control.Monad.IO.Class (liftIO)
@@ -352,14 +301,6 @@
352301-- > )
353302-- > import Control.Distributed.Process.Closure (remotable, mkClosure)
354303-- > import Control.Distributed.Process.Backend.Azure
355−-- > ( Backend
356−-- > , ProcessPair(..)
357−-- > , RemoteProcess
358−-- > , LocalProcess
359−-- > , localExpect
360−-- > , remoteSend
361−-- > )
362−-- > import Control.Distributed.Process.Backend.Azure.GenericMain (genericMain)
363304-- > import qualified Data.ByteString.Lazy as BSL (readFile, writeFile)
364305-- >
365306-- > pingServer :: () -> Backend -> Process ()
@@ -395,28 +336,34 @@
395336-- > pingClientLocal = localExpect >>= liftIO . putStrLn
396337-- >
397338-- > main :: IO ()
398−-- > main = genericMain __remoteTable callable spawnable
399−-- > where
400−-- > callable :: String -> IO (ProcessPair ())
401−-- > callable "client" = return $ ProcessPair ($(mkClosure 'pingClientRemote) ()) pingClientLocal
402−-- > callable _ = error "callable: unknown"
403−-- >
404−-- > spawnable :: String -> IO (RemoteProcess ())
405−-- > spawnable "server" = return $ ($(mkClosure 'pingServer) ())
406−-- > spawnable _ = error "spawnable: unknown"
407−--
408−-- The ping server stores its PID in a file, which the ping client attempts to
409−-- read. This kind of pattern is typical, and is provided for in the high-level API.
410−--
411−-- [Using the High-Level API]
412−--
413−-- TODO: Does not yet exist.
339+-- > main = do
340+-- > args <- getArgs
341+-- > case args of
342+-- > "onvm":args' -> onVmMain __remoteTable args'
343+-- > "list":sid:x509:pkey:_ -> do
344+-- > params <- defaultAzureParameters sid x509 pkey
345+-- > css <- cloudServices (azureSetup params)
346+-- > mapM_ print css
347+-- > cmd:sid:x509:pkey:user:cloudService:virtualMachine:port:_ -> do
348+-- > params <- defaultAzureParameters sid x509 pkey
349+-- > let params' = params { azureSshUserName = user }
350+-- > backend <- initializeBackend params' cloudService
351+-- > Just vm <- findNamedVM backend virtualMachine
352+-- > case cmd of
353+-- > "server" -> spawnOnVM backend vm port ($(mkClosure 'pingServer) ())
354+-- > "client" -> callOnVM backend vm port $
355+-- > ProcessPair ($(mkClosure 'pingClientRemote) ())
356+-- > pingClientLocal
414357module Control.Distributed.Process.Backend.Azure
415358 ( -- * Initialization
416359 Backend(..)
417360 , AzureParameters(..)
418361 , defaultAzureParameters
419362 , initializeBackend
363+ -- * Utilities
364+ , findNamedVM
365+ -- * On-VM main
366+ , onVmMain
420367 -- * Re-exports from Azure Service Management
421368 , CloudService(..)
422369 , VirtualMachine(..)
@@ -433,10 +380,22 @@ module Control.Distributed.Process.Backend.Azure
433380 , remoteThrow
434381 ) where
435382
383+import Prelude hiding (catch)
436384import System.Environment (getEnv)
437385import System.FilePath ((</>), takeFileName)
438386import System.Environment.Executable (getExecutablePath)
439−import System.IO (stdout, hFlush)
387+import System.IO
388+ ( stdout
389+ , hFlush
390+ , hSetBinaryMode
391+ , stdin
392+ , stdout
393+ , stderr
394+ , Handle
395+ , hClose
396+ )
397+import qualified System.Posix.Process as Posix (forkProcess, createSession)
398+import Data.Maybe (listToMaybe)
440399import Data.Binary (Binary(get, put), encode, decode)
441400import Data.Digest.Pure.MD5 (md5, MD5Digest)
442401import qualified Data.ByteString as BSS
@@ -444,6 +403,7 @@ import qualified Data.ByteString as BSS
444403 , length
445404 , concat
446405 , hPut
406+ , hGet
447407 )
448408import qualified Data.ByteString.Char8 as BSSC (pack)
449409import qualified Data.ByteString.Lazy as BSL
@@ -453,14 +413,23 @@ import qualified Data.ByteString.Lazy as BSL
453413 , fromChunks
454414 , toChunks
455415 , hPut
416+ , hGet
456417 )
457418import qualified Data.ByteString.Lazy.Char8 as BSLC (unpack)
458419import Data.Typeable (Typeable)
420+import Data.Foldable (forM_)
459421import Control.Applicative ((<$>), (<*>))
460422import Control.Monad (void, unless)
461423import Control.Monad.Reader (MonadReader, ReaderT, runReaderT, ask)
462−import Control.Exception (Exception, catches, Handler(Handler), throwIO)
424+import Control.Exception
425+ ( Exception
426+ , catches
427+ , Handler(Handler)
428+ , throwIO
429+ , SomeException
430+ )
463431import Control.Monad.IO.Class (MonadIO, liftIO)
432+import Control.Concurrent.MVar (MVar, newEmptyMVar, readMVar, putMVar)
464433
465434-- Azure
466435import Network.Azure.ServiceManagement
@@ -499,12 +468,30 @@ import qualified Network.SSH.Client.LibSSH2.Errors as SSH
499468 )
500469
501470-- CH
502−import Control.Distributed.Process (Process, Closure)
471+import Control.Distributed.Process
472+ ( Process
473+ , Closure
474+ , RemoteTable
475+ , catch
476+ , unClosure
477+ )
503478import Control.Distributed.Process.Serializable (Serializable)
504−import Control.Distributed.Process.Internal.Types
505− ( messageToPayload
479+import qualified Control.Distributed.Process.Internal.Types as CH
480+ ( LocalNode
481+ , LocalProcess(processQueue)
482+ , Message
483+ , payloadToMessage
484+ , messageToPayload
506485 , createMessage
507486 )
487+import Control.Distributed.Process.Node
488+ ( runProcess
489+ , forkProcess
490+ , newLocalNode
491+ , initRemoteTable
492+ )
493+import Control.Distributed.Process.Internal.CQueue (CQueue, enqueue)
494+import Network.Transport.TCP (createTransport, defaultTCPParameters)
508495import Network.Transport.Internal (encodeInt32, decodeInt32, prependLength)
509496
510497-- | Azure backend
@@ -609,10 +596,11 @@ apiCallOnVM :: AzureParameters
609596apiCallOnVM params cloudService vm port ppair =
610597 withSSH2 params vm $ \s -> do
611598 let exe = "PATH=. " ++ azureSshRemotePath params
612− ++ " onvm run "
613− ++ " --host " ++ vmIpAddress vm
614− ++ " --port " ++ port
615− ++ " --cloud-service " ++ cloudService
599+ ++ " onvm"
600+ ++ " " ++ vmIpAddress vm
601+ ++ " " ++ port
602+ ++ " " ++ cloudService
603+ ++ " False"
616604 ++ " 2>&1"
617605 let paramsEnc = encode params
618606 (status, r) <- SSH.withChannelBy (SSH.openChannelSession s) id $ \ch -> do
@@ -639,11 +627,11 @@ apiSpawnOnVM params cloudService vm port proc =
639627 withSSH2 params vm $ \s -> do
640628 -- TODO: reduce duplication with apiCallOnVM
641629 let exe = "PATH=. " ++ azureSshRemotePath params
642− ++ " onvm run "
643− ++ " --host " ++ vmIpAddress vm
644− ++ " --port " ++ port
645− ++ " --cloud-service " ++ cloudService
646− ++ " --background "
630+ ++ " onvm"
631+ ++ " " ++ vmIpAddress vm
632+ ++ " " ++ port
633+ ++ " " ++ cloudService
634+ ++ " True"
647635 ++ " 2>&1"
648636 let paramsEnc = encode params
649637 (status, r) <- SSH.withChannelBy (SSH.openChannelSession s) id $ \ch -> do
@@ -702,6 +690,15 @@ catchSshError s io =
702690localHash :: AzureParameters -> IO MD5Digest
703691localHash params = md5 <$> BSL.readFile (azureSshLocalPath params)
704692
693+--------------------------------------------------------------------------------
694+-- Utilities --
695+--------------------------------------------------------------------------------
696+
697+-- | Find a virtual machine with a particular name
698+findNamedVM :: Backend -> String -> IO (Maybe VirtualMachine)
699+findNamedVM backend vm =
700+ listToMaybe . filter ((== vm) . vmName) <$> findVMs backend
701+
705702--------------------------------------------------------------------------------
706703-- Local and remote processes --
707704--------------------------------------------------------------------------------
@@ -735,8 +732,8 @@ localSend x = LocalProcess $ do
735732 ch <- ask
736733 liftIO $ mapM_ (SSH.writeChannel ch)
737734 . prependLength
738− . messageToPayload
739− . createMessage
735+ . CH.messageToPayload
736+ . CH.createMessage
740737 $ x
741738
742739-- | Wait for a message from the remote process (see 'ProcessPair').
@@ -773,6 +770,80 @@ remoteSend' flags x = do
773770 BSL.hPut stdout enc
774771 hFlush stdout
775772
773+--------------------------------------------------------------------------------
774+-- On-VM main --
775+--------------------------------------------------------------------------------
776+
777+-- | Program main when run on the VM. A typical 'main' function looks like
778+--
779+-- > main :: IO ()
780+-- > main = do
781+-- > args <- getArgs
782+-- > case args of
783+-- > "onvm":args' -> onVmMain __remoteTable args'
784+-- > _ -> -- your normal main
785+onVmMain :: (RemoteTable -> RemoteTable) -> [String] -> IO ()
786+onVmMain rtable [host, port, cloudService, bg] = do
787+ hSetBinaryMode stdin True
788+ hSetBinaryMode stdout True
789+ Just procEnc <- getWithLength stdin
790+ Just paramsEnc <- getWithLength stdin
791+ backend <- initializeBackend (decode paramsEnc) cloudService
792+ let proc = decode procEnc
793+ lprocMVar <- newEmptyMVar :: IO (MVar CH.LocalProcess)
794+ if read bg
795+ then detach $ startCH proc lprocMVar backend runProcess (\_ -> return ())
796+ else do
797+ startCH proc lprocMVar backend forkProcess (liftIO . remoteThrow)
798+ lproc <- readMVar lprocMVar
799+ queueFromHandle stdin (CH.processQueue lproc)
800+ where
801+ startCH :: RemoteProcess ()
802+ -> MVar CH.LocalProcess
803+ -> Backend
804+ -> (CH.LocalNode -> Process () -> IO a)
805+ -> (SomeException -> Process ())
806+ -> IO ()
807+ startCH rproc lprocMVar backend go exceptionHandler = do
808+ mTransport <- createTransport host port defaultTCPParameters
809+ case mTransport of
810+ Left err -> remoteThrow err
811+ Right transport -> do
812+ node <- newLocalNode transport (rtable initRemoteTable)
813+ void . go node $ do
814+ ask >>= liftIO . putMVar lprocMVar
815+ proc <- unClosure rproc :: Process (Backend -> Process ())
816+ catch (proc backend) exceptionHandler
817+onVmMain _ _
818+ = error "Invalid arguments passed on onVmMain"
819+
820+-- | Read a 4-byte length @l@ and then an @l@-byte payload
821+--
822+-- Returns Nothing on EOF
823+getWithLength :: Handle -> IO (Maybe BSL.ByteString)
824+getWithLength h = do
825+ lenEnc <- BSS.hGet h 4
826+ if BSS.length lenEnc < 4
827+ then return Nothing
828+ else do
829+ let len = decodeInt32 lenEnc
830+ bs <- BSL.hGet h len
831+ if BSL.length bs < fromIntegral len
832+ then return Nothing
833+ else return (Just bs)
834+
835+queueFromHandle :: Handle -> CQueue CH.Message -> IO ()
836+queueFromHandle h q = do
837+ mPayload <- getWithLength stdin
838+ forM_ mPayload $ \payload -> do
839+ enqueue q $ CH.payloadToMessage (BSL.toChunks payload)
840+ queueFromHandle h q
841+
842+detach :: IO () -> IO ()
843+detach io = do
844+ mapM_ hClose [stdin, stdout, stderr]
845+ void . Posix.forkProcess $ void Posix.createSession >> io
846+
776847--------------------------------------------------------------------------------
777848-- SSH utilities --
778849--------------------------------------------------------------------------------