distributed-process 0.5.5.1 → 0.7.8
raw patch · 30 files changed
Files
- ChangeLog +111/−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/−106
- src/Control/Distributed/Process.hs +55/−6
- src/Control/Distributed/Process/Debug.hs +10/−14
- src/Control/Distributed/Process/Internal/BiMultiMap.hs +135/−0
- src/Control/Distributed/Process/Internal/CQueue.hs +63/−18
- src/Control/Distributed/Process/Internal/Closure/Explicit.hs +4/−3
- src/Control/Distributed/Process/Internal/Closure/TH.hs +42/−8
- src/Control/Distributed/Process/Internal/Messaging.hs +36/−26
- src/Control/Distributed/Process/Internal/Primitives.hs +245/−117
- src/Control/Distributed/Process/Internal/Spawn.hs +3/−3
- src/Control/Distributed/Process/Internal/StrictMVar.hs +4/−19
- src/Control/Distributed/Process/Internal/Types.hs +124/−26
- src/Control/Distributed/Process/Internal/WeakTQueue.hs +4/−3
- src/Control/Distributed/Process/Management.hs +41/−83
- src/Control/Distributed/Process/Management/Internal/Agent.hs +2/−1
- src/Control/Distributed/Process/Management/Internal/Table.hs +0/−222
- src/Control/Distributed/Process/Management/Internal/Trace/Primitives.hs +10/−3
- src/Control/Distributed/Process/Management/Internal/Trace/Tracer.hs +8/−13
- src/Control/Distributed/Process/Management/Internal/Trace/Types.hs +0/−10
- src/Control/Distributed/Process/Management/Internal/Types.hs +9/−20
- src/Control/Distributed/Process/Node.hs +305/−210
- src/Control/Distributed/Process/Serializable.hs +8/−20
- src/Control/Distributed/Process/UnsafePrimitives.hs +110/−29
ChangeLog view
@@ -1,3 +1,114 @@+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)++2016-10-12 Facundo Domínguez <facundo.dominguez@tweag.io> 0.6.5++* Use only one connection to communicate between NCs. (#296, #297)+* Improve documentation of CQueue.+* Implement bidirectional multimaps for links and monitors. (#293, #294)+* Fix various warnings. (#292)+* Fix some of the intermittent failures in tests.+* Improve error messages when node controllers receive invalid requests.+(#290)++2016-06-09 Facundo Domínguez <facundo.dominguez@tweag.io> 0.6.4++* Fixup build errors.++2016-06-09 Facundo Domínguez <facundo.dominguez@tweag.io> 0.6.3++* Relax template-haskell upper bound.++2016-06-09 Facundo Domínguez <facundo.dominguez@tweag.io> 0.6.2++* Provide compatibility with ghc-8.0.1+* Remove dependency on ghc-prim.+* Don't throw exceptions asynchronously.+* Bump upper bounds of dependencies.+* Fix exception handling in callLocal.+* Have spawnLocal inherit the masking state of the caller.+* Have nsend send unencoded messages to local processes.++2016-03-03 Facundo Domínguez <facundo.dominguez@tweag.io> 0.6.1++* Implement MonadCatch, MonadThrow, MonadMask for Process.++2016-02-18 Facundo Domínguez <facundo.dominguez@tweag.io> 0.6.0++* Have nsendRemote skip the transport for local communication.+* Unsafe primitives for usend and nsendRemote.+* Stop using the transport for local communication.+* Skip the transport for whereisRemoteAsync and registerRemoteAsync.+* Have nsendRemote skip the transport for local communication.+* Have runProcess forward exceptions.+* Reimport distributed-process-tests. d-p and d-p-test now can be kept in+sync.+* Add a stack.yaml file for building tests and d-p all at once conveniently.+* Implement unreliable forward (uforward).+* Add Functor instance for Match data type+* Have `spawnAsync` not use the transport in the local case.+* Fix monitor race in 'call'.+* Add compatibility with ghc-7.10: support new typeable, loosen deps, write+proper NFData instances, support new TH.+* Kill processes on a local node upon closeLocalNode.+* Fix getNodeStats function, see DP-97+* Return size of the queue in ProcessInfo.+* Implement unreliable send (usend).+* Implement MonadFix instance for Process.+* Introduce callLocal primitive.+* Prevent message loss due to timeouts in CQueue.+* More informative ProcessRegistrationException. Now includes the identifier+of the process that owns the name, if any.+* Avoid message loop between threads when tracing received messages.+ 2015-06-15 Facundo Domínguez <facundo.dominguez@tweag.io> 0.5.5 * Fix dependencies.
− 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,154 +1,176 @@+cabal-version: 3.0 Name: distributed-process-Version: 0.5.5.1-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.4.2 GHC==7.6.3 GHC==7.8.4 GHC==7.10.1+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.8,- hashable >= 1.2.0.5 && < 1.3,- network-transport >= 0.4.1.0 && < 0.5,- stm >= 2.4 && < 2.5,- transformers >= 0.2 && < 0.5,+ 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,- ghc-prim >= 0.2 && < 0.5,+ 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- Exposed-modules: Control.Distributed.Process,- Control.Distributed.Process.Closure,- Control.Distributed.Process.Debug,- 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.11+ 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.5,- bytestring >= 0.9 && < 0.11,- binary >= 0.6.3 && < 0.8- 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.5,- bytestring >= 0.9 && < 0.11,- binary >= 0.6.3 && < 0.8- 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.5,- bytestring >= 0.9 && < 0.11,- binary >= 0.6.3 && < 0.8- 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.5,- bytestring >= 0.9 && < 0.11,- binary >= 0.6.3 && < 0.8- 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.5,- bytestring >= 0.9 && < 0.11,- binary >= 0.6.3 && < 0.8- 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
@@ -25,6 +25,7 @@ , liftIO -- Reexported for convenience -- * Basic messaging , send+ , usend , expect , expectTimeout -- * Channels@@ -39,8 +40,10 @@ , mergePortsRR -- * Unsafe messaging variants , unsafeSend+ , unsafeUSend , unsafeSendChan , unsafeNSend+ , unsafeNSendRemote , unsafeWrapMessage -- * Advanced messaging , Match@@ -64,6 +67,7 @@ , handleMessage_ , handleMessageIf_ , forward+ , uforward , delegate , relay , proxy@@ -98,6 +102,7 @@ , monitorPort , unmonitor , withMonitor+ , withMonitor_ , MonitorRef -- opaque , ProcessLinkException(..) , NodeLinkException(..)@@ -149,19 +154,22 @@ -- * Local versions of 'spawn' , spawnLocal , spawnChannelLocal+ , callLocal -- * Reconnecting , reconnect , reconnectPort ) where -#if ! MIN_VERSION_base(4,6,0)-import Prelude hiding (catch)-#endif- import Control.Monad.IO.Class (liftIO)-import Control.Applicative ((<$>))+import Control.Applicative import Control.Monad.Reader (ask)-import Control.Concurrent.MVar (newEmptyMVar, takeMVar, putMVar)+import Control.Concurrent (killThread)+import Control.Concurrent.MVar+ ( MVar+ , newEmptyMVar+ , takeMVar+ , putMVar+ ) import Control.Distributed.Static ( Closure , closure@@ -190,11 +198,13 @@ , RegisterReply(..) , LocalProcess(processNode) , Message+ , localProcessWithId ) import Control.Distributed.Process.Serializable (Serializable) import Control.Distributed.Process.Internal.Primitives ( -- Basic messaging send+ , usend , expect -- Channels , newChan@@ -203,8 +213,10 @@ , mergePortsBiased , mergePortsRR , unsafeSend+ , unsafeUSend , unsafeSendChan , unsafeNSend+ , unsafeNSendRemote -- Advanced messaging , Match , receiveWait@@ -227,6 +239,7 @@ , handleMessage_ , handleMessageIf_ , forward+ , uforward , delegate , relay , proxy@@ -257,6 +270,7 @@ , monitorPort , unmonitor , withMonitor+ , withMonitor_ -- Logging , say -- Registry@@ -293,6 +307,10 @@ , reconnectPort ) import Control.Distributed.Process.Node (forkProcess)+import Control.Distributed.Process.Internal.Types+ ( processThread+ , withValidLocalState+ ) import Control.Distributed.Process.Internal.Spawn ( -- Spawning Processes/Channels spawn@@ -302,7 +320,14 @@ , spawnSupervised , call )+import qualified Control.Monad.Catch as Catch +import Prelude+import qualified Control.Exception as Exception (onException)+import Data.Accessor ((^.))+import Data.Foldable (forM_)++ -- INTERNAL NOTES -- -- 1. 'send' never fails. If you want to know that the remote process received@@ -392,3 +417,27 @@ liftIO $ putMVar mvar sport proc rport takeMVar mvar++-- | Local version of 'call'. Running a process in this way isolates it from+-- messages sent to the caller process, and also allows silently dropping late+-- or duplicate messages sent to the isolated process after it exits.+-- Silently dropping messages may not always be the best approach.+callLocal :: Process a -> Process a+callLocal proc = Catch.mask $ \release -> do+ mv <- liftIO newEmptyMVar :: Process (MVar (Either Catch.SomeException a))+ child <- spawnLocal $ Catch.try (release proc) >>= liftIO . putMVar mv+ lproc <- ask+ liftIO $ do+ rs <- Exception.onException (takeMVar mv) $ Catch.uninterruptibleMask_ $+ -- Exceptions need to be prevented from interrupting the clean up or+ -- the original exception which caused entering the handler could be+ -- forgotten. For instance, this could have a problematic effect+ -- when the original exception was meant to kill the thread and the+ -- second exception doesn't (like the exception thrown by+ -- 'System.Timeout.timeout').+ do mchildThreadId <- withValidLocalState (processNode lproc) $+ \vst -> return $ fmap processThread $+ vst ^. localProcessWithId (processLocalId child)+ forM_ mchildThreadId killThread+ takeMVar mv+ either Catch.throwM return rs
src/Control/Distributed/Process/Debug.hs view
@@ -83,13 +83,13 @@ -- environment variable accepts the following flags, which enable tracing specific -- event types: ----- p = trace the spawning of new processes--- d = trace the death of processes--- n = trace registration of names (i.e., named processes)--- u = trace un-registration of names (i.e., named processes)--- s = trace the sending of messages to other processes--- r = trace the receipt of messages from other processes--- l = trace node up/down events+-- * @p@ = trace the spawning of new processes+-- * @d@ = trace the death of processes+-- * @n@ = trace registration of names (i.e., named processes)+-- * @u@ = trace un-registration of names (i.e., named processes)+-- * @s@ = trace the sending of messages to other processes+-- * @r@ = trace the receipt of messages from other processes+-- * @l@ = trace node up/down events -- -- Users of the /simplelocalnet/ Cloud Haskell backend should also note that -- because the trace file option only supports trace output from a single node@@ -141,17 +141,14 @@ ) where -import Control.Applicative ((<$>))+import Control.Applicative import Control.Distributed.Process.Internal.Primitives ( proxy- , finally , die , whereis , send , receiveWait , matchIf- , finally- , try , monitor ) import Control.Distributed.Process.Internal.Types@@ -196,12 +193,11 @@ import Control.Monad.IO.Class (liftIO) import Control.Monad.Reader (ask)+import Control.Monad.Catch (finally, try) import Data.Binary() -#if ! MIN_VERSION_base(4,6,0)-import Prelude hiding (catch)-#endif+import Prelude -------------------------------------------------------------------------------- -- Debugging/Tracing API --
+ src/Control/Distributed/Process/Internal/BiMultiMap.hs view
@@ -0,0 +1,135 @@+-- | This is an implementation of bidirectional multimaps.+module Control.Distributed.Process.Internal.BiMultiMap+ ( BiMultiMap+ , empty+ , singleton+ , size+ , insert+ , lookupBy1st+ , lookupBy2nd+ , delete+ , deleteAllBy1st+ , deleteAllBy2nd+ , partitionWithKeyBy1st+ , partitionWithKeyBy2nd+ , flip+ ) where++import Data.List (foldl')+import Data.Map.Strict (Map)+import qualified Data.Map.Strict as Map+import Data.Set (Set)+import qualified Data.Set as Set+import Prelude hiding (flip, lookup)++-- | A bidirectional multimaps @BiMultiMap a b v@ is a set of triplets of type+-- @(a, b, v)@.+--+-- It is possible to lookup values by using either @a@ or @b@ as keys.+--+data BiMultiMap a b v = BiMultiMap !(Map a (Set (b, v))) !(Map b (Set (a, v)))++-- The bidirectional multimap is implemented with a pair of multimaps.+--+-- Each multimap represents a set of triples, and one invariant is that both+-- multimaps should represent exactly the same set of triples.+--+-- Each of the multimaps, however, uses a different component of the triplets+-- as key. This allows to do efficient deletions by any of the two components.++-- | The empty bidirectional multimap.+empty :: BiMultiMap a b v+empty = BiMultiMap Map.empty Map.empty++-- | A bidirectional multimap containing a single triplet.+singleton :: (Ord a, Ord b, Ord v) => a -> b -> v -> BiMultiMap a b v+singleton a b v = insert a b v empty++-- | Yields the amount of triplets in the multimap.+size :: BiMultiMap a b v -> Int+size (BiMultiMap m _) = foldl' (+) 0 $ map Set.size $ Map.elems m++-- | Inserts a triplet in the multimap.+insert :: (Ord a, Ord b, Ord v)+ => a -> b -> v -> BiMultiMap a b v -> BiMultiMap a b v+insert a b v (BiMultiMap m r) =+ BiMultiMap (Map.insertWith (\_new old -> Set.insert (b, v) old)+ a+ (Set.singleton (b, v))+ m)+ (Map.insertWith (\_new old -> Set.insert (a, v) old)+ b+ (Set.singleton (a, v))+ r)++-- | Looks up all the triplets whose first component is the given value.+lookupBy1st :: Ord a => a -> BiMultiMap a b v -> Set (b, v)+lookupBy1st a (BiMultiMap m _) = maybe Set.empty id $ Map.lookup a m++-- | Looks up all the triplets whose second component is the given value.+lookupBy2nd :: Ord b => b -> BiMultiMap a b v -> Set (a, v)+lookupBy2nd b = lookupBy1st b . flip++-- | Deletes a triplet. It yields the original multimap if the triplet is+-- not present.+delete :: (Ord a, Ord b, Ord v)+ => a -> b -> v -> BiMultiMap a b v -> BiMultiMap a b v+delete a b v (BiMultiMap m r) =+ let m' = Map.update (nothingWhen Set.null . Set.delete (b, v)) a m+ r' = Map.update (nothingWhen Set.null . Set.delete (a, v)) b r+ in BiMultiMap m' r'++-- | Deletes all triplets whose first component is the given value.+deleteAllBy1st :: (Ord a, Ord b, Ord v) => a -> BiMultiMap a b v -> BiMultiMap a b v+deleteAllBy1st a (BiMultiMap m r) =+ let (mm, m') = Map.updateLookupWithKey (\_ _ -> Nothing) a m+ r' = case mm of+ Nothing -> r+ Just mb -> reverseDelete a (Set.toList mb) r+ in BiMultiMap m' r'++-- | Like 'deleteAllBy1st' but deletes by the second component of the triplets.+deleteAllBy2nd :: (Ord a, Ord b, Ord v)+ => b -> BiMultiMap a b v -> BiMultiMap a b v+deleteAllBy2nd b = flip . deleteAllBy1st b . flip++-- | Yields the triplets satisfying the given predicate, and a multimap+-- with all this triplets removed.+partitionWithKeyBy1st :: (Ord a, Ord b, Ord v)+ => (a -> Set (b, v) -> Bool) -> BiMultiMap a b v+ -> (Map a (Set (b, v)), BiMultiMap a b v)+partitionWithKeyBy1st p (BiMultiMap m r) =+ let (m0, m1) = Map.partitionWithKey p m+ r1 = foldl' (\rr (a, mb) -> reverseDelete a (Set.toList mb) rr) r $+ Map.toList m0+ in (m0, BiMultiMap m1 r1)++-- | Like 'partitionWithKeyBy1st' but the predicates takes the second component+-- of the triplets as first argument.+partitionWithKeyBy2nd :: (Ord a, Ord b, Ord v)+ => (b -> Set (a, v) -> Bool) -> BiMultiMap a b v+ -> (Map b (Set (a, v)), BiMultiMap a b v)+partitionWithKeyBy2nd p b = let (m, b') = partitionWithKeyBy1st p $ flip b+ in (m, flip b')++-- | Exchange the first and the second components of all triplets.+flip :: BiMultiMap a b v -> BiMultiMap b a v+flip (BiMultiMap m r) = BiMultiMap r m++-- Internal functions++-- | @reverseDelete a bs m@ removes from @m@ all the triplets wich have @a@ as+-- first component and second and third components in @bs@.+--+-- The @m@ map is in reversed form, meaning that the second component of the+-- triplets is used as key.+reverseDelete :: (Ord a, Ord b, Ord v)+ => a -> [(b, v)] -> Map b (Set (a, v)) -> Map b (Set (a, v))+reverseDelete a bs r = foldl' (\rr (b, v) -> Map.update (rmb v) b rr) r bs+ where+ rmb v = nothingWhen Set.null . Set.delete (a, v)++-- | @nothingWhen p a@ is @Just a@ when @a@ satisfies predicate @p@.+-- Yields @Nothing@ otherwise.+nothingWhen :: (a -> Bool) -> a -> Maybe a+nothingWhen p a = if p a then Nothing else Just a
src/Control/Distributed/Process/Internal/CQueue.hs view
@@ -1,3 +1,5 @@+{-# LANGUAGE CPP #-}+{-# LANGUAGE DeriveFunctor #-} {-# LANGUAGE BangPatterns #-} {-# LANGUAGE MagicHash, UnboxedTuples, PatternGuards, ScopedTypeVariables, RankNTypes #-} -- | Concurrent queue for single reader, single writer@@ -10,6 +12,7 @@ , enqueueSTM , dequeue , mkWeakCQueue+ , queueSize ) where import Prelude hiding (length, reverse)@@ -17,14 +20,17 @@ ( atomically , STM , TChan+ , TVar+ , modifyTVar' , tryReadTChan , newTChan+ , newTVarIO , writeTChan , readTChan+ , readTVarIO , orElse , retry )-import Control.Applicative ((<$>), (<*>)) import Control.Exception (mask_, onException) import System.Timeout (timeout) import Control.Distributed.Process.Internal.StrictMVar@@ -39,38 +45,54 @@ ) import Data.Maybe (fromJust) import GHC.MVar (MVar(MVar))-import GHC.IO (IO(IO))-import GHC.Prim (mkWeak#)+import GHC.IO (IO(IO), unIO)+import GHC.Exts (mkWeak#) import GHC.Weak (Weak(Weak)) -- We use a TCHan rather than a Chan so that we have a non-blocking read data CQueue a = CQueue (StrictMVar (StrictList a)) -- Arrived (TChan a) -- Incoming+ (TVar Int) -- Queue size newCQueue :: IO (CQueue a)-newCQueue = CQueue <$> newMVar Nil <*> atomically newTChan+newCQueue = CQueue <$> newMVar Nil <*> atomically newTChan <*> newTVarIO 0 -- | Enqueue an element -- -- Enqueue is strict. enqueue :: CQueue a -> a -> IO ()-enqueue (CQueue _arrived incoming) !a = atomically $ writeTChan incoming a+enqueue c !a = atomically (enqueueSTM c a) -- | Variant of enqueue for use in the STM monad. enqueueSTM :: CQueue a -> a -> STM ()-enqueueSTM (CQueue _arrived incoming) !a = writeTChan incoming a+enqueueSTM (CQueue _arrived incoming size) !a = do+ writeTChan incoming a+ modifyTVar' size succ data BlockSpec = NonBlocking | Blocking- | Timeout Int+ | Timeout Int -- ^ Timeout in microseconds +-- Match operations+--+-- They can be either a message match or a channel match. data MatchOn m a = MatchMsg (m -> Maybe a) | MatchChan (STM a)+ deriving (Functor) +-- Lists of chunks of matches+--+-- Two consecutive chunks never have the same kind of matches. i.e. if one chunk+-- contains message matches then the next one must contain channel matches and+-- viceversa. type MatchChunks m a = [Either [m -> Maybe a] [STM a]] +-- Splits a list of matches into chunks.+--+-- > concatMap (either (map MatchMsg) (map MatchChan)) . chunkMatches == id+-- chunkMatches :: [MatchOn m a] -> MatchChunks m a chunkMatches [] = [] chunkMatches (MatchMsg m : ms) = Left (m : chk) : chunkMatches rest@@ -78,6 +100,7 @@ chunkMatches (MatchChan r : ms) = Right (r : chk) : chunkMatches rest where (chk, rest) = spanMatchChan ms +-- | @spanMatchMsg = first (map (\(MatchMsg x) -> x)) . span isMatchMsg@ spanMatchMsg :: [MatchOn m a] -> ([m -> Maybe a], [MatchOn m a]) spanMatchMsg [] = ([],[]) spanMatchMsg (m : ms)@@ -85,6 +108,7 @@ | otherwise = ([], m:ms) where !(msgs,rest) = spanMatchMsg ms +-- | @spanMatchMsg = first (map (\(MatchChan x) -> x)) . span isMatchChan@ spanMatchChan :: [MatchOn m a] -> ([STM a], [MatchOn m a]) spanMatchChan [] = ([],[]) spanMatchChan (m : ms)@@ -101,7 +125,7 @@ -> BlockSpec -- ^ Blocking behaviour -> [MatchOn m a] -- ^ List of matches -> IO (Maybe a) -- ^ 'Nothing' only on timeout-dequeue (CQueue arrived incoming) blockSpec matchons =+dequeue (CQueue arrived incoming size) blockSpec matchons = mask_ $ decrementJust $ case blockSpec of Timeout n -> timeout n $ fmap fromJust run _other ->@@ -113,9 +137,17 @@ -- no onException needed _other -> run where+ -- Decrement counter is smth is returned from the queue,+ -- this is safe to use as method is called under a mask+ -- and there is no 'unmasked' operation inside+ decrementJust :: IO (Maybe (Either a a)) -> IO (Maybe a)+ decrementJust f =+ traverse (either return (\x -> decrement >> return x)) =<< f+ decrement = atomically $ modifyTVar' size pred+ chunks = chunkMatches matchons - run = mask_ $ do+ run = do arr <- takeMVar arrived let grabNew xs = do r <- atomically $ tryReadTChan incoming@@ -125,17 +157,25 @@ arr' <- grabNew arr goCheck chunks arr' + -- Yields the value of the first succesful STM transaction as+ -- @Just (Left v)@. If all transactions fail, yields the value of the second+ -- argument.+ waitChans :: [STM a] -> STM (Maybe (Either a a)) -> STM (Maybe (Either a a)) waitChans ports on_block =- foldr orElse on_block (map (fmap Just) ports)+ foldr orElse on_block (map (fmap (Just . Left)) ports) -- -- First check the MatchChunks against the messages already in the -- mailbox. For channel matches, we do a non-blocking check at -- this point. --+ -- Yields @Just (Left a)@ when a channel is matched, @Just (Right a)@+ -- when a message is matched and @Nothing@ when there are no messages and we+ -- aren't blocking.+ -- goCheck :: MatchChunks m a -> StrictList m -- messages to check, in this order- -> IO (Maybe a)+ -> IO (Maybe (Either a a)) goCheck [] old = goWait old @@ -152,7 +192,7 @@ -- of passing around restore and setting up exception handlers is -- high. So just don't use expensive matchIfs! case checkArrived matches old of- (old', Just r) -> returnOld old' (Just r)+ (old', Just r) -> returnOld old' (Just (Right r)) (old', Nothing) -> goCheck rest old' -- use the result list, which is now left-biased @@ -181,6 +221,7 @@ -- Contents of 'arrived' from now on is (old ++ new), and -- messages that arrive are snocced onto new. --+ goWait :: StrictList m -> IO (Maybe (Either a a)) goWait old = do r <- waitIncoming `onException` putMVar arrived old case r of@@ -198,7 +239,7 @@ -- -- Right => message arrived on a channel first --- Right a -> returnOld old (Just a)+ Right a -> returnOld old (Just (Left a)) -- -- A message arrived in the process inbox; check the MatchChunks for@@ -207,7 +248,7 @@ goCheck1 :: MatchChunks m a -> m -- single message to check -> StrictList m -- old messages we have already checked- -> IO (Maybe a)+ -> IO (Maybe (Either a a)) goCheck1 [] m old = goWait (Snoc old m) @@ -220,16 +261,17 @@ goCheck1 (Left matches : rest) m old = do case checkMatches matches m of Nothing -> goCheck1 rest m old- Just p -> returnOld old (Just p)+ Just p -> returnOld old (Just (Right p)) -- a common pattern for putting back the arrived queue at the end- returnOld :: StrictList m -> Maybe a -> IO (Maybe a)+ returnOld :: StrictList m -> Maybe (Either a a) -> IO (Maybe (Either a a)) returnOld old r = do putMVar arrived old; return r -- as a side-effect, this left-biases the list checkArrived :: [m -> Maybe a] -> StrictList m -> (StrictList m, Maybe a) checkArrived matches list = go list Nil where+ -- @go xs ys@ searches for a message match in @append xs ys@ go Nil Nil = (Nil, Nothing) go Nil r = go r Nil go (Append xs ys) tl = go xs (append ys tl)@@ -245,5 +287,8 @@ -- | Weak reference to a CQueue mkWeakCQueue :: CQueue a -> IO () -> IO (Weak (CQueue a))-mkWeakCQueue m@(CQueue (StrictMVar (MVar m#)) _) f = IO $ \s ->- case mkWeak# m# m f s of (# s1, w #) -> (# s1, Weak w #)+mkWeakCQueue m@(CQueue (StrictMVar (MVar m#)) _ _) f = IO $ \s ->+ case mkWeak# m# m (unIO f) s of (# s1, w #) -> (# s1, Weak w #)++queueSize :: CQueue a -> IO Int+queueSize (CQueue _ _ size) = readTVarIO size
src/Control/Distributed/Process/Internal/Closure/Explicit.hs view
@@ -6,8 +6,8 @@ , UndecidableInstances , KindSignatures , GADTs- , OverlappingInstances , EmptyDataDecls+ , TypeOperators , DeriveDataTypeable #-} module Control.Distributed.Process.Internal.Closure.Explicit (@@ -30,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@@ -109,7 +110,7 @@ class Curry a b | a -> b where curryFun :: a -> b -instance Curry ((a,EndOfTuple) -> b) (a -> b) where+instance Curry ((a, EndOfTuple) -> b) (a -> b) where curryFun f = \x -> f (x,undefined) instance Curry (b -> c) r => Curry ((a,b) -> c) (a -> r) where@@ -119,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,12 +27,18 @@ , Exp , Type(AppT, ForallT, VarT, ArrowT) , Info(VarI)+#if MIN_VERSION_template_haskell(2,17,0)+ , Specificity+#endif , TyVarBndr(PlainTV, KindedTV)-#if ! MIN_VERSION_template_haskell(2,10,0) , Pred+#if MIN_VERSION_template_haskell(2,10,0)+ , conT+ , appT+#else+ , classP #endif , varT- , classP -- Lifted constructors -- .. Literals , stringL@@ -71,10 +76,6 @@ ) import Control.Distributed.Process.Internal.Closure.BuiltIn (staticDecode) -#if MIN_VERSION_template_haskell(2,10,0)-type Pred = Type-#endif- -------------------------------------------------------------------------------- -- User-level API -- --------------------------------------------------------------------------------@@ -204,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@@ -223,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))@@ -248,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@@ -260,8 +274,17 @@ , 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- typeable tv = classP (mkName "Typeable") [varT (tyVarBndrName tv)]+#endif+ typeable tv =+#if MIN_VERSION_template_haskell(2,10,0)+ conT (mkName "Typeable") `appT` varT (tyVarBndrName tv)+#else+ classP (mkName "Typeable") [varT (tyVarBndrName tv)]+#endif -- | Generate a serialization dictionary with name 'n' for type 'typ' generateDict :: Name -> Type -> Q [Dec]@@ -299,7 +322,11 @@ getType name = do info <- reify name case info of+#if MIN_VERSION_template_haskell(2,11,0)+ VarI origName typ _ -> return (origName, typ)+#else VarI origName typ _ _ -> return (origName, typ)+#endif _ -> fail $ show name ++ " not found" -- | Variation on 'funD' which takes a single expression to define the function@@ -307,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/Messaging.hs view
@@ -13,10 +13,11 @@ import qualified Data.Map as Map (partitionWithKey, elems) import qualified Data.ByteString.Lazy as BSL (toChunks) import qualified Data.ByteString as BSS (ByteString)-import Control.Distributed.Process.Internal.StrictMVar (withMVar, modifyMVar_) import Control.Distributed.Process.Serializable () +import Control.Concurrent (forkIO) import Control.Concurrent.Chan (writeChan)+import Control.Exception (mask_) import Control.Monad (unless) import Control.Monad.IO.Class (liftIO) import Control.Monad.Reader (ask)@@ -29,7 +30,10 @@ , close ) import Control.Distributed.Process.Internal.Types- ( LocalNode(localState, localEndPoint, localCtrlChan)+ ( LocalNode(localEndPoint, localCtrlChan)+ , withValidLocalState+ , modifyValidLocalState+ , modifyValidLocalState_ , Identifier , localConnections , localConnectionBetween@@ -50,6 +54,7 @@ , Identifier(NodeIdentifier, ProcessIdentifier, SendPortIdentifier) ) import Control.Distributed.Process.Serializable (Serializable)+import Data.Foldable (forM_) -------------------------------------------------------------------------------- -- Message sending --@@ -73,7 +78,7 @@ unless didSend $ do writeChan (localCtrlChan node) NCMsg { ctrlMsgSender = to- , ctrlMsgSignal = Died to DiedDisconnect+ , ctrlMsgSignal = Died (NodeIdentifier $ nodeOf to) DiedDisconnect } sendBinary :: Binary a@@ -102,7 +107,7 @@ -> ImplicitReconnect -> IO (Maybe NT.Connection) setupConnBetween node from to implicitReconnect = do- mConn <- NT.connect endPoint+ mConn <- NT.connect (localEndPoint node) (nodeAddress . nodeOf $ to) NT.ReliableOrdered NT.defaultConnectHints@@ -113,14 +118,11 @@ Left _ -> return Nothing Right () -> do- modifyMVar_ nodeState $ return .- (localConnectionBetween from to ^= Just (conn, implicitReconnect))+ modifyValidLocalState_ node $+ return . (localConnectionBetween from to ^= Just (conn, implicitReconnect)) return $ Just conn Left _ -> return Nothing- where- endPoint = localEndPoint node- nodeState = localState node connBetween :: LocalNode -> Identifier@@ -128,33 +130,37 @@ -> ImplicitReconnect -> IO (Maybe NT.Connection) connBetween node from to implicitReconnect = do- mConn <- withMVar nodeState $ return . (^. localConnectionBetween from to)+ mConn <- withValidLocalState node $+ return . (^. localConnectionBetween from to) case mConn of Just (conn, _) -> return $ Just conn Nothing -> setupConnBetween node from to implicitReconnect- where- nodeState = localState node disconnect :: LocalNode -> Identifier -> Identifier -> IO ()-disconnect node from to =- modifyMVar_ (localState node) $ \st ->- case st ^. localConnectionBetween from to of+disconnect node from to = mask_ $ do+ mio <- modifyValidLocalState node $ \vst ->+ case vst ^. localConnectionBetween from to of Nothing ->- return st+ return (vst, return ()) Just (conn, _) -> do- NT.close conn- return (localConnectionBetween from to ^= Nothing $ st)+ return ( localConnectionBetween from to ^= Nothing $ vst+ , NT.close conn+ )+ forM_ mio forkIO closeImplicitReconnections :: LocalNode -> Identifier -> IO ()-closeImplicitReconnections node to =- modifyMVar_ (localState node) $ \st -> do+closeImplicitReconnections node to = mask_ $ do+ mconns <- modifyValidLocalState node $ \vst -> do let shouldClose (_, to') (_, WithImplicitReconnect) = to `impliesDeathOf` to' shouldClose _ _ = False- let (affected, unaffected) = Map.partitionWithKey shouldClose (st ^. localConnections)- mapM_ (NT.close . fst) (Map.elems affected)- return (localConnections ^= unaffected $ st)+ let (affected, unaffected) =+ Map.partitionWithKey shouldClose (vst ^. localConnections)+ return ( localConnections ^= unaffected $ vst+ , map fst $ Map.elems affected+ )+ forM_ mconns $ forkIO . mapM_ NT.close -- | @a `impliesDeathOf` b@ is true if the death of @a@ (for instance, a node) -- implies the death of @b@ (for instance, a process on that node)@@ -177,7 +183,11 @@ False --- Send a control message+-- Send a control message. Evaluates the message to HNF before sending it.+--+-- The message shouldn't produce more errors when further evaluated. If+-- evaluation threw errors the node controller or the receiver would crash when+-- inspecting it. sendCtrlMsg :: Maybe NodeId -- ^ Nothing for the local node -> ProcessSignal -- ^ Message to send -> Process ()@@ -188,10 +198,10 @@ } case mNid of Nothing -> do- liftIO $ writeChan (localCtrlChan (processNode proc)) msg+ liftIO $ writeChan (localCtrlChan (processNode proc)) $! msg Just nid -> liftIO $ sendBinary (processNode proc)- (ProcessIdentifier (processId proc))+ (NodeIdentifier (processNodeId $ processId proc)) (NodeIdentifier nid) WithImplicitReconnect msg
src/Control/Distributed/Process/Internal/Primitives.hs view
@@ -1,6 +1,7 @@ {-# LANGUAGE CPP #-} {-# LANGUAGE RankNTypes #-} {-# LANGUAGE DeriveDataTypeable #-}+{-# LANGUAGE DeriveFunctor #-} {-# LANGUAGE ScopedTypeVariables #-} {-# LANGUAGE ExistentialQuantification #-} {-# LANGUAGE BangPatterns #-}@@ -12,6 +13,7 @@ module Control.Distributed.Process.Internal.Primitives ( -- * Basic messaging send+ , usend , expect -- * Channels , newChan@@ -21,8 +23,10 @@ , mergePortsRR -- * Unsafe messaging variants , unsafeSend+ , unsafeUSend , unsafeSendChan , unsafeNSend+ , unsafeNSendRemote -- * Advanced messaging , Match , receiveWait@@ -45,6 +49,7 @@ , handleMessage_ , handleMessageIf_ , forward+ , uforward , delegate , relay , proxy@@ -71,8 +76,11 @@ , unlink , monitor , unmonitor+ , unmonitorAsync , withMonitor+ , withMonitor_ -- * Logging+ , SayMessage(..) , say -- * Registry , register@@ -116,25 +124,23 @@ , 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.Applicative ((<$>))-import Control.Exception (Exception(..), throw, throwIO, SomeException)-import qualified Control.Exception as Ex (catch, mask, mask_, try)+import Control.Monad.Catch+ ( Exception+ , SomeException+ , throwM+ , fromException+ )+import qualified Control.Monad.Catch as Catch+import Control.Applicative import Control.Distributed.Process.Internal.StrictMVar ( StrictMVar , modifyMVar@@ -174,6 +180,7 @@ , SpawnRef(..) , ProcessSignal(..) , NodeMonitorNotification(..)+ , ProcessMonitorNotification(..) , monitorCounter , spawnCounter , SendPort(..)@@ -198,7 +205,6 @@ , isEncoded , createMessage , createUnencodedMessage- , runLocalProcess , ImplicitReconnect( NoImplicitReconnect) , LocalProcessState , LocalSendPortId@@ -222,6 +228,7 @@ , readTQueue , mkWeakTQueue )+import Prelude import Unsafe.Coerce @@ -238,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@@ -258,6 +263,34 @@ unsafeSend :: Serializable a => ProcessId -> a -> Process () unsafeSend = Unsafe.send +-- | Send a message unreliably.+--+-- Unlike 'send', this function is insensitive to 'reconnect'. It will+-- try to send the message regardless of the history of connection failures+-- between the nodes.+--+-- Message passing with 'usend' is ordered for a given sender and receiver+-- if the messages arrive at all.+--+usend :: Serializable a => ProcessId -> a -> Process ()+usend them msg = do+ proc <- ask+ let there = processNodeId them+ 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)++-- | /Unsafe/ variant of 'usend'. This function makes /no/ attempt to serialize+-- the message when the destination process resides on the same local+-- node. Therefore, a local receiver would need to be prepared to cope with any+-- errors resulting from evaluation of the message.+unsafeUSend :: Serializable a => ProcessId -> a -> Process ()+unsafeUSend = Unsafe.usend+ -- | Wait for a message of a specific type expect :: forall a. Serializable a => Process a expect = receiveWait [match return]@@ -266,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@@ -294,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@@ -355,20 +397,26 @@ -- | Opaque type used in 'receiveWait' and 'receiveTimeout' newtype Match b = Match { unMatch :: MatchOn Message (Process b) }+ deriving (Functor) -- | Test the matches in order against each message in the queue 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@@ -439,19 +487,32 @@ 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'.+--+-- Unlike 'forward', this function is insensitive to 'reconnect'. It will+-- try to send the message regardless of the history of connection failures+-- between the nodes.+uforward :: Message -> ProcessId -> Process ()+uforward msg them = do+ proc <- ask+ let node = processNode proc+ us = processId proc+ nid = localNodeId node+ 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@@ -648,7 +709,7 @@ -- | Terminate immediately (throws a ProcessTerminationException) terminate :: Process a-terminate = liftIO $ throwIO ProcessTerminationException+terminate = throwM ProcessTerminationException -- [Issue #110] -- | Die immediately - throws a 'ProcessExitException' with the given @reason@.@@ -658,7 +719,7 @@ -- could be decoded by a handler passed to 'catchExit', re-thrown or even -- passed to another process via the tracing mechanism. pid <- getSelfPid- liftIO $ throwIO (ProcessExitException pid (createMessage reason))+ throwM (ProcessExitException pid (createMessage reason)) -- | Forceful request to kill a process. Where 'exit' provides an exception -- that can be caught and handled, 'kill' throws an unexposed exception type@@ -691,12 +752,12 @@ => Process b -> (ProcessId -> a -> Process b) -> Process b-catchExit act exitHandler = catch act handleExit+catchExit act exitHandler = Catch.catch act handleExit where handleExit ex@(ProcessExitException from msg) = if messageFingerprint msg == fingerprint (undefined :: a) then exitHandler from decoded- else liftIO $ throwIO ex+ else throwM ex where decoded :: a -- Make sure the value is fully decoded so that we don't hang to@@ -715,12 +776,12 @@ catchesExit :: Process b -> [(ProcessId -> Message -> (Process (Maybe b)))] -> Process b-catchesExit act handlers = catch act ((flip handleExit) handlers)+catchesExit act handlers = Catch.catch act ((flip handleExit) handlers) where handleExit :: ProcessExitException -> [(ProcessId -> Message -> Process (Maybe b))] -> Process b- handleExit ex [] = liftIO $ throwIO ex+ handleExit ex [] = throwM ex handleExit ex@(ProcessExitException from msg) (h:hs) = do r <- h from msg case r of@@ -742,11 +803,11 @@ selfNode <- getSelfNode if nid == selfNode then Right `fmap` getLocalNodeStats -- optimisation- else getNodeStatsRemote+ else getNodeStatsRemote selfNode where- getNodeStatsRemote :: Process (Either DiedReason NodeStats)- getNodeStatsRemote = do- sendCtrlMsg (Just nid) $ GetNodeStats nid+ getNodeStatsRemote :: NodeId -> Process (Either DiedReason NodeStats)+ getNodeStatsRemote selfNode = do+ sendCtrlMsg (Just nid) $ GetNodeStats selfNode bracket (monitorNode nid) unmonitor $ \mRef -> receiveWait [ match (\(stats :: NodeStats) -> return $ Right stats) , matchIf (\(NodeMonitorNotification ref _ _) -> ref == mRef)@@ -829,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@@ -872,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 --@@ -885,56 +962,43 @@ -- | Lift 'Control.Exception.catch' catch :: Exception e => Process a -> (e -> Process a) -> Process a-catch p h = do- lproc <- ask- liftIO $ Ex.catch (runLocalProcess lproc p) (runLocalProcess lproc . h)+catch = Catch.catch+{-# DEPRECATED catch "Use Control.Monad.Catch.catch instead" #-} -- | Lift 'Control.Exception.try' try :: Exception e => Process a -> Process (Either e a)-try p = do- lproc <- ask- liftIO $ Ex.try (runLocalProcess lproc p)+try = Catch.try+{-# DEPRECATED try "Use Control.Monad.Catch.try instead" #-} -- | Lift 'Control.Exception.mask' mask :: ((forall a. Process a -> Process a) -> Process b) -> Process b-mask p = do- lproc <- ask- liftIO $ Ex.mask $ \restore ->- runLocalProcess lproc (p (liftRestore restore))- where- liftRestore :: (forall a. IO a -> IO a)- -> (forall a. Process a -> Process a)- liftRestore restoreIO = \p2 -> do- ourLocalProc <- ask- liftIO $ restoreIO $ runLocalProcess ourLocalProc p2+mask = Catch.mask+{-# DEPRECATED mask "Use Control.Monad.Catch.mask_ instead" #-} -- | Lift 'Control.Exception.mask_' mask_ :: Process a -> Process a-mask_ p = do- lproc <- ask- liftIO $ Ex.mask_ $ runLocalProcess lproc p+mask_ = Catch.mask_+{-# DEPRECATED mask_ "Use Control.Monad.Catch.mask_ instead" #-} -- | Lift 'Control.Exception.onException' onException :: Process a -> Process b -> Process a-onException p what = p `catch` \e -> do _ <- what- liftIO $ throwIO (e :: SomeException)+onException = Catch.onException+{-# DEPRECATED onException "Use Control.Monad.Catch.onException instead" #-} -- | Lift 'Control.Exception.bracket' bracket :: Process a -> (a -> Process b) -> (a -> Process c) -> Process c-bracket before after thing =- mask $ \restore -> do- a <- before- r <- restore (thing a) `onException` after a- _ <- after a- return r+bracket = Catch.bracket+{-# DEPRECATED bracket "Use Control.Monad.Catch.bracket instead" #-} -- | Lift 'Control.Exception.bracket_' bracket_ :: Process a -> Process b -> Process c -> Process c-bracket_ before after thing = bracket before (const after) (const thing)+bracket_ = Catch.bracket_+{-# DEPRECATED bracket_ "Use Control.Monad.Catch.bracket_ instead" #-} -- | Lift 'Control.Exception.finally' finally :: Process a -> Process b -> Process a-finally a sequel = bracket_ (return ()) sequel a+finally = Catch.finally+{-# DEPRECATED finally "Use Control.Monad.Catch.finally instead" #-} -- | You need this when using 'catches' data Handler a = forall e . Exception e => Handler (e -> Process a)@@ -944,10 +1008,10 @@ -- | Lift 'Control.Exception.catches' catches :: Process a -> [Handler a] -> Process a-catches proc handlers = proc `catch` catchesHandler handlers+catches proc handlers = proc `Catch.catch` catchesHandler handlers catchesHandler :: [Handler a] -> SomeException -> Process a-catchesHandler handlers e = foldr tryHandler (throw e) handlers+catchesHandler handlers e = foldr tryHandler (throwM e) handlers where tryHandler (Handler handler) res = case fromException e of Just e' -> handler e'@@ -958,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'@@ -967,7 +1033,10 @@ spawnAsync :: NodeId -> Closure (Process ()) -> Process SpawnRef spawnAsync nid proc = do spawnRef <- getSpawnRef- sendCtrlMsg (Just nid) $ Spawn proc spawnRef+ node <- getSelfNode+ if nid == node+ then sendCtrlMsg Nothing $ Spawn proc spawnRef+ else sendCtrlMsg (Just nid) $ Spawn proc spawnRef return spawnRef -- | Monitor a node (asynchronous)@@ -1013,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 --@@ -1046,9 +1150,10 @@ registerImpl force label pid = do mynid <- getSelfNode sendCtrlMsg Nothing (Register label mynid (Just pid) force)- receiveWait [ matchIf (\(RegisterReply label' _) -> label == label')- (\(RegisterReply _ ok) -> handleRegistrationReply label ok)- ]+ receiveWait+ [ matchIf (\(RegisterReply label' _ _) -> label == label')+ (\(RegisterReply _ ok owner) -> handleRegistrationReply label ok owner)+ ] -- | Register a process with a remote registry (asynchronous). --@@ -1057,8 +1162,10 @@ -- -- See comments in 'whereisRemoteAsync' registerRemoteAsync :: NodeId -> String -> ProcessId -> Process ()-registerRemoteAsync nid label pid =- sendCtrlMsg (Just nid) (Register label nid (Just pid) False)+registerRemoteAsync nid label pid = do+ here <- getSelfNode+ sendCtrlMsg (if nid == here then Nothing else Just nid)+ (Register label nid (Just pid) False) reregisterRemoteAsync :: NodeId -> String -> ProcessId -> Process () reregisterRemoteAsync nid label pid =@@ -1071,16 +1178,17 @@ unregister label = do mynid <- getSelfNode sendCtrlMsg Nothing (Register label mynid Nothing False)- receiveWait [ matchIf (\(RegisterReply label' _) -> label == label')- (\(RegisterReply _ ok) -> handleRegistrationReply label ok)- ]+ receiveWait+ [ matchIf (\(RegisterReply label' _ _) -> label == label')+ (\(RegisterReply _ ok owner) -> handleRegistrationReply label ok owner)+ ] -- | Deal with the result from an attempted registration or unregistration -- by throwing an exception if necessary-handleRegistrationReply :: String -> Bool -> Process ()-handleRegistrationReply label ok =+handleRegistrationReply :: String -> Bool -> Maybe ProcessId -> Process ()+handleRegistrationReply label ok owner = when (not ok) $- liftIO $ throwIO $ ProcessRegistrationException label+ throwM $ ProcessRegistrationException label owner -- | Remove a process from a remote registry (asynchronous). --@@ -1109,13 +1217,18 @@ -- use 'monitorNode' and take appropriate action when you receive a -- 'NodeMonitorNotification'). whereisRemoteAsync :: NodeId -> String -> Process ()-whereisRemoteAsync nid label =- sendCtrlMsg (Just nid) (WhereIs label)+whereisRemoteAsync nid label = do+ here <- getSelfNode+ sendCtrlMsg (if nid == here then Nothing else Just nid) (WhereIs label) -- | Named send to a process in the local registry (asynchronous) nsend :: Serializable a => String -> a -> Process ()-nsend label msg =- sendCtrlMsg Nothing (NamedSend label (createMessage 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@@ -1126,8 +1239,23 @@ -- | Named send to a process in a remote registry (asynchronous) nsendRemote :: Serializable a => NodeId -> String -> a -> Process ()-nsendRemote nid label msg =- sendCtrlMsg (Just nid) (NamedSend label (createMessage msg))+nsendRemote nid label msg = do+ 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+-- destination process resides on the same local node) therefore ensure that+-- the payload is fully evaluated before it is delivered.+unsafeNSendRemote :: Serializable a => NodeId -> String -> a -> Process ()+unsafeNSendRemote = Unsafe.nsendRemote -------------------------------------------------------------------------------- -- Closures --
src/Control/Distributed/Process/Internal/Spawn.hs view
@@ -43,7 +43,7 @@ ) import Control.Distributed.Process.Internal.Primitives ( -- Basic messaging- send+ usend , expect , receiveWait , match@@ -71,7 +71,7 @@ receiveWait [ matchIf (\(DidSpawn ref _) -> ref == sRef) $ \(DidSpawn _ pid) -> do unmonitor mRef- send pid ()+ usend pid () return pid , matchIf (\(NodeMonitorNotification ref _ _) -> ref == mRef) $ \_ -> return (nullProcessId nid)@@ -118,7 +118,7 @@ cpDelayed us (returnCP sdictUnit ()) ) mResult <- receiveWait- [ match $ \a -> send pid () >> return (Right a)+ [ match $ \a -> usend pid () >> return (Right a) , matchIf (\(ProcessMonitorNotification ref _ _) -> ref == mRef) (\(ProcessMonitorNotification _ _ reason) -> return (Left reason)) ]
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,13 +26,10 @@ , withMVar , modifyMVar_ , modifyMVar-#if MIN_VERSION_base(4,6,0)- , modifyMVarMasked-#endif ) import GHC.MVar (MVar(MVar))-import GHC.IO (IO(IO))-import GHC.Prim (mkWeak#)+import GHC.IO (IO(IO), unIO)+import GHC.Exts (mkWeak#) import GHC.Weak (Weak(Weak)) newtype StrictMVar a = StrictMVar (MVar.MVar a)@@ -71,19 +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 m@(StrictMVar (MVar m#)) f = IO $ \s ->- case mkWeak# m# m f s of (# s1, w #) -> (# s1, Weak w #)+mkWeakMVar q@(StrictMVar (MVar m#)) f = IO $ \s ->+ case mkWeak# m# q (unIO f) s of (# s', w #) -> (# s', Weak w #)
src/Control/Distributed/Process/Internal/Types.hs view
@@ -4,6 +4,9 @@ {-# LANGUAGE GADTs #-} {-# LANGUAGE DeriveGeneric #-} {-# LANGUAGE CPP #-}+{-# LANGUAGE MultiParamTypeClasses #-}+{-# LANGUAGE TypeFamilies #-}+{-# LANGUAGE Rank2Types #-} -- | Types used throughout the Cloud Haskell framework --@@ -21,9 +24,14 @@ , nullProcessId -- * Local nodes and processes , LocalNode(..)+ , LocalNodeState(..)+ , ValidLocalNodeState(..)+ , NodeClosedException(..)+ , withValidLocalState+ , modifyValidLocalState+ , modifyValidLocalState_ , Tracer(..) , MxEventBus(..)- , LocalNodeState(..) , LocalProcess(..) , LocalProcessState(..) , Process(..)@@ -102,15 +110,17 @@ import Data.Accessor (Accessor, accessor) import Control.Category ((>>>)) import Control.DeepSeq (NFData(..))-import Control.Exception (Exception)+import Control.Exception (Exception, throwIO) import Control.Concurrent (ThreadId) import Control.Concurrent.Chan (Chan) import Control.Concurrent.STM (STM) import Control.Concurrent.STM.TChan (TChan)+import Control.Monad.Catch (MonadThrow(..), MonadCatch(..), MonadMask(..)) import qualified Network.Transport as NT (EndPoint, EndPointAddress, Connection)-import Control.Applicative (Applicative, Alternative, (<$>), (<*>))+import Control.Applicative+import Control.Monad.Fix (MonadFix) import Control.Monad.Reader (MonadReader(..), ReaderT, runReaderT)-import Control.Monad.IO.Class (MonadIO)+import Control.Monad.IO.Class (MonadIO(..)) import Control.Distributed.Process.Serializable ( Fingerprint , Serializable@@ -121,13 +131,19 @@ , showFingerprint ) import Control.Distributed.Process.Internal.CQueue (CQueue)-import Control.Distributed.Process.Internal.StrictMVar (StrictMVar)+import Control.Distributed.Process.Internal.StrictMVar+ ( StrictMVar+ , withMVar+ , modifyMVar+ , modifyMVar_+ ) import Control.Distributed.Process.Internal.WeakTQueue (TQueue) import Control.Distributed.Static (RemoteTable, Closure) import qualified Control.Distributed.Process.Internal.StrictContainerAccessors as DAC (mapMaybe) import Data.Hashable import GHC.Generics+import Prelude -------------------------------------------------------------------------------- -- Node and process identifiers --@@ -257,7 +273,11 @@ deriving (Eq, Show) -- | Local node state-data LocalNodeState = LocalNodeState+data LocalNodeState =+ LocalNodeValid {-# UNPACK #-} !ValidLocalNodeState+ | LocalNodeClosed++data ValidLocalNodeState = ValidLocalNodeState { -- | Processes running on this node _localProcesses :: !(Map LocalProcessId LocalProcess) -- | Counter to assign PIDs@@ -270,6 +290,40 @@ (NT.Connection, ImplicitReconnect)) } +-- | Thrown by some primitives when they notice the node has been closed.+data NodeClosedException = NodeClosedException NodeId+ deriving (Show, Typeable)++instance Exception NodeClosedException++-- | Wrapper around 'withMVar' that checks that the local node is still in+-- a valid state.+withValidLocalState :: LocalNode+ -> (ValidLocalNodeState -> IO r)+ -> IO r+withValidLocalState node f = withMVar (localState node) $ \st -> case st of+ LocalNodeValid vst -> f vst+ LocalNodeClosed -> throwIO $ NodeClosedException (localNodeId node)++-- | Wrapper around 'modifyMVar' that checks that the local node is still in+-- a valid state.+modifyValidLocalState :: LocalNode+ -> (ValidLocalNodeState -> IO (ValidLocalNodeState, a))+ -> IO (Maybe a)+modifyValidLocalState node f = modifyMVar (localState node) $ \st -> case st of+ LocalNodeValid vst -> do (vst', a) <- f vst+ return (LocalNodeValid vst', Just a)+ LocalNodeClosed -> return (LocalNodeClosed, Nothing)++-- | Wrapper around 'modifyMVar_' that checks that the local node is still in+-- a valid state.+modifyValidLocalState_ :: LocalNode+ -> (ValidLocalNodeState -> IO ValidLocalNodeState)+ -> IO ()+modifyValidLocalState_ node f = modifyMVar_ (localState node) $ \st -> case st of+ LocalNodeValid vst -> LocalNodeValid <$> f vst+ LocalNodeClosed -> return LocalNodeClosed+ -- | Processes running on our local node data LocalProcess = LocalProcess { processQueue :: !(CQueue Message)@@ -296,8 +350,51 @@ newtype Process a = Process { unProcess :: ReaderT LocalProcess IO a }- deriving (Functor, Monad, MonadIO, MonadReader LocalProcess, Typeable, Applicative)+ deriving ( Applicative+ , Functor+ , Monad+ , MonadFail+ , MonadFix+ , MonadIO+ , MonadReader LocalProcess+ , Typeable+ ) +instance MonadThrow Process where+ throwM = liftIO . throwIO+instance MonadCatch Process where+ catch p h = do+ 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 ->+ runLocalProcess lproc (p (liftRestore restore))+ where+ liftRestore :: (forall a. IO a -> IO a)+ -> (forall a. Process a -> Process a)+ liftRestore restoreIO = \p2 -> do+ ourLocalProc <- ask+ liftIO $ restoreIO $ runLocalProcess ourLocalProc p2+ uninterruptibleMask p = do+ lproc <- ask+ liftIO $ uninterruptibleMask $ \restore ->+ runLocalProcess lproc (p (liftRestore restore))+ where+ liftRestore :: (forall a. IO a -> IO a)+ -> (forall a. Process a -> Process a)+ liftRestore restoreIO = \p2 -> do+ ourLocalProc <- ask+ liftIO $ restoreIO $ runLocalProcess ourLocalProc p2+ -------------------------------------------------------------------------------- -- Typed channels -- --------------------------------------------------------------------------------@@ -374,12 +471,9 @@ 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) = a `seq` () -- forced to WHNF only+ rnf (UnencodedMessage _ a) = e `seq` ()+ where e = BSL.length (encode a) instance Show Message where show (EncodedMessage fp enc) = show enc ++ " :: " ++ showFingerprint fp []@@ -481,9 +575,11 @@ -- | Exception thrown when a process attempts to register -- a process under an already-registered name or to--- unregister a name that hasn't been registered+-- unregister a name that hasn't been registered. Returns+-- the name and the identifier of the process that owns it,+-- if any. data ProcessRegistrationException =- ProcessRegistrationException !String+ ProcessRegistrationException !String !(Maybe ProcessId) deriving (Typeable, Show) -- | Internal exception thrown indirectly by 'exit'@@ -537,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@@ -548,7 +644,7 @@ deriving (Show, Typeable) -- | (Asynchronous) reply from 'register' and 'unregister'-data RegisterReply = RegisterReply String Bool+data RegisterReply = RegisterReply String Bool (Maybe ProcessId) deriving (Show, Typeable) data NodeStats = NodeStats {@@ -564,7 +660,7 @@ data ProcessInfo = ProcessInfo { infoNode :: NodeId , infoRegisteredNames :: [String]- , infoMessageQueueLength :: Maybe Int+ , infoMessageQueueLength :: Int , infoMonitors :: [(ProcessId, MonitorRef)] , infoLinks :: [ProcessId] } deriving (Show, Eq, Typeable)@@ -594,6 +690,7 @@ | WhereIs !String | Register !String !NodeId !(Maybe ProcessId) !Bool -- Use 'Nothing' to unregister, use True to force reregister | NamedSend !String !Message+ | UnreliableSend !LocalProcessId !Message | LocalSend !ProcessId !Message | LocalPortSend !SendPortId !Message | Kill !ProcessId !String@@ -649,6 +746,7 @@ put (Exit pid reason) = putWord8 10 >> put pid >> put (messageToPayload reason) put (LocalSend to' msg) = putWord8 11 >> put to' >> put (messageToPayload msg) put (LocalPortSend sid msg) = putWord8 12 >> put sid >> put (messageToPayload msg)+ put (UnreliableSend lpid msg) = putWord8 13 >> put lpid >> put (messageToPayload msg) put (GetInfo about) = putWord8 30 >> put about put (SigShutdown) = putWord8 31 put (GetNodeStats nid) = putWord8 32 >> put nid@@ -668,6 +766,7 @@ 10 -> Exit <$> get <*> (payloadToMessage <$> get) 11 -> LocalSend <$> get <*> (payloadToMessage <$> get) 12 -> LocalPortSend <$> get <*> (payloadToMessage <$> get)+ 13 -> UnreliableSend <$> get <*> (payloadToMessage <$> get) 30 -> GetInfo <$> get 31 -> return SigShutdown 32 -> GetNodeStats <$> get@@ -714,8 +813,8 @@ get = WhereIsReply <$> get <*> get instance Binary RegisterReply where- put (RegisterReply label ok) = put label >> put ok- get = RegisterReply <$> get <*> get+ put (RegisterReply label ok owner) = put label >> put ok >> put owner+ get = RegisterReply <$> get <*> get <*> get instance Binary ProcessInfo where get = ProcessInfo <$> get <*> get <*> get <*> get <*> get@@ -741,22 +840,22 @@ -- Accessors -- -------------------------------------------------------------------------------- -localProcesses :: Accessor LocalNodeState (Map LocalProcessId LocalProcess)+localProcesses :: Accessor ValidLocalNodeState (Map LocalProcessId LocalProcess) localProcesses = accessor _localProcesses (\procs st -> st { _localProcesses = procs }) -localPidCounter :: Accessor LocalNodeState Int32+localPidCounter :: Accessor ValidLocalNodeState Int32 localPidCounter = accessor _localPidCounter (\ctr st -> st { _localPidCounter = ctr }) -localPidUnique :: Accessor LocalNodeState Int32+localPidUnique :: Accessor ValidLocalNodeState Int32 localPidUnique = accessor _localPidUnique (\unq st -> st { _localPidUnique = unq }) -localConnections :: Accessor LocalNodeState (Map (Identifier, Identifier) (NT.Connection, ImplicitReconnect))+localConnections :: Accessor ValidLocalNodeState (Map (Identifier, Identifier) (NT.Connection, ImplicitReconnect)) localConnections = accessor _localConnections (\conns st -> st { _localConnections = conns }) -localProcessWithId :: LocalProcessId -> Accessor LocalNodeState (Maybe LocalProcess)+localProcessWithId :: LocalProcessId -> Accessor ValidLocalNodeState (Maybe LocalProcess) localProcessWithId lpid = localProcesses >>> DAC.mapMaybe lpid -localConnectionBetween :: Identifier -> Identifier -> Accessor LocalNodeState (Maybe (NT.Connection, ImplicitReconnect))+localConnectionBetween :: Identifier -> Identifier -> Accessor ValidLocalNodeState (Maybe (NT.Connection, ImplicitReconnect)) localConnectionBetween from' to' = localConnections >>> DAC.mapMaybe (from', to') monitorCounter :: Accessor LocalProcessState Int32@@ -782,4 +881,3 @@ {-# INLINE forever' #-} forever' :: Monad m => m a -> m b forever' a = let a' = a >> a' in a'-
src/Control/Distributed/Process/Internal/WeakTQueue.hs view
@@ -1,3 +1,4 @@+{-# LANGUAGE CPP #-} {-# LANGUAGE DeriveDataTypeable #-} {-# LANGUAGE MagicHash, UnboxedTuples #-} -- | Clone of Control.Concurrent.STM.TQueue with support for mkWeakTQueue@@ -22,8 +23,8 @@ import Prelude hiding (read) import GHC.Conc import Data.Typeable (Typeable)-import GHC.IO (IO(IO))-import GHC.Prim (mkWeak#)+import GHC.IO (IO(IO), unIO)+import GHC.Exts (mkWeak#) import GHC.Weak (Weak(Weak)) --------------------------------------------------------------------------------@@ -99,4 +100,4 @@ mkWeakTQueue :: TQueue a -> IO () -> IO (Weak (TQueue a)) mkWeakTQueue q@(TQueue _read (TVar write#)) f = IO $ \s ->- case mkWeak# write# q f s of (# s', w #) -> (# s', Weak w #)+ case mkWeak# write# q (unIO f) s of (# s', w #) -> (# s', Weak w #)
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,16 +272,9 @@ , mxGetLocal , mxUpdateLocal , liftMX- -- * Mx Data API- , mxPublish- , mxSet- , mxGet- , mxClear- , mxPurgeTable- , mxDropTable ) where -import Control.Applicative ((<$>))+import Control.Applicative import Control.Concurrent.STM (atomically) import Control.Concurrent.STM.TChan ( readTChan@@ -281,14 +282,10 @@ , TChan ) import Control.Distributed.Process.Internal.Primitives- ( newChan- , nsend- , receiveWait- , matchChan+ ( receiveWait , matchAny , matchSTM , unwrapMessage- , onException , register , whereis , die@@ -303,26 +300,26 @@ , 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(..) ) import Control.Distributed.Process.Serializable (Serializable) import Control.Monad.IO.Class (liftIO) import Control.Monad.Reader (ask)+import Control.Monad.Catch (onException) import qualified Control.Monad.State as ST ( get , modify , lift , runStateT )+import Prelude -- | Publishes an arbitrary @Serializable@ message to the management event bus. -- Note that /no attempt is made to force the argument/, therefore it is very@@ -334,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) -- --------------------------------------------------------------------------------@@ -489,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/Agent.hs view
@@ -3,7 +3,7 @@ module Control.Distributed.Process.Management.Internal.Agent where -import Control.Applicative ((<$>))+import Control.Applicative import Control.Concurrent (forkIO) import Control.Concurrent.MVar (MVar, newEmptyMVar, putMVar, takeMVar) import Control.Concurrent.STM (atomically)@@ -44,6 +44,7 @@ import Control.Monad.IO.Class (liftIO) import Control.Monad.Reader (ask) import GHC.Weak (Weak, deRefWeak)+import Prelude -------------------------------------------------------------------------------- -- Agent Controller Implementation --
− 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
@@ -28,11 +28,12 @@ , withRegisteredTracer ) where -import Control.Applicative ((<$>))+import Control.Applicative import Control.Distributed.Process.Internal.Primitives ( whereis , newChan , receiveChan+ , die ) import Control.Distributed.Process.Management.Internal.Trace.Types ( TraceArg(..)@@ -67,6 +68,7 @@ import Control.Monad.Reader (ask) import qualified Data.Set as Set (fromList)+import Prelude -------------------------------------------------------------------------------- -- Main API --@@ -167,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
@@ -11,7 +11,7 @@ , eventLogTracer ) where -import Control.Applicative ((<$>))+import Control.Applicative import Control.Concurrent.Chan (writeChan) import Control.Concurrent.MVar ( MVar@@ -21,9 +21,7 @@ ( CQueue ) import Control.Distributed.Process.Internal.Primitives- ( catch- , finally- , die+ ( die , receiveWait , forward , sendChan@@ -63,6 +61,10 @@ import Control.Monad.IO.Class (liftIO) import Control.Monad.Reader (ask)+import Control.Monad.Catch+ ( catch+ , finally+ ) import Data.Set (Set) import qualified Data.Set as Set@@ -73,10 +75,7 @@ import Data.Time.Clock (getCurrentTime) import Data.Time.Format (formatTime) import Debug.Trace (traceEventIO)--#if ! MIN_VERSION_base(4,6,0)-import Prelude hiding (catch)-#endif+import Prelude import System.Environment (getEnv) import System.IO@@ -88,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 )@@ -150,7 +145,7 @@ emptyPid <- return $ (nullProcessId (localNodeId node)) traceMsg <- return $ NCMsg { ctrlMsgSender = ProcessIdentifier (emptyPid)- , ctrlMsgSignal = (NamedSend "logger"+ , ctrlMsgSignal = (NamedSend "trace.logger" (createUnencodedMessage msg)) } liftIO $ writeChan (localCtrlChan node) traceMsg
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
@@ -1,31 +1,27 @@ {-# LANGUAGE DeriveDataTypeable #-}-{-# LANGUAGE StandaloneDeriving #-} {-# LANGUAGE ExistentialQuantification #-} {-# LANGUAGE GeneralizedNewtypeDeriving #-} {-# 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 )@@ -34,6 +30,7 @@ ( MonadState , StateT )+import Control.Monad.Fix (MonadFix) import Data.Binary import Data.Typeable (Typeable) import GHC.Generics@@ -60,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@@ -100,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 } @@ -123,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
@@ -5,6 +5,7 @@ {-# LANGUAGE BangPatterns #-} {-# LANGUAGE GeneralizedNewtypeDeriving #-} {-# LANGUAGE RecordWildCards #-}+{-# LANGUAGE MagicHash #-} -- | Local nodes --@@ -20,10 +21,6 @@ -- TODO: Calls to 'sendBinary' and co (by the NC) may stall the node controller. -#if ! MIN_VERSION_base(4,6,0)-import Prelude hiding (catch)-#endif- import System.IO (fixIO, hPutStrLn, stderr) import System.Mem.Weak (Weak, deRefWeak) import qualified Data.ByteString.Lazy as BSL (fromChunks)@@ -33,50 +30,61 @@ ( 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 , insert , delete+ , map , member , toList+ , union ) import Data.Foldable (forM_) import Data.Maybe (isJust, fromJust, isNothing, catMaybes) import Data.Typeable (Typeable) import Control.Category ((>>>))-import Control.Applicative (Applicative, (<$>))-import Control.Monad (void, when)+import Control.Applicative+import Control.Monad (void, when, join) import Control.Monad.IO.Class (MonadIO, liftIO) import Control.Monad.State.Strict (MonadState, StateT, evalStateT, gets) import qualified Control.Monad.State.Strict as StateT (get, put) import Control.Monad.Reader (MonadReader, ReaderT, runReaderT, ask) import Control.Exception ( throwIO- , AsyncException(ThreadKilled) , SomeException , Exception , throwTo , uninterruptibleMask_+ , getMaskingState+ , MaskingState(..) )-import qualified Control.Exception as Exception (Handler(..), catches, finally)-import Control.Concurrent (forkIO, forkIOWithUnmask, myThreadId)+import qualified Control.Exception as Exception+ ( Handler(..)+ , catch+ , catches+ , finally+ )+import Control.Concurrent (forkIO, killThread)+import Control.Distributed.Process.Internal.BiMultiMap (BiMultiMap)+import qualified Control.Distributed.Process.Internal.BiMultiMap as BiMultiMap import Control.Distributed.Process.Internal.StrictMVar ( newMVar , withMVar , modifyMVarMasked- , modifyMVar_+ , modifyMVar , newEmptyMVar , putMVar , takeMVar- , readMVar ) import Control.Concurrent.Chan (newChan, writeChan, readChan) import qualified Control.Concurrent.MVar as MVar (newEmptyMVar, takeMVar)@@ -88,6 +96,7 @@ , enqueue , newCQueue , mkWeakCQueue+ , queueSize ) import qualified Network.Transport as NT ( Transport@@ -99,6 +108,7 @@ , TransportError(..) , address , closeEndPoint+ , Connection , ConnectionId , close , EndPointAddress@@ -118,6 +128,9 @@ , LocalNode(..) , MxEventBus(..) , LocalNodeState(..)+ , ValidLocalNodeState(..)+ , withValidLocalState+ , modifyValidLocalState , LocalProcess(..) , LocalProcessState(..) , Process(..)@@ -131,6 +144,7 @@ , localConnections , forever' , MonitorRef(..)+ , NodeClosedException(..) , ProcessMonitorNotification(..) , NodeMonitorNotification(..) , PortMonitorNotification(..)@@ -156,18 +170,17 @@ , RegisterReply(..) , WhereIsReply(..) , payloadToMessage- , messageToPayload , createUnencodedMessage+ , unsafeCreateUnencodedMessage , runLocalProcess , firstNonReservedProcessId- , ImplicitReconnect(WithImplicitReconnect,NoImplicitReconnect)+ , ImplicitReconnect(WithImplicitReconnect) ) 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@@ -181,25 +194,19 @@ , traceLogFmt , enableTrace )-import Control.Distributed.Process.Management.Internal.Types- ( MxEvent(..)- ) import Control.Distributed.Process.Serializable (Serializable) import Control.Distributed.Process.Internal.Messaging ( sendBinary- , sendMessage- , sendPayload , closeImplicitReconnections , impliesDeathOf ) import Control.Distributed.Process.Internal.Primitives ( register- , finally , receiveWait , match , sendChan- , catch , unwrapMessage+ , SayMessage(..) ) import Control.Distributed.Process.Internal.Types (SendPort, Tracer(..)) import qualified Control.Distributed.Process.Internal.Closure.BuiltIn as BuiltIn (remoteTable)@@ -208,8 +215,21 @@ ( mapMaybe , mapDefault )+import Control.Monad.Catch (try)+import GHC.IO (IO(..), unsafeUnmask)+import GHC.Base ( maskAsyncExceptions# )+ import Unsafe.Coerce+import Prelude +-- Remove these definitions when the fix for+-- https://ghc.haskell.org/trac/ghc/ticket/10149+-- is included in all supported compilers:+block :: IO a -> IO a+block (IO io) = IO $ maskAsyncExceptions# io+unblock :: IO a -> IO a+unblock = unsafeUnmask+ -------------------------------------------------------------------------------- -- Initialization -- --------------------------------------------------------------------------------@@ -232,7 +252,7 @@ createBareLocalNode :: NT.EndPoint -> RemoteTable -> IO LocalNode createBareLocalNode endPoint rtable = do unq <- randomIO- state <- newMVar LocalNodeState+ state <- newMVar $ LocalNodeValid $ ValidLocalNodeState { _localProcesses = Map.empty , _localPidCounter = firstNonReservedProcessId , _localPidUnique = unq@@ -294,17 +314,19 @@ -- 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 $ register "logger" logger+ runProcess node $ do+ register "logger" logger+ -- The trace.logger is used for tracing to the console to avoid feedback+ -- loops during tracing if the user reregisters the "logger" with a custom+ -- 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@@ -314,33 +336,48 @@ sendChan ch () ] --- | Force-close a local node------ TODO: for now we just close the associated endpoint+-- | Force-close a local node, killing all processes on that node. closeLocalNode :: LocalNode -> IO ()-closeLocalNode node =- -- TODO: close all our processes, surely!?+closeLocalNode node = do+ -- Kill processes after refilling the mvar. Otherwise, there is potential for+ -- deadlock as a dying process tries to get the mvar while masking exceptions+ -- uninterruptibly.+ join $ modifyMVar (localState node) $ \st -> case st of+ LocalNodeValid vst -> do+ return ( LocalNodeClosed+ , forM_ (vst ^. localProcesses) $ \lproc ->+ -- Semantics of 'throwTo' guarantee that target thread will get+ -- delivered an exception. Therefore, target thread will be+ -- killed eventually and that's as good as we can do. No need+ -- to wait for thread to actually finish dying.+ killThread (processThread lproc)+ )+ LocalNodeClosed -> return (LocalNodeClosed, return ())+ -- This call will have the effect of shutting down the NC as well (see+ -- 'createBareLocalNode'). NT.closeEndPoint (localEndPoint node) -- | Run a process on a local node and wait for it to finish runProcess :: LocalNode -> Process () -> IO () runProcess node proc = do done <- newEmptyMVar- tid <- myThreadId- void $ forkProcess node $ do- catch (proc `finally` liftIO (putMVar done ()))- (\(ex :: SomeException) -> liftIO $ throwTo tid ex)- takeMVar done+ -- TODO; When forkProcess inherits the masking state, protect the forked+ -- thread against async exceptions that could occur before 'try' is evaluated.+ void $ forkProcess node $ try proc >>= liftIO . putMVar done+ takeMVar done >>= either (throwIO :: SomeException -> IO a) return -- | Spawn a new process on a local node forkProcess :: LocalNode -> Process () -> IO ProcessId-forkProcess node proc =- modifyMVarMasked (localState node) startProcess+forkProcess node proc = do+ ms <- getMaskingState+ modifyMVarMasked (localState node) (startProcess ms) where- startProcess :: LocalNodeState -> IO (LocalNodeState, ProcessId)- startProcess st = do- let lpid = LocalProcessId { lpidCounter = st ^. localPidCounter- , lpidUnique = st ^. localPidUnique+ startProcess :: MaskingState+ -> LocalNodeState+ -> IO (LocalNodeState, ProcessId)+ startProcess ms (LocalNodeValid vst) = do+ let lpid = LocalProcessId { lpidCounter = vst ^. localPidCounter+ , lpidUnique = vst ^. localPidUnique } let pid = ProcessId { processNodeId = localNodeId node , processLocalId = lpid@@ -360,7 +397,13 @@ , processThread = tid , processNode = node }- tid' <- uninterruptibleMask_ $ forkIOWithUnmask $ \unmask -> do+ -- Rewrite this code when this is fixed:+ -- https://ghc.haskell.org/trac/ghc/ticket/10149+ let unmask = case ms of+ Unmasked -> unblock+ MaskedInterruptible -> block+ MaskedUninterruptible -> id+ tid' <- uninterruptibleMask_ $ forkIO $ do reason <- Exception.catches (unmask $ runLocalProcess lproc proc >> return DiedNormal) [ (Exception.Handler (\ex@(ProcessExitException from msg) -> do@@ -372,7 +415,12 @@ (return . DiedException . (show :: SomeException -> String)))] -- [Unified: Table 4, rules termination and exiting]- modifyMVar_ (localState node) (cleanupProcess pid)+ mconns <- modifyValidLocalState node (cleanupProcess pid)+ -- XXX: Revisit after agreeing on the bigger picture for the semantics+ -- of transport operations.+ -- https://github.com/haskell-distributed/distributed-process/issues/204+ forM_ mconns $ forkIO . mapM_ NT.close+ writeChan (localCtrlChan node) NCMsg { ctrlMsgSender = ProcessIdentifier pid , ctrlMsgSignal = Died (ProcessIdentifier pid) reason@@ -387,27 +435,34 @@ -- TODO: this doesn't look right at all - how do we know -- that newUnique represents a process id that is available!? newUnique <- randomIO- return ( (localProcessWithId lpid ^= Just lproc)+ return ( LocalNodeValid+ $ (localProcessWithId lpid ^= Just lproc) . (localPidCounter ^= firstNonReservedProcessId) . (localPidUnique ^= newUnique)- $ st+ $ vst , pid ) else- return ( (localProcessWithId lpid ^= Just lproc)+ return ( LocalNodeValid+ $ (localProcessWithId lpid ^= Just lproc) . (localPidCounter ^: (+ 1))- $ st+ $ vst , pid )+ startProcess _ LocalNodeClosed =+ throwIO $ NodeClosedException $ localNodeId node - cleanupProcess :: ProcessId -> LocalNodeState -> IO LocalNodeState- cleanupProcess pid st = do+ cleanupProcess :: ProcessId+ -> ValidLocalNodeState+ -> IO (ValidLocalNodeState, [NT.Connection])+ cleanupProcess pid vst = do let pid' = ProcessIdentifier pid- let (affected, unaffected) = Map.partitionWithKey (\(fr, _to) !_v -> impliesDeathOf pid' fr) (st ^. localConnections)- mapM_ (NT.close . fst) (Map.elems affected)- return $ (localProcessWithId (processLocalId pid) ^= Nothing)- . (localConnections ^= unaffected)- $ st+ let (affected, unaffected) = Map.partitionWithKey (\(fr, _to) !_v -> impliesDeathOf pid' fr) (vst ^. localConnections)+ return ( (localProcessWithId (processLocalId pid) ^= Nothing)+ . (localConnections ^= unaffected)+ $ vst+ , map fst $ Map.elems affected+ ) -- note [tracer/forkProcess races] --@@ -427,7 +482,7 @@ data IncomingTarget = Uninit | ToProc ProcessId (Weak (CQueue Message))- | ToChan TypedChannel+ | ToChan SendPortId TypedChannel | ToNode data ConnectionState = ConnectionState {@@ -454,6 +509,7 @@ handleIncomingMessages :: LocalNode -> IO () handleIncomingMessages node = go initConnectionState+ `Exception.catch` \(NodeClosedException _) -> return () where go :: ConnectionState -> IO () go !st = do@@ -468,7 +524,8 @@ . (incomingFrom theirAddr ^: Set.insert cid) $ st )- else invalidRequest cid st+ else invalidRequest cid st $+ "attempt to connect with unsupported reliability " ++ show rel NT.Received cid payload -> case st ^. incomingAt cid of Just (_, ToProc pid weakQueue) -> do@@ -480,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@@ -496,36 +557,47 @@ case decode (BSL.fromChunks payload) of ProcessIdentifier pid -> do let lpid = processLocalId pid- mProc <- withMVar state $ return . (^. localProcessWithId lpid)+ mProc <- withValidLocalState node $ return . (^. localProcessWithId lpid) case mProc of Just proc -> go (incomingAt cid ^= Just (src, ToProc pid (processWeakQ proc)) $ st) Nothing ->- invalidRequest cid st+ -- incoming attempt to connect to unknown process - might+ -- be dead already+ go (incomingAt cid ^= Nothing $ st) SendPortIdentifier chId -> do let lcid = sendPortLocalId chId lpid = processLocalId (sendPortProcessId chId)- mProc <- withMVar state $ return . (^. localProcessWithId lpid)+ mProc <- withValidLocalState node $ return . (^. localProcessWithId lpid) case mProc of Just proc -> do 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+ invalidRequest cid st $+ "incoming attempt to connect to unknown channel of"+ ++ " process " ++ show (sendPortProcessId chId) Nothing ->- invalidRequest cid st+ -- incoming attempt to connect to channel of unknown+ -- process - might be dead already+ go (incomingAt cid ^= Nothing $ st) NodeIdentifier nid -> if nid == localNodeId node then go (incomingAt cid ^= Just (src, ToNode) $ st)- else invalidRequest cid st+ else invalidRequest cid st $+ "incoming attempt to connect to a different node -"+ ++ " I'm " ++ show (localNodeId node)+ ++ " but the remote peer wants to connect to "+ ++ show nid Nothing -> invalidRequest cid st+ "message received from an unknown connection" NT.ConnectionClosed cid -> case st ^. incomingAt cid of Nothing ->- invalidRequest cid st+ invalidRequest cid st "closed unknown connection" Just (src, _) -> do trace node (MxDisconnected cid src) go ( (incomingAt cid ^= Nothing)@@ -556,19 +628,20 @@ -- and we just give up fail "Cloud Haskell fatal error: received unexpected multicast" - invalidRequest :: NT.ConnectionId -> ConnectionState -> IO ()- invalidRequest cid st = do+ invalidRequest :: NT.ConnectionId -> ConnectionState -> String -> IO ()+ invalidRequest cid st msg = do -- TODO: We should treat this as a fatal error on the part of the remote -- node. That is, we should report the remote node as having died, and we -- should close incoming connections (this requires a Transport layer -- extension).- traceEventFmtIO node "" [(TraceStr " [network] invalid request: "),- (Trace cid)]+ traceEventFmtIO node "" [ TraceStr $ " [network] invalid request"+ ++ " (" ++ msg ++ "): "+ , (Trace cid)+ ] go ( incomingAt cid ^= Nothing $ st ) - state = localState node endpoint = localEndPoint node ctrlChan = localCtrlChan node @@ -577,8 +650,9 @@ -------------------------------------------------------------------------------- runNodeController :: LocalNode -> IO ()-runNodeController =- runReaderT (evalStateT (unNC nodeController) initNCState)+runNodeController node =+ runReaderT (evalStateT (unNC nodeController) initNCState) node+ `Exception.catch` \(NodeClosedException _) -> return () -------------------------------------------------------------------------------- -- Internal data types --@@ -586,9 +660,9 @@ data NCState = NCState { -- Mapping from remote processes to linked local processes- _links :: !(Map Identifier (Set ProcessId))+ _links :: !(BiMultiMap Identifier ProcessId ()) -- Mapping from remote processes to monitoring local processes- , _monitors :: !(Map Identifier (Set (ProcessId, MonitorRef)))+ , _monitors :: !(BiMultiMap Identifier ProcessId MonitorRef) -- Process registry: names and where they live, mapped to the PIDs , _registeredHere :: !(Map String ProcessId) , _registeredOnNodes :: !(Map ProcessId [(NodeId,Int)])@@ -604,8 +678,8 @@ ) initNCState :: NCState-initNCState = NCState { _links = Map.empty- , _monitors = Map.empty+initNCState = NCState { _links = BiMultiMap.empty+ , _monitors = BiMultiMap.empty , _registeredHere = Map.empty , _registeredOnNodes = Map.empty }@@ -621,6 +695,33 @@ show (ProcessKillException pid reason) = "killed-by=" ++ show pid ++ ",reason=" ++ reason +ncSendToProcess :: ProcessId -> Message -> NC ()+ncSendToProcess = ncSendToProcessAndTrace True++ncSendToProcessAndTrace :: Bool -> ProcessId -> Message -> NC ()+ncSendToProcessAndTrace shouldTrace pid msg = do+ node <- ask+ if processNodeId pid == localNodeId node+ then ncEffectLocalSendAndTrace shouldTrace node pid msg+ else liftIO $ sendBinary node+ (NodeIdentifier $ localNodeId node)+ (NodeIdentifier $ processNodeId pid)+ WithImplicitReconnect+ NCMsg { ctrlMsgSender = NodeIdentifier $ localNodeId node+ , ctrlMsgSignal = UnreliableSend (processLocalId pid) msg+ }++ncSendToNode :: NodeId -> NCMsg -> NC ()+ncSendToNode to msg = do+ node <- ask+ liftIO $ if to == localNodeId node+ then writeChan (localCtrlChan node) $! msg+ else sendBinary node+ (NodeIdentifier $ localNodeId node)+ (NodeIdentifier to)+ WithImplicitReconnect+ msg+ -------------------------------------------------------------------------------- -- Tracing/Debugging -- --------------------------------------------------------------------------------@@ -663,11 +764,7 @@ -- [Unified: Table 7, rule nc_forward] case destNid (ctrlMsgSignal msg) of Just nid' | nid' /= localNodeId node ->- liftIO $ sendBinary node- (ctrlMsgSender msg)- (NodeIdentifier nid')- WithImplicitReconnect- msg+ ncSendToNode nid' msg _ -> return () @@ -688,8 +785,10 @@ ncEffectRegister from label atnode pid force NCMsg (ProcessIdentifier from) (WhereIs label) -> ncEffectWhereIs from label- NCMsg (ProcessIdentifier from) (NamedSend label msg') ->- ncEffectNamedSend from label msg'+ NCMsg _ (NamedSend label msg') ->+ ncEffectNamedSend label msg'+ NCMsg _ (UnreliableSend lpid msg') ->+ ncEffectLocalSend node (ProcessId (localNodeId node) lpid) msg' NCMsg _ (LocalSend to msg') -> ncEffectLocalSend node to msg' NCMsg _ (LocalPortSend to msg') ->@@ -703,8 +802,7 @@ NCMsg _ SigShutdown -> liftIO $ do NT.closeEndPoint (localEndPoint node)- `Exception.finally` throwIO ThreadKilled- -- ThreadKilled seems to make more sense than fail/error here+ `Exception.finally` throwIO (NodeClosedException $ localNodeId node) NCMsg (ProcessIdentifier from) (GetNodeStats nid) -> ncEffectGetNodeStats from nid unexpected ->@@ -724,22 +822,18 @@ case (shouldLink, isLocal node (ProcessIdentifier from)) of (True, _) -> -- [Unified: first rule] case mRef of- Just ref -> modify' $ monitorsFor them ^: Set.insert (from, ref)- Nothing -> modify' $ linksFor them ^: Set.insert from+ Just ref -> modify' $ monitors ^: BiMultiMap.insert them from ref+ Nothing -> modify' $ links ^: BiMultiMap.insert them from () (False, True) -> -- [Unified: second rule] notifyDied from them DiedUnknownId mRef (False, False) -> -- [Unified: third rule] -- TODO: this is the right sender according to the Unified semantics, -- but perhaps having 'them' as the sender would make more sense -- (see also: notifyDied)- liftIO $ sendBinary node- (NodeIdentifier $ localNodeId node)- (NodeIdentifier $ processNodeId from)- WithImplicitReconnect- NCMsg- { ctrlMsgSender = NodeIdentifier (localNodeId node)- , ctrlMsgSignal = Died them DiedUnknownId- }+ ncSendToNode (processNodeId from) $ NCMsg+ { ctrlMsgSender = NodeIdentifier (localNodeId node)+ , ctrlMsgSignal = Died them DiedUnknownId+ } -- [Unified: Table 11] ncEffectUnlink :: ProcessId -> Identifier -> NC ()@@ -753,7 +847,7 @@ postAsMessage from $ DidUnlinkNode nid SendPortIdentifier cid -> postAsMessage from $ DidUnlinkPort cid- modify' $ linksFor them ^: Set.delete from+ modify' $ links ^: BiMultiMap.delete them from () -- [Unified: Table 11] ncEffectUnmonitor :: ProcessId -> MonitorRef -> NC ()@@ -761,7 +855,7 @@ node <- ask when (isLocal node (ProcessIdentifier from)) $ postAsMessage from $ DidUnmonitor ref- modify' $ monitorsFor (monitorRefIdent ref) ^: Set.delete (from, ref)+ modify' $ monitors ^: BiMultiMap.delete (monitorRefIdent ref) from ref -- [Unified: Table 12] ncEffectDied :: Identifier -> DiedReason -> NC ()@@ -776,7 +870,7 @@ let localOnly = case ident of NodeIdentifier _ -> True ; _ -> False forM_ (Map.toList affectedLinks) $ \(them, uss) ->- forM_ uss $ \us ->+ forM_ uss $ \(us, _) -> when (localOnly <= isLocal node (ProcessIdentifier us)) $ notifyDied us them reason Nothing @@ -785,30 +879,51 @@ when (localOnly <= isLocal node (ProcessIdentifier us)) $ notifyDied us them reason (Just ref) - modify' $ (links ^= unaffectedLinks) . (monitors ^= unaffectedMons)+ -- Notify remote nodes that the process died so it can be removed from monitor+ -- lists.+ mapM_ (forwardDeath node) $+ [ nid | ProcessIdentifier pid <- [ident]+ , i <- Set.toList $ Set.union+ (Set.map fst $ BiMultiMap.lookupBy2nd pid unaffectedLinks)+ (Set.map fst $ BiMultiMap.lookupBy2nd pid unaffectedMons)+ , let nid = nodeOf i+ , nid /= localNodeId node+ ] - modify' $ registeredHere ^: Map.filter (\pid -> not $ ident `impliesDeathOf` ProcessIdentifier pid)+ -- Delete monitors in the local node.+ let deleteDeads :: (Ord a, Ord v)+ => BiMultiMap a ProcessId v -> BiMultiMap a ProcessId v+ deleteDeads = case ident of+ -- deleteAllBy2nd is faster than partitionWithKeyBy2nd+ ProcessIdentifier pid -> BiMultiMap.deleteAllBy2nd pid+ _ -> snd . BiMultiMap.partitionWithKeyBy2nd+ (\pid _ -> ident `impliesDeathOf` ProcessIdentifier pid)+ unaffectedLinks' = deleteDeads unaffectedLinks+ unaffectedMons' = deleteDeads unaffectedMons + modify' $ (links ^= unaffectedLinks') . (monitors ^= unaffectedMons')++ -- 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) -> case ident `impliesDeathOf` ProcessIdentifier pid of True -> do forM_ nidlist $ \(nid,_) -> when (not $ isLocal node (NodeIdentifier nid))- (forwardNameDeath node nid)+ (forwardDeath node nid) return Nothing False -> return $ Just (pid,nidlist) ) modify' $ registeredOnNodes ^= (Map.fromList (catMaybes remaining)) where- forwardNameDeath node nid =- liftIO $ sendBinary node- (NodeIdentifier $ localNodeId node)- (NodeIdentifier $ nid)- WithImplicitReconnect- NCMsg- { ctrlMsgSender = NodeIdentifier (localNodeId node)- , ctrlMsgSignal = Died ident reason- }+ forwardDeath node nid = ncSendToNode nid+ NCMsg { ctrlMsgSender = NodeIdentifier (localNodeId node)+ , ctrlMsgSignal = Died ident reason+ } -- [Unified: Table 13] ncEffectSpawn :: ProcessId -> Closure (Process ()) -> SpawnRef -> NC ()@@ -822,11 +937,7 @@ Right p -> p node <- ask pid' <- liftIO $ forkProcess node proc- liftIO $ sendMessage node- (NodeIdentifier (localNodeId node))- (ProcessIdentifier pid)- WithImplicitReconnect- (DidSpawn ref pid')+ ncSendToProcess pid $ unsafeCreateUnencodedMessage $ DidSpawn ref pid' -- Unified semantics does not explicitly describe how to implement 'register', -- but mentions it's "very similar to nsend" (Table 14)@@ -844,17 +955,19 @@ return $ (isNothing currentVal /= reregistration) && (not (isLocal node (ProcessIdentifier thepid) ) || isvalidlocal ) if isLocal node (NodeIdentifier atnode)- then do when (isOk) $+ then do when isOk $ 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)- liftIO $ sendMessage node- (NodeIdentifier (localNodeId node))- (ProcessIdentifier from)- WithImplicitReconnect- (RegisterReply label isOk)+ newVal <- gets (^. registeredHereFor label)+ ncSendToProcess from $ unsafeCreateUnencodedMessage $+ RegisterReply label isOk newVal else let operation = case reregistration of True -> flip decList@@ -884,46 +997,36 @@ decList (x:xs) tag = x:decList xs tag forward node to reg = when (not $ isLocal node (NodeIdentifier to)) $- liftIO $ sendBinary node- (ProcessIdentifier from)- (NodeIdentifier to)- WithImplicitReconnect- NCMsg- { ctrlMsgSender = ProcessIdentifier from- , ctrlMsgSignal = reg- }+ ncSendToNode to $ NCMsg { ctrlMsgSender = ProcessIdentifier from+ , ctrlMsgSignal = reg+ } -- Unified semantics does not explicitly describe 'whereis' ncEffectWhereIs :: ProcessId -> String -> NC () ncEffectWhereIs from label = do- node <- ask mPid <- gets (^. registeredHereFor label)- liftIO $ sendMessage node- (NodeIdentifier (localNodeId node))- (ProcessIdentifier from)- WithImplicitReconnect- (WhereIsReply label mPid)+ ncSendToProcess from $ unsafeCreateUnencodedMessage $ WhereIsReply label mPid -- [Unified: Table 14]-ncEffectNamedSend :: ProcessId -> String -> Message -> NC ()-ncEffectNamedSend from label msg = do+ncEffectNamedSend :: String -> Message -> NC ()+ncEffectNamedSend label msg = do mPid <- gets (^. registeredHereFor label)- node <- ask -- If mPid is Nothing, we just ignore the named send (as per Table 14)- forM_ mPid $ \pid ->- liftIO $ sendPayload node- (ProcessIdentifier from)- (ProcessIdentifier pid)- NoImplicitReconnect- (messageToPayload msg)+ forM_ mPid $ \to ->+ -- If this is a trace message we don't trace it to avoid entering a loop+ -- where trace messages produce more trace messages.+ ncSendToProcessAndTrace (label /= "trace.logger") to msg -- [Issue #DP-20] ncEffectLocalSend :: LocalNode -> ProcessId -> Message -> NC ()-ncEffectLocalSend node to msg =+ncEffectLocalSend = ncEffectLocalSendAndTrace True++ncEffectLocalSendAndTrace :: Bool -> LocalNode -> ProcessId -> Message -> NC ()+ncEffectLocalSendAndTrace shouldTrace node to msg = liftIO $ withLocalProc node to $ \p -> do enqueue (processQueue p) msg- trace node (MxReceived to msg)+ when shouldTrace $ trace node (MxReceived to msg) -- [Issue #DP-20] ncEffectLocalPortSend :: SendPortId -> Message -> NC ()@@ -941,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" @@ -970,41 +1074,36 @@ them = (ProcessIdentifier pid) in do node <- ask- mProc <- liftIO $- withMVar (localState node) $ return . (^. localProcessWithId lpid)+ mProc <- liftIO $ withValidLocalState node+ $ return . (^. localProcessWithId lpid) case mProc of Nothing -> dispatch (isLocal node (ProcessIdentifier from))- from node (ProcessInfoNone DiedUnknownId)- Just _ -> do- itsLinks <- gets (^. linksFor them)- itsMons <- gets (^. monitorsFor them)+ from (ProcessInfoNone DiedUnknownId)+ Just proc -> do+ itsLinks <- Set.map fst . BiMultiMap.lookupBy1st them <$>+ gets (^. links)+ itsMons <- BiMultiMap.lookupBy1st them <$> gets (^. monitors) registered <- gets (^. registeredHere)+ size <- liftIO $ queueSize $ processQueue $ proc let reg = registeredNames registered dispatch (isLocal node (ProcessIdentifier from)) from- node ProcessInfo { infoNode = (processNodeId pid) , infoRegisteredNames = reg- -- we cannot populate this field- , infoMessageQueueLength = Nothing- , infoMonitors = Set.toList itsMons- , infoLinks = Set.toList itsLinks+ , infoMessageQueueLength = size+ , infoMonitors = Set.toList itsMons+ , infoLinks = Set.toList itsLinks }- where dispatch :: (Serializable a, Show a)+ where dispatch :: (Serializable a) => Bool -> ProcessId- -> LocalNode -> a -> NC ()- dispatch True dest _ pInfo = postAsMessage dest $ pInfo- dispatch False dest node pInfo = do- liftIO $ sendMessage node- (NodeIdentifier (localNodeId node))- (ProcessIdentifier dest)- WithImplicitReconnect- pInfo+ dispatch True dest pInfo = postAsMessage dest $ pInfo+ dispatch False dest pInfo =+ ncSendToProcess dest $ unsafeCreateUnencodedMessage pInfo registeredNames = Map.foldlWithKey (\ks k v -> if v == pid then (k:ks)@@ -1014,17 +1113,17 @@ ncEffectGetNodeStats from _nid = do node <- ask ncState <- StateT.get- nodeState <- liftIO $ readMVar (localState node)- let localProcesses' = nodeState ^. localProcesses- stats =+ nodeState <- liftIO $ withValidLocalState node return+ let stats = NodeStats { nodeStatsNode = localNodeId node , nodeStatsRegisteredNames = Map.size $ ncState ^. registeredHere- , nodeStatsMonitors = Map.size $ ncState ^. monitors- , nodeStatsLinks = Map.size $ ncState ^. links- , nodeStatsProcesses = Map.size localProcesses'+ , nodeStatsMonitors = BiMultiMap.size $ ncState ^. monitors+ , nodeStatsLinks = BiMultiMap.size $ ncState ^. links+ , nodeStatsProcesses = Map.size (nodeState ^. localProcesses) } postAsMessage from stats+ -------------------------------------------------------------------------------- -- Auxiliary -- --------------------------------------------------------------------------------@@ -1051,14 +1150,10 @@ throwException dest $ PortLinkException pid reason (False, _, _) -> -- The change in sender comes from [Unified: Table 10]- liftIO $ sendBinary node- (NodeIdentifier $ localNodeId node)- (NodeIdentifier $ processNodeId dest)- WithImplicitReconnect- NCMsg- { ctrlMsgSender = NodeIdentifier (localNodeId node)- , ctrlMsgSignal = Died src reason- }+ ncSendToNode (processNodeId dest) $ NCMsg+ { ctrlMsgSender = NodeIdentifier (localNodeId node)+ , ctrlMsgSignal = Died src reason+ } -- | [Unified: Table 8] destNid :: ProcessSignal -> Maybe NodeId@@ -1070,6 +1165,7 @@ destNid (Register _ _ _ _) = Nothing destNid (WhereIs _) = Nothing destNid (NamedSend _ _) = Nothing+destNid (UnreliableSend _ _) = Nothing -- We don't need to forward 'Died' signals; if monitoring/linking is setup, -- then when a local process dies the monitoring/linking machinery will take -- care of notifying remote nodes@@ -1096,7 +1192,7 @@ isValidLocalIdentifier :: Identifier -> NC Bool isValidLocalIdentifier ident = do node <- ask- liftIO . withMVar (localState node) $ \nSt ->+ liftIO . withValidLocalState node $ \nSt -> case ident of NodeIdentifier nid -> return $ nid == localNodeId node@@ -1127,6 +1223,9 @@ throwException :: Exception e => ProcessId -> e -> NC () throwException pid e = do node <- ask+ -- throwTo blocks until the exception is received by the target thread.+ -- We cannot easily make it happen asynchronpusly because then 'unlink'+ -- semantics would break. liftIO $ withLocalProc node pid $ \p -> throwTo (processThread p) e withLocalProc :: LocalNode -> ProcessId -> (LocalProcess -> IO ()) -> IO ()@@ -1134,17 +1233,17 @@ -- By [Unified: table 6, rule missing_process] messages to dead processes -- can silently be dropped let lpid = processLocalId pid in do- mProc <- withMVar (localState node) $ return . (^. localProcessWithId lpid)- forM_ mProc p+ join $ withValidLocalState node $ \vst ->+ return $ forM_ (vst ^. localProcessWithId lpid) p -------------------------------------------------------------------------------- -- Accessors -- -------------------------------------------------------------------------------- -links :: Accessor NCState (Map Identifier (Set ProcessId))+links :: Accessor NCState (BiMultiMap Identifier ProcessId ()) links = accessor _links (\ls st -> st { _links = ls }) -monitors :: Accessor NCState (Map Identifier (Set (ProcessId, MonitorRef)))+monitors :: Accessor NCState (BiMultiMap Identifier ProcessId MonitorRef) monitors = accessor _monitors (\ms st -> st { _monitors = ms }) registeredHere :: Accessor NCState (Map String ProcessId)@@ -1153,12 +1252,6 @@ registeredOnNodes :: Accessor NCState (Map ProcessId [(NodeId, Int)]) registeredOnNodes = accessor _registeredOnNodes (\ry st -> st { _registeredOnNodes = ry }) -linksFor :: Identifier -> Accessor NCState (Set ProcessId)-linksFor ident = links >>> DAC.mapDefault Set.empty ident--monitorsFor :: Identifier -> Accessor NCState (Set (ProcessId, MonitorRef))-monitorsFor ident = monitors >>> DAC.mapDefault Set.empty ident- registeredHereFor :: String -> Accessor NCState (Maybe ProcessId) registeredHereFor ident = registeredHere >>> DAC.mapMaybe ident @@ -1185,10 +1278,12 @@ -- * the notifications for typed channels to that process. -- -- See https://github.com/haskell/containers/issues/14 for the bang on _v.-splitNotif :: Identifier- -> Map Identifier a- -> (Map Identifier a, Map Identifier a)-splitNotif ident = Map.partitionWithKey (\k !_v -> ident `impliesDeathOf` k)+splitNotif :: (Ord a, Ord v)+ => Identifier+ -> BiMultiMap Identifier a v+ -> (Map Identifier (Set (a,v)), BiMultiMap Identifier a v)+splitNotif ident =+ BiMultiMap.partitionWithKeyBy1st (\k !_v -> ident `impliesDeathOf` k) -------------------------------------------------------------------------------- -- Auxiliary --
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,25 +18,17 @@ 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) import GHC.Fingerprint.Type (Fingerprint(..)) import Data.ByteString (ByteString) import qualified Data.ByteString as BS-import qualified Data.ByteString.Internal as BSI ( unsafeCreate- , inlinePerformIO- , toForeignPtr- )+import qualified Data.ByteString.Internal as BSI ( unsafeCreate, toForeignPtr ) import Foreign.Storable (pokeByteOff, peekByteOff, sizeOf) import Foreign.ForeignPtr (withForeignPtr)+import System.IO.Unsafe (unsafePerformIO) -- | Reification of 'Serializable' (see "Control.Distributed.Process.Closure") data SerializableDict a where@@ -48,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@@ -63,7 +55,7 @@ decodeFingerprint bs | BS.length bs /= sizeOfFingerprint = throw $ userError "decodeFingerprint: Invalid length"- | otherwise = BSI.inlinePerformIO $ do+ | otherwise = unsafePerformIO $ do let (fp, offset, _) = BSI.toForeignPtr bs withForeignPtr fp $ \p -> peekByteOff p offset @@ -73,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"@@ -45,6 +62,8 @@ send , sendChan , nsend+ , nsendRemote+ , usend , wrapMessage ) where @@ -53,9 +72,15 @@ , 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(..) , LocalNode(..) , LocalProcess(..) , Process(..)@@ -65,6 +90,7 @@ , ImplicitReconnect(..) , SendPortId(..) , Message+ , createMessage , sendPortProcessId , unsafeCreateUnencodedMessage )@@ -75,45 +101,100 @@ -- | 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+ let us = processId proc+ let node = processNode proc+ if localNodeId node == nid+ then nsend label 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.+--+-- Unlike 'send', this function is insensitive to 'reconnect'. It will+-- try to send the message regardless of the history of connection failures+-- between the nodes.+--+-- Message passing with 'usend' is ordered for a given sender and receiver+-- if the messages arrive at all.+--+usend :: Serializable a => ProcessId -> a -> Process ()+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 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