distributed-process 0.6.1 → 0.6.2
raw patch · 13 files changed
+121/−57 lines, 13 filesdep −ghc-primdep ~basedep ~binarydep ~transformers
Dependencies removed: ghc-prim
Dependency ranges changed: base, binary, transformers
Files
- ChangeLog +10/−0
- distributed-process.cabal +8/−9
- src/Control/Distributed/Process.hs +27/−4
- src/Control/Distributed/Process/Debug.hs +3/−7
- src/Control/Distributed/Process/Internal/CQueue.hs +7/−2
- src/Control/Distributed/Process/Internal/Closure/TH.hs +4/−0
- src/Control/Distributed/Process/Internal/Primitives.hs +1/−1
- src/Control/Distributed/Process/Internal/StrictMVar.hs +8/−4
- src/Control/Distributed/Process/Internal/WeakTQueue.hs +7/−2
- src/Control/Distributed/Process/Management.hs +3/−2
- src/Control/Distributed/Process/Management/Internal/Agent.hs +2/−1
- src/Control/Distributed/Process/Management/Internal/Trace/Primitives.hs +2/−1
- src/Control/Distributed/Process/Node.hs +39/−24
ChangeLog view
@@ -1,3 +1,13 @@+2016-06-09 Facundo Domínguez <facundo.dominguez@tweag.io> 0.6.2++* Provide compatibility with ghc-8.0.1+* Remove dependency on ghc-prim.+* Don't throw exceptions asynchronously.+* Bump upper bounds of dependencies.+* Fix exception handling in callLocal.+* Have spawnLocal inherit the masking state of the caller.+* Have nsend send unencoded messages to local processes.+ 2016-03-03 Facundo Domínguez <facundo.dominguez@tweag.io> 0.6.1 * Implement MonadCatch, MonadThrow, MonadMask for Process.
distributed-process.cabal view
@@ -1,5 +1,5 @@ Name: distributed-process-Version: 0.6.1+Version: 0.6.2 Cabal-Version: >=1.8 Build-Type: Simple License: BSD3@@ -42,16 +42,15 @@ Library Build-Depends: base >= 4.4 && < 5,- binary >= 0.6.3 && < 0.8,+ binary >= 0.6.3 && < 0.9, hashable >= 1.2.0.5 && < 1.3, network-transport >= 0.4.1.0 && < 0.5, stm >= 2.4 && < 2.5,- transformers >= 0.2 && < 0.5,+ transformers >= 0.2 && < 0.6, mtl >= 2.0 && < 2.4, data-accessor >= 0.2 && < 0.3, bytestring >= 0.9 && < 0.11, random >= 1.0 && < 1.2,- ghc-prim >= 0.2 && < 0.5, distributed-static >= 0.2 && < 0.4, rank1dynamic >= 0.1 && < 0.4, syb >= 0.3 && < 0.7,@@ -110,7 +109,7 @@ distributed-process, network-transport-tcp >= 0.3 && < 0.6, bytestring >= 0.9 && < 0.11,- binary >= 0.6.3 && < 0.8+ binary >= 0.6.3 && < 0.9 Main-Is: benchmarks/Throughput.hs ghc-options: -Wall @@ -120,7 +119,7 @@ distributed-process, network-transport-tcp >= 0.3 && < 0.6, bytestring >= 0.9 && < 0.11,- binary >= 0.6.3 && < 0.8+ binary >= 0.6.3 && < 0.9 Main-Is: benchmarks/Latency.hs ghc-options: -Wall @@ -130,7 +129,7 @@ distributed-process, network-transport-tcp >= 0.3 && < 0.6, bytestring >= 0.9 && < 0.11,- binary >= 0.6.3 && < 0.8+ binary >= 0.6.3 && < 0.9 Main-Is: benchmarks/Channels.hs ghc-options: -Wall @@ -140,7 +139,7 @@ distributed-process, network-transport-tcp >= 0.3 && < 0.6, bytestring >= 0.9 && < 0.11,- binary >= 0.6.3 && < 0.8+ binary >= 0.6.3 && < 0.9 Main-Is: benchmarks/Spawns.hs ghc-options: -Wall @@ -150,6 +149,6 @@ distributed-process, network-transport-tcp >= 0.3 && < 0.6, bytestring >= 0.9 && < 0.11,- binary >= 0.6.3 && < 0.8+ binary >= 0.6.3 && < 0.9 Main-Is: benchmarks/ProcessRing.hs ghc-options: -Wall -threaded -O2 -rtsopts
src/Control/Distributed/Process.hs view
@@ -162,6 +162,7 @@ import Control.Monad.IO.Class (liftIO) import Control.Applicative import Control.Monad.Reader (ask)+import Control.Concurrent (killThread) import Control.Concurrent.MVar ( MVar , newEmptyMVar@@ -196,6 +197,7 @@ , RegisterReply(..) , LocalProcess(processNode) , Message+ , localProcessWithId ) import Control.Distributed.Process.Serializable (Serializable) import Control.Distributed.Process.Internal.Primitives@@ -303,6 +305,10 @@ , reconnectPort ) import Control.Distributed.Process.Node (forkProcess)+import Control.Distributed.Process.Internal.Types+ ( processThread+ , withValidLocalState+ ) import Control.Distributed.Process.Internal.Spawn ( -- Spawning Processes/Channels spawn@@ -319,7 +325,12 @@ #else import Prelude hiding (catch) #endif+import Control.Distributed.Process.Internal.StrictMVar (readMVar)+import qualified Control.Exception as Exception (onException)+import Data.Accessor ((^.))+import Data.Foldable (forM_) + -- INTERNAL NOTES -- -- 1. 'send' never fails. If you want to know that the remote process received@@ -416,8 +427,20 @@ -- Silently dropping messages may not always be the best approach. callLocal :: Process a -> Process a callLocal proc = Catch.mask $ \release -> do- mv <- liftIO newEmptyMVar :: Process (MVar (Either Catch.SomeException a))+ mv <- liftIO newEmptyMVar :: Process (MVar (Either Catch.SomeException a)) child <- spawnLocal $ Catch.try (release proc) >>= liftIO . putMVar mv- rs <- liftIO (takeMVar mv) `Catch.onException`- (kill child "exception in parent process" >> liftIO (takeMVar mv))- either Catch.throwM return rs+ lproc <- ask+ liftIO $ do+ rs <- Exception.onException (takeMVar mv) $ Catch.uninterruptibleMask_ $+ -- Exceptions need to be prevented from interrupting the clean up or+ -- the original exception which caused entering the handler could be+ -- forgotten. For instance, this could have a problematic effect+ -- when the original exception was meant to kill the thread and the+ -- second exception doesn't (like the exception thrown by+ -- 'System.Timeout.timeout').+ do mchildThreadId <- withValidLocalState (processNode lproc) $+ \vst -> return $ fmap processThread $+ vst ^. localProcessWithId (processLocalId child)+ forM_ mchildThreadId killThread+ takeMVar mv+ either Catch.throwM return rs
src/Control/Distributed/Process/Debug.hs view
@@ -141,17 +141,14 @@ ) where -import Control.Applicative ((<$>))+import Control.Applicative import Control.Distributed.Process.Internal.Primitives ( proxy- , finally , die , whereis , send , receiveWait , matchIf- , finally- , try , monitor ) import Control.Distributed.Process.Internal.Types@@ -196,12 +193,11 @@ import Control.Monad.IO.Class (liftIO) import Control.Monad.Reader (ask)+import Control.Monad.Catch (finally, try) import Data.Binary() -#if ! MIN_VERSION_base(4,6,0)-import Prelude hiding (catch)-#endif+import Prelude -------------------------------------------------------------------------------- -- Debugging/Tracing API --
src/Control/Distributed/Process/Internal/CQueue.hs view
@@ -1,3 +1,4 @@+{-# LANGUAGE CPP #-} {-# LANGUAGE DeriveFunctor #-} {-# LANGUAGE BangPatterns #-} {-# LANGUAGE MagicHash, UnboxedTuples, PatternGuards, ScopedTypeVariables, RankNTypes #-}@@ -46,8 +47,8 @@ import Data.Maybe (fromJust) import Data.Traversable (traverse) import GHC.MVar (MVar(MVar))-import GHC.IO (IO(IO))-import GHC.Prim (mkWeak#)+import GHC.IO (IO(IO), unIO)+import GHC.Exts (mkWeak#) import GHC.Weak (Weak(Weak)) -- We use a TCHan rather than a Chan so that we have a non-blocking read@@ -265,7 +266,11 @@ -- | Weak reference to a CQueue mkWeakCQueue :: CQueue a -> IO () -> IO (Weak (CQueue a)) mkWeakCQueue m@(CQueue (StrictMVar (MVar m#)) _ _) f = IO $ \s ->+#if MIN_VERSION_base(4,9,0)+ case mkWeak# m# m (unIO f) s of (# s1, w #) -> (# s1, Weak w #)+#else case mkWeak# m# m f s of (# s1, w #) -> (# s1, Weak w #)+#endif queueSize :: CQueue a -> IO Int queueSize (CQueue _ _ size) = readTVarIO size
src/Control/Distributed/Process/Internal/Closure/TH.hs view
@@ -299,7 +299,11 @@ getType name = do info <- reify name case info of+#if MIN_VERSION_template_haskell(2,11,0)+ VarI origName typ _ -> return (origName, typ)+#else VarI origName typ _ _ -> return (origName, typ)+#endif _ -> fail $ show name ++ " not found" -- | Variation on 'funD' which takes a single expression to define the function
src/Control/Distributed/Process/Internal/Primitives.hs view
@@ -1166,7 +1166,7 @@ -- | Named send to a process in the local registry (asynchronous) nsend :: Serializable a => String -> a -> Process () nsend label msg =- sendCtrlMsg Nothing (NamedSend label (createMessage msg))+ sendCtrlMsg Nothing (NamedSend label (createUnencodedMessage msg)) -- | Named send to a process in the local registry (asynchronous). -- This function makes /no/ attempt to serialize and (in the case when the
src/Control/Distributed/Process/Internal/StrictMVar.hs view
@@ -36,8 +36,8 @@ #endif ) import GHC.MVar (MVar(MVar))-import GHC.IO (IO(IO))-import GHC.Prim (mkWeak#)+import GHC.IO (IO(IO), unIO)+import GHC.Exts (mkWeak#) import GHC.Weak (Weak(Weak)) newtype StrictMVar a = StrictMVar (MVar.MVar a)@@ -85,5 +85,9 @@ evaluateFst (x, y) = evaluate x >> return (x, y) mkWeakMVar :: StrictMVar a -> IO () -> IO (Weak (StrictMVar a))-mkWeakMVar m@(StrictMVar (MVar m#)) f = IO $ \s ->- case mkWeak# m# m f s of (# s1, w #) -> (# s1, Weak w #)+mkWeakMVar q@(StrictMVar (MVar m#)) f = IO $ \s ->+#if MIN_VERSION_base(4,9,0)+ case mkWeak# m# q (unIO f) s of (# s', w #) -> (# s', Weak w #)+#else+ case mkWeak# m# q f s of (# s', w #) -> (# s', Weak w #)+#endif
src/Control/Distributed/Process/Internal/WeakTQueue.hs view
@@ -1,3 +1,4 @@+{-# LANGUAGE CPP #-} {-# LANGUAGE DeriveDataTypeable #-} {-# LANGUAGE MagicHash, UnboxedTuples #-} -- | Clone of Control.Concurrent.STM.TQueue with support for mkWeakTQueue@@ -22,8 +23,8 @@ import Prelude hiding (read) import GHC.Conc import Data.Typeable (Typeable)-import GHC.IO (IO(IO))-import GHC.Prim (mkWeak#)+import GHC.IO (IO(IO), unIO)+import GHC.Exts (mkWeak#) import GHC.Weak (Weak(Weak)) --------------------------------------------------------------------------------@@ -99,4 +100,8 @@ mkWeakTQueue :: TQueue a -> IO () -> IO (Weak (TQueue a)) mkWeakTQueue q@(TQueue _read (TVar write#)) f = IO $ \s ->+#if MIN_VERSION_base(4,9,0)+ case mkWeak# write# q (unIO f) s of (# s', w #) -> (# s', Weak w #)+#else case mkWeak# write# q f s of (# s', w #) -> (# s', Weak w #)+#endif
src/Control/Distributed/Process/Management.hs view
@@ -273,7 +273,7 @@ , mxDropTable ) where -import Control.Applicative ((<$>))+import Control.Applicative import Control.Concurrent.STM (atomically) import Control.Concurrent.STM.TChan ( readTChan@@ -288,7 +288,6 @@ , matchAny , matchSTM , unwrapMessage- , onException , register , whereis , die@@ -317,12 +316,14 @@ import Control.Distributed.Process.Serializable (Serializable) import Control.Monad.IO.Class (liftIO) import Control.Monad.Reader (ask)+import Control.Monad.Catch (onException) import qualified Control.Monad.State as ST ( get , modify , lift , runStateT )+import Prelude -- | Publishes an arbitrary @Serializable@ message to the management event bus. -- Note that /no attempt is made to force the argument/, therefore it is very
src/Control/Distributed/Process/Management/Internal/Agent.hs view
@@ -3,7 +3,7 @@ module Control.Distributed.Process.Management.Internal.Agent where -import Control.Applicative ((<$>))+import Control.Applicative import Control.Concurrent (forkIO) import Control.Concurrent.MVar (MVar, newEmptyMVar, putMVar, takeMVar) import Control.Concurrent.STM (atomically)@@ -44,6 +44,7 @@ import Control.Monad.IO.Class (liftIO) import Control.Monad.Reader (ask) import GHC.Weak (Weak, deRefWeak)+import Prelude -------------------------------------------------------------------------------- -- Agent Controller Implementation --
src/Control/Distributed/Process/Management/Internal/Trace/Primitives.hs view
@@ -28,7 +28,7 @@ , withRegisteredTracer ) where -import Control.Applicative ((<$>))+import Control.Applicative import Control.Distributed.Process.Internal.Primitives ( whereis , newChan@@ -67,6 +67,7 @@ import Control.Monad.Reader (ask) import qualified Data.Set as Set (fromList)+import Prelude -------------------------------------------------------------------------------- -- Main API --
src/Control/Distributed/Process/Node.hs view
@@ -5,6 +5,7 @@ {-# LANGUAGE BangPatterns #-} {-# LANGUAGE GeneralizedNewtypeDeriving #-} {-# LANGUAGE RecordWildCards #-}+{-# LANGUAGE MagicHash #-} -- | Local nodes --@@ -20,10 +21,6 @@ -- TODO: Calls to 'sendBinary' and co (by the NC) may stall the node controller. -#if ! MIN_VERSION_base(4,6,0)-import Prelude hiding (catch)-#endif- import System.IO (fixIO, hPutStrLn, stderr) import System.Mem.Weak (Weak, deRefWeak) import qualified Data.ByteString.Lazy as BSL (fromChunks)@@ -52,7 +49,7 @@ import Data.Maybe (isJust, fromJust, isNothing, catMaybes) import Data.Typeable (Typeable) import Control.Category ((>>>))-import Control.Applicative (Applicative, (<$>))+import Control.Applicative import Control.Monad (void, when, join) import Control.Monad.IO.Class (MonadIO, liftIO) import Control.Monad.State.Strict (MonadState, StateT, evalStateT, gets)@@ -60,11 +57,12 @@ import Control.Monad.Reader (MonadReader, ReaderT, runReaderT, ask) import Control.Exception ( throwIO- , AsyncException(ThreadKilled) , SomeException , Exception , throwTo , uninterruptibleMask_+ , getMaskingState+ , MaskingState(..) ) import qualified Control.Exception as Exception ( Handler(..)@@ -72,12 +70,11 @@ , catches , finally )-import Control.Concurrent (forkIO, forkIOWithUnmask, killThread, myThreadId)+import Control.Concurrent (forkIO, forkIOWithUnmask, killThread) import Control.Distributed.Process.Internal.StrictMVar ( newMVar , withMVar , modifyMVarMasked- , modifyMVar_ , modifyMVar , newEmptyMVar , putMVar@@ -167,12 +164,11 @@ , RegisterReply(..) , WhereIsReply(..) , payloadToMessage- , messageToPayload , createUnencodedMessage , unsafeCreateUnencodedMessage , runLocalProcess , firstNonReservedProcessId- , ImplicitReconnect(WithImplicitReconnect,NoImplicitReconnect)+ , ImplicitReconnect(WithImplicitReconnect) ) import Control.Distributed.Process.Management.Internal.Agent ( mxAgentController@@ -199,7 +195,6 @@ import Control.Distributed.Process.Serializable (Serializable) import Control.Distributed.Process.Internal.Messaging ( sendBinary- , sendPayload , closeImplicitReconnections , impliesDeathOf )@@ -208,7 +203,6 @@ , receiveWait , match , sendChan- , try , unwrapMessage ) import Control.Distributed.Process.Internal.Types (SendPort, Tracer(..))@@ -218,8 +212,21 @@ ( mapMaybe , mapDefault )+import Control.Monad.Catch (try)+import GHC.IO (IO(..), unsafeUnmask)+import GHC.Base ( maskAsyncExceptions# )+ import Unsafe.Coerce+import Prelude +-- Remove these definitions when the fix for+-- https://ghc.haskell.org/trac/ghc/ticket/10149+-- is included in all supported compilers:+block :: IO a -> IO a+block (IO io) = IO $ maskAsyncExceptions# io+unblock :: IO a -> IO a+unblock = unsafeUnmask+ -------------------------------------------------------------------------------- -- Initialization -- --------------------------------------------------------------------------------@@ -314,8 +321,6 @@ -- process which uses 'send' or other primitives which are traced. register "trace.logger" logger where- fork = forkProcess node- loop = do receiveWait [ match $ \((time, pid, string) ::(String, ProcessId, String)) -> do@@ -361,11 +366,14 @@ -- | Spawn a new process on a local node forkProcess :: LocalNode -> Process () -> IO ProcessId-forkProcess node proc =- modifyMVarMasked (localState node) startProcess+forkProcess node proc = do+ ms <- getMaskingState+ modifyMVarMasked (localState node) (startProcess ms) where- startProcess :: LocalNodeState -> IO (LocalNodeState, ProcessId)- startProcess (LocalNodeValid vst) = do+ startProcess :: MaskingState+ -> LocalNodeState+ -> IO (LocalNodeState, ProcessId)+ startProcess ms (LocalNodeValid vst) = do let lpid = LocalProcessId { lpidCounter = vst ^. localPidCounter , lpidUnique = vst ^. localPidUnique }@@ -387,7 +395,13 @@ , processThread = tid , processNode = node }- tid' <- uninterruptibleMask_ $ forkIOWithUnmask $ \unmask -> do+ -- Rewrite this code when this is fixed:+ -- https://ghc.haskell.org/trac/ghc/ticket/10149+ let unmask = case ms of+ Unmasked -> unblock+ MaskedInterruptible -> block+ MaskedUninterruptible -> id+ tid' <- uninterruptibleMask_ $ forkIO $ do reason <- Exception.catches (unmask $ runLocalProcess lproc proc >> return DiedNormal) [ (Exception.Handler (\ex@(ProcessExitException from msg) -> do@@ -433,7 +447,7 @@ $ vst , pid )- startProcess LocalNodeClosed =+ startProcess _ LocalNodeClosed = throwIO $ NodeClosedException $ localNodeId node cleanupProcess :: ProcessId@@ -1158,16 +1172,17 @@ throwException pid e = do node <- ask -- throwTo blocks until the exception is received by the target thread.- -- We fork a helper thread to avoid blocking the node controller.- liftIO $ withLocalProc node pid $ \p -> void $ forkIO $ throwTo (processThread p) e+ -- We cannot easily make it happen asynchronpusly because then 'unlink'+ -- semantics would break.+ liftIO $ withLocalProc node pid $ \p -> throwTo (processThread p) e withLocalProc :: LocalNode -> ProcessId -> (LocalProcess -> IO ()) -> IO () withLocalProc node pid p = -- By [Unified: table 6, rule missing_process] messages to dead processes -- can silently be dropped let lpid = processLocalId pid in do- withValidLocalState node $ \vst ->- forM_ (vst ^. localProcessWithId lpid) p+ join $ withValidLocalState node $ \vst ->+ return $ forM_ (vst ^. localProcessWithId lpid) p -------------------------------------------------------------------------------- -- Accessors --