distributed-process 0.5.1 → 0.5.2

raw patch · 12 files changed

+102/−37 lines, 12 filesdep ~network-transportPVP: major bump suggested

API removals or changes: PVP suggests a major version bump

Dependency ranges changed: network-transport

API changes (from Hackage documentation)

Files

• ChangeLog +17/−1
• distributed-process.cabal +2/−2
• src/Control/Distributed/Process/Closure.hs +4/−0
• src/Control/Distributed/Process/Debug.hs +13/−0
• src/Control/Distributed/Process/Internal/Closure/BuiltIn.hs +11/−0
• src/Control/Distributed/Process/Internal/Messaging.hs +1/−2
• src/Control/Distributed/Process/Internal/Primitives.hs +7/−9
• src/Control/Distributed/Process/Internal/StrictMVar.hs +24/−1
• src/Control/Distributed/Process/Internal/Types.hs +4/−3
• src/Control/Distributed/Process/Management.hs +3/−9
• src/Control/Distributed/Process/Node.hs +15/−8
• src/Control/Distributed/Process/UnsafePrimitives.hs +1/−2

@@ -1,4 +1,20 @@-2014-30-05  Tim Watson  <watson.timothy@gmail.com>  0.5.0+2014-12-09  Tim Watson  <watson.timothy@gmail.com>  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  <watson.timothy@gmail.com>  0.5.1++* Fix cabal docs (thanks Markus Barenhoff)+* Expose lifted version of Control.Exception.mask_ (thanks Alexander Vershilov)++2014-05-30  Tim Watson  <watson.timothy@gmail.com>  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

@@ -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,

@@ -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

@@ -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

@@ -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                                                              --

@@ -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-

@@ -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 ()

@@ -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)

@@ -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

@@ -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')-

@@ -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

@@ -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-