distributed-process 0.6.6 → 0.7.8
raw patch · 25 files changed
Files
- ChangeLog +47/−0
- Setup.hs +0/−2
- benchmarks/Channels.hs +6/−4
- benchmarks/Latency.hs +7/−4
- benchmarks/ProcessRing.hs +3/−2
- benchmarks/Spawns.hs +6/−4
- benchmarks/Throughput.hs +6/−4
- distributed-process.cabal +128/−107
- src/Control/Distributed/Process.hs +2/−4
- src/Control/Distributed/Process/Internal/CQueue.hs +1/−7
- 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 +139/−71
- src/Control/Distributed/Process/Internal/StrictMVar.hs +0/−19
- src/Control/Distributed/Process/Internal/Types.hs +9/−5
- src/Control/Distributed/Process/Internal/WeakTQueue.hs +0/−4
- src/Control/Distributed/Process/Management.hs +38/−81
- src/Control/Distributed/Process/Management/Internal/Table.hs +0/−222
- src/Control/Distributed/Process/Management/Internal/Trace/Primitives.hs +8/−2
- src/Control/Distributed/Process/Management/Internal/Trace/Tracer.hs +0/−8
- src/Control/Distributed/Process/Management/Internal/Trace/Types.hs +0/−10
- src/Control/Distributed/Process/Management/Internal/Types.hs +9/−19
- src/Control/Distributed/Process/Node.hs +37/−29
- src/Control/Distributed/Process/Serializable.hs +5/−15
- src/Control/Distributed/Process/UnsafePrimitives.hs +83/−33
ChangeLog view
@@ -1,3 +1,50 @@+2025-02-04 Laurent P. René de Cotret <laurent.decotret@outlook.com> 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 <laurent.decotret@outlook.com> 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 <dsd@caltech.edu> 0.7.6++* Bumped hashable upper bound.++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++* Added support for exceptions >= 0.10++2017-08-31 Facundo Domínguez <facundo.dominguez@tweag.io> 0.7.3++* Drop support for ghc-7.8.* and earlier.++2017-08-31 Facundo Domínguez <facundo.dominguez@tweag.io> 0.7.2++* Fixed build errors with ghc-8.2.1.++2017-08-22 Facundo Domínguez <facundo.dominguez@tweag.io> 0.7.1++* Relax upper bounds in dependencies to build with ghc-8.2.1.++2017-08-21 Facundo Domínguez <facundo.dominguez@tweag.io> 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 <facundo.dominguez@tweag.io> 0.6.6 * Remove monitors from remote nodes when a process dies. (#295)
− Setup.hs
@@ -1,2 +0,0 @@-import Distribution.Simple-main = defaultMain
benchmarks/Channels.hs view
@@ -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
benchmarks/Latency.hs view
@@ -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
benchmarks/ProcessRing.hs view
@@ -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))
benchmarks/Spawns.hs view
@@ -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
benchmarks/Throughput.hs view
@@ -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
distributed-process.cabal view
@@ -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 <facundo.dominguez@tweag.io>+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- (<http://research.microsoft.com/en-us/um/people/simonpj/papers/parallel/>),+ (<https://simon.peytonjones.org/haskell-cloud/>), 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
src/Control/Distributed/Process.hs view
@@ -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_)
src/Control/Distributed/Process/Internal/CQueue.hs view
@@ -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
src/Control/Distributed/Process/Internal/Closure/Explicit.hs view
@@ -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
src/Control/Distributed/Process/Internal/Closure/TH.hs view
@@ -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 --
src/Control/Distributed/Process/Internal/Primitives.hs view
@@ -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
src/Control/Distributed/Process/Internal/StrictMVar.hs view
@@ -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
src/Control/Distributed/Process/Internal/Types.hs view
@@ -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
src/Control/Distributed/Process/Internal/WeakTQueue.hs view
@@ -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
src/Control/Distributed/Process/Management.hs view
@@ -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 ()
− src/Control/Distributed/Process/Management/Internal/Table.hs
@@ -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 })
src/Control/Distributed/Process/Management/Internal/Trace/Primitives.hs view
@@ -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-
src/Control/Distributed/Process/Management/Internal/Trace/Tracer.hs view
@@ -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 )
src/Control/Distributed/Process/Management/Internal/Trace/Types.hs view
@@ -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
src/Control/Distributed/Process/Management/Internal/Types.hs view
@@ -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
src/Control/Distributed/Process/Node.hs view
@@ -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
src/Control/Distributed/Process/Serializable.hs view
@@ -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
src/Control/Distributed/Process/UnsafePrimitives.hs view
@@ -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