distributed-process 0.4.2 → 0.5.0

raw patch · 31 files changed

+3874/−2533 lines, 31 filesdep +deepseqdep +hashabledep −HUnitdep −ansi-terminaldep −networkdep ~binarydep ~containersdep ~distributed-staticPVP ok

version bump matches the API change (PVP)

Dependencies added: deepseq, hashable

Dependencies removed: HUnit, ansi-terminal, network, test-framework, test-framework-hunit

Dependency ranges changed: binary, containers, distributed-static, mtl, network-transport, network-transport-tcp, rank1dynamic, stm, syb, template-haskell, transformers

API changes (from Hackage documentation)

Files

• ChangeLog +120/−0
• benchmarks/ProcessRing.hs +113/−0
• distributed-process.cabal +101/−144
• src/Control/Distributed/Process.hs +64/−132
• src/Control/Distributed/Process/Closure.hs +18/−0
• src/Control/Distributed/Process/Debug.hs +279/−0
• src/Control/Distributed/Process/Internal/CQueue.hs +8/−2
• src/Control/Distributed/Process/Internal/Closure/BuiltIn.hs +23/−0
• src/Control/Distributed/Process/Internal/Closure/Explicit.hs +148/−0
• src/Control/Distributed/Process/Internal/Messaging.hs +31/−2
• src/Control/Distributed/Process/Internal/Primitives.hs +380/−149
• src/Control/Distributed/Process/Internal/Spawn.hs +161/−0
• src/Control/Distributed/Process/Internal/StrictMVar.hs +5/−0
• src/Control/Distributed/Process/Internal/Trace.hs +0/−122
• src/Control/Distributed/Process/Internal/Types.hs +153/−40
• src/Control/Distributed/Process/Internal/WeakTQueue.hs +2/−1
• src/Control/Distributed/Process/Management.hs +557/−0
• src/Control/Distributed/Process/Management/Internal/Agent.hs +156/−0
• src/Control/Distributed/Process/Management/Internal/Bus.hs +21/−0
• src/Control/Distributed/Process/Management/Internal/Table.hs +222/−0
• src/Control/Distributed/Process/Management/Internal/Trace/Primitives.hs +172/−0
• src/Control/Distributed/Process/Management/Internal/Trace/Remote.hs +61/−0
• src/Control/Distributed/Process/Management/Internal/Trace/Tracer.hs +347/−0
• src/Control/Distributed/Process/Management/Internal/Trace/Types.hs +169/−0
• src/Control/Distributed/Process/Management/Internal/Types.hs +151/−0
• src/Control/Distributed/Process/Node.hs +279/−115
• src/Control/Distributed/Process/Serializable.hs +12/−0
• src/Control/Distributed/Process/UnsafePrimitives.hs +121/−0
• tests/TestCH.hs +0/−1184
• tests/TestClosure.hs +0/−480
• tests/TestStats.hs +0/−162

@@ -0,0 +1,120 @@+2014-30-05  Tim Watson  <watson.timothy@gmail.com>  0.5.0++* Dependency on STM implicitly changed from 1.3 to 1.4, but was not reflected in the cabal file+* Race condition in local monitoring when using call+* mask now works correctly if unmask is called by another process+* Improve efficiency of local message passing+* nsend uses local communication channels+* Link Node Controller and Network Listener+* Label spawned processes using labelThread+* Relaxed upper bound on syb in the cabal manifest+* Bump binary version to include 0.7.*+* Exposed process info+* Exposed node statistics+* Moved tests to https://github.com/haskell-distributed/distributed-process-tests+* Added "polymorphic expect"+* Exposed Message type and broaden scope of polymorphic expect+* Added Management API (for working with internal/system events)+* Tracing can no longer be disabled+* We now report node statistics for monitoring/management+* Node.runProcess now propagates exceptions to its caller+* Added simple micro benchmarks++2013-01-27  Tim Watson  <watson.timothy@gmail.com>  0.4.2++* Improved exception handling for deferred type checked exit reasons+* Add matchChan primitive (thanks Simon Marlow)+* Expose deferred message handling/checking for AbstractMessage+* Add `getProcessInfo' API+* Add `trace' API backed by the GHC eventlog++2012-11-22  Edsko de Vries  <edsko@well-typed.com>  0.4.1++* Make behaviour of 'register' more Erlang-like (register will now fail if the+name is already registered).  Patch by Jeff Epstein.+* Functor, Applicative, Alternative and Monad instances for ReceivePort+* Add support for receiveChanTimeout+* Improved documentation+* Avoid name clashes in the TH generation for closures+* Relax package bounds to allow for Binary 0.6++2012-10-23  Edsko de Vries  <edsko@well-typed.com>  0.4.0.2++* Fix race condition in spawn++2012-10-04  Edsko de Vries  <edsko@well-typed.com>  0.4.0.1++* Relax package boundaries++2012-10-03  Edsko de Vries  <edsko@well-typed.com>  0.4.0++* Improved treatment of network failure, using new failure semantics of+Network.Transport.+* Make NodeId Typeable+* Extend Template Haskell support with "remotableDec" so that you can refer to+$(mkClosure 'f) within the body of "f".+* Fix bug in spawnChannelLocal+* Numerous memory leaks plugged+* Relax upper bound on dependency on 'network'+* New primitive 'matchAny'+* Remove 'whereisRemote' (see comment of 'whereisRemoteAsync') ++2012-08-16  Edsko de Vries  <edsko@well-typed.com>  0.3.1++* Fix memory leaks+* Make Template Haskell support optional+* Relax dependency constraints++2012-08-07  Edsko de Vries  <edsko@well-typed.com>  0.3.0++* Extract 'static' into a separate package (C.D.Static)+* Use new package rank1dynamic to proper runtime checks for polymorphic values++2012-08-02  Edsko de Vries  <edsko@well-typed.com>  0.2.3.0++* Expose the constructors of Closure+* Add instance (Typeable a => Serializable (Static a)) and make sure we only+use the internal representation of Static where really necessary+* Improved docs++2012-07-31  Edsko de Vries  <edsko@well-typed.com>  0.2.2.0++* Add exception handling primitives+* Fix runProcess: if the process threw an exception, a 'waiting indefinitely on+MVar' exception would be thrown.++2012-07-21  Edsko de Vries  <edsko@well-typed.com>  0.2.1.4++* Bugfix in the node controller+(one way this bug materialized: when using the SimpleLocalnet backend,+slave nodes could not be reused)+* Improved documentation in Control.Distributed.Process.Closure++2012-07-20  Edsko de Vries  <edsko@well-typed.com>  0.2.1.3++* Improve docs+* Local versions of spawn++2012-07-16  Edsko de Vries  <edsko@well-typed.com>  0.2.1.2++* Base 4.6 compatibility+* Relax constraints on bytestring and containers++2012-07-16  Edsko de Vries  <edsko@well-typed.com>  0.2.1.1++* Relax upper bound on 'time' dependency++2012-07-11  Edsko de Vries  <edsko@well-typed.com>  0.2.1++* Complete redesign of the underlying implementation of static values and+closures. ++* Add support for 'spawnChannel' ++2012-07-09  Edsko de Vries  <edsko@well-typed.com>  0.2.0.1++* Bugfix: Continue processing messages when a connection breaks.++2012-07-07  Edsko de Vries  <edsko@well-typed.com>  0.2.0++* Initial release.

@@ -0,0 +1,113 @@+{- ProcessRing benchmarks.++To run the benchmarks, select a value for the ring size (sz) and+the number of times to send a message around the ring++-}+import Control.Monad+import Control.Distributed.Process hiding (catch)+import Control.Distributed.Process.Node+import Control.Exception (catch, SomeException)+import Network.Transport.TCP (createTransport, defaultTCPParameters)+import System.Environment+import System.Console.GetOpt++data Options = Options+  { optRingSize   :: Int+  , optIterations :: Int+  , optForward    :: Bool+  , optParallel   :: Bool+  , optUnsafe     :: Bool+  } deriving Show++initialProcess :: Options -> Process ()+initialProcess op =+  let ringSz = optRingSize op+      msgCnt = optIterations op+      fwd    = optForward op+      unsafe = optUnsafe op+      msg    = ("foobar", "baz")+  in do+    self <- getSelfPid+    ring <- makeRing fwd unsafe ringSz self+    forM_ [1..msgCnt] (\_ -> send ring msg)+    collect msgCnt+  where relay fsend pid = do+          msg <- expect :: Process (String, String)+          fsend pid msg+          relay fsend pid++        forward' pid =+          receiveWait [ matchAny (\m -> forward m pid) ] >> forward' pid++        makeRing :: Bool -> Bool -> Int -> ProcessId -> Process ProcessId+        makeRing !f !u !n !pid+          | n == 0    = go f u pid+          | otherwise = go f u pid >>= makeRing f u (n - 1)++        go :: Bool -> Bool -> ProcessId -> Process ProcessId+        go False False next = spawnLocal $ relay send next+        go False True  next = spawnLocal $ relay unsafeSend next+        go True  _     next = spawnLocal $ forward' next++        collect :: Int -> Process ()+        collect !n+          | n == 0    = return ()+          | otherwise = do+                receiveWait [+                    matchIf    (\(a, b) -> a == "foobar" && b == "baz")+                               (\_ -> return ())+                  , matchAny   (\_ -> error "unexpected input!")+                  ]+                collect (n - 1)++defaultOptions :: Options+defaultOptions = Options+  { optRingSize   = 10+  , optIterations = 100+  , optForward    = False+  , optParallel   = False+  , optUnsafe     = False+  }++options :: [OptDescr (Options -> Options)]+options =+    [ Option ['s'] ["ring-size"] (OptArg optSz "SIZE") "# of processes in ring"+    , Option ['i'] ["iterations"] (OptArg optMsgCnt "ITER") "# of times to send"+    , Option ['f'] ["forward"]+        (NoArg (\opts -> opts { optForward = True }))+        "use `forward' instead of send - default = False"+    , Option ['u'] ["unsafe-send"]+        (NoArg (\opts -> opts { optUnsafe = True }))+        "use 'unsafeSend' (ignored with -f) - default = False"+    , Option ['p'] ["parallel"]+        (NoArg (\opts -> opts { optParallel = True }))+        "send in parallel and consume sequentially - default = False"+    ]++optMsgCnt :: Maybe String -> Options -> Options+optMsgCnt Nothing  opts = opts+optMsgCnt (Just c) opts = opts { optIterations = ((read c) :: Int) }++optSz :: Maybe String -> Options -> Options+optSz Nothing  opts = opts+optSz (Just s) opts = opts { optRingSize = ((read s) :: Int) }++parseArgv :: [String] -> IO (Options, [String])+parseArgv argv = do+  pn <- getProgName+  case getOpt Permute options argv of+    (o,n,[]  ) -> return (foldl (flip id) defaultOptions o, n)+    (_,_,errs) -> ioError (userError (concat errs ++ usageInfo (header pn) options))+  where header pn' = "Usage: " ++ pn' ++ " [OPTION...]"++main :: IO ()+main = do+  argv <- getArgs+  (opt, _) <- parseArgv argv+  putStrLn $ "options: " ++ (show opt)+  Right transport <- createTransport "127.0.0.1" "8090" defaultTCPParameters+  node <- newLocalNode transport initRemoteTable+  catch (void $ runProcess node $ initialProcess opt)+        (\(e :: SomeException) -> putStrLn $ "ERROR: " ++ (show e))+

@@ -1,5 +1,5 @@ Name:          distributed-process-Version:       0.4.2+Version:       0.5.0 Cabal-Version: >=1.8 Build-Type:    Simple License:       BSD3@@ -8,8 +8,8 @@ Author:        Duncan Coutts, Nicolas Wu, Edsko de Vries Maintainer:    watson.timothy@gmail.com, edsko@well-typed.com, duncan@well-typed.com Stability:     experimental-Homepage:      http://github.com/haskell-distributed/distributed-process-Bug-Reports:   http://github.com/haskell-distributed/distributed-process/issues+Homepage:      http://haskell-distributed.github.com/+Bug-Reports:   http://cloud-haskell.atlassian.net 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,@@ -21,8 +21,9 @@                 You will probably also want to install a Cloud Haskell backend such                as distributed-process-simplelocalnet.-Tested-With:   GHC==7.2.2 GHC==7.4.1 GHC==7.4.2+Tested-With:   GHC==7.2.2 GHC==7.4.1 GHC==7.4.2 GHC==7.6.2 Category:      Control+extra-source-files: ChangeLog  Source-Repository head   Type:     git@@ -33,171 +34,127 @@   description: Build with Template Haskell support   default: True -flag benchmarks-  description: Build benchmarks+flag prof+  description: Compiling with profiling enabled   default: False  Library   Build-Depends:     base >= 4.4 && < 5,-                     binary >= 0.5 && < 0.7,-                     network-transport >= 0.3 && < 0.4,-                     stm >= 2.3 && < 2.5,-                     transformers >= 0.2 && < 0.4,-                     mtl >= 2.0 && < 2.2,+                     binary >= 0.6.3 && < 0.8,+                     hashable >= 1.2.0.5 && < 1.3,+                     network-transport >= 0.4.0.0 && < 0.5,+                     stm >= 2.4 && < 2.5,+                     transformers >= 0.2 && < 0.5,+                     mtl >= 2.0 && < 2.3,                      data-accessor >= 0.2 && < 0.3,                      bytestring >= 0.9 && < 0.11,-                     containers >= 0.4 && < 0.6,                      old-locale >= 1.0 && < 1.1,                      time >= 1.2 && < 1.5,                      random >= 1.0 && < 1.1,                      ghc-prim >= 0.2 && < 0.4,-                     distributed-static >= 0.2 && < 0.3,-                     rank1dynamic >= 0.1 && < 0.2,-                     syb >= 0.3 && < 0.4+                     distributed-static >= 0.2 && < 0.4,+                     rank1dynamic >= 0.1 && < 0.3,+                     syb >= 0.3 && < 0.5   Exposed-modules:   Control.Distributed.Process,-                     Control.Distributed.Process.Serializable,                      Control.Distributed.Process.Closure,-                     Control.Distributed.Process.Node,-                     Control.Distributed.Process.Internal.Primitives,-                     Control.Distributed.Process.Internal.CQueue,-                     Control.Distributed.Process.Internal.Types,-                     Control.Distributed.Process.Internal.Trace,+                     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.WeakTQueue-                     Control.Distributed.Process.Internal.StrictContainerAccessors-  Extensions:        RankNTypes,-                     ScopedTypeVariables,-                     FlexibleInstances,-                     UndecidableInstances,-                     ExistentialQuantification,-                     GADTs,-                     GeneralizedNewtypeDeriving,-                     DeriveDataTypeable,-                     CPP,-                     BangPatterns+                     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.Trace.Tracer+  -- Extensions:     -- we use explicit per-file LANGUAGE pragmas   ghc-options:       -Wall   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.4   if flag(th)-     Build-Depends:   template-haskell >= 2.6 && < 2.9+     if impl(ghc <= 7.4.2)+       Build-Depends: template-haskell >= 2.7 && < 2.8+     else+       Build-Depends: template-haskell >= 2.6 && < 2.10      Exposed-modules: Control.Distributed.Process.Internal.Closure.TH      CPP-Options:     -DTemplateHaskellSupport -Test-Suite TestCH-  Type:              exitcode-stdio-1.0-  Main-Is:           TestCH.hs-  Build-Depends:     base >= 4.4 && < 5,-                     random >= 1.0 && < 1.1,-                     ansi-terminal >= 0.5 && < 0.6,-                     distributed-process,-                     network-transport >= 0.3 && < 0.4,-                     network-transport-tcp >= 0.3 && < 0.4,-                     binary >= 0.5 && < 0.7,-                     network >= 2.3 && < 2.5,-                     HUnit >= 1.2 && < 1.3,-                     test-framework >= 0.6 && < 0.9,-                     test-framework-hunit >= 0.2.0 && < 0.4-  Extensions:        CPP,-                     ScopedTypeVariables,-                     DeriveDataTypeable,-                     GeneralizedNewtypeDeriving-  ghc-options:       -Wall -threaded -rtsopts -with-rtsopts=-N -fno-warn-unused-do-bind-  HS-Source-Dirs:    tests--Test-Suite TestClosure-  Type:              exitcode-stdio-1.0-  Main-Is:           TestClosure.hs-  Build-Depends:     base >= 4.4 && < 5,-                     random >= 1.0 && < 1.1,-                     ansi-terminal >= 0.5 && < 0.6,-                     distributed-static >= 0.2 && < 0.3,-                     distributed-process,-                     network-transport >= 0.3 && < 0.4,-                     network-transport-tcp >= 0.3 && < 0.4,-                     bytestring >= 0.9 && < 0.11,-                     network >= 2.3 && < 2.5,-                     HUnit >= 1.2 && < 1.3,-                     test-framework >= 0.6 && < 0.9,-                     test-framework-hunit >= 0.2.0 && < 0.4-  Extensions:        CPP,-                     ScopedTypeVariables-  ghc-options:       -Wall -threaded -rtsopts -with-rtsopts=-N -fno-warn-unused-do-bind-  HS-Source-Dirs:    tests--Test-Suite TestStats-  Type:              exitcode-stdio-1.0-  Main-Is:           TestStats.hs-  Build-Depends:     base >= 4.4 && < 5,-                     random >= 1.0 && < 1.1,-                     ansi-terminal >= 0.5 && < 0.6,-                     containers >= 0.4 && < 0.6,-                     stm >= 2.3 && < 2.5,-                     distributed-process,-                     network-transport >= 0.3 && < 0.4,-                     network-transport-tcp >= 0.3 && < 0.4,-                     binary >= 0.5 && < 0.7,-                     network >= 2.3 && < 2.5,-                     HUnit >= 1.2 && < 1.3,-                     test-framework >= 0.6 && < 0.9,-                     test-framework-hunit >= 0.2.0 && < 0.4-  Extensions:        CPP,-                     ScopedTypeVariables,-                     DeriveDataTypeable,-                     GeneralizedNewtypeDeriving-  ghc-options:       -Wall -debug -eventlog -threaded -rtsopts -with-rtsopts=-N -fno-warn-unused-do-bind-  HS-Source-Dirs:    tests +-- 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+  Extensions:      BangPatterns -Executable distributed-process-throughput -  if flag(benchmarks)-    Build-Depends:   base >= 4.4 && < 5,-                     distributed-process,-                     network-transport-tcp >= 0.3 && < 0.4,-                     bytestring >= 0.9 && < 0.11,-                     binary >= 0.5 && < 0.7-  else-    buildable: False-  Main-Is:           benchmarks/Throughput.hs-  ghc-options:       -Wall-  Extensions:        BangPatterns+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+  Extensions:      BangPatterns -Executable distributed-process-latency-  if flag(benchmarks)-    Build-Depends:   base >= 4.4 && < 5,-                     distributed-process,-                     network-transport-tcp >= 0.3 && < 0.4,-                     bytestring >= 0.9 && < 0.11,-                     binary >= 0.5 && < 0.7-  else-    buildable: False-  Main-Is:           benchmarks/Latency.hs-  ghc-options:       -Wall-  Extensions:        BangPatterns+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+  Extensions:      BangPatterns -Executable distributed-process-channels-  if flag(benchmarks)-    Build-Depends:   base >= 4.4 && < 5,-                     distributed-process,-                     network-transport-tcp >= 0.3 && < 0.4,-                     bytestring >= 0.9 && < 0.11,-                     binary >= 0.5 && < 0.7-  else-    buildable: False-  Main-Is:           benchmarks/Channels.hs-  ghc-options:       -Wall-  Extensions:        BangPatterns+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+  Extensions:      BangPatterns -Executable distributed-process-spawns-  if flag(benchmarks)-    Build-Depends:   base >= 4.4 && < 5,-                     distributed-process,-                     network-transport-tcp >= 0.3 && < 0.4,-                     bytestring >= 0.9 && < 0.11,-                     binary >= 0.5 && < 0.7+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+  if flag(prof)+    ghc-options:   -Wall -threaded -prof -auto-all -fno-prof-count-entries -rtsopts   else-    buildable: False-  Main-Is:           benchmarks/Spawns.hs-  ghc-options:       -Wall-  Extensions:        BangPatterns+    ghc-options:   -Wall -threaded -O2 -rtsopts+  Extensions:      BangPatterns,+                   ScopedTypeVariables

@@ -1,8 +1,10 @@+{-# LANGUAGE CPP  #-} {- | [Cloud Haskell]  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/>),+Peyton Jones (see+<http://research.microsoft.com/en-us/um/people/simonpj/papers/parallel/>), 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.@@ -15,7 +17,7 @@ module Control.Distributed.Process   ( -- * Basic types     ProcessId-  , NodeId+  , NodeId(..)   , Process   , SendPortId   , processNodeId@@ -35,6 +37,11 @@   , receiveChanTimeout   , mergePortsBiased   , mergePortsRR+    -- * Unsafe messaging variants+  , unsafeSend+  , unsafeSendChan+  , unsafeNSend+  , unsafeWrapMessage     -- * Advanced messaging   , Match   , receiveWait@@ -42,10 +49,24 @@   , match   , matchIf   , matchUnknown-  , AbstractMessage(..)   , matchAny   , matchAnyIf   , matchChan+  , matchSTM+  , Message+  , matchMessage+  , matchMessageIf+  , isEncoded+  , wrapMessage+  , unwrapMessage+  , handleMessage+  , handleMessageIf+  , handleMessage_+  , handleMessageIf_+  , forward+  , delegate+  , relay+  , proxy     -- * Process management   , spawn   , call@@ -62,6 +83,9 @@   , getSelfNode   , ProcessInfo(..)   , getProcessInfo+  , NodeStats(..)+  , getNodeStats+  , getLocalNodeStats     -- * Monitoring and linking   , link   , linkNode@@ -133,7 +157,6 @@ import Prelude hiding (catch) #endif -import Data.Typeable (Typeable) import Control.Monad.IO.Class (liftIO) import Control.Applicative ((<$>)) import Control.Monad.Reader (ask)@@ -143,8 +166,6 @@   , closure   , Static   , RemoteTable-  , closureCompose-  , staticClosure   ) import Control.Distributed.Process.Internal.Types   ( NodeId(..)@@ -167,21 +188,9 @@   , WhereIsReply(..)   , RegisterReply(..)   , LocalProcess(processNode)-  , nullProcessId-  )-import Control.Distributed.Process.Serializable (Serializable, SerializableDict)-import Control.Distributed.Process.Internal.Closure.BuiltIn-  ( sdictSendPort-  , sndStatic-  , idCP-  , seqCP-  , bindCP-  , splitCP-  , cpLink-  , cpSend-  , cpNewChan-  , cpDelay+  , Message   )+import Control.Distributed.Process.Serializable (Serializable) import Control.Distributed.Process.Internal.Primitives   ( -- Basic messaging     send@@ -192,6 +201,9 @@   , receiveChan   , mergePortsBiased   , mergePortsRR+  , unsafeSend+  , unsafeSendChan+  , unsafeNSend     -- Advanced messaging   , Match   , receiveWait@@ -199,10 +211,24 @@   , match   , matchIf   , matchUnknown-  , AbstractMessage(..)   , matchAny   , matchAnyIf   , matchChan+  , matchSTM+  , matchMessage+  , matchMessageIf+  , isEncoded+  , wrapMessage+  , unsafeWrapMessage+  , unwrapMessage+  , handleMessage+  , handleMessageIf+  , handleMessage_+  , handleMessageIf_+  , forward+  , delegate+  , relay+  , proxy     -- Process management   , terminate   , ProcessTerminationException(..)@@ -215,6 +241,9 @@   , getSelfNode   , ProcessInfo(..)   , getProcessInfo+  , NodeStats(..)+  , getNodeStats+  , getLocalNodeStats     -- Monitoring and linking   , link   , linkNode@@ -262,6 +291,15 @@   , reconnectPort   ) import Control.Distributed.Process.Node (forkProcess)+import Control.Distributed.Process.Internal.Spawn+  ( -- Spawning Processes/Channels+    spawn+  , spawnLink+  , spawnMonitor+  , spawnChannel+  , spawnSupervised+  , call+  )  -- INTERNAL NOTES --@@ -273,7 +311,11 @@ -- 2.  'send' may block (when the system TCP buffers are full, while we are --     trying to establish a connection to the remote endpoint, etc.) but its --     return does not imply that the remote process received the message (much---     less processed it)+--     less processed it). When 'send' targets a local process, it is dispatched+--     via the node controller however, in which cases it will /not/ block. This+--     isn't part of the semantics, so it should be ok. The ordering is maintained+--     because the ctrlChannel still has FIFO semantics with regards interactions+--     between two disparate forkIO threads. -- -- 3.  Message delivery is reliable and ordered. That means that if process A --     sends messages m1, m2, m3 to process B, B will either arrive all three@@ -321,116 +363,6 @@ --       http://www.erlang.org/faq/academic.html -- [6] "Delivery of Messages", post on erlang-questions --       http://erlang.org/pipermail/erlang-questions/2012-February/064767.html------------------------------------------------------------------------------------- Primitives are defined in a separate module; here we only define derived   ----- constructs                                                                 --------------------------------------------------------------------------------------- | Spawn a process------ For more information about 'Closure', see--- "Control.Distributed.Process.Closure".------ See also 'call'.-spawn :: NodeId -> Closure (Process ()) -> Process ProcessId-spawn nid proc = do-  us   <- getSelfPid-  mRef <- monitorNode nid-  sRef <- spawnAsync nid (cpDelay us proc)-  receiveWait [-      matchIf (\(DidSpawn ref _) -> ref == sRef) $ \(DidSpawn _ pid) -> do-        unmonitor mRef-        send pid ()-        return pid-    , matchIf (\(NodeMonitorNotification ref _ _) -> ref == mRef) $ \_ ->-        return (nullProcessId nid)-    ]---- | Spawn a process and link to it------ Note that this is just the sequential composition of 'spawn' and 'link'.--- (The "Unified" semantics that underlies Cloud Haskell does not even support--- a synchronous link operation)-spawnLink :: NodeId -> Closure (Process ()) -> Process ProcessId-spawnLink nid proc = do-  pid <- spawn nid proc-  link pid-  return pid---- | Like 'spawnLink', but monitor the spawned process-spawnMonitor :: NodeId -> Closure (Process ()) -> Process (ProcessId, MonitorRef)-spawnMonitor nid proc = do-  pid <- spawn nid proc-  ref <- monitor pid-  return (pid, ref)---- | Run a process remotely and wait for it to reply------ We monitor the remote process: if it dies before it can send a reply, we die--- too.------ For more information about 'Static', 'SerializableDict', and 'Closure', see--- "Control.Distributed.Process.Closure".------ See also 'spawn'.-call :: Serializable a-        => Static (SerializableDict a)-        -> NodeId-        -> Closure (Process a)-        -> Process a-call dict nid proc = do-  us <- getSelfPid-  (pid, mRef) <- spawnMonitor nid (proc `bindCP` cpSend dict us)-  -- We are guaranteed to receive the reply before the monitor notification-  -- (if a reply is sent at all)-  -- NOTE: This might not be true if we switch to unreliable delivery.-  mResult <- receiveWait-    [ match (return . Right)-    , matchIf (\(ProcessMonitorNotification ref _ _) -> ref == mRef)-              (\(ProcessMonitorNotification _ _ reason) -> return (Left reason))-    ]-  case mResult of-    Right a  -> do-      -- Wait for the monitor message so that we the mailbox doesn't grow-      receiveWait-        [ matchIf (\(ProcessMonitorNotification ref _ _) -> ref == mRef)-                  (\(ProcessMonitorNotification {}) -> return ())-        ]-      -- Clean up connection to pid-      reconnect pid-      return a-    Left err ->-      fail $ "call: remote process died: " ++ show err---- | Spawn a child process, have the child link to the parent and the parent--- monitor the child-spawnSupervised :: NodeId-                -> Closure (Process ())-                -> Process (ProcessId, MonitorRef)-spawnSupervised nid proc = do-  us   <- getSelfPid-  them <- spawn nid (cpLink us `seqCP` proc)-  ref  <- monitor them-  return (them, ref)---- | Spawn a new process, supplying it with a new 'ReceivePort' and return--- the corresponding 'SendPort'.-spawnChannel :: forall a. Typeable a => Static (SerializableDict a)-             -> NodeId-             -> Closure (ReceivePort a -> Process ())-             -> Process (SendPort a)-spawnChannel dict nid proc = do-    us <- getSelfPid-    _ <- spawn nid (go us)-    expect-  where-    go :: ProcessId -> Closure (Process ())-    go pid = cpNewChan dict-           `bindCP`-             (cpSend (sdictSendPort dict) pid `splitCP` proc)-           `bindCP`-             (idCP `closureCompose` staticClosure sndStatic)  -------------------------------------------------------------------------------- -- Local versions of spawn                                                    --

@@ -173,9 +173,17 @@     -- * CP versions of Cloud Haskell primitives   , cpLink   , cpUnlink+  , cpRelay   , cpSend   , cpExpect   , cpNewChan+    -- * Working with static values and closures (without Template Haskell)+  , RemoteRegister+  , MkTDict(..)+  , mkStaticVal+  , mkClosureValSingle+  , mkClosureVal+  , call' #ifdef TemplateHaskellSupport     -- * Template Haskell support for creating static values and closures   , remotable@@ -205,9 +213,19 @@     -- CP versions of Cloud Haskell primitives   , cpLink   , cpUnlink+  , cpRelay   , cpSend   , cpExpect   , cpNewChan+  )+import Control.Distributed.Process.Internal.Closure.Explicit+  (+    RemoteRegister+  , MkTDict(..)+  , mkStaticVal+  , mkClosureValSingle+  , mkClosureVal+  , call'   ) #ifdef TemplateHaskellSupport import Control.Distributed.Process.Internal.Closure.TH

@@ -0,0 +1,279 @@+{-# LANGUAGE CPP  #-}+{-# LANGUAGE ScopedTypeVariables  #-}+{-# LANGUAGE RankNTypes  #-}+-----------------------------------------------------------------------------+-- |+-- Module      :  Control.Distributed.Process.Debug+-- Copyright   :  (c) Well-Typed / Tim Watson+-- License     :  BSD3 (see the file LICENSE)+--+-- Maintainer  :  Tim Watson <watson.timothy@gmail.com>+-- Stability   :  experimental+-- Portability :  non-portable (requires concurrency)+--+-- [Tracing/Debugging Facilities]+--+-- Cloud Haskell provides a general purpose tracing mechanism, allowing a+-- user supplied /tracer process/ to receive messages when certain classes of+-- system events occur. It's possible to use this facility to aid in debugging+-- and/or perform other diagnostic tasks to a program at runtime.+--+-- [Enabling Tracing]+--+-- Throughout the lifecycle of a local node, the distributed-process runtime+-- generates /trace events/, describing internal runtime activities such as+-- the spawning and death of processes, message sending, delivery and so on.+-- See the 'MxEvent' type's documentation for a list of all the published+-- event types, which correspond directly to the types of /management/ events.+-- Users can additionally publish custom trace events in the form of+-- 'MxLog' log messages or pass custom (i.e., completely user defined)+-- event data using the 'traceMessage' function.+--+-- All published traces are forwarded to a /tracer process/, which can be+-- specified (and changed) at runtime using 'traceEnable'. Some pre-defined+-- tracer processes are provided for conveniently printing to stderr, a log file+-- or the GHC eventlog.+--+-- If a tracer process crashes, no attempt is made to restart it.+--+-- [Working with multiple tracer processes]+--+-- The tracing facility only ever writes to a single tracer process. This+-- invariant insulates the tracer controller and ensures a fast path for+-- handling all trace events. /This/ module provides facilities for layering+-- trace handlers using Cloud Haskell's built-in delegation primitives.+--+-- The 'startTracer' function wraps the registered @tracer@ process with the+-- supplied handler and also forwards trace events to the original tracer.+-- The corresponding 'stopTracer' function terminates tracer processes in+-- reverse of the order in which they were started, and re-registers the+-- previous tracer process.+--+-- [Built in tracers]+--+-- The built in tracers provide a simple /logging/ facility that writes trace+-- events out to either a log file, @stderr@ or the GHC eventlog. These tracers+-- can be configured using environment variables, or specified manually using+-- the 'traceEnable' function.+--+-- When a new local node is started, the contents of several environment+-- variables are checked to determine which default tracer process is selected.+-- If none of these variables is set, a no-op tracer process is installed,+-- which effectively ignores all trace messages. Note that in this case,+-- trace events are still generated and passed through the system.+-- Only one default tracer will be chosen - the first that contains a (valid)+-- value. These environment variables, in the order they're examined, are:+--+-- 1. @DISTRIBUTED_PROCESS_TRACE_FILE@+-- This is checked for a valid file path. If it exists and the file can be+-- opened for writing, all trace output will be directed thence. If the supplied+-- path is invalid, or the file is unavailable for writing, this tracer will not+-- be selected.+--+-- 2. @DISTRIBUTED_PROCESS_TRACE_CONSOLE@+-- This is checked for /any/ non-empty value. If set, then all trace output will+-- be directed to the system logger process.+--+-- 3. @DISTRIBUTED_PROCESS_TRACE_EVENTLOG@+-- This is checked for /any/ non-empty value. If set, all internal traces are+-- written to the GHC eventlog.+--+-- Users of the /simplelocalnet/ Cloud Haskell backend should also note that+-- because the trace file option only supports trace output from a single node+-- (so as to avoid interleaving), a file trace configured for the master node+-- will prevent slaves from tracing to the file. They will need to fall back to+-- the console or eventlog tracers instead, which can be accomplished by setting+-- one of these environment variables /as well/, since the latter will only be+-- selected on slaves (when the file tracer selection fails).+--+-- Support for writing to the eventlog requires specific intervention to work,+-- without which, written traces are silently dropped/ignored and no output will+-- be generated. The GHC eventlog documentation provides information about+-- enabling, viewing and working with event traces at+-- <http://hackage.haskell.org/trac/ghc/wiki/EventLog>.+--+module Control.Distributed.Process.Debug+  ( -- * Exported Data Types+    TraceArg(..)+  , TraceFlags(..)+  , TraceSubject(..)+    -- * Configuring Tracing+  , enableTrace+  , enableTraceAsync+  , disableTrace+  , withTracer+  , withFlags+  , getTraceFlags+  , setTraceFlags+  , setTraceFlagsAsync+  , defaultTraceFlags+  , traceOn+  , traceOnly+  , traceOff+    -- * Debugging+  , startTracer+  , stopTracer+    -- * Sending Custom Trace Data+  , traceLog+  , traceLogFmt+  , traceMessage+    -- * Working with remote nodes+  , Remote.remoteTable+  , Remote.startTraceRelay+  , Remote.setTraceFlagsRemote+    -- * Built in tracers+  , systemLoggerTracer+  , logfileTracer+  , eventLogTracer+  )+  where++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+  ( ProcessId+  , Process+  , LocalProcess(..)+  , ProcessMonitorNotification(..)+  )+import Control.Distributed.Process.Management.Internal.Types+  ( MxEvent(..)+  )+import Control.Distributed.Process.Management.Internal.Trace.Types+  ( TraceArg(..)+  , TraceFlags(..)+  , TraceSubject(..)+  , defaultTraceFlags+  )+import Control.Distributed.Process.Management.Internal.Trace.Tracer+  ( systemLoggerTracer+  , logfileTracer+  , eventLogTracer+  )+import Control.Distributed.Process.Management.Internal.Trace.Primitives+  ( withRegisteredTracer+  , enableTrace+  , enableTraceAsync+  , disableTrace+  , setTraceFlags+  , setTraceFlagsAsync+  , getTraceFlags+  , traceOn+  , traceOff+  , traceOnly+  , traceLog+  , traceLogFmt+  , traceMessage+  )+import qualified Control.Distributed.Process.Management.Internal.Trace.Remote as Remote+import Control.Distributed.Process.Node++import Control.Exception (SomeException)++import Control.Monad.IO.Class (liftIO)+import Control.Monad.Reader (ask)++import Data.Binary()++#if ! MIN_VERSION_base(4,6,0)+import Prelude hiding (catch)+#endif++--------------------------------------------------------------------------------+-- Debugging/Tracing API                                                      --+--------------------------------------------------------------------------------++-- | Starts a new tracer, using the supplied trace function.+-- Only one tracer can be registered at a time, however /this/ function overlays+-- the registered tracer with the supplied handler, allowing the user to layer+-- multiple tracers on top of one another, with trace events forwarded down+-- through all the layers in turn. Once the top layer is stopped, the user+-- is responsible for re-registering the original (prior) tracer pid before+-- terminating. See 'withTracer' for a mechanism that handles that.+startTracer :: (MxEvent -> Process ()) -> Process ProcessId+startTracer handler = do+  withRegisteredTracer $ \pid -> do+    node <- processNode <$> ask+    newPid <- liftIO $ forkProcess node $ traceProxy pid handler+    enableTrace newPid  -- invokes sync + registration+    return newPid++-- | Evaluate @proc@ with tracing enabled via @handler@, and immediately+-- disable tracing thereafter, before giving the result (or exception+-- in case of failure).+withTracer :: forall a.+              (MxEvent -> Process ())+           -> Process a+           -> Process (Either SomeException a)+withTracer handler proc = do+    previous <- whereis "tracer"+    tracer <- startTracer handler+    finally (try proc)+            (stopTracing tracer previous)+  where+    stopTracing :: ProcessId -> Maybe ProcessId -> Process ()+    stopTracing tracer previousTracer = do+      case previousTracer of+        Nothing -> return ()+        Just _  -> do+          ref <- monitor tracer+          send tracer MxTraceDisable+          receiveWait [+              matchIf (\(ProcessMonitorNotification ref' _ _) -> ref == ref')+                      (\_ -> return ())+            ]++-- | Evaluate @proc@ with the supplied flags enabled. Any previously set+-- trace flags are restored immediately afterwards.+withFlags :: forall a.+             TraceFlags+          -> Process a+          -> Process (Either SomeException a)+withFlags flags proc = do+  oldFlags <- getTraceFlags+  finally (setTraceFlags flags >> try proc)+          (setTraceFlags oldFlags)++traceProxy :: ProcessId -> (MxEvent -> Process ()) -> Process ()+traceProxy pid act = do+  proxy pid $ \(ev :: MxEvent) ->+    case ev of+      (MxTraceTakeover _) -> return False+      MxTraceDisable      -> die "disabled"+      _                   -> act ev >> return True++-- | Stops a user supplied tracer started with 'startTracer'.+-- Note that only one tracer process can be active at any given time.+-- This process will stop the last process started with 'startTracer'.+-- If 'startTracer' is called multiple times, successive calls to this+-- function will stop the tracers in the reverse order which they were+-- started.+--+-- This function will never stop the system tracer (i.e., the tracer+-- initially started when the node is created), therefore once all user+-- supplied tracers (i.e., processes started via 'startTracer') have exited,+-- subsequent calls to this function will have no effect.+--+-- If the last tracer to have been registered was not started+-- with 'startTracer' then the behaviour of this function is /undefined/.+stopTracer :: Process ()+stopTracer =+  withRegisteredTracer $ \pid -> do+    -- we need to avoid killing the initial (base) tracer, as+    -- nothing we rely on having exactly 1 registered tracer+    -- process at all times.+    basePid <- whereis "tracer.initial"+    case basePid == (Just pid) of+      True  -> return ()+      False -> send pid MxTraceDisable

@@ -1,11 +1,13 @@+{-# LANGUAGE BangPatterns  #-}+{-# LANGUAGE MagicHash, UnboxedTuples, PatternGuards, ScopedTypeVariables, RankNTypes #-} -- | Concurrent queue for single reader, single writer-{-# LANGUAGE MagicHash, UnboxedTuples, PatternGuards #-} module Control.Distributed.Process.Internal.CQueue   ( CQueue   , BlockSpec(..)   , MatchOn(..)   , newCQueue   , enqueue+  , enqueueSTM   , dequeue   , mkWeakCQueue   ) where@@ -52,7 +54,11 @@ -- -- Enqueue is strict. enqueue :: CQueue a -> a -> IO ()-enqueue (CQueue _arrived incoming) !a = atomically $ writeTChan incoming a +enqueue (CQueue _arrived incoming) !a = atomically $ writeTChan incoming a++-- | Variant of enqueue for use in the STM monad.+enqueueSTM :: CQueue a -> a -> STM ()+enqueueSTM (CQueue _arrived incoming) !a = writeTChan incoming a  data BlockSpec =     NonBlocking

@@ -1,3 +1,5 @@+{-# LANGUAGE ScopedTypeVariables  #-}+{-# LANGUAGE RankNTypes  #-} module Control.Distributed.Process.Internal.Closure.BuiltIn   ( -- * Remote table     remoteTable@@ -16,13 +18,16 @@   , bindCP   , seqCP     -- * CP versions of Cloud Haskell primitives+  , decodeProcessIdStatic   , cpLink   , cpUnlink+  , cpRelay   , cpSend   , cpExpect   , cpNewChan     -- * Support for some CH operations   , cpDelay+  , cpEnableTraceRemote   ) where  import Data.ByteString.Lazy (ByteString)@@ -56,6 +61,7 @@ import Control.Distributed.Process.Internal.Primitives   ( link   , unlink+  , relay   , send   , expect   , newChan@@ -82,6 +88,7 @@     . registerStatic "$decodeProcessId" (toDynamic (decode           :: ByteString -> ProcessId))     . registerStatic "$link"            (toDynamic link)     . registerStatic "$unlink"          (toDynamic unlink)+    . registerStatic "$relay"           (toDynamic relay)     . registerStatic "$sendDict"        (toDynamic (sendDict         :: SerializableDict ANY -> ProcessId -> ANY -> Process ()))     . registerStatic "$expectDict"      (toDynamic (expectDict       :: SerializableDict ANY -> Process ANY))     . registerStatic "$newChanDict"     (toDynamic (newChanDict      :: SerializableDict ANY -> Process (SendPort ANY, ReceivePort ANY)))@@ -245,6 +252,22 @@     newChanDictStatic :: Typeable a                       => Static (SerializableDict a -> Process (SendPort a, ReceivePort a))     newChanDictStatic = staticLabel "$newChanDict"++-- | 'CP' version of 'relay'+cpRelay :: ProcessId -> Closure (Process ())+cpRelay = closure (relayStatic `staticCompose` decodeProcessIdStatic) . encode+  where+    relayStatic :: Static (ProcessId -> Process ())+    relayStatic = staticLabel "$relay"++-- TODO: move cpEnableTraceRemote into Trace/Primitives.hs++cpEnableTraceRemote :: ProcessId -> Closure (Process ())+cpEnableTraceRemote =+    closure (enableTraceStatic `staticCompose` decodeProcessIdStatic) . encode+  where+    enableTraceStatic :: Static (ProcessId -> Process ())+    enableTraceStatic = staticLabel "$enableTraceRemote"  -------------------------------------------------------------------------------- -- Support for spawn                                                          --

@@ -0,0 +1,148 @@+{-# LANGUAGE ScopedTypeVariables+  , MultiParamTypeClasses+  , FlexibleInstances+  , FunctionalDependencies+  , FlexibleContexts+  , UndecidableInstances+  , KindSignatures+  , GADTs+  , OverlappingInstances+  , EmptyDataDecls+  , DeriveDataTypeable #-}+module Control.Distributed.Process.Internal.Closure.Explicit+  (+    RemoteRegister+  , MkTDict(..)+  , mkStaticVal+  , mkClosureValSingle+  , mkClosureVal+  , call'+  ) where++import Control.Distributed.Static+import Control.Distributed.Process.Serializable+import Control.Distributed.Process.Internal.Closure.BuiltIn+  ( -- Static dictionaries and associated operations+    staticDecode+  )+import Control.Distributed.Process+import Data.Rank1Dynamic+import Data.Rank1Typeable+import Data.Binary(encode,put,get,Binary)+import qualified Data.ByteString.Lazy as B++-- | A RemoteRegister is a trasformer on a RemoteTable to register additional static values.+type RemoteRegister = RemoteTable -> RemoteTable++-- | This takes an explicit name and a value, and produces both a static reference to the name and a RemoteRegister for it.+mkStaticVal :: Serializable a => String -> a -> (Static a, RemoteRegister)+mkStaticVal n v = (staticLabel n_s, registerStatic n_s (toDynamic v))+    where n_s = n++class MkTDict a where+    mkTDict :: String -> a -> RemoteRegister++instance (Serializable b) => MkTDict (Process b) where+    mkTDict _ _ = registerStatic (show (typeOf (undefined :: b)) ++ "__staticDict") (toDynamic (SerializableDict :: SerializableDict b))++instance MkTDict a where+    mkTDict _ _ = id++-- | This takes an explicit name, a function of arity one, and creates a creates a function yielding a closure and a remote register for it.+mkClosureValSingle :: forall a b. (Serializable a, Typeable b, MkTDict b) => String -> (a -> b) -> (a -> Closure b, RemoteRegister)+mkClosureValSingle n v = (c, registerStatic n_s (toDynamic v) .+                             registerStatic n_sdict (toDynamic sdict) .+                             mkTDict n_tdict (undefined :: b)+                         ) where+    n_s = n+    n_sdict = n ++ "__sdict"+    n_tdict = n ++ "__tdict"++    c = closure decoder . encode++    decoder = (staticLabel n_s :: Static (a -> b)) `staticCompose` staticDecode (staticLabel n_sdict :: Static (SerializableDict a))++    sdict :: (SerializableDict a)+    sdict = SerializableDict++-- | This takes an explict name, a function of any arity, and creates a function yielding a closure and a remote register for it.+mkClosureVal :: forall func argTuple result closureFunction.+                (Curry (argTuple -> Closure result) closureFunction,+                 MkTDict result,+                 Uncurry HTrue argTuple func result,+                 Typeable result, Serializable argTuple, IsFunction func HTrue) =>+                String -> func -> (closureFunction, RemoteRegister)+mkClosureVal n v = (curryFun c, rtable)+    where+      uv :: argTuple -> result+      uv = uncurry' reify v++      n_s = n+      n_sdict = n ++ "__sdict"+      n_tdict = n ++ "__tdict"++      c :: argTuple -> Closure result+      c = closure decoder . encode++      decoder :: Static (B.ByteString -> result)+      decoder = (staticLabel n_s :: Static (argTuple -> result)) `staticCompose` staticDecode (staticLabel n_sdict :: Static (SerializableDict argTuple))++      rtable = registerStatic n_s (toDynamic uv) .+               registerStatic n_sdict (toDynamic sdict) .+               mkTDict n_tdict (undefined :: result)+++      sdict :: (SerializableDict argTuple)+      sdict = SerializableDict++-- | Works just like standard call, but with a simpler signature.+call' :: forall a. Serializable a => NodeId -> Closure (Process a) -> Process a+call' = call (staticLabel $ (show $ typeOf $ (undefined :: a)) ++ "__staticDict")+++data EndOfTuple deriving Typeable+instance Binary EndOfTuple where+    put _ = return ()+    get = return undefined++-- This generic curry is straightforward+class Curry a b | a -> b where+    curryFun :: a -> b++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+    curryFun f = \x -> curryFun (\y -> (f (x,y)))+++-- This generic uncurry courtesy Andrea Vezzosi+data HTrue+data HFalse+data Fun :: * -> * -> * -> * where+  Done :: Fun EndOfTuple r r+  Moar :: Fun xs f r -> Fun (x,xs) (x -> f) r++class Uncurry'' args func result | func -> args, func -> result, args result -> func where+    reify :: Fun args func result++class Uncurry flag args func result | flag func -> args, flag func -> result, args result -> func where+    reify' :: flag -> Fun args func result++instance Uncurry'' rest f r => Uncurry HTrue (a,rest) (a -> f) r where+    reify' _ = Moar reify++instance Uncurry HFalse EndOfTuple a a where+    reify' _ = Done++instance (IsFunction func b, Uncurry b args func result) => Uncurry'' args func result where+    reify = reify' (undefined :: b)++uncurry' :: Fun args func result -> func -> args -> result+uncurry' Done r _ = r+uncurry' (Moar fun) f (x,xs) = uncurry' fun (f x) xs++class IsFunction t b | t -> b+instance (b ~ HTrue) => IsFunction (a -> c) b+instance (b ~ HFalse) => IsFunction a b+

@@ -5,6 +5,7 @@   , disconnect   , closeImplicitReconnections   , impliesDeathOf+  , sendCtrlMsg   ) where  import Data.Accessor ((^.), (^=))@@ -13,8 +14,12 @@ 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.Chan (writeChan) import Control.Monad (unless)+import Control.Monad.IO.Class (MonadIO, liftIO)+import Control.Monad.Reader (ask) import qualified Network.Transport as NT   ( Connection   , send@@ -36,8 +41,11 @@   , ProcessSignal(Died)   , DiedReason(DiedDisconnect)   , ImplicitReconnect(WithImplicitReconnect)-  , NodeId-  , ProcessId(processNodeId)+  , NodeId(..)+  , ProcessId(..)+  , LocalNode(..)+  , LocalProcess(..)+  , Process(..)   , SendPortId(sendPortProcessId)   , Identifier(NodeIdentifier, ProcessIdentifier, SendPortIdentifier)   )@@ -167,3 +175,24 @@   cid' == cid _ `impliesDeathOf` _ =   False+++-- Send a control message+sendCtrlMsg :: Maybe NodeId  -- ^ Nothing for the local node+            -> ProcessSignal -- ^ Message to send+            -> Process ()+sendCtrlMsg mNid signal = do+  proc <- ask+  let msg = NCMsg { ctrlMsgSender = ProcessIdentifier (processId proc)+                  , ctrlMsgSignal = signal+                  }+  case mNid of+    Nothing -> do+      liftIO $ writeChan (localCtrlChan (processNode proc)) msg+    Just nid ->+      liftIO $ sendBinary (processNode proc)+                          (ProcessIdentifier (processId proc))+                          (NodeIdentifier nid)+                          WithImplicitReconnect+                          msg+

@@ -1,3 +1,10 @@+{-# LANGUAGE CPP  #-}+{-# LANGUAGE RankNTypes  #-}+{-# LANGUAGE DeriveDataTypeable  #-}+{-# LANGUAGE ScopedTypeVariables  #-}+{-# LANGUAGE ExistentialQuantification  #-}+{-# LANGUAGE BangPatterns #-}+ -- | Cloud Haskell primitives -- -- We define these in a separate module so that we don't have to rely on@@ -12,6 +19,10 @@   , receiveChan   , mergePortsBiased   , mergePortsRR+    -- * Unsafe messaging variants+  , unsafeSend+  , unsafeSendChan+  , unsafeNSend     -- * Advanced messaging   , Match   , receiveWait@@ -19,10 +30,24 @@   , match   , matchIf   , matchUnknown-  , AbstractMessage(..)   , matchAny   , matchAnyIf   , matchChan+  , matchSTM+  , matchMessage+  , matchMessageIf+  , isEncoded+  , wrapMessage+  , unsafeWrapMessage+  , unwrapMessage+  , handleMessage+  , handleMessageIf+  , handleMessage_+  , handleMessageIf_+  , forward+  , delegate+  , relay+  , proxy     -- * Process management   , terminate   , ProcessTerminationException(..)@@ -38,6 +63,9 @@   , getSelfNode   , ProcessInfo(..)   , getProcessInfo+  , NodeStats(..)+  , getNodeStats+  , getLocalNodeStats     -- * Monitoring and linking   , link   , unlink@@ -83,8 +111,8 @@     -- * Reconnecting   , reconnect   , reconnectPort-    -- * Tracing/Debugging-  , trace+    -- * Internal Exports+  , sendCtrlMsg   ) where  #if ! MIN_VERSION_base(4,6,0)@@ -107,7 +135,6 @@   , modifyMVar   , modifyMVar_   )-import Control.Concurrent.Chan (writeChan) import Control.Concurrent.STM   ( STM   , TVar@@ -124,9 +151,13 @@   ) import Control.Distributed.Process.Serializable (Serializable, fingerprint) import Data.Accessor ((^.), (^:), (^=))-import Control.Distributed.Static (Closure, Static)+import Control.Distributed.Static+  ( Static+  , Closure+  ) import Data.Rank1Typeable (Typeable) import qualified Control.Distributed.Static as Static (unstatic, unclosure)+import qualified Control.Distributed.Process.UnsafePrimitives as Unsafe import Control.Distributed.Process.Internal.Types   ( NodeId(..)   , ProcessId(..)@@ -136,8 +167,8 @@   , Message(..)   , MonitorRef(..)   , SpawnRef(..)-  , NCMsg(..)   , ProcessSignal(..)+  , NodeMonitorNotification(..)   , monitorCounter   , spawnCounter   , SendPort(..)@@ -148,6 +179,7 @@   , SendPortId(..)   , Identifier(..)   , ProcessExitException(..)+  , DiedReason(..)   , DidUnmonitor(..)   , DidUnlinkProcess(..)   , DidUnlinkNode(..)@@ -157,9 +189,12 @@   , ProcessRegistrationException(..)   , ProcessInfo(..)   , ProcessInfoNone(..)+  , NodeStats(..)+  , isEncoded   , createMessage+  , createUnencodedMessage   , runLocalProcess-  , ImplicitReconnect(WithImplicitReconnect, NoImplicitReconnect)+  , ImplicitReconnect( NoImplicitReconnect)   , LocalProcessState   , LocalSendPortId   , messageToPayload@@ -169,14 +204,22 @@   , sendBinary   , sendPayload   , disconnect+  , sendCtrlMsg   )-import qualified Control.Distributed.Process.Internal.Trace as Trace+import Control.Distributed.Process.Management.Internal.Types+  ( MxEvent(..)+  )+import Control.Distributed.Process.Management.Internal.Trace.Types+  ( traceEvent+  ) import Control.Distributed.Process.Internal.WeakTQueue   ( newTQueueIO   , readTQueue   , mkWeakTQueue   ) +import Unsafe.Coerce+ -------------------------------------------------------------------------------- -- Basic messaging                                                            -- --------------------------------------------------------------------------------@@ -184,15 +227,32 @@ -- | Send a message send :: Serializable a => ProcessId -> a -> Process () -- This requires a lookup on every send. If we want to avoid that we need to--- modify serializable to allow for stateful (IO) deserialization+-- modify serializable to allow for stateful (IO) deserialization. send them msg = do   proc <- ask-  liftIO $ sendMessage (processNode proc)-                       (ProcessIdentifier (processId proc))-                       (ProcessIdentifier them)-                       NoImplicitReconnect-                       msg+  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.+  liftIO $ traceEvent (localEventBus node)+                      (MxSent them us (createUnencodedMessage 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+-- node) therefore ensure that the payload is fully evaluated before it is+-- delivered.+unsafeSend :: Serializable a => ProcessId -> a -> Process ()+unsafeSend = Unsafe.send+ -- | Wait for a message of a specific type expect :: forall a. Serializable a => Process a expect = receiveWait [match return]@@ -229,12 +289,24 @@ sendChan :: Serializable a => SendPort a -> a -> Process () sendChan (SendPort cid) msg = do   proc <- ask-  liftIO $ sendBinary (processNode proc)-                      (ProcessIdentifier (processId proc))-                      (SendPortIdentifier cid)-                      NoImplicitReconnect-                      msg+  let node     = localNodeId (processNode proc)+      destNode = processNodeId (sendPortProcessId cid) in do+  case destNode == node of+    True  -> sendChanLocal cid msg+    False -> do+      liftIO $ sendBinary (processNode proc)+                          (ProcessIdentifier (processId proc))+                          (SendPortIdentifier cid)+                          NoImplicitReconnect+                          msg +-- | Send a message on a typed channel. This function makes /no/ attempt to+-- serialize and (in the case when the @ReceivePort@ resides on the same local+-- node) therefore ensure that the payload is fully evaluated before it is+-- delivered.+unsafeSendChan :: Serializable a => SendPort a -> a -> Process ()+unsafeSendChan = Unsafe.sendChan+ -- | Wait for a message on a typed channel receiveChan :: Serializable a => ReceivePort a -> Process a receiveChan = liftIO . atomically . receiveSTM@@ -300,9 +372,26 @@     Nothing   -> return Nothing     Just proc -> Just <$> proc +-- | Match on a typed channel matchChan :: ReceivePort a -> (a -> Process b) -> Match b matchChan p fn = Match $ MatchChan (fmap fn (receiveSTM p)) +-- | Match on an arbitrary STM action.+--+-- This rather unusaul /match primitive/ allows us to compose arbitrary STM+-- actions with checks against our process' mailbox and/or any typed channel+-- @ReceivePort@s we may hold.+--+-- This allows us to process multiple input streams /along with our mailbox/,+-- in just the same way that 'matchChan' supports checking both the mailbox+-- /and/ an arbitrary set of typed channels in one atomic transaction.+--+-- Note there are no ordering guarnatees with respect to these disparate input+-- sources.+--+matchSTM :: STM a -> (a -> Process b) -> Match b+matchSTM stm fn = Match $ MatchChan (fmap fn stm)+ -- | Match against any message of the right type match :: forall a b. Serializable a => (a -> Process b) -> Match b match = matchIf (const True)@@ -310,83 +399,238 @@ -- | Match against any message of the right type that satisfies a predicate matchIf :: forall a b. Serializable a => (a -> Bool) -> (a -> Process b) -> Match b matchIf c p = Match $ MatchMsg $ \msg ->-   case messageFingerprint msg == fingerprint (undefined :: a) of-     True | c decoded -> Just (p decoded)-       where-         decoded :: a-         -- Make sure the value is fully decoded so that we don't hang to-         -- bytestrings when the process calling 'matchIf' doesn't process-         -- the values immediately-         !decoded = decode (messageEncoding msg)-     _ -> Nothing+  case messageFingerprint msg == fingerprint (undefined :: a) of+    False -> Nothing+    True  -> case msg of+      (UnencodedMessage _ m) ->+        let m' = unsafeCoerce m :: a in+        case (c m') of+          True  -> Just (p m')+          False -> Nothing+      (EncodedMessage _ _) ->+        if (c decoded) then Just (p decoded) else Nothing+        where+          decoded :: a+            -- Make sure the value is fully decoded so that we don't hang to+            -- bytestrings when the process calling 'matchIf' doesn't process+            -- the values immediately+          !decoded = decode (messageEncoding msg) --- | Represents a received message and provides two basic operations on it.-data AbstractMessage = AbstractMessage {-    forward :: ProcessId -> Process () -- ^ forward the message to @ProcessId@-  , maybeHandleMessage :: forall a b. (Serializable a)-            => (a -> Process b) -> Process (Maybe b) {- ^ Handle the message.-        If the type of the message matches the type of the first argument to-        the supplied expression, then the expression will be evaluated against-        it. If this runtime type checking fails, then @Nothing@ will be returned-        to indicate the fact. If the check succeeds and evaluation proceeds-        however, the resulting value with be wrapped with @Just@.-    -}-  }+-- | Match against any message, regardless of the underlying (contained) type+matchMessage :: (Message -> Process Message) -> Match Message+matchMessage p = Match $ MatchMsg $ \msg -> Just (p msg) +-- | Match against any message (regardless of underlying type) that satisfies a predicate+matchMessageIf :: (Message -> Bool) -> (Message -> Process Message) -> Match Message+matchMessageIf c p = Match $ MatchMsg $ \msg ->+    case (c msg) of+      True  -> Just (p msg)+      False -> Nothing++-- | Forward a raw 'Message' to the given 'ProcessId'.+forward :: Message -> ProcessId -> Process ()+forward msg them = do+  proc <- ask+  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)+++-- | Wrap a 'Serializable' value in a 'Message'. Note that 'Message's are+-- 'Serializable' - like the datum they contain - but also note, deserialising+-- such a 'Message' will yield a 'Message', not the type within it! To obtain the+-- wrapped datum, use 'unwrapMessage' or 'handleMessage' with a specific type.+--+-- @+-- do+--    self <- getSelfPid+--    send self (wrapMessage "blah")+--    Nothing  <- expectTimeout 1000000 :: Process (Maybe String)+--    (Just m) <- expectTimeout 1000000 :: Process (Maybe Message)+--    (Just "blah") <- unwrapMessage m :: Process (Maybe String)+-- @+--+wrapMessage :: Serializable a => a -> Message+wrapMessage = createUnencodedMessage -- see [note Serializable UnencodedMessage]++-- | This is the /unsafe/ variant of 'wrapMessage'. See+-- "Control.Distributed.Process.UnsafePrimitives" for details.+unsafeWrapMessage :: Serializable a => a -> Message+unsafeWrapMessage = Unsafe.wrapMessage++-- [note Serializable UnencodedMessage]+-- if we attempt to serialise an UnencodedMessage, it will be converted to an+-- EncodedMessage first, so we're ok here. See 'messageToPayload' in+-- Primitives.hs for the gory details++-- | Attempt to unwrap a raw 'Message'.+-- If the type of the decoded message payload matches the expected type, the+-- value will be returned with @Just@, otherwise @Nothing@ indicates the types+-- do not match.+--+-- This expression, for example, will evaluate to @Nothing@+-- > unwrapMessage (wrapMessage "foobar") :: Process (Maybe Int)+--+-- Whereas this expression, will yield @Just "foo"@+-- > unwrapMessage (wrapMessage "foo") :: Process (Maybe String)+--+unwrapMessage :: forall m a. (Monad m, Serializable a) => Message -> m (Maybe a)+unwrapMessage msg =+  case messageFingerprint msg == fingerprint (undefined :: a) of+    False -> return Nothing :: m (Maybe a)+    True  -> case msg of+      (UnencodedMessage _ ms) ->+        let ms' = unsafeCoerce ms :: a+        in return (Just ms')+      (EncodedMessage _ _) ->+        return (Just (decoded))+        where+          decoded :: a -- note [decoding]+          !decoded = decode (messageEncoding msg)++-- | Attempt to handle a raw 'Message'.+-- If the type of the message matches the type of the first argument to+-- the supplied expression, then the message will be decoded and the expression+-- evaluated against its value. If this runtime type checking fails however,+-- @Nothing@ will be returned to indicate the fact. If the check succeeds and+-- evaluation proceeds, the resulting value with be wrapped with @Just@.+--+-- Intended for use in `catchesExit` and `matchAny` primitives.+--+handleMessage :: forall m a b. (Monad m, Serializable a)+              => Message -> (a -> m b) -> m (Maybe b)+handleMessage msg proc = handleMessageIf msg (const True) proc++-- | Conditionally handle a raw 'Message'.+-- If the predicate @(a -> Bool)@ evaluates to @True@, invokes the supplied+-- handler, other returns @Nothing@ to indicate failure. See 'handleMessage'+-- for further information about runtime type checking.+--+handleMessageIf :: forall m a b . (Monad m, Serializable a)+                => Message+                -> (a -> Bool)+                -> (a -> m b)+                -> m (Maybe b)+handleMessageIf msg c proc = do+  case messageFingerprint msg == fingerprint (undefined :: a) of+    False -> return Nothing :: m (Maybe b)+    True  -> case msg of+      (UnencodedMessage _ ms) ->+        let ms' = unsafeCoerce ms :: a in+        case (c ms') of+          True  -> do { r <- proc ms'; return (Just r) }+          False -> return Nothing :: m (Maybe b)+      (EncodedMessage _ _) ->+        case (c decoded) of+          True  -> do { r <- proc (decoded :: a); return (Just r) }+          False -> return Nothing :: m (Maybe b)+        where+          decoded :: a -- note [decoding]+          !decoded = decode (messageEncoding msg)++-- | As 'handleMessage' but ignores result, which is useful if you don't+-- care whether or not the handler succeeded.+--+handleMessage_ :: forall m a . (Monad m, Serializable a)+               => Message -> (a -> m ()) -> m ()+handleMessage_ msg proc = handleMessageIf_ msg (const True) proc++-- | Conditional version of 'handleMessage_'.+--+handleMessageIf_ :: forall m a . (Monad m, Serializable a)+                => Message+                -> (a -> Bool)+                -> (a -> m ())+                -> m ()+handleMessageIf_ msg c proc = handleMessageIf msg c proc >> return ()+ -- | Match against an arbitrary message. 'matchAny' removes the first available--- message from the process mailbox, and via the 'AbstractMessage' type,--- supports forwarding /or/ handling the message /if/ it is of the correct--- type. If /not/ of the right type, then the 'AbstractMessage'--- @maybeHandleMessage@ function will not evaluate the supplied expression,--- /but/ the message will still have been removed from the process mailbox!+-- message from the process mailbox. To handle arbitrary /raw/ messages once+-- removed from the mailbox, see 'handleMessage' and 'unwrapMessage'. ---matchAny :: forall b. (AbstractMessage -> Process b) -> Match b-matchAny p = Match $ MatchMsg $ Just . p . abstract+matchAny :: forall b. (Message -> Process b) -> Match b+matchAny p = Match $ MatchMsg $ \msg -> Just (p msg) --- | Match against an arbitrary message. 'matchAnyIf' will /only/ remove the--- message from the process mailbox, /if/ the supplied condition matches. The--- success (or failure) of runtime type checks in @maybeHandleMessage@ does not--- count here, i.e., if the condition evaluates to @True@ then the message will+-- | Match against an arbitrary message. Intended for use with 'handleMessage'+-- and 'unwrapMessage', this function /only/ removes a message from the process+-- mailbox, /if/ the supplied condition matches. The success (or failure) of+-- runtime type checks deferred to @handleMessage@ and friends is irrelevant+-- here, i.e., if the condition evaluates to @True@ then the message will -- be removed from the process mailbox and decoded, but that does /not/--- guarantee that an expression passed to @maybeHandleMessage@ will pass the--- runtime type checks and therefore be evaluated. If the types do not match--- up, then @maybeHandleMessage@ returns 'Nothing'.+-- guarantee that an expression passed to @handleMessage@ will pass the+-- runtime type checks and therefore be evaluated.+-- matchAnyIf :: forall a b. (Serializable a)                        => (a -> Bool)-                       -> (AbstractMessage -> Process b)+                       -> (Message -> Process b)                        -> Match b matchAnyIf c p = Match $ MatchMsg $ \msg ->-   case messageFingerprint msg == fingerprint (undefined :: a) of-     True | c decoded -> Just (p (abstract msg))+  case messageFingerprint msg == fingerprint (undefined :: a) of+     True | check -> Just (p msg)        where-         decoded :: a-         -- Make sure the value is fully decoded so that we don't hang to-         -- bytestrings when the calling process doesn't evaluate immediately+         check :: Bool+         !check =+           case msg of+             (EncodedMessage _ _)    -> c decoded+             (UnencodedMessage _ m') -> c (unsafeCoerce m')++         decoded :: a -- note [decoding]          !decoded = decode (messageEncoding msg)      _ -> Nothing -abstract :: Message -> AbstractMessage-abstract msg = AbstractMessage {-    forward = \them -> do-      proc <- ask-      liftIO $ sendPayload (processNode proc)-                           (ProcessIdentifier (processId proc))-                           (ProcessIdentifier them)-                           NoImplicitReconnect-                           (messageToPayload msg)-  , maybeHandleMessage = \(proc :: (a -> Process b)) -> do-      case messageFingerprint msg == fingerprint (undefined :: a) of-        True -> do { r <- proc (decoded :: a); return (Just r) }-          where-            decoded :: a-            !decoded = decode (messageEncoding msg)-        _ -> return Nothing-  }+{- note [decoding]+For an EncodedMessage, we need to ensure the value is fully decoded so that+we don't hang to bytestrings if the calling process doesn't evaluate+immediately. For UnencodedMessage we know (because the fingerprint comparison+succeeds) that unsafeCoerce will not fail.+-}  -- | Remove any message from the queue matchUnknown :: Process b -> Match b matchUnknown p = Match $ MatchMsg (const (Just p)) +-- | Receives messages and forwards them to @pid@ if @p msg == True@.+delegate :: ProcessId -> (Message -> Bool) -> Process ()+delegate pid p = do+  receiveWait [+      matchAny (\m -> case (p m) of+                        True  -> forward m pid+                        False -> return ())+    ]+  delegate pid p++-- | A straight relay that forwards all messages to the supplied pid.+relay :: ProcessId -> Process ()+relay !pid = receiveWait [ matchAny (\m -> forward m pid) ] >> relay pid++-- | Proxies @pid@ and forwards messages whenever @proc@ evaluates to @True@.+-- Unlike 'delegate' the predicate @proc@ runs in the 'Process' monad, allowing+-- for richer proxy behaviour. If @proc@ returns @False@ or the /runtime type check/+-- fails, no action is taken and the proxy process will continue running.+proxy :: Serializable a => ProcessId -> (a -> Process Bool) -> Process ()+proxy pid proc = do+  receiveWait [+      matchAny (\m -> do+                   next <- handleMessage m proc+                   case next of+                     Just True  -> forward m pid+                     Just False -> return ()  -- explicitly ignored+                     Nothing    -> return ()) -- un-routable+    ]+  proxy pid proc+ -------------------------------------------------------------------------------- -- Process management                                                         -- --------------------------------------------------------------------------------@@ -405,6 +649,9 @@ -- | Die immediately - throws a 'ProcessExitException' with the given @reason@. die :: Serializable a => a -> Process b die reason = do+-- NOTE: We do /not/ want to use UnencodedMessage here, as the exception+-- 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)) @@ -454,23 +701,23 @@ -- | Lift 'Control.Exception.catches' (almost). -- -- As 'ProcessExitException' stores the exit @reason@ as a typed, encoded--- message, a handler must accept an input of the expected type. In order to+-- message, a handler must accept inputs of the expected type. In order to -- handle a list of potentially different handlers (and therefore input types),--- a handler passed to 'catchesExit' must accept 'AbstractMessage' and return+-- a handler passed to 'catchesExit' must accept 'Message' and return -- @Maybe@ (i.e., @Just p@ if it handled the exit reason, otherwise @Nothing@). ----- See 'maybeHandleMessage' and 'AsbtractMessage' for more details.+-- See 'maybeHandleMessage' and 'Message' for more details. catchesExit :: Process b-            -> [(ProcessId -> AbstractMessage -> (Process (Maybe b)))]+            -> [(ProcessId -> Message -> (Process (Maybe b)))]             -> Process b catchesExit act handlers = catch act ((flip handleExit) handlers)   where     handleExit :: ProcessExitException-               -> [(ProcessId -> AbstractMessage -> Process (Maybe b))]+               -> [(ProcessId -> Message -> Process (Maybe b))]                -> Process b     handleExit ex [] = liftIO $ throwIO ex     handleExit ex@(ProcessExitException from msg) (h:hs) = do-      r <- h from (abstract msg)+      r <- h from msg       case r of         Nothing -> handleExit ex hs         Just p  -> return p@@ -483,6 +730,31 @@ getSelfNode :: Process NodeId getSelfNode = localNodeId . processNode <$> ask ++-- | Get statistics about the specified node+getNodeStats :: NodeId -> Process (Either DiedReason NodeStats)+getNodeStats nid = do+    selfNode <- getSelfNode+    if nid == selfNode+      then Right `fmap` getLocalNodeStats -- optimisation+      else getNodeStatsRemote nid+  where+    getNodeStatsRemote :: NodeId -> Process (Either DiedReason NodeStats)+    getNodeStatsRemote nid = do+        sendCtrlMsg (Just nid) $ GetNodeStats nid+        bracket (monitorNode nid) unmonitor $ \mRef ->+            receiveWait [ match (\(stats :: NodeStats) -> return $ Right stats)+                        , matchIf (\(NodeMonitorNotification ref _ _) -> ref == mRef)+                                  (\(NodeMonitorNotification _ _ dr) -> return $ Left dr)+                        ]++-- | Get statistics about our local node+getLocalNodeStats :: Process NodeStats+getLocalNodeStats = do+  self <- getSelfNode+  sendCtrlMsg Nothing $ GetNodeStats self+  receiveWait [ match (\(stats :: NodeStats) -> return stats) ]+ -- | Get information about the specified process getProcessInfo :: ProcessId -> Process (Maybe ProcessInfo) getProcessInfo pid =@@ -623,10 +895,13 @@ mask p = do     lproc <- ask     liftIO $ Ex.mask $ \restore ->-      runLocalProcess lproc (p (liftRestore lproc restore))+      runLocalProcess lproc (p (liftRestore restore))   where-    liftRestore :: LocalProcess -> (forall a. IO a -> IO a) -> (forall a. Process a -> Process a)-    liftRestore lproc restoreIO = liftIO . restoreIO . runLocalProcess lproc+    liftRestore :: (forall a. IO a -> IO a)+                -> (forall a. Process a -> Process a)+    liftRestore restoreIO = \p2 -> do+      ourLocalProc <- ask+      liftIO $ restoreIO $ runLocalProcess ourLocalProc p2  -- | Lift 'Control.Exception.onException' onException :: Process a -> Process b -> Process a@@ -654,7 +929,7 @@ data Handler a = forall e . Exception e => Handler (e -> Process a)  instance Functor Handler where-     fmap f (Handler h) = Handler (fmap f . h)+    fmap f (Handler h) = Handler (fmap f . h)  -- | Lift 'Control.Exception.catches' catches :: Process a -> [Handler a] -> Process a@@ -831,8 +1106,15 @@ -- | 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))+  sendCtrlMsg Nothing (NamedSend label (createUnencodedMessage msg)) +-- | Named send to a process in the local 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.+unsafeNSend :: Serializable a => String -> a -> Process ()+unsafeNSend = Unsafe.nsend+ -- | Named send to a process in a remote registry (asynchronous) nsendRemote :: Serializable a => NodeId -> String -> a -> Process () nsendRemote nid label msg =@@ -896,51 +1178,19 @@   liftIO $ disconnect node (ProcessIdentifier us) (SendPortIdentifier (sendPortId them))  ----------------------------------------------------------------------------------- Debugging/Tracing                                                          --+-- Auxiliary functions                                                        -- -------------------------------------------------------------------------------- --- | Send a message to the internal (system) trace facility. If tracing is--- enabled, this will create a custom trace event. Note that several Cloud Haskell--- sub-systems also generate trace events for informational/debugging purposes,--- thus traces generated this way will not be the only output seen.------ Just as with the "Debug.Trace" module, this is a debugging/tracing facility--- for use in development, and should not be used in a production setting ---- which is why the default behaviour is to trace to the GHC eventlog. For a--- general purpose logging facility, you should consider 'say'.------ Trace events can be written to the GHC event log, a text file, or to the--- standard system logger process (see 'say'). The default behaviour for writing--- to the eventlog requires specific intervention to work, without which traces--- are silently dropped/ignored and no output will be generated.--- The GHC eventlog documentation provides information about enabling, viewing--- and working with event traces: <http://hackage.haskell.org/trac/ghc/wiki/EventLog>.------ When a new local node is started, the contents of the environment variable--- @DISTRIBUTED_PROCESS_TRACE_FILE@ are checked for a valid file path. If this--- exists and the file can be opened for writing, all trace output will be directed--- thence. If the environment variable is empty, the path invalid, or the file--- unavailable for writing - e.g., because another node has already started--- tracing to it - then the @DISTRIBUTED_PROCESS_TRACE_CONSOLE@ environment--- variable is checked for /any/ non-empty value. If this is set, then all trace--- output will be directed to the system logger process. If neither evironment--- variable provides a valid trace configuration, all internal traces are written--- to "Debug.Trace.traceEventIO", which writes to the GHC eventlog.------ Users of the /simplelocalnet/ Cloud Haskell backend should also note that--- because the trace file option only supports trace output from a single node--- (so as to avoid interleaving), a file trace configured for the master node will--- prevent slaves from tracing to the file and they will fall back to using the--- console/'say' or eventlog instead.----trace :: String -> Process ()-trace s = do-  node <- processNode <$> ask-  liftIO $ Trace.trace (localTracer node) s+sendLocal :: (Serializable a) => ProcessId -> a -> Process ()+sendLocal to msg =+  sendCtrlMsg Nothing $ LocalSend to (createUnencodedMessage msg) ------------------------------------------------------------------------------------ Auxiliary functions                                                        -----------------------------------------------------------------------------------+sendChanLocal :: (Serializable a) => SendPortId -> a -> Process ()+sendChanLocal spId msg =+  -- we *must* fully serialize/encode the message here, because+  -- attempting to use `unsafeCoerce' in the node controller+  -- won't work since we know nothing about the required type+  sendCtrlMsg Nothing $ LocalPortSend spId (createUnencodedMessage msg)  getMonitorRefFor :: Identifier -> Process MonitorRef getMonitorRefFor ident = do@@ -970,22 +1220,3 @@ -- | Link to a process/node/channel link' :: Identifier -> Process () link' = sendCtrlMsg Nothing . Link---- Send a control message-sendCtrlMsg :: Maybe NodeId  -- ^ Nothing for the local node-            -> ProcessSignal -- ^ Message to send-            -> Process ()-sendCtrlMsg mNid signal = do-  proc <- ask-  let msg = NCMsg { ctrlMsgSender = ProcessIdentifier (processId proc)-                  , ctrlMsgSignal = signal-                  }-  case mNid of-    Nothing -> do-      liftIO $ writeChan (localCtrlChan (processNode proc)) msg-    Just nid ->-      liftIO $ sendBinary (processNode proc)-                          (ProcessIdentifier (processId proc))-                          (NodeIdentifier nid)-                          WithImplicitReconnect-                          msg

@@ -0,0 +1,161 @@+{-# LANGUAGE RankNTypes  #-}+module Control.Distributed.Process.Internal.Spawn+  ( spawn+  , spawnLink+  , spawnMonitor+  , call+  , spawnSupervised+  , spawnChannel+  ) where++import Control.Distributed.Static+  ( Static+  , Closure+  , closureCompose+  , staticClosure+  )+import Control.Distributed.Process.Internal.Types+  ( NodeId(..)+  , ProcessId(..)+  , Process(..)+  , MonitorRef(..)+  , ProcessMonitorNotification(..)+  , NodeMonitorNotification(..)+  , DidSpawn(..)+  , SendPort(..)+  , ReceivePort(..)+  , nullProcessId+  )+import Control.Distributed.Process.Serializable (Serializable, SerializableDict)+import Control.Distributed.Process.Internal.Closure.BuiltIn+  ( sdictSendPort+  , sndStatic+  , idCP+  , seqCP+  , bindCP+  , splitCP+  , cpLink+  , cpSend+  , cpNewChan+  , cpDelay+  )+import Control.Distributed.Process.Internal.Primitives+  ( -- Basic messaging+    send+  , expect+  , receiveWait+  , match+  , matchIf+  , link+  , getSelfPid+  , monitor+  , monitorNode+  , unmonitor+  , spawnAsync+  , reconnect+  )++-- | Spawn a process+--+-- For more information about 'Closure', see+-- "Control.Distributed.Process.Closure".+--+-- See also 'call'.+spawn :: NodeId -> Closure (Process ()) -> Process ProcessId+spawn nid proc = do+  us   <- getSelfPid+  mRef <- monitorNode nid+  sRef <- spawnAsync nid (cpDelay us proc)+  receiveWait [+      matchIf (\(DidSpawn ref _) -> ref == sRef) $ \(DidSpawn _ pid) -> do+        unmonitor mRef+        send pid ()+        return pid+    , matchIf (\(NodeMonitorNotification ref _ _) -> ref == mRef) $ \_ ->+        return (nullProcessId nid)+    ]++-- | Spawn a process and link to it+--+-- Note that this is just the sequential composition of 'spawn' and 'link'.+-- (The "Unified" semantics that underlies Cloud Haskell does not even support+-- a synchronous link operation)+spawnLink :: NodeId -> Closure (Process ()) -> Process ProcessId+spawnLink nid proc = do+  pid <- spawn nid proc+  link pid+  return pid++-- | Like 'spawnLink', but monitor the spawned process+spawnMonitor :: NodeId -> Closure (Process ()) -> Process (ProcessId, MonitorRef)+spawnMonitor nid proc = do+  pid <- spawn nid proc+  ref <- monitor pid+  return (pid, ref)++-- | Run a process remotely and wait for it to reply+--+-- We monitor the remote process: if it dies before it can send a reply, we die+-- too.+--+-- For more information about 'Static', 'SerializableDict', and 'Closure', see+-- "Control.Distributed.Process.Closure".+--+-- See also 'spawn'.+call :: Serializable a+        => Static (SerializableDict a)+        -> NodeId+        -> Closure (Process a)+        -> Process a+call dict nid proc = do+  us <- getSelfPid+  (pid, mRef) <- spawnMonitor nid (proc `bindCP` cpSend dict us)+  -- We are guaranteed to receive the reply before the monitor notification+  -- (if a reply is sent at all)+  -- NOTE: This might not be true if we switch to unreliable delivery.+  mResult <- receiveWait+    [ match (return . Right)+    , matchIf (\(ProcessMonitorNotification ref _ _) -> ref == mRef)+              (\(ProcessMonitorNotification _ _ reason) -> return (Left reason))+    ]+  case mResult of+    Right a  -> do+      -- Wait for the monitor message so that we the mailbox doesn't grow+      receiveWait+        [ matchIf (\(ProcessMonitorNotification ref _ _) -> ref == mRef)+                  (\(ProcessMonitorNotification {}) -> return ())+        ]+      -- Clean up connection to pid+      reconnect pid+      return a+    Left err ->+      fail $ "call: remote process died: " ++ show err++-- | Spawn a child process, have the child link to the parent and the parent+-- monitor the child+spawnSupervised :: NodeId+                -> Closure (Process ())+                -> Process (ProcessId, MonitorRef)+spawnSupervised nid proc = do+  us   <- getSelfPid+  them <- spawn nid (cpLink us `seqCP` proc)+  ref  <- monitor them+  return (them, ref)++-- | Spawn a new process, supplying it with a new 'ReceivePort' and return+-- the corresponding 'SendPort'.+spawnChannel :: forall a. Serializable a => Static (SerializableDict a)+             -> NodeId+             -> Closure (ReceivePort a -> Process ())+             -> Process (SendPort a)+spawnChannel dict nid proc = do+    us <- getSelfPid+    _ <- spawn nid (go us)+    expect+  where+    go :: ProcessId -> Closure (Process ())+    go pid = cpNewChan dict+           `bindCP`+             (cpSend (sdictSendPort dict) pid `splitCP` proc)+           `bindCP`+             (idCP `closureCompose` staticClosure sndStatic)

@@ -6,6 +6,7 @@   , newMVar   , takeMVar   , putMVar+  , readMVar   , withMVar   , modifyMVar_   , modifyMVar@@ -21,6 +22,7 @@   , newMVar   , takeMVar   , putMVar+  , readMVar   , withMVar   , modifyMVar_   , modifyMVar@@ -43,6 +45,9 @@  putMVar :: StrictMVar a -> a -> IO () putMVar (StrictMVar v) x = evaluate x >> MVar.putMVar v x++readMVar :: StrictMVar a -> IO a+readMVar (StrictMVar v) = MVar.readMVar v  withMVar :: StrictMVar a -> (a -> IO b) -> IO b withMVar (StrictMVar v) = MVar.withMVar v

@@ -1,122 +0,0 @@--- | Simple (internal) system logging/tracing support.-module Control.Distributed.Process.Internal.Trace-  ( Tracer-  , TraceArg(..)-  , trace-  , traceFormat-  , startTracing-  , stopTracer-  ) where--import Control.Concurrent (forkIO)-import Control.Concurrent.Chan (writeChan)-import Control.Concurrent.STM-  ( TQueue-  , newTQueueIO-  , readTQueue-  , writeTQueue-  , atomically-  )-import Control.Distributed.Process.Internal.Types-  ( Tracer(..)-  , LocalNode(..)-  , NCMsg(..)-  , Identifier(ProcessIdentifier)-  , ProcessSignal(NamedSend)-  , forever'-  , nullProcessId-  , createMessage-  )-import Control.Exception-  ( catch-  , throwTo-  , SomeException-  , AsyncException(ThreadKilled)-  )-import Data.List (intersperse)-import Data.Time.Clock (getCurrentTime)-import Data.Time.Format (formatTime)-import Debug.Trace (traceEventIO)--import Prelude hiding (catch)--import System.Environment (getEnv)-import System.IO-  ( Handle-  , IOMode(AppendMode)-  , BufferMode(..)-  , openFile-  , hClose-  , hPutStrLn-  , hSetBuffering-  )-import System.Locale (defaultTimeLocale)--data TraceArg = -    TraceStr String-  | forall a. (Show a) => Trace a--startTracing :: LocalNode -> IO LocalNode-startTracing node = do-  tracer <- defaultTracer node-  return node { localTracer = tracer }--defaultTracer :: LocalNode -> IO Tracer-defaultTracer node = do-  catch (getEnv "DISTRIBUTED_PROCESS_TRACE_FILE" >>= logfileTracer)-        (\(_ :: IOError) -> defaultTracerAux node)--defaultTracerAux :: LocalNode -> IO Tracer-defaultTracerAux node = do-  catch (getEnv "DISTRIBUTED_PROCESS_TRACE_CONSOLE" >> procTracer node)-        (\(_ :: IOError) -> return (EventLogTracer traceEventIO))-  where procTracer :: LocalNode -> IO Tracer-        procTracer n = return $ (LocalNodeTracer n)--logfileTracer :: FilePath -> IO Tracer-logfileTracer p = do-    q <- newTQueueIO-    h <- openFile p AppendMode-    hSetBuffering h LineBuffering-    tid <- forkIO $ logger h q `catch` (\(_ :: SomeException) ->-                                         hClose h >> return ())-    return $ LogFileTracer tid q h-  where logger :: Handle -> TQueue String -> IO ()-        logger h q' = forever' $ do-          msg <- atomically $ readTQueue q'-          now <- getCurrentTime-          hPutStrLn h $ msg ++ (formatTime defaultTimeLocale " - %c" now)---- TODO: compatibility layer for GHC/base versions (e.g., where's killThread?)--stopTracer :: Tracer -> IO ()  -- overzealous but harmless duplication of hClose-stopTracer (LogFileTracer tid _ h) = throwTo tid ThreadKilled >> hClose h-stopTracer _                       = return ()--trace :: Tracer -> String -> IO ()-trace (LogFileTracer _ q _) msg = atomically $ writeTQueue q msg-trace (LocalNodeTracer n)   msg = sendTraceMsg n msg-trace (EventLogTracer t)    msg = t msg-trace InactiveTracer        _   = return ()--traceFormat :: Tracer-            -> String-            -> [TraceArg]-            -> IO ()-traceFormat t d ls =-    trace t $ concat (intersperse d (map toS ls))-  where toS :: TraceArg -> String-        toS (TraceStr s) = s-        toS (Trace    a) = show a--sendTraceMsg :: LocalNode -> String -> IO ()-sendTraceMsg node string = do-  now <- getCurrentTime-  msg <- return $ (formatTime defaultTimeLocale "%c" now, string)-  emptyPid <- return $ (nullProcessId (localNodeId node))-  traceMsg <- return $ NCMsg {-                         ctrlMsgSender = ProcessIdentifier (emptyPid)-                       , ctrlMsgSignal = (NamedSend "logger" (createMessage msg))-                       }-  writeChan (localCtrlChan node) traceMsg-

@@ -1,3 +1,9 @@+{-# LANGUAGE DeriveDataTypeable  #-}+{-# LANGUAGE ExistentialQuantification  #-}+{-# LANGUAGE GeneralizedNewtypeDeriving  #-}+{-# LANGUAGE GADTs  #-}+{-# LANGUAGE DeriveGeneric #-}+ -- | Types used throughout the Cloud Haskell framework -- -- We collect all types used internally in a single module because@@ -15,6 +21,7 @@     -- * Local nodes and processes   , LocalNode(..)   , Tracer(..)+  , MxEventBus(..)   , LocalNodeState(..)   , LocalProcess(..)   , LocalProcessState(..)@@ -29,7 +36,10 @@   , ReceivePort(..)     -- * Messages   , Message(..)+  , isEncoded   , createMessage+  , createUnencodedMessage+  , unsafeCreateUnencodedMessage   , messageToPayload   , payloadToMessage     -- * Node controller user-visible data types@@ -53,6 +63,7 @@   , RegisterReply(..)   , ProcessInfo(..)   , ProcessInfoNone(..)+  , NodeStats(..)     -- * Node controller internal data types   , NCMsg(..)   , ProcessSignal(..)@@ -75,7 +86,7 @@ import System.Mem.Weak (Weak) import Data.Map (Map) import Data.Int (Int32)-import Data.Typeable (Typeable)+import Data.Typeable (Typeable, typeOf) import Data.Binary (Binary(put, get), putWord8, getWord8, encode) import qualified Data.ByteString as BSS (ByteString, concat, copy) import qualified Data.ByteString.Lazy as BSL@@ -83,15 +94,17 @@   , toChunks   , splitAt   , fromChunks+  , length   ) import qualified Data.ByteString.Lazy.Internal as BSL (ByteString(..)) import Data.Accessor (Accessor, accessor) import Control.Category ((>>>))+import Control.DeepSeq (NFData(..)) import Control.Exception (Exception) import Control.Concurrent (ThreadId) import Control.Concurrent.Chan (Chan) import Control.Concurrent.STM (STM)-import qualified Control.Concurrent.STM as STM (TQueue)+import Control.Concurrent.STM.TChan (TChan) import qualified Network.Transport as NT (EndPoint, EndPointAddress, Connection) import Control.Applicative (Applicative, Alternative, (<$>), (<*>)) import Control.Monad.Reader (MonadReader(..), ReaderT, runReaderT)@@ -110,16 +123,20 @@ import Control.Distributed.Process.Internal.WeakTQueue (TQueue) import Control.Distributed.Static (RemoteTable, Closure) import qualified Control.Distributed.Process.Internal.StrictContainerAccessors as DAC (mapMaybe)-import System.IO (Handle) +import Data.Hashable+import GHC.Generics+ -------------------------------------------------------------------------------- -- Node and process identifiers                                               -- --------------------------------------------------------------------------------  -- | Node identifier newtype NodeId = NodeId { nodeAddress :: NT.EndPointAddress }-  deriving (Eq, Ord, Binary, Typeable)-+  deriving (Eq, Ord, Typeable, Generic)+instance Binary NodeId where+instance NFData NodeId+instance Hashable NodeId where instance Show NodeId where   show (NodeId addr) = "nid://" ++ show addr @@ -129,8 +146,10 @@   { lpidUnique  :: {-# UNPACK #-} !Int32   , lpidCounter :: {-# UNPACK #-} !Int32   }-  deriving (Eq, Ord, Typeable, Show)+  deriving (Eq, Ord, Typeable, Generic, Show) +instance Hashable LocalProcessId where+ -- | Process identifier data ProcessId = ProcessId   { -- | The ID of the node the process is running on@@ -138,8 +157,12 @@     -- | Node-local identifier for the process   , processLocalId :: {-# UNPACK #-} !LocalProcessId   }-  deriving (Eq, Ord, Typeable)+  deriving (Eq, Ord, Typeable, Generic) +instance Binary ProcessId where+instance NFData ProcessId where+instance Hashable ProcessId where+ instance Show ProcessId where   show (ProcessId (NodeId addr) (LocalProcessId _ lid))     = "pid://" ++ show addr ++ ":" ++ show lid@@ -149,8 +172,10 @@     NodeIdentifier !NodeId   | ProcessIdentifier !ProcessId   | SendPortIdentifier !SendPortId-  deriving (Eq, Ord)+  deriving (Eq, Ord, Generic) +instance Hashable Identifier where+ instance Show Identifier where   show (NodeIdentifier nid)     = show nid   show (ProcessIdentifier pid)  = show pid@@ -180,13 +205,31 @@ -- Local nodes and processes                                                  -- -------------------------------------------------------------------------------- --- | Required for system tracing in the node controller-data Tracer =-    LogFileTracer   !ThreadId !(STM.TQueue String) !Handle-  | EventLogTracer  !(String -> IO ())-  | LocalNodeTracer !LocalNode-  | InactiveTracer  -- NB: never used, this is required to initialize LocalNode+-- | Provides access to the trace controller+data Tracer = Tracer+              {+                -- | Process id for the currently active trace handler+                tracerPid :: !ProcessId+                -- | Weak reference to the tracer controller's mailbox+              , weakQ     :: !(Weak (CQueue Message))+              } +-- | Local system management event bus state+data MxEventBus =+    MxEventBusInitialising+  | MxEventBus+    {+      -- | Process id of the management agent controller process+      agent  :: !ProcessId+      -- | Configuration for the local trace controller+    , tracer :: !Tracer+      -- | Weak reference to the management agent controller's mailbox+    , evbuss :: !(Weak (CQueue Message))+      -- | API for adding management agents to a running node+    , mxNew  :: !(((TChan Message, TChan Message) -> Process ()) -> IO ProcessId)+--    , mxReg  :: !(StrictMVar (Map MxAgentId ))+    }+ -- | Local nodes data LocalNode = LocalNode   { -- | 'NodeId' of the node@@ -197,8 +240,8 @@   , localState      :: !(StrictMVar LocalNodeState)     -- | Channel for the node controller   , localCtrlChan   :: !(Chan NCMsg)-    -- | Current active system debug/trace log-  , localTracer     :: !Tracer+    -- | Internal management event bus+  , localEventBus   :: !MxEventBus     -- | Runtime lookup table for supporting closures     -- TODO: this should be part of the CH state, not the local endpoint state   , remoteTable     :: !RemoteTable@@ -265,8 +308,9 @@     -- | Process-local ID of the channel   , sendPortLocalId   :: {-# UNPACK #-} !LocalSendPortId   }-  deriving (Eq, Ord)+  deriving (Eq, Ord, Typeable, Generic) +instance Hashable SendPortId where instance Show SendPortId where   show (SendPortId (ProcessId (NodeId addr) (LocalProcessId _ plid)) clid)     = "cid://" ++ show addr ++ ":" ++ show plid ++ ":" ++ show clid@@ -278,8 +322,12 @@     -- | The (unique) ID of this send port     sendPortId :: SendPortId   }-  deriving (Typeable, Binary, Show, Eq, Ord)+  deriving (Typeable, Generic, Show, Eq, Ord) +instance (Serializable a) => Binary (SendPort a) where+instance (Hashable a) => Hashable (SendPort a) where+instance (NFData a) => NFData (SendPort a) where+ -- | The receive end of a typed channel (not serializable) -- -- Note that 'ReceivePort' implements 'Functor', 'Applicative', 'Alternative'@@ -303,25 +351,57 @@ --------------------------------------------------------------------------------  -- | Messages consist of their typeRep fingerprint and their encoding-data Message = Message+data Message =+  EncodedMessage   { messageFingerprint :: !Fingerprint   , messageEncoding    :: !BSL.ByteString+  } |+  forall a . Serializable a =>+  UnencodedMessage+  {+    messageFingerprint :: !Fingerprint+  , messagePayload     :: !a   }+  deriving (Typeable)  instance Show Message where-  show (Message fp enc) = show enc ++ " :: " ++ showFingerprint fp []+  show (EncodedMessage fp enc) = show enc ++ " :: " ++ showFingerprint fp []+  show (UnencodedMessage _ uenc) = "[unencoded message] :: " ++ (show $ typeOf uenc) +-- | /internal use only/.+isEncoded :: Message -> Bool+isEncoded (EncodedMessage _ _) = True+isEncoded _                    = False+-- [note] isEncoded:+-- This is just as internal as it looks and yes, it does feel a bit odd that+-- we're exporting it for use, however DPP does a /lot/ of work with low+-- level APIs such as unwrapMessage and handleMessage, and in the process+-- tries very hard to avoid copying (and re-serialisation) where possible.+-- Being able to determine that a message is encoded (or otherwise) makes+-- that a lot more manageable.+ -- | Turn any serialiable term into a message createMessage :: Serializable a => a -> Message-createMessage a = Message (fingerprint a) (encode a)+createMessage a = EncodedMessage (fingerprint a) (encode a) +-- | Turn any serializable term into an unencoded/local message+createUnencodedMessage :: Serializable a => a -> Message+createUnencodedMessage a =+  let encoded = encode a in BSL.length encoded `seq` UnencodedMessage (fingerprint a) a++-- | Turn any serializable term into an unencodede/local message, without+-- evalutaing it! This is a dangerous business.+unsafeCreateUnencodedMessage :: Serializable a => a -> Message+unsafeCreateUnencodedMessage a = UnencodedMessage (fingerprint a) a+ -- | Serialize a message messageToPayload :: Message -> [BSS.ByteString]-messageToPayload (Message fp enc) = encodeFingerprint fp : BSL.toChunks enc+messageToPayload (EncodedMessage fp enc) = encodeFingerprint fp : BSL.toChunks enc+messageToPayload (UnencodedMessage fp m) = messageToPayload ((EncodedMessage fp (encode m)))  -- | Deserialize a message payloadToMessage :: [BSS.ByteString] -> Message-payloadToMessage payload = Message fp (copy msg)+payloadToMessage payload = EncodedMessage fp (copy msg)   where     encFp :: BSL.ByteString     msg   :: BSL.ByteString@@ -346,7 +426,8 @@     -- | Unique to distinguish multiple monitor requests by the same process   , monitorRefCounter :: !Int32   }-  deriving (Eq, Ord, Show)+  deriving (Eq, Ord, Show, Typeable, Generic)+instance Hashable MonitorRef where  -- | Message sent by process monitors data ProcessMonitorNotification =@@ -446,6 +527,15 @@ data RegisterReply = RegisterReply String Bool   deriving (Show, Typeable) +data NodeStats = NodeStats {+     nodeStatsNode            :: NodeId+   , nodeStatsRegisteredNames :: Int+   , nodeStatsMonitors        :: Int+   , nodeStatsLinks           :: Int+   , nodeStatsProcesses       :: Int+   }+   deriving (Show, Eq, Typeable)+ -- | Provide information about a running process data ProcessInfo = ProcessInfo {     infoNode               :: NodeId@@ -480,23 +570,27 @@   | WhereIs !String   | Register !String !NodeId !(Maybe ProcessId) !Bool -- Use 'Nothing' to unregister, use True to force reregister   | NamedSend !String !Message+  | LocalSend !ProcessId !Message+  | LocalPortSend !SendPortId !Message   | Kill !ProcessId !String   | Exit !ProcessId !Message   | GetInfo !ProcessId+  | SigShutdown+  | GetNodeStats !NodeId   deriving Show  -------------------------------------------------------------------------------- -- Binary instances                                                           -- -------------------------------------------------------------------------------- +instance Binary Message where+  put msg = put $ messageToPayload msg+  get = payloadToMessage <$> get+ instance Binary LocalProcessId where   put lpid = put (lpidUnique lpid) >> put (lpidCounter lpid)   get      = LocalProcessId <$> get <*> get -instance Binary ProcessId where-  put pid = put (processNodeId pid) >> put (processLocalId pid)-  get     = ProcessId <$> get <*> get- instance Binary ProcessMonitorNotification where   put (ProcessMonitorNotification ref pid reason) = put ref >> put pid >> put reason   get = ProcessMonitorNotification <$> get <*> get <*> get@@ -518,18 +612,22 @@   get     = MonitorRef <$> get <*> get  instance Binary ProcessSignal where-  put (Link pid)            = putWord8 0 >> put pid-  put (Unlink pid)          = putWord8 1 >> put pid-  put (Monitor ref)         = putWord8 2 >> put ref-  put (Unmonitor ref)       = putWord8 3 >> put ref-  put (Died who reason)     = putWord8 4 >> put who >> put reason-  put (Spawn proc ref)      = putWord8 5 >> put proc >> put ref-  put (WhereIs label)       = putWord8 6 >> put label+  put (Link pid)              = putWord8 0 >> put pid+  put (Unlink pid)            = putWord8 1 >> put pid+  put (Monitor ref)           = putWord8 2 >> put ref+  put (Unmonitor ref)         = putWord8 3 >> put ref+  put (Died who reason)       = putWord8 4 >> put who >> put reason+  put (Spawn proc ref)        = putWord8 5 >> put proc >> put ref+  put (WhereIs label)         = putWord8 6 >> put label   put (Register label nid pid force) = putWord8 7 >> put label >> put nid >> put pid >> put force-  put (NamedSend label msg) = putWord8 8 >> put label >> put (messageToPayload msg)-  put (Kill pid reason)     = putWord8 9 >> put pid >> put reason-  put (Exit pid reason)     = putWord8 10 >> put pid >> put (messageToPayload reason)-  put (GetInfo about)       = putWord8 30 >> put about+  put (NamedSend label msg)   = putWord8 8 >> put label >> put (messageToPayload msg)+  put (Kill pid reason)       = putWord8 9 >> put pid >> put reason+  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 (GetInfo about)         = putWord8 30 >> put about+  put (SigShutdown)         = putWord8 31+  put (GetNodeStats nid)         = putWord8 32 >> put nid   get = do     header <- getWord8     case header of@@ -544,7 +642,11 @@       8  -> NamedSend <$> get <*> (payloadToMessage <$> get)       9  -> Kill <$> get <*> get       10 -> Exit <$> get <*> (payloadToMessage <$> get)+      11 -> LocalSend <$> get <*> (payloadToMessage <$> get)+      12 -> LocalPortSend <$> get <*> (payloadToMessage <$> get)       30 -> GetInfo <$> get+      31 -> return SigShutdown+      32 -> GetNodeStats <$> get       _ -> fail "ProcessSignal.get: invalid"  instance Binary DiedReason where@@ -571,6 +673,9 @@   put cid = put (sendPortProcessId cid) >> put (sendPortLocalId cid)   get = SendPortId <$> get <*> get +instance NFData SendPortId where+  rnf cid = (sendPortProcessId cid) `seq` (sendPortLocalId cid) `seq` ()+ instance Binary Identifier where   put (ProcessIdentifier pid)  = putWord8 0 >> put pid   put (NodeIdentifier nid)     = putWord8 1 >> put nid@@ -599,6 +704,14 @@            >> put (infoMonitors pInfo)            >> put (infoLinks pInfo) +instance Binary NodeStats where+  get = NodeStats <$> get <*> get <*> get <*> get <*> get+  put nStats =  put (nodeStatsNode nStats)+             >> put (nodeStatsRegisteredNames nStats)+             >> put (nodeStatsMonitors nStats)+             >> put (nodeStatsLinks nStats)+             >> put (nodeStatsProcesses nStats)+ instance Binary ProcessInfoNone where   get = ProcessInfoNone <$> get   put (ProcessInfoNone r) = put r@@ -623,7 +736,7 @@ localProcessWithId lpid = localProcesses >>> DAC.mapMaybe lpid  localConnectionBetween :: Identifier -> Identifier -> Accessor LocalNodeState (Maybe (NT.Connection, ImplicitReconnect))-localConnectionBetween from to = localConnections >>> DAC.mapMaybe (from, to)+localConnectionBetween from' to' = localConnections >>> DAC.mapMaybe (from', to')  monitorCounter :: Accessor LocalProcessState Int32 monitorCounter = accessor _monitorCounter (\cnt st -> st { _monitorCounter = cnt })

@@ -1,10 +1,11 @@+{-# LANGUAGE DeriveDataTypeable  #-}+{-# LANGUAGE MagicHash, UnboxedTuples #-} -- | Clone of Control.Concurrent.STM.TQueue with support for mkWeakTQueue -- -- Not all functionality from the original module is available: unGetTQueue, -- peekTQueue and tryPeekTQueue are missing. In order to implement these we'd -- need to be able to touch# the write end of the queue inside unGetTQueue, but -- that means we need a version of touch# that works within the STM monad.-{-# LANGUAGE MagicHash, UnboxedTuples #-} module Control.Distributed.Process.Internal.WeakTQueue (   -- * Original functionality   TQueue,

@@ -0,0 +1,557 @@+{-# LANGUAGE DeriveDataTypeable         #-}+{-# LANGUAGE DeriveGeneric              #-}+{-# LANGUAGE StandaloneDeriving         #-}+{-# LANGUAGE ExistentialQuantification  #-}+{-# LANGUAGE ScopedTypeVariables        #-}+{-# LANGUAGE RecordWildCards            #-}+{-# LANGUAGE PatternGuards              #-}+{-# LANGUAGE FlexibleInstances          #-}+{-# LANGUAGE UndecidableInstances       #-}++-----------------------------------------------------------------------------+-- |+-- Module      :  Control.Distributed.Process.Management+-- Copyright   :  (c) Well-Typed / Tim Watson+-- License     :  BSD3 (see the file LICENSE)+--+-- Maintainer  :  Tim Watson <watson.timothy@gmail.com>+-- Stability   :  experimental+-- Portability :  non-portable (requires concurrency)+--+-- [Management Extensions API]+--+-- This module presents an API for creating /Management Agents/:+-- special processes that are capable of receiving and responding to+-- a node's internal system events. These /system events/ are delivered by+-- the management event bus: An internal subsystem maintained for each+-- running node, to which all agents are automatically subscribed.+--+-- /Agents/ are defined in terms of /event sinks/, taking a particular+-- @Serializable@ type and evaluating to an action in the 'MxAgent' monad in+-- response. Each 'MxSink' evaluates to an 'MxAction' that specifies whether+-- the agent should continue processing it's inputs or stop. If the type of a+-- message cannot be matched to any of the agent's sinks, it will be discarded.+-- A sink can also deliberately skip processing a message, deferring to the+-- remaining handlers. This is the /only/ way that more than one event sink+-- can handle the same data type, since otherwise the first /type match/ will+-- /win/ every time a message arrives. See 'mxSkip' for details.+--+-- Various events are published to the management event bus automatically,+-- the full list of which can be found in the definition of the 'MxEvent' data+-- type. Additionally, clients of the /Management API/ can publish arbitrary+-- @Serializable@ data to the event bus using 'mxNotify'. All running agents+-- receive all events (from the primary event bus to which they're subscribed).+--+-- Agent processes are automatically registered on the local node, and can+-- receive messages via their mailbox just like ordinary processes. Unlike+-- ordinary @Process@ code however, it is unnecessary (though possible) for+-- agents to use the base @expect@ and @receiveX@ primitives to do this, since+-- the management infrastructure will continuously read from both the primary+-- event bus /and/ the process' own mailbox. Messages are transparently passed+-- to the agent's event sinks from both sources, so an agent need only concern+-- itself with how to respond to its inputs.+--+-- Some agents may wish to prioritise messages from their mailbox over traffic+-- on the management event bus, or vice versa. The 'mxReceive' and+-- 'mxReceiveChan' API calls do this for the mailbox and event bus,+-- respectively. The /prioritisation/ these APIs offer is simply that the chosen+-- data stream will be checked first. No blocking will occur if the chosen+-- (prioritised) source is devoid of input messages, instead the agent handling+-- code will revert to switching between the alternatives in /round-robin/ as+-- usual. If messages exist in one or more channels, they will be consumed as+-- soon as they're available, priority is effectively a hint about which+-- channel to consume from, should messages be available in both.+--+-- Prioritisation then, is a /hint/ about the preference of data source from+-- which the next input should be chosen. No guarantee can be made that the+-- chosen source will in fact be selected at runtime.+--+-- [Management API Semantics]+--+-- The management API provides /no guarantees whatsoever/, viz:+--+--  * The ordering of messages delivered to the event bus.+--+--  * The order in which agents will be executed.+--+--  * Whether messages will be taken from the mailbox first, or the event bus.+--+-- [Management Data API]+--+-- Both management agents and clients of the API have access to a variety of+-- data storage capabilities, to facilitate publishing and consuming useful+-- 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.+--+-- [Defining Agents]+--+-- New agents are defined with 'mxAgent' and require a unique 'MxAgentId', an+-- initial state - 'MxAgent' runs in a state transformer - and a list of the+-- agent's event sinks. Each 'MxSink' is defined in terms of a specific+-- @Serializable@ type, via the 'mxSink' function, binding the event handler+-- expression to inputs of only that type.+--+-- Apart from modifying its own local state, an agent can execute arbitrary+-- @Process a@ code via lifting (see 'liftMX') and even publish its own messages+-- back to the primary event bus (see 'mxBroadcast').+--+-- Since messages are delivered to agents from both the management event bus and+-- the agent processes mailbox, agents (i.e., event sinks) will generally have+-- no idea as to their origin. An agent can, however, choose to prioritise the+-- choice of input (source) each time one of its event sinks runs. The /standard/+-- way for an event sink to indicate that the agent is ready for its next input+-- is to evaluate 'mxReady'. When this happens, the management infrastructure+-- will obtain data from the event bus and process' mailbox in a round robbin+-- fashion, i.e., one after the other, changing each time.+--+-- [Example Code]+--+-- What follows is a grossly over-simplified example of a management agent that+-- provides a basic name monitoring facility. Whenever a process name is+-- registered or unregistered, clients are informed of the fact.+--+-- > -- simple notification data type+-- >+-- > data Registration = Reg { added  :: Bool+-- >                         , procId :: ProcessId+-- >                         , name   :: String+-- >                         }+-- >+-- > -- start a /name monitoring agent/+-- > nameMonitorAgent = do+-- >   mxAgent (MxAgentId "name-monitor") Set.empty [+-- >         (mxSink $ \(pid :: ProcessId) -> do+-- >            mxUpdateState $ Set.insert pid+-- >            mxReady)+-- >       , (mxSink $+-- >             let act =+-- >                   case ev of+-- >                     (MxRegistered   p n) -> notify True  n p+-- >                     (MxUnRegistered p n) -> notify False n p+-- >                     _                    -> return ()+-- >             act >> mxReady)+-- >     ]+-- >   where+-- >     notify a n p = do+-- >       Foldable.mapM_ (liftMX . deliver (Reg a n p)) =<< mxGetLocal+-- >+--+-- The client interface (for sending their pid) can take one of two forms:+--+-- > 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:+-- broadcast communication with them can have a lower latency, they offer+-- simplified messgage (i.e., input type) handling and they have access to+-- internal system information that would be otherwise unobtainable.+--+-- Do not be tempted to implement everything (e.g., the kitchen sink) using the+-- management API though. There are overheads associated with management agents+-- which is why they're presented as tools for consuming low level system+-- information, instead of as /application level/ development tools.+--+-- Agents that rely heavily on a busy mailbox can cause the management event+-- bus to backlog un-GC'ed data, leading to increased heap space. Producers that+-- do not take care to avoid passing unevaluated thunks to the API can crash+-- /all/ the agents in the system. Agents are not monitored or managed in any+-- way, and those that crash will not be restarted.+--+-- The management event bus can receive a great deal of traffic. Every time+-- a message is sent and/or received, an event is passed to the agent controller+-- and broadcast to all agents (plus the trace controller, if tracing is enabled+-- for the node). This is already a significant overhead - though profiling and+-- benchmarks have demonstrated that it does not adversely affect performance+-- if few agents are installed. Agents will typically use more cycles than plain+-- processes, since they perform additional work: selecting input data from both+-- 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+-- change without prior notice. The description that follows is provided for+-- informational purposes only.+--+-- When a node initially starts, two special, internal system processes are+-- started to support the management infrastructure. The first, known as the+-- /trace controller/, is responsible for consuming 'MxEvent's and forwarding+-- them to the configured tracer - see "Control.Distributed.Process.Debug" for+-- further details. The second is the /management agent controller/, and is the+-- primary worker process underpinning the management infrastructure. All+-- published management events are routed to this process, which places them+-- onto a system wide /event bus/ and additionally passes them directly to the+-- /trace controller/.+--+-- There are several reasons for segregating the tracing and management control+-- planes in this fashion. Tracing can be enabled or disabled by clients, whilst+-- the management event bus cannot, since in addition to providing+-- runtime instrumentation, its intended use-cases include node monitoring, peer+-- discovery (via topology providing backends) and other essential system+-- services that require knowledge of otherwise hidden system internals. Tracing+-- is also subject to /trace flags/ that limit the specific 'MxEvent's delivered+-- to trace clients - an overhead/complexity not shared by management agents.+-- Finally, tracing and management agents are implemented using completely+-- different signalling techniques - more on this later - which would introduce+-- considerable complexity if the shared the same /event loop/.+--+-- The management control plane is driven by a shared broadcast channel, which+-- is written to by the agent controller and subscribed to by all agent+-- processes. Agents are spawned as regular processes, whose primary+-- implementation (i.e., /server loop/) is responsible for consuming+-- messages from both the broadcast channel and their own mailbox. Once+-- consumed, messages are applied to the agent's /event sinks/ until one+-- matches the input, at which point it is applied and the loop continues.+-- The implementation chooses from the event bus and the mailbox in a+-- round-robin fashion, until a message is received. This polling activity would+-- lead to management agents consuming considerable system resources if left+-- unchecked, therefore the implementation will poll for a limitted number of+-- retries, after which it will perform a blocking read on the event bus.+--+-----------------------------------------------------------------------------+module Control.Distributed.Process.Management+  (+    MxEvent(..)+    -- * Firing Arbitrary /Mx Events/+  , mxNotify+    -- * Constructing Mx Agents+  , MxAction()+  , MxAgentId(..)+  , MxAgent()+  , mxAgent+  , mxAgentWithFinalize+  , MxSink()+  , mxSink+  , mxGetId+  , mxDeactivate+  , mxReady+  , mxSkip+  , mxReceive+  , mxReceiveChan+  , mxBroadcast+  , mxSetLocal+  , mxGetLocal+  , mxUpdateLocal+  , liftMX+    -- * Mx Data API+  , mxPublish+  , mxSet+  , mxGet+  , mxClear+  , mxPurgeTable+  , mxDropTable+  ) where++import Control.Applicative ((<$>))+import Control.Concurrent.STM (atomically)+import Control.Concurrent.STM.TChan+  ( tryReadTChan+  , readTChan+  , writeTChan+  , TChan+  )+import Control.Distributed.Process.Internal.Primitives+  ( newChan+  , nsend+  , receiveWait+  , receiveTimeout+  , matchChan+  , matchAny+  , matchSTM+  , unwrapMessage+  , onException+  , register+  , whereis+  , die+  )+import Control.Distributed.Process.Internal.Types+  ( Process+  , ProcessId+  , Message+  , LocalProcess(..)+  , LocalNode(..)+  , MxEventBus(..)+  , 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 qualified Control.Monad.State as ST+  ( MonadState+  , StateT+  , get+  , modify+  , lift+  , runStateT+  )++-- | Publishes an arbitrary @Serializable@ message to the management event bus.+-- 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 the+-- receiving process via this API, since /all/ registered agents will gain+-- access to the data structure once it is broadcast by the agent controller.+mxNotify :: (Serializable a) => a -> Process ()+mxNotify msg = do+  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)          --+--------------------------------------------------------------------------------++-- | Return the 'MxAgentId' for the currently executing agent.+--+mxGetId :: MxAgent s MxAgentId+mxGetId = ST.get >>= return . mxAgentId++-- | The 'MxAgent' version of 'mxNotify'.+--+mxBroadcast :: (Serializable m) => m -> MxAgent s ()+mxBroadcast msg = do+  state <- ST.get+  liftMX $ liftIO $ atomically $ do+    writeTChan (mxBus state) (unsafeCreateUnencodedMessage msg)++-- | Gracefully terminate an agent.+--+mxDeactivate :: forall s. String -> MxAgent s MxAction+mxDeactivate = return . MxAgentDeactivate++-- | Continue executing (i.e., receiving and processing messages).+--+mxReady :: forall s. MxAgent s MxAction+mxReady = return MxAgentReady++-- | Causes the currently executing /event sink/ to be skipped.+-- The remaining declared event sinks will be evaluated to find+-- a matching handler. Can be used to allow multiple event sinks+-- to process data of the same type.+--+mxSkip :: forall s. MxAgent s MxAction+mxSkip = return MxAgentSkip++-- | Continue exeucting, prioritising inputs from the process' own+-- /mailbox/ ahead of data from the management event bus.+--+mxReceive :: forall s. MxAgent s MxAction+mxReceive = return $ MxAgentPrioritise Mailbox++-- | Continue exeucting, prioritising inputs from the management event bus+-- over the process' own /mailbox/.+--+mxReceiveChan :: forall s. MxAgent s MxAction+mxReceiveChan = return $ MxAgentPrioritise InputChan++-- | Lift a @Process@ action.+--+liftMX :: Process a -> MxAgent s a+liftMX p = MxAgent $ ST.lift p++-- | Set the agent's local state.+--+mxSetLocal :: s -> MxAgent s ()+mxSetLocal s = ST.modify $ \st -> st { mxLocalState = s }++-- | Update the agent's local state.+--+mxUpdateLocal :: (s -> s) -> MxAgent s ()+mxUpdateLocal f = ST.modify $ \st -> st { mxLocalState = (f $ mxLocalState st) }++-- | Fetch the agent's local state.+--+mxGetLocal :: MxAgent s s+mxGetLocal = ST.get >>= return . mxLocalState++-- | Create an 'MxSink' from an expression taking a @Serializable@ type @m@,+-- that yields an 'MxAction' in the 'MxAgent' monad.+--+mxSink :: forall s m . (Serializable m)+       => (m -> MxAgent s MxAction)+       -> MxSink s+mxSink act msg = do+  msg' <- liftMX $ (unwrapMessage msg :: Process (Maybe m))+  case msg' of+    Nothing -> return Nothing+    Just m  -> do+      r <- act m+      case r of+        MxAgentSkip -> return Nothing+        _           -> return $ Just r++-- private ADT: a linked list of event sinks+data MxPipeline s =+  MxPipeline+  {+    current  :: !(MxSink s)+  , next     :: !(MxPipeline s)+  } | MxStop++-- | Activates a new agent.+--+mxAgent :: MxAgentId -> s -> [MxSink s] -> Process ProcessId+mxAgent mxId st hs = mxAgentWithFinalize mxId st hs $ return ()++-- | Activates a new agent. This variant takes a /finalizer/ expression,+-- that is run once the agent shuts down (even in case of failure/exceptions).+-- The /finalizer/ expression runs in the mx monad -  @MxAgent s ()@ - such+-- that the agent's internal state remains accessible to the shutdown/cleanup+-- code.+--+mxAgentWithFinalize :: MxAgentId+        -> s+        -> [MxSink s]+        -> MxAgent s ()+        -> Process ProcessId+mxAgentWithFinalize mxId initState handlers dtor = do+    let name = agentId mxId+    existing <- whereis name+    case existing of+      Just _  -> die "DuplicateAgentId"  -- TODO: better error handling policy+      Nothing -> do+        node <- processNode <$> ask+        pid <- liftIO $ mxNew (localEventBus node) $ start+        register name pid+        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+      runAgent dtor handlers InputChan recvTChan nState++    runAgent :: MxAgent s ()+             -> [MxSink s]+             -> ChannelSelector+             -> TChan Message+             -> MxAgentState s+             -> Process ()+    runAgent eh hs cs c s =+      runAgentWithFinalizer eh hs cs c s+        `onException` runAgentFinalizer eh s++    runAgentWithFinalizer :: MxAgent s ()+                          -> [MxSink s]+                          -> ChannelSelector+                          -> TChan Message+                          -> MxAgentState s+                          -> Process ()+    runAgentWithFinalizer eh' hs' cs' c' s' = do+      msg <- getNextInput cs' c'+      (action, state) <- runPipeline msg s' $ pipeline hs'+      case action of+        MxAgentReady               -> runAgent eh' hs' InputChan c' state+        MxAgentPrioritise priority -> runAgent eh' hs' priority  c' state+        MxAgentDeactivate _        -> runAgentFinalizer eh' state+        MxAgentSkip                -> error "IllegalState"+--      MxAgentBecome h'           -> runAgent h' c state++    getNextInput sel chan =+      let stmRead = atomically . readTChan+          matches =+            case sel of+              Mailbox   -> [ matchAny return+                           , matchSTM (readTChan chan) return]+              InputChan -> [ matchSTM (readTChan chan) return+                           , matchAny return]+      in receiveWait matches++    runAgentFinalizer :: MxAgent s () -> MxAgentState s -> Process ()+    runAgentFinalizer f s = ST.runStateT (unAgent f) s >>= return . fst++    pipeline :: forall s . [MxSink s] -> MxPipeline s+    pipeline []           = MxStop+    pipeline (sink:sinks) = MxPipeline sink (pipeline sinks)++    runPipeline :: forall s .+                   Message+                -> MxAgentState s+                -> MxPipeline s+                -> Process (MxAction, MxAgentState s)+    runPipeline _   state MxStop         = return (MxAgentReady, state)+    runPipeline msg state MxPipeline{..} = do+      let act = current msg+      (pass, state') <- ST.runStateT (unAgent act) state+      case pass of+        Nothing     -> runPipeline msg state next+        Just result -> return (result, state')+

@@ -0,0 +1,156 @@+{-# LANGUAGE ScopedTypeVariables  #-}+{-# LANGUAGE RecordWildCards #-}++module Control.Distributed.Process.Management.Internal.Agent where++import Control.Applicative ((<$>))+import Control.Concurrent (forkIO)+import Control.Concurrent.MVar (MVar, newEmptyMVar, putMVar, takeMVar)+import Control.Concurrent.STM (atomically)+import Control.Concurrent.STM.TChan+  ( TChan+  , newBroadcastTChanIO+  , readTChan+  , writeTChan+  , dupTChan+  )+import Control.Distributed.Process.Internal.Primitives+  ( receiveWait+  , matchAny+  , die+  , catches+  , Handler(..)+  )+import Control.Distributed.Process.Internal.CQueue+  ( enqueueSTM+  , CQueue+  )+import Control.Distributed.Process.Management.Internal.Types+  ( Fork+  )+import Control.Distributed.Process.Management.Internal.Trace.Tracer+  ( traceController+  )+import Control.Distributed.Process.Internal.Types+  ( Process+  , Message+  , Tracer(..)+  , LocalProcess(..)+  , ProcessId+  , forever'+  )+import Control.Exception (AsyncException(ThreadKilled), SomeException)+import Control.Monad (void)+import Control.Monad.IO.Class (liftIO)+import Control.Monad.Reader (ask)+import GHC.Weak (Weak, deRefWeak)++--------------------------------------------------------------------------------+-- Agent Controller Implementation                                            --+--------------------------------------------------------------------------------++-- | A triple containing a configured tracer, weak pointer to the+-- agent controller's mailbox (CQueue) and an expression used to+-- instantiate new agents on the current node.+type AgentConfig =+  (Tracer, Weak (CQueue Message),+   (((TChan Message, TChan Message) -> Process ()) -> IO ProcessId))++-- | Starts a management agent for the current node. The agent process+-- must not crash or be killed, so we generally avoid publishing its+-- @ProcessId@ where possible.+--+-- Our process is also responsible for forwarding messages to the trace+-- controller, since having two /special processes/ handled via the+-- @LocalNode@ would be inelegant. We forward messages directly to the+-- trace controller's message queue, just as the @MxEventBus@ that's+-- set up on the @LocalNode@ forwards messages directly to us. This+-- optimises the code path for tracing and avoids overloading the node+-- node controller's internal control plane with additional routing, at the+-- cost of a little more complexity and two cases where we break+-- encapsulation.+--+mxAgentController :: Fork+                  -> MVar AgentConfig+                  -> Process ()+mxAgentController forkProcess mv = do+    trc <- liftIO $ startTracing forkProcess+    sigbus <- liftIO $ newBroadcastTChanIO+    liftIO $ startDeadLetterQueue sigbus+    weakQueue <- processWeakQ <$> ask+    liftIO $ putMVar mv (trc, weakQueue, mxStartAgent forkProcess sigbus)+    go sigbus trc+  where+    go bus tracer = forever' $ do+      void $ receiveWait [+          -- This is exactly what it appears to be: a "catch all" handler.+          -- Since mxNotify can potentially pass an unevaluated thunk to+          -- our mailbox, the dequeue (i.e., matchMessage) can fail and+          -- crash this process, which we DO NOT want. Alternatively,+          -- we handle IO exceptions here explicitly, since we don't want+          -- this process to ever crash, and the assumption we therefore+          -- make is thus:+          --+          -- 1. only ThreadKilled can tell this process to terminate+          -- 2. all other exceptions are invalid and should be ignored+          --+          -- The outcome of course, is that /bad/ calls to mxNotify+          -- (e.g., passing unevaluated thunks that will crash when+          -- they're eventually forced) are thus silently ignored.+          --+          matchAny (liftIO . broadcast bus tracer)+        ] `catches` [Handler (\ThreadKilled -> die "Killed"),+                     Handler (\(_ :: SomeException) -> return ())]++    broadcast :: TChan Message -> Tracer -> Message -> IO ()+    broadcast ch tr msg = do+      tmQueue <- tracerQueue tr+      atomicBroadcast ch tmQueue msg++    tracerQueue :: Tracer -> IO (Maybe (CQueue Message))+    tracerQueue (Tracer _ wQ) = deRefWeak wQ++    atomicBroadcast :: TChan Message+                    -> Maybe (CQueue Message)+                    -> Message -> IO ()+    atomicBroadcast ch Nothing  msg = liftIO $ atomically $ writeTChan ch msg+    atomicBroadcast ch (Just q) msg = do+      -- liftIO $ putStrLn $ "broadcasting " ++ (show msg)+      liftIO $ atomically $ enqueueSTM q msg >> writeTChan ch msg++-- | Forks a new process in which an mxAgent is run.+--+mxStartAgent :: Fork+             -> TChan Message+             -> ((TChan Message, TChan Message) -> Process ())+             -> IO ProcessId+mxStartAgent fork chan handler = do+  chan' <- atomically (dupTChan chan)+  let proc = handler (chan, chan')+  fork proc++-- | Start the tracer controller.+--+startTracing :: Fork -> IO Tracer+startTracing forkProcess = do+  mv  <- newEmptyMVar+  pid <- forkProcess $ traceController mv+  wQ  <- liftIO $ takeMVar mv+  return $ Tracer pid wQ++-- | Start a dead letter (agent) queue.+--+-- If no agents are registered on the system, the management+-- event bus will fill up and its data won't be GC'ed until someone+-- comes along and reads from the broadcast channel (via dupTChan+-- of course). This is effectively a leak, so to mitigate it, we+-- start a /dead letter queue/ that drains the event bus continuously,+-- thus ensuring if there are no other consumers that we won't use+-- up heap space unnecessarily.+--+startDeadLetterQueue :: TChan Message+                     -> IO ()+startDeadLetterQueue sigbus = do+  chan' <- atomically (dupTChan sigbus)+  void $ forkIO $ forever' $ do+    void $ atomically $ readTChan chan'

@@ -0,0 +1,21 @@+-- | Interface to the management event bus.+module Control.Distributed.Process.Management.Internal.Bus+  ( publishEvent+  ) where++import Control.Distributed.Process.Internal.CQueue+  ( enqueue+  )+import Control.Distributed.Process.Internal.Types+  ( MxEventBus(..)+  , Message+  )+import Data.Foldable (forM_)+import System.Mem.Weak (deRefWeak)++publishEvent :: MxEventBus -> Message -> IO ()+publishEvent MxEventBusInitialising _     = return ()+publishEvent (MxEventBus _ _ wqRef _) msg =  do+  mQueue <- deRefWeak wqRef+  forM_ mQueue $ \queue -> enqueue queue msg+

@@ -0,0 +1,222 @@+{-# 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 })

@@ -0,0 +1,172 @@+{-# LANGUAGE FlexibleInstances  #-}+{-# LANGUAGE TypeSynonymInstances  #-}+-- | Keeps the tracing API calls separate from the Tracer implementation,+-- which allows us to avoid a nasty import cycle between tracing and+-- the messaging primitives that rely on it, and also between the node+-- controller (which requires access to the tracing related elements of+-- our RemoteTable) and the Debug module, which requires @forkProcess@.+-- This module is also used by the management agent, which relies on the+-- tracing infrastructure's messaging fabric.+module Control.Distributed.Process.Management.Internal.Trace.Primitives+  ( -- * Sending Trace Data+    traceLog+  , traceLogFmt+  , traceMessage+    -- * Configuring A Tracer+  , defaultTraceFlags+  , enableTrace+  , enableTraceAsync+  , disableTrace+  , disableTraceAsync+  , getTraceFlags+  , setTraceFlags+  , setTraceFlagsAsync+  , traceOnly+  , traceOn+  , traceOff+  , withLocalTracer+  , withRegisteredTracer+  ) where++import Control.Applicative ((<$>))+import Control.Distributed.Process.Internal.Primitives+  ( whereis+  , newChan+  , receiveChan+  )+import Control.Distributed.Process.Management.Internal.Trace.Types+  ( TraceArg(..)+  , TraceFlags(..)+  , TraceOk(..)+  , TraceSubject(..)+  , defaultTraceFlags+  )+import qualified Control.Distributed.Process.Management.Internal.Trace.Types as Tracer+  ( traceLog+  , traceLogFmt+  , traceMessage+  , enableTrace+  , enableTraceSync+  , disableTrace+  , disableTraceSync+  , setTraceFlags+  , setTraceFlagsSync+  , getTraceFlags+  , getCurrentTraceClient+  )+import Control.Distributed.Process.Internal.Types+  ( Process+  , ProcessId+  , LocalProcess(..)+  , LocalNode(localEventBus)+  , SendPort+  , MxEventBus(..)+  )+import Control.Distributed.Process.Serializable+import Control.Monad.IO.Class (liftIO)+import Control.Monad.Reader (ask)++import qualified Data.Set as Set (fromList)++--------------------------------------------------------------------------------+-- Main API                                                                   --+--------------------------------------------------------------------------------++-- | Converts a list of identifiers (that can be+-- mapped to process ids), to a 'TraceSubject'.+class Traceable a where+  uod :: [a] -> TraceSubject++instance Traceable ProcessId where+  uod = TraceProcs . Set.fromList++instance Traceable String where+  uod = TraceNames . Set.fromList++-- | Turn tracing for for a subset of trace targets.+traceOnly :: Traceable a => [a] -> Maybe TraceSubject+traceOnly = Just . uod++-- | Trace all targets.+traceOn :: Maybe TraceSubject+traceOn = Just TraceAll++-- | Trace no targets.+traceOff :: Maybe TraceSubject+traceOff = Nothing++-- | Enable tracing to the supplied process.+enableTraceAsync :: ProcessId -> Process ()+enableTraceAsync pid = withLocalTracer $ \t -> liftIO $ Tracer.enableTrace t pid++-- TODO: refactor _Sync versions of trace configuration functions...++-- | Enable tracing to the supplied process and wait for a @TraceOk@+-- response from the trace coordinator process.+enableTrace :: ProcessId -> Process ()+enableTrace pid =+  withLocalTracerSync $ \t sp -> Tracer.enableTraceSync t sp pid++-- | Disable the currently configured trace.+disableTraceAsync :: Process ()+disableTraceAsync = withLocalTracer $ \t -> liftIO $ Tracer.disableTrace t++-- | Disable the currently configured trace and wait for a @TraceOk@+-- response from the trace coordinator process.+disableTrace :: Process ()+disableTrace =+  withLocalTracerSync $ \t sp -> Tracer.disableTraceSync t sp++getTraceFlags :: Process TraceFlags+getTraceFlags = do+  (sp, rp) <- newChan+  withLocalTracer $ \t -> liftIO $ Tracer.getTraceFlags t sp+  receiveChan rp++-- | Set the given flags for the current tracer.+setTraceFlagsAsync :: TraceFlags -> Process ()+setTraceFlagsAsync f = withLocalTracer $ \t -> liftIO $ Tracer.setTraceFlags t f++-- | Set the given flags for the current tracer and wait for a @TraceOk@+-- response from the trace coordinator process.+setTraceFlags :: TraceFlags -> Process ()+setTraceFlags f =+  withLocalTracerSync $ \t sp -> Tracer.setTraceFlagsSync t sp f++-- | Send a log message to the internal tracing facility. If tracing is+-- enabled, this will create a custom trace log event.+--+traceLog :: String -> Process ()+traceLog s = withLocalTracer $ \t -> liftIO $ Tracer.traceLog t s++-- | Send a log message to the internal tracing facility, using the given+-- list of printable 'TraceArg's interspersed with the preceding delimiter.+--+traceLogFmt :: String -> [TraceArg] -> Process ()+traceLogFmt d ls = withLocalTracer $ \t -> liftIO $ Tracer.traceLogFmt t d ls++-- | Send an arbitrary 'Message' to the tracer process.+traceMessage :: Serializable m => m -> Process ()+traceMessage msg = withLocalTracer $ \t -> liftIO $ Tracer.traceMessage t msg++withLocalTracer :: (MxEventBus -> Process ()) -> Process ()+withLocalTracer act = do+  node <- processNode <$> ask+  act (localEventBus node)++withLocalTracerSync :: (MxEventBus -> SendPort TraceOk -> IO ()) -> Process ()+withLocalTracerSync act = do+  (sp, rp) <- newChan+  withLocalTracer $ \t -> liftIO $ (act t sp)+  TraceOk <- receiveChan rp+  return ()++withRegisteredTracer :: (ProcessId -> Process a) -> Process a+withRegisteredTracer act = do+  (sp, rp) <- newChan+  withLocalTracer $ \t -> liftIO $ Tracer.getCurrentTraceClient t sp+  currentTracer <- receiveChan rp+  case currentTracer of+    Nothing  -> do { (Just p') <- whereis "tracer.initial"; act p' }+    (Just p) -> act p+

@@ -0,0 +1,61 @@+module Control.Distributed.Process.Management.Internal.Trace.Remote+  ( -- * Configuring A Remote Tracer+    setTraceFlagsRemote+  , startTraceRelay+    -- * Remote Table+  , remoteTable+  ) where++import Control.Distributed.Process.Internal.Closure.BuiltIn+  ( cpEnableTraceRemote+  )+import Control.Distributed.Process.Internal.Primitives+  ( getSelfPid+  , relay+  , nsendRemote+  )+import Control.Distributed.Process.Management.Internal.Trace.Types+  ( TraceFlags(..)+  , TraceOk(..)+  )+import Control.Distributed.Process.Management.Internal.Trace.Primitives+  ( withRegisteredTracer+  , enableTrace+  )+import Control.Distributed.Process.Internal.Spawn+  ( spawn+  )+import Control.Distributed.Process.Internal.Types+  ( Process+  , ProcessId+  , SendPort+  , NodeId+  )+import Control.Distributed.Static+  ( RemoteTable+  , registerStatic+  )+import Data.Rank1Dynamic (toDynamic)++-- | Remote Table.+remoteTable :: RemoteTable -> RemoteTable+remoteTable = registerStatic "$enableTraceRemote" (toDynamic enableTraceRemote)++enableTraceRemote :: ProcessId -> Process ()+enableTraceRemote pid =+  getSelfPid >>= enableTrace >> relay pid++-- | Starts a /trace relay/ process on the remote node, which forwards all trace+-- events to the registered tracer on /this/ (the calling process') node.+startTraceRelay :: NodeId -> Process ProcessId+startTraceRelay nodeId = do+  withRegisteredTracer $ \pid ->+    spawn nodeId $ cpEnableTraceRemote pid++-- | Set the given flags for a remote node (asynchronous).+setTraceFlagsRemote :: TraceFlags -> NodeId -> Process ()+setTraceFlagsRemote flags node = do+  nsendRemote node+              "trace.controller"+              ((Nothing :: Maybe (SendPort TraceOk)), flags)+

@@ -0,0 +1,347 @@+{-# LANGUAGE CPP  #-}+{-# LANGUAGE ScopedTypeVariables  #-}+-- | Tracing/Debugging support - Trace Implementation+module Control.Distributed.Process.Management.Internal.Trace.Tracer+  ( -- * API for the Management Agent+    traceController+    -- * Built in tracers+  , defaultTracer+  , systemLoggerTracer+  , logfileTracer+  , eventLogTracer+  ) where++import Control.Applicative ((<$>))+import Control.Concurrent.Chan (writeChan)+import Control.Concurrent.MVar+  ( MVar+  , putMVar+  )+import Control.Distributed.Process.Internal.CQueue+  ( CQueue+  )+import Control.Distributed.Process.Internal.Primitives+  ( catch+  , finally+  , die+  , receiveWait+  , forward+  , sendChan+  , match+  , matchAny+  , matchIf+  , handleMessage+  , matchUnknown+  )+import Control.Distributed.Process.Management.Internal.Types+  ( MxEvent(..)+  , Addressable(..)+  )+import Control.Distributed.Process.Management.Internal.Trace.Types+  ( SetTrace(..)+  , TraceSubject(..)+  , TraceFlags(..)+  , TraceOk(..)+  , defaultTraceFlags+  )+import Control.Distributed.Process.Management.Internal.Trace.Primitives+  ( traceOn )+import Control.Distributed.Process.Internal.Types+  ( LocalNode(..)+  , NCMsg(..)+  , ProcessId+  , Process+  , LocalProcess(..)+  , Identifier(..)+  , ProcessSignal(NamedSend)+  , Message+  , SendPort+  , forever'+  , nullProcessId+  , createUnencodedMessage+  )++import Control.Monad.IO.Class (liftIO)+import Control.Monad.Reader (ask)++import Data.Set (Set)+import qualified Data.Set as Set+import Data.Map (Map)+import qualified Data.Map as Map++import Data.Maybe (fromMaybe)+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 System.Environment (getEnv)+import System.IO+  ( Handle+  , IOMode(AppendMode)+  , BufferMode(..)+  , openFile+  , hClose+  , hPutStrLn+  , hSetBuffering+  )+import System.Locale (defaultTimeLocale)+import System.Mem.Weak+  ( Weak+  )++data TracerState =+  TracerST+  {+    client   :: !(Maybe ProcessId)+  , flags    :: !TraceFlags+  , regNames :: !(Map ProcessId (Set String))+  }++--------------------------------------------------------------------------------+-- Trace Handlers                                                             --+--------------------------------------------------------------------------------++defaultTracer :: Process ()+defaultTracer =+  catch (checkEnv "DISTRIBUTED_PROCESS_TRACE_FILE" >>= logfileTracer)+        (\(_ :: IOError) -> defaultTracerAux)++defaultTracerAux :: Process ()+defaultTracerAux =+  catch (checkEnv "DISTRIBUTED_PROCESS_TRACE_CONSOLE" >> systemLoggerTracer)+        (\(_ :: IOError) -> defaultEventLogTracer)++-- TODO: it would be /nice/ if we had some way of checking the runtime+-- options to see if +RTS -v (or similar) has been given...+defaultEventLogTracer :: Process ()+defaultEventLogTracer =+  catch (checkEnv "DISTRIBUTED_PROCESS_TRACE_EVENTLOG" >> eventLogTracer)+        (\(_ :: IOError) -> nullTracer)++checkEnv :: String -> Process String+checkEnv s = liftIO $ getEnv s++-- This trace client is (intentionally) a noop - it simply provides+-- an intial client for the trace controller to talk to, until some+-- other (hopefully more useful) client is installed over the top of+-- it. This is the default trace client.+nullTracer :: Process ()+nullTracer =+  forever' $ receiveWait [ matchUnknown (return ()) ]++systemLoggerTracer :: Process ()+systemLoggerTracer = do+  node <- processNode <$> ask+  let tr = sendTraceLog node+  forever' $ receiveWait [ matchAny (\m -> handleMessage m tr) ]+  where+    sendTraceLog :: LocalNode -> MxEvent -> Process ()+    sendTraceLog node ev = do+      now <- liftIO $ getCurrentTime+      msg <- return $ (formatTime defaultTimeLocale "%c" now, buildTxt ev)+      emptyPid <- return $ (nullProcessId (localNodeId node))+      traceMsg <- return $ NCMsg {+                             ctrlMsgSender = ProcessIdentifier (emptyPid)+                           , ctrlMsgSignal = (NamedSend "logger"+                                                 (createUnencodedMessage msg))+                           }+      liftIO $ writeChan (localCtrlChan node) traceMsg++    buildTxt :: MxEvent -> String+    buildTxt (MxLog msg) = msg+    buildTxt ev          = show ev++eventLogTracer :: Process ()+eventLogTracer =+  -- NB: when the GHC event log supports tracing arbitrary (ish) data, we will+  -- almost certainly use *that* facility independently of whether or not there+  -- is a tracer process installed. This is just a stop gap until then.+  forever' $ receiveWait [ matchAny (\m -> handleMessage m writeTrace) ]+  where+    writeTrace :: MxEvent -> Process ()+    writeTrace ev = liftIO $ traceEventIO (show ev)++logfileTracer :: FilePath -> Process ()+logfileTracer p = do+  -- TODO: error handling if the handle cannot be opened+  h <- liftIO $ openFile p AppendMode+  liftIO $ hSetBuffering h LineBuffering+  logger h `finally` (liftIO $ hClose h)+  where+    logger :: Handle -> Process ()+    logger h' = forever' $ do+      receiveWait [+          matchIf (\ev -> case ev of+                            MxTraceDisable      -> True+                            (MxTraceTakeover _) -> True+                            _                   -> False)+                  (\_ -> (liftIO $ hClose h') >> die "trace stopped")+        , matchAny (\ev -> handleMessage ev (writeTrace h'))+        ]++    writeTrace :: Handle -> MxEvent -> Process ()+    writeTrace h ev = do+      liftIO $ do+        now <- getCurrentTime+        hPutStrLn h $ (formatTime defaultTimeLocale "%c - " now) ++ (show ev)++--------------------------------------------------------------------------------+-- Tracer Implementation                                                      --+--------------------------------------------------------------------------------++traceController :: MVar ((Weak (CQueue Message))) -> Process ()+traceController mv = do+    -- See the documentation for mxAgentController for a+    -- commentary that explains this breach of encapsulation+    weakQueue <- processWeakQ <$> ask+    liftIO $ putMVar mv weakQueue+    initState <- initialState+    traceLoop initState { client = Nothing }+  where+    traceLoop :: TracerState -> Process ()+    traceLoop st = do+      let client' = client st+      -- Trace events are forwarded to the enabled trace target.+      -- At some point in the future, we're going to start writing these custom+      -- events to the ghc eventlog, at which point this design might change.+      st' <- receiveWait [+          match (\(setResp, set :: SetTrace) -> do+                  -- We consider at most one trace client, which is a process.+                  -- Tracking multiple clients represents too high an overhead,+                  -- so we leave that kind of thing to our consumers (e.g., the+                  -- high level Debug client module) to figure out.+                  case set of+                    (TraceEnable pid) -> do+                      -- notify the previous tracer it has been replaced+                      sendTraceMsg client' (createUnencodedMessage (MxTraceTakeover pid))+                      sendOk setResp+                      return st { client = (Just pid) }+                    TraceDisable -> do+                      sendTraceMsg client' (createUnencodedMessage MxTraceDisable)+                      sendOk setResp+                      return st { client = Nothing })+        , match (\(confResp, flags') ->+                  sendOk confResp >> applyTraceFlags flags' st)+        , match (\chGetFlags -> sendChan chGetFlags (flags st) >> return st)+        , match (\chGetCurrent -> sendChan chGetCurrent (client st) >> return st)+          -- we dequeue incoming events even if we don't process them+        , matchAny (\ev ->+            handleMessage ev (handleTrace st ev) >>= return . fromMaybe st)+        ]+      traceLoop st'++    sendOk :: Maybe (SendPort TraceOk) -> Process ()+    sendOk Nothing   = return ()+    sendOk (Just sp) = sendChan sp TraceOk++    initialState :: Process TracerState+    initialState = do+      flags' <- checkEnvFlags+      return $ TracerST { client   = Nothing+                        , flags    = flags'+                        , regNames = Map.empty+                        }++    checkEnvFlags :: Process TraceFlags+    checkEnvFlags =+      catch (checkEnv "DISTRIBUTED_PROCESS_TRACE_FLAGS" >>= return . parseFlags)+            (\(_ :: IOError) -> return defaultTraceFlags)++    parseFlags :: String -> TraceFlags+    parseFlags s = parseFlags' s defaultTraceFlags+      where parseFlags' :: String -> TraceFlags -> TraceFlags+            parseFlags' [] parsedFlags = parsedFlags+            parseFlags' (x:xs) parsedFlags+              | x == 'p'  = parseFlags' xs parsedFlags { traceSpawned = traceOn }+              | x == 'n'  = parseFlags' xs parsedFlags { traceRegistered = traceOn }+              | x == 'u'  = parseFlags' xs parsedFlags { traceUnregistered = traceOn }+              | x == 'd'  = parseFlags' xs parsedFlags { traceDied = traceOn }+              | x == 's'  = parseFlags' xs parsedFlags { traceSend = traceOn }+              | x == 'r'  = parseFlags' xs parsedFlags { traceRecv = traceOn }+              | x == 'l'  = parseFlags' xs parsedFlags { traceNodes = True }+              | otherwise = parseFlags' xs parsedFlags++applyTraceFlags :: TraceFlags -> TracerState -> Process TracerState+applyTraceFlags flags' state = return state { flags = flags' }++handleTrace :: TracerState -> Message -> MxEvent -> Process TracerState+handleTrace st msg ev@(MxRegistered p n) =+  let regNames' =+        Map.insertWith (\_ ns -> Set.insert n ns) p+                       (Set.singleton n)+                       (regNames st)+  in do+    traceEv ev msg (traceRegistered (flags st)) st+    return st { regNames = regNames' }+handleTrace st msg ev@(MxUnRegistered p n) =+  let f ns = case ns of+               Nothing  -> Nothing+               Just ns' -> Just (Set.delete n ns')+      regNames' = Map.alter f p (regNames st)+  in do+    traceEv ev msg (traceUnregistered (flags st)) st+    return st { regNames = regNames' }+handleTrace st msg ev@(MxSpawned  _)   = do+  traceEv ev msg (traceSpawned (flags st)) st >> return st+handleTrace st msg ev@(MxProcessDied _ _)     = do+  traceEv ev msg (traceDied (flags st)) st >> return st+handleTrace st msg ev@(MxSent _ _ _)   =+  traceEv ev msg (traceSend (flags st)) st >> return st+handleTrace st msg ev@(MxReceived _ _) =+  traceEv ev msg (traceRecv (flags st)) st >> return st+handleTrace st msg ev = do+  case ev of+    (MxNodeDied _ _) ->+      case (traceNodes (flags st)) of+        True  -> sendTrace st ev msg+        False -> return ()+    (MxUser _) -> sendTrace st ev msg+    (MxLog _)  -> sendTrace st ev msg+    _ ->+      case (traceConnections (flags st)) of+        True  -> sendTrace st ev msg+        False -> return ()+  return st++traceEv :: MxEvent+        -> Message+        -> Maybe TraceSubject+        -> TracerState+        -> Process ()+traceEv _  _   Nothing                  _  = return ()+traceEv ev msg (Just TraceAll)          st = sendTrace st ev msg+traceEv ev msg (Just (TraceProcs pids)) st = do+  node <- processNode <$> ask+  let p = case resolveToPid ev of+            Nothing  -> (nullProcessId (localNodeId node))+            Just pid -> pid+  case (Set.member p pids) of+    True  -> sendTrace st ev msg+    False -> return ()+traceEv ev msg (Just (TraceNames names)) st = do+  -- if we have recorded regnames for p, then we forward the trace iif+  -- there are overlapping trace targets+  node <- processNode <$> ask+  let p = case resolveToPid ev of+            Nothing  -> (nullProcessId (localNodeId node))+            Just pid -> pid+  case (Map.lookup p (regNames st)) of+    Nothing -> return ()+    Just ns -> if (Set.null (Set.intersection ns names))+                 then return ()+                 else sendTrace st ev msg++sendTrace :: TracerState -> MxEvent -> Message -> Process ()+sendTrace st ev msg = do+  let c = client st+  if c == (resolveToPid ev)  -- we do not send the tracer events about itself...+     then return ()+     else sendTraceMsg c msg++sendTraceMsg :: Maybe ProcessId -> Message -> Process ()+sendTraceMsg Nothing  _   = return ()+sendTraceMsg (Just p) msg = (flip forward) p msg

@@ -0,0 +1,169 @@+{-# LANGUAGE DeriveDataTypeable  #-}+{-# LANGUAGE ExistentialQuantification  #-}+{-# LANGUAGE FlexibleInstances  #-}+{-# LANGUAGE TypeSynonymInstances  #-}+{-# LANGUAGE GADTs  #-}+{-# LANGUAGE DeriveGeneric  #-}++-- | Tracing/Debugging support - Types+module Control.Distributed.Process.Management.Internal.Trace.Types+  ( SetTrace(..)+  , TraceSubject(..)+  , TraceFlags(..)+  , TraceArg(..)+  , TraceOk(..)+  , traceLog+  , traceLogFmt+  , traceEvent+  , traceMessage+  , defaultTraceFlags+  , enableTrace+  , enableTraceSync+  , disableTrace+  , disableTraceSync+  , getTraceFlags+  , setTraceFlags+  , setTraceFlagsSync+  , getCurrentTraceClient+  ) where++import Control.Distributed.Process.Internal.Types+  ( MxEventBus(..)+  , ProcessId+  , SendPort+  , unsafeCreateUnencodedMessage+  )+import Control.Distributed.Process.Management.Internal.Bus+  ( publishEvent+  )+import Control.Distributed.Process.Management.Internal.Types+  ( MxEvent(..)+  )+import Control.Distributed.Process.Serializable+import Data.Binary+import Data.List (intersperse)+import Data.Set (Set)+import qualified Data.Set as Set (fromList)+import Data.Typeable+import GHC.Generics++--------------------------------------------------------------------------------+-- Types                                                                      --+--------------------------------------------------------------------------------++data SetTrace = TraceEnable !ProcessId | TraceDisable+  deriving (Typeable, Generic, Eq, Show)+instance Binary SetTrace where++-- | Defines which processes will be traced by a given 'TraceFlag',+-- either by name, or @ProcessId@. Choosing @TraceAll@ is /by far/+-- the most efficient approach, as the tracer process therefore+-- avoids deciding whether or not a trace event is viable.+--+data TraceSubject =+    TraceAll                     -- enable tracing for all running processes+  | TraceProcs !(Set ProcessId)  -- enable tracing for a set of processes+  | TraceNames !(Set String)     -- enable tracing for a set of named/registered processes+  deriving (Typeable, Generic, Show)+instance Binary TraceSubject where++-- | Defines /what/ will be traced. Flags that control tracing of+-- @Process@ events, take a 'TraceSubject' controlling which processes+-- should generate trace events in the target process.+data TraceFlags = TraceFlags {+    traceSpawned      :: !(Maybe TraceSubject) -- filter process spawned tracing+  , traceDied         :: !(Maybe TraceSubject) -- filter process died tracing+  , traceRegistered   :: !(Maybe TraceSubject) -- filter process registration tracing+  , traceUnregistered :: !(Maybe TraceSubject) -- filter process un-registration+  , traceSend         :: !(Maybe TraceSubject) -- filter process/message tracing by sender+  , traceRecv         :: !(Maybe TraceSubject) -- filter process/message tracing by receiver+  , traceNodes        :: !Bool                 -- enable node status trace events+  , traceConnections  :: !Bool                 -- enable connection status trace events+  } deriving (Typeable, Generic, Show)+instance Binary TraceFlags where++defaultTraceFlags :: TraceFlags+defaultTraceFlags =+  TraceFlags {+    traceSpawned      = Nothing+  , traceDied         = Nothing+  , traceRegistered   = Nothing+  , traceUnregistered = Nothing+  , traceSend         = Nothing+  , traceRecv         = Nothing+  , traceNodes        = False+  , traceConnections  = False+  }++data TraceArg =+    TraceStr String+  | forall a. (Show a) => Trace a++-- | A generic 'ok' response from the trace coordinator.+data TraceOk = TraceOk+  deriving (Typeable, Generic)+instance Binary TraceOk where++--------------------------------------------------------------------------------+-- Internal/Common API                                                        --+--------------------------------------------------------------------------------++traceLog :: MxEventBus -> String -> IO ()+traceLog tr s = publishEvent tr (unsafeCreateUnencodedMessage $ MxLog s)++traceLogFmt :: MxEventBus+            -> String+            -> [TraceArg]+            -> IO ()+traceLogFmt t d ls =+  traceLog t $ concat (intersperse d (map toS ls))+  where toS :: TraceArg -> String+        toS (TraceStr s) = s+        toS (Trace    a) = show a++traceEvent :: MxEventBus -> MxEvent -> IO ()+traceEvent tr ev = publishEvent tr (unsafeCreateUnencodedMessage ev)++traceMessage :: Serializable m => MxEventBus -> m -> IO ()+traceMessage tr msg = traceEvent tr (MxUser (unsafeCreateUnencodedMessage msg))++enableTrace :: MxEventBus -> ProcessId -> IO ()+enableTrace t p =+  publishEvent t (unsafeCreateUnencodedMessage ((Nothing :: Maybe (SendPort TraceOk)),+                                                (TraceEnable p)))++enableTraceSync :: MxEventBus -> SendPort TraceOk -> ProcessId -> IO ()+enableTraceSync t s p =+  publishEvent t (unsafeCreateUnencodedMessage (Just s, TraceEnable p))++disableTrace :: MxEventBus -> IO ()+disableTrace t =+  publishEvent t (unsafeCreateUnencodedMessage ((Nothing :: Maybe (SendPort TraceOk)),+                                     TraceDisable))++disableTraceSync :: MxEventBus -> SendPort TraceOk -> IO ()+disableTraceSync t s =+  publishEvent t (unsafeCreateUnencodedMessage ((Just s), TraceDisable))++setTraceFlags :: MxEventBus -> TraceFlags -> IO ()+setTraceFlags t f =+  publishEvent t (unsafeCreateUnencodedMessage ((Nothing :: Maybe (SendPort TraceOk)), f))++setTraceFlagsSync :: MxEventBus -> SendPort TraceOk -> TraceFlags -> IO ()+setTraceFlagsSync t s f =+  publishEvent t (unsafeCreateUnencodedMessage ((Just s), f))++getTraceFlags :: MxEventBus -> SendPort TraceFlags -> IO ()+getTraceFlags t s = publishEvent t (unsafeCreateUnencodedMessage s)++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

@@ -0,0 +1,151 @@+{-# 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+  , Message+  , SendPort+  , DiedReason+  , NodeId+  )+import Control.Monad.IO.Class (MonadIO)+import qualified Control.Monad.State as ST+  ( MonadState+  , StateT+  )+import Data.Binary+import Data.Typeable (Typeable)+import GHC.Generics+import Network.Transport+  ( ConnectionId+  , EndPointAddress+  )++-- | This is the /default/ management event, fired for various internal+-- events around the NT connection and Process lifecycle. All published+-- events that conform to this type, are eligible for tracing - i.e.,+-- they will be delivered to the trace controller.+--+data MxEvent =+    MxSpawned          ProcessId+    -- ^ fired whenever a local process is spawned+  | MxRegistered       ProcessId    String+    -- ^ fired whenever a process/name is registered (locally)+  | MxUnRegistered     ProcessId    String+    -- ^ fired whenever a process/name is unregistered (locally)+  | MxProcessDied      ProcessId    DiedReason+    -- ^ fired whenever a process dies+  | MxNodeDied         NodeId       DiedReason+    -- ^ 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+  | MxReceived         ProcessId    Message+    -- ^ fired whenever a message is received by a local process+  | MxConnected        ConnectionId EndPointAddress+    -- ^ fired when a network-transport connection is first established+  | MxDisconnected     ConnectionId EndPointAddress+    -- ^ fired when a network-transport connection is broken/disconnected+  | MxUser             Message+    -- ^ a user defined trace event+  | MxLog              String+    -- ^ a /logging/ event - used for debugging purposes only+  | MxTraceTakeover    ProcessId+    -- ^ notifies a trace listener that all subsequent traces will be sent to /pid/+  | MxTraceDisable+    -- ^ notifies a trace listener that it has been disabled/removed+    deriving (Typeable, Generic, Show)++instance Binary MxEvent where++-- | The class of things that we might be able to resolve to+-- a @ProcessId@ (or not).+class Addressable a where+  resolveToPid :: a -> Maybe ProcessId++instance Addressable MxEvent where+  resolveToPid (MxSpawned     p)     = Just p+  resolveToPid (MxProcessDied p _)   = Just p+  resolveToPid (MxSent        _ p _) = Just p+  resolveToPid (MxReceived    p _)   = Just p+  resolveToPid _                     = Nothing++-- | Gross though it is, this synonym represents a function+-- used to forking new processes, which has to be passed as a HOF+-- when calling mxAgentController, since there's no other way to+-- avoid a circular dependency with Node.hs+type Fork = (Process () -> IO ProcessId)++-- | A newtype wrapper for an agent id (which is a string).+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+                      }++-- | Monad for management agents.+--+newtype MxAgent s a =+  MxAgent+  {+    unAgent :: ST.StateT (MxAgentState s) Process a+  } deriving ( Functor+             , Monad+             , MonadIO+             , 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++-- | Represents the actions a management agent can take+-- when evaluating an /event sink/.+--+data MxAction =+    MxAgentDeactivate !String+  | MxAgentPrioritise !ChannelSelector+  | MxAgentReady+  | MxAgentSkip++-- | Type of a management agent's event sink.+type MxSink s = Message -> MxAgent s (Maybe MxAction)

@@ -1,3 +1,11 @@+{-# LANGUAGE CPP  #-}+{-# LANGUAGE DeriveDataTypeable  #-}+{-# LANGUAGE ScopedTypeVariables  #-}+{-# LANGUAGE RankNTypes  #-}+{-# LANGUAGE BangPatterns  #-}+{-# LANGUAGE GeneralizedNewtypeDeriving  #-}+{-# LANGUAGE RecordWildCards #-}+ -- | Local nodes -- module Control.Distributed.Process.Node@@ -28,6 +36,7 @@   , filter   , partitionWithKey   , elems+  , size   , filterWithKey   , foldlWithKey   )@@ -40,7 +49,7 @@   , toList   ) import Data.Foldable (forM_)-import Data.Maybe (isJust, isNothing, catMaybes)+import Data.Maybe (isJust, fromJust, isNothing, catMaybes) import Data.Typeable (Typeable) import Control.Category ((>>>)) import Control.Applicative ((<$>))@@ -49,9 +58,15 @@ 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, SomeException, Exception, throwTo)-import qualified Control.Exception as Exception (catch)-import Control.Concurrent (forkIO)+import Control.Exception+  ( throwIO+  , AsyncException(ThreadKilled)+  , SomeException+  , Exception+  , throwTo+  )+import qualified Control.Exception as Exception (Handler(..), catches, finally)+import Control.Concurrent (forkIO, myThreadId) import Control.Distributed.Process.Internal.StrictMVar   ( newMVar   , withMVar@@ -60,8 +75,10 @@   , newEmptyMVar   , putMVar   , takeMVar+  , readMVar   ) import Control.Concurrent.Chan (newChan, writeChan, readChan)+import qualified Control.Concurrent.MVar as MVar (newEmptyMVar, takeMVar) import Control.Concurrent.STM   ( atomically   )@@ -99,7 +116,7 @@   , LocalProcessId(..)   , ProcessId(..)   , LocalNode(..)-  , Tracer(InactiveTracer)+  , MxEventBus(..)   , LocalNodeState(..)   , LocalProcess(..)   , LocalProcessState(..)@@ -110,6 +127,7 @@   , localPidCounter   , localPidUnique   , localProcessWithId+  , localProcesses   , localConnections   , forever'   , MonitorRef(..)@@ -126,34 +144,49 @@   , DidUnlinkPort(..)   , SpawnRef   , DidSpawn(..)-  , Message+  , Message(..)   , TypedChannel(..)   , Identifier(..)   , nodeOf   , ProcessInfo(..)   , ProcessInfoNone(..)+  , NodeStats(..)   , SendPortId(..)   , typedChannelWithId   , RegisterReply(..)   , WhereIsReply(..)-  , messageToPayload   , payloadToMessage-  , createMessage+  , createUnencodedMessage   , runLocalProcess   , firstNonReservedProcessId-  , ImplicitReconnect(WithImplicitReconnect, NoImplicitReconnect)+  , ImplicitReconnect(WithImplicitReconnect)   )-import Control.Distributed.Process.Internal.Trace+import Control.Distributed.Process.Management.Internal.Agent+  ( mxAgentController+  )+import qualified Control.Distributed.Process.Management.Internal.Table as Table+  ( mxTableCoordinator+  , startTableCoordinator+  )+import qualified Control.Distributed.Process.Management.Internal.Trace.Remote as Trace+  ( remoteTable+  )+import Control.Distributed.Process.Management.Internal.Trace.Tracer+  ( defaultTracer+  )+import Control.Distributed.Process.Management.Internal.Trace.Types   ( TraceArg(..)-  , traceFormat-  , startTracing-  , stopTracer+  , traceEvent+  , 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   )@@ -163,21 +196,24 @@   , receiveWait   , match   , sendChan+  , catch+  , unwrapMessage   )-import Control.Distributed.Process.Internal.Types (SendPort)+import Control.Distributed.Process.Internal.Types (SendPort, Tracer(..)) import qualified Control.Distributed.Process.Internal.Closure.BuiltIn as BuiltIn (remoteTable)-import Control.Distributed.Process.Internal.WeakTQueue (writeTQueue)+import Control.Distributed.Process.Internal.WeakTQueue (TQueue, writeTQueue) import qualified Control.Distributed.Process.Internal.StrictContainerAccessors as DAC   ( mapMaybe   , mapDefault   )+import Unsafe.Coerce  -------------------------------------------------------------------------------- -- Initialization                                                             -- --------------------------------------------------------------------------------  initRemoteTable :: RemoteTable-initRemoteTable = BuiltIn.remoteTable Static.initRemoteTable+initRemoteTable = Trace.remoteTable $ BuiltIn.remoteTable Static.initRemoteTable  -- | Initialize a new local node. newLocalNode :: NT.Transport -> RemoteTable -> IO LocalNode@@ -193,58 +229,105 @@ -- | Create a new local node (without any service processes running) createBareLocalNode :: NT.EndPoint -> RemoteTable -> IO LocalNode createBareLocalNode endPoint rtable = do-  unq <- randomIO-  state <- newMVar LocalNodeState-    { _localProcesses   = Map.empty-    , _localPidCounter  = firstNonReservedProcessId-    , _localPidUnique   = unq-    , _localConnections = Map.empty-    }-  ctrlChan <- newChan-  let node = LocalNode { localNodeId   = NodeId $ NT.address endPoint-                       , localEndPoint = endPoint-                       , localState    = state-                       , localCtrlChan = ctrlChan-                       , localTracer   = InactiveTracer-                       , remoteTable   = rtable-                       }-  tracedNode <- startTracing node-  void . forkIO $ runNodeController tracedNode-  void . forkIO $ handleIncomingMessages tracedNode-  return tracedNode+    unq <- randomIO+    state <- newMVar LocalNodeState+      { _localProcesses   = Map.empty+      , _localPidCounter  = firstNonReservedProcessId+      , _localPidUnique   = unq+      , _localConnections = Map.empty+      }+    ctrlChan <- newChan+    let node = LocalNode { localNodeId   = NodeId $ NT.address endPoint+                         , localEndPoint = endPoint+                         , localState    = state+                         , localCtrlChan = ctrlChan+                         , localEventBus = MxEventBusInitialising+                         , remoteTable   = rtable+                         }+    tracedNode <- startMxAgent node +    -- Once the NC terminates, the endpoint isn't much use,+    void $ forkIO $ Exception.finally (runNodeController tracedNode)+                                      (NT.closeEndPoint (localEndPoint node))++    -- whilst a closed/failing endpoint will terminate the NC+    void $ forkIO $ Exception.finally (handleIncomingMessages tracedNode)+                                      (stopNC node)++    return tracedNode+  where+    stopNC node =+       writeChan (localCtrlChan node) NCMsg+            { ctrlMsgSender = NodeIdentifier (localNodeId node)+            , ctrlMsgSignal = SigShutdown+            }++startMxAgent :: LocalNode -> IO LocalNode+startMxAgent node = do+  -- see note [tracer/forkProcess races]+  let fork = forkProcess node+  mv <- MVar.newEmptyMVar+  pid <- fork $ mxAgentController fork mv+  (tracer', wqRef, mxNew') <- MVar.takeMVar mv+  return node { localEventBus = (MxEventBus pid tracer' wqRef mxNew') }++startDefaultTracer :: LocalNode -> IO ()+startDefaultTracer node' = do+  let t = localEventBus node'+  case t of+    MxEventBus _ (Tracer pid _) _ _ -> do+      runProcess node' $ register "trace.controller" pid+      pid' <- forkProcess node' defaultTracer+      enableTrace (localEventBus node') pid'+      runProcess node' $ register "tracer.initial" pid'+    _ -> return ()++-- TODO: we need a better mechanism for defining and registering services+ -- | Start and register the service processes on a node--- (for now, this is only the logger) startServiceProcesses :: LocalNode -> IO () startServiceProcesses node = do+  -- tracing /spawns/ relies on the tracer being enabled, but we start+  -- the default tracer first, even though it might @nsend@ to the logger+  -- 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  where-  loop = do-    receiveWait-      [ match $ \((time, pid, string) ::(String, ProcessId, String)) -> do-          liftIO . hPutStrLn stderr $ time ++ " " ++ show pid ++ ": " ++ string-          loop-      , match $ \((time, string) :: (String, String)) -> do-          -- this is a 'trace' message from the local node tracer-          liftIO . hPutStrLn stderr $ time ++ " [trace] " ++ string-          loop-      , match $ \(ch :: SendPort ()) -> -- a shutdown request-          sendChan ch ()-      ]+   fork = forkProcess node +   loop = do+     receiveWait+       [ match $ \((time, pid, string) ::(String, ProcessId, String)) -> do+           liftIO . hPutStrLn stderr $ time ++ " " ++ show pid ++ ": " ++ string+           loop+       , match $ \((time, string) :: (String, String)) -> do+           -- this is a 'trace' message from the local node tracer+           liftIO . hPutStrLn stderr $ time ++ " [trace] " ++ string+           loop+       , match $ \(ch :: SendPort ()) -> -- a shutdown request+           sendChan ch ()+       ]+ -- | Force-close a local node -- -- TODO: for now we just close the associated endpoint closeLocalNode :: LocalNode -> IO () closeLocalNode node =+  -- TODO: close all our processes, surely!?   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-  void $ forkProcess node (proc `finally` liftIO (putMVar done ()))+  tid <- myThreadId+  void $ forkProcess node $ do+    catch (proc `finally` liftIO (putMVar done ()))+          (\(ex :: SomeException) -> liftIO $ throwTo tid ex)   takeMVar done  -- | Spawn a new process on a local node@@ -275,9 +358,15 @@                                  , processNode   = node                                  }         tid' <- forkIO $ do-          reason <- Exception.catch+          reason <- Exception.catches             (runLocalProcess lproc proc >> return DiedNormal)-            (return . DiedException . (show :: SomeException -> String))+            [ (Exception.Handler (\ex@(ProcessExitException from msg) -> do+                 mMsg <- unwrapMessage msg :: IO (Maybe String)+                 case mMsg of+                   Nothing -> return $ DiedException $ show ex+                   Just m  -> return $ DiedException ("exit-from=" ++ (show from) ++ ",reason=" ++ m)))+            , (Exception.Handler+                (return . DiedException . (show :: SomeException -> String)))]            -- [Unified: Table 4, rules termination and exiting]           modifyMVar_ (localState node) (cleanupProcess pid)@@ -287,8 +376,13 @@             }         return (tid', lproc) +      -- see note [tracer/forkProcess races]+      trace node (MxSpawned pid)+       if lpidCounter lpid == maxBound         then do+          -- 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)                  . (localPidCounter ^= firstNonReservedProcessId)@@ -312,6 +406,15 @@              . (localConnections ^= unaffected)              $ st +-- note [tracer/forkProcess races]+--+-- Our startTracing function uses forkProcess to start the trace controller+-- process, and of course forkProcess attempts to call traceEvent once the+-- process has started. This is harmless, as the localEventBus is not updated+-- until /after/ the initial forkProcess completes, so the first call to+-- traceEvent behaves as if tracing were disabled (i.e., it is ignored).+--+ -------------------------------------------------------------------------------- -- Handle incoming messages                                                   -- --------------------------------------------------------------------------------@@ -320,7 +423,7 @@  data IncomingTarget =     Uninit-  | ToProc (Weak (CQueue Message))+  | ToProc ProcessId (Weak (CQueue Message))   | ToChan TypedChannel   | ToNode @@ -355,20 +458,24 @@       case event of         NT.ConnectionOpened cid rel theirAddr ->           if rel == NT.ReliableOrdered-            then go ( (incomingAt cid ^= Just (theirAddr, Uninit))+            then+              trace node (MxConnected cid theirAddr)+              >> go (+                      (incomingAt cid ^= Just (theirAddr, Uninit))                     . (incomingFrom theirAddr ^: Set.insert cid)                     $ st                     )             else invalidRequest cid st         NT.Received cid payload ->           case st ^. incomingAt cid of-            Just (_, ToProc weakQueue) -> do+            Just (_, ToProc pid weakQueue) -> do               mQueue <- deRefWeak weakQueue               forM_ mQueue $ \queue -> do                 -- TODO: if we find that the queue is Nothing, should we remove                 -- it from the NC state? (and same for channels, below)                 let msg = payloadToMessage payload                 enqueue queue msg -- 'enqueue' is strict+                trace node (MxReceived pid msg)               go st             Just (_, ToChan (TypedChannel chan')) -> do               mChan <- deRefWeak chan'@@ -389,7 +496,7 @@                   mProc <- withMVar state $ return . (^. localProcessWithId lpid)                   case mProc of                     Just proc ->-                      go (incomingAt cid ^= Just (src, ToProc (processWeakQ proc)) $ st)+                      go (incomingAt cid ^= Just (src, ToProc pid (processWeakQ proc)) $ st)                     Nothing ->                       invalidRequest cid st                 SendPortIdentifier chId -> do@@ -416,7 +523,8 @@           case st ^. incomingAt cid of             Nothing ->               invalidRequest cid st-            Just (src, _) ->+            Just (src, _) -> do+              trace node (MxDisconnected cid src)               go ( (incomingAt cid ^= Nothing)                  . (incomingFrom src ^: Set.delete cid)                  $ st@@ -435,18 +543,15 @@              $ st              )         NT.ErrorEvent (NT.TransportError NT.EventEndPointFailed str) ->-          fatal $ "Cloud Haskell fatal error: end point failed: " ++ str+          fail $ "Cloud Haskell fatal error: end point failed: " ++ str         NT.ErrorEvent (NT.TransportError NT.EventTransportFailed str) ->-          fatal $ "Cloud Haskell fatal error: transport failed: " ++ str+          fail $ "Cloud Haskell fatal error: transport failed: " ++ str         NT.EndPointClosed ->-          stopTracer (localTracer node) >> return ()+          return ()         NT.ReceivedMulticast _ _ ->           -- If we received a multicast message, something went horribly wrong           -- and we just give up-          fatal "Cloud Haskell fatal error: received unexpected multicast"--    fatal :: String -> IO ()-    fatal msg = stopTracer (localTracer node) >> fail msg+          fail "Cloud Haskell fatal error: received unexpected multicast"      invalidRequest :: NT.ConnectionId -> ConnectionState -> IO ()     invalidRequest cid st = do@@ -454,7 +559,7 @@       -- 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: "),+      traceEventFmtIO node "" [(TraceStr " [network] invalid request: "),                                (Trace cid)]       go ( incomingAt cid ^= Nothing          $ st@@ -511,31 +616,30 @@ -- Tracing/Debugging                                                          -- -------------------------------------------------------------------------------- --- [Issue #104]+-- [Issue #104 / DP-13]  traceNotifyDied :: LocalNode -> Identifier -> DiedReason -> NC () traceNotifyDied node ident reason =-  case reason of-    DiedNormal -> return ()-    _ -> traceNcEventFmt node " "-                         [(TraceStr "[node-controller]"),-                          (Trace ident),-                          (Trace reason)]--traceNcEventFmt :: LocalNode -> String -> [TraceArg] -> NC ()-traceNcEventFmt node fmt args =-  liftIO $ traceEventFmtIO node fmt args+  -- TODO: sendPortDied notifications+  liftIO $ withLocalTracer node $ \t ->+    case ident of+      (NodeIdentifier nid)    -> traceEvent t (MxNodeDied nid reason)+      (ProcessIdentifier pid) -> traceEvent t (MxProcessDied pid reason)+      _                       -> return ()  traceEventFmtIO :: LocalNode                 -> String                 -> [TraceArg]                 -> IO () traceEventFmtIO node fmt args =-  withLocalTracer node $ \t -> traceFormat t fmt args+  withLocalTracer node $ \t -> traceLogFmt t fmt args -withLocalTracer :: LocalNode -> (Tracer -> IO ()) -> IO ()-withLocalTracer node act = act (localTracer node)+trace :: LocalNode -> MxEvent -> IO ()+trace node ev = withLocalTracer node $ \t -> traceEvent t ev +withLocalTracer :: LocalNode -> (MxEventBus -> IO ()) -> IO ()+withLocalTracer node act = act (localEventBus node)+ -------------------------------------------------------------------------------- -- Core functionality                                                         -- --------------------------------------------------------------------------------@@ -575,14 +679,25 @@         ncEffectRegister from label atnode pid force       NCMsg (ProcessIdentifier from) (WhereIs label) ->         ncEffectWhereIs from label-      NCMsg from (NamedSend label msg') ->-        ncEffectNamedSend from label msg'+      NCMsg _ (NamedSend label msg') ->+        ncEffectNamedSend label msg'+      NCMsg _ (LocalSend to msg') ->+        ncEffectLocalSend node to msg'+      NCMsg _ (LocalPortSend to msg') ->+        ncEffectLocalPortSend to msg'       NCMsg (ProcessIdentifier from) (Kill to reason) ->         ncEffectKill from to reason       NCMsg (ProcessIdentifier from) (Exit to reason) ->         ncEffectExit from to reason       NCMsg (ProcessIdentifier from) (GetInfo pid) ->         ncEffectGetInfo from pid+      NCMsg _ SigShutdown ->+        liftIO $ do+          NT.closeEndPoint (localEndPoint node)+            `Exception.finally` throwIO ThreadKilled+        -- ThreadKilled seems to make more sense than fail/error here+      NCMsg (ProcessIdentifier from) (GetNodeStats nid) ->+        ncEffectGetNodeStats from nid       unexpected ->         error $ "nodeController: unexpected message " ++ show unexpected @@ -692,6 +807,7 @@   mProc <- unClosure cProc   -- If the closure does not exist, we spawn a process that throws an exception   -- This allows the remote node to find out what's happening+  -- TODO:   let proc = case mProc of                Left err -> fail $ "Error: Could not resolve closure: " ++ err                Right p  -> p@@ -722,6 +838,9 @@      then do when (isOk) $                do modify' $ registeredHereFor label ^= mPid                   updateRemote node currentVal mPid+                  case mPid of+                    (Just p) -> liftIO $ trace node (MxRegistered p label)+                    Nothing  -> liftIO $ trace node (MxUnRegistered (fromJust currentVal) label)              liftIO $ sendMessage node                        (NodeIdentifier (localNodeId node))                        (ProcessIdentifier from)@@ -744,6 +863,7 @@             updateRemote _ _ _ = return ()             maybeify f Nothing = unmaybeify $ f []             maybeify f (Just x) = unmaybeify $ f x+             unmaybeify [] = Nothing             unmaybeify x = Just x             incList [] tag = [(tag,1)]@@ -777,18 +897,43 @@                        (WhereIsReply label mPid)  -- [Unified: Table 14]-ncEffectNamedSend :: Identifier -> String -> Message -> NC ()-ncEffectNamedSend from label msg = do-  node <- ask+ncEffectNamedSend :: String -> Message -> NC ()+ncEffectNamedSend label msg = do   mPid <- gets (^. registeredHereFor label)   -- If mPid is Nothing, we just ignore the named send (as per Table 14)-  forM_ mPid $ \pid ->-    liftIO $ sendPayload node-                         from-                         (ProcessIdentifier pid)-                         NoImplicitReconnect-                         (messageToPayload msg)+  forM_ mPid $ \pid -> postMessage pid msg +-- [Issue #DP-20]+ncEffectLocalSend :: LocalNode -> ProcessId -> Message -> NC ()+ncEffectLocalSend node to msg =+  liftIO $ withLocalProc node to $ \p -> do+    enqueue (processQueue p) msg+    trace node (MxReceived to msg)++-- [Issue #DP-20]+ncEffectLocalPortSend :: SendPortId -> Message -> NC ()+ncEffectLocalPortSend from msg = do+  node <- ask+  let pid = sendPortProcessId from+      cid = sendPortLocalId   from+  liftIO $ withLocalProc node pid $ \proc -> do+    mChan <- withMVar (processState proc) $ return . (^. typedChannelWithId cid)+    case mChan of+      -- in the unlikely event we know nothing about this channel id,+      -- there's little to be done - perhaps some logging/tracing though...+      Nothing -> return ()+      Just (TypedChannel chan') -> do+        -- 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' =+            atomically $ writeTQueue chan' ((unsafeCoerce raw) :: a)+        deliverChan (EncodedMessage   _ _) _ =+            -- this will not happen unless someone screws with Primitives.hs+            error "invalid local channel delivery"+ -- [Issue #69] ncEffectKill :: ProcessId -> ProcessId -> String -> NC () ncEffectKill from to reason = do@@ -850,6 +995,21 @@                                                  then (k:ks)                                                  else ks) [] +ncEffectGetNodeStats :: ProcessId -> NodeId -> NC ()+ncEffectGetNodeStats from _nid = do+  node <- ask+  ncState <- StateT.get+  nodeState <- liftIO $ readMVar (localState node)+  let localProcesses' = nodeState ^. localProcesses+      stats =+        NodeStats {+            nodeStatsNode = localNodeId node+          , nodeStatsRegisteredNames = Map.size $ ncState ^. registeredHere+          , nodeStatsMonitors = Map.size $ ncState ^. monitors+          , nodeStatsLinks = Map.size $ ncState ^. links+          , nodeStatsProcesses = Map.size localProcesses'+          }+  postAsMessage from stats -------------------------------------------------------------------------------- -- Auxiliary                                                                  -- --------------------------------------------------------------------------------@@ -887,21 +1047,25 @@  -- | [Unified: Table 8] destNid :: ProcessSignal -> Maybe NodeId-destNid (Link ident)        = Just $ nodeOf ident-destNid (Unlink ident)      = Just $ nodeOf ident-destNid (Monitor ref)       = Just $ nodeOf (monitorRefIdent ref)-destNid (Unmonitor ref)     = Just $ nodeOf (monitorRefIdent ref)-destNid (Spawn _ _)         = Nothing-destNid (Register _ _ _ _)  = Nothing-destNid (WhereIs _)         = Nothing-destNid (NamedSend _ _)     = Nothing+destNid (Link ident)          = Just $ nodeOf ident+destNid (Unlink ident)        = Just $ nodeOf ident+destNid (Monitor ref)         = Just $ nodeOf (monitorRefIdent ref)+destNid (Unmonitor ref)       = Just $ nodeOf (monitorRefIdent ref)+destNid (Spawn _ _)           = Nothing+destNid (Register _ _ _ _)    = Nothing+destNid (WhereIs _)           = Nothing+destNid (NamedSend _ _)       = 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-destNid (Died _ _)          = Nothing-destNid (Kill pid _)        = Just $ processNodeId pid-destNid (Exit pid _)        = Just $ processNodeId pid-destNid (GetInfo pid)       = Just $ processNodeId pid+destNid (Died _ _)            = Nothing+destNid (Kill pid _)          = Just $ processNodeId pid+destNid (Exit pid _)          = Just $ processNodeId pid+destNid (GetInfo pid)         = Just $ processNodeId pid+destNid (GetNodeStats nid)    = Just nid+destNid (LocalSend pid _)     = Just $ processNodeId pid+destNid (LocalPortSend cid _) = Just $ processNodeId (sendPortProcessId cid)+destNid (SigShutdown)       = Nothing  -- | Check if a process is local to our own node isLocal :: LocalNode -> Identifier -> Bool@@ -938,25 +1102,25 @@ --------------------------------------------------------------------------------  postAsMessage :: Serializable a => ProcessId -> a -> NC ()-postAsMessage pid = postMessage pid . createMessage+postAsMessage pid = postMessage pid . createUnencodedMessage  postMessage :: ProcessId -> Message -> NC () postMessage pid msg = do-  withLocalProc pid $ \p -> enqueue (processQueue p) msg+  node <- ask+  liftIO $ withLocalProc node pid $ \p -> enqueue (processQueue p) msg  throwException :: Exception e => ProcessId -> e -> NC ()-throwException pid e = withLocalProc pid $ \p ->-  throwTo (processThread p) e--withLocalProc :: ProcessId -> (LocalProcess -> IO ()) -> NC ()-withLocalProc pid p = do+throwException pid e = do   node <- ask-  liftIO $ do-    -- By [Unified: table 6, rule missing_process] messages to dead processes-    -- can silently be dropped-    let lpid = processLocalId pid-    mProc <- withMVar (localState node) $ return . (^. localProcessWithId lpid)-    forM_ mProc p+  liftIO $ withLocalProc node pid $ \p -> throwTo (processThread p) e++withLocalProc :: LocalNode -> ProcessId -> (LocalProcess -> IO ()) -> IO ()+withLocalProc node pid p =+  -- 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  -------------------------------------------------------------------------------- -- Accessors                                                                  --

@@ -1,3 +1,8 @@+{-# LANGUAGE DeriveDataTypeable  #-}+{-# LANGUAGE UndecidableInstances  #-}+{-# LANGUAGE FlexibleInstances  #-}+{-# LANGUAGE GADTs  #-}+{-# LANGUAGE CPP    #-} module Control.Distributed.Process.Serializable   ( Serializable   , encodeFingerprint@@ -10,8 +15,15 @@   ) where  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 Numeric (showHex) import Control.Exception (throw) import GHC.Fingerprint.Type (Fingerprint(..))

@@ -0,0 +1,121 @@+-----------------------------------------------------------------------------+-- |+-- Module      :  Control.Distributed.Process.UnsafePrimitives+-- Copyright   :  (c) Well-Typed / Tim Watson+-- License     :  BSD3 (see the file LICENSE)+--+-- Maintainer  :  Tim Watson <watson.timothy@gmail.com>+-- Stability   :  experimental+-- Portability :  non-portable (requires concurrency)+--+-- [Unsafe Variants of Cloud Haskell's Messaging Primitives]+--+-- Cloud Haskell's semantics do not mention evaluation/strictness properties+-- at all; Only the promise (or lack) of signal/message delivery between+-- processes is considered. For practical purposes, Cloud Haskell optimises+-- communication between (intra-node) local processes by skipping the+-- network-transport layer. In order to ensure the same strictness semantics+-- however, messages /still/ undergo binary serialisation before (internal)+-- transmission takes place. Thus we ensure that in both (the local and remote)+-- cases, message payloads are fully evaluated. Whilst this provides the user+-- with /unsurprising/ behaviour, the resulting performance overhead is quite+-- severe. Communicating data between Haskell threads /without/ forcing binary+-- serialisation reduces (intra-node, inter-process) communication overheads+-- by several orders of magnitude.+--+-- This module provides variants of Cloud Haskell's messaging primitives+-- ('send', 'sendChan', 'nsend' and 'wrapMessage') which do /not/ force binary+-- serialisation in the local case, thereby offering superior intra-node+-- messaging performance. The /catch/ is that any evaluation problems lurking+-- within the passed data structure (e.g., fields set to @undefined@ and so on)+-- will show up in the receiver rather than in the caller (as they would with+-- the /normal/ strategy).+--+-- Use of the functions in this module can potentially change the runtime+-- behaviour of your application. 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"+-- also re-exports these functions under different names, using the @unsafe@+-- prefix.+--+module Control.Distributed.Process.UnsafePrimitives+  ( -- * Unsafe Basic Messaging+    send+  , sendChan+  , nsend+  , wrapMessage+  ) where++import Control.Distributed.Process.Internal.Messaging+  ( sendMessage+  , sendBinary+  , sendCtrlMsg+  )++import Control.Distributed.Process.Internal.Types+  ( ProcessId(..)+  , LocalNode(..)+  , LocalProcess(..)+  , Process(..)+  , SendPort(..)+  , ProcessSignal(..)+  , Identifier(..)+  , ImplicitReconnect(..)+  , SendPortId(..)+  , Message+  , sendPortProcessId+  , unsafeCreateUnencodedMessage+  )+import Control.Distributed.Process.Serializable (Serializable)++import Control.Monad.IO.Class (MonadIO, liftIO)+import Control.Monad.Reader (ask)++-- | 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))++-- | 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')++-- | 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+  where+    unsafeSendChanLocal :: (Serializable a) => SendPortId -> a -> Process ()+    unsafeSendChanLocal spId msg' =+      sendCtrlMsg Nothing $ LocalPortSend spId (unsafeCreateUnencodedMessage msg')++-- | Create an unencoded @Message@ for any @Serializable@ type.+wrapMessage :: Serializable a => a -> Message+wrapMessage = unsafeCreateUnencodedMessage+

@@ -1,1184 +0,0 @@-module Main where--#if ! MIN_VERSION_base(4,6,0)-import Prelude hiding (catch)-#endif--import Data.Binary (Binary(..))-import Data.Typeable (Typeable)-import Data.Foldable (forM_)-import Control.Concurrent (forkIO, threadDelay, myThreadId, throwTo, ThreadId)-import Control.Concurrent.MVar-  ( MVar-  , newEmptyMVar-  , putMVar-  , takeMVar-  , readMVar-  )-import Control.Monad (replicateM_, replicateM, forever)-import Control.Exception (SomeException, throwIO)-import qualified Control.Exception as Ex (catch)-import Control.Applicative ((<$>), (<*>), pure, (<|>))-import qualified Network.Transport as NT (Transport, closeEndPoint)-import Network.Socket (sClose)-import Network.Transport.TCP-  ( createTransportExposeInternals-  , TransportInternals(socketBetween)-  , defaultTCPParameters-  )-import Control.Distributed.Process-import Control.Distributed.Process.Internal.Types-  ( NodeId(nodeAddress)-  , LocalNode(localEndPoint)-  , ProcessExitException(..)-  , nullProcessId-  )-import Control.Distributed.Process.Node-import Control.Distributed.Process.Serializable (Serializable)--import Test.HUnit (Assertion)-import Test.Framework (Test, defaultMain, testGroup)-import Test.Framework.Providers.HUnit (testCase)--newtype Ping = Ping ProcessId-  deriving (Typeable, Binary, Show)--newtype Pong = Pong ProcessId-  deriving (Typeable, Binary, Show)------------------------------------------------------------------------------------- Supporting definitions                                                     --------------------------------------------------------------------------------------- | Like fork, but throw exceptions in the child thread to the parent-forkTry :: IO () -> IO ThreadId-forkTry p = do-  tid <- myThreadId-  forkIO $ Ex.catch p (\e -> throwTo tid (e :: SomeException))---- | The ping server from the paper-ping :: Process ()-ping = do-  Pong partner <- expect-  self <- getSelfPid-  send partner (Ping self)-  ping---- | Quick and dirty synchronous version of whereisRemoteAsync-whereisRemote :: NodeId -> String -> Process (Maybe ProcessId)-whereisRemote nid string = do-  whereisRemoteAsync nid string-  WhereIsReply _ mPid <- expect-  return mPid--data Add       = Add    ProcessId Double Double deriving (Typeable)-data Divide    = Divide ProcessId Double Double deriving (Typeable)-data DivByZero = DivByZero deriving (Typeable)--instance Binary Add where-  put (Add pid x y) = put pid >> put x >> put y-  get = Add <$> get <*> get <*> get--instance Binary Divide where-  put (Divide pid x y) = put pid >> put x >> put y-  get = Divide <$> get <*> get <*> get--instance Binary DivByZero where-  put DivByZero = return ()-  get = return DivByZero---- The math server from the paper-math :: Process ()-math = do-  receiveWait-    [ match (\(Add pid x y) -> send pid (x + y))-    , matchIf (\(Divide _   _ y) -> y /= 0)-              (\(Divide pid x y) -> send pid (x / y))-    , match (\(Divide pid _ _) -> send pid DivByZero)-    ]-  math---- | Monitor or link to a remote node-monitorOrLink :: Bool            -- ^ 'True' for monitor, 'False' for link-              -> ProcessId       -- Process to monitor/link to-              -> Maybe (MVar ()) -- MVar to signal on once the monitor has been set up-              -> Process (Maybe MonitorRef)-monitorOrLink mOrL pid mSignal = do-  result <- if mOrL then Just <$> monitor pid-                    else link pid >> return Nothing-  -- Monitor is asynchronous, which usually does not matter but if we want a-  -- *specific* signal then it does. Therefore we wait an arbitrary delay and-  -- hope that this means the monitor has been set up-  forM_ mSignal $ \signal -> liftIO . forkIO $ threadDelay 100000 >> putMVar signal ()-  return result--monitorTestProcess :: ProcessId       -- Process to monitor/link to-                   -> Bool            -- 'True' for monitor, 'False' for link-                   -> Bool            -- Should we unmonitor?-                   -> DiedReason      -- Expected cause of death-                   -> Maybe (MVar ()) -- Signal for 'monitor set up'-                   -> MVar ()         -- Signal for successful termination-                   -> Process ()-monitorTestProcess theirAddr mOrL un reason monitorSetup done =-  catch (do mRef <- monitorOrLink mOrL theirAddr monitorSetup-            case (un, mRef) of-              (True, Nothing) -> do-                unlink theirAddr-                liftIO $ putMVar done ()-              (True, Just ref) -> do-                unmonitor ref-                liftIO $ putMVar done ()-              (False, ref) -> do-                ProcessMonitorNotification ref' pid reason' <- expect-                True <- return $ Just ref' == ref && pid == theirAddr && mOrL && reason == reason'-                liftIO $ putMVar done ()-        )-        (\(ProcessLinkException pid reason') -> do-            True <- return $ pid == theirAddr && not mOrL && not un && reason == reason'-            liftIO $ putMVar done ()-        )------------------------------------------------------------------------------------- The tests proper                                                           --------------------------------------------------------------------------------------- | Basic ping test-testPing :: NT.Transport -> Assertion-testPing transport = do-  serverAddr <- newEmptyMVar-  clientDone <- newEmptyMVar--  -- Server-  forkIO $ do-    localNode <- newLocalNode transport initRemoteTable-    addr <- forkProcess localNode ping-    putMVar serverAddr addr--  -- Client-  forkIO $ do-    localNode <- newLocalNode transport initRemoteTable-    pingServer <- readMVar serverAddr--    let numPings = 10000--    runProcess localNode $ do-      pid <- getSelfPid-      replicateM_ numPings $ do-        send pingServer (Pong pid)-        Ping _ <- expect-        return ()--    putMVar clientDone ()--  takeMVar clientDone---- | Monitor a process on an unreachable node-testMonitorUnreachable :: NT.Transport -> Bool -> Bool -> Assertion-testMonitorUnreachable transport mOrL un = do-  deadProcess <- newEmptyMVar-  done <- newEmptyMVar--  forkIO $ do-    localNode <- newLocalNode transport initRemoteTable-    addr <- forkProcess localNode . liftIO $ threadDelay 1000000-    closeLocalNode localNode-    putMVar deadProcess addr--  forkIO $ do-    localNode <- newLocalNode transport initRemoteTable-    theirAddr <- readMVar deadProcess-    runProcess localNode $-      monitorTestProcess theirAddr mOrL un DiedDisconnect Nothing done--  takeMVar done---- | Monitor a process which terminates normally-testMonitorNormalTermination :: NT.Transport -> Bool -> Bool -> Assertion-testMonitorNormalTermination transport mOrL un = do-  monitorSetup <- newEmptyMVar-  monitoredProcess <- newEmptyMVar-  done <- newEmptyMVar--  forkIO $ do-    localNode <- newLocalNode transport initRemoteTable-    addr <- forkProcess localNode $-      liftIO $ readMVar monitorSetup-    putMVar monitoredProcess addr--  forkIO $ do-    localNode <- newLocalNode transport initRemoteTable-    theirAddr <- readMVar monitoredProcess-    runProcess localNode $-      monitorTestProcess theirAddr mOrL un DiedNormal (Just monitorSetup) done--  takeMVar done---- | Monitor a process which terminates abnormally-testMonitorAbnormalTermination :: NT.Transport -> Bool -> Bool -> Assertion-testMonitorAbnormalTermination transport mOrL un = do-  monitorSetup <- newEmptyMVar-  monitoredProcess <- newEmptyMVar-  done <- newEmptyMVar--  let err = userError "Abnormal termination"--  forkIO $ do-    localNode <- newLocalNode transport initRemoteTable-    addr <- forkProcess localNode . liftIO $ do-      readMVar monitorSetup-      throwIO err-    putMVar monitoredProcess addr--  forkIO $ do-    localNode <- newLocalNode transport initRemoteTable-    theirAddr <- readMVar monitoredProcess-    runProcess localNode $-      monitorTestProcess theirAddr mOrL un (DiedException (show err)) (Just monitorSetup) done--  takeMVar done---- | Monitor a local process that is already dead-testMonitorLocalDeadProcess :: NT.Transport -> Bool -> Bool -> Assertion-testMonitorLocalDeadProcess transport mOrL un = do-  processDead <- newEmptyMVar-  processAddr <- newEmptyMVar-  localNode <- newLocalNode transport initRemoteTable-  done <- newEmptyMVar--  forkIO $ do-    addr <- forkProcess localNode . liftIO $ putMVar processDead ()-    putMVar processAddr addr--  forkIO $ do-    theirAddr <- readMVar processAddr-    readMVar processDead-    runProcess localNode $ do-      monitorTestProcess theirAddr mOrL un DiedUnknownId Nothing done--  takeMVar done---- | Monitor a remote process that is already dead-testMonitorRemoteDeadProcess :: NT.Transport -> Bool -> Bool -> Assertion-testMonitorRemoteDeadProcess transport mOrL un = do-  processDead <- newEmptyMVar-  processAddr <- newEmptyMVar-  done <- newEmptyMVar--  forkIO $ do-    localNode <- newLocalNode transport initRemoteTable-    addr <- forkProcess localNode . liftIO $ putMVar processDead ()-    putMVar processAddr addr--  forkIO $ do-    localNode <- newLocalNode transport initRemoteTable-    theirAddr <- readMVar processAddr-    readMVar processDead-    runProcess localNode $ do-      monitorTestProcess theirAddr mOrL un DiedUnknownId Nothing done--  takeMVar done---- | Monitor a process that becomes disconnected-testMonitorDisconnect :: NT.Transport -> Bool -> Bool -> Assertion-testMonitorDisconnect transport mOrL un = do-  processAddr <- newEmptyMVar-  monitorSetup <- newEmptyMVar-  done <- newEmptyMVar--  forkIO $ do-    localNode <- newLocalNode transport initRemoteTable-    addr <- forkProcess localNode . liftIO $ threadDelay 1000000-    putMVar processAddr addr-    readMVar monitorSetup-    NT.closeEndPoint (localEndPoint localNode)--  forkIO $ do-    localNode <- newLocalNode transport initRemoteTable-    theirAddr <- readMVar processAddr-    runProcess localNode $ do-      monitorTestProcess theirAddr mOrL un DiedDisconnect (Just monitorSetup) done--  takeMVar done---- | Test the math server (i.e., receiveWait)-testMath :: NT.Transport -> Assertion-testMath transport = do-  serverAddr <- newEmptyMVar-  clientDone <- newEmptyMVar--  -- Server-  forkIO $ do-    localNode <- newLocalNode transport initRemoteTable-    addr <- forkProcess localNode math-    putMVar serverAddr addr--  -- Client-  forkIO $ do-    localNode <- newLocalNode transport initRemoteTable-    mathServer <- readMVar serverAddr--    runProcess localNode $ do-      pid <- getSelfPid-      send mathServer (Add pid 1 2)-      3 <- expect :: Process Double-      send mathServer (Divide pid 8 2)-      4 <- expect :: Process Double-      send mathServer (Divide pid 8 0)-      DivByZero <- expect-      liftIO $ putMVar clientDone ()--  takeMVar clientDone---- | Send first message (i.e. connect) to an already terminated process--- (without monitoring); then send another message to a second process on--- the same remote node (we're checking that the remote node did not die)-testSendToTerminated :: NT.Transport -> Assertion-testSendToTerminated transport = do-  serverAddr1 <- newEmptyMVar-  serverAddr2 <- newEmptyMVar-  clientDone <- newEmptyMVar--  forkIO $ do-    terminated <- newEmptyMVar-    localNode <- newLocalNode transport initRemoteTable-    addr1 <- forkProcess localNode $ liftIO $ putMVar terminated ()-    addr2 <- forkProcess localNode $ ping-    readMVar terminated-    putMVar serverAddr1 addr1-    putMVar serverAddr2 addr2--  forkIO $ do-    localNode <- newLocalNode transport initRemoteTable-    server1 <- readMVar serverAddr1-    server2 <- readMVar serverAddr2-    runProcess localNode $ do-      pid <- getSelfPid-      send server1 "Hi"-      send server2 (Pong pid)-      Ping pid' <- expect-      True <- return $ pid' == server2-      liftIO $ putMVar clientDone ()--  takeMVar clientDone---- | Test (non-zero) timeout-testTimeout :: NT.Transport -> Assertion-testTimeout transport = do-  localNode <- newLocalNode transport initRemoteTable-  done <- newEmptyMVar--  runProcess localNode $ do-    Nothing <- receiveTimeout 1000000 [match (\(Add _ _ _) -> return ())]-    liftIO $ putMVar done ()--  takeMVar done---- | Test zero timeout-testTimeout0 :: NT.Transport -> Assertion-testTimeout0 transport = do-  serverAddr <- newEmptyMVar-  clientDone <- newEmptyMVar-  messagesSent <- newEmptyMVar--  forkIO $ do-    localNode <- newLocalNode transport initRemoteTable-    addr <- forkProcess localNode $ do-      liftIO $ readMVar messagesSent >> threadDelay 1000000-      -- Variation on the venerable ping server which uses a zero timeout-      -- Since we wait for all messages to be sent before doing this receive,-      -- we should nevertheless find the right message immediately-      Just partner <- receiveTimeout 0 [match (\(Pong partner) -> return partner)]-      self <- getSelfPid-      send partner (Ping self)-    putMVar serverAddr addr--  forkIO $ do-    localNode <- newLocalNode transport initRemoteTable-    server <- readMVar serverAddr-    runProcess localNode $ do-      pid <- getSelfPid-      -- Send a bunch of messages. A large number of messages that the server-      -- is not interested in, and then a single message that it wants-      replicateM_ 10000 $ send server "Irrelevant message"-      send server (Pong pid)-      liftIO $ putMVar messagesSent ()-      Ping _ <- expect-      liftIO $ putMVar clientDone ()--  takeMVar clientDone---- | Test typed channels-testTypedChannels :: NT.Transport -> Assertion-testTypedChannels transport = do-  serverChannel <- newEmptyMVar :: IO (MVar (SendPort (SendPort Bool, Int)))-  clientDone <- newEmptyMVar--  forkIO $ do-    localNode <- newLocalNode transport initRemoteTable-    forkProcess localNode $ do-      (serverSendPort, rport) <- newChan-      liftIO $ putMVar serverChannel serverSendPort-      (clientSendPort, i) <- receiveChan rport-      sendChan clientSendPort (even i)-    return ()--  forkIO $ do-    localNode <- newLocalNode transport initRemoteTable-    serverSendPort <- readMVar serverChannel-    runProcess localNode $ do-      (clientSendPort, rport) <- newChan-      sendChan serverSendPort (clientSendPort, 5)-      False <- receiveChan rport-      liftIO $ putMVar clientDone ()--  takeMVar clientDone---- | Test merging receive ports-testMergeChannels :: NT.Transport -> Assertion-testMergeChannels transport = do-    localNode <- newLocalNode transport initRemoteTable-    testFlat localNode True          "aaabbbccc"-    testFlat localNode False         "abcabcabc"-    testNested localNode True True   "aaabbbcccdddeeefffggghhhiii"-    testNested localNode True False  "adgadgadgbehbehbehcficficfi"-    testNested localNode False True  "abcabcabcdefdefdefghighighi"-    testNested localNode False False "adgbehcfiadgbehcfiadgbehcfi"-    testBlocked localNode True-    testBlocked localNode False-  where-    -- Single layer of merging-    testFlat :: LocalNode -> Bool -> String -> IO ()-    testFlat localNode biased expected = do-      done <- newEmptyMVar-      forkProcess localNode $ do-        rs  <- mapM charChannel "abc"-        m   <- mergePorts biased rs-        xs  <- replicateM 9 $ receiveChan m-        True <- return $ xs == expected-        liftIO $ putMVar done ()-      takeMVar done--    -- Two layers of merging-    testNested :: LocalNode -> Bool -> Bool -> String -> IO ()-    testNested localNode biasedInner biasedOuter expected = do-      done <- newEmptyMVar-      forkProcess localNode $ do-        rss  <- mapM (mapM charChannel) ["abc", "def", "ghi"]-        ms   <- mapM (mergePorts biasedInner) rss-        m    <- mergePorts biasedOuter ms-        xs   <- replicateM (9 * 3) $ receiveChan m-        True <- return $ xs == expected-        liftIO $ putMVar done ()-      takeMVar done--    -- Test that if no messages are (immediately) available, the scheduler makes no difference-    testBlocked :: LocalNode -> Bool -> IO ()-    testBlocked localNode biased = do-      vs <- replicateM 3 newEmptyMVar-      done <- newEmptyMVar--      forkProcess localNode $ do-        [sa, sb, sc] <- liftIO $ mapM readMVar vs-        mapM_ ((>> liftIO (threadDelay 10000)) . uncurry sendChan)-          [ -- a, b, c-            (sa, 'a')-          , (sb, 'b')-          , (sc, 'c')-            -- a, c, b-          , (sa, 'a')-          , (sc, 'c')-          , (sb, 'b')-            -- b, a, c-          , (sb, 'b')-          , (sa, 'a')-          , (sc, 'c')-            -- b, c, a-          , (sb, 'b')-          , (sc, 'c')-          , (sa, 'a')-            -- c, a, b-          , (sc, 'c')-          , (sa, 'a')-          , (sb, 'b')-            -- c, b, a-          , (sc, 'c')-          , (sb, 'b')-          , (sa, 'a')-          ]--      forkProcess localNode $ do-        (ss, rs) <- unzip <$> replicateM 3 newChan-        liftIO $ mapM_ (uncurry putMVar) $ zip vs ss-        m  <- mergePorts biased rs-        xs <- replicateM (6 * 3) $ receiveChan m-        True <- return $ xs == "abcacbbacbcacabcba"-        liftIO $ putMVar done ()--      takeMVar done--    mergePorts :: Serializable a => Bool -> [ReceivePort a] -> Process (ReceivePort a)-    mergePorts True  = mergePortsBiased-    mergePorts False = mergePortsRR--    charChannel :: Char -> Process (ReceivePort Char)-    charChannel c = do-      (sport, rport) <- newChan-      replicateM_ 3 $ sendChan sport c-      liftIO $ threadDelay 10000 -- Make sure messages have been sent-      return rport--testTerminate :: NT.Transport -> Assertion-testTerminate transport = do-  localNode <- newLocalNode transport initRemoteTable-  done <- newEmptyMVar--  pid <- forkProcess localNode $ do-    liftIO $ threadDelay 100000-    terminate--  runProcess localNode $ do-    ref <- monitor pid-    ProcessMonitorNotification ref' pid' (DiedException ex) <- expect-    True <- return $ ref == ref' && pid == pid' && ex == show ProcessTerminationException-    liftIO $ putMVar done ()--  takeMVar done--testMonitorNode :: NT.Transport -> Assertion-testMonitorNode transport = do-  [node1, node2] <- replicateM 2 $ newLocalNode transport initRemoteTable-  done <- newEmptyMVar--  closeLocalNode node1--  runProcess node2 $ do-    ref <- monitorNode (localNodeId node1)-    NodeMonitorNotification ref' nid DiedDisconnect <- expect-    True <- return $ ref == ref' && nid == localNodeId node1-    liftIO $ putMVar done ()--  takeMVar done--testMonitorChannel :: NT.Transport -> Assertion-testMonitorChannel transport = do-    [node1, node2] <- replicateM 2 $ newLocalNode transport initRemoteTable-    gotNotification <- newEmptyMVar--    pid <- forkProcess node1 $ do-      sport <- expect :: Process (SendPort ())-      ref <- monitorPort sport-      PortMonitorNotification ref' port' reason <- expect-      -- reason might be DiedUnknownId if the receive port is GCed before the-      -- monitor is established (TODO: not sure that this is reasonable)-      return $ ref' == ref && port' == sendPortId sport && (reason == DiedNormal || reason == DiedUnknownId)-      liftIO $ putMVar gotNotification ()--    runProcess node2 $ do-      (sport, _) <- newChan :: Process (SendPort (), ReceivePort ())-      send pid sport-      liftIO $ threadDelay 100000--    takeMVar gotNotification--testRegistry :: NT.Transport -> Assertion-testRegistry transport = do-  node <- newLocalNode transport initRemoteTable-  done <- newEmptyMVar--  pingServer <- forkProcess node ping--  runProcess node $ do-    register "ping" pingServer-    Just pid <- whereis "ping"-    True <- return $ pingServer == pid-    us <- getSelfPid-    nsend "ping" (Pong us)-    Ping pid' <- expect-    True <- return $ pingServer == pid'-    liftIO $ putMVar done ()--  takeMVar done--testRemoteRegistry :: NT.Transport -> Assertion-testRemoteRegistry transport = do-  node1 <- newLocalNode transport initRemoteTable-  node2 <- newLocalNode transport initRemoteTable-  done <- newEmptyMVar--  pingServer <- forkProcess node1 ping--  runProcess node2 $ do-    let nid1 = localNodeId node1-    registerRemoteAsync nid1 "ping" pingServer-    receiveWait [-       matchIf (\(RegisterReply label' _) -> "ping" == label')-               (\(RegisterReply _ _) -> return ()) ]--    Just pid <- whereisRemote nid1 "ping"-    True <- return $ pingServer == pid-    us <- getSelfPid-    nsendRemote nid1 "ping" (Pong us)-    Ping pid' <- expect-    True <- return $ pingServer == pid'-    liftIO $ putMVar done ()--  takeMVar done--testSpawnLocal :: NT.Transport -> Assertion-testSpawnLocal transport = do-  node <- newLocalNode transport initRemoteTable-  done <- newEmptyMVar--  runProcess node $ do-    us <- getSelfPid--    pid <- spawnLocal $ do-      sport <- expect-      sendChan sport (1234 :: Int)--    sport <- spawnChannelLocal $ \rport -> do-      (1234 :: Int) <- receiveChan rport-      send us ()--    send pid sport-    () <- expect-    liftIO $ putMVar done ()--  takeMVar done--testReconnect :: NT.Transport -> TransportInternals -> Assertion-testReconnect transport transportInternals = do-  [node1, node2] <- replicateM 2 $ newLocalNode transport initRemoteTable-  let nid1 = localNodeId node1-      nid2 = localNodeId node2-  processA <- newEmptyMVar-  [sendTestOk, registerTestOk] <- replicateM 2 newEmptyMVar--  forkProcess node1 $ do-    us <- getSelfPid-    liftIO $ putMVar processA us-    msg1 <- expect-    msg2 <- expect-    True <- return $ msg1 == "message 1" && msg2 == "message 3"-    liftIO $ putMVar sendTestOk ()--  forkProcess node2 $ do-    {--     - Make sure there is no implicit reconnect on normal message sending-     -}--    them <- liftIO $ readMVar processA-    send them "message 1" >> liftIO (threadDelay 100000)--    -- Simulate network failure-    liftIO $ do-      sock <- socketBetween transportInternals (nodeAddress nid1) (nodeAddress nid2)-      sClose sock-      threadDelay 10000--    -- Should not arrive-    send them "message 2"--    -- Should arrive-    reconnect them-    send them "message 3"--    liftIO $ takeMVar sendTestOk--    {--     - Test that there *is* implicit reconnect on node controller messages-     -}--    us <- getSelfPid-    registerRemoteAsync nid1 "a" us -- registerRemote is asynchronous-    receiveWait [-        matchIf (\(RegisterReply label' _) -> "a" == label')-                (\(RegisterReply _ _) -> return ()) ]--    Just _  <- whereisRemote nid1 "a"---    -- Simulate network failure-    liftIO $ do-      sock <- socketBetween transportInternals (nodeAddress nid1) (nodeAddress nid2)-      sClose sock-      threadDelay 10000--    -- This will happen due to implicit reconnect-    registerRemoteAsync nid1 "b" us-    receiveWait [-        matchIf (\(RegisterReply label' _) -> "b" == label')-                (\(RegisterReply _ _) -> return ()) ]--    -- Should happen-    registerRemoteAsync nid1 "c" us-    receiveWait [-        matchIf (\(RegisterReply label' _) -> "c" == label')-                (\(RegisterReply _ _) -> return ()) ]--    -- Check-    Nothing  <- whereisRemote nid1 "a"  -- this will fail because the name is removed when the node is disconnected-    Just _  <- whereisRemote nid1 "b"  -- this will suceed because the value is set after thereconnect-    Just _  <- whereisRemote nid1 "c"--    liftIO $ putMVar registerTestOk ()--  takeMVar registerTestOk---- | Test 'matchAny'. This repeats the 'testMath' but with a proxy server--- in between-testMatchAny :: NT.Transport -> Assertion-testMatchAny transport = do-  proxyAddr <- newEmptyMVar-  clientDone <- newEmptyMVar--  -- Math server-  forkIO $ do-    localNode <- newLocalNode transport initRemoteTable-    mathServer <- forkProcess localNode math-    proxyServer <- forkProcess localNode $ forever $ do-      msg <- receiveWait [ matchAny return ]-      forward msg mathServer-    putMVar proxyAddr proxyServer--  -- Client-  forkIO $ do-    localNode <- newLocalNode transport initRemoteTable-    mathServer <- readMVar proxyAddr--    runProcess localNode $ do-      pid <- getSelfPid-      send mathServer (Add pid 1 2)-      3 <- expect :: Process Double-      send mathServer (Divide pid 8 2)-      4 <- expect :: Process Double-      send mathServer (Divide pid 8 0)-      DivByZero <- expect-      liftIO $ putMVar clientDone ()--  takeMVar clientDone---- | Test 'matchAny'. This repeats the 'testMath' but with a proxy server--- in between, however we block 'Divide' requests ....-testMatchAnyHandle :: NT.Transport -> Assertion-testMatchAnyHandle transport = do-  proxyAddr <- newEmptyMVar-  clientDone <- newEmptyMVar--  -- Math server-  forkIO $ do-    localNode <- newLocalNode transport initRemoteTable-    mathServer <- forkProcess localNode math-    proxyServer <- forkProcess localNode $ forever $ do-        receiveWait [-            matchAny (maybeForward mathServer)-          ]-    putMVar proxyAddr proxyServer--  -- Client-  forkIO $ do-    localNode <- newLocalNode transport initRemoteTable-    mathServer <- readMVar proxyAddr--    runProcess localNode $ do-      pid <- getSelfPid-      send mathServer (Add pid 1 2)-      3 <- expect :: Process Double-      send mathServer (Divide pid 8 2)-      Nothing <- (expectTimeout 100000) :: Process (Maybe Double)-      liftIO $ putMVar clientDone ()--  takeMVar clientDone-  where maybeForward :: ProcessId -> AbstractMessage -> Process (Maybe ())-        maybeForward s msg =-            maybeHandleMessage msg (\m@(Add _ _ _) -> send s m)--testMatchAnyNoHandle :: NT.Transport -> Assertion-testMatchAnyNoHandle transport = do-  addr <- newEmptyMVar-  clientDone <- newEmptyMVar-  serverDone <- newEmptyMVar--  -- Math server-  forkIO $ do-    localNode <- newLocalNode transport initRemoteTable-    server <- forkProcess localNode $ forever $ do-        receiveWait [-          matchAnyIf-            -- the condition has type `Add -> Bool`-            (\(Add _ _ _) -> True)-            -- the match `AbstractMessage -> Process ()` will succeed!-            (\m -> do-              -- `String -> Process ()` does *not* match the input types however-              r <- (maybeHandleMessage m (\(_ :: String) -> die "NONSENSE" ))-              case r of-                Nothing -> return ()-                Just _  -> die "NONSENSE")-          ]-        -- we *must* have removed the message from our mailbox though!!!-        Nothing <- receiveTimeout 100000 [ match (\(Add _ _ _) -> return ()) ]-        liftIO $ putMVar serverDone ()-    putMVar addr server--  -- Client-  forkIO $ do-    localNode <- newLocalNode transport initRemoteTable-    server <- readMVar addr--    runProcess localNode $ do-      pid <- getSelfPid-      send server (Add pid 1 2)-      -- we only care about the client having sent a message, so we're done-      liftIO $ putMVar clientDone ()--  takeMVar clientDone-  takeMVar serverDone---- | Test 'matchAnyIf'. We provide an /echo/ server, but it ignores requests--- unless the text body @/= "bar"@ - this case should time out rather than--- removing the message from the process mailbox.-testMatchAnyIf :: NT.Transport -> Assertion-testMatchAnyIf transport = do-  echoAddr <- newEmptyMVar-  clientDone <- newEmptyMVar--  -- echo server-  forkIO $ do-    localNode <- newLocalNode transport initRemoteTable-    echoServer <- forkProcess localNode $ forever $ do-        receiveWait [-            matchAnyIf (\(_ :: ProcessId, (s :: String)) -> s /= "bar")-                       handleMessage-          ]-    putMVar echoAddr echoServer--  -- Client-  forkIO $ do-    localNode <- newLocalNode transport initRemoteTable-    server <- readMVar echoAddr--    runProcess localNode $ do-      pid <- getSelfPid-      send server (pid, "foo")-      "foo" <- expect-      send server (pid, "baz")-      "baz" <- expect-      send server (pid, "bar")-      Nothing <- (expectTimeout 100000) :: Process (Maybe Double)-      liftIO $ putMVar clientDone ()--  takeMVar clientDone-  where handleMessage :: AbstractMessage -> Process (Maybe ())-        handleMessage msg =-          maybeHandleMessage msg (\(pid :: ProcessId, (m :: String))-                                  -> do { send pid m; return () })---- Test 'receiveChanTimeout'-testReceiveChanTimeout :: NT.Transport -> Assertion-testReceiveChanTimeout transport = do-  done <- newEmptyMVar-  sendPort <- newEmptyMVar--  forkTry $ do-    localNode <- newLocalNode transport initRemoteTable-    runProcess localNode $ do-      -- Create a typed channel-      (sp, rp) <- newChan :: Process (SendPort Bool, ReceivePort Bool)-      liftIO $ putMVar sendPort sp--      -- Wait for a message with a delay. No message arrives, we should get Nothing after 1 second-      Nothing <- receiveChanTimeout 1000000 rp--      -- Wait for a message with a delay again. Now a message arrives after 0.5 seconds-      Just True <- receiveChanTimeout 1000000 rp--      -- Wait for a message with zero timeout: non-blocking check. No message is available, we get Nothing-      Nothing <- receiveChanTimeout 0 rp--      -- Again, but now there is a message available-      liftIO $ threadDelay 1000000-      Just False <- receiveChanTimeout 0 rp--      liftIO $ putMVar done ()--  forkTry $ do-    localNode <- newLocalNode transport initRemoteTable-    runProcess localNode $ do-      sp <- liftIO $ readMVar sendPort--      liftIO $ threadDelay 1500000-      sendChan sp True--      liftIO $ threadDelay 500000-      sendChan sp False--  takeMVar done---- | Test Functor, Applicative, Alternative and Monad instances for ReceiveChan-testReceiveChanFeatures :: NT.Transport -> Assertion-testReceiveChanFeatures transport = do-  done <- newEmptyMVar--  forkTry $ do-    localNode <- newLocalNode transport initRemoteTable-    runProcess localNode $ do-      (spInt,  rpInt)  <- newChan :: Process (SendPort Int, ReceivePort Int)-      (spBool, rpBool) <- newChan :: Process (SendPort Bool, ReceivePort Bool)--      -- Test Functor instance--      sendChan spInt 2-      sendChan spBool False--      rp1 <- mergePortsBiased [even <$> rpInt, rpBool]--      True <- receiveChan rp1-      False <- receiveChan rp1--      -- Test Applicative instance--      sendChan spInt 3-      sendChan spInt 4--      let rp2 = pure (+) <*> rpInt <*> rpInt--      7 <- receiveChan rp2--      -- Test Alternative instance--      sendChan spInt 3-      sendChan spBool True--      let rp3 = (even <$> rpInt) <|> rpBool--      False <- receiveChan rp3-      True <- receiveChan rp3--      -- Test Monad instance--      sendChan spBool True-      sendChan spBool False-      sendChan spInt 5--      let rp4 :: ReceivePort Int-          rp4 = do b <- rpBool-                   if b-                     then rpInt-                     else return 7--      5 <- receiveChan rp4-      7 <- receiveChan rp4--      liftIO $ putMVar done ()--  takeMVar done--testKillLocal :: NT.Transport -> Assertion-testKillLocal transport = do-  localNode <- newLocalNode transport initRemoteTable-  done <- newEmptyMVar--  pid <- forkProcess localNode $ do-    liftIO $ threadDelay 1000000--  runProcess localNode $ do-    ref <- monitor pid-    us <- getSelfPid-    kill pid "TestKill"-    ProcessMonitorNotification ref' pid' (DiedException ex) <- expect-    True <- return $ ref == ref' && pid == pid' && ex == "killed-by=" ++ show us ++ ",reason=TestKill"-    liftIO $ putMVar done ()--  takeMVar done--testKillRemote :: NT.Transport -> Assertion-testKillRemote transport = do-  node1 <- newLocalNode transport initRemoteTable-  node2 <- newLocalNode transport initRemoteTable-  done <- newEmptyMVar--  pid <- forkProcess node1 $ do-    liftIO $ threadDelay 1000000--  runProcess node2 $ do-    ref <- monitor pid-    us <- getSelfPid-    kill pid "TestKill"-    ProcessMonitorNotification ref' pid' (DiedException reason) <- expect-    True <- return $ ref == ref' && pid == pid' && reason == "killed-by=" ++ show us ++ ",reason=TestKill"-    liftIO $ putMVar done ()--  takeMVar done--testCatchesExit :: NT.Transport -> Assertion-testCatchesExit transport = do-  localNode <- newLocalNode transport initRemoteTable-  done <- newEmptyMVar--  _ <- forkProcess localNode $ do-      (die ("foobar", 123 :: Int))-      `catchesExit` [-           (\_ m -> maybeHandleMessage m (\(_ :: String) -> return ()))-         , (\_ m -> maybeHandleMessage m (\(_ :: Maybe Int) -> return ()))-         , (\_ m -> maybeHandleMessage m (\(_ :: String, _ :: Int)-                    -> (liftIO $ putMVar done ()) >> return ()))-         ]--  takeMVar done--testCatches :: NT.Transport -> Assertion-testCatches transport = do-  localNode <- newLocalNode transport initRemoteTable-  done <- newEmptyMVar--  _ <- forkProcess localNode $ do-    node <- getSelfNode-    (liftIO $ throwIO (ProcessLinkException (nullProcessId node) DiedNormal))-    `catches` [-        Handler (\(ProcessLinkException _ _) -> liftIO $ putMVar done ())-      ]--  takeMVar done--testDie :: NT.Transport -> Assertion-testDie transport = do-  localNode <- newLocalNode transport initRemoteTable-  done <- newEmptyMVar--  _ <- forkProcess localNode $ do-      (die ("foobar", 123 :: Int))-      `catchExit` \_from reason -> do-        -- TODO: should verify that 'from' has the right value-        True <- return $ reason == ("foobar", 123 :: Int)-        liftIO $ putMVar done ()--  takeMVar done--testPrettyExit :: NT.Transport -> Assertion-testPrettyExit transport = do-  localNode <- newLocalNode transport initRemoteTable-  done <- newEmptyMVar--  _ <- forkProcess localNode $ do-      (die "timeout")-      `catch` \ex@(ProcessExitException from _) ->-        let expected = "exit-from=" ++ (show from)-        in do-          True <- return $ (show ex) == expected-          liftIO $ putMVar done ()--  takeMVar done--testExitLocal :: NT.Transport -> Assertion-testExitLocal transport = do-  localNode <- newLocalNode transport initRemoteTable-  supervisedDone <- newEmptyMVar-  supervisorDone <- newEmptyMVar--  pid <- forkProcess localNode $ do-    (liftIO $ threadDelay 100000)-      `catchExit` \_from reason -> do-        -- TODO: should verify that 'from' has the right value-        True <- return $ reason == "TestExit"-        liftIO $ putMVar supervisedDone ()--  runProcess localNode $ do-    ref <- monitor pid-    exit pid "TestExit"-    -- This time the client catches the exception, so it dies normally-    ProcessMonitorNotification ref' pid' DiedNormal <- expect-    True <- return $ ref == ref' && pid == pid'-    liftIO $ putMVar supervisorDone ()--  takeMVar supervisedDone-  takeMVar supervisorDone--testExitRemote :: NT.Transport -> Assertion-testExitRemote transport = do-  node1 <- newLocalNode transport initRemoteTable-  node2 <- newLocalNode transport initRemoteTable-  supervisedDone <- newEmptyMVar-  supervisorDone <- newEmptyMVar--  pid <- forkProcess node1 $ do-    (liftIO $ threadDelay 100000)-      `catchExit` \_from reason -> do-        -- TODO: should verify that 'from' has the right value-        True <- return $ reason == "TestExit"-        liftIO $ putMVar supervisedDone ()--  runProcess node2 $ do-    ref <- monitor pid-    exit pid "TestExit"-    ProcessMonitorNotification ref' pid' DiedNormal <- expect-    True <- return $ ref == ref' && pid == pid'-    liftIO $ putMVar supervisorDone ()--  takeMVar supervisedDone-  takeMVar supervisorDone--tests :: (NT.Transport, TransportInternals)  -> [Test]-tests (transport, transportInternals) = [-    testGroup "Basic features" [-        testCase "Ping"                (testPing                transport)-      , testCase "Math"                (testMath                transport)-      , testCase "Timeout"             (testTimeout             transport)-      , testCase "Timeout0"            (testTimeout0            transport)-      , testCase "SendToTerminated"    (testSendToTerminated    transport)-      , testCase "TypedChannnels"      (testTypedChannels       transport)-      , testCase "MergeChannels"       (testMergeChannels       transport)-      , testCase "Terminate"           (testTerminate           transport)-      , testCase "Registry"            (testRegistry            transport)-      , testCase "RemoteRegistry"      (testRemoteRegistry      transport)-      , testCase "SpawnLocal"          (testSpawnLocal          transport)-      , testCase "MatchAny"            (testMatchAny            transport)-      , testCase "MatchAnyHandle"      (testMatchAnyHandle      transport)-      , testCase "MatchAnyNoHandle"    (testMatchAnyNoHandle    transport)-      , testCase "MatchAnyIf"          (testMatchAnyIf          transport)-      , testCase "ReceiveChanTimeout"  (testReceiveChanTimeout  transport)-      , testCase "ReceiveChanFeatures" (testReceiveChanFeatures transport)-      , testCase "KillLocal"           (testKillLocal           transport)-      , testCase "KillRemote"          (testKillRemote          transport)-      , testCase "Die"                 (testDie                 transport)-      , testCase "PrettyExit"          (testPrettyExit          transport)-      , testCase "CatchesExit"         (testCatchesExit         transport)-      , testCase "Catches"             (testCatches             transport)-      , testCase "ExitLocal"           (testExitLocal           transport)-      , testCase "ExitRemote"          (testExitRemote          transport)-      ]-  , testGroup "Monitoring and Linking" [-      -- Monitoring processes-      ---      -- The "missing" combinations in the list below don't make much sense, as-      -- we cannot guarantee that the monitor reply or link exception will not-      -- happen before the unmonitor or unlink-      testCase "MonitorUnreachable"           (testMonitorUnreachable         transport True  False)-    , testCase "MonitorNormalTermination"     (testMonitorNormalTermination   transport True  False)-    , testCase "MonitorAbnormalTermination"   (testMonitorAbnormalTermination transport True  False)-    , testCase "MonitorLocalDeadProcess"      (testMonitorLocalDeadProcess    transport True  False)-    , testCase "MonitorRemoteDeadProcess"     (testMonitorRemoteDeadProcess   transport True  False)-    , testCase "MonitorDisconnect"            (testMonitorDisconnect          transport True  False)-    , testCase "LinkUnreachable"              (testMonitorUnreachable         transport False False)-    , testCase "LinkNormalTermination"        (testMonitorNormalTermination   transport False False)-    , testCase "LinkAbnormalTermination"      (testMonitorAbnormalTermination transport False False)-    , testCase "LinkLocalDeadProcess"         (testMonitorLocalDeadProcess    transport False False)-    , testCase "LinkRemoteDeadProcess"        (testMonitorRemoteDeadProcess   transport False False)-    , testCase "LinkDisconnect"               (testMonitorDisconnect          transport False False)-    , testCase "UnmonitorNormalTermination"   (testMonitorNormalTermination   transport True  True)-    , testCase "UnmonitorAbnormalTermination" (testMonitorAbnormalTermination transport True  True)-    , testCase "UnmonitorDisconnect"          (testMonitorDisconnect          transport True  True)-    , testCase "UnlinkNormalTermination"      (testMonitorNormalTermination   transport False True)-    , testCase "UnlinkAbnormalTermination"    (testMonitorAbnormalTermination transport False True)-    , testCase "UnlinkDisconnect"             (testMonitorDisconnect          transport False True)-      -- Monitoring nodes and channels-    , testCase "MonitorNode"                  (testMonitorNode                transport)-    , testCase "MonitorChannel"               (testMonitorChannel             transport)-      -- Reconnect-    , testCase "Reconnect"                    (testReconnect                  transport transportInternals)-    ]-  ]--main :: IO ()-main = do-  Right transport <- createTransportExposeInternals "127.0.0.1" "8080" defaultTCPParameters-  defaultMain (tests transport)

@@ -1,480 +0,0 @@-{-# LANGUAGE TemplateHaskell #-}-module Main where--import Data.ByteString.Lazy (empty)-import Data.Typeable (Typeable)-import Control.Monad (join, replicateM, forever, replicateM_, void)-import Control.Exception (IOException, throw)-import Control.Concurrent (forkIO, threadDelay)-import Control.Concurrent.MVar-  ( MVar-  , newEmptyMVar-  , readMVar-  , takeMVar-  , putMVar-  , modifyMVar_-  , newMVar-  )-import Control.Applicative ((<$>))-import System.Random (randomIO)-import Network.Transport (Transport)-import Network.Transport.TCP-  ( createTransportExposeInternals-  , defaultTCPParameters-  , TransportInternals(socketBetween)-  )-import Network.Socket (sClose)-import Control.Distributed.Process-import Control.Distributed.Process.Closure-import Control.Distributed.Process.Node-import Control.Distributed.Process.Internal.Types (NodeId(nodeAddress))-import Control.Distributed.Static (staticLabel, staticClosure)--import Test.HUnit (Assertion)-import Test.Framework (Test, defaultMain)-import Test.Framework.Providers.HUnit (testCase)------------------------------------------------------------------------------------- Supporting definitions                                                     -------------------------------------------------------------------------------------quintuple :: a -> b -> c -> d -> e -> (a, b, c, d, e)-quintuple a b c d e = (a, b, c, d, e)--sdictInt :: SerializableDict Int-sdictInt = SerializableDict--factorial :: Int -> Process Int-factorial 0 = return 1-factorial n = (n *) <$> factorial (n - 1)--addInt :: Int -> Int -> Int-addInt x y = x + y--putInt :: Int -> MVar Int -> IO ()-putInt = flip putMVar--sendPid :: ProcessId -> Process ()-sendPid toPid = do-  fromPid <- getSelfPid-  send toPid fromPid--wait :: Int -> Process ()-wait = liftIO . threadDelay--isPrime :: Integer -> Process Bool-isPrime n = return . (n `elem`) . takeWhile (<= n) . sieve $ [2..]-  where-    sieve :: [Integer] -> [Integer]-    sieve (p : xs) = p : sieve [x | x <- xs, x `mod` p > 0]-    sieve [] = error "Uh oh -- we've run out of primes"---- | First argument indicates empty closure environment-typedPingServer :: () -> ReceivePort (SendPort ()) -> Process ()-typedPingServer () rport = forever $ do-  sport <- receiveChan rport-  sendChan sport ()--signal :: ProcessId -> Process ()-signal pid = send pid ()--remotable [ 'factorial-          , 'addInt-          , 'putInt-          , 'sendPid-          , 'sdictInt-          , 'wait-          , 'typedPingServer-          , 'isPrime-          , 'quintuple-          , 'signal-          ]--randomElement :: [a] -> IO a-randomElement xs = do-  ix <- randomIO-  return (xs !! (ix `mod` length xs))--remotableDecl [-    [d| dfib :: ([NodeId], SendPort Integer, Integer) -> Process () ;-        dfib (_, reply, 0) = sendChan reply 0-        dfib (_, reply, 1) = sendChan reply 1-        dfib (nids, reply, n) = do-          nid1 <- liftIO $ randomElement nids-          nid2 <- liftIO $ randomElement nids-          (sport, rport) <- newChan-          spawn nid1 $ $(mkClosure 'dfib) (nids, sport, n - 2)-          spawn nid2 $ $(mkClosure 'dfib) (nids, sport, n - 1)-          n1 <- receiveChan rport-          n2 <- receiveChan rport-          sendChan reply $ n1 + n2-      |]-  ]---- Just try creating a static polymorphic value-staticQuintuple :: (Typeable a, Typeable b, Typeable c, Typeable d, Typeable e)-                => Static (a -> b -> c -> d -> e -> (a, b, c, d, e))-staticQuintuple = $(mkStatic 'quintuple)--factorialClosure :: Int -> Closure (Process Int)-factorialClosure = $(mkClosure 'factorial)--addIntClosure :: Int -> Closure (Int -> Int)-addIntClosure = $(mkClosure 'addInt)--putIntClosure :: Int -> Closure (MVar Int -> IO ())-putIntClosure = $(mkClosure 'putInt)--sendPidClosure :: ProcessId -> Closure (Process ())-sendPidClosure = $(mkClosure 'sendPid)--sendFac :: Int -> ProcessId -> Closure (Process ())-sendFac n pid = factorialClosure n `bindCP` cpSend $(mkStatic 'sdictInt) pid--factorialOf :: Closure (Int -> Process Int)-factorialOf = staticClosure $(mkStatic 'factorial)--factorial' :: Int -> Closure (Process Int)-factorial' n = returnCP $(mkStatic 'sdictInt) n `bindCP` factorialOf--waitClosure :: Int -> Closure (Process ())-waitClosure = $(mkClosure 'wait)--simulateNetworkFailure :: TransportInternals -> NodeId -> NodeId -> Process ()-simulateNetworkFailure transportInternals fr to = liftIO $ do-  threadDelay 10000-  sock <- socketBetween transportInternals (nodeAddress fr) (nodeAddress to)-  sClose sock-  threadDelay 10000------------------------------------------------------------------------------------- The tests proper                                                           -------------------------------------------------------------------------------------testUnclosure :: Transport -> RemoteTable -> Assertion-testUnclosure transport rtable = do-  node <- newLocalNode transport rtable-  done <- newEmptyMVar-  forkProcess node $ do-    120 <- join . unClosure $ factorialClosure 5-    liftIO $ putMVar done ()-  takeMVar done--testBind :: Transport -> RemoteTable -> Assertion-testBind transport rtable = do-  node <- newLocalNode transport rtable-  done <- newEmptyMVar-  runProcess node $ do-    us <- getSelfPid-    join . unClosure $ sendFac 6 us-    (720 :: Int) <- expect-    liftIO $ putMVar done ()-  takeMVar done--testSendPureClosure :: Transport -> RemoteTable -> Assertion-testSendPureClosure transport rtable = do-  serverAddr <- newEmptyMVar-  serverDone <- newEmptyMVar--  forkIO $ do-    node <- newLocalNode transport rtable-    addr <- forkProcess node $ do-      cl <- expect-      fn <- unClosure cl :: Process (Int -> Int)-      13 <- return $ fn 6-      liftIO $ putMVar serverDone ()-    putMVar serverAddr addr--  forkIO $ do-    node <- newLocalNode transport rtable-    theirAddr <- readMVar serverAddr-    runProcess node $ send theirAddr (addIntClosure 7)--  takeMVar serverDone--testSendIOClosure :: Transport -> RemoteTable -> Assertion-testSendIOClosure transport rtable = do-  serverAddr <- newEmptyMVar-  serverDone <- newEmptyMVar--  forkIO $ do-    node <- newLocalNode transport rtable-    addr <- forkProcess node $ do-      cl <- expect-      io <- unClosure cl :: Process (MVar Int -> IO ())-      liftIO $ do-        someMVar <- newEmptyMVar-        io someMVar-        5 <- readMVar someMVar-        putMVar serverDone ()-    putMVar serverAddr addr--  forkIO $ do-    node <- newLocalNode transport rtable-    theirAddr <- readMVar serverAddr-    runProcess node $ send theirAddr (putIntClosure 5)--  takeMVar serverDone--testSendProcClosure :: Transport -> RemoteTable -> Assertion-testSendProcClosure transport rtable = do-  serverAddr <- newEmptyMVar-  clientDone <- newEmptyMVar--  forkIO $ do-    node <- newLocalNode transport rtable-    addr <- forkProcess node $ do-      cl <- expect-      pr <- unClosure cl :: Process (Int -> Process ())-      pr 5-    putMVar serverAddr addr--  forkIO $ do-    node <- newLocalNode transport rtable-    theirAddr <- readMVar serverAddr-    runProcess node $ do-      pid <- getSelfPid-      send theirAddr (cpSend $(mkStatic 'sdictInt) pid)-      5 <- expect :: Process Int-      liftIO $ putMVar clientDone ()--  takeMVar clientDone--testSpawn :: Transport -> RemoteTable -> Assertion-testSpawn transport rtable = do-  serverNodeAddr <- newEmptyMVar-  clientDone <- newEmptyMVar--  forkIO $ do-    node <- newLocalNode transport rtable-    putMVar serverNodeAddr (localNodeId node)--  forkIO $ do-    node <- newLocalNode transport rtable-    nid <- readMVar serverNodeAddr-    runProcess node $ do-      pid   <- getSelfPid-      pid'  <- spawn nid (sendPidClosure pid)-      pid'' <- expect-      True <- return $ pid' == pid''-      liftIO $ putMVar clientDone ()--  takeMVar clientDone--testCall :: Transport -> RemoteTable -> Assertion-testCall transport rtable = do-  serverNodeAddr <- newEmptyMVar-  clientDone <- newEmptyMVar--  forkIO $ do-    node <- newLocalNode transport rtable-    putMVar serverNodeAddr (localNodeId node)--  forkIO $ do-    node <- newLocalNode transport rtable-    nid <- readMVar serverNodeAddr-    runProcess node $ do-      (120 :: Int) <- call $(mkStatic 'sdictInt) nid (factorialClosure 5)-      liftIO $ putMVar clientDone ()--  takeMVar clientDone--testCallBind :: Transport -> RemoteTable -> Assertion-testCallBind transport rtable = do-  serverNodeAddr <- newEmptyMVar-  clientDone <- newEmptyMVar--  forkIO $ do-    node <- newLocalNode transport rtable-    putMVar serverNodeAddr (localNodeId node)--  forkIO $ do-    node <- newLocalNode transport rtable-    nid <- readMVar serverNodeAddr-    runProcess node $ do-      (120 :: Int) <- call $(mkStatic 'sdictInt) nid (factorial' 5)-      liftIO $ putMVar clientDone ()--  takeMVar clientDone--testSeq :: Transport -> RemoteTable -> Assertion-testSeq transport rtable = do-  node <- newLocalNode transport rtable-  done <- newEmptyMVar-  runProcess node $ do-    us <- getSelfPid-    join . unClosure $ sendFac 5 us `seqCP` sendFac 6 us-    120 :: Int <- expect-    720 :: Int <- expect-    liftIO $ putMVar done ()-  takeMVar done---- Test 'spawnSupervised'------ Set up a supervisor, spawn a child, then have a third process monitor the--- child. The supervisor then throws an exception, the child dies because it--- was linked to the supervisor, and the third process notices that the child--- dies.-testSpawnSupervised :: Transport -> RemoteTable -> Assertion-testSpawnSupervised transport rtable = do-    [node1, node2]       <- replicateM 2 $ newLocalNode transport rtable-    [superPid, childPid] <- replicateM 2 $ newEmptyMVar-    thirdProcessDone     <- newEmptyMVar--    forkProcess node1 $ do-      us <- getSelfPid-      liftIO $ putMVar superPid us-      (child, _ref) <- spawnSupervised (localNodeId node2) (waitClosure 1000000)-      liftIO $ do-        putMVar childPid child-        threadDelay 500000 -- Give the child a chance to link to us-        throw supervisorDeath--    forkProcess node2 $ do-      [super, child] <- liftIO $ mapM readMVar [superPid, childPid]-      ref <- monitor child-      ProcessMonitorNotification ref' pid' (DiedException e) <- expect-      True <- return $ ref' == ref-                    && pid' == child-                    && e == show (ProcessLinkException super (DiedException (show supervisorDeath)))-      liftIO $ putMVar thirdProcessDone ()--    takeMVar thirdProcessDone-  where-    supervisorDeath :: IOException-    supervisorDeath = userError "Supervisor died"--testSpawnInvalid :: Transport -> RemoteTable -> Assertion-testSpawnInvalid transport rtable = do-  node <- newLocalNode transport rtable-  done <- newEmptyMVar-  forkProcess node $ do-    (pid, ref) <- spawnMonitor (localNodeId node) (closure (staticLabel "ThisDoesNotExist") empty)-    ProcessMonitorNotification ref' pid' _reason <- expect-    -- Depending on the exact interleaving, reason might be NoProc or the exception thrown by the absence of the static closure-    True <- return $ ref' == ref && pid == pid'-    liftIO $ putMVar done ()-  takeMVar done--testClosureExpect :: Transport -> RemoteTable -> Assertion-testClosureExpect transport rtable = do-  node <- newLocalNode transport rtable-  done <- newEmptyMVar-  runProcess node $ do-    nodeId <- getSelfNode-    us     <- getSelfPid-    them   <- spawn nodeId $ cpExpect $(mkStatic 'sdictInt) `bindCP` cpSend $(mkStatic 'sdictInt) us-    send them (1234 :: Int)-    (1234 :: Int) <- expect-    liftIO $ putMVar done ()-  takeMVar done--testSpawnChannel :: Transport -> RemoteTable -> Assertion-testSpawnChannel transport rtable = do-  done <- newEmptyMVar-  [node1, node2] <- replicateM 2 $ newLocalNode transport rtable--  forkProcess node1 $ do-    pingServer <- spawnChannel-                    (sdictSendPort sdictUnit)-                    (localNodeId node2)-                    ($(mkClosure 'typedPingServer) ())-    (sendReply, receiveReply) <- newChan-    sendChan pingServer sendReply-    receiveChan receiveReply-    liftIO $ putMVar done ()--  takeMVar done--testTDict :: Transport -> RemoteTable -> Assertion-testTDict transport rtable = do-  done <- newEmptyMVar-  [node1, node2] <- replicateM 2 $ newLocalNode transport rtable-  forkProcess node1 $ do-    True <- call $(functionTDict 'isPrime) (localNodeId node2) ($(mkClosure 'isPrime) (79 :: Integer))-    liftIO $ putMVar done ()-  takeMVar done--testFib :: Transport -> RemoteTable -> Assertion-testFib transport rtable = do-  nodes <- replicateM 4 $ newLocalNode transport rtable-  done <- newEmptyMVar--  forkProcess (head nodes) $ do-    (sport, rport) <- newChan-    spawnLocal $ dfib (map localNodeId nodes, sport, 10)-    55 <- receiveChan rport :: Process Integer-    liftIO $ putMVar done ()--  takeMVar done--testSpawnReconnect :: Transport -> RemoteTable -> TransportInternals -> Assertion-testSpawnReconnect transport rtable transportInternals = do-  [node1, node2] <- replicateM 2 $ newLocalNode transport rtable-  let nid1 = localNodeId node1-      nid2 = localNodeId node2-  done <- newEmptyMVar-  iv <- newMVar (0 :: Int)--  incr <- forkProcess node1 $ forever $ do-    () <- expect-    liftIO $ modifyMVar_ iv (return . (+ 1))--  forkProcess node2 $ do-    _pid1 <- spawn nid1 ($(mkClosure 'signal) incr)-    simulateNetworkFailure transportInternals nid2 nid1-    _pid2 <- spawn nid1 ($(mkClosure 'signal) incr)-    _pid3 <- spawn nid1 ($(mkClosure 'signal) incr)--    liftIO $ threadDelay 100000--    count <- liftIO $ takeMVar iv-    True <- return $ count == 2 || count == 3 -- It depends on which message we get first in 'spawn'--    liftIO $ putMVar done ()--  takeMVar done---- | 'spawn' used to ave a race condition which would be triggered if the--- spawning process terminates immediately after spawning-testSpawnTerminate :: Transport -> RemoteTable -> Assertion-testSpawnTerminate transport rtable = do-  slave  <- newLocalNode transport rtable-  master <- newLocalNode transport rtable-  masterDone <- newEmptyMVar--  runProcess master $ do-    us <- getSelfPid-    replicateM_ 1000 . spawnLocal . void . spawn (localNodeId slave) $ $(mkClosure 'signal) us-    replicateM_ 1000 $ (expect :: Process ())-    liftIO $ putMVar masterDone ()--  takeMVar masterDone--tests :: (Transport, TransportInternals) -> RemoteTable -> [Test]-tests (transport, transportInternals) rtable = [-    testCase "Unclosure"       (testUnclosure       transport rtable)-  , testCase "Bind"            (testBind            transport rtable)-  , testCase "SendPureClosure" (testSendPureClosure transport rtable)-  , testCase "SendIOClosure"   (testSendIOClosure   transport rtable)-  , testCase "SendProcClosure" (testSendProcClosure transport rtable)-  , testCase "Spawn"           (testSpawn           transport rtable)-  , testCase "Call"            (testCall            transport rtable)-  , testCase "CallBind"        (testCallBind        transport rtable)-  , testCase "Seq"             (testSeq             transport rtable)-  , testCase "SpawnSupervised" (testSpawnSupervised transport rtable)-  , testCase "SpawnInvalid"    (testSpawnInvalid    transport rtable)-  , testCase "ClosureExpect"   (testClosureExpect   transport rtable)-  , testCase "SpawnChannel"    (testSpawnChannel    transport rtable)-  , testCase "TDict"           (testTDict           transport rtable)-  , testCase "Fib"             (testFib             transport rtable)-  , testCase "SpawnTerminate"  (testSpawnTerminate  transport rtable)-  , testCase "SpawnReconnect"  (testSpawnReconnect  transport rtable transportInternals)-  ]---main :: IO ()-main = do-  Right transport <- createTransportExposeInternals "127.0.0.1" "8080" defaultTCPParameters-  let rtable = __remoteTable . __remoteTableDecl $ initRemoteTable-  defaultMain (tests transport rtable)