diff --git a/distributed-process.cabal b/distributed-process.cabal --- a/distributed-process.cabal +++ b/distributed-process.cabal @@ -1,5 +1,5 @@ Name: distributed-process -Version: 0.2.1.4 +Version: 0.2.2.0 Cabal-Version: >=1.8 Build-Type: Simple License: BSD3 @@ -123,6 +123,7 @@ random >= 1.0 && < 1.1, ghc-prim >= 0.2 && < 0.3, ansi-terminal >= 0.5 && < 0.6 + Other-modules: TestAuxiliary Extensions: RankNTypes, ScopedTypeVariables, FlexibleInstances, diff --git a/src/Control/Distributed/Process.hs b/src/Control/Distributed/Process.hs --- a/src/Control/Distributed/Process.hs +++ b/src/Control/Distributed/Process.hs @@ -80,8 +80,14 @@ , whereisRemoteAsync , nsendRemote , WhereIsReply(..) - -- * Auxiliary API + -- * Exception handling , catch + , mask + , onException + , bracket + , bracket_ + , finally + -- * Auxiliary API , expectTimeout , spawnAsync , spawnSupervised @@ -190,8 +196,14 @@ , nsendRemote -- Closures , unClosure - -- Auxiliary API + -- Exception handling , catch + , mask + , onException + , bracket + , bracket_ + , finally + -- Auxiliary API , expectTimeout , spawnAsync ) diff --git a/src/Control/Distributed/Process/Internal/Primitives.hs b/src/Control/Distributed/Process/Internal/Primitives.hs --- a/src/Control/Distributed/Process/Internal/Primitives.hs +++ b/src/Control/Distributed/Process/Internal/Primitives.hs @@ -43,8 +43,14 @@ , nsendRemote -- * Closures , unClosure - -- * Auxiliary API + -- * Exception handling , catch + , mask + , onException + , bracket + , bracket_ + , finally + -- * Auxiliary API , expectTimeout , spawnAsync , linkNode @@ -67,8 +73,8 @@ import Control.Monad.Reader (ask) import Control.Monad.IO.Class (MonadIO, liftIO) import Control.Applicative ((<$>)) -import Control.Exception (Exception, throw) -import qualified Control.Exception as Ex (catch) +import Control.Exception (Exception, throw, throwIO, SomeException) +import qualified Control.Exception as Ex (catch, mask) import Control.Concurrent.MVar (modifyMVar) import Control.Concurrent.Chan (writeChan) import Control.Concurrent.STM @@ -258,7 +264,7 @@ -- | Terminate (throws a ProcessTerminationException) terminate :: Process a -terminate = liftIO $ throw ProcessTerminationException +terminate = liftIO $ throwIO ProcessTerminationException -- | Our own process ID getSelfPid :: Process ProcessId @@ -317,15 +323,51 @@ ] -------------------------------------------------------------------------------- --- Auxiliary API -- +-- Exception handling -- -------------------------------------------------------------------------------- --- | Catch exceptions within a process +-- | Lift 'Control.Exception.catch' catch :: Exception e => Process a -> (e -> Process a) -> Process a catch p h = do lproc <- ask liftIO $ Ex.catch (runLocalProcess lproc p) (runLocalProcess lproc . h) +-- | Lift 'Control.Exception.mask' +mask :: ((forall a. Process a -> Process a) -> Process b) -> Process b +mask p = do + lproc <- ask + liftIO $ Ex.mask $ \restore -> + runLocalProcess lproc (p (liftRestore lproc restore)) + where + liftRestore :: LocalProcess -> (forall a. IO a -> IO a) -> (forall a. Process a -> Process a) + liftRestore lproc restoreIO = liftIO . restoreIO . runLocalProcess lproc + +-- | Lift 'Control.Exception.onException' +onException :: Process a -> Process b -> Process a +onException p what = p `catch` \e -> do _ <- what + liftIO $ throwIO (e :: SomeException) + +-- | Lift 'Control.Exception.bracket' +bracket :: Process a -> (a -> Process b) -> (a -> Process c) -> Process c +bracket before after thing = do + mask $ \restore -> do + a <- before + r <- restore (thing a) `onException` after a + _ <- after a + return r + +-- | Lift 'Control.Exception.bracket_' +bracket_ :: Process a -> Process b -> Process c -> Process c +bracket_ before after thing = bracket before (const after) (const thing) + +-- | Lift 'Control.Exception.finally' +finally :: Process a -> Process b -> Process a +finally a sequel = bracket_ (return ()) sequel a + +-------------------------------------------------------------------------------- +-- Auxiliary API -- +-------------------------------------------------------------------------------- + -- | Like 'expect' but with a timeout expectTimeout :: forall a. Serializable a => Int -> Process (Maybe a) expectTimeout timeout = receiveTimeout timeout [match return] @@ -464,7 +506,7 @@ unClosure (Closure (Static label) env) = do rtable <- remoteTable . processNode <$> ask case resolveClosure rtable label env of - Nothing -> throw . userError $ "Unregistered closure " ++ show label + Nothing -> error $ "Unregistered closure " ++ show label Just dyn -> return $ fromDyn dyn (throw (typeError dyn)) where typeError dyn = userError $ "lookupStatic type error: " diff --git a/src/Control/Distributed/Process/Node.hs b/src/Control/Distributed/Process/Node.hs --- a/src/Control/Distributed/Process/Node.hs +++ b/src/Control/Distributed/Process/Node.hs @@ -121,7 +121,7 @@ , sendMessage , sendPayload ) -import Control.Distributed.Process.Internal.Primitives (expect, register) +import Control.Distributed.Process.Internal.Primitives (expect, register, finally) import qualified Control.Distributed.Process.Internal.Closure.Static as Static (__remoteTable) import qualified Control.Distributed.Process.Internal.Closure.CP as CP (__remoteTable) @@ -189,7 +189,7 @@ runProcess :: LocalNode -> Process () -> IO () runProcess node proc = do done <- newEmptyMVar - void $ forkProcess node (proc >> liftIO (putMVar done ())) + void $ forkProcess node (proc `finally` liftIO (putMVar done ())) takeMVar done -- | Spawn a new process on a local node diff --git a/tests/TestAuxiliary.hs b/tests/TestAuxiliary.hs new file mode 100644 --- /dev/null +++ b/tests/TestAuxiliary.hs @@ -0,0 +1,75 @@ +{-# OPTIONS_GHC -fno-warn-orphans #-} +module TestAuxiliary ( -- Running tests + runTest + , runTests + -- Writing tests + , forkTry + , trySome + , randomThreadDelay + ) where + + +#if ! MIN_VERSION_base(4,6,0) +import Prelude hiding (catch) +#endif + +import Control.Concurrent (myThreadId, forkIO, ThreadId, throwTo, threadDelay) +import Control.Monad (liftM2, unless) +import Control.Exception (SomeException, try, catch) +import System.Timeout (timeout) +import System.IO (stdout, hFlush) +import System.Console.ANSI ( SGR(SetColor, Reset) + , Color(Red, Green) + , ConsoleLayer(Foreground) + , ColorIntensity(Vivid) + , setSGR + ) +import System.Random (randomIO) + +-- | Like fork, but throw exceptions in the child thread to the parent +forkTry :: IO () -> IO ThreadId +forkTry p = do + tid <- myThreadId + forkIO $ catch p (\e -> throwTo tid (e :: SomeException)) + +-- | Like try, but specialized to SomeException +trySome :: IO a -> IO (Either SomeException a) +trySome = try + +-- | Run the given test, catching timeouts and exceptions +runTest :: String -> IO () -> IO Bool +runTest description test = do + putStr $ "Running " ++ show description ++ ": " + hFlush stdout + done <- try . timeout 60000000 $ test -- 60 seconds + case done of + Left err -> failed $ "(exception: " ++ show (err :: SomeException) ++ ")" + Right Nothing -> failed "(timeout)" + Right (Just ()) -> ok + where + failed :: String -> IO Bool + failed err = do + setSGR [SetColor Foreground Vivid Red] + putStr "failed " + setSGR [Reset] + putStrLn err + return False + + ok :: IO Bool + ok = do + setSGR [SetColor Foreground Vivid Green] + putStrLn "ok" + setSGR [Reset] + return True + +-- | Run a bunch of tests and throw an exception if any fails +runTests :: [(String, IO ())] -> IO () +runTests tests = do + success <- foldr (liftM2 (&&) . uncurry runTest) (return True) tests + unless success $ fail "Some tests failed" + +-- | Random thread delay between 0 and the specified max +randomThreadDelay :: Int -> IO () +randomThreadDelay maxDelay = do + delay <- randomIO :: IO Int + threadDelay (delay `mod` maxDelay)