diff --git a/ChangeLog b/ChangeLog --- a/ChangeLog +++ b/ChangeLog @@ -1,4 +1,20 @@ -2014-30-05 Tim Watson 0.5.0 +2014-12-09 Tim Watson 0.5.2 + +* Fix docstring for `register` +* Added Data instance to ProcessId, LocalProcessId and NodeId +* Add static serialiation dictionary for 'Static', for completeness +* Add Closure static serialization dictionary +* Replacement for modifyMVarMasked for GHC <= 7.4 +* Document the use of built-in trace flags +* Make forkProcess exception-safe +* Make -Wall clean + +2014-08-13 Tim Watson 0.5.1 + +* Fix cabal docs (thanks Markus Barenhoff) +* Expose lifted version of Control.Exception.mask_ (thanks Alexander Vershilov) + +2014-05-30 Tim Watson 0.5.0 * Dependency on STM implicitly changed from 1.3 to 1.4, but was not reflected in the cabal file * Race condition in local monitoring when using call 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.5.1 +Version: 0.5.2 Cabal-Version: >=1.8 Build-Type: Simple License: BSD3 @@ -42,7 +42,7 @@ Build-Depends: base >= 4.4 && < 5, binary >= 0.6.3 && < 0.8, hashable >= 1.2.0.5 && < 1.3, - network-transport >= 0.4.0.0 && < 0.5, + network-transport >= 0.4.1.0 && < 0.5, stm >= 2.4 && < 2.5, transformers >= 0.2 && < 0.5, mtl >= 2.0 && < 2.3, diff --git a/src/Control/Distributed/Process/Closure.hs b/src/Control/Distributed/Process/Closure.hs --- a/src/Control/Distributed/Process/Closure.hs +++ b/src/Control/Distributed/Process/Closure.hs @@ -163,6 +163,8 @@ , sdictUnit , sdictProcessId , sdictSendPort + , sdictStatic + , sdictClosure -- * The CP type and associated combinators , CP , idCP @@ -203,6 +205,8 @@ , sdictUnit , sdictProcessId , sdictSendPort + , sdictStatic + , sdictClosure -- The CP type and associated combinators , CP , idCP diff --git a/src/Control/Distributed/Process/Debug.hs b/src/Control/Distributed/Process/Debug.hs --- a/src/Control/Distributed/Process/Debug.hs +++ b/src/Control/Distributed/Process/Debug.hs @@ -78,6 +78,19 @@ -- This is checked for /any/ non-empty value. If set, all internal traces are -- written to the GHC eventlog. -- +-- By default, the built in tracers will ignore all trace events! In order to +-- enable tracing the incoming 'MxEvent' stream, the @DISTRIBUTED_PROCESS_TRACE_FLAGS@ +-- environment variable accepts the following flags, which enable tracing specific +-- event types: +-- +-- p = trace the spawning of new processes +-- d = trace the death of processes +-- n = trace registration of names (i.e., named processes) +-- u = trace un-registration of names (i.e., named processes) +-- s = trace the sending of messages to other processes +-- r = trace the receipt of messages from other processes +-- l = trace node up/down events +-- -- Users of the /simplelocalnet/ Cloud Haskell backend should also note that -- because the trace file option only supports trace output from a single node -- (so as to avoid interleaving), a file trace configured for the master node diff --git a/src/Control/Distributed/Process/Internal/Closure/BuiltIn.hs b/src/Control/Distributed/Process/Internal/Closure/BuiltIn.hs --- a/src/Control/Distributed/Process/Internal/Closure/BuiltIn.hs +++ b/src/Control/Distributed/Process/Internal/Closure/BuiltIn.hs @@ -8,6 +8,8 @@ , sdictUnit , sdictProcessId , sdictSendPort + , sdictStatic + , sdictClosure -- * Some static values , sndStatic -- * The CP type and associated combinators @@ -82,6 +84,7 @@ . registerStatic "$sdictUnit" (toDynamic (SerializableDict :: SerializableDict ())) . registerStatic "$sdictProcessId" (toDynamic (SerializableDict :: SerializableDict ProcessId)) . registerStatic "$sdictSendPort_" (toDynamic (sdictSendPort_ :: SerializableDict ANY -> SerializableDict (SendPort ANY))) + . registerStatic "$sdictClosure" (toDynamic (SerializableDict :: SerializableDict (Closure ANY))) . registerStatic "$returnProcess" (toDynamic (return :: ANY -> Process ANY)) . registerStatic "$seqProcess" (toDynamic ((>>) :: Process ANY1 -> Process ANY2 -> Process ANY2)) . registerStatic "$bindProcess" (toDynamic ((>>=) :: Process ANY1 -> (ANY1 -> Process ANY2) -> Process ANY2)) @@ -143,6 +146,14 @@ sdictSendPort :: Typeable a => Static (SerializableDict a) -> Static (SerializableDict (SendPort a)) sdictSendPort = staticApply (staticLabel "$sdictSendPort_") + +-- | Serialization dictionary for 'Static'. +sdictStatic :: Typeable a => Static (SerializableDict (Static a)) +sdictStatic = staticLabel "$sdictStatic" + +-- | Serialization dictionary for 'Closure'. +sdictClosure :: Typeable a => Static (SerializableDict (Closure a)) +sdictClosure = staticLabel "$sdictClosure" -------------------------------------------------------------------------------- -- Static values -- diff --git a/src/Control/Distributed/Process/Internal/Messaging.hs b/src/Control/Distributed/Process/Internal/Messaging.hs --- a/src/Control/Distributed/Process/Internal/Messaging.hs +++ b/src/Control/Distributed/Process/Internal/Messaging.hs @@ -18,7 +18,7 @@ import Control.Concurrent.Chan (writeChan) import Control.Monad (unless) -import Control.Monad.IO.Class (MonadIO, liftIO) +import Control.Monad.IO.Class (liftIO) import Control.Monad.Reader (ask) import qualified Network.Transport as NT ( Connection @@ -195,4 +195,3 @@ (NodeIdentifier nid) WithImplicitReconnect msg - 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 @@ -127,7 +127,7 @@ import System.Timeout (timeout) import Control.Monad (when) import Control.Monad.Reader (ask) -import Control.Monad.IO.Class (MonadIO, liftIO) +import Control.Monad.IO.Class (liftIO) import Control.Applicative ((<$>)) import Control.Exception (Exception(..), throw, throwIO, SomeException) import qualified Control.Exception as Ex (catch, mask, mask_, try) @@ -738,10 +738,10 @@ selfNode <- getSelfNode if nid == selfNode then Right `fmap` getLocalNodeStats -- optimisation - else getNodeStatsRemote nid + else getNodeStatsRemote where - getNodeStatsRemote :: NodeId -> Process (Either DiedReason NodeStats) - getNodeStatsRemote nid = do + getNodeStatsRemote :: Process (Either DiedReason NodeStats) + getNodeStatsRemote = do sendCtrlMsg (Just nid) $ GetNodeStats nid bracket (monitorNode nid) unmonitor $ \mRef -> receiveWait [ match (\(stats :: NodeStats) -> return $ Right stats) @@ -1025,11 +1025,9 @@ -- Registry -- -------------------------------------------------------------------------------- --- | Register a process with the local registry (asynchronous). --- This version will wait until a response is gotten from the --- management process. The name must not already be registered. --- The process need not be on this node. --- A bad registration will result in a 'ProcessRegistrationException' +-- | Register a process with the local registry (synchronous). The name must not +-- already be registered. The process need not be on this node. A bad +-- registration will result in a 'ProcessRegistrationException' -- -- The process to be registered does not have to be local itself. register :: String -> ProcessId -> Process () diff --git a/src/Control/Distributed/Process/Internal/StrictMVar.hs b/src/Control/Distributed/Process/Internal/StrictMVar.hs --- a/src/Control/Distributed/Process/Internal/StrictMVar.hs +++ b/src/Control/Distributed/Process/Internal/StrictMVar.hs @@ -1,5 +1,5 @@ -- | Like Control.Concurrent.MVar.Strict but reduce to HNF, not NF -{-# LANGUAGE MagicHash, UnboxedTuples #-} +{-# LANGUAGE CPP, MagicHash, UnboxedTuples #-} module Control.Distributed.Process.Internal.StrictMVar ( StrictMVar(StrictMVar) , newEmptyMVar @@ -10,12 +10,17 @@ , withMVar , modifyMVar_ , modifyMVar + , modifyMVarMasked , mkWeakMVar ) where import Control.Applicative ((<$>)) import Control.Monad ((>=>)) +#if MIN_VERSION_base(4,6,0) import Control.Exception (evaluate) +#else +import Control.Exception (evaluate, mask_, onException) +#endif import qualified Control.Concurrent.MVar as MVar ( MVar , newEmptyMVar @@ -26,6 +31,9 @@ , withMVar , modifyMVar_ , modifyMVar +#if MIN_VERSION_base(4,6,0) + , modifyMVarMasked +#endif ) import GHC.MVar (MVar(MVar)) import GHC.IO (IO(IO)) @@ -57,6 +65,21 @@ modifyMVar :: StrictMVar a -> (a -> IO (a, b)) -> IO b modifyMVar (StrictMVar v) f = MVar.modifyMVar v (f >=> evaluateFst) + where + evaluateFst :: (a, b) -> IO (a, b) + evaluateFst (x, y) = evaluate x >> return (x, y) + +modifyMVarMasked :: StrictMVar a -> (a -> IO (a, b)) -> IO b +modifyMVarMasked (StrictMVar v) f = +#if MIN_VERSION_base(4,6,0) + MVar.modifyMVarMasked v (f >=> evaluateFst) +#else + mask_ $ do + a <- MVar.takeMVar v + (a',b) <- (f a >>= evaluate) `onException` MVar.putMVar v a + MVar.putMVar v a' + return b +#endif where evaluateFst :: (a, b) -> IO (a, b) evaluateFst (x, y) = evaluate x >> return (x, y) diff --git a/src/Control/Distributed/Process/Internal/Types.hs b/src/Control/Distributed/Process/Internal/Types.hs --- a/src/Control/Distributed/Process/Internal/Types.hs +++ b/src/Control/Distributed/Process/Internal/Types.hs @@ -86,6 +86,7 @@ import System.Mem.Weak (Weak) import Data.Map (Map) import Data.Int (Int32) +import Data.Data (Data) import Data.Typeable (Typeable, typeOf) import Data.Binary (Binary(put, get), putWord8, getWord8, encode) import qualified Data.ByteString as BSS (ByteString, concat, copy) @@ -133,7 +134,7 @@ -- | Node identifier newtype NodeId = NodeId { nodeAddress :: NT.EndPointAddress } - deriving (Eq, Ord, Typeable, Generic) + deriving (Eq, Ord, Typeable, Data, Generic) instance Binary NodeId where instance NFData NodeId instance Hashable NodeId where @@ -146,7 +147,7 @@ { lpidUnique :: {-# UNPACK #-} !Int32 , lpidCounter :: {-# UNPACK #-} !Int32 } - deriving (Eq, Ord, Typeable, Generic, Show) + deriving (Eq, Ord, Typeable, Data, Generic, Show) instance Hashable LocalProcessId where @@ -157,7 +158,7 @@ -- | Node-local identifier for the process , processLocalId :: {-# UNPACK #-} !LocalProcessId } - deriving (Eq, Ord, Typeable, Generic) + deriving (Eq, Ord, Typeable, Data, Generic) instance Binary ProcessId where instance NFData ProcessId where diff --git a/src/Control/Distributed/Process/Management.hs b/src/Control/Distributed/Process/Management.hs --- a/src/Control/Distributed/Process/Management.hs +++ b/src/Control/Distributed/Process/Management.hs @@ -276,8 +276,7 @@ import Control.Applicative ((<$>)) import Control.Concurrent.STM (atomically) import Control.Concurrent.STM.TChan - ( tryReadTChan - , readTChan + ( readTChan , writeTChan , TChan ) @@ -285,7 +284,6 @@ ( newChan , nsend , receiveWait - , receiveTimeout , matchChan , matchAny , matchSTM @@ -320,9 +318,7 @@ import Control.Monad.IO.Class (liftIO) import Control.Monad.Reader (ask) import qualified Control.Monad.State as ST - ( MonadState - , StateT - , get + ( get , modify , lift , runStateT @@ -526,8 +522,7 @@ -- MxAgentBecome h' -> runAgent h' c state getNextInput sel chan = - let stmRead = atomically . readTChan - matches = + let matches = case sel of Mailbox -> [ matchAny return , matchSTM (readTChan chan) return] @@ -554,4 +549,3 @@ case pass of Nothing -> runPipeline msg state next Just result -> return (result, state') - 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 @@ -52,7 +52,7 @@ import Data.Maybe (isJust, fromJust, isNothing, catMaybes) import Data.Typeable (Typeable) import Control.Category ((>>>)) -import Control.Applicative ((<$>)) +import Control.Applicative (Applicative, (<$>)) import Control.Monad (void, when) import Control.Monad.IO.Class (MonadIO, liftIO) import Control.Monad.State.Strict (MonadState, StateT, evalStateT, gets) @@ -64,13 +64,14 @@ , SomeException , Exception , throwTo + , uninterruptibleMask_ ) import qualified Control.Exception as Exception (Handler(..), catches, finally) -import Control.Concurrent (forkIO, myThreadId) +import Control.Concurrent (forkIO, forkIOWithUnmask, myThreadId) import Control.Distributed.Process.Internal.StrictMVar ( newMVar , withMVar - , modifyMVar + , modifyMVarMasked , modifyMVar_ , newEmptyMVar , putMVar @@ -99,7 +100,6 @@ , address , closeEndPoint , ConnectionId - , Connection , close , EndPointAddress , Reliability(ReliableOrdered) @@ -332,7 +332,8 @@ -- | Spawn a new process on a local node forkProcess :: LocalNode -> Process () -> IO ProcessId -forkProcess node proc = modifyMVar (localState node) startProcess +forkProcess node proc = + modifyMVarMasked (localState node) startProcess where startProcess :: LocalNodeState -> IO (LocalNodeState, ProcessId) startProcess st = do @@ -357,9 +358,9 @@ , processThread = tid , processNode = node } - tid' <- forkIO $ do + tid' <- uninterruptibleMask_ $ forkIOWithUnmask $ \unmask -> do reason <- Exception.catches - (runLocalProcess lproc proc >> return DiedNormal) + (unmask $ runLocalProcess lproc proc >> return DiedNormal) [ (Exception.Handler (\ex@(ProcessExitException from msg) -> do mMsg <- unwrapMessage msg :: IO (Maybe String) case mMsg of @@ -592,7 +593,13 @@ } newtype NC a = NC { unNC :: StateT NCState (ReaderT LocalNode IO) a } - deriving (Functor, Monad, MonadIO, MonadState NCState, MonadReader LocalNode) + deriving ( Applicative + , Functor + , Monad + , MonadIO + , MonadState NCState + , MonadReader LocalNode + ) initNCState :: NCState initNCState = NCState { _links = Map.empty diff --git a/src/Control/Distributed/Process/UnsafePrimitives.hs b/src/Control/Distributed/Process/UnsafePrimitives.hs --- a/src/Control/Distributed/Process/UnsafePrimitives.hs +++ b/src/Control/Distributed/Process/UnsafePrimitives.hs @@ -70,7 +70,7 @@ ) import Control.Distributed.Process.Serializable (Serializable) -import Control.Monad.IO.Class (MonadIO, liftIO) +import Control.Monad.IO.Class (liftIO) import Control.Monad.Reader (ask) -- | Named send to a process in the local registry (asynchronous) @@ -118,4 +118,3 @@ -- | Create an unencoded @Message@ for any @Serializable@ type. wrapMessage :: Serializable a => a -> Message wrapMessage = unsafeCreateUnencodedMessage -