diff --git a/ChangeLog b/ChangeLog --- a/ChangeLog +++ b/ChangeLog @@ -1,3 +1,50 @@ +2025-02-04 Laurent P. René de Cotret 0.7.8 + +* Added documentation on the unit of measurement for timeout durations (#340) +* Added upper bound on `template-haskell` to prevent future breakage. +* Addressed some compilation warnings (#467) + +2024-09-03 Laurent P. René de Cotret 0.7.7 + +* Bumped dependency bounds to support GHC 8.10.7 - GHC 9.10.1 +* Updated links to point to Distributed Haskell monorepo + +2024-04-03 David Simmons-Duffin 0.7.6 + +* Bumped hashable upper bound. + +2018-06-12 David Simmons-Duffin 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 0.7.4 + +* Added support for exceptions >= 0.10 + +2017-08-31 Facundo Domínguez 0.7.3 + +* Drop support for ghc-7.8.* and earlier. + +2017-08-31 Facundo Domínguez 0.7.2 + +* Fixed build errors with ghc-8.2.1. + +2017-08-22 Facundo Domínguez 0.7.1 + +* Relax upper bounds in dependencies to build with ghc-8.2.1. + +2017-08-21 Facundo Domínguez 0.7.0 + +* Change type of message sent by `say` from a 3-tuple to a proper +type (`SayMessage`) with a proper `UTCTime`. (#291) +* Expose the MonitorRef in the withMonitor call. +* Have unmonitor remove the monitor message in the inbox. (#268) +* Remove Mx Data Tables. This API isn't used, is easy to replace with various +other packages. (#276) +* Add Ord instance to SpawnRef. + 2016-10-13 Facundo Domínguez 0.6.6 * Remove monitors from remote nodes when a process dies. (#295) diff --git a/Setup.hs b/Setup.hs deleted file mode 100644 --- a/Setup.hs +++ /dev/null @@ -1,2 +0,0 @@ -import Distribution.Simple -main = defaultMain diff --git a/benchmarks/Channels.hs b/benchmarks/Channels.hs --- a/benchmarks/Channels.hs +++ b/benchmarks/Channels.hs @@ -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,6 +36,8 @@ main :: IO () main = do [role, host, port] <- getArgs - Right transport <- createTransport 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 diff --git a/benchmarks/Latency.hs b/benchmarks/Latency.hs --- a/benchmarks/Latency.hs +++ b/benchmarks/Latency.hs @@ -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,6 +31,9 @@ main :: IO () main = do [role, host, port] <- getArgs - Right transport <- createTransport 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 diff --git a/benchmarks/ProcessRing.hs b/benchmarks/ProcessRing.hs --- a/benchmarks/ProcessRing.hs +++ b/benchmarks/ProcessRing.hs @@ -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 @@ -110,7 +110,8 @@ argv <- getArgs (opt, _) <- parseArgv argv putStrLn $ "options: " ++ (show opt) - Right transport <- createTransport "127.0.0.1" "8090" defaultTCPParameters + Right transport <- createTransport + (defaultTCPAddr "127.0.0.1" "8090" ) defaultTCPParameters node <- newLocalNode transport initRemoteTable catch (void $ runProcess node $ initialProcess opt) (\(e :: SomeException) -> putStrLn $ "ERROR: " ++ (show e)) diff --git a/benchmarks/Spawns.hs b/benchmarks/Spawns.hs --- a/benchmarks/Spawns.hs +++ b/benchmarks/Spawns.hs @@ -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,6 +42,8 @@ main :: IO () main = do [role, host, port] <- getArgs - Right transport <- createTransport 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 diff --git a/benchmarks/Throughput.hs b/benchmarks/Throughput.hs --- a/benchmarks/Throughput.hs +++ b/benchmarks/Throughput.hs @@ -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,6 +67,8 @@ main :: IO () main = do [role, host, port] <- getArgs - Right transport <- createTransport 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 diff --git a/distributed-process.cabal b/distributed-process.cabal --- a/distributed-process.cabal +++ b/distributed-process.cabal @@ -1,155 +1,176 @@ +cabal-version: 3.0 Name: distributed-process -Version: 0.6.6 -Cabal-Version: >=1.8 +Version: 0.7.8 Build-Type: Simple -License: BSD3 +License: BSD-3-Clause License-File: LICENSE Copyright: Well-Typed LLP, Tweag I/O Limited Author: Duncan Coutts, Nicolas Wu, Edsko de Vries -Maintainer: Facundo Domínguez +maintainer: The Distributed Haskell team Stability: experimental -Homepage: http://haskell-distributed.github.com/ +Homepage: https://haskell-distributed.github.io/ Bug-Reports: https://github.com/haskell-distributed/distributed-process/issues Synopsis: Cloud Haskell: Erlang-style concurrency in Haskell Description: This is an implementation of Cloud Haskell, as described in /Towards Haskell in the Cloud/ by Jeff Epstein, Andrew Black, and Simon Peyton Jones - (), + (), although some of the details are different. The precise message passing semantics are based on /A unified semantics for future Erlang/ by Hans Svensson, Lars-Åke Fredlund and Clara Benac Earle. You will probably also want to install a Cloud Haskell backend such as distributed-process-simplelocalnet. -Tested-With: GHC==7.2.2 GHC==7.4.1 GHC==7.4.2 GHC==7.6.2 +tested-with: GHC==8.10.7 GHC==9.0.2 GHC==9.2.8 GHC==9.4.5 GHC==9.6.4 GHC==9.8.2 GHC==9.10.1 GHC==9.12.1 Category: Control -extra-source-files: ChangeLog +extra-doc-files: ChangeLog -Source-Repository head +common warnings + ghc-options: -Wall + -Wcompat + -Widentities + -Wincomplete-uni-patterns + -Wincomplete-record-updates + -Wredundant-constraints + -fhide-source-paths + -Wpartial-fields + -Wunused-packages + +source-repository head Type: git Location: https://github.com/haskell-distributed/distributed-process - SubDir: distributed-process + SubDir: packages/distributed-process flag th description: Build with Template Haskell support default: True -flag old-locale - description: If false then depend on time >= 1.5. - . - If true then depend on time < 1.5 together with old-locale. - default: False - Library - Build-Depends: base >= 4.4 && < 5, - 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.6, + import: warnings + Build-Depends: base >= 4.14 && < 5, + binary >= 0.8 && < 0.10, + hashable >= 1.2.0.5 && < 1.6, + network-transport >= 0.4.1.0 && < 0.6, + stm >= 2.4 && < 2.6, mtl >= 2.0 && < 2.4, data-accessor >= 0.2 && < 0.3, - bytestring >= 0.9 && < 0.11, - random >= 1.0 && < 1.2, + bytestring >= 0.10 && < 0.13, + random >= 1.0 && < 1.4, distributed-static >= 0.2 && < 0.4, - rank1dynamic >= 0.1 && < 0.4, - syb >= 0.3 && < 0.7, - exceptions >= 0.5 - Exposed-modules: Control.Distributed.Process, - Control.Distributed.Process.Closure, - Control.Distributed.Process.Debug, - Control.Distributed.Process.Internal.BiMultiMap, - Control.Distributed.Process.Internal.Closure.BuiltIn, - Control.Distributed.Process.Internal.Closure.Explicit, - Control.Distributed.Process.Internal.CQueue, - Control.Distributed.Process.Internal.Messaging, - Control.Distributed.Process.Internal.Primitives, - Control.Distributed.Process.Internal.Spawn, - Control.Distributed.Process.Internal.StrictContainerAccessors, - Control.Distributed.Process.Internal.StrictList, - Control.Distributed.Process.Internal.StrictMVar, - Control.Distributed.Process.Internal.Types, - Control.Distributed.Process.Internal.WeakTQueue, - Control.Distributed.Process.Management, - Control.Distributed.Process.Node, - Control.Distributed.Process.Serializable, + rank1dynamic >= 0.1 && < 0.5, + syb >= 0.3 && < 0.8, + exceptions >= 0.10, + containers >= 0.6 && < 0.8, + deepseq >= 1.4 && < 1.7, + time >= 1.9 + Exposed-modules: Control.Distributed.Process + Control.Distributed.Process.Closure + Control.Distributed.Process.Debug + Control.Distributed.Process.Internal.BiMultiMap + Control.Distributed.Process.Internal.Closure.BuiltIn + Control.Distributed.Process.Internal.Closure.Explicit + Control.Distributed.Process.Internal.CQueue + Control.Distributed.Process.Internal.Messaging + Control.Distributed.Process.Internal.Primitives + Control.Distributed.Process.Internal.Spawn + Control.Distributed.Process.Internal.StrictContainerAccessors + Control.Distributed.Process.Internal.StrictList + Control.Distributed.Process.Internal.StrictMVar + Control.Distributed.Process.Internal.Types + Control.Distributed.Process.Internal.WeakTQueue + Control.Distributed.Process.Management + Control.Distributed.Process.Node + Control.Distributed.Process.Serializable Control.Distributed.Process.UnsafePrimitives - Control.Distributed.Process.Management.Internal.Agent, - Control.Distributed.Process.Management.Internal.Bus, - Control.Distributed.Process.Management.Internal.Table, - Control.Distributed.Process.Management.Internal.Types, - Control.Distributed.Process.Management.Internal.Trace.Primitives, - Control.Distributed.Process.Management.Internal.Trace.Remote, - Control.Distributed.Process.Management.Internal.Trace.Types, + Control.Distributed.Process.Management.Internal.Agent + Control.Distributed.Process.Management.Internal.Bus + Control.Distributed.Process.Management.Internal.Types + Control.Distributed.Process.Management.Internal.Trace.Primitives + Control.Distributed.Process.Management.Internal.Trace.Remote + Control.Distributed.Process.Management.Internal.Trace.Types Control.Distributed.Process.Management.Internal.Trace.Tracer - ghc-options: -Wall + default-language: Haskell2010 HS-Source-Dirs: src - if impl(ghc <= 7.4.2) - Build-Depends: containers >= 0.4 && < 0.5, - deepseq == 1.3.0.0 - else - Build-Depends: containers >= 0.4 && < 0.6, - deepseq >= 1.3.0.1 && < 1.6 - if flag(old-locale) - Build-Depends: time < 1.5, old-locale >= 1.0 && <1.1 - else - Build-Depends: time >= 1.5 + other-extensions: BangPatterns + CPP + DeriveDataTypeable + DeriveFunctor + DeriveGeneric + ExistentialQuantification + FlexibleInstances + GADTs + GeneralizedNewtypeDeriving + KindSignatures + MagicHash + PatternGuards + RankNTypes + RecordWildCards + ScopedTypeVariables + StandaloneDeriving + TypeFamilies + TypeSynonymInstances + UnboxedTuples + UndecidableInstances if flag(th) - if impl(ghc <= 7.4.2) - Build-Depends: template-haskell >= 2.7 && < 2.8 - else - Build-Depends: template-haskell >= 2.6 && < 2.12 + other-extensions: TemplateHaskell + Build-Depends: template-haskell >= 2.6 && <2.24 Exposed-modules: Control.Distributed.Process.Internal.Closure.TH CPP-Options: -DTemplateHaskellSupport -- Tests are in distributed-process-test package, for convenience. benchmark distributed-process-throughput - Type: exitcode-stdio-1.0 - Build-Depends: base >= 4.4 && < 5, - distributed-process, - network-transport-tcp >= 0.3 && < 0.6, - bytestring >= 0.9 && < 0.11, - binary >= 0.6.3 && < 0.9 - Main-Is: benchmarks/Throughput.hs - ghc-options: -Wall + import: warnings + Type: exitcode-stdio-1.0 + Build-Depends: base >= 4.14 && < 5, + distributed-process, + network-transport-tcp >= 0.3 && <= 0.9, + bytestring >= 0.10 && < 0.13, + binary >= 0.8 && < 0.10 + Main-Is: benchmarks/Throughput.hs + default-language: Haskell2010 benchmark distributed-process-latency - Type: exitcode-stdio-1.0 - Build-Depends: base >= 4.4 && < 5, - distributed-process, - network-transport-tcp >= 0.3 && < 0.6, - bytestring >= 0.9 && < 0.11, - binary >= 0.6.3 && < 0.9 - Main-Is: benchmarks/Latency.hs - ghc-options: -Wall + import: warnings + Type: exitcode-stdio-1.0 + Build-Depends: base >= 4.14 && < 5, + distributed-process, + network-transport-tcp >= 0.3 && <= 0.9, + bytestring >= 0.10 && < 0.13, + binary >= 0.8 && < 0.10 + Main-Is: benchmarks/Latency.hs + default-language: Haskell2010 benchmark distributed-process-channels - Type: exitcode-stdio-1.0 - Build-Depends: base >= 4.4 && < 5, - distributed-process, - network-transport-tcp >= 0.3 && < 0.6, - bytestring >= 0.9 && < 0.11, - binary >= 0.6.3 && < 0.9 - Main-Is: benchmarks/Channels.hs - ghc-options: -Wall + import: warnings + Type: exitcode-stdio-1.0 + Build-Depends: base >= 4.14 && < 5, + distributed-process, + network-transport-tcp >= 0.3 && <= 0.9, + bytestring >= 0.10 && < 0.13, + binary >= 0.8 && < 0.10 + Main-Is: benchmarks/Channels.hs + default-language: Haskell2010 benchmark distributed-process-spawns - Type: exitcode-stdio-1.0 - Build-Depends: base >= 4.4 && < 5, - distributed-process, - network-transport-tcp >= 0.3 && < 0.6, - bytestring >= 0.9 && < 0.11, - binary >= 0.6.3 && < 0.9 - Main-Is: benchmarks/Spawns.hs - ghc-options: -Wall + import: warnings + Type: exitcode-stdio-1.0 + Build-Depends: base >= 4.14 && < 5, + distributed-process, + network-transport-tcp >= 0.3 && <= 0.9, + bytestring >= 0.10 && < 0.13, + binary >= 0.8 && < 0.10 + Main-Is: benchmarks/Spawns.hs + default-language: Haskell2010 benchmark distributed-process-ring - Type: exitcode-stdio-1.0 - Build-Depends: base >= 4.4 && < 5, - distributed-process, - network-transport-tcp >= 0.3 && < 0.6, - bytestring >= 0.9 && < 0.11, - binary >= 0.6.3 && < 0.9 - Main-Is: benchmarks/ProcessRing.hs - ghc-options: -Wall -threaded -O2 -rtsopts + import: warnings + Type: exitcode-stdio-1.0 + Build-Depends: base >= 4.14 && < 5, + distributed-process, + network-transport-tcp >= 0.3 && <= 0.9, + bytestring >= 0.10 && < 0.13, + binary >= 0.8 && < 0.10 + Main-Is: benchmarks/ProcessRing.hs + default-language: Haskell2010 + ghc-options: -threaded -O2 -rtsopts 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 @@ -102,6 +102,7 @@ , monitorPort , unmonitor , withMonitor + , withMonitor_ , MonitorRef -- opaque , ProcessLinkException(..) , NodeLinkException(..) @@ -269,6 +270,7 @@ , monitorPort , unmonitor , withMonitor + , withMonitor_ -- Logging , say -- Registry @@ -320,11 +322,7 @@ ) import qualified Control.Monad.Catch as Catch -#if MIN_VERSION_base(4,6,0) import Prelude -#else -import Prelude hiding (catch) -#endif import qualified Control.Exception as Exception (onException) import Data.Accessor ((^.)) import Data.Foldable (forM_) diff --git a/src/Control/Distributed/Process/Internal/CQueue.hs b/src/Control/Distributed/Process/Internal/CQueue.hs --- a/src/Control/Distributed/Process/Internal/CQueue.hs +++ b/src/Control/Distributed/Process/Internal/CQueue.hs @@ -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#) @@ -74,7 +72,7 @@ data BlockSpec = NonBlocking | Blocking - | Timeout Int + | Timeout Int -- ^ Timeout in microseconds -- Match operations -- @@ -290,11 +288,7 @@ -- | 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 diff --git a/src/Control/Distributed/Process/Internal/Closure/Explicit.hs b/src/Control/Distributed/Process/Internal/Closure/Explicit.hs --- a/src/Control/Distributed/Process/Internal/Closure/Explicit.hs +++ b/src/Control/Distributed/Process/Internal/Closure/Explicit.hs @@ -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 diff --git a/src/Control/Distributed/Process/Internal/Closure/TH.hs b/src/Control/Distributed/Process/Internal/Closure/TH.hs --- a/src/Control/Distributed/Process/Internal/Closure/TH.hs +++ b/src/Control/Distributed/Process/Internal/Closure/TH.hs @@ -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 -- 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 @@ -76,8 +76,11 @@ , unlink , monitor , unmonitor + , unmonitorAsync , withMonitor + , withMonitor_ -- * Logging + , SayMessage(..) , say -- * Registry , register @@ -121,20 +124,13 @@ , sendCtrlMsg ) where -#if ! MIN_VERSION_base(4,6,0) -import Prelude hiding (catch) -#endif - -import Data.Binary (decode) -import Data.Time.Clock (getCurrentTime) +import Data.Binary (Binary(..), Put, Get, decode) +import Data.Time.Clock (getCurrentTime, UTCTime(..)) +import Data.Time.Calendar (Day(..)) import Data.Time.Format (formatTime) -#if MIN_VERSION_time(1,5,0) import Data.Time.Format (defaultTimeLocale) -#else -import System.Locale (defaultTimeLocale) -#endif import System.Timeout (timeout) -import Control.Monad (when) +import Control.Monad (when, void) import Control.Monad.Reader (ask) import Control.Monad.IO.Class (liftIO) import Control.Monad.Catch @@ -184,6 +180,7 @@ , SpawnRef(..) , ProcessSignal(..) , NodeMonitorNotification(..) + , ProcessMonitorNotification(..) , monitorCounter , spawnCounter , SendPort(..) @@ -248,18 +245,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 @@ -279,9 +274,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) @@ -301,7 +299,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 @@ -329,13 +333,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 @@ -396,15 +403,20 @@ 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. -- -- If the timeout is zero do a non-blocking check for matching messages. A -- non-zero timeout is applied only when waiting for incoming messages (that is, -- /after/ we have checked the messages that are already in the mailbox). -receiveTimeout :: Int -> [Match b] -> Process (Maybe b) +receiveTimeout :: Int -- ^ Timeout in microseconds + -> [Match b] + -> Process (Maybe b) receiveTimeout t ms = do queue <- processQueue <$> ask let blockSpec = if t == 0 then NonBlocking else Timeout t @@ -475,18 +487,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'. -- @@ -499,15 +508,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 @@ -885,13 +890,20 @@ -- @withMonitor@ returns, there might still be unreceived monitor -- messages in the queue. -- -withMonitor :: ProcessId -> Process a -> Process a -withMonitor pid code = bracket (monitor pid) unmonitor (\_ -> code) - -- unmonitor blocks waiting for the response, so there's a possibility - -- that an exception might interrupt withMonitor before the unmonitor - -- has completed. I think that's better than making the unmonitor - -- uninterruptible. +withMonitor :: ProcessId -> (MonitorRef -> Process a) -> Process a +withMonitor pid = bracket (monitor pid) unmonitor +-- | Establishes temporary monitoring of another process. +-- +-- @withMonitor_ pid code@ sets up monitoring of @pid@ for the duration +-- of @code@. Note: although monitoring is no longer active when +-- @withMonitor_@ returns, there might still be unreceived monitor +-- messages in the queue. +-- +-- Since 0.6.1 +withMonitor_ :: ProcessId -> Process a -> Process a +withMonitor_ p = withMonitor p . const + -- | Remove a link -- -- This is synchronous in the sense that once it returns you are guaranteed @@ -928,12 +940,21 @@ -- | Remove a monitor -- -- This has the same synchronous/asynchronous nature as 'unlink'. +-- +-- ProcessMonitorNotification messages for the given MonitorRef are removed from +-- the mailbox. unmonitor :: MonitorRef -> Process () unmonitor ref = do unmonitorAsync ref receiveWait [ matchIf (\(DidUnmonitor ref') -> ref' == ref) (\_ -> return ()) ] + -- Discard the notification if any. With the current NC implementation at most + -- one notification is in the mailbox for any given ref. + void $ receiveTimeout 0 + [ matchIf (\(ProcessMonitorNotification ref' _ _) -> ref' == ref) + (const $ return ()) + ] -------------------------------------------------------------------------------- -- Exception handling -- @@ -947,7 +968,7 @@ -- | Lift 'Control.Exception.try' try :: Exception e => Process a -> Process (Either e a) try = Catch.try -{-# DEPRECATED try "Use Control.Monad.Catch.mask_ instead" #-} +{-# DEPRECATED try "Use Control.Monad.Catch.try instead" #-} -- | Lift 'Control.Exception.mask' mask :: ((forall a. Process a -> Process a) -> Process b) -> Process b @@ -1001,7 +1022,9 @@ -------------------------------------------------------------------------------- -- | Like 'expect' but with a timeout -expectTimeout :: forall a. Serializable a => Int -> Process (Maybe a) +expectTimeout :: forall a. Serializable a + => Int -- ^ Timeout in microseconds + -> Process (Maybe a) expectTimeout n = receiveTimeout n [match return] -- | Asynchronous version of 'spawn' @@ -1059,17 +1082,52 @@ -- Logging -- -------------------------------------------------------------------------------- +data SayMessage = SayMessage { sayTime :: UTCTime + , sayProcess :: ProcessId + , sayMessage :: String } + deriving (Typeable) + +-- There is sadly no Show UTCTime instance +instance Show SayMessage where + showsPrec p msg = + showParen (p >= 11) + $ showString "SayMessage " + . showString (formatTime defaultTimeLocale "%c" (sayTime msg)) + . showChar ' ' + . showsPrec 11 (sayProcess msg) . showChar ' ' + . showsPrec 11 (sayMessage msg) . showChar ' ' + +instance Binary SayMessage where + put s = do + putUTCTime (sayTime s) + put (sayProcess s) + put (sayMessage s) + get = SayMessage <$> getUTCTime <*> get <*> get + +-- Sadly there is no Binary UTCTime instance +putUTCTime :: UTCTime -> Put +putUTCTime (UTCTime (ModifiedJulianDay day) tod) = do + put day + put (toRational tod) + +getUTCTime :: Get UTCTime +getUTCTime = do + day <- get + tod <- get + return $! UTCTime (ModifiedJulianDay day) + (fromRational tod) + -- | Log a string -- --- @say message@ sends a message (time, pid of the current process, message) --- to the process registered as 'logger'. By default, this process simply --- sends the string to 'stderr'. Individual Cloud Haskell backends might --- replace this with a different logger process, however. +-- @say message@ sends a message of type 'SayMessage' with the current time and +-- 'ProcessId' of the current process to the process registered as @logger@. By +-- default, this process simply sends the string to @stderr@. Individual Cloud +-- Haskell backends might replace this with a different logger process, however. say :: String -> Process () say string = do now <- liftIO getCurrentTime us <- getSelfPid - nsend "logger" (formatTime defaultTimeLocale "%c" now, us, string) + nsend "logger" (SayMessage now us string) -------------------------------------------------------------------------------- -- Registry -- @@ -1165,8 +1223,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 @@ -1178,9 +1240,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 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 @@ -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,23 +63,12 @@ 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 -> -#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 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 @@ -353,6 +353,7 @@ deriving ( Applicative , Functor , Monad + , MonadFail , MonadFix , MonadIO , MonadReader LocalProcess @@ -366,6 +367,13 @@ lproc <- ask liftIO $ catch (runLocalProcess lproc p) (runLocalProcess lproc . h) instance MonadMask Process where + generalBracket acquire release inner = do + lproc <- ask + liftIO $ + generalBracket (runLocalProcess lproc acquire) + (\a e -> runLocalProcess lproc $ release a e) + (runLocalProcess lproc . inner) + mask p = do lproc <- ask liftIO $ mask $ \restore -> @@ -463,11 +471,7 @@ deriving (Typeable) instance NFData Message where -#if MIN_VERSION_bytestring(0,10,0) rnf (EncodedMessage _ e) = rnf e `seq` () -#else - rnf (EncodedMessage _ e) = BSL.length e `seq` () -#endif rnf (UnencodedMessage _ a) = e `seq` () where e = BSL.length (encode a) @@ -629,7 +633,7 @@ -- | 'SpawnRef' are used to return pids of spawned processes newtype SpawnRef = SpawnRef Int32 - deriving (Show, Binary, Typeable, Eq) + deriving (Show, Binary, Typeable, Eq, Ord) -- | (Asynchronius) reply from 'spawn' data DidSpawn = DidSpawn SpawnRef ProcessId diff --git a/src/Control/Distributed/Process/Internal/WeakTQueue.hs b/src/Control/Distributed/Process/Internal/WeakTQueue.hs --- a/src/Control/Distributed/Process/Internal/WeakTQueue.hs +++ b/src/Control/Distributed/Process/Internal/WeakTQueue.hs @@ -100,8 +100,4 @@ 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 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 @@ -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 @@ -83,28 +117,8 @@ -- system information. Agents maintain their own internal state privately (via a -- state transformer - see 'mxGetLocal' et al), however it is possible for -- agents to share additional data with each other (and the outside world) --- using /data tables/. --- --- Each agent is assigned its own data table, which acts as a shared map, where --- the keys are @String@s and the values are @Serializable@ datum of whatever --- type the agent or its clients stores. --- --- Because an agent's /data table/ stores its values in raw 'Message' format, --- it works effectively as an /un-typed dictionary/, into which data of varying --- types can be fed and later retrieved. The upside of this is that different --- keys can be mapped to various types without any additional work on the part --- of the developer. The downside is that the code reading these values must --- know in advance what type(s) to expect, and the API provides no additional --- support for handling that. --- --- Publishing is accomplished using the 'mxPublish' and 'mxSet' APIs, whilst --- querying and deletion are handled by 'mxGet', 'mxClear', 'mxPurgeTable' and --- 'mxDropTable' respectively. --- --- When a management agent terminates, their tables are left in memory despite --- termination, such that an agent may resume its role (by restarting) or have --- its 'MxAgentId' taken over by another subsequent agent, leaving the data --- originally captured in place. +-- using whatever mechanism the user wishes, e.g., acidstate, or shared memory +-- primitives. -- -- [Defining Agents] -- @@ -164,8 +178,6 @@ -- > monitorNames = getSelfPid >>= nsend "name-monitor" -- > monitorNames2 = getSelfPid >>= mxNotify -- --- For some real-world examples, see the distributed-process-platform package. --- -- [Performance, Stablity and Scalability] -- -- /Management Agents/ offer numerous advantages over regular processes: @@ -194,10 +206,6 @@ -- the event bus /and/ their own mailboxes, plus searching through the set of -- event sinks (for each agent) to determine the right handler for the event. -- --- Each management agent requires not only its own @Process@ (in which the agent --- code is run), but also a peer process that provides its /data table/. These --- data tables also have to be coordinated and manaaged on each agent's behalf. --- -- [Architecture Overview] -- -- The architecture of the management event bus is internal and subject to @@ -264,13 +272,6 @@ , mxGetLocal , mxUpdateLocal , liftMX - -- * Mx Data API - , mxPublish - , mxSet - , mxGet - , mxClear - , mxPurgeTable - , mxDropTable ) where import Control.Applicative @@ -281,10 +282,7 @@ , TChan ) import Control.Distributed.Process.Internal.Primitives - ( newChan - , nsend - , receiveWait - , matchChan + ( receiveWait , matchAny , matchSTM , unwrapMessage @@ -302,14 +300,12 @@ , unsafeCreateUnencodedMessage ) import Control.Distributed.Process.Management.Internal.Bus (publishEvent) -import qualified Control.Distributed.Process.Management.Internal.Table as Table import Control.Distributed.Process.Management.Internal.Types ( MxAgentId(..) , MxAgent(..) , MxAction(..) , ChannelSelector(..) , MxAgentState(..) - , MxAgentStart(..) , MxSink , MxEvent(..) ) @@ -335,42 +331,6 @@ bus <- localEventBus . processNode <$> ask liftIO $ publishEvent bus $ unsafeCreateUnencodedMessage msg --- | Publish an arbitrary @Message@ as a property in the management database. --- --- For publishing @Serializable@ data, use 'mxSet' instead. --- -mxPublish :: MxAgentId -> String -> Message -> Process () -mxPublish a k v = Table.set k v (Table.MxForAgent a) - --- | Sets an arbitrary @Serializable@ datum against a key in the management --- database. Note that /no attempt is made to force the argument/, therefore --- it is very important that you do not pass unevaluated thunks that might --- crash some other, arbitrary process (or management agent!) that obtains --- and attempts to force the value later on. --- -mxSet :: Serializable a => MxAgentId -> String -> a -> Process () -mxSet mxId key msg = do - Table.set key (unsafeCreateUnencodedMessage msg) (Table.MxForAgent mxId) - --- | Fetches a property from the management database for the given key. --- If the property is not set, or does not match the expected type when --- typechecked (at runtime), returns @Nothing@. -mxGet :: Serializable a => MxAgentId -> String -> Process (Maybe a) -mxGet = Table.fetch . Table.MxForAgent - --- | Clears a property from the management database using the given key. --- If the key does not exist in the database, this is a noop. -mxClear :: MxAgentId -> String -> Process () -mxClear mxId key = Table.clear key (Table.MxForAgent mxId) - --- | Purges a table in the management database of all its stored properties. -mxPurgeTable :: MxAgentId -> Process () -mxPurgeTable = Table.purge . Table.MxForAgent - --- | Deletes a table from the management database. -mxDropTable :: MxAgentId -> Process () -mxDropTable = Table.delete . Table.MxForAgent - -------------------------------------------------------------------------------- -- API for writing user defined management extensions (i.e., agents) -- -------------------------------------------------------------------------------- @@ -490,10 +450,7 @@ return pid where start (sendTChan, recvTChan) = do - (sp, rp) <- newChan - nsend Table.mxTableCoordinator (MxAgentStart sp mxId) - tablePid <- receiveWait [ matchChan rp (\(p :: ProcessId) -> return p) ] - let nState = MxAgentState mxId sendTChan tablePid initState + let nState = MxAgentState mxId sendTChan initState runAgent dtor handlers InputChan recvTChan nState runAgent :: MxAgent s () diff --git a/src/Control/Distributed/Process/Management/Internal/Table.hs b/src/Control/Distributed/Process/Management/Internal/Table.hs deleted file mode 100644 --- a/src/Control/Distributed/Process/Management/Internal/Table.hs +++ /dev/null @@ -1,222 +0,0 @@ -{-# LANGUAGE DeriveDataTypeable #-} -{-# LANGUAGE ScopedTypeVariables #-} -{-# LANGUAGE RankNTypes #-} -{-# LANGUAGE DeriveGeneric #-} -{-# LANGUAGE RecordWildCards #-} - -module Control.Distributed.Process.Management.Internal.Table - ( MxTableRequest(..) - , MxTableId(..) - , mxTableCoordinator - , startTableCoordinator - , delete - , purge - , clear - , set - , get - , fetch - ) where - -import Control.Distributed.Process.Internal.Primitives - ( receiveWait - , receiveChan - , match - , matchAny - , matchIf - , matchChan - , send - , nsend - , sendChan - , getSelfPid - , link - , monitor - , unwrapMessage - , newChan - , withMonitor - ) -import Control.Distributed.Process.Internal.Types - ( Process - , ProcessId - , ProcessMonitorNotification(..) - , SendPort - , ReceivePort - , Message - , unsafeCreateUnencodedMessage - ) -import Control.Distributed.Process.Management.Internal.Types - ( MxTableId(..) - , MxAgentId(..) - , MxAgentStart(..) - , Fork) -import Control.Distributed.Process.Serializable (Serializable) -import Control.Monad.IO.Class (liftIO) -import Data.Accessor (Accessor, accessor, (^=), (^:)) -import Data.Binary (Binary) -import Data.Map (Map) -import qualified Data.Map as Map -import Data.Typeable (Typeable) - -import GHC.Generics - --- An extremely lightweight shared Map implementation, for use --- by /management agents/ and their cohorts. Each agent is assigned --- a table, into which any serializable @Message@ can be inserted. --- Data are inserted, removed and searched for via their key, which --- is a string. Tables can be purged, values can be set, fetched or --- cleared/removed. --- - -data MxTableRequest = - Delete - | Purge - | Clear !String - | Set !String !Message - | Get !String !(SendPort (Maybe Message)) -- see [note: un-typed send port] - deriving (Typeable, Generic) -instance Binary MxTableRequest where - -data MxTableState = MxTableState { _name :: !String - , _entries :: !(Map String Message) - } - -type MxTables = Map MxAgentId ProcessId - -mxTableCoordinator :: String -mxTableCoordinator = "mx.table.coordinator" - -delete :: MxTableId -> Process () -delete = sendReq Delete - -purge :: MxTableId -> Process () -purge = sendReq Purge - -clear :: String -> MxTableId -> Process () -clear k = sendReq (Clear k) - -set :: String -> Message -> MxTableId -> Process () -set k v = sendReq (Set k v) - -fetch :: forall a. (Serializable a) - => MxTableId - -> String - -> Process (Maybe a) -fetch (MxForPid pid) key = get pid key -fetch mxId@(MxForAgent _) key = do - (sp, rp) <- newChan :: Process (SendPort (Maybe Message), - ReceivePort (Maybe Message)) - sendReq (Get key sp) mxId - receiveChan rp >>= maybe (return Nothing) - (unwrapMessage :: Message -> Process (Maybe a)) - --- [note: un-typed send port] --- Here, fetch uses a typed channel over a raw Message to obtain --- its result, so type checking is deferred until receipt and will --- be handled in the caller's thread. This is necessary because --- the server portion of the code knows nothing about the types --- involved, nor should it, since these tables can be used to --- store arbitrary serializable data. - -get :: forall a. (Serializable a) - => ProcessId - -> String - -> Process (Maybe a) -get pid key = do - safeFetch pid key >>= maybe (return Nothing) - (unwrapMessage :: Message -> Process (Maybe a)) - -safeFetch :: ProcessId -> String -> Process (Maybe Message) -safeFetch pid key = do - (sp, rp) <- newChan - send pid $ Get key sp - withMonitor pid $ do - receiveWait [ - matchChan rp return - , matchIf (\(ProcessMonitorNotification _ pid' _) -> pid' == pid) - (\_ -> return $ Just (unsafeCreateUnencodedMessage ())) - ] - -sendReq :: MxTableRequest -> MxTableId -> Process () -sendReq req tid = (resolve tid) req - -resolve :: Serializable a => MxTableId -> (a -> Process ()) -resolve (MxForAgent agent) = \msg -> nsend mxTableCoordinator (agent, msg) -resolve (MxForPid pid) = \msg -> send pid msg - -startTableCoordinator :: Fork -> Process () -startTableCoordinator fork = run Map.empty - where - run :: MxTables -> Process () - run tables = - receiveWait [ - -- note that this state change can race with MxAgentStart requests - match (\(ProcessMonitorNotification _ pid _) -> do - return $ Map.filter (/= pid) tables) - , match (\(MxAgentStart ch agent) -> do - lookupAgent tables agent >>= \(p, t) -> do - sendChan ch p >> return t) - , match (\req@(agent, tReq :: MxTableRequest) -> do - case tReq of - Get k sp -> do - lookupAgent tables agent >>= \(p, t) -> do - safeFetch p k >>= sendChan sp >> return t - _ -> do - handleRequest tables req) - , matchAny (\_ -> return tables) -- unrecognised messages are dropped - ] >>= run - - handleRequest :: MxTables - -> (MxAgentId, MxTableRequest) - -> Process MxTables - handleRequest tables' (agent, req) = do - lookupAgent tables' agent >>= \(p, t) -> send p req >> return t - - lookupAgent :: MxTables -> MxAgentId -> Process (ProcessId, MxTables) - lookupAgent tables' agentId' = do - case Map.lookup agentId' tables' of - Nothing -> launchNew agentId' tables' - Just p -> return (p, tables') - - launchNew :: MxAgentId - -> MxTables - -> Process (ProcessId, MxTables) - launchNew mxId tblMap = do - let initState = MxTableState { _name = (agentId mxId) - , _entries = Map.empty - } - (pid, _) <- spawnSup $ tableHandler initState - return $ (pid, mxId `seq` pid `seq` Map.insert mxId pid tblMap) - - spawnSup proc = do - us <- getSelfPid - -- we need to use that passed in "fork", in order to - -- break an import cycle with Node.hs courtesy of the - -- management agent, API and tracing modules - them <- liftIO $ fork $ link us >> proc - ref <- monitor them - return (them, ref) - -tableHandler :: MxTableState -> Process () -tableHandler state = do - ns <- receiveWait [ - match (handleTableRequest state) - , matchAny (\_ -> return (Just state)) - ] - case ns of - Nothing -> return () - Just s' -> tableHandler s' - where - handleTableRequest _ Delete = return Nothing - handleTableRequest st Purge = return $ Just $ (entries ^= Map.empty) $ st - handleTableRequest st (Clear k) = return $ Just $ (entries ^: (k `seq` Map.delete k)) $ st - handleTableRequest st (Set k v) = return $ Just $ (entries ^: (k `seq` v `seq` Map.insert k v)) st - handleTableRequest st (Get k c) = getEntry k c st >> return (Just st) - -getEntry :: String - -> SendPort (Maybe Message) - -> MxTableState - -> Process () -getEntry k m MxTableState{..} = do - sendChan m =<< return (Map.lookup k _entries) - -entries :: Accessor MxTableState (Map String Message) -entries = accessor _entries (\ls st -> st { _entries = ls }) diff --git a/src/Control/Distributed/Process/Management/Internal/Trace/Primitives.hs b/src/Control/Distributed/Process/Management/Internal/Trace/Primitives.hs --- a/src/Control/Distributed/Process/Management/Internal/Trace/Primitives.hs +++ b/src/Control/Distributed/Process/Management/Internal/Trace/Primitives.hs @@ -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 - diff --git a/src/Control/Distributed/Process/Management/Internal/Trace/Tracer.hs b/src/Control/Distributed/Process/Management/Internal/Trace/Tracer.hs --- a/src/Control/Distributed/Process/Management/Internal/Trace/Tracer.hs +++ b/src/Control/Distributed/Process/Management/Internal/Trace/Tracer.hs @@ -77,10 +77,6 @@ import Debug.Trace (traceEventIO) import Prelude -#if ! MIN_VERSION_base(4,6,0) -import Prelude hiding (catch) -#endif - import System.Environment (getEnv) import System.IO ( Handle @@ -91,11 +87,7 @@ , hPutStrLn , hSetBuffering ) -#if MIN_VERSION_time(1,5,0) import Data.Time.Format (defaultTimeLocale) -#else -import System.Locale (defaultTimeLocale) -#endif import System.Mem.Weak ( Weak ) diff --git a/src/Control/Distributed/Process/Management/Internal/Trace/Types.hs b/src/Control/Distributed/Process/Management/Internal/Trace/Types.hs --- a/src/Control/Distributed/Process/Management/Internal/Trace/Types.hs +++ b/src/Control/Distributed/Process/Management/Internal/Trace/Types.hs @@ -43,7 +43,6 @@ import Data.Binary import Data.List (intersperse) import Data.Set (Set) -import qualified Data.Set as Set (fromList) import Data.Typeable import GHC.Generics @@ -158,12 +157,3 @@ getCurrentTraceClient :: MxEventBus -> SendPort (Maybe ProcessId) -> IO () getCurrentTraceClient t s = publishEvent t (unsafeCreateUnencodedMessage s) - -class Traceable a where - uod :: [a] -> TraceSubject - -instance Traceable ProcessId where - uod = TraceProcs . Set.fromList - -instance Traceable String where - uod = TraceNames . Set.fromList diff --git a/src/Control/Distributed/Process/Management/Internal/Types.hs b/src/Control/Distributed/Process/Management/Internal/Types.hs --- a/src/Control/Distributed/Process/Management/Internal/Types.hs +++ b/src/Control/Distributed/Process/Management/Internal/Types.hs @@ -4,27 +4,24 @@ {-# LANGUAGE DeriveGeneric #-} module Control.Distributed.Process.Management.Internal.Types ( MxAgentId(..) - , MxTableId(..) , MxAgentState(..) , MxAgent(..) , MxAction(..) , ChannelSelector(..) - , MxAgentStart(..) , Fork , MxSink , MxEvent(..) , Addressable(..) ) where -import Control.Applicative (Applicative) import Control.Concurrent.STM ( TChan ) import Control.Distributed.Process.Internal.Types ( Process , ProcessId + , SendPortId , Message - , SendPort , DiedReason , NodeId ) @@ -33,6 +30,7 @@ ( MonadState , StateT ) +import Control.Monad.Fix (MonadFix) import Data.Binary import Data.Typeable (Typeable) import GHC.Generics @@ -59,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 @@ -99,17 +103,10 @@ newtype MxAgentId = MxAgentId { agentId :: String } deriving (Typeable, Binary, Eq, Ord) -data MxTableId = - MxForAgent !MxAgentId - | MxForPid !ProcessId - deriving (Typeable, Generic) -instance Binary MxTableId where - data MxAgentState s = MxAgentState { mxAgentId :: !MxAgentId , mxBus :: !(TChan Message) - , mxSharedTable :: !ProcessId , mxLocalState :: !s } @@ -122,18 +119,11 @@ } deriving ( Functor , Monad , MonadIO + , MonadFix , ST.MonadState (MxAgentState s) , Typeable , Applicative ) - -data MxAgentStart = MxAgentStart - { - mxAgentTableChan :: SendPort ProcessId - , mxAgentIdStart :: MxAgentId - } - deriving (Typeable, Generic) -instance Binary MxAgentStart where data ChannelSelector = InputChan | Mailbox 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 @@ -30,13 +30,15 @@ ( empty , toList , fromList - , filter + , partition , partitionWithKey , elems , size , filterWithKey , foldlWithKey ) +import Data.Time.Format (formatTime) +import Data.Time.Format (defaultTimeLocale) import Data.Set (Set) import qualified Data.Set as Set ( empty @@ -48,7 +50,6 @@ , union ) import Data.Foldable (forM_) -import Data.List (foldl') import Data.Maybe (isJust, fromJust, isNothing, catMaybes) import Data.Typeable (Typeable) import Control.Category ((>>>)) @@ -178,9 +179,8 @@ import Control.Distributed.Process.Management.Internal.Agent ( mxAgentController ) -import qualified Control.Distributed.Process.Management.Internal.Table as Table - ( mxTableCoordinator - , startTableCoordinator +import Control.Distributed.Process.Management.Internal.Types + ( MxEvent(..) ) import qualified Control.Distributed.Process.Management.Internal.Trace.Remote as Trace ( remoteTable @@ -194,9 +194,6 @@ , traceLogFmt , enableTrace ) -import Control.Distributed.Process.Management.Internal.Types - ( MxEvent(..) - ) import Control.Distributed.Process.Serializable (Serializable) import Control.Distributed.Process.Internal.Messaging ( sendBinary @@ -209,6 +206,7 @@ , match , sendChan , unwrapMessage + , SayMessage(..) ) import Control.Distributed.Process.Internal.Types (SendPort, Tracer(..)) import qualified Control.Distributed.Process.Internal.Closure.BuiltIn as BuiltIn (remoteTable) @@ -316,8 +314,6 @@ -- before /that/ process has started - this is a totally harmless race -- however, so we deliberably ignore it startDefaultTracer node - tableCoordinatorPid <- fork $ Table.startTableCoordinator fork - runProcess node $ register Table.mxTableCoordinator tableCoordinatorPid logger <- forkProcess node loop runProcess node $ do register "logger" logger @@ -326,12 +322,11 @@ -- 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 - liftIO . hPutStrLn stderr $ time ++ " " ++ show pid ++ ": " ++ string + [ match $ \(SayMessage time pid string) -> do + let time' = formatTime defaultTimeLocale "%c" time + liftIO . hPutStrLn stderr $ time' ++ " " ++ show pid ++ ": " ++ string loop , match $ \((time, string) :: (String, String)) -> do -- this is a 'trace' message from the local node tracer @@ -487,7 +482,7 @@ data IncomingTarget = Uninit | ToProc ProcessId (Weak (CQueue Message)) - | ToChan TypedChannel + | ToChan SendPortId TypedChannel | ToNode data ConnectionState = ConnectionState { @@ -542,13 +537,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 @@ -575,7 +574,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" @@ -904,7 +903,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) -> @@ -956,7 +959,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 $ @@ -1037,11 +1044,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" @@ -1071,7 +1079,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) @@ -1085,8 +1093,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 diff --git a/src/Control/Distributed/Process/Serializable.hs b/src/Control/Distributed/Process/Serializable.hs --- a/src/Control/Distributed/Process/Serializable.hs +++ b/src/Control/Distributed/Process/Serializable.hs @@ -1,8 +1,9 @@ {-# LANGUAGE DeriveDataTypeable #-} +{-# LANGUAGE ConstraintKinds #-} {-# LANGUAGE UndecidableInstances #-} {-# LANGUAGE FlexibleInstances #-} +{-# LANGUAGE ScopedTypeVariables #-} {-# LANGUAGE GADTs #-} -{-# LANGUAGE CPP #-} module Control.Distributed.Process.Serializable ( Serializable , encodeFingerprint @@ -17,13 +18,7 @@ import Data.Binary (Binary) -#if MIN_VERSION_base(4,7,0) -import Data.Typeable (Typeable) -import Data.Typeable.Internal (TypeRep(TypeRep), typeOf) -#else -import Data.Typeable (Typeable(..)) -import Data.Typeable.Internal (TypeRep(TypeRep)) -#endif +import Data.Typeable (Typeable, typeRepFingerprint, typeOf) import Numeric (showHex) import Control.Exception (throw) @@ -46,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 @@ -71,11 +65,7 @@ -- | The fingerprint of the typeRep of the argument fingerprint :: Typeable a => a -> Fingerprint -#if MIN_VERSION_base(4,8,0) -fingerprint a = let TypeRep fp _ _ _ = typeOf a in fp -#else -fingerprint a = let TypeRep fp _ _ = typeOf a in fp -#endif +fingerprint = typeRepFingerprint . typeOf -- | Show fingerprint (for debugging purposes) showFingerprint :: Fingerprint -> ShowS 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 @@ -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