git.iko.soy/distributed-process
on master help sign in
Deal with remote exceptions in callOnVM
commit43721cac12abf1a50a8b752bd4d8d9765f1b00ed authorEdsko de Vries <edsko@well-typed.com> dateWed 8 Aug 2012, 10:22 (14 years ago) parent770b18b
2 files changed+43−26
compare with master
unifiedsplit
distributed-process-azure/src/Control/Distributed/Process/Backend/Azure.hs+21−4 distributed-process-azure/src/Control/Distributed/Process/Backend/Azure/GenericMain.hs+22−22
distributed-process-azure/src/Control/Distributed/Process/Backend/Azure.hs+21−4
@@ -14,6 +14,8 @@ module Control.Distributed.Process.Backend.Azure
1414 , LocalProcess
1515 , localExpect
1616 , remoteSend
17+ , remoteThrow
18+ , remoteSend'
1719 , localSend
1820 ) where
1921
@@ -43,7 +45,7 @@ import Data.Typeable (Typeable)
4345import Control.Applicative ((<$>))
4446import Control.Monad (void, unless)
4547import Control.Monad.Reader (MonadReader, ReaderT, runReaderT, ask)
46−import Control.Exception (catches, Handler(Handler))
48+import Control.Exception (Exception, catches, Handler(Handler))
4749import Control.Monad.IO.Class (MonadIO, liftIO)
4850
4951-- Azure
@@ -285,8 +287,12 @@ localExpect :: Serializable a => LocalProcess a
285287localExpect = LocalProcess $ do
286288 ch <- ask
287289 liftIO $ do
288− len <- decodeInt32 . BSS.concat . BSL.toChunks <$> readSizeChannel ch 4
289− decode <$> readSizeChannel ch len
290+ isE <- readIntChannel ch
291+ len <- readIntChannel ch
292+ msg <- readSizeChannel ch len
293+ if isE /= 0
294+ then error (decode msg)
295+ else return (decode msg)
290296
291297localSend :: Serializable a => a -> LocalProcess ()
292298localSend x = LocalProcess $ do
@@ -298,8 +304,15 @@ localSend x = LocalProcess $ do
298304 $ x
299305
300306remoteSend :: Serializable a => a -> Process ()
301−remoteSend x = liftIO $ do
307+remoteSend = liftIO . remoteSend' 0
308+
309+remoteThrow :: Exception e => e -> Process ()
310+remoteThrow = liftIO . remoteSend' 1 . show
311+
312+remoteSend' :: Serializable a => Int -> a -> IO ()
313+remoteSend' flags x = do
302314 let enc = encode x
315+ BSS.hPut stdout (encodeInt32 flags)
303316 BSS.hPut stdout (encodeInt32 (BSL.length enc))
304317 BSL.hPut stdout enc
305318 hFlush stdout
@@ -316,3 +329,7 @@ readSizeChannel ch = go []
316329 go acc size = do
317330 bs <- SSH.readChannel ch (fromIntegral (0x400 `min` size))
318331 go (bs : acc) (size - BSS.length bs)
332+
333+readIntChannel :: SSH.Channel -> IO Int
334+readIntChannel ch =
335+ decodeInt32 . BSS.concat . BSL.toChunks <$> readSizeChannel ch 4
distributed-process-azure/src/Control/Distributed/Process/Backend/Azure/GenericMain.hs+22−22
@@ -21,7 +21,7 @@ import qualified Data.ByteString.Lazy as BSL (ByteString, hGet, toChunks, length
2121import qualified Data.ByteString as BSS (hGet, length)
2222import Control.Monad (unless, forM, void)
2323import Control.Monad.Reader (ask)
24−import Control.Exception (throwIO, SomeException)
24+import Control.Exception (SomeException)
2525import Control.Applicative ((<$>), (<*>), optional)
2626import Control.Monad.IO.Class (liftIO)
2727import Control.Concurrent.MVar (MVar, newEmptyMVar, putMVar, readMVar)
@@ -73,6 +73,9 @@ import Control.Distributed.Process.Internal.Types
7373 , payloadToMessage
7474 , Message
7575 )
76+import Control.Distributed.Process.Internal.CQueue (CQueue, enqueue)
77+
78+-- Azure
7679import Control.Distributed.Process.Backend.Azure
7780 ( AzureParameters(azureSshUserName, azureSetup)
7881 , defaultAzureParameters
@@ -82,11 +85,11 @@ import Control.Distributed.Process.Backend.Azure
8285 , Backend(findVMs, copyToVM, checkMD5, callOnVM, spawnOnVM)
8386 , ProcessPair(..)
8487 , RemoteProcess
88+ , remoteSend'
89+ , remoteThrow
8590 )
86−import Control.Distributed.Process.Internal.CQueue (CQueue, enqueue)
8791
8892-- Transport
89−import Network.Transport (Transport)
9093import Network.Transport.Internal (decodeInt32)
9194import Network.Transport.TCP (createTransport, defaultTCPParameters)
9295
@@ -185,31 +188,28 @@ onVmRun rtable host port bg = do
185188 let proc = decode procEnc
186189 lprocMVar <- newEmptyMVar :: IO (MVar LocalProcess)
187190 if bg
188− then detach $ startCH proc lprocMVar runProcess
191+ then detach $ startCH proc lprocMVar runProcess (\_ -> return ())
189192 else do
190− _pid <- startCH proc lprocMVar forkProcess
193+ startCH proc lprocMVar forkProcess remoteThrow
191194 lproc <- readMVar lprocMVar
192195 queueFromHandle stdin (processQueue lproc)
193196 where
194− startCH :: RemoteProcess () -> MVar LocalProcess -> (LocalNode -> Process () -> IO a) -> IO a
195− startCH rproc lprocMVar go = do
196− transport <-newTransport
197− node <- newLocalNode transport rtable
198− go node $ do
199− ask >>= liftIO . putMVar lprocMVar
200− let backend = error "TODO: backend not initialized in onVmRun"
201− proc <- unClosure rproc :: Process (Backend -> Process ())
202− catch (proc backend) exceptionHandler
203−
204− newTransport :: IO Transport
205− newTransport = do
197+ startCH :: RemoteProcess ()
198+ -> MVar LocalProcess
199+ -> (LocalNode -> Process () -> IO a)
200+ -> (SomeException -> Process ())
201+ -> IO ()
202+ startCH rproc lprocMVar go exceptionHandler = do
206203 mTransport <- createTransport host port defaultTCPParameters
207204 case mTransport of
208− Left err -> throwIO err
209− Right transport -> return transport
210−
211− exceptionHandler :: SomeException -> Process ()
212− exceptionHandler e = liftIO $ appendFile "error.log" (show e)
205+ Left err -> remoteSend' 1 (show err)
206+ Right transport -> do
207+ node <- newLocalNode transport rtable
208+ void . go node $ do
209+ ask >>= liftIO . putMVar lprocMVar
210+ let backend = error "TODO: backend not initialized in onVmRun"
211+ proc <- unClosure rproc :: Process (Backend -> Process ())
212+ catch (proc backend) exceptionHandler
213213
214214-- | Read a 4-byte length @l@ and then an @l@-byte payload
215215--