7a81bd5 · distributed-process

git.iko.soy/distributed-process

on master help sign in

Have spawnOnVM return a PID

7a81bd5Edsko de Vries14 years ago

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

compare with master

unifiedsplit

@@ -76,7 +76,10 @@ main = do

76 -- The same binary can behave as the client or the server,

77 -- depending on the command line arguments

78 case cmd of

79+ "server" -> do

79− "server" -> spawnOnVM backend vm port ($(mkClosure 'pingServer) ())

80+ pid <- spawnOnVM backend vm port ($(mkClosure 'pingServer) ())

81+ putStrLn $ "Ping server started at " ++ show pid

82+ "client" ->

80− "client" -> callOnVM backend vm port $

83+ callOnVM backend vm port $

81− ProcessPair ($(mkClosure 'pingClientRemote) ())

84+ ProcessPair ($(mkClosure 'pingClientRemote) ())

82− pingClientLocal

85+ pingClientLocal

distributed-process-azure/src/Control/Distributed/Process/Backend/Azure.hs+24−13

@@ -455,7 +455,7 @@ import qualified Data.ByteString.Lazy.Char8 as BSLC (unpack)

455import Data.Typeable (Typeable)

456import Data.Foldable (forM_)

457import Control.Applicative ((<$>), (<*>))

458−import Control.Monad (void, unless)

458+import Control.Monad (void)

459import Control.Monad.Reader (MonadReader, ReaderT, runReaderT, ask)

460import Control.Exception

461 ( Exception

@@ -510,6 +510,8 @@ import Control.Distributed.Process

510 , RemoteTable

511 , catch

512 , unClosure

513+ , ProcessId

514+ , getSelfPid

513 )

515 )

514import Control.Distributed.Process.Serializable (Serializable)

516import Control.Distributed.Process.Serializable (Serializable)

515import qualified Control.Distributed.Process.Internal.Types as CH

517import qualified Control.Distributed.Process.Internal.Types as CH

@@ -546,7 +548,7 @@ data Backend = Backend {

546 -- | Create a new CH node and run the specified process.

548 -- | Create a new CH node and run the specified process.

547 -- The CH node will shut down when the /remote/ process exists. @spawnOnVM@

549 -- The CH node will shut down when the /remote/ process exists. @spawnOnVM@

548 -- returns as soon as the process has been spawned.

550 -- returns as soon as the process has been spawned.

549− , spawnOnVM :: VirtualMachine -> String -> RemoteProcess () -> IO ()

551+ , spawnOnVM :: VirtualMachine -> String -> RemoteProcess () -> IO ProcessId

550 } deriving (Typeable)

552 } deriving (Typeable)

551

553

552-- | Azure connection parameters

554-- | Azure connection parameters

@@ -658,7 +660,7 @@ apiSpawnOnVM :: AzureParameters

658 -> VirtualMachine

660 -> VirtualMachine

659 -> String

661 -> String

660 -> Closure (Backend -> Process ())

662 -> Closure (Backend -> Process ())

661− -> IO ()

663+ -> IO ProcessId

662apiSpawnOnVM params cloudService vm port proc =

664apiSpawnOnVM params cloudService vm port proc =

663 withSSH2 params vm $ \s -> do

665 withSSH2 params vm $ \s -> do

664 -- TODO: reduce duplication with apiCallOnVM

666 -- TODO: reduce duplication with apiCallOnVM

@@ -678,7 +680,9 @@ apiSpawnOnVM params cloudService vm port proc =

678 SSH.writeAllChannel ch paramsEnc

680 SSH.writeAllChannel ch paramsEnc

679 SSH.channelSendEOF ch

681 SSH.channelSendEOF ch

680 SSH.readAllChannel ch

682 SSH.readAllChannel ch

683+ if status == 0

684+ then return (decode r)

681− unless (status == 0) $ error (BSLC.unpack r)

685+ else error (BSLC.unpack r)

682 where

686 where

683 procEnc :: BSL.ByteString

687 procEnc :: BSL.ByteString

684 procEnc = encode proc

688 procEnc = encode proc

@@ -822,15 +826,27 @@ onVmMain :: (RemoteTable -> RemoteTable) -> [String] -> IO ()

822onVmMain rtable [host, port, cloudService, bg] = do

826onVmMain rtable [host, port, cloudService, bg] = do

823 hSetBinaryMode stdin True

827 hSetBinaryMode stdin True

824 hSetBinaryMode stdout True

828 hSetBinaryMode stdout True

825− Just procEnc <- getWithLength stdin

829+ Just rprocEnc <- getWithLength stdin

826 Just paramsEnc <- getWithLength stdin

830 Just paramsEnc <- getWithLength stdin

827 backend <- initializeBackend (decode paramsEnc) cloudService

831 backend <- initializeBackend (decode paramsEnc) cloudService

828− let proc = decode procEnc

832+ let rproc = decode rprocEnc

829 lprocMVar <- newEmptyMVar :: IO (MVar CH.LocalProcess)

833 lprocMVar <- newEmptyMVar :: IO (MVar CH.LocalProcess)

830 if read bg

834 if read bg

835+ then

836+ void . Posix.forkProcess $ do

837+ -- We inherit the file descriptors from the parent, so the SSH

838+ -- session will not be terminated until we close them

839+ void Posix.createSession

831− then detach $ startCH proc lprocMVar backend runProcess (\_ -> return ())

840+ startCH rproc lprocMVar backend

841+ (\node proc -> runProcess node $ do

842+ us <- getSelfPid

843+ liftIO $ do

844+ BSL.hPut stdout (encode us)

845+ mapM_ hClose [stdin, stdout, stderr]

846+ proc)

847+ (\_ -> return ())

832 else do

848 else do

833− startCH proc lprocMVar backend forkProcess (liftIO . remoteThrow)

849+ startCH rproc lprocMVar backend forkProcess (liftIO . remoteThrow)

834 lproc <- readMVar lprocMVar

850 lproc <- readMVar lprocMVar

835 queueFromHandle stdin (CH.processQueue lproc)

851 queueFromHandle stdin (CH.processQueue lproc)

836 where

852 where

@@ -875,11 +891,6 @@ queueFromHandle h q = do

875 enqueue q $ CH.payloadToMessage (BSL.toChunks payload)

891 enqueue q $ CH.payloadToMessage (BSL.toChunks payload)

876 queueFromHandle h q

892 queueFromHandle h q

877

893

878−detach :: IO () -> IO ()

879−detach io = do

880− mapM_ hClose [stdin, stdout, stderr]

881− void . Posix.forkProcess $ void Posix.createSession >> io

882−

883--------------------------------------------------------------------------------

894--------------------------------------------------------------------------------

884-- SSH utilities --

895-- SSH utilities --

885--------------------------------------------------------------------------------

896--------------------------------------------------------------------------------