distributed-process 0.7.4 → 0.7.5

raw patch · 19 files changed

+355/−199 lines, 19 filesdep ~basedep ~bytestringdep ~containersnew-uploaderPVP: major bump suggested

API removals or changes: PVP suggests a major version bump

Dependency ranges changed: base, bytestring, containers, hashable, network-transport-tcp, random, stm, transformers

API changes (from Hackage documentation)

Files

• ChangeLog +4/−2
• benchmarks/Channels.hs +6/−5
• benchmarks/Latency.hs +7/−5
• benchmarks/ProcessRing.hs +2/−2
• benchmarks/Spawns.hs +6/−5
• benchmarks/Throughput.hs +6/−5
• distributed-process.cabal +57/−51
• src/Control/Distributed/Process/Internal/CQueue.hs +0/−2
• src/Control/Distributed/Process/Internal/Closure/Explicit.hs +3/−1
• src/Control/Distributed/Process/Internal/Closure/TH.hs +27/−1
• src/Control/Distributed/Process/Internal/Primitives.hs +63/−46
• src/Control/Distributed/Process/Internal/StrictMVar.hs +0/−15
• src/Control/Distributed/Process/Internal/Types.hs +6/−0
• src/Control/Distributed/Process/Management.hs +34/−2
• src/Control/Distributed/Process/Management/Internal/Trace/Primitives.hs +8/−2
• src/Control/Distributed/Process/Management/Internal/Types.hs +9/−1
• src/Control/Distributed/Process/Node.hs +32/−19
• src/Control/Distributed/Process/Serializable.hs +2/−2
• src/Control/Distributed/Process/UnsafePrimitives.hs +83/−33

@@ -1,6 +1,8 @@-[Unrealeased]+2018-06-12 David Simmons-Duffin <dsd@caltech.edu> 0.7.5 -*+* Bumped dependencies to build with ghc-9.8+* Turn Serializable into a type synonym+* Remove tests for ghc-8.8.* and earlier  2018-06-12 Facundo Domínguez <facundo.dominguez@tweag.io> 0.7.4 

@@ -4,7 +4,7 @@ import Control.Applicative import Control.Distributed.Process import Control.Distributed.Process.Node-import Network.Transport.TCP (createTransport, defaultTCPParameters)+import Network.Transport.TCP (createTransport, defaultTCPAddr, defaultTCPParameters) import Data.Binary (encode, decode) import qualified Data.ByteString.Lazy as BSL @@ -36,7 +36,8 @@ main :: IO () main = do   [role, host, port] <- getArgs-  Right transport <- createTransport-      host port (\sn -> (host, port)) defaultTCPParameters-  node <- newLocalNode transport initRemoteTable-  runProcess node $ initialProcess role+  trans <- createTransport (defaultTCPAddr host port) defaultTCPParameters+  case trans of+    Right transport -> do node <- newLocalNode transport initRemoteTable+                          runProcess node $ initialProcess role+    Left other -> error $ show other

@@ -3,7 +3,7 @@ import Control.Applicative import Control.Distributed.Process import Control.Distributed.Process.Node-import Network.Transport.TCP (createTransport, defaultTCPParameters)+import Network.Transport.TCP (createTransport, defaultTCPAddr, defaultTCPParameters) import Data.Binary (encode, decode) import qualified Data.ByteString.Lazy as BSL @@ -31,7 +31,9 @@ main :: IO () main = do   [role, host, port] <- getArgs-  Right transport <- createTransport-      host port (\sn -> (host, sn)) defaultTCPParameters-  node <- newLocalNode transport initRemoteTable-  runProcess node $ initialProcess role+  trans <- createTransport+                      (defaultTCPAddr host port) defaultTCPParameters+  case trans of+    Right transport -> do node <- newLocalNode transport initRemoteTable+                          runProcess node $ initialProcess role+    Left other -> error $ show other

@@ -12,7 +12,7 @@ import Control.Distributed.Process hiding (catch) import Control.Distributed.Process.Node import Control.Exception (catch, SomeException)-import Network.Transport.TCP (createTransport, defaultTCPParameters)+import Network.Transport.TCP (createTransport, defaultTCPAddr, defaultTCPParameters) import System.Environment import System.Console.GetOpt @@ -111,7 +111,7 @@   (opt, _) <- parseArgv argv   putStrLn $ "options: " ++ (show opt)   Right transport <- createTransport-    "127.0.0.1" "8090" (\sn -> ("127.0.0.1", sn)) defaultTCPParameters+                        (defaultTCPAddr "127.0.0.1" "8090" ) defaultTCPParameters   node <- newLocalNode transport initRemoteTable   catch (void $ runProcess node $ initialProcess opt)         (\(e :: SomeException) -> putStrLn $ "ERROR: " ++ (show e))

@@ -7,7 +7,7 @@ import Control.Applicative import Control.Distributed.Process import Control.Distributed.Process.Node-import Network.Transport.TCP (createTransport, defaultTCPParameters)+import Network.Transport.TCP (createTransport, defaultTCPAddr, defaultTCPParameters) import Data.Binary (encode, decode) import qualified Data.ByteString.Lazy as BSL @@ -42,7 +42,8 @@ main :: IO () main = do   [role, host, port] <- getArgs-  Right transport <- createTransport-    host port (\sn -> (host, sn)) defaultTCPParameters-  node <- newLocalNode transport initRemoteTable-  runProcess node $ initialProcess role+  trans <- createTransport (defaultTCPAddr host port) defaultTCPParameters+  case trans of+    Right transport -> do node <- newLocalNode transport initRemoteTable+                          runProcess node $ initialProcess role+    Left other -> error $ show other

@@ -6,7 +6,7 @@ import Control.Applicative import Control.Distributed.Process import Control.Distributed.Process.Node-import Network.Transport.TCP (createTransport, defaultTCPParameters)+import Network.Transport.TCP (createTransport, defaultTCPParameters, defaultTCPAddr) import Data.Binary import qualified Data.ByteString.Lazy as BSL import Data.Typeable@@ -67,7 +67,8 @@ main :: IO () main = do   [role, host, port] <- getArgs-  Right transport <- createTransport-      host port (\sn -> (host, sn)) defaultTCPParameters-  node <- newLocalNode transport initRemoteTable-  runProcess node $ initialProcess role+  trans <- createTransport (defaultTCPAddr host port) defaultTCPParameters+  case trans of+    Right transport -> do node <- newLocalNode transport initRemoteTable+                          runProcess node $ initialProcess role+    Left other -> error $ show other

@@ -1,12 +1,12 @@ Name:          distributed-process-Version:       0.7.4-Cabal-Version: >=1.8+Version:       0.7.5+Cabal-Version: >=1.10 Build-Type:    Simple License:       BSD3 License-File:  LICENSE Copyright:     Well-Typed LLP, Tweag I/O Limited Author:        Duncan Coutts, Nicolas Wu, Edsko de Vries-Maintainer:    Facundo Domínguez <facundo.dominguez@tweag.io>+Maintainer:    Tim Watson <watson.timothy@gmail.com> Stability:     experimental Homepage:      http://haskell-distributed.github.com/ Bug-Reports:   https://github.com/haskell-distributed/distributed-process/issues@@ -21,7 +21,7 @@                 You will probably also want to install a Cloud Haskell backend such                as distributed-process-simplelocalnet.-Tested-With:   GHC==7.6.3 GHC==7.8.4 GHC==7.10.3 GHC==8.0.1+Tested-With:   GHC==7.10.3 GHC==8.0.2 GHC==8.2.2 GHC==8.4.4 Category:      Control extra-source-files: ChangeLog @@ -41,21 +41,21 @@  default: False  Library-  Build-Depends:     base >= 4.6 && < 5,+  Build-Depends:     base >= 4.9 && < 5,                      binary >= 0.6.3 && < 0.10,-                     hashable >= 1.2.0.5 && < 1.3,+                     hashable >= 1.2.0.5 && <= 1.4.3.0,                      network-transport >= 0.4.1.0 && < 0.6,-                     stm >= 2.4 && < 2.5,-                     transformers >= 0.2 && < 0.6,+                     stm >= 2.4 && < 2.6,+                     transformers >= 0.2 && < 0.7,                      mtl >= 2.0 && < 2.4,                      data-accessor >= 0.2 && < 0.3,-                     bytestring >= 0.9 && < 0.11,-                     random >= 1.0 && < 1.2,+                     bytestring >= 0.9 && < 0.13,+                     random >= 1.0 && < 1.3,                      distributed-static >= 0.2 && < 0.4,                      rank1dynamic >= 0.1 && < 0.5,                      syb >= 0.3 && < 0.8,                      exceptions >= 0.5,-                     containers >= 0.5 && < 0.6,+                     containers >= 0.5 && < 0.7,                      deepseq >= 1.3.0.1 && < 1.6   Exposed-modules:   Control.Distributed.Process,                      Control.Distributed.Process.Closure,@@ -83,6 +83,7 @@                      Control.Distributed.Process.Management.Internal.Trace.Remote,                      Control.Distributed.Process.Management.Internal.Trace.Types,                      Control.Distributed.Process.Management.Internal.Trace.Tracer+  default-language:  Haskell2010   ghc-options:       -Wall   HS-Source-Dirs:    src   other-extensions:  BangPatterns@@ -118,51 +119,56 @@ -- Tests are in distributed-process-test package, for convenience.  benchmark distributed-process-throughput-  Type:            exitcode-stdio-1.0-  Build-Depends:   base >= 4.6 && < 5,-                   distributed-process,-                   network-transport-tcp >= 0.3 && < 0.7,-                   bytestring >= 0.9 && < 0.11,-                   binary >= 0.6.3 && < 0.10-  Main-Is:         benchmarks/Throughput.hs-  ghc-options:     -Wall+  Type:             exitcode-stdio-1.0+  Build-Depends:    base >= 4.9 && < 5,+                    distributed-process,+                    network-transport-tcp >= 0.3 && <= 0.81,+                    bytestring >= 0.9 && < 0.13,+                    binary >= 0.6.3 && < 0.10+  Main-Is:          benchmarks/Throughput.hs+  default-language: Haskell2010+  ghc-options:      -Wall  benchmark distributed-process-latency-  Type:            exitcode-stdio-1.0-  Build-Depends:   base >= 4.6 && < 5,-                   distributed-process,-                   network-transport-tcp >= 0.3 && < 0.7,-                   bytestring >= 0.9 && < 0.11,-                   binary >= 0.6.3 && < 0.10-  Main-Is:         benchmarks/Latency.hs-  ghc-options:     -Wall+  Type:             exitcode-stdio-1.0+  Build-Depends:    base >= 4.9 && < 5,+                    distributed-process,+                    network-transport-tcp >= 0.3 && <= 0.81,+                    bytestring >= 0.9 && < 0.13,+                    binary >= 0.6.3 && < 0.10+  Main-Is:          benchmarks/Latency.hs+  default-language: Haskell2010+  ghc-options:      -Wall  benchmark distributed-process-channels-  Type:            exitcode-stdio-1.0-  Build-Depends:   base >= 4.6 && < 5,-                   distributed-process,-                   network-transport-tcp >= 0.3 && < 0.7,-                   bytestring >= 0.9 && < 0.11,-                   binary >= 0.6.3 && < 0.10-  Main-Is:         benchmarks/Channels.hs-  ghc-options:     -Wall+  Type:             exitcode-stdio-1.0+  Build-Depends:    base >= 4.9 && < 5,+                    distributed-process,+                    network-transport-tcp >= 0.3 && <= 0.81,+                    bytestring >= 0.9 && < 0.13,+                    binary >= 0.6.3 && < 0.10+  Main-Is:          benchmarks/Channels.hs+  default-language: Haskell2010+  ghc-options:      -Wall  benchmark distributed-process-spawns-  Type:            exitcode-stdio-1.0-  Build-Depends:   base >= 4.6 && < 5,-                   distributed-process,-                   network-transport-tcp >= 0.3 && < 0.7,-                   bytestring >= 0.9 && < 0.11,-                   binary >= 0.6.3 && < 0.10-  Main-Is:         benchmarks/Spawns.hs-  ghc-options:     -Wall+  Type:             exitcode-stdio-1.0+  Build-Depends:    base >= 4.9 && < 5,+                    distributed-process,+                    network-transport-tcp >= 0.3 && <= 0.81,+                    bytestring >= 0.9 && < 0.13,+                    binary >= 0.6.3 && < 0.10+  Main-Is:          benchmarks/Spawns.hs+  default-language: Haskell2010+  ghc-options:      -Wall  benchmark distributed-process-ring-  Type:            exitcode-stdio-1.0-  Build-Depends:   base >= 4.6 && < 5,-                   distributed-process,-                   network-transport-tcp >= 0.3 && < 0.7,-                   bytestring >= 0.9 && < 0.11,-                   binary >= 0.6.3 && < 0.10-  Main-Is:         benchmarks/ProcessRing.hs-  ghc-options:     -Wall -threaded -O2 -rtsopts+  Type:             exitcode-stdio-1.0+  Build-Depends:    base >= 4.9 && < 5,+                    distributed-process,+                    network-transport-tcp >= 0.3 && <= 0.81,+                    bytestring >= 0.9 && < 0.13,+                    binary >= 0.6.3 && < 0.10+  Main-Is:          benchmarks/ProcessRing.hs+  default-language: Haskell2010+  ghc-options:      -Wall -threaded -O2 -rtsopts

@@ -31,7 +31,6 @@   , orElse   , retry   )-import Control.Applicative ((<$>), (<*>)) import Control.Exception (mask_, onException) import System.Timeout (timeout) import Control.Distributed.Process.Internal.StrictMVar@@ -45,7 +44,6 @@   , append   ) import Data.Maybe (fromJust)-import Data.Traversable (traverse) import GHC.MVar (MVar(MVar)) import GHC.IO (IO(IO), unIO) import GHC.Exts (mkWeak#)

@@ -7,6 +7,7 @@   , KindSignatures   , GADTs   , EmptyDataDecls+  , TypeOperators   , DeriveDataTypeable #-} module Control.Distributed.Process.Internal.Closure.Explicit   (@@ -29,6 +30,7 @@ import Data.Rank1Typeable import Data.Binary(encode,put,get,Binary) import qualified Data.ByteString.Lazy as B+import Data.Kind (Type)  -- | A RemoteRegister is a trasformer on a RemoteTable to register additional static values. type RemoteRegister = RemoteTable -> RemoteTable@@ -118,7 +120,7 @@ -- This generic uncurry courtesy Andrea Vezzosi data HTrue data HFalse-data Fun :: * -> * -> * -> * where+data Fun :: Type -> Type -> Type -> Type where   Done :: Fun EndOfTuple r r   Moar :: Fun xs f r -> Fun (x,xs) (x -> f) r 

@@ -12,7 +12,6 @@   ) where  import Prelude hiding (succ, any)-import Control.Applicative ((<$>)) import Language.Haskell.TH   ( -- Q monad and operations     Q@@ -28,6 +27,9 @@   , Exp   , Type(AppT, ForallT, VarT, ArrowT)   , Info(VarI)+#if MIN_VERSION_template_haskell(2,17,0)+  , Specificity+#endif   , TyVarBndr(PlainTV, KindedTV)   , Pred #if MIN_VERSION_template_haskell(2,10,0)@@ -203,7 +205,11 @@            , concat [register, registerSDict, registerTDict]            )   where+#if MIN_VERSION_template_haskell(2,17,0)+    makeStatic :: [TyVarBndr Specificity] -> Type -> Q ([Dec], [Q Exp])+#else     makeStatic :: [TyVarBndr] -> Type -> Q ([Dec], [Q Exp])+#endif     makeStatic typVars typ = do       static <- generateStatic origName typVars typ       let dyn = case typVars of@@ -222,7 +228,12 @@              )  -- | Turn a polymorphic type into a monomorphic type using ANY and co+#if MIN_VERSION_template_haskell(2,17,0)+monomorphize :: [TyVarBndr Specificity] -> Type -> Q Type+#else monomorphize :: [TyVarBndr] -> Type -> Q Type+#endif+ monomorphize tvs =     let subst = zip (map tyVarBndrName tvs) anys     in everywhereM (mkM (applySubst subst))@@ -247,7 +258,11 @@     applySubst s t = gmapM (mkM (applySubst s)) t  -- | Generate a static value+#if MIN_VERSION_template_haskell(2,17,0)+generateStatic :: Name -> [TyVarBndr Specificity] -> Type -> Q [Dec]+#else generateStatic :: Name -> [TyVarBndr] -> Type -> Q [Dec]+#endif generateStatic n xs typ = do     staticTyp <- [t| Static |]     sequence@@ -259,7 +274,11 @@       , sfnD (staticName n) [| staticLabel $(showFQN n) |]       ]   where+#if MIN_VERSION_template_haskell(2,17,0)+    typeable :: TyVarBndr Specificity -> Q Pred+#else     typeable :: TyVarBndr -> Q Pred+#endif     typeable tv = #if MIN_VERSION_template_haskell(2,10,0)       conT (mkName "Typeable") `appT` varT (tyVarBndrName tv)@@ -315,9 +334,16 @@ sfnD n e = funD n [clause [] (normalB e) []]  -- | The name of a type variable binding occurrence+#if MIN_VERSION_template_haskell(2,17,0)+tyVarBndrName :: TyVarBndr Specificity -> Name+tyVarBndrName (PlainTV n _)    = n+tyVarBndrName (KindedTV n _ _) = n+#else tyVarBndrName :: TyVarBndr -> Name tyVarBndrName (PlainTV n)    = n tyVarBndrName (KindedTV n _) = n+#endif+  -- | Fully qualified name; that is, the name and the _current_ module --

@@ -76,6 +76,7 @@   , unlink   , monitor   , unmonitor+  , unmonitorAsync   , withMonitor   , withMonitor_     -- * Logging@@ -252,18 +253,16 @@   let us       = processId proc       node     = processNode proc       nodeId   = localNodeId node-      destNode = (processNodeId them) in do-  case destNode == nodeId of-    True  -> sendLocal them msg-    False -> liftIO $ sendMessage (processNode proc)-                                  (ProcessIdentifier (processId proc))-                                  (ProcessIdentifier them)-                                  NoImplicitReconnect-                                  msg-  -- We do not fire the trace event until after the sending is complete;-  -- In the remote case, 'sendMessage' can block in the networking stack.+      destNode = (processNodeId them)   liftIO $ traceEvent (localEventBus node)                       (MxSent them us (createUnencodedMessage msg))+  if destNode == nodeId+    then sendLocal them msg+    else liftIO $ sendMessage (processNode proc)+                              (ProcessIdentifier (processId proc))+                              (ProcessIdentifier them)+                              NoImplicitReconnect+                              msg  -- | /Unsafe/ variant of 'send'. This function makes /no/ attempt to serialize -- and (in the case when the destination process resides on the same local@@ -283,9 +282,12 @@ -- usend :: Serializable a => ProcessId -> a -> Process () usend them msg = do-    here <- getSelfNode+    proc <- ask     let there = processNodeId them-    if here == there+    let (us, node) = (processId proc, processNode proc)+    let msg' = wrapMessage msg+    liftIO $ traceEvent (localEventBus node) (MxSent them us msg')+    if localNodeId (processNode proc) == there       then sendLocal them msg       else sendCtrlMsg (Just there) $ UnreliableSend (processLocalId them)                                                      (createMessage msg)@@ -305,7 +307,13 @@ -- Channels                                                                   -- -------------------------------------------------------------------------------- --- | Create a new typed channel+-- | Create a new typed channel, bound to the calling Process+--+-- Note that the channel is bound to the lifecycle of the process that evaluates+-- this function, such that when it dies/exits, the channel will no longer+-- function, but will remain accessible. Thus reading from the ReceivePort will+-- fail silently thereafter, blocking indefinitely (unless a timeout is used).+-- newChan :: Serializable a => Process (SendPort a, ReceivePort a) newChan = do     proc <- ask@@ -333,13 +341,16 @@ sendChan :: Serializable a => SendPort a -> a -> Process () sendChan (SendPort cid) msg = do   proc <- ask-  let node     = localNodeId (processNode proc)-      destNode = processNodeId (sendPortProcessId cid) in do-  case destNode == node of+  let node = processNode proc+      pid  = processId proc+      us   = localNodeId node+      them = processNodeId (sendPortProcessId cid)+  liftIO $ traceEvent (localEventBus node) (MxSentToPort pid cid $ wrapMessage msg)+  case them == us of     True  -> sendChanLocal cid msg     False -> do-      liftIO $ sendBinary (processNode proc)-                          (ProcessIdentifier (processId proc))+      liftIO $ sendBinary node+                          (ProcessIdentifier pid)                           (SendPortIdentifier cid)                           NoImplicitReconnect                           msg@@ -400,8 +411,11 @@ receiveWait :: [Match b] -> Process b receiveWait ms = do   queue <- processQueue <$> ask-  Just proc <- liftIO $ dequeue queue Blocking (map unMatch ms)-  proc+  mProc <- liftIO $ dequeue queue Blocking (map unMatch ms)+  case mProc of+    Just proc' -> proc'+    Nothing    -> die $ "System Invariant Violation: CQueue.hs returned `Nothing` "+                     ++ "in the absence of a timeout value."  -- | Like 'receiveWait' but with a timeout. --@@ -479,18 +493,15 @@   let node     = processNode proc       us       = processId proc       nid      = localNodeId node-      destNode = (processNodeId them) in do-  case destNode == nid of-    True  -> sendCtrlMsg Nothing (LocalSend them msg)-    False -> liftIO $ sendPayload (processNode proc)-                                  (ProcessIdentifier (processId proc))-                                  (ProcessIdentifier them)-                                  NoImplicitReconnect-                                  (messageToPayload msg)-  -- We do not fire the trace event until after the sending is complete;-  -- In the remote case, 'sendMessage' can block in the networking stack.-  liftIO $ traceEvent (localEventBus node)-                      (MxSent them us msg)+      destNode = (processNodeId them)+  liftIO $ traceEvent (localEventBus node) (MxSent them us msg)+  if destNode == nid+    then sendCtrlMsg Nothing (LocalSend them msg)+    else liftIO $ sendPayload (processNode proc)+                              (ProcessIdentifier (processId proc))+                              (ProcessIdentifier them)+                              NoImplicitReconnect+                              (messageToPayload msg)  -- | Forward a raw 'Message' to the given 'ProcessId'. --@@ -503,15 +514,11 @@   let node     = processNode proc       us       = processId proc       nid      = localNodeId node-      destNode = (processNodeId them) in do-  case destNode == nid of-    True  -> sendCtrlMsg Nothing (LocalSend them msg)-    False -> sendCtrlMsg (Just destNode) $ UnreliableSend (processLocalId them)-                                                          msg-  -- We do not fire the trace event until after the sending is complete;-  -- In the remote case, 'sendCtrlMsg' can block in the networking stack.-  liftIO $ traceEvent (localEventBus node)-                      (MxSent them us msg)+      destNode = (processNodeId them)+  liftIO $ traceEvent (localEventBus node) (MxSent them us msg)+  if destNode == nid+    then sendCtrlMsg Nothing (LocalSend them msg)+    else sendCtrlMsg (Just destNode) $ UnreliableSend (processLocalId them) msg  -- | Wrap a 'Serializable' value in a 'Message'. Note that 'Message's are -- 'Serializable' - like the datum they contain - but also note, deserialising@@ -1220,8 +1227,12 @@  -- | Named send to a process in the local registry (asynchronous) nsend :: Serializable a => String -> a -> Process ()-nsend label msg =-  sendCtrlMsg Nothing (NamedSend label (createUnencodedMessage msg))+nsend label msg = do+  proc <- ask+  let msg' = createUnencodedMessage msg+  liftIO $ traceEvent (localEventBus (processNode proc))+                      (MxSentToName label (processId proc) msg')+  sendCtrlMsg Nothing (NamedSend label msg')  -- | Named send to a process in the local registry (asynchronous). -- This function makes /no/ attempt to serialize and (in the case when the@@ -1233,9 +1244,15 @@ -- | Named send to a process in a remote registry (asynchronous) nsendRemote :: Serializable a => NodeId -> String -> a -> Process () nsendRemote nid label msg = do-  here <- getSelfNode-  if here == nid then nsend label msg-    else sendCtrlMsg (Just nid) (NamedSend label (createMessage msg))+  proc <- ask+  let us = processId proc+  let node = processNode proc+  if localNodeId node == nid+    then nsend label msg+    else let lbl = label ++ "@" ++ show nid in do+            liftIO $ traceEvent (localEventBus node)+                                (MxSentToName lbl us (wrapMessage msg))+            sendCtrlMsg (Just nid) (NamedSend label (createMessage msg))  -- | Named send to a process in a remote registry (asynchronous) -- This function makes /no/ attempt to serialize and (in the case when the

@@ -14,13 +14,8 @@   , 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@@ -31,9 +26,6 @@   , withMVar   , modifyMVar_   , modifyMVar-#if MIN_VERSION_base(4,6,0)-  , modifyMVarMasked-#endif   ) import GHC.MVar (MVar(MVar)) import GHC.IO (IO(IO), unIO)@@ -71,18 +63,11 @@  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)  mkWeakMVar :: StrictMVar a -> IO () -> IO (Weak (StrictMVar a)) mkWeakMVar q@(StrictMVar (MVar m#)) f = IO $ \s ->

@@ -118,6 +118,9 @@ import Control.Monad.Catch (MonadThrow(..), MonadCatch(..), MonadMask(..)) import qualified Network.Transport as NT (EndPoint, EndPointAddress, Connection) import Control.Applicative+#if !MIN_VERSION_base(4,13,0) && MIN_VERSION_base(4,9,0)+import Control.Monad.Fail (MonadFail)+#endif import Control.Monad.Fix (MonadFix) import Control.Monad.Reader (MonadReader(..), ReaderT, runReaderT) import Control.Monad.IO.Class (MonadIO(..))@@ -353,6 +356,9 @@   deriving ( Applicative            , Functor            , Monad+#if MIN_VERSION_base(4,9,0)+           , MonadFail+#endif            , MonadFix            , MonadIO            , MonadReader LocalProcess

@@ -76,6 +76,40 @@ -- --  * Whether messages will be taken from the mailbox first, or the event bus. --+-- Since the event bus uses STM broadcast channels to communicate with agents,+-- no message written to the bus successfully can be lost.+--+-- Agents can also receive messages via their mailboxes - these are subject to+-- the same guarantees as all inter-process message sending.+--+-- Messages dispatched on an STM broadcast channel (i.e., management event bus)+-- are guaranteed to be delivered with the same FIFO ordering guarantees that+-- exist between two communicating processes, such that communication from the+-- node controller's threads (i.e., MxEvent's) will never be re-ordered, but+-- messages dispatched to the event bus by other processes (including, but not+-- limited to agents) are only guaranteed to be ordered between one sender and+-- one receiver.+--+-- No guarantee exists for the ordering in which messages sent to an agent's+-- mailbox will be delivered, vs messages dispatched via the event bus.+--+-- Because of the above, there are no ordering guarantees for messages sent+-- between agents, or for processes to agents, except for those that apply to+-- messages sent between regular processes, since agents are+-- implemented as such.+--+-- The event bus is serial and single threaded. Anything that is published by+-- the node controller will be seen in FIFO order. There are no ordering+-- guarantees pertaining to entries published to the event bus by other+-- processes or agents.+--+-- It should not be possible to see, for example, an @MxReceived@ before the+-- corresponding @MxSent@ event, since the places where we issue the @MxSent@+-- write directly to the event bus (using STM) in the calling (green) thread,+-- before dispatching instructions to the node controller to perform the+-- necessary routing to deliver the message to a process (or registered name,+-- or typed channel) locally or remotely.+-- -- [Management Data API] -- -- Both management agents and clients of the API have access to a variety of@@ -143,8 +177,6 @@ -- -- > monitorNames = getSelfPid >>= nsend "name-monitor" -- > monitorNames2 = getSelfPid >>= mxNotify------ For some real-world examples, see the distributed-process-platform package. -- -- [Performance, Stablity and Scalability] --

@@ -33,6 +33,7 @@   ( whereis   , newChan   , receiveChan+  , die   ) import Control.Distributed.Process.Management.Internal.Trace.Types   ( TraceArg(..)@@ -168,6 +169,11 @@   withLocalTracer $ \t -> liftIO $ Tracer.getCurrentTraceClient t sp   currentTracer <- receiveChan rp   case currentTracer of-    Nothing  -> do { (Just p') <- whereis "tracer.initial"; act p' }+    Nothing  -> do mTP <- whereis "tracer.initial"+                   -- NB: this should NOT ever happen, but forcing pattern matches+                   --     is not considered cool in later versions of MonadFail+                   case mTP of+                     Just p' -> act p'+                     Nothing -> die $ "System Invariant Violation: Tracer Process "+                                   ++ "Name Not Found (whereis tracer.initial)"     (Just p) -> act p-

@@ -14,13 +14,13 @@   , Addressable(..)   ) where -import Control.Applicative (Applicative) import Control.Concurrent.STM   ( TChan   ) import Control.Distributed.Process.Internal.Types   ( Process   , ProcessId+  , SendPortId   , Message   , DiedReason   , NodeId@@ -30,6 +30,7 @@   ( MonadState   , StateT   )+import Control.Monad.Fix (MonadFix) import Data.Binary import Data.Typeable (Typeable) import GHC.Generics@@ -56,8 +57,14 @@     -- ^ fired whenever a node /dies/ (i.e., the connection is broken/disconnected)   | MxSent             ProcessId    ProcessId Message     -- ^ fired whenever a message is sent from a local process+  | MxSentToName       String       ProcessId Message+    -- ^ fired whenever a named send occurs+  | MxSentToPort       ProcessId    SendPortId Message+    -- ^ fired whenever a sendChan occurs   | MxReceived         ProcessId    Message     -- ^ fired whenever a message is received by a local process+  | MxReceivedPort     SendPortId   Message+    -- ^ fired whenever a message is received via a typed channel   | MxConnected        ConnectionId EndPointAddress     -- ^ fired when a network-transport connection is first established   | MxDisconnected     ConnectionId EndPointAddress@@ -112,6 +119,7 @@   } deriving ( Functor              , Monad              , MonadIO+             , MonadFix              , ST.MonadState (MxAgentState s)              , Typeable              , Applicative

@@ -30,7 +30,7 @@   ( empty   , toList   , fromList-  , filter+  , partition   , partitionWithKey   , elems   , size@@ -183,6 +183,9 @@ import Control.Distributed.Process.Management.Internal.Agent   ( mxAgentController   )+import Control.Distributed.Process.Management.Internal.Types+  ( MxEvent(..)+  ) import qualified Control.Distributed.Process.Management.Internal.Trace.Remote as Trace   ( remoteTable   )@@ -195,9 +198,6 @@   , traceLogFmt   , enableTrace   )-import Control.Distributed.Process.Management.Internal.Types-  ( MxEvent(..)-  ) import Control.Distributed.Process.Serializable (Serializable) import Control.Distributed.Process.Internal.Messaging   ( sendBinary@@ -486,7 +486,7 @@ data IncomingTarget =     Uninit   | ToProc ProcessId (Weak (CQueue Message))-  | ToChan TypedChannel+  | ToChan SendPortId TypedChannel   | ToNode  data ConnectionState = ConnectionState {@@ -541,13 +541,17 @@                 enqueue queue msg -- 'enqueue' is strict                 trace node (MxReceived pid msg)               go st-            Just (_, ToChan (TypedChannel chan')) -> do+            Just (_, ToChan chId (TypedChannel chan')) -> do               mChan <- deRefWeak chan'               -- If mChan is Nothing, the process has given up the read end of               -- the channel and we simply ignore the incoming message-              forM_ mChan $ \chan -> atomically $-                -- We make sure the message is fully decoded when it is enqueued-                writeTQueue chan $! decode (BSL.fromChunks payload)+              forM_ mChan $ \chan -> do+                msg' <- atomically $ do+                  msg <- return $! decode (BSL.fromChunks payload)+                  -- We make sure the message is fully decoded when it is enqueued+                  writeTQueue chan msg+                  return msg+                trace node $ MxReceivedPort chId $ unsafeCreateUnencodedMessage msg'               go st             Just (_, ToNode) -> do               let ctrlMsg = decode . BSL.fromChunks $ payload@@ -574,7 +578,7 @@                       mChannel <- withMVar (processState proc) $ return . (^. typedChannelWithId lcid)                       case mChannel of                         Just channel ->-                          go (incomingAt cid ^= Just (src, ToChan channel) $ st)+                          go (incomingAt cid ^= Just (src, ToChan chId channel) $ st)                         Nothing ->                           invalidRequest cid st $                             "incoming attempt to connect to unknown channel of"@@ -903,7 +907,11 @@    modify' $ (links ^= unaffectedLinks') . (monitors ^= unaffectedMons') -  modify' $ registeredHere ^: Map.filter (\pid -> not $ ident `impliesDeathOf` ProcessIdentifier pid)+  -- we now consider all labels for this identifier unregistered+  let toDrop pid = not $ ident `impliesDeathOf` ProcessIdentifier pid+  (keepNames, dropNames) <- Map.partition toDrop <$> gets (^. registeredHere)+  mapM_ (\(p, l) -> liftIO $ trace node (MxUnRegistered l p)) (Map.toList dropNames)+  modify' $ registeredHere ^= keepNames    remaining <- fmap Map.toList (gets (^. registeredOnNodes)) >>=       mapM (\(pid,nidlist) ->@@ -955,7 +963,11 @@                do modify' $ registeredHereFor label ^= mPid                   updateRemote node currentVal mPid                   case mPid of-                    (Just p) -> liftIO $ trace node (MxRegistered p label)+                    (Just p) -> do+                      if reregistration+                        then liftIO $ trace node (MxUnRegistered (fromJust currentVal) label)+                        else return ()+                      liftIO $ trace node (MxRegistered p label)                     Nothing  -> liftIO $ trace node (MxUnRegistered (fromJust currentVal) label)              newVal <- gets (^. registeredHereFor label)              ncSendToProcess from $ unsafeCreateUnencodedMessage $@@ -1036,11 +1048,12 @@         -- If ch is Nothing, the process has given up the read end of         -- the channel and we simply ignore the incoming message - this         ch <- deRefWeak chan'-        forM_ ch $ \chan -> deliverChan msg chan-  where deliverChan :: forall a . Message -> TQueue a -> IO ()-        deliverChan (UnencodedMessage _ raw) chan' =+        forM_ ch $ \chan -> deliverChan node from msg chan+  where deliverChan :: forall a . LocalNode -> SendPortId -> Message -> TQueue a -> IO ()+        deliverChan n p (UnencodedMessage _ raw) chan' = do             atomically $ writeTQueue chan' ((unsafeCoerce raw) :: a)-        deliverChan (EncodedMessage   _ _) _ =+            trace n (MxReceivedPort p $ unsafeCreateUnencodedMessage raw)+        deliverChan _ _ (EncodedMessage   _ _) _ =             -- this will not happen unless someone screws with Primitives.hs             error "invalid local channel delivery" @@ -1070,7 +1083,7 @@   case mProc of     Nothing   -> dispatch (isLocal node (ProcessIdentifier from))                           from (ProcessInfoNone DiedUnknownId)-    Just proc    -> do+    Just proc -> do       itsLinks    <- Set.map fst . BiMultiMap.lookupBy1st them <$>                        gets (^. links)       itsMons     <- BiMultiMap.lookupBy1st them <$> gets (^. monitors)@@ -1084,8 +1097,8 @@                    infoNode               = (processNodeId pid)                  , infoRegisteredNames    = reg                  , infoMessageQueueLength = size-                 , infoMonitors       = Set.toList itsMons-                 , infoLinks          = Set.toList itsLinks+                 , infoMonitors           = Set.toList itsMons+                 , infoLinks              = Set.toList itsLinks                  }   where dispatch :: (Serializable a)                  => Bool

@@ -1,4 +1,5 @@ {-# LANGUAGE DeriveDataTypeable  #-}+{-# LANGUAGE ConstraintKinds  #-} {-# LANGUAGE UndecidableInstances  #-} {-# LANGUAGE FlexibleInstances  #-} {-# LANGUAGE ScopedTypeVariables  #-}@@ -40,8 +41,7 @@   deriving (Typeable)  -- | Objects that can be sent across the network-class (Binary a, Typeable a) => Serializable a-instance (Binary a, Typeable a) => Serializable a+type Serializable a = (Binary a, Typeable a)  -- | Encode type representation as a bytestring encodeFingerprint :: Fingerprint -> ByteString

@@ -32,8 +32,25 @@ -- the /normal/ strategy). -- -- Use of the functions in this module can potentially change the runtime--- behaviour of your application. You have been warned!+-- behaviour of your application. In addition, messages passed between Cloud+-- Haskell processes are written to a tracing infrastructure on the local node,+-- to provide improved introspection and debugging facilities for complex actor+-- based systems. This module makes no attempt to force evaluation in these+-- cases either, thus evaluation problems in passed data structures could not+-- only crash your processes, but could also bring down critical internal+-- services on which the node relies to function correctly. --+-- If you wish to repudiate such issues, you are advised to consider the use+-- of NFSerialisable in the distributed-process-extras package, which type+-- class brings NFData into scope along with Serializable, such that we can+-- force evaluation. Intended for use with modules such as this one, this+-- approach guarantees correct evaluatedness in terms of @NFData@. Please note+-- however, that we /cannot/ guarantee that an @NFData@ instance will behave the+-- same way as a @Binary@ one with regards evaluation, so it is still possible+-- to introduce unexpected behaviour by using /unsafe/ primitives in this way.+--+-- You have been warned!+-- -- This module is exported so that you can replace the use of Cloud Haskell's -- /safe/ messaging primitives. If you want to use both variants, then you can -- take advantage of qualified imports, however "Control.Distributed.Process"@@ -55,7 +72,12 @@   , sendBinary   , sendCtrlMsg   )-+import Control.Distributed.Process.Management.Internal.Types+  ( MxEvent(..)+  )+import Control.Distributed.Process.Management.Internal.Trace.Types+  ( traceEvent+  ) import Control.Distributed.Process.Internal.Types   ( ProcessId(..)   , NodeId(..)@@ -79,34 +101,48 @@  -- | Named send to a process in the local registry (asynchronous) nsend :: Serializable a => String -> a -> Process ()-nsend label msg =-  sendCtrlMsg Nothing (NamedSend label (unsafeCreateUnencodedMessage msg))+nsend label msg = do+  proc <- ask+  let us = processId proc+  let msg' = wrapMessage msg+  -- see [note: tracing]+  liftIO $ traceEvent (localEventBus (processNode proc))+                      (MxSentToName label us msg')+  sendCtrlMsg Nothing (NamedSend label msg')  -- | Named send to a process in a remote registry (asynchronous) nsendRemote :: Serializable a => NodeId -> String -> a -> Process () nsendRemote nid label msg = do   proc <- ask-  if localNodeId (processNode proc) == nid+  let us = processId proc+  let node = processNode proc+  if localNodeId node == nid     then nsend label msg-    else sendCtrlMsg (Just nid) (NamedSend label (createMessage msg))+    else+      let lbl = label ++ "@" ++ show nid in do+        -- see [note: tracing] NB: this is a remote call to another NC...+        liftIO $ traceEvent (localEventBus node)+                            (MxSentToName lbl us (wrapMessage msg))+        sendCtrlMsg (Just nid) (NamedSend label (createMessage msg))  -- | Send a message send :: Serializable a => ProcessId -> a -> Process () send them msg = do   proc <- ask   let node     = localNodeId (processNode proc)-      destNode = (processNodeId them) in do-  case destNode == node of-    True  -> unsafeSendLocal them msg-    False -> liftIO $ sendMessage (processNode proc)-                                  (ProcessIdentifier (processId proc))-                                  (ProcessIdentifier them)-                                  NoImplicitReconnect-                                  msg-  where-    unsafeSendLocal :: (Serializable a) => ProcessId -> a -> Process ()-    unsafeSendLocal pid msg' =-      sendCtrlMsg Nothing $ LocalSend pid (unsafeCreateUnencodedMessage msg')+      destNode = (processNodeId them)+      us       = (processId proc)+      msg'     = wrapMessage msg in do+    -- see [note: tracing]+    liftIO $ traceEvent (localEventBus (processNode proc))+                        (MxSent them us msg')+    if destNode == node+      then sendCtrlMsg Nothing $ LocalSend them msg'+      else liftIO $ sendMessage (processNode proc)+                                (ProcessIdentifier (processId proc))+                                (ProcessIdentifier them)+                                NoImplicitReconnect+                                msg  -- | Send a message unreliably. --@@ -121,30 +157,44 @@ usend them msg = do     proc <- ask     let there = processNodeId them+    let (us, node) = (processId proc, processNode proc)+    let msg' = wrapMessage msg+    -- see [note: tracing]+    liftIO $ traceEvent (localEventBus node) (MxSent them us msg')     if localNodeId (processNode proc) == there-      then sendCtrlMsg Nothing $-             LocalSend them (unsafeCreateUnencodedMessage msg)+      then sendCtrlMsg Nothing $ LocalSend them msg'       else sendCtrlMsg (Just there) $ UnreliableSend (processLocalId them)                                                      (createMessage msg) +-- [note: tracing]+-- Note that tracing writes to the local node's control channel, and this+-- module explicitly specifies to its clients that it does unsafe message+-- encoding. The same is true for the messages it puts onto the Management+-- event bus, however we do *not* want unevaluated thunks hitting the event+-- bus control thread. Hence the word /Unsafe/ in this module's name!+--+ -- | Send a message on a typed channel sendChan :: Serializable a => SendPort a -> a -> Process () sendChan (SendPort cid) msg = do   proc <- ask-  let node     = localNodeId (processNode proc)-      destNode = processNodeId (sendPortProcessId cid) in do-  case destNode == node of-    True  -> unsafeSendChanLocal cid msg-    False -> do-      liftIO $ sendBinary (processNode proc)-                          (ProcessIdentifier (processId proc))-                          (SendPortIdentifier cid)-                          NoImplicitReconnect-                          msg+  let+    node = processNode proc+    pid  = processId proc+    us   = localNodeId node+    them = processNodeId (sendPortProcessId cid)+    msg' = wrapMessage msg+  liftIO $ traceEvent (localEventBus node) (MxSentToPort pid cid msg')+  if them == us+    then unsafeSendChanLocal cid msg' -- NB: we wrap to P.Message !!!+    else liftIO $ sendBinary node+                             (ProcessIdentifier pid)+                             (SendPortIdentifier cid)+                             NoImplicitReconnect+                             msg   where-    unsafeSendChanLocal :: (Serializable a) => SendPortId -> a -> Process ()-    unsafeSendChanLocal spId msg' =-      sendCtrlMsg Nothing $ LocalPortSend spId (unsafeCreateUnencodedMessage msg')+    unsafeSendChanLocal :: SendPortId -> Message -> Process ()+    unsafeSendChanLocal p m = sendCtrlMsg Nothing $ LocalPortSend p m  -- | Create an unencoded @Message@ for any @Serializable@ type. wrapMessage :: Serializable a => a -> Message