distributed-process 0.4.1 → 0.4.2

raw patch · 23 files changed

+2108/−883 lines, 23 filesdep ~test-frameworkdep ~test-framework-hunitnew-uploaderPVP: major bump suggested

API removals or changes: PVP suggests a major version bump

Dependency ranges changed: test-framework, test-framework-hunit

API changes (from Hackage documentation)

Files

• benchmarks/Channels.hs +4/−4
• benchmarks/Latency.hs +4/−4
• benchmarks/Spawns.hs +4/−4
• benchmarks/Throughput.hs +7/−7
• distributed-process.cabal +41/−16
• src/Control/Distributed/Process.hs +89/−52
• src/Control/Distributed/Process/Closure.hs +41/−39
• src/Control/Distributed/Process/Internal/CQueue.hs +180/−67
• src/Control/Distributed/Process/Internal/Closure/BuiltIn.hs +41/−41
• src/Control/Distributed/Process/Internal/Closure/TH.hs +65/−56
• src/Control/Distributed/Process/Internal/Messaging.hs +42/−42
• src/Control/Distributed/Process/Internal/Primitives.hs +353/−111
• src/Control/Distributed/Process/Internal/StrictContainerAccessors.hs +1/−1
• src/Control/Distributed/Process/Internal/StrictList.hs +20/−18
• src/Control/Distributed/Process/Internal/StrictMVar.hs +4/−4
• src/Control/Distributed/Process/Internal/Trace.hs +122/−0
• src/Control/Distributed/Process/Internal/Types.hs +150/−81
• src/Control/Distributed/Process/Internal/WeakTQueue.hs +2/−2
• src/Control/Distributed/Process/Node.hs +300/−148
• src/Control/Distributed/Process/Serializable.hs +8/−8
• tests/TestCH.hs +402/−112
• tests/TestClosure.hs +66/−66
• tests/TestStats.hs +162/−0

@@ -23,11 +23,11 @@     receiveChan rc   liftIO . putStrLn $ "Did " ++ show n ++ " pings" -initialProcess :: String -> Process () +initialProcess :: String -> Process () initialProcess "SERVER" = do   us <- getSelfPid   liftIO $ BSL.writeFile "pingServer.pid" (encode us)-  pingServer +  pingServer initialProcess "CLIENT" = do   n <- liftIO $ getLine   them <- liftIO $ decode <$> BSL.readFile "pingServer.pid"@@ -37,5 +37,5 @@ main = do   [role, host, port] <- getArgs   Right transport <- createTransport host port defaultTCPParameters-  node <- newLocalNode transport initRemoteTable -  runProcess node $ initialProcess role +  node <- newLocalNode transport initRemoteTable+  runProcess node $ initialProcess role

@@ -18,11 +18,11 @@   replicateM_ n $ send them us >> (expect :: Process ())   liftIO . putStrLn $ "Did " ++ show n ++ " pings" -initialProcess :: String -> Process () +initialProcess :: String -> Process () initialProcess "SERVER" = do   us <- getSelfPid   liftIO $ BSL.writeFile "pingServer.pid" (encode us)-  pingServer +  pingServer initialProcess "CLIENT" = do   n <- liftIO $ getLine   them <- liftIO $ decode <$> BSL.readFile "pingServer.pid"@@ -32,5 +32,5 @@ main = do   [role, host, port] <- getArgs   Right transport <- createTransport host port defaultTCPParameters-  node <- newLocalNode transport initRemoteTable -  runProcess node $ initialProcess role +  node <- newLocalNode transport initRemoteTable+  runProcess node $ initialProcess role

@@ -12,7 +12,7 @@ counter :: Process () counter = go 0   where-    go :: Int -> Process () +    go :: Int -> Process ()     go !n = do       b <- expect       case b of@@ -27,7 +27,7 @@   n' <- expect   liftIO $ print (n == n') -initialProcess :: String -> Process () +initialProcess :: String -> Process () initialProcess "SERVER" = do   us <- getSelfPid   liftIO $ BSL.writeFile "counter.pid" (encode us)@@ -41,5 +41,5 @@ main = do   [role, host, port] <- getArgs   Right transport <- createTransport host port defaultTCPParameters-  node <- newLocalNode transport initRemoteTable -  runProcess node $ initialProcess role +  node <- newLocalNode transport initRemoteTable+  runProcess node $ initialProcess role

@@ -14,8 +14,8 @@  instance Binary a => Binary (SizedList a) where   put (SizedList sz xs) = put sz >> mapM_ put xs-  get = do -    sz <- get +  get = do+    sz <- get     xs <- getMany sz     return (SizedList sz xs) @@ -37,9 +37,9 @@ counter :: Process () counter = go 0   where-    go :: Int -> Process () +    go :: Int -> Process ()     go !n =-      receiveWait +      receiveWait         [ match $ \xs   -> go (n + size (xs :: SizedList Int))         , match $ \them -> send them n >> go 0         ]@@ -52,7 +52,7 @@   n' <- expect   liftIO $ print (packets * sz, n' == packets * sz) -initialProcess :: String -> Process () +initialProcess :: String -> Process () initialProcess "SERVER" = do   us <- getSelfPid   liftIO $ BSL.writeFile "counter.pid" (encode us)@@ -66,5 +66,5 @@ main = do   [role, host, port] <- getArgs   Right transport <- createTransport host port defaultTCPParameters-  node <- newLocalNode transport initRemoteTable -  runProcess node $ initialProcess role +  node <- newLocalNode transport initRemoteTable+  runProcess node $ initialProcess role

@@ -1,28 +1,28 @@-Name:          distributed-process -Version:       0.4.1+Name:          distributed-process+Version:       0.4.2 Cabal-Version: >=1.8 Build-Type:    Simple-License:       BSD3 +License:       BSD3 License-File:  LICENSE Copyright:     Well-Typed LLP Author:        Duncan Coutts, Nicolas Wu, Edsko de Vries-Maintainer:    edsko@well-typed.com, duncan@well-typed.com+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:   mailto:edsko@well-typed.com-Synopsis:      Cloud Haskell: Erlang-style concurrency in Haskell +Bug-Reports:   http://github.com/haskell-distributed/distributed-process/issues+Synopsis:      Cloud Haskell: Erlang-style concurrency in Haskell Description:   This is an implementation of Cloud Haskell, as described in                /Towards Haskell in the Cloud/ by Jeff Epstein, Andrew Black,                and Simon Peyton Jones                (<http://research.microsoft.com/en-us/um/people/simonpj/papers/parallel/>),                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.+               by Hans Svensson, Lars-Åke Fredlund and Clara Benac Earle.                 You will probably also want to install a Cloud Haskell backend such                as distributed-process-simplelocalnet. Tested-With:   GHC==7.2.2 GHC==7.4.1 GHC==7.4.2-Category:      Control +Category:      Control  Source-Repository head   Type:     git@@ -61,6 +61,7 @@                      Control.Distributed.Process.Internal.Primitives,                      Control.Distributed.Process.Internal.CQueue,                      Control.Distributed.Process.Internal.Types,+                     Control.Distributed.Process.Internal.Trace,                      Control.Distributed.Process.Internal.Closure.BuiltIn,                      Control.Distributed.Process.Internal.Messaging,                      Control.Distributed.Process.Internal.StrictList,@@ -96,14 +97,14 @@                      binary >= 0.5 && < 0.7,                      network >= 2.3 && < 2.5,                      HUnit >= 1.2 && < 1.3,-                     test-framework >= 0.6 && < 0.7,-                     test-framework-hunit >= 0.2 && < 0.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 +  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@@ -118,12 +119,36 @@                      bytestring >= 0.9 && < 0.11,                      network >= 2.3 && < 2.5,                      HUnit >= 1.2 && < 1.3,-                     test-framework >= 0.6 && < 0.7,-                     test-framework-hunit >= 0.2 && < 0.3-  Extensions:        CPP, +                     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 +  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 +  Executable distributed-process-throughput    if flag(benchmarks)

@@ -1,14 +1,19 @@--- | [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/>),--- 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.-module Control.Distributed.Process -  ( -- * Basic types +{- | [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/>),+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.++For a detailed description of the package and other reference materials,+please see the distributed-process wiki page on github:+<https://github.com/haskell-distributed/distributed-process/wiki>.++-}+module Control.Distributed.Process+  ( -- * Basic types     ProcessId   , NodeId   , Process@@ -17,7 +22,7 @@   , sendPortProcessId   , liftIO -- Reexported for convenience     -- * Basic messaging-  , send +  , send   , expect   , expectTimeout     -- * Channels@@ -37,17 +42,26 @@   , match   , matchIf   , matchUnknown-  , AbstractMessage(..) +  , AbstractMessage(..)   , matchAny+  , matchAnyIf+  , matchChan     -- * Process management   , spawn   , call   , terminate+  , die+  , kill+  , exit+  , catchExit+  , catchesExit   , ProcessTerminationException(..)   , ProcessRegistrationException(..)   , SpawnRef   , getSelfPid   , getSelfNode+  , ProcessInfo(..)+  , getProcessInfo     -- * Monitoring and linking   , link   , linkNode@@ -59,10 +73,11 @@   , monitorNode   , monitorPort   , unmonitor+  , withMonitor+  , MonitorRef -- opaque   , ProcessLinkException(..)   , NodeLinkException(..)   , PortLinkException(..)-  , MonitorRef -- opaque   , ProcessMonitorNotification(..)   , NodeMonitorNotification(..)   , PortMonitorNotification(..)@@ -88,8 +103,12 @@   , whereisRemoteAsync   , nsendRemote   , WhereIsReply(..)+  , RegisterReply(..)     -- * Exception handling   , catch+  , Handler(..)+  , catches+  , try   , mask   , onException   , bracket@@ -102,7 +121,7 @@   , spawnMonitor   , spawnChannel   , DidSpawn(..)-    -- * Local versions of 'spawn' +    -- * Local versions of 'spawn'   , spawnLocal   , spawnChannelLocal     -- * Reconnecting@@ -119,7 +138,7 @@ import Control.Applicative ((<$>)) import Control.Monad.Reader (ask) import Control.Concurrent.MVar (newEmptyMVar, takeMVar, putMVar)-import Control.Distributed.Static +import Control.Distributed.Static   ( Closure   , closure   , Static@@ -127,7 +146,7 @@   , closureCompose   , staticClosure   )-import Control.Distributed.Process.Internal.Types +import Control.Distributed.Process.Internal.Types   ( NodeId(..)   , ProcessId(..)   , Process(..)@@ -146,6 +165,7 @@   , ReceivePort(..)   , SendPortId(..)   , WhereIsReply(..)+  , RegisterReply(..)   , LocalProcess(processNode)   , nullProcessId   )@@ -154,8 +174,8 @@   ( sdictSendPort   , sndStatic   , idCP-  , seqCP -  , bindCP +  , seqCP+  , bindCP   , splitCP   , cpLink   , cpSend@@ -164,7 +184,7 @@   ) import Control.Distributed.Process.Internal.Primitives   ( -- Basic messaging-    send +    send   , expect     -- Channels   , newChan@@ -179,13 +199,22 @@   , match   , matchIf   , matchUnknown-  , AbstractMessage(..) +  , AbstractMessage(..)   , matchAny+  , matchAnyIf+  , matchChan     -- Process management   , terminate   , ProcessTerminationException(..)+  , die+  , exit+  , catchExit+  , catchesExit+  , kill   , getSelfPid   , getSelfNode+  , ProcessInfo(..)+  , getProcessInfo     -- Monitoring and linking   , link   , linkNode@@ -197,6 +226,7 @@   , monitorNode   , monitorPort   , unmonitor+  , withMonitor     -- Logging   , say     -- Registry@@ -215,6 +245,9 @@   , unClosure     -- Exception handling   , catch+  , Handler(..)+  , catches+  , try   , mask   , onException   , bracket@@ -231,7 +264,7 @@ import Control.Distributed.Process.Node (forkProcess)  -- INTERNAL NOTES--- +-- -- 1.  'send' never fails. If you want to know that the remote process received --     your message, you will need to send an explicit acknowledgement. If you --     want to know when the remote process failed, you will need to monitor@@ -269,17 +302,17 @@ --       http://research.microsoft.com/en-us/um/people/simonpj/papers/parallel/remote.pdf -- -- The precise semantics for message passing is based on--- +-- -- [2] "A Unified Semantics for Future Erlang", Hans Svensson, Lars-Ake Fredlund --     and Clara Benac Earle (not freely available online, unfortunately) -- -- Some pointers to related documentation about Erlang, for comparison and--- inspiration: +-- inspiration: -- -- [3] "Programming Distributed Erlang Applications: Pitfalls and Recipes",---     Hans Svensson and Lars-Ake Fredlund +--     Hans Svensson and Lars-Ake Fredlund --       http://man.lupaworld.com/content/develop/p37-svensson.pdf--- [4] The Erlang manual, sections "Message Sending" and "Send" +-- [4] The Erlang manual, sections "Message Sending" and "Send" --       http://www.erlang.org/doc/reference_manual/processes.html#id82409 --       http://www.erlang.org/doc/reference_manual/expressions.html#send -- [5] Questions "Is the order of message reception guaranteed?" and@@ -297,9 +330,9 @@ -- | Spawn a process -- -- For more information about 'Closure', see--- "Control.Distributed.Process.Closure". +-- "Control.Distributed.Process.Closure". ----- See also 'call'. +-- See also 'call'. spawn :: NodeId -> Closure (Process ()) -> Process ProcessId spawn nid proc = do   us   <- getSelfPid@@ -316,7 +349,7 @@  -- | Spawn a process and link to it ----- Note that this is just the sequential composition of 'spawn' and 'link'. +-- 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@@ -333,16 +366,20 @@   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". +-- "Control.Distributed.Process.Closure". ----- See also 'spawn'. -call :: Serializable a => Static (SerializableDict a) -> NodeId -> Closure (Process a) -> Process a-call dict nid proc = do +-- 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@@ -355,25 +392,25 @@     ]   case mResult of     Right a  -> do-      -- Wait for the monitor message so that we the mailbox doesn't grow -      receiveWait +      -- 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 +    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 ()) +spawnSupervised :: NodeId+                -> Closure (Process ())                 -> Process (ProcessId, MonitorRef) spawnSupervised nid proc = do   us   <- getSelfPid-  them <- spawn nid (cpLink us `seqCP` proc) +  them <- spawn nid (cpLink us `seqCP` proc)   ref  <- monitor them   return (them, ref) @@ -381,19 +418,19 @@ -- the corresponding 'SendPort'. spawnChannel :: forall a. Typeable a => Static (SerializableDict a)              -> NodeId-             -> Closure (ReceivePort a -> Process ()) +             -> Closure (ReceivePort a -> Process ())              -> Process (SendPort a) spawnChannel dict nid proc = do     us <- getSelfPid-    spawn nid (go us) +    _ <- spawn nid (go us)     expect   where     go :: ProcessId -> Closure (Process ())-    go pid = cpNewChan dict -           `bindCP` +    go pid = cpNewChan dict+           `bindCP`              (cpSend (sdictSendPort dict) pid `splitCP` proc)            `bindCP`-             (idCP `closureCompose` staticClosure sndStatic) +             (idCP `closureCompose` staticClosure sndStatic)  -------------------------------------------------------------------------------- -- Local versions of spawn                                                    --@@ -402,22 +439,22 @@ -- | Spawn a process on the local node spawnLocal :: Process () -> Process ProcessId spawnLocal proc = do-  node <- processNode <$> ask +  node <- processNode <$> ask   liftIO $ forkProcess node proc  -- | Create a new typed channel, spawn a process on the local node, passing it -- the receive port, and return the send port spawnChannelLocal :: Serializable a-                  => (ReceivePort a -> Process ()) +                  => (ReceivePort a -> Process ())                   -> Process (SendPort a)-spawnChannelLocal proc = do -  node <- processNode <$> ask +spawnChannelLocal proc = do+  node <- processNode <$> ask   liftIO $ do     mvar <- newEmptyMVar-    forkProcess node $ do+    _ <- forkProcess node $ do       -- It is important that we allocate the new channel in the new process,       -- because otherwise it will be associated with the wrong process ID       (sport, rport) <- newChan       liftIO $ putMVar mvar sport       proc rport-    takeMVar mvar +    takeMVar mvar

@@ -15,15 +15,15 @@ -- > f = ... -- -- you can use a Template Haskell splice to create a static version of 'f':--- +-- -- > $(mkStatic 'f) :: forall a1 .. an. (Typeable a1, .., Typeable an) => Static T--- +-- -- Every module that you write that contains calls to 'mkStatic' needs to -- have a call to 'remotable': -- -- > remotable [ 'f, 'g, ... ] ----- where you must pass every function (or other value) that you pass as an +-- where you must pass every function (or other value) that you pass as an -- argument to 'mkStatic'. The call to 'remotable' will create a definition -- -- > __remoteTable :: RemoteTable -> RemoteTable@@ -32,12 +32,12 @@ -- Cloud Haskell. You should have (at most) one call to 'remotable' per module, -- and compose all created functions when initializing Cloud Haskell: ----- > let rtable :: RemoteTable +-- > let rtable :: RemoteTable -- >     rtable = M1.__remoteTable -- >            . M2.__remoteTable -- >            . ... -- >            . Mn.__remoteTable--- >            $ initRemoteTable +-- >            $ initRemoteTable -- -- NOTE: If you get a type error from ghc along these lines --@@ -52,19 +52,19 @@ -- -- > call :: Serializable a => Static (SerializableDict a) -> NodeId -> Closure (Process a) -> Process a ----- Given some serializable type 'T' you can define +-- Given some serializable type 'T' you can define -- -- > sdictT :: SerializableDict T -- > sdictT = SerializableDict -- -- and then have--- +-- -- > $(mkStatic 'sdictT) :: Static (SerializableDict T) -- -- However, since these dictionaries are so frequently required Cloud Haskell -- provides special support for them.  When you call 'remotable' on a -- /monomorphic/ function @f :: T1 -> T2@--- +-- -- > remotable ['f] -- -- then a serialization dictionary is automatically created for you, which you@@ -80,26 +80,26 @@ -- -- Suppose you have a process ----- > isPrime :: Integer -> Process Bool +-- > isPrime :: Integer -> Process Bool ----- Then +-- Then -- -- > $(mkClosure 'isPrime) :: Integer -> Closure (Process Bool) ----- which you can then 'call', for example, to have a remote node check if +-- which you can then 'call', for example, to have a remote node check if -- a number is prime. -- -- In general, if you have a /monomorphic/ function -- -- > f :: T1 -> T2--- +-- -- then -- -- > $(mkClosure 'f) :: T1 -> Closure T2 -- -- provided that 'T1' is serializable (*) (remember to pass 'f' to 'remotable'). ----- (You can also create closures manually--see the documentation of +-- (You can also create closures manually--see the documentation of -- "Control.Distributed.Static" for examples.) -- -- [Example]@@ -114,34 +114,34 @@ -- > import Control.Distributed.Process.Closure -- > import Control.Distributed.Process.Backend.SimpleLocalnet -- > import Control.Distributed.Process.Node (initRemoteTable)--- > +-- > -- > 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]--- > +-- > -- > remotable ['isPrime]--- > +-- > -- > master :: [NodeId] -> Process () -- > master [] = liftIO $ putStrLn "no slaves" -- > master (slave:_) = do -- >   isPrime79 <- call $(functionTDict 'isPrime) slave ($(mkClosure 'isPrime) (79 :: Integer))--- >   liftIO $ print isPrime79 --- > +-- >   liftIO $ print isPrime79+-- > -- > main :: IO () -- > main = do -- >   args <- getArgs -- >   case args of -- >     ["master", host, port] -> do--- >       backend <- initializeBackend host port rtable --- >       startMaster backend master +-- >       backend <- initializeBackend host port rtable+-- >       startMaster backend master -- >     ["slave", host, port] -> do--- >       backend <- initializeBackend host port rtable +-- >       backend <- initializeBackend host port rtable -- >       startSlave backend -- >   where -- >     rtable :: RemoteTable--- >     rtable = __remoteTable initRemoteTable +-- >     rtable = __remoteTable initRemoteTable -- -- [Notes] --@@ -150,42 +150,43 @@ --     a priori if 'T1' is serializable or not due to a bug in the Template --     Haskell libraries (<http://hackage.haskell.org/trac/ghc/ticket/7066>) ----- (**) Even though 'call' is passed an explicit serialization ---      dictionary, we still need the 'Serializable' constraint because +-- (**) Even though 'call' is passed an explicit serialization+--      dictionary, we still need the 'Serializable' constraint because --      'Static' is not the /true/ static. If it was, we could 'unstatic' --      the dictionary and pattern match on it to bring the 'Typeable' --      instance into scope, but unless proper 'static' support is added to---      ghc we need both the type class argument and the explicit dictionary. -module Control.Distributed.Process.Closure +--      ghc we need both the type class argument and the explicit dictionary.+module Control.Distributed.Process.Closure   ( -- * Serialization dictionaries (and their static versions)     SerializableDict(..)   , staticDecode   , sdictUnit   , sdictProcessId   , sdictSendPort-    -- * The CP type and associated combinators +    -- * The CP type and associated combinators   , CP   , idCP   , splitCP   , returnCP-  , bindCP -  , seqCP -    -- * CP versions of Cloud Haskell primitives  +  , bindCP+  , seqCP+    -- * CP versions of Cloud Haskell primitives   , cpLink   , cpUnlink   , cpSend   , cpExpect   , cpNewChan-#ifdef TemplateHaskellSupport -    -- * Template Haskell support for creating static values and closures +#ifdef TemplateHaskellSupport+    -- * Template Haskell support for creating static values and closures   , remotable   , remotableDecl   , mkStatic   , mkClosure+  , mkStaticClosure   , functionSDict   , functionTDict #endif-  ) where +  ) where  import Control.Distributed.Process.Serializable (SerializableDict(..)) import Control.Distributed.Process.Internal.Closure.BuiltIn@@ -194,27 +195,28 @@   , sdictUnit   , sdictProcessId   , sdictSendPort-    -- The CP type and associated combinators +    -- The CP type and associated combinators   , CP   , idCP   , splitCP   , returnCP-  , bindCP -  , seqCP -    -- CP versions of Cloud Haskell primitives  +  , bindCP+  , seqCP+    -- CP versions of Cloud Haskell primitives   , cpLink   , cpUnlink   , cpSend   , cpExpect   , cpNewChan   )-#ifdef TemplateHaskellSupport -import Control.Distributed.Process.Internal.Closure.TH +#ifdef TemplateHaskellSupport+import Control.Distributed.Process.Internal.Closure.TH   ( remotable   , remotableDecl   , mkStatic   , functionSDict   , functionTDict   , mkClosure+  , mkStaticClosure   ) #endif

@@ -1,8 +1,9 @@ -- | Concurrent queue for single reader, single writer-{-# LANGUAGE MagicHash, UnboxedTuples #-}-module Control.Distributed.Process.Internal.CQueue +{-# LANGUAGE MagicHash, UnboxedTuples, PatternGuards #-}+module Control.Distributed.Process.Internal.CQueue   ( CQueue   , BlockSpec(..)+  , MatchOn(..)   , newCQueue   , enqueue   , dequeue@@ -10,18 +11,21 @@   ) where  import Prelude hiding (length, reverse)-import Control.Concurrent.STM +import Control.Concurrent.STM   ( atomically+  , STM   , TChan+  , tryReadTChan   , newTChan   , writeTChan   , readTChan-  , tryReadTChan+  , orElse+  , retry   ) import Control.Applicative ((<$>), (<*>))-import Control.Exception (mask, onException)+import Control.Exception (mask_, onException) import System.Timeout (timeout)-import Control.Distributed.Process.Internal.StrictMVar +import Control.Distributed.Process.Internal.StrictMVar   ( StrictMVar(StrictMVar)   , newMVar   , takeMVar@@ -29,11 +33,11 @@   ) import Control.Distributed.Process.Internal.StrictList   ( StrictList(..)-  , reverse-  , reverse'+  , append   )+import Data.Maybe (fromJust) import GHC.MVar (MVar(MVar))-import GHC.IO (IO(IO)) +import GHC.IO (IO(IO)) import GHC.Prim (mkWeak#) import GHC.Weak (Weak(Weak)) @@ -50,76 +54,185 @@ enqueue :: CQueue a -> a -> IO () enqueue (CQueue _arrived incoming) !a = atomically $ writeTChan incoming a  -data BlockSpec = +data BlockSpec =     NonBlocking   | Blocking   | Timeout Int +data MatchOn m a+ = MatchMsg  (m -> Maybe a)+ | MatchChan (STM a)++type MatchChunks m a = [Either [m -> Maybe a] [STM a]]++chunkMatches :: [MatchOn m a] -> MatchChunks m a+chunkMatches [] = []+chunkMatches (MatchMsg m : ms) = Left (m : chk) : chunkMatches rest+   where (chk, rest) = spanMatchMsg ms+chunkMatches (MatchChan r : ms) = Right (r : chk) : chunkMatches rest+   where (chk, rest) = spanMatchChan ms++spanMatchMsg :: [MatchOn m a] -> ([m -> Maybe a], [MatchOn m a])+spanMatchMsg [] = ([],[])+spanMatchMsg (m : ms)+    | MatchMsg msg <- m = (msg:msgs, rest)+    | otherwise         = ([], m:ms)+    where !(msgs,rest) = spanMatchMsg ms++spanMatchChan :: [MatchOn m a] -> ([STM a], [MatchOn m a])+spanMatchChan [] = ([],[])+spanMatchChan (m : ms)+    | MatchChan stm <- m = (stm:stms, rest)+    | otherwise          = ([], m:ms)+    where !(stms,rest)  = spanMatchChan ms+ -- | Dequeue an element -- -- The timeout (if any) is applied only to waiting for incoming messages, not -- to checking messages that have already arrived-dequeue :: forall a b. -           CQueue a          -- ^ Queue-        -> BlockSpec         -- ^ Blocking behaviour -        -> [a -> Maybe b]    -- ^ List of matches-        -> IO (Maybe b)      -- ^ 'Nothing' only on timeout-dequeue (CQueue arrived incoming) blockSpec matches = go -  where    -    go :: IO (Maybe b)-    go = mask $ \restore -> do-      arr <- takeMVar arrived -      -- We first check the arrived messages. If we get interrupted during this-      -- search, we just put the MVar back (we haven't read from the Chan yet)-      (arr', mb) <- onException (restore (checkArrived Nil arr))-                                (putMVar arrived arr) -      case (mb, blockSpec) of-        (Just b, _) -> do -          putMVar arrived arr'-          return (Just b)-        (Nothing, NonBlocking) ->-          checkNonBlocking arr'-        (Nothing, Blocking) ->-          Just <$> checkBlocking arr' -        (Nothing, Timeout n) ->-          timeout n $ checkBlocking arr'+dequeue :: forall m a.+           CQueue m          -- ^ Queue+        -> BlockSpec         -- ^ Blocking behaviour+        -> [MatchOn m a]     -- ^ List of matches+        -> IO (Maybe a)      -- ^ 'Nothing' only on timeout+dequeue (CQueue arrived incoming) blockSpec matchons =+  case blockSpec of+    Timeout n -> timeout n $ fmap fromJust run+    _other    ->+       case chunks of+         [Right ports] -> -- channels only, this is easy:+           case blockSpec of+             NonBlocking -> atomically $ waitChans ports (return Nothing)+             _           -> atomically $ waitChans ports retry+                              -- no onException needed+         _other -> run+  where+    chunks = chunkMatches matchons -    -- We reverse the accumulator on return only if we find a match-    checkArrived :: StrictList a -> StrictList a -> IO (StrictList a, Maybe b)-    checkArrived acc Nil = return (acc, Nothing)-    checkArrived acc (Cons x xs) = -      case check x of-        Just y  -> return (reverse' acc xs, Just y)-        Nothing -> checkArrived (Cons x acc) xs+    run = mask_ $ do+           arr <- takeMVar arrived+           let grabNew xs = do+                 r <- atomically $ tryReadTChan incoming+                 case r of+                   Nothing -> return xs+                   Just x  -> grabNew (Snoc xs x)+           arr' <- grabNew arr+           goCheck chunks arr' -    -- If we call checkBlocking there may or may not be a timeout-    checkBlocking :: StrictList a -> IO b-    checkBlocking acc = do-      -- readTChan is a blocking call, and hence is interruptable. If it is -      -- interrupted, we put the value of the accumulator in 'arrived'  -      -- (as opposed to the original value), so that no messages get lost-      -- (hence the low-level structure using mask rather than modifyMVar)-      x <- onException (atomically $ readTChan incoming)-                       (putMVar arrived $ reverse acc)-      case check x of-        Nothing -> checkBlocking (Cons x acc)-        Just y  -> putMVar arrived (reverse acc) >> return y +    waitChans ports on_block =+        foldr orElse on_block (map (fmap Just) ports) -    -- checkNonBlocking is only called if there is no timeout-    checkNonBlocking :: StrictList a -> IO (Maybe b)-    checkNonBlocking acc = do-      -- tryReadTChan is *not* interruptible-      mx <- atomically $ tryReadTChan incoming-      case mx of-        Nothing -> putMVar arrived (reverse acc) >> return Nothing-        Just x  -> case check x of-          Nothing -> checkNonBlocking (Cons x acc)-          Just y  -> putMVar arrived (reverse acc) >> return (Just y)-        -    check :: a -> Maybe b-    check = checkMatches matches +    --+    -- First check the MatchChunks against the messages already in the+    -- mailbox.  For channel matches, we do a non-blocking check at+    -- this point.+    --+    goCheck :: MatchChunks m a+            -> StrictList m  -- messages to check, in this order+            -> IO (Maybe a) -    checkMatches :: [a -> Maybe b] -> a -> Maybe b+    goCheck [] old = goWait old++    goCheck (Right ports : rest) old = do+      r <- atomically $ waitChans ports (return Nothing) -- does not block+      case r of+        Just _  -> returnOld old r+        Nothing -> goCheck rest old++    goCheck (Left matches : rest) old = do+           -- checkArrived might in principle take arbitrary time, so+           -- we ought to call restore and use an exception handler.  However,+           -- the check is usually fast (just a comparison), and the overhead+           -- of passing around restore and setting up exception handlers is+           -- high.  So just don't use expensive matchIfs!+      case checkArrived matches old of+        (old', Just r)  -> returnOld old' (Just r)+        (old', Nothing) -> goCheck rest old'+          -- use the result list, which is now left-biased++    --+    -- Construct an STM transaction that looks at the relevant channels+    -- in the correct order.+    --+    mkSTM :: MatchChunks m a -> STM (Either m a)+    mkSTM [] = retry+    mkSTM (Left _ : rest)+      = fmap Left (readTChan incoming) `orElse` mkSTM rest+    mkSTM (Right ports : rest)+      = foldr orElse (mkSTM rest) (map (fmap Right) ports)++    waitIncoming :: IO (Maybe (Either m a))+    waitIncoming = case blockSpec of+      NonBlocking -> atomically $ fmap Just stm `orElse` return Nothing+      _           -> atomically $ fmap Just stm+     where+      stm = mkSTM chunks++    --+    -- The initial pass didn't find a message, so now we go into blocking+    -- mode.+    --+    -- Contents of 'arrived' from now on is (old ++ new), and+    -- messages that arrive are snocced onto new.+    --+    goWait old = do+      r <- waitIncoming `onException` putMVar arrived old+      case r of+        --  Nothing => non-blocking and no message+        Nothing -> returnOld old Nothing+        Just e  -> case e of+          --+          -- Left => message arrived in the process mailbox.  We now have to+          -- run through the MatchChunks checking each one, because we might+          -- have a situation where the first chunk fails to match and the+          -- second chunk is a channel match and there *is* a message in the+          -- channel.  In that case the channel wins.+          --+          Left m -> goCheck1 chunks m old+          --+          -- Right => message arrived on a channel first+          --+          Right a -> returnOld old (Just a)++    --+    -- A message arrived in the process inbox; check the MatchChunks for+    -- a valid match.+    --+    goCheck1 :: MatchChunks m a+             -> m               -- single message to check+             -> StrictList m    -- old messages we have already checked+             -> IO (Maybe a)++    goCheck1 [] m old = goWait (Snoc old m)++    goCheck1 (Right ports : rest) m old = do+      r <- atomically $ waitChans ports (return Nothing) -- does not block+      case r of+        Nothing -> goCheck1 rest m old+        Just _  -> returnOld (Snoc old m) r++    goCheck1 (Left matches : rest) m old = do+      case checkMatches matches m of+        Nothing -> goCheck1 rest m old+        Just p  -> returnOld old (Just p)++    -- a common pattern for putting back the arrived queue at the end+    returnOld :: StrictList m -> Maybe a -> IO (Maybe a)+    returnOld old r = do putMVar arrived old; return r++    -- as a side-effect, this left-biases the list+    checkArrived :: [m -> Maybe a] -> StrictList m -> (StrictList m, Maybe a)+    checkArrived matches list = go list Nil+      where+        go Nil Nil           = (Nil, Nothing)+        go Nil r             = go r Nil+        go (Append xs ys) tl = go xs (append ys tl)+        go (Snoc xs x)    tl = go xs (Cons x tl)+        go (Cons x xs)    tl+          | Just y <- checkMatches matches x = (append xs tl, Just y)+          | otherwise = let !(rest,r) = go xs tl in (Cons x rest, r)++    checkMatches :: [m -> Maybe a] -> m -> Maybe a     checkMatches []     _ = Nothing     checkMatches (m:ms) a = case m a of Nothing -> checkMatches ms a                                         Just b  -> Just b

@@ -1,5 +1,5 @@-module Control.Distributed.Process.Internal.Closure.BuiltIn -  ( -- * Remote table +module Control.Distributed.Process.Internal.Closure.BuiltIn+  ( -- * Remote table     remoteTable     -- * Static dictionaries and associated operations   , staticDecode@@ -8,14 +8,14 @@   , sdictSendPort     -- * Some static values   , sndStatic-    -- * The CP type and associated combinators +    -- * The CP type and associated combinators   , CP   , idCP   , splitCP   , returnCP-  , bindCP -  , seqCP -    -- * CP versions of Cloud Haskell primitives  +  , bindCP+  , seqCP+    -- * CP versions of Cloud Haskell primitives   , cpLink   , cpUnlink   , cpSend@@ -25,11 +25,11 @@   , cpDelay   ) where -import Data.ByteString.Lazy (ByteString)  +import Data.ByteString.Lazy (ByteString) import Data.Binary (decode, encode) import Data.Rank1Typeable (Typeable, ANY, ANY1, ANY2, ANY3, ANY4) import Data.Rank1Dynamic (toDynamic)-import Control.Distributed.Static +import Control.Distributed.Static   ( RemoteTable   , registerStatic   , Static@@ -42,18 +42,18 @@   , staticCompose   , staticClosure   )-import Control.Distributed.Process.Serializable +import Control.Distributed.Process.Serializable   ( SerializableDict(..)   , Serializable   )-import Control.Distributed.Process.Internal.Types +import Control.Distributed.Process.Internal.Types   ( Process   , ProcessId   , SendPort   , ReceivePort   , ProcessMonitorNotification(ProcessMonitorNotification)   )-import Control.Distributed.Process.Internal.Primitives +import Control.Distributed.Process.Internal.Primitives   ( link   , unlink   , send@@ -71,10 +71,10 @@ --------------------------------------------------------------------------------  remoteTable :: RemoteTable -> RemoteTable-remoteTable = +remoteTable =       registerStatic "$decodeDict"      (toDynamic (decodeDict       :: SerializableDict ANY -> ByteString -> ANY))     . registerStatic "$sdictUnit"       (toDynamic (SerializableDict :: SerializableDict ()))-    . registerStatic "$sdictProcessId"  (toDynamic (SerializableDict :: SerializableDict ProcessId)) +    . registerStatic "$sdictProcessId"  (toDynamic (SerializableDict :: SerializableDict ProcessId))     . registerStatic "$sdictSendPort_"  (toDynamic (sdictSendPort_   :: SerializableDict ANY -> SerializableDict (SendPort ANY)))     . registerStatic "$returnProcess"   (toDynamic (return           :: ANY -> Process ANY))     . registerStatic "$seqProcess"      (toDynamic ((>>)             :: Process ANY1 -> Process ANY2 -> Process ANY2))@@ -119,23 +119,23 @@ -- See module documentation of "Control.Distributed.Process.Closure" for an -- example. staticDecode :: Typeable a => Static (SerializableDict a) -> Static (ByteString -> a)-staticDecode dict = decodeDictStatic `staticApply` dict +staticDecode dict = decodeDictStatic `staticApply` dict   where     decodeDictStatic :: Typeable a => Static (SerializableDict a -> ByteString -> a)     decodeDictStatic = staticLabel "$decodeDict" --- | Serialization dictionary for '()' +-- | Serialization dictionary for '()' sdictUnit :: Static (SerializableDict ())-sdictUnit = staticLabel "$sdictUnit" +sdictUnit = staticLabel "$sdictUnit" --- | Serialization dictionary for 'ProcessId' +-- | Serialization dictionary for 'ProcessId' sdictProcessId :: Static (SerializableDict ProcessId)-sdictProcessId = staticLabel "$sdictProcessId" +sdictProcessId = staticLabel "$sdictProcessId"  -- | Serialization dictionary for 'SendPort'-sdictSendPort :: Typeable a +sdictSendPort :: Typeable a               => Static (SerializableDict a) -> Static (SerializableDict (SendPort a))-sdictSendPort = staticApply (staticLabel "$sdictSendPort_") +sdictSendPort = staticApply (staticLabel "$sdictSendPort_")  -------------------------------------------------------------------------------- -- Static values                                                              --@@ -154,20 +154,20 @@ returnProcessStatic :: Typeable a => Static (a -> Process a) returnProcessStatic = staticLabel "$returnProcess" --- | 'CP' version of 'Control.Category.id' +-- | 'CP' version of 'Control.Category.id' idCP :: Typeable a => CP a a idCP = staticClosure returnProcessStatic  -- | 'CP' version of ('Control.Arrow.***')-splitCP :: (Typeable a, Typeable b, Typeable c, Typeable d) +splitCP :: (Typeable a, Typeable b, Typeable c, Typeable d)         => CP a c -> CP b d -> CP (a, b) (c, d) splitCP p q = cpSplitStatic `closureApplyStatic` p `closureApply` q   where     cpSplitStatic :: Static ((a -> Process c) -> (b -> Process d) -> (a, b) -> Process (c, d))-    cpSplitStatic = staticLabel "$cpSplit" +    cpSplitStatic = staticLabel "$cpSplit"  -- | 'CP' version of 'Control.Monad.return'-returnCP :: forall a. Serializable a +returnCP :: forall a. Serializable a          => Static (SerializableDict a) -> a -> Closure (Process a) returnCP dict x = closure decoder (encode x)   where@@ -179,18 +179,18 @@ -- | 'CP' version of ('Control.Monad.>>') seqCP :: (Typeable a, Typeable b)       => Closure (Process a) -> Closure (Process b) -> Closure (Process b)-seqCP p q = seqProcessStatic `closureApplyStatic` p `closureApply` q +seqCP p q = seqProcessStatic `closureApplyStatic` p `closureApply` q   where     seqProcessStatic :: (Typeable a, Typeable b)                      => Static (Process a -> Process b -> Process b)     seqProcessStatic = staticLabel "$seqProcess" --- | (Not quite the) 'CP' version of ('Control.Monad.>>=') +-- | (Not quite the) 'CP' version of ('Control.Monad.>>=') bindCP :: forall a b. (Typeable a, Typeable b)        => Closure (Process a) -> CP a b -> Closure (Process b)-bindCP x f = bindProcessStatic `closureApplyStatic` x `closureApply` f +bindCP x f = bindProcessStatic `closureApplyStatic` x `closureApply` f   where-    bindProcessStatic :: (Typeable a, Typeable b) +    bindProcessStatic :: (Typeable a, Typeable b)                       => Static (Process a -> (a -> Process b) -> Process b)     bindProcessStatic = staticLabel "$bindProcess" @@ -203,7 +203,7 @@  -- | 'CP' version of 'link' cpLink :: ProcessId -> Closure (Process ())-cpLink = closure (linkStatic `staticCompose` decodeProcessIdStatic) . encode +cpLink = closure (linkStatic `staticCompose` decodeProcessIdStatic) . encode   where     linkStatic :: Static (ProcessId -> Process ())     linkStatic = staticLabel "$link"@@ -216,18 +216,18 @@     unlinkStatic = staticLabel "$unlink"  -- | 'CP' version of 'send'-cpSend :: forall a. Typeable a -       => Static (SerializableDict a) -> ProcessId -> CP a () +cpSend :: forall a. Typeable a+       => Static (SerializableDict a) -> ProcessId -> CP a () cpSend dict pid = closure decoder (encode pid)   where     decoder :: Static (ByteString -> a -> Process ())     decoder = (sendDictStatic `staticApply` dict)-            `staticCompose` -              decodeProcessIdStatic +            `staticCompose`+              decodeProcessIdStatic -    sendDictStatic :: Typeable a +    sendDictStatic :: Typeable a                    => Static (SerializableDict a -> ProcessId -> a -> Process ())-    sendDictStatic = staticLabel "$sendDict" +    sendDictStatic = staticLabel "$sendDict"  -- | 'CP' version of 'expect' cpExpect :: Typeable a => Static (SerializableDict a) -> Closure (Process a)@@ -237,14 +237,14 @@     expectDictStatic = staticLabel "$expectDict"  -- | 'CP' version of 'newChan'-cpNewChan :: Typeable a -          => Static (SerializableDict a) +cpNewChan :: Typeable a+          => Static (SerializableDict a)           -> Closure (Process (SendPort a, ReceivePort a)) cpNewChan dict = staticClosure (newChanDictStatic `staticApply` dict)   where-    newChanDictStatic :: Typeable a +    newChanDictStatic :: Typeable a                       => Static (SerializableDict a -> Process (SendPort a, ReceivePort a))-    newChanDictStatic = staticLabel "$newChanDict"                  +    newChanDictStatic = staticLabel "$newChanDict"  -------------------------------------------------------------------------------- -- Support for spawn                                                          --@@ -252,7 +252,7 @@  -- | @delay them p@ is a process that waits for a signal (a message of type @()@) -- from 'them' (origin is not verified) before proceeding as @p@. In order to--- avoid waiting forever, @delay them p@ monitors 'them'. If it receives a +-- avoid waiting forever, @delay them p@ monitors 'them'. If it receives a -- monitor message instead it simply terminates. delay :: ProcessId -> Process () -> Process () delay them p = do@@ -269,7 +269,7 @@   where     cpDelay' :: ProcessId -> Closure (Process () -> Process ())     cpDelay' pid = closure decoder (encode pid)-    +     decoder :: Static (ByteString -> Process () -> Process ())     decoder = delayStatic `staticCompose` decodeProcessIdStatic 

@@ -1,6 +1,6 @@ -- | Template Haskell support {-# LANGUAGE TemplateHaskell #-}-module Control.Distributed.Process.Internal.Closure.TH +module Control.Distributed.Process.Internal.Closure.TH   ( -- * User-level API     remotable   , remotableDecl@@ -8,11 +8,12 @@   , functionSDict   , functionTDict   , mkClosure+  , mkStaticClosure   ) where  import Prelude hiding (succ, any) import Control.Applicative ((<$>))-import Language.Haskell.TH +import Language.Haskell.TH   ( -- Q monad and operations     Q   , reify@@ -42,7 +43,7 @@   , funD   , sigD   )-import Data.Maybe (catMaybes)  +import Data.Maybe (catMaybes) import Data.Binary (encode) import Data.Generics (everywhereM, mkM, gmapM) import Data.Rank1Dynamic (toDynamic)@@ -51,16 +52,17 @@   , Succ   , TypVar   )-import Control.Distributed.Static +import Control.Distributed.Static   ( RemoteTable   , registerStatic   , Static   , staticLabel   , closure   , staticCompose+  , staticClosure   ) import Control.Distributed.Process.Internal.Types (Process)-import Control.Distributed.Process.Serializable +import Control.Distributed.Process.Serializable   ( SerializableDict(SerializableDict)   ) import Control.Distributed.Process.Internal.Closure.BuiltIn (staticDecode)@@ -71,12 +73,12 @@  -- | Create the closure, decoder, and metadata definitions for the given list -- of functions-remotable :: [Name] -> Q [Dec] +remotable :: [Name] -> Q [Dec] remotable ns = do-    types <- mapM getType ns -    (closures, inserts) <- unzip <$> mapM generateDefs types +    types <- mapM getType ns+    (closures, inserts) <- unzip <$> mapM generateDefs types     rtable <- createMetaData (mkName "__remoteTable") (concat inserts)-    return $ concat closures ++ rtable +    return $ concat closures ++ rtable  -- | Like 'remotable', but parameterized by the declaration of a function -- instead of the function name. So where for 'remotable' you'd do@@ -106,9 +108,9 @@ remotableDecl qDecs = do     decs <- concat <$> sequence qDecs     let types = catMaybes (map typeOf decs)-    (closures, inserts) <- unzip <$> mapM generateDefs types +    (closures, inserts) <- unzip <$> mapM generateDefs types     rtable <- createMetaData (mkName "__remoteTableDecl") (concat inserts)-    return $ decs ++ concat closures ++ rtable +    return $ decs ++ concat closures ++ rtable   where     typeOf :: Dec -> Maybe (Name, Type)     typeOf (SigD name typ) = Just (name, typ)@@ -116,15 +118,15 @@  -- | Construct a static value. ----- If @f : forall a1 .. an. T@ --- then @$(mkStatic 'f) :: forall a1 .. an. Static T@. --- Be sure to pass 'f' to 'remotable'. +-- If @f : forall a1 .. an. T@+-- then @$(mkStatic 'f) :: forall a1 .. an. Static T@.+-- Be sure to pass 'f' to 'remotable'. mkStatic :: Name -> Q Exp mkStatic = varE . staticName --- | If @f : T1 -> T2@ is a monomorphic function +-- | If @f : T1 -> T2@ is a monomorphic function -- then @$(functionSDict 'f) :: Static (SerializableDict T1)@.--- +-- -- Be sure to pass 'f' to 'remotable'. functionSDict :: Name -> Q Exp functionSDict = varE . sdictName@@ -136,24 +138,31 @@ functionTDict :: Name -> Q Exp functionTDict = varE . tdictName --- | If @f : T1 -> T2@ then @$(mkClosure 'f) :: T1 -> Closure T2@. +-- | If @f : T1 -> T2@ then @$(mkClosure 'f) :: T1 -> Closure T2@. ----- TODO: The current version of mkClosure is too polymorphic +-- TODO: The current version of mkClosure is too polymorphic -- (@forall a. Binary a => a -> Closure T2). mkClosure :: Name -> Q Exp-mkClosure n = -  [|   closure ($(mkStatic n) `staticCompose` staticDecode $(functionSDict n)) +mkClosure n =+  [|   closure ($(mkStatic n) `staticCompose` staticDecode $(functionSDict n))      . encode   |] +-- | Make a 'Closure' from a static function.  This is useful for+-- making a closure for a top-level @Process ()@ function, because+-- using 'mkClosure' would require adding a dummy @()@ argument.+--+mkStaticClosure :: Name -> Q Exp+mkStaticClosure n = [| staticClosure $( mkStatic n ) |]+ -------------------------------------------------------------------------------- -- Internal (Template Haskell)                                                -- --------------------------------------------------------------------------------  -- | Generate the code to add the metadata to the CH runtime createMetaData :: Name -> [Q Exp] -> Q [Dec]-createMetaData name is = -  sequence [ sigD name [t| RemoteTable -> RemoteTable |] +createMetaData name is =+  sequence [ sigD name [t| RemoteTable -> RemoteTable |]            , sfnD name (compose is)            ] @@ -163,52 +172,52 @@     let (typVars, typ') = case fullType of ForallT vars [] mono -> (vars, mono)                                            _                    -> ([], fullType) -    -- The main "static" entry                                  -    (static, register) <- makeStatic typVars typ' -     -    -- If n :: T1 -> T2, static serializable dictionary for T1 +    -- The main "static" entry+    (static, register) <- makeStatic typVars typ'++    -- If n :: T1 -> T2, static serializable dictionary for T1     -- TODO: we should check if arg is an instance of Serializable, but we cannot     -- http://hackage.haskell.org/trac/ghc/ticket/7066     (sdict, registerSDict) <- case (typVars, typ') of-      ([], ArrowT `AppT` arg `AppT` _res) -> +      ([], ArrowT `AppT` arg `AppT` _res) ->         makeDict (sdictName origName) arg-      _ -> +      _ ->         return ([], [])-    +     -- If n :: T1 -> Process T2, static serializable dictionary for T2     -- TODO: check if T2 is serializable (same as above)     (tdict, registerTDict) <- case (typVars, typ') of-      ([], ArrowT `AppT` _arg `AppT` (proc' `AppT` res)) | proc' == proc -> -        makeDict (tdictName origName) res +      ([], ArrowT `AppT` _arg `AppT` (proc' `AppT` res)) | proc' == proc ->+        makeDict (tdictName origName) res       _ ->         return ([], [])-    +     return ( concat [static, sdict, tdict]            , concat [register, registerSDict, registerTDict]            )   where     makeStatic :: [TyVarBndr] -> Type -> Q ([Dec], [Q Exp])-    makeStatic typVars typ = do +    makeStatic typVars typ = do       static <- generateStatic origName typVars typ-      let dyn = case typVars of +      let dyn = case typVars of                   [] -> [| toDynamic $(varE origName) |]                   _  -> [| toDynamic ($(varE origName) :: $(monomorphize typVars typ)) |]       return ( static              , [ [| registerStatic $(showFQN origName) $dyn |] ]              ) -    makeDict :: Name -> Type -> Q ([Dec], [Q Exp]) +    makeDict :: Name -> Type -> Q ([Dec], [Q Exp])     makeDict dictName typ = do-      sdict <- generateDict dictName typ +      sdict <- generateDict dictName typ       let dyn = [| toDynamic (SerializableDict :: SerializableDict $(return typ)) |]       return ( sdict-             , [ [| registerStatic $(showFQN dictName) $dyn |] ] +             , [ [| registerStatic $(showFQN dictName) $dyn |] ]              )  -- | Turn a polymorphic type into a monomorphic type using ANY and co monomorphize :: [TyVarBndr] -> Type -> Q Type-monomorphize tvs = -    let subst = zip (map tyVarBndrName tvs) anys +monomorphize tvs =+    let subst = zip (map tyVarBndrName tvs) anys     in everywhereM (mkM (applySubst subst))   where     anys :: [Q Type]@@ -219,34 +228,34 @@      zero :: Q Type     zero = [t| Zero |]-    +     succ :: Q Type -> Q Type     succ t = [t| Succ $t |]- +     applySubst :: [(Name, Q Type)] -> Type -> Q Type-    applySubst s (VarT n) = -      case lookup n s of  +    applySubst s (VarT n) =+      case lookup n s of         Nothing -> return (VarT n)         Just t  -> t     applySubst s t = gmapM (mkM (applySubst s)) t --- | Generate a static value +-- | Generate a static value generateStatic :: Name -> [TyVarBndr] -> Type -> Q [Dec] generateStatic n xs typ = do     staticTyp <- [t| Static |]     sequence-      [ sigD (staticName n) $ -          return (ForallT xs -                  (map typeable xs) +      [ sigD (staticName n) $+          return (ForallT xs+                  (map typeable xs)                   (staticTyp `AppT` typ)           )       , sfnD (staticName n) [| staticLabel $(showFQN n) |]       ]   where     typeable :: TyVarBndr -> Pred-    typeable tv = ClassP (mkName "Typeable") [VarT (tyVarBndrName tv)] +    typeable tv = ClassP (mkName "Typeable") [VarT (tyVarBndrName tv)] --- | Generate a serialization dictionary with name 'n' for type 'typ' +-- | Generate a serialization dictionary with name 'n' for type 'typ' generateDict :: Name -> Type -> Q [Dec] generateDict n typ = do     sequence@@ -270,7 +279,7 @@ -- | Compose a set of expressions compose :: [Q Exp] -> Q Exp compose []     = [| id |]-compose [e]    = e +compose [e]    = e compose (e:es) = [| $e . $(compose es) |]  -- | Literal string as an expression@@ -279,17 +288,17 @@  -- | Look up the "original name" (module:name) and type of a top-level function getType :: Name -> Q (Name, Type)-getType name = do +getType name = do   info <- reify name-  case info of +  case info of     VarI origName typ _ _ -> return (origName, typ)     _                     -> fail $ show name ++ " not found"  -- | Variation on 'funD' which takes a single expression to define the function sfnD :: Name -> Q Exp -> Q Dec-sfnD n e = funD n [clause [] (normalB e) []] -    --- | The name of a type variable binding occurrence    +sfnD n e = funD n [clause [] (normalB e) []]++-- | The name of a type variable binding occurrence tyVarBndrName :: TyVarBndr -> Name tyVarBndrName (PlainTV n)    = n tyVarBndrName (KindedTV n _) = n@@ -299,7 +308,7 @@ -- We ignore the module part of the Name argument (which may or may not exist) -- because we construct various names (`staticName`, `sdictName`, `tdictName`) -- and those names certainly won't have Module components.-showFQN :: Name -> Q Exp +showFQN :: Name -> Q Exp showFQN n = do   loc <- location   stringE (loc_module loc ++ "." ++ nameBase n)

@@ -2,8 +2,8 @@   ( sendPayload   , sendBinary   , sendMessage-  , disconnect -  , closeImplicitReconnections +  , disconnect+  , closeImplicitReconnections   , impliesDeathOf   ) where @@ -15,7 +15,7 @@ import Control.Distributed.Process.Internal.StrictMVar (withMVar, modifyMVar_) import Control.Concurrent.Chan (writeChan) import Control.Monad (unless)-import qualified Network.Transport as NT +import qualified Network.Transport as NT   ( Connection   , send   , defaultConnectHints@@ -23,7 +23,7 @@   , Reliability(ReliableOrdered)   , close   )-import Control.Distributed.Process.Internal.Types +import Control.Distributed.Process.Internal.Types   ( LocalNode(localState, localEndPoint, localCtrlChan)   , Identifier   , localConnections@@ -44,14 +44,14 @@ import Control.Distributed.Process.Serializable (Serializable)  ----------------------------------------------------------------------------------- Message sending                                                            -- +-- Message sending                                                            -- -------------------------------------------------------------------------------- -sendPayload :: LocalNode +sendPayload :: LocalNode             -> Identifier             -> Identifier             -> ImplicitReconnect-            -> [BSS.ByteString] +            -> [BSS.ByteString]             -> IO () sendPayload node from to implicitReconnect payload = do   mConn <- connBetween node from to implicitReconnect@@ -59,53 +59,53 @@     Just conn -> do       didSend <- NT.send conn payload       case didSend of-        Left _err -> return False -        Right ()  -> return True +        Left _err -> return False+        Right ()  -> return True     Nothing -> return False   unless didSend $ do     writeChan (localCtrlChan node) NCMsg-      { ctrlMsgSender = to +      { ctrlMsgSender = to       , ctrlMsgSignal = Died to DiedDisconnect       } -sendBinary :: Binary a -           => LocalNode -           -> Identifier -           -> Identifier +sendBinary :: Binary a+           => LocalNode+           -> Identifier+           -> Identifier            -> ImplicitReconnect-           -> a +           -> a            -> IO ()-sendBinary node from to implicitReconnect +sendBinary node from to implicitReconnect   = sendPayload node from to implicitReconnect . BSL.toChunks . encode -sendMessage :: Serializable a -            => LocalNode -            -> Identifier -            -> Identifier +sendMessage :: Serializable a+            => LocalNode+            -> Identifier+            -> Identifier             -> ImplicitReconnect-            -> a +            -> a             -> IO ()-sendMessage node from to implicitReconnect = +sendMessage node from to implicitReconnect =   sendPayload node from to implicitReconnect . messageToPayload . createMessage -setupConnBetween :: LocalNode -                 -> Identifier -                 -> Identifier -                 -> ImplicitReconnect +setupConnBetween :: LocalNode+                 -> Identifier+                 -> Identifier+                 -> ImplicitReconnect                  -> IO (Maybe NT.Connection) setupConnBetween node from to implicitReconnect = do-    mConn <- NT.connect endPoint -                        (nodeAddress . nodeOf $ to) -                        NT.ReliableOrdered +    mConn <- NT.connect endPoint+                        (nodeAddress . nodeOf $ to)+                        NT.ReliableOrdered                         NT.defaultConnectHints-    case mConn of +    case mConn of       Right conn -> do         didSend <- NT.send conn (BSL.toChunks . encode $ to)         case didSend of           Left _ ->             return Nothing           Right () -> do-            modifyMVar_ nodeState $ return . +            modifyMVar_ nodeState $ return .               (localConnectionBetween from to ^= Just (conn, implicitReconnect))             return $ Just conn       Left _ ->@@ -114,33 +114,33 @@     endPoint  = localEndPoint node     nodeState = localState node -connBetween :: LocalNode -            -> Identifier -            -> Identifier +connBetween :: LocalNode+            -> Identifier+            -> Identifier             -> ImplicitReconnect             -> IO (Maybe NT.Connection) connBetween node from to implicitReconnect = do     mConn <- withMVar nodeState $ return . (^. localConnectionBetween from to)     case mConn of-      Just (conn, _) -> +      Just (conn, _) ->         return $ Just conn-      Nothing -> -        setupConnBetween node from to implicitReconnect +      Nothing ->+        setupConnBetween node from to implicitReconnect   where     nodeState = localState node  disconnect :: LocalNode -> Identifier -> Identifier -> IO () disconnect node from to =-  modifyMVar_ (localState node) $ \st -> +  modifyMVar_ (localState node) $ \st ->     case st ^. localConnectionBetween from to of-      Nothing -> +      Nothing ->         return st       Just (conn, _) -> do-        NT.close conn +        NT.close conn         return (localConnectionBetween from to ^= Nothing $ st)  closeImplicitReconnections :: LocalNode -> Identifier -> IO ()-closeImplicitReconnections node to = +closeImplicitReconnections node to =   modifyMVar_ (localState node) $ \st -> do     let shouldClose (_, to') (_, WithImplicitReconnect) = to `impliesDeathOf` to'         shouldClose _ _ = False@@ -153,7 +153,7 @@ impliesDeathOf :: Identifier                -> Identifier                -> Bool-NodeIdentifier nid `impliesDeathOf` NodeIdentifier nid' = +NodeIdentifier nid `impliesDeathOf` NodeIdentifier nid' =   nid' == nid NodeIdentifier nid `impliesDeathOf` ProcessIdentifier pid =   processNodeId pid == nid

@@ -1,10 +1,10 @@ -- | Cloud Haskell primitives ----- We define these in a separate module so that we don't have to rely on +-- We define these in a separate module so that we don't have to rely on -- the closure combinators-module Control.Distributed.Process.Internal.Primitives +module Control.Distributed.Process.Internal.Primitives   ( -- * Basic messaging-    send +    send   , expect     -- * Channels   , newChan@@ -19,18 +19,31 @@   , match   , matchIf   , matchUnknown-  , AbstractMessage(..) +  , AbstractMessage(..)   , matchAny+  , matchAnyIf+  , matchChan     -- * Process management   , terminate   , ProcessTerminationException(..)+  , die+  , kill+  , exit+  , catchExit+  , catchesExit+    -- keep the exception constructor hidden, so that handling exit+    -- reasons /must/ take place via the 'catchExit' family of primitives+  , ProcessExitException()   , getSelfPid   , getSelfNode+  , ProcessInfo(..)+  , getProcessInfo     -- * Monitoring and linking   , link   , unlink   , monitor   , unmonitor+  , withMonitor     -- * Logging   , say     -- * Registry@@ -49,6 +62,9 @@   , unStatic     -- * Exception handling   , catch+  , Handler(..)+  , catches+  , try   , mask   , onException   , bracket@@ -67,6 +83,8 @@     -- * Reconnecting   , reconnect   , reconnectPort+    -- * Tracing/Debugging+  , trace   ) where  #if ! MIN_VERSION_base(4,6,0)@@ -82,15 +100,15 @@ import Control.Monad.Reader (ask) import Control.Monad.IO.Class (MonadIO, liftIO) import Control.Applicative ((<$>))-import Control.Exception (Exception, throwIO, SomeException)-import qualified Control.Exception as Ex (catch, mask)-import Control.Distributed.Process.Internal.StrictMVar +import Control.Exception (Exception(..), throw, throwIO, SomeException)+import qualified Control.Exception as Ex (catch, mask, try)+import Control.Distributed.Process.Internal.StrictMVar   ( StrictMVar   , modifyMVar   , modifyMVar_   ) import Control.Concurrent.Chan (writeChan)-import Control.Concurrent.STM +import Control.Concurrent.STM   ( STM   , TVar   , atomically@@ -99,13 +117,17 @@   , readTVar   , writeTVar   )-import Control.Distributed.Process.Internal.CQueue (dequeue, BlockSpec(..))+import Control.Distributed.Process.Internal.CQueue+  ( dequeue+  , BlockSpec(..)+  , MatchOn(..)+  ) import Control.Distributed.Process.Serializable (Serializable, fingerprint) import Data.Accessor ((^.), (^:), (^=)) import Control.Distributed.Static (Closure, Static) import Data.Rank1Typeable (Typeable) import qualified Control.Distributed.Static as Static (unstatic, unclosure)-import Control.Distributed.Process.Internal.Types +import Control.Distributed.Process.Internal.Types   ( NodeId(..)   , ProcessId(..)   , LocalNode(..)@@ -116,7 +138,7 @@   , SpawnRef(..)   , NCMsg(..)   , ProcessSignal(..)-  , monitorCounter +  , monitorCounter   , spawnCounter   , SendPort(..)   , ReceivePort(..)@@ -125,6 +147,7 @@   , TypedChannel(..)   , SendPortId(..)   , Identifier(..)+  , ProcessExitException(..)   , DidUnmonitor(..)   , DidUnlinkProcess(..)   , DidUnlinkNode(..)@@ -132,6 +155,8 @@   , WhereIsReply(..)   , RegisterReply(..)   , ProcessRegistrationException(..)+  , ProcessInfo(..)+  , ProcessInfoNone(..)   , createMessage   , runLocalProcess   , ImplicitReconnect(WithImplicitReconnect, NoImplicitReconnect)@@ -139,13 +164,14 @@   , LocalSendPortId   , messageToPayload   )-import Control.Distributed.Process.Internal.Messaging +import Control.Distributed.Process.Internal.Messaging   ( sendMessage   , sendBinary   , sendPayload   , disconnect-  ) -import Control.Distributed.Process.Internal.WeakTQueue +  )+import qualified Control.Distributed.Process.Internal.Trace as Trace+import Control.Distributed.Process.Internal.WeakTQueue   ( newTQueueIO   , readTQueue   , mkWeakTQueue@@ -160,16 +186,16 @@ -- This requires a lookup on every send. If we want to avoid that we need to -- modify serializable to allow for stateful (IO) deserialization send them msg = do-  proc <- ask -  liftIO $ sendMessage (processNode proc) -                       (ProcessIdentifier (processId proc)) +  proc <- ask+  liftIO $ sendMessage (processNode proc)+                       (ProcessIdentifier (processId proc))                        (ProcessIdentifier them)                        NoImplicitReconnect                        msg  -- | Wait for a message of a specific type expect :: forall a. Serializable a => Process a-expect = receiveWait [match return] +expect = receiveWait [match return]  -------------------------------------------------------------------------------- -- Channels                                                                   --@@ -178,17 +204,17 @@ -- | Create a new typed channel newChan :: Serializable a => Process (SendPort a, ReceivePort a) newChan = do-    proc <- ask +    proc <- ask     liftIO . modifyMVar (processState proc) $ \st -> do       let lcid  = st ^. channelCounter       let cid   = SendPortId { sendPortProcessId = processId proc                              , sendPortLocalId   = lcid                              }-      let sport = SendPort cid +      let sport = SendPort cid       chan  <- liftIO newTQueueIO       chan' <- mkWeakTQueue chan $ finalizer (processState proc) lcid       let rport = ReceivePort $ readTQueue chan-      let tch   = TypedChannel chan' +      let tch   = TypedChannel chan'       return ( (channelCounter ^: (+ 1))              . (typedChannelWithId lcid ^= Just tch)              $ st@@ -196,7 +222,7 @@              )   where     finalizer :: StrictMVar LocalProcessState -> LocalSendPortId -> IO ()-    finalizer st lcid = modifyMVar_ st $ +    finalizer st lcid = modifyMVar_ st $       return . (typedChannelWithId lcid ^= Nothing)  -- | Send a message on a typed channel@@ -205,24 +231,24 @@   proc <- ask   liftIO $ sendBinary (processNode proc)                       (ProcessIdentifier (processId proc))-                      (SendPortIdentifier cid) +                      (SendPortIdentifier cid)                       NoImplicitReconnect-                      msg +                      msg  -- | Wait for a message on a typed channel receiveChan :: Serializable a => ReceivePort a -> Process a-receiveChan = liftIO . atomically . receiveSTM +receiveChan = liftIO . atomically . receiveSTM --- | Like 'receiveChan' but with a timeout. If the timeout is 0, do a +-- | Like 'receiveChan' but with a timeout. If the timeout is 0, do a -- non-blocking check for a message. receiveChanTimeout :: Serializable a => Int -> ReceivePort a -> Process (Maybe a)-receiveChanTimeout 0 ch = liftIO . atomically $ +receiveChanTimeout 0 ch = liftIO . atomically $   (Just <$> receiveSTM ch) `orElse` return Nothing-receiveChanTimeout n ch = liftIO . timeout n . atomically $ +receiveChanTimeout n ch = liftIO . timeout n . atomically $   receiveSTM ch  -- | Merge a list of typed channels.--- +-- -- The result port is left-biased: if there are messages available on more -- than one port, the first available message is returned. mergePortsBiased :: Serializable a => [ReceivePort a] -> Process (ReceivePort a)@@ -233,7 +259,7 @@ mergePortsRR :: Serializable a => [ReceivePort a] -> Process (ReceivePort a) mergePortsRR = \ps -> do     psVar <- liftIO . atomically $ newTVar (map receiveSTM ps)-    return $ ReceivePort (rr psVar) +    return $ ReceivePort (rr psVar)   where     rotate :: [a] -> [a]     rotate []     = []@@ -242,16 +268,16 @@     rr :: TVar [STM a] -> STM a     rr psVar = do       ps <- readTVar psVar-      a  <- foldr1 orElse ps +      a  <- foldr1 orElse ps       writeTVar psVar (rotate ps)       return a  ----------------------------------------------------------------------------------- Advanced messaging                                                         -- +-- Advanced messaging                                                         -- --------------------------------------------------------------------------------  -- | Opaque type used in 'receiveWait' and 'receiveTimeout'-newtype Match b = Match { unMatch :: Message -> Maybe (Process b) }+newtype Match b = Match { unMatch :: MatchOn Message (Process b) }  -- | Test the matches in order against each message in the queue receiveWait :: [Match b] -> Process b@@ -261,7 +287,7 @@   proc  -- | Like 'receiveWait' but with a timeout.--- +-- -- If the timeout is zero do a non-blocking check for matching messages. A -- non-zero timeout is applied only when waiting for incoming messages (that is, -- /after/ we have checked the messages that are already in the mailbox).@@ -274,45 +300,92 @@     Nothing   -> return Nothing     Just proc -> Just <$> proc +matchChan :: ReceivePort a -> (a -> Process b) -> Match b+matchChan p fn = Match $ MatchChan (fmap fn (receiveSTM p))+ -- | Match against any message of the right type match :: forall a b. Serializable a => (a -> Process b) -> Match b-match = matchIf (const True) +match = matchIf (const True)  -- | 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 $ \msg -> +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 +         -- 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 +-- | Represents a received message and provides two basic operations on it. data AbstractMessage = AbstractMessage {-    forward :: ProcessId -> Process ()+    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 an arbitrary message+-- | 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!+-- matchAny :: forall b. (AbstractMessage -> Process b) -> Match b-matchAny p = Match $ Just . p . abstract-  where -    abstract :: Message -> AbstractMessage-    abstract msg = AbstractMessage {-        forward = \them -> do-          proc <- ask-          liftIO $ sendPayload (processNode proc)-                               (ProcessIdentifier (processId proc))-                               (ProcessIdentifier them)-                               NoImplicitReconnect -                               (messageToPayload msg)-      }+matchAny p = Match $ MatchMsg $ Just . p . abstract +-- | 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+-- 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'.+matchAnyIf :: forall a b. (Serializable a)+                       => (a -> Bool)+                       -> (AbstractMessage -> Process b)+                       -> Match b+matchAnyIf c p = Match $ MatchMsg $ \msg ->+   case messageFingerprint msg == fingerprint (undefined :: a) of+     True | c decoded -> Just (p (abstract 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+         !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+  }+ -- | Remove any message from the queue matchUnknown :: Process b -> Match b-matchUnknown = Match . const . Just+matchUnknown p = Match $ MatchMsg (const (Just p))  -------------------------------------------------------------------------------- -- Process management                                                         --@@ -324,18 +397,108 @@  instance Exception ProcessTerminationException --- | Terminate (throws a ProcessTerminationException)+-- | Terminate immediately (throws a ProcessTerminationException) terminate :: Process a terminate = liftIO $ throwIO ProcessTerminationException +-- [Issue #110]+-- | Die immediately - throws a 'ProcessExitException' with the given @reason@.+die :: Serializable a => a -> Process b+die reason = do+  pid <- getSelfPid+  liftIO $ throwIO (ProcessExitException pid (createMessage reason))++-- | Forceful request to kill a process. Where 'exit' provides an exception+-- that can be caught and handled, 'kill' throws an unexposed exception type+-- which cannot be handled explicitly (by type).+kill :: ProcessId -> String -> Process ()+-- NOTE: We send the message to our local node controller, which will then+-- forward it to a remote node controller (if applicable). Sending it directly+-- to a remote node controller means that that the message may overtake a+-- 'monitor' or 'link' request.+kill them reason = sendCtrlMsg Nothing (Kill them reason)++-- | Graceful request to exit a process. Throws 'ProcessExitException' with the+-- supplied @reason@ encoded as a message. Any /exit signal/ raised in this+-- manner can be handled using the 'catchExit' family of functions.+exit :: Serializable a => ProcessId -> a -> Process ()+-- NOTE: We send the message to our local node controller, which will then+-- forward it to a remote node controller (if applicable). Sending it directly+-- to a remote node controller means that that the message may overtake a+-- 'monitor' or 'link' request.+exit them reason = sendCtrlMsg Nothing (Exit them (createMessage reason))++-- | Catches 'ProcessExitException'. The handler will not be applied unless its+-- type matches the encoded data stored in the exception (see the /reason/+-- argument given to the 'exit' primitive). If the handler cannot be applied,+-- the exception will be re-thrown.+--+-- To handle 'ProcessExitException' without regard for /reason/, see 'catch'.+-- To handle multiple /reasons/ of differing types, see 'catchesExit'.+catchExit :: forall a b . (Show a, Serializable a)+                       => Process b+                       -> (ProcessId -> a -> Process b)+                       -> Process b+catchExit act exitHandler = catch act handleExit+  where+    handleExit ex@(ProcessExitException from msg) =+        if messageFingerprint msg == fingerprint (undefined :: a)+          then exitHandler from decoded+          else liftIO $ throwIO ex+     where+       decoded :: a+       -- Make sure the value is fully decoded so that we don't hang to+       -- bytestrings if the caller doesn't use the value immediately+       !decoded = decode (messageEncoding msg)++-- | 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+-- handle a list of potentially different handlers (and therefore input types),+-- a handler passed to 'catchesExit' must accept 'AbstractMessage' and return+-- @Maybe@ (i.e., @Just p@ if it handled the exit reason, otherwise @Nothing@).+--+-- See 'maybeHandleMessage' and 'AsbtractMessage' for more details.+catchesExit :: Process b+            -> [(ProcessId -> AbstractMessage -> (Process (Maybe b)))]+            -> Process b+catchesExit act handlers = catch act ((flip handleExit) handlers)+  where+    handleExit :: ProcessExitException+               -> [(ProcessId -> AbstractMessage -> Process (Maybe b))]+               -> Process b+    handleExit ex [] = liftIO $ throwIO ex+    handleExit ex@(ProcessExitException from msg) (h:hs) = do+      r <- h from (abstract msg)+      case r of+        Nothing -> handleExit ex hs+        Just p  -> return p+ -- | Our own process ID getSelfPid :: Process ProcessId-getSelfPid = processId <$> ask +getSelfPid = processId <$> ask  -- | Get the node ID of our local node getSelfNode :: Process NodeId-getSelfNode = localNodeId . processNode <$> ask +getSelfNode = localNodeId . processNode <$> ask +-- | Get information about the specified process+getProcessInfo :: ProcessId -> Process (Maybe ProcessInfo)+getProcessInfo pid =+  let them = processNodeId pid in do+  us <- getSelfNode+  dest <- mkNode them us+  sendCtrlMsg dest $ GetInfo pid+  receiveWait [+       match (\(p :: ProcessInfo)     -> return $ Just p)+     , match (\(_ :: ProcessInfoNone) -> return Nothing)+     ]+  where mkNode :: NodeId -> NodeId -> Process (Maybe NodeId)+        mkNode them us = case them == us of+                           True -> return Nothing+                           _    -> return $ Just them+ -------------------------------------------------------------------------------- -- Monitoring and linking                                                     -- --------------------------------------------------------------------------------@@ -354,7 +517,7 @@ -- > unlink pidB -- Unlink again -- -- doesn't quite do what one might expect: if process B sends a message to--- process A, and /subsequently terminates/, then process A might or might not +-- process A, and /subsequently terminates/, then process A might or might not -- be terminated too, depending on whether the exception is thrown before or -- after the 'unlink' (i.e., this code has a race condition). --@@ -369,7 +532,7 @@  -- | Monitor another process (asynchronous) ----- When process A monitors process B (that is, process A calls +-- When process A monitors process B (that is, process A calls -- @monitor pidB@) then process A will receive a 'ProcessMonitorNotification' -- when process B terminates (normally or abnormally), or when process A gets -- disconnected from process B. You receive this message like any other (using@@ -377,27 +540,41 @@ -- 'DiedDisconnect', etc.). -- -- Every call to 'monitor' returns a new monitor reference 'MonitorRef'; if--- multiple monitors are set up, multiple notifications will be delivered +-- multiple monitors are set up, multiple notifications will be delivered -- and monitors can be disabled individually using 'unmonitor'.-monitor :: ProcessId -> Process MonitorRef -monitor = monitor' . ProcessIdentifier +monitor :: ProcessId -> Process MonitorRef+monitor = monitor' . ProcessIdentifier --- | Remove a link +-- | Establishes temporary monitoring of another process. --+-- @withMonitor pid code@ sets up monitoring of @pid@ for the duration+-- of @code@.  Note: although monitoring is no longer active when+-- @withMonitor@ returns, there might still be unreceived monitor+-- messages in the queue.+--+withMonitor :: ProcessId -> Process a -> Process a+withMonitor pid code = bracket (monitor pid) unmonitor (\_ -> code)+  -- unmonitor blocks waiting for the response, so there's a possibility+  -- that an exception might interrupt withMonitor before the unmonitor+  -- has completed.  I think that's better than making the unmonitor+  -- uninterruptible.++-- | Remove a link+-- -- This is synchronous in the sense that once it returns you are guaranteed -- that no exception will be raised if the remote process dies. However, it is--- asynchronous in the sense that we do not wait for a response from the remote +-- asynchronous in the sense that we do not wait for a response from the remote -- node. unlink :: ProcessId -> Process () unlink pid = do   unlinkAsync pid-  receiveWait [ matchIf (\(DidUnlinkProcess pid') -> pid' == pid) -                        (\_ -> return ()) +  receiveWait [ matchIf (\(DidUnlinkProcess pid') -> pid' == pid)+                        (\_ -> return ())               ] --- | Remove a node link +-- | Remove a node link ----- This has the same synchronous/asynchronous nature as 'unlink'. +-- This has the same synchronous/asynchronous nature as 'unlink'. unlinkNode :: NodeId -> Process () unlinkNode nid = do   unlinkNodeAsync nid@@ -407,7 +584,7 @@  -- | Remove a channel (send port) link ----- This has the same synchronous/asynchronous nature as 'unlink'. +-- This has the same synchronous/asynchronous nature as 'unlink'. unlinkPort :: SendPort a -> Process () unlinkPort sport = do   unlinkPortAsync sport@@ -415,9 +592,9 @@                         (\_ -> return ())               ] --- | Remove a monitor +-- | Remove a monitor ----- This has the same synchronous/asynchronous nature as 'unlink'. +-- This has the same synchronous/asynchronous nature as 'unlink'. unmonitor :: MonitorRef -> Process () unmonitor ref = do   unmonitorAsync ref@@ -433,11 +610,17 @@ catch :: Exception e => Process a -> (e -> Process a) -> Process a catch p h = do   lproc <- ask-  liftIO $ Ex.catch (runLocalProcess lproc p) (runLocalProcess lproc . h) +  liftIO $ Ex.catch (runLocalProcess lproc p) (runLocalProcess lproc . h) --- | Lift 'Control.Exception.mask' +-- | Lift 'Control.Exception.try'+try :: Exception e => Process a -> Process (Either e a)+try p = do+  lproc <- ask+  liftIO $ Ex.try (runLocalProcess lproc p)++-- | Lift 'Control.Exception.mask' mask :: ((forall a. Process a -> Process a) -> Process b) -> Process b-mask p = do +mask p = do     lproc <- ask     liftIO $ Ex.mask $ \restore ->       runLocalProcess lproc (p (liftRestore lproc restore))@@ -465,18 +648,35 @@  -- | Lift 'Control.Exception.finally' finally :: Process a -> Process b -> Process a-finally a sequel = bracket_ (return ()) sequel a +finally a sequel = bracket_ (return ()) sequel a +-- | You need this when using 'catches'+data Handler a = forall e . Exception e => Handler (e -> Process a)++instance Functor Handler where+     fmap f (Handler h) = Handler (fmap f . h)++-- | Lift 'Control.Exception.catches'+catches :: Process a -> [Handler a] -> Process a+catches proc handlers = proc `catch` catchesHandler handlers++catchesHandler :: [Handler a] -> SomeException -> Process a+catchesHandler handlers e = foldr tryHandler (throw e) handlers+    where tryHandler (Handler handler) res+              = case fromException e of+                Just e' -> handler e'+                Nothing -> res+ -------------------------------------------------------------------------------- -- Auxiliary API                                                              -- --------------------------------------------------------------------------------  -- | Like 'expect' but with a timeout expectTimeout :: forall a. Serializable a => Int -> Process (Maybe a)-expectTimeout n = receiveTimeout n [match return] +expectTimeout n = receiveTimeout n [match return]  -- | Asynchronous version of 'spawn'--- +-- -- ('spawn' is defined in terms of 'spawnAsync' and 'expect') spawnAsync :: NodeId -> Closure (Process ()) -> Process SpawnRef spawnAsync nid proc = do@@ -486,41 +686,41 @@  -- | Monitor a node (asynchronous) monitorNode :: NodeId -> Process MonitorRef-monitorNode = +monitorNode =   monitor' . NodeIdentifier  -- | Monitor a typed channel (asynchronous) monitorPort :: forall a. Serializable a => SendPort a -> Process MonitorRef-monitorPort (SendPort cid) = -  monitor' (SendPortIdentifier cid) +monitorPort (SendPort cid) =+  monitor' (SendPortIdentifier cid)  -- | Remove a monitor (asynchronous) unmonitorAsync :: MonitorRef -> Process ()-unmonitorAsync = +unmonitorAsync =   sendCtrlMsg Nothing . Unmonitor  -- | Link to a node (asynchronous) linkNode :: NodeId -> Process ()-linkNode = link' . NodeIdentifier +linkNode = link' . NodeIdentifier  -- | Link to a channel (asynchronous) linkPort :: SendPort a -> Process ()-linkPort (SendPort cid) = +linkPort (SendPort cid) =   link' (SendPortIdentifier cid)  -- | Remove a process link (asynchronous) unlinkAsync :: ProcessId -> Process ()-unlinkAsync = +unlinkAsync =   sendCtrlMsg Nothing . Unlink . ProcessIdentifier  -- | Remove a node link (asynchronous) unlinkNodeAsync :: NodeId -> Process ()-unlinkNodeAsync = +unlinkNodeAsync =   sendCtrlMsg Nothing . Unlink . NodeIdentifier  -- | Remove a channel (send port) link (asynchronous) unlinkPortAsync :: SendPort a -> Process ()-unlinkPortAsync (SendPort cid) = +unlinkPortAsync (SendPort cid) =   sendCtrlMsg Nothing . Unlink $ SendPortIdentifier cid  --------------------------------------------------------------------------------@@ -573,12 +773,12 @@ -- -- See comments in 'whereisRemoteAsync' registerRemoteAsync :: NodeId -> String -> ProcessId -> Process ()-registerRemoteAsync nid label pid = -  sendCtrlMsg (Just nid) (Register label nid (Just pid) False) +registerRemoteAsync nid label pid =+  sendCtrlMsg (Just nid) (Register label nid (Just pid) False)  reregisterRemoteAsync :: NodeId -> String -> ProcessId -> Process ()-reregisterRemoteAsync nid label pid = -  sendCtrlMsg (Just nid) (Register label nid (Just pid) True) +reregisterRemoteAsync nid label pid =+  sendCtrlMsg (Just nid) (Register label nid (Just pid) True)  -- | Remove a process from the local registry (asynchronous). -- This version will wait until a response is gotten from the@@ -594,12 +794,12 @@ -- | Deal with the result from an attempted registration or unregistration -- by throwing an exception if necessary handleRegistrationReply :: String -> Bool -> Process ()-handleRegistrationReply label ok = +handleRegistrationReply label ok =   when (not ok) $      liftIO $ throwIO $ ProcessRegistrationException label  -- | Remove a process from a remote registry (asynchronous).--- +-- -- Reply wil come in the form of a 'RegisterReply' message -- -- See comments in 'whereisRemoteAsync'@@ -617,25 +817,25 @@  -- | Query a remote process registry (asynchronous) ----- Reply will come in the form of a 'WhereIsReply' message. +-- Reply will come in the form of a 'WhereIsReply' message. -- -- There is currently no synchronous version of 'whereisRemoteAsync': if -- you implement one yourself, be sure to take into account that the remote -- node might die or get disconnect before it can respond (i.e. you should--- use 'monitorNode' and take appropriate action when you receive a +-- use 'monitorNode' and take appropriate action when you receive a -- 'NodeMonitorNotification'). whereisRemoteAsync :: NodeId -> String -> Process ()-whereisRemoteAsync nid label = +whereisRemoteAsync nid label =   sendCtrlMsg (Just nid) (WhereIs label) --- | Named send to a process in the local registry (asynchronous) +-- | Named send to a process in the local registry (asynchronous) nsend :: Serializable a => String -> a -> Process ()-nsend label msg = +nsend label msg =   sendCtrlMsg Nothing (NamedSend label (createMessage msg))  -- | Named send to a process in a remote registry (asynchronous) nsendRemote :: Serializable a => NodeId -> String -> a -> Process ()-nsendRemote nid label msg = +nsendRemote nid label msg =   sendCtrlMsg (Just nid) (NamedSend label (createMessage msg))  --------------------------------------------------------------------------------@@ -653,9 +853,9 @@ -- | Resolve a closure unClosure :: Typeable a => Closure a -> Process a unClosure closure = do-  rtable <- remoteTable . processNode <$> ask +  rtable <- remoteTable . processNode <$> ask   case Static.unclosure rtable closure of-    Left err -> fail $ "Could not resolve closure: " ++ err +    Left err -> fail $ "Could not resolve closure: " ++ err     Right x  -> return x  --------------------------------------------------------------------------------@@ -666,11 +866,11 @@ -- message passing. However, when network connections get disrupted this -- illusion cannot always be maintained. Once a network connection breaks (even -- temporarily) no further communication on that connection will be possible.--- For example, if process A sends a message to process B, and A is then +-- For example, if process A sends a message to process B, and A is then -- notified (by monitor notification) that it got disconnected from B, A will--- not be able to send any further messages to B, /unless/ A explicitly +-- not be able to send any further messages to B, /unless/ A explicitly -- indicates that it is acceptable to attempt to reconnect to B using the--- Cloud Haskell 'reconnect' primitive. +-- Cloud Haskell 'reconnect' primitive. -- -- Importantly, when A calls 'reconnect' it acknowledges that some messages to -- B might have been lost. For instance, if A sends messages m1 and m2 to B,@@ -690,22 +890,65 @@  -- | Reconnect to a sendport. See 'reconnect' for more information. reconnectPort :: SendPort a -> Process ()-reconnectPort them = do +reconnectPort them = do   us <- getSelfPid   node <- processNode <$> ask   liftIO $ disconnect node (ProcessIdentifier us) (SendPortIdentifier (sendPortId them))  --------------------------------------------------------------------------------+-- Debugging/Tracing                                                          --+--------------------------------------------------------------------------------++-- | 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++-------------------------------------------------------------------------------- -- Auxiliary functions                                                        -- --------------------------------------------------------------------------------  getMonitorRefFor :: Identifier -> Process MonitorRef getMonitorRefFor ident = do   proc <- ask-  liftIO $ modifyMVar (processState proc) $ \st -> do -    let counter = st ^. monitorCounter +  liftIO $ modifyMVar (processState proc) $ \st -> do+    let counter = st ^. monitorCounter     return ( monitorCounter ^: (+ 1) $ st-           , MonitorRef ident counter +           , MonitorRef ident counter            )  getSpawnRef :: Process SpawnRef@@ -720,7 +963,7 @@ -- | Monitor a process/node/channel monitor' :: Identifier -> Process MonitorRef monitor' ident = do-  monitorRef <- getMonitorRefFor ident +  monitorRef <- getMonitorRefFor ident   sendCtrlMsg Nothing $ Monitor monitorRef   return monitorRef @@ -730,17 +973,16 @@  -- Send a control message sendCtrlMsg :: Maybe NodeId  -- ^ Nothing for the local node-            -> ProcessSignal -- ^ Message to send +            -> ProcessSignal -- ^ Message to send             -> Process ()-sendCtrlMsg mNid signal = do            +sendCtrlMsg mNid signal = do   proc <- ask-  let msg = NCMsg { ctrlMsgSender = ProcessIdentifier (processId proc) +  let msg = NCMsg { ctrlMsgSender = ProcessIdentifier (processId proc)                   , ctrlMsgSignal = signal                   }   case mNid of     Nothing -> do-      ctrlChan <- localCtrlChan . processNode <$> ask -      liftIO $ writeChan ctrlChan msg +      liftIO $ writeChan (localCtrlChan (processNode proc)) msg     Just nid ->       liftIO $ sendBinary (processNode proc)                           (ProcessIdentifier (processId proc))

@@ -1,4 +1,4 @@-module Control.Distributed.Process.Internal.StrictContainerAccessors +module Control.Distributed.Process.Internal.StrictContainerAccessors   ( mapMaybe   , mapDefault   ) where

@@ -1,25 +1,27 @@ -- | Spine and element strict list-module Control.Distributed.Process.Internal.StrictList -  ( StrictList(Cons, Nil)-  , length -  , reverse-  , reverse'+module Control.Distributed.Process.Internal.StrictList+  ( StrictList(..)+  , append+  , foldr   ) where -import Prelude hiding (length, reverse)+import Prelude hiding (length, reverse, foldr)  -- | Strict list-data StrictList a = Cons !a !(StrictList a) | Nil--length :: StrictList a -> Int-length Nil         = 0-length (Cons _ xs) = 1 + length xs+data StrictList a+   = Cons !a !(StrictList a)+   | Nil+   | Snoc !(StrictList a) !a+   | Append !(StrictList a) !(StrictList a) --- | Reverse a strict list-reverse :: StrictList a -> StrictList a -reverse xs = reverse' xs Nil+append :: StrictList a -> StrictList a -> StrictList a+append Nil l = l+append l Nil = l+append l1 l2 = l1 `Append` l2 --- | @reverseStrict' xs ys@ is 'reverse xs ++ ys' if they were lists-reverse' :: StrictList a -> StrictList a -> StrictList a -reverse' Nil         ys = ys-reverse' (Cons x xs) ys = reverse' xs (Cons x ys)+foldr :: (a -> b -> b) -> b -> StrictList a -> b+foldr f c0 xs0 = go xs0 c0+  where go Nil            c = c+        go (Cons x xs)    c = f x (go xs c)+        go (Snoc xs x)    c = go xs (f x c)+        go (Append xs ys) c = go xs (go ys c)

@@ -1,6 +1,6 @@ -- | Like Control.Concurrent.MVar.Strict but reduce to HNF, not NF {-# LANGUAGE MagicHash, UnboxedTuples #-}-module Control.Distributed.Process.Internal.StrictMVar +module Control.Distributed.Process.Internal.StrictMVar   ( StrictMVar(StrictMVar)   , newEmptyMVar   , newMVar@@ -15,7 +15,7 @@ import Control.Applicative ((<$>)) import Control.Monad ((>=>)) import Control.Exception (evaluate)-import qualified Control.Concurrent.MVar as MVar +import qualified Control.Concurrent.MVar as MVar   ( MVar   , newEmptyMVar   , newMVar@@ -26,7 +26,7 @@   , modifyMVar   ) import GHC.MVar (MVar(MVar))-import GHC.IO (IO(IO)) +import GHC.IO (IO(IO)) import GHC.Prim (mkWeak#) import GHC.Weak (Weak(Weak)) @@ -48,7 +48,7 @@ withMVar (StrictMVar v) = MVar.withMVar v  modifyMVar_ :: StrictMVar a -> (a -> IO a) -> IO ()-modifyMVar_ (StrictMVar v) f = MVar.modifyMVar_ v (f >=> evaluate) +modifyMVar_ (StrictMVar v) f = MVar.modifyMVar_ v (f >=> evaluate)  modifyMVar :: StrictMVar a -> (a -> IO (a, b)) -> IO b modifyMVar (StrictMVar v) f = MVar.modifyMVar v (f >=> evaluateFst)

@@ -0,0 +1,122 @@+-- | 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,10 +1,10 @@ -- | Types used throughout the Cloud Haskell framework ----- We collect all types used internally in a single module because +-- We collect all types used internally in a single module because -- many of these data types are mutually recursive and cannot be split across -- modules. module Control.Distributed.Process.Internal.Types-  ( -- * Node and process identifiers +  ( -- * Node and process identifiers     NodeId(..)   , LocalProcessId(..)   , ProcessId(..)@@ -14,6 +14,7 @@   , nullProcessId     -- * Local nodes and processes   , LocalNode(..)+  , Tracer(..)   , LocalNodeState(..)   , LocalProcess(..)   , LocalProcessState(..)@@ -21,21 +22,22 @@   , runLocalProcess   , ImplicitReconnect(..)     -- * Typed channels-  , LocalSendPortId +  , LocalSendPortId   , SendPortId(..)   , TypedChannel(..)   , SendPort(..)   , ReceivePort(..)-    -- * Messages +    -- * Messages   , Message(..)   , createMessage   , messageToPayload   , payloadToMessage-    -- * Node controller user-visible data types +    -- * Node controller user-visible data types   , MonitorRef(..)   , ProcessMonitorNotification(..)   , NodeMonitorNotification(..)   , PortMonitorNotification(..)+  , ProcessExitException(..)   , ProcessLinkException(..)   , NodeLinkException(..)   , PortLinkException(..)@@ -49,7 +51,9 @@   , DidSpawn(..)   , WhereIsReply(..)   , RegisterReply(..)-    -- * Node controller internal data types +  , ProcessInfo(..)+  , ProcessInfoNone(..)+    -- * Node controller internal data types   , NCMsg(..)   , ProcessSignal(..)     -- * Accessors@@ -64,6 +68,8 @@   , channelCounter   , typedChannels   , typedChannelWithId+    -- * Utilities+  , forever'   ) where  import System.Mem.Weak (Weak)@@ -72,7 +78,7 @@ import Data.Typeable (Typeable) 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 +import qualified Data.ByteString.Lazy as BSL   ( ByteString   , toChunks   , splitAt@@ -85,11 +91,12 @@ import Control.Concurrent (ThreadId) import Control.Concurrent.Chan (Chan) import Control.Concurrent.STM (STM)+import qualified Control.Concurrent.STM as STM (TQueue) import qualified Network.Transport as NT (EndPoint, EndPointAddress, Connection) import Control.Applicative (Applicative, Alternative, (<$>), (<*>)) import Control.Monad.Reader (MonadReader(..), ReaderT, runReaderT) import Control.Monad.IO.Class (MonadIO)-import Control.Distributed.Process.Serializable +import Control.Distributed.Process.Serializable   ( Fingerprint   , Serializable   , fingerprint@@ -103,43 +110,44 @@ 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)  -------------------------------------------------------------------------------- -- Node and process identifiers                                               -- -------------------------------------------------------------------------------- --- | Node identifier +-- | Node identifier newtype NodeId = NodeId { nodeAddress :: NT.EndPointAddress }   deriving (Eq, Ord, Binary, Typeable)  instance Show NodeId where-  show (NodeId addr) = "nid://" ++ show addr +  show (NodeId addr) = "nid://" ++ show addr  -- | A local process ID consists of a seed which distinguishes processes from -- different instances of the same local node and a counter-data LocalProcessId = LocalProcessId +data LocalProcessId = LocalProcessId   { lpidUnique  :: {-# UNPACK #-} !Int32   , lpidCounter :: {-# UNPACK #-} !Int32   }   deriving (Eq, Ord, Typeable, Show)  -- | Process identifier-data ProcessId = ProcessId +data ProcessId = ProcessId   { -- | The ID of the node the process is running on     processNodeId  :: !NodeId     -- | Node-local identifier for the process-  , processLocalId :: {-# UNPACK #-} !LocalProcessId +  , processLocalId :: {-# UNPACK #-} !LocalProcessId   }   deriving (Eq, Ord, Typeable)  instance Show ProcessId where-  show (ProcessId (NodeId addr) (LocalProcessId _ lid)) +  show (ProcessId (NodeId addr) (LocalProcessId _ lid))     = "pid://" ++ show addr ++ ":" ++ show lid --- | Union of all kinds of identifiers -data Identifier = +-- | Union of all kinds of identifiers+data Identifier =     NodeIdentifier !NodeId-  | ProcessIdentifier !ProcessId +  | ProcessIdentifier !ProcessId   | SendPortIdentifier !SendPortId   deriving (Eq, Ord) @@ -161,7 +169,7 @@ firstNonReservedProcessId = 1  nullProcessId :: NodeId -> ProcessId-nullProcessId nid = +nullProcessId nid =   ProcessId { processNodeId  = nid             , processLocalId = LocalProcessId { lpidUnique  = 0                                               , lpidCounter = 0@@ -172,40 +180,49 @@ -- 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+ -- | Local nodes-data LocalNode = LocalNode +data LocalNode = LocalNode   { -- | 'NodeId' of the node-    localNodeId :: !NodeId-    -- | The network endpoint associated with this node -  , localEndPoint :: !NT.EndPoint -    -- | Local node state -  , localState :: !(StrictMVar LocalNodeState)+    localNodeId     :: !NodeId+    -- | The network endpoint associated with this node+  , localEndPoint   :: !NT.EndPoint+    -- | Local node state+  , localState      :: !(StrictMVar LocalNodeState)     -- | Channel for the node controller-  , localCtrlChan :: !(Chan NCMsg)+  , localCtrlChan   :: !(Chan NCMsg)+    -- | Current active system debug/trace log+  , localTracer     :: !Tracer     -- | Runtime lookup table for supporting closures     -- TODO: this should be part of the CH state, not the local endpoint state-  , remoteTable :: !RemoteTable +  , remoteTable     :: !RemoteTable   }  data ImplicitReconnect = WithImplicitReconnect | NoImplicitReconnect   deriving (Eq, Show)  -- | Local node state-data LocalNodeState = LocalNodeState +data LocalNodeState = LocalNodeState   { -- | Processes running on this node     _localProcesses   :: !(Map LocalProcessId LocalProcess)-    -- | Counter to assign PIDs +    -- | Counter to assign PIDs   , _localPidCounter  :: !Int32     -- | The 'unique' value used to create PIDs (so that processes on     -- restarted nodes have new PIDs)   , _localPidUnique   :: !Int32     -- | Outgoing connections-  , _localConnections :: !(Map (Identifier, Identifier) +  , _localConnections :: !(Map (Identifier, Identifier)                                (NT.Connection, ImplicitReconnect))   }  -- | Processes running on our local node-data LocalProcess = LocalProcess +data LocalProcess = LocalProcess   { processQueue  :: !(CQueue Message)   , processWeakQ  :: !(Weak (CQueue Message))   , processId     :: !ProcessId@@ -214,7 +231,7 @@   , processNode   :: !LocalNode   } --- | Deconstructor for 'Process' (not exported to the public API) +-- | Deconstructor for 'Process' (not exported to the public API) runLocalProcess :: LocalProcess -> Process a -> IO a runLocalProcess lproc proc = runReaderT (unProcess proc) lproc @@ -227,8 +244,8 @@   }  -- | The Cloud Haskell 'Process' type-newtype Process a = Process { -    unProcess :: ReaderT LocalProcess IO a +newtype Process a = Process {+    unProcess :: ReaderT LocalProcess IO a   }   deriving (Functor, Monad, MonadIO, MonadReader LocalProcess, Typeable, Applicative) @@ -251,15 +268,15 @@   deriving (Eq, Ord)  instance Show SendPortId where-  show (SendPortId (ProcessId (NodeId addr) (LocalProcessId _ plid)) clid)  +  show (SendPortId (ProcessId (NodeId addr) (LocalProcessId _ plid)) clid)     = "cid://" ++ show addr ++ ":" ++ show plid ++ ":" ++ show clid  data TypedChannel = forall a. Serializable a => TypedChannel (Weak (TQueue a))  -- | The send send of a typed channel (serializable)-newtype SendPort a = SendPort { +newtype SendPort a = SendPort {     -- | The (unique) ID of this send port-    sendPortId :: SendPortId +    sendPortId :: SendPortId   }   deriving (Typeable, Binary, Show, Eq, Ord) @@ -271,13 +288,13 @@   deriving (Typeable, Functor, Applicative, Alternative, Monad)  {--data ReceivePort a = +data ReceivePort a =     -- | A single receive port     ReceivePortSingle (TQueue a)-    -- | A left-biased combination of receive ports -  | ReceivePortBiased [ReceivePort a] +    -- | A left-biased combination of receive ports+  | ReceivePortBiased [ReceivePort a]     -- | A round-robin combination of receive ports-  | ReceivePortRR (TVar [ReceivePort a]) +  | ReceivePortRR (TVar [ReceivePort a])   deriving Typeable -} @@ -286,13 +303,13 @@ --------------------------------------------------------------------------------  -- | Messages consist of their typeRep fingerprint and their encoding-data Message = Message -  { messageFingerprint :: !Fingerprint +data Message = Message+  { messageFingerprint :: !Fingerprint   , messageEncoding    :: !BSL.ByteString   }  instance Show Message where-  show (Message fp enc) = show enc ++ " :: " ++ showFingerprint fp [] +  show (Message fp enc) = show enc ++ " :: " ++ showFingerprint fp []  -- | Turn any serialiable term into a message createMessage :: Serializable a => a -> Message@@ -304,26 +321,26 @@  -- | Deserialize a message payloadToMessage :: [BSS.ByteString] -> Message-payloadToMessage payload = Message fp (copy msg) +payloadToMessage payload = Message fp (copy msg)   where     encFp :: BSL.ByteString     msg   :: BSL.ByteString-    (encFp, msg) = BSL.splitAt (fromIntegral sizeOfFingerprint) -                 $ BSL.fromChunks payload +    (encFp, msg) = BSL.splitAt (fromIntegral sizeOfFingerprint)+                 $ BSL.fromChunks payload      fp :: Fingerprint     fp = decodeFingerprint . BSS.concat . BSL.toChunks $ encFp      copy :: BSL.ByteString -> BSL.ByteString     copy (BSL.Chunk bs BSL.Empty) = BSL.Chunk (BSS.copy bs) BSL.Empty-    copy bsl = BSL.fromChunks . return . BSS.concat . BSL.toChunks $ bsl +    copy bsl = BSL.fromChunks . return . BSS.concat . BSL.toChunks $ bsl  -------------------------------------------------------------------------------- -- Node controller user-visible data types                                    -- -------------------------------------------------------------------------------- --- | MonitorRef is opaque for regular Cloud Haskell processes -data MonitorRef = MonitorRef +-- | MonitorRef is opaque for regular Cloud Haskell processes+data MonitorRef = MonitorRef   { -- | ID of the entity to be monitored     monitorRefIdent   :: !Identifier     -- | Unique to distinguish multiple monitor requests by the same process@@ -332,49 +349,58 @@   deriving (Eq, Ord, Show)  -- | Message sent by process monitors-data ProcessMonitorNotification = +data ProcessMonitorNotification =     ProcessMonitorNotification !MonitorRef !ProcessId !DiedReason   deriving (Typeable, Show)  -- | Message sent by node monitors-data NodeMonitorNotification = +data NodeMonitorNotification =     NodeMonitorNotification !MonitorRef !NodeId !DiedReason   deriving (Typeable, Show)  -- | Message sent by channel (port) monitors-data PortMonitorNotification = +data PortMonitorNotification =     PortMonitorNotification !MonitorRef !SendPortId !DiedReason   deriving (Typeable, Show)  -- | Exceptions thrown when a linked process dies-data ProcessLinkException = +data ProcessLinkException =     ProcessLinkException !ProcessId !DiedReason   deriving (Typeable, Show)  -- | Exception thrown when a linked node dies-data NodeLinkException = +data NodeLinkException =     NodeLinkException !NodeId !DiedReason   deriving (Typeable, Show)  -- | Exception thrown when a linked channel (port) dies-data PortLinkException = +data PortLinkException =     PortLinkException !SendPortId !DiedReason   deriving (Typeable, Show)  -- | Exception thrown when a process attempts to register -- a process under an already-registered name or to -- unregister a name that hasn't been registered-data ProcessRegistrationException = +data ProcessRegistrationException =     ProcessRegistrationException !String   deriving (Typeable, Show) +-- | Internal exception thrown indirectly by 'exit'+data ProcessExitException =+    ProcessExitException !ProcessId !Message+  deriving Typeable++instance Exception ProcessExitException+instance Show ProcessExitException where+  show (ProcessExitException pid _) = "exit-from=" ++ (show pid)+ instance Exception ProcessLinkException instance Exception NodeLinkException instance Exception PortLinkException instance Exception ProcessRegistrationException  -- | Why did a process die?-data DiedReason = +data DiedReason =     -- | Normal termination     DiedNormal     -- | The process exited with an exception@@ -384,7 +410,7 @@   | DiedDisconnect     -- | The process node died   | DiedNodeDown-    -- | Invalid (process/node/channel) identifier +    -- | Invalid (process/node/channel) identifier   | DiedUnknownId   deriving (Show, Eq) @@ -401,7 +427,7 @@   deriving (Typeable, Binary)  -- | (Asynchronous) reply from unlinkPort-newtype DidUnlinkPort = DidUnlinkPort SendPortId +newtype DidUnlinkPort = DidUnlinkPort SendPortId   deriving (Typeable, Binary)  -- | 'SpawnRef' are used to return pids of spawned processes@@ -420,28 +446,43 @@ data RegisterReply = RegisterReply String Bool   deriving (Show, Typeable) +-- | Provide information about a running process+data ProcessInfo = ProcessInfo {+    infoNode               :: NodeId+  , infoRegisteredNames    :: [String]+  , infoMessageQueueLength :: Maybe Int+  , infoMonitors           :: [(ProcessId, MonitorRef)]+  , infoLinks              :: [ProcessId]+  } deriving (Show, Eq, Typeable)++data ProcessInfoNone = ProcessInfoNone DiedReason+    deriving (Show, Typeable)+ -------------------------------------------------------------------------------- -- Node controller internal data types                                        -- --------------------------------------------------------------------------------  -- | Messages to the node controller-data NCMsg = NCMsg -  { ctrlMsgSender :: !Identifier +data NCMsg = NCMsg+  { ctrlMsgSender :: !Identifier   , ctrlMsgSignal :: !ProcessSignal   }   deriving Show  -- | Signals to the node controller (see 'NCMsg') data ProcessSignal =-    Link !Identifier -  | Unlink !Identifier +    Link !Identifier+  | Unlink !Identifier   | Monitor !MonitorRef   | Unmonitor !MonitorRef   | Died Identifier !DiedReason-  | Spawn !(Closure (Process ())) !SpawnRef +  | Spawn !(Closure (Process ())) !SpawnRef   | WhereIs !String   | Register !String !NodeId !(Maybe ProcessId) !Bool -- Use 'Nothing' to unregister, use True to force reregister   | NamedSend !String !Message+  | Kill !ProcessId !String+  | Exit !ProcessId !Message+  | GetInfo !ProcessId   deriving Show  --------------------------------------------------------------------------------@@ -480,29 +521,35 @@   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 (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 (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   get = do     header <- getWord8     case header of-      0 -> Link <$> get-      1 -> Unlink <$> get-      2 -> Monitor <$> get-      3 -> Unmonitor <$> get-      4 -> Died <$> get <*> get-      5 -> Spawn <$> get <*> get-      6 -> WhereIs <$> get-      7 -> Register <$> get <*> get <*> get <*> get-      8 -> NamedSend <$> get <*> (payloadToMessage <$> get)+      0  -> Link <$> get+      1  -> Unlink <$> get+      2  -> Monitor <$> get+      3  -> Unmonitor <$> get+      4  -> Died <$> get <*> get+      5  -> Spawn <$> get <*> get+      6  -> WhereIs <$> get+      7  -> Register <$> get <*> get <*> get <*> get+      8  -> NamedSend <$> get <*> (payloadToMessage <$> get)+      9  -> Kill <$> get <*> get+      10 -> Exit <$> get <*> (payloadToMessage <$> get)+      30 -> GetInfo <$> get       _ -> fail "ProcessSignal.get: invalid"  instance Binary DiedReason where   put DiedNormal        = putWord8 0-  put (DiedException e) = putWord8 1 >> put e +  put (DiedException e) = putWord8 1 >> put e   put DiedDisconnect    = putWord8 2   put DiedNodeDown      = putWord8 3   put DiedUnknownId     = putWord8 4@@ -513,7 +560,7 @@       1 -> DiedException <$> get       2 -> return DiedDisconnect       3 -> return DiedNodeDown-      4 -> return DiedUnknownId +      4 -> return DiedUnknownId       _ -> fail "DiedReason.get: invalid"  instance Binary DidSpawn where@@ -522,18 +569,18 @@  instance Binary SendPortId where   put cid = put (sendPortProcessId cid) >> put (sendPortLocalId cid)-  get = SendPortId <$> get <*> get +  get = SendPortId <$> get <*> get  instance Binary Identifier where   put (ProcessIdentifier pid)  = putWord8 0 >> put pid   put (NodeIdentifier nid)     = putWord8 1 >> put nid   put (SendPortIdentifier cid) = putWord8 2 >> put cid   get = do-    header <- getWord8 +    header <- getWord8     case header of       0 -> ProcessIdentifier <$> get       1 -> NodeIdentifier <$> get-      2 -> SendPortIdentifier <$> get +      2 -> SendPortIdentifier <$> get       _ -> fail "Identifier.get: invalid"  instance Binary WhereIsReply where@@ -541,9 +588,21 @@   get = WhereIsReply <$> get <*> get  instance Binary RegisterReply where-  put (RegisterReply label ok) = put label >> put ok +  put (RegisterReply label ok) = put label >> put ok   get = RegisterReply <$> get <*> get +instance Binary ProcessInfo where+  get = ProcessInfo <$> get <*> get <*> get <*> get <*> get+  put pInfo = put (infoNode pInfo)+           >> put (infoRegisteredNames pInfo)+           >> put (infoMessageQueueLength pInfo)+           >> put (infoMonitors pInfo)+           >> put (infoLinks pInfo)++instance Binary ProcessInfoNone where+  get = ProcessInfoNone <$> get+  put (ProcessInfoNone r) = put r+ -------------------------------------------------------------------------------- -- Accessors                                                                  -- --------------------------------------------------------------------------------@@ -580,3 +639,13 @@  typedChannelWithId :: LocalSendPortId -> Accessor LocalProcessState (Maybe TypedChannel) typedChannelWithId cid = typedChannels >>> DAC.mapMaybe cid++--------------------------------------------------------------------------------+-- Utilities                                                                  --+--------------------------------------------------------------------------------++-- Like 'Control.Monad.forever' but sans space leak+{-# INLINE forever' #-}+forever' :: Monad m => m a -> m b+forever' a = let a' = a >> a' in a'+

@@ -21,7 +21,7 @@ import Prelude hiding (read) import GHC.Conc import Data.Typeable (Typeable)-import GHC.IO (IO(IO)) +import GHC.IO (IO(IO)) import GHC.Prim (mkWeak#) import GHC.Weak (Weak(Weak)) @@ -98,4 +98,4 @@  mkWeakTQueue :: TQueue a -> IO () -> IO (Weak (TQueue a)) mkWeakTQueue q@(TQueue _read (TVar write#)) f = IO $ \s ->-  case mkWeak# write# q f s of (# s', w #) -> (# s', Weak w #) +  case mkWeak# write# q f s of (# s', w #) -> (# s', Weak w #)

@@ -1,8 +1,6 @@ -- | Local nodes ----- TODO: Calls to 'sendBinary' and co by the node controller may stall the--- node controller.-module Control.Distributed.Process.Node +module Control.Distributed.Process.Node   ( LocalNode   , newLocalNode   , closeLocalNode@@ -12,6 +10,8 @@   , localNodeId   ) where +-- TODO: Calls to 'sendBinary' and co (by the NC) may stall the node controller.+ #if ! MIN_VERSION_base(4,6,0) import Prelude hiding (catch) #endif@@ -21,7 +21,7 @@ import qualified Data.ByteString.Lazy as BSL (fromChunks) import Data.Binary (decode) import Data.Map (Map)-import qualified Data.Map as Map +import qualified Data.Map as Map   ( empty   , toList   , fromList@@ -29,9 +29,16 @@   , partitionWithKey   , elems   , filterWithKey+  , foldlWithKey   ) import Data.Set (Set)-import qualified Data.Set as Set (empty, insert, delete, member)+import qualified Data.Set as Set+  ( empty+  , insert+  , delete+  , member+  , toList+  ) import Data.Foldable (forM_) import Data.Maybe (isJust, isNothing, catMaybes) import Data.Typeable (Typeable)@@ -46,7 +53,7 @@ import qualified Control.Exception as Exception (catch) import Control.Concurrent (forkIO) import Control.Distributed.Process.Internal.StrictMVar-  ( newMVar +  ( newMVar   , withMVar   , modifyMVar   , modifyMVar_@@ -55,14 +62,16 @@   , takeMVar   ) import Control.Concurrent.Chan (newChan, writeChan, readChan)-import Control.Concurrent.STM (atomically)-import Control.Distributed.Process.Internal.CQueue +import Control.Concurrent.STM+  ( atomically+  )+import Control.Distributed.Process.Internal.CQueue   ( CQueue   , enqueue   , newCQueue   , mkWeakCQueue   )-import qualified Network.Transport as NT +import qualified Network.Transport as NT   ( Transport   , EndPoint   , newEndPoint@@ -81,15 +90,16 @@ import Data.Accessor (Accessor, accessor, (^.), (^=), (^:)) import System.Random (randomIO) import Control.Distributed.Static (RemoteTable, Closure)-import qualified Control.Distributed.Static as Static +import qualified Control.Distributed.Static as Static   ( unclosure   , initRemoteTable   )-import Control.Distributed.Process.Internal.Types +import Control.Distributed.Process.Internal.Types   ( NodeId(..)   , LocalProcessId(..)   , ProcessId(..)   , LocalNode(..)+  , Tracer(InactiveTracer)   , LocalNodeState(..)   , LocalProcess(..)   , LocalProcessState(..)@@ -101,10 +111,12 @@   , localPidUnique   , localProcessWithId   , localConnections+  , forever'   , MonitorRef(..)   , ProcessMonitorNotification(..)   , NodeMonitorNotification(..)   , PortMonitorNotification(..)+  , ProcessExitException(..)   , ProcessLinkException(..)   , NodeLinkException(..)   , PortLinkException(..)@@ -118,6 +130,8 @@   , TypedChannel(..)   , Identifier(..)   , nodeOf+  , ProcessInfo(..)+  , ProcessInfoNone(..)   , SendPortId(..)   , typedChannelWithId   , RegisterReply(..)@@ -129,15 +143,28 @@   , firstNonReservedProcessId   , ImplicitReconnect(WithImplicitReconnect, NoImplicitReconnect)   )+import Control.Distributed.Process.Internal.Trace+  ( TraceArg(..)+  , traceFormat+  , startTracing+  , stopTracer+  ) import Control.Distributed.Process.Serializable (Serializable) import Control.Distributed.Process.Internal.Messaging   ( sendBinary   , sendMessage   , sendPayload-  , closeImplicitReconnections +  , closeImplicitReconnections   , impliesDeathOf   )-import Control.Distributed.Process.Internal.Primitives (expect, register, finally)+import Control.Distributed.Process.Internal.Primitives+  ( register+  , finally+  , receiveWait+  , match+  , sendChan+  )+import Control.Distributed.Process.Internal.Types (SendPort) import qualified Control.Distributed.Process.Internal.Closure.BuiltIn as BuiltIn (remoteTable) import Control.Distributed.Process.Internal.WeakTQueue (writeTQueue) import qualified Control.Distributed.Process.Internal.StrictContainerAccessors as DAC@@ -150,27 +177,27 @@ --------------------------------------------------------------------------------  initRemoteTable :: RemoteTable-initRemoteTable = BuiltIn.remoteTable Static.initRemoteTable +initRemoteTable = BuiltIn.remoteTable Static.initRemoteTable --- | Initialize a new local node. +-- | Initialize a new local node. newLocalNode :: NT.Transport -> RemoteTable -> IO LocalNode newLocalNode transport rtable = do     mEndPoint <- NT.newEndPoint transport     case mEndPoint of       Left ex -> throwIO ex       Right endPoint -> do-        localNode <- createBareLocalNode endPoint rtable +        localNode <- createBareLocalNode endPoint rtable         startServiceProcesses localNode         return localNode-    + -- | 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 +  state <- newMVar LocalNodeState     { _localProcesses   = Map.empty-    , _localPidCounter  = firstNonReservedProcessId -    , _localPidUnique   = unq +    , _localPidCounter  = firstNonReservedProcessId+    , _localPidUnique   = unq     , _localConnections = Map.empty     }   ctrlChan <- newChan@@ -178,25 +205,33 @@                        , localEndPoint = endPoint                        , localState    = state                        , localCtrlChan = ctrlChan+                       , localTracer   = InactiveTracer                        , remoteTable   = rtable                        }-  void . forkIO $ runNodeController node -  void . forkIO $ handleIncomingMessages node-  return node---- Like 'Control.Monad.forever' but sans space leak-{-# INLINE forever' #-}-forever' :: Monad m => m a -> m b-forever' a = let a' = a >> a' in a'+  tracedNode <- startTracing node+  void . forkIO $ runNodeController tracedNode+  void . forkIO $ handleIncomingMessages tracedNode+  return tracedNode --- | Start and register the service processes on a node +-- | Start and register the service processes on a node -- (for now, this is only the logger) startServiceProcesses :: LocalNode -> IO () startServiceProcesses node = do-  logger <- forkProcess node . forever' $ do-    (time, pid, string) <- expect :: Process (String, ProcessId, String)-    liftIO . hPutStrLn stderr $ time ++ " " ++ show pid ++ ": " ++ string +  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 ()+      ]  -- | Force-close a local node --@@ -221,7 +256,7 @@       let lpid  = LocalProcessId { lpidCounter = st ^. localPidCounter                                  , lpidUnique  = st ^. localPidUnique                                  }-      let pid   = ProcessId { processNodeId  = localNodeId node +      let pid   = ProcessId { processNodeId  = localNodeId node                             , processLocalId = lpid                             }       pst <- newMVar LocalProcessState { _monitorCounter = 0@@ -235,19 +270,20 @@         let lproc = LocalProcess { processQueue  = queue                                  , processWeakQ  = weakQueue                                  , processId     = pid-                                 , processState  = pst +                                 , processState  = pst                                  , processThread = tid                                  , processNode   = node                                  }         tid' <- forkIO $ do-          reason <- Exception.catch +          reason <- Exception.catch             (runLocalProcess lproc proc >> return DiedNormal)             (return . DiedException . (show :: SomeException -> String))+           -- [Unified: Table 4, rules termination and exiting]           modifyMVar_ (localState node) (cleanupProcess pid)-          writeChan (localCtrlChan node) NCMsg -            { ctrlMsgSender = ProcessIdentifier pid -            , ctrlMsgSignal = Died (ProcessIdentifier pid) reason +          writeChan (localCtrlChan node) NCMsg+            { ctrlMsgSender = ProcessIdentifier pid+            , ctrlMsgSignal = Died (ProcessIdentifier pid) reason             }         return (tid', lproc) @@ -264,12 +300,12 @@           return ( (localProcessWithId lpid ^= Just lproc)                  . (localPidCounter ^: (+ 1))                  $ st-                 , pid +                 , pid                  )      cleanupProcess :: ProcessId -> LocalNodeState -> IO LocalNodeState     cleanupProcess pid st = do-      let pid' = ProcessIdentifier pid +      let pid' = ProcessIdentifier pid       let (affected, unaffected) = Map.partitionWithKey (\(fr, _to) !_v -> impliesDeathOf pid' fr) (st ^. localConnections)       mapM_ (NT.close . fst) (Map.elems affected)       return $ (localProcessWithId (processLocalId pid) ^= Nothing)@@ -282,7 +318,7 @@  type IncomingConnection = (NT.EndPointAddress, IncomingTarget) -data IncomingTarget =  +data IncomingTarget =     Uninit   | ToProc (Weak (CQueue Message))   | ToChan TypedChannel@@ -305,20 +341,20 @@ incomingAt :: NT.ConnectionId -> Accessor ConnectionState (Maybe IncomingConnection) incomingAt cid = incoming >>> DAC.mapMaybe cid -incomingFrom :: NT.EndPointAddress -> Accessor ConnectionState (Set NT.ConnectionId) +incomingFrom :: NT.EndPointAddress -> Accessor ConnectionState (Set NT.ConnectionId) incomingFrom addr = aux >>> DAC.mapDefault Set.empty addr   where     aux = accessor _incomingFrom (\fr st -> st { _incomingFrom = fr })  handleIncomingMessages :: LocalNode -> IO ()-handleIncomingMessages node = go initConnectionState +handleIncomingMessages node = go initConnectionState   where-    go :: ConnectionState -> IO () +    go :: ConnectionState -> IO ()     go !st = do       event <- NT.receive endpoint       case event of         NT.ConnectionOpened cid rel theirAddr ->-          if rel == NT.ReliableOrdered +          if rel == NT.ReliableOrdered             then go ( (incomingAt cid ^= Just (theirAddr, Uninit))                     . (incomingFrom theirAddr ^: Set.insert cid)                     $ st@@ -328,33 +364,33 @@           case st ^. incomingAt cid of             Just (_, ToProc weakQueue) -> do               mQueue <- deRefWeak weakQueue-              forM_ mQueue $ \queue -> do +              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-              go st +              go st             Just (_, ToChan (TypedChannel chan')) -> do               mChan <- deRefWeak chan'               -- If mChan is Nothing, the process has given up the read end of               -- the channel and we simply ignore the incoming message-              forM_ mChan $ \chan -> atomically $  +              forM_ mChan $ \chan -> atomically $                 -- We make sure the message is fully decoded when it is enqueued                 writeTQueue chan $! decode (BSL.fromChunks payload)-              go st +              go st             Just (_, ToNode) -> do               let ctrlMsg = decode . BSL.fromChunks $ payload               writeChan ctrlChan $! ctrlMsg-              go st -            Just (src, Uninit) -> +              go st+            Just (src, Uninit) ->               case decode (BSL.fromChunks payload) of                 ProcessIdentifier pid -> do                   let lpid = processLocalId pid-                  mProc <- withMVar state $ return . (^. localProcessWithId lpid) +                  mProc <- withMVar state $ return . (^. localProcessWithId lpid)                   case mProc of-                    Just proc -> -                      go (incomingAt cid ^= Just (src, ToProc (processWeakQ proc)) $ st) -                    Nothing -> +                    Just proc ->+                      go (incomingAt cid ^= Just (src, ToProc (processWeakQ proc)) $ st)+                    Nothing ->                       invalidRequest cid st                 SendPortIdentifier chId -> do                   let lcid = sendPortLocalId chId@@ -364,7 +400,7 @@                     Just proc -> do                       mChannel <- withMVar (processState proc) $ return . (^. typedChannelWithId lcid)                       case mChannel of-                        Just channel -> +                        Just channel ->                           go (incomingAt cid ^= Just (src, ToChan channel) $ st)                         Nothing ->                           invalidRequest cid st@@ -373,50 +409,55 @@                 NodeIdentifier nid ->                   if nid == localNodeId node                     then go (incomingAt cid ^= Just (src, ToNode) $ st)-                    else invalidRequest cid st +                    else invalidRequest cid st             Nothing ->               invalidRequest cid st-        NT.ConnectionClosed cid -> +        NT.ConnectionClosed cid ->           case st ^. incomingAt cid of-            Nothing -> +            Nothing ->               invalidRequest cid st-            Just (src, _) -> +            Just (src, _) ->               go ( (incomingAt cid ^= Nothing)                  . (incomingFrom src ^: Set.delete cid)                  $ st                  )-        NT.ErrorEvent (NT.TransportError (NT.EventConnectionLost theirAddr) _) -> do +        NT.ErrorEvent (NT.TransportError (NT.EventConnectionLost theirAddr) _) -> do           -- [Unified table 9, rule node_disconnect]           let nid = NodeIdentifier $ NodeId theirAddr-          writeChan ctrlChan NCMsg +          writeChan ctrlChan NCMsg             { ctrlMsgSender = nid             , ctrlMsgSignal = Died nid DiedDisconnect             }           let notLost k = not (k `Set.member` (st ^. incomingFrom theirAddr))-          closeImplicitReconnections node nid +          closeImplicitReconnections node nid           go ( (incomingFrom theirAddr ^= Set.empty)              . (incoming ^: Map.filterWithKey (const . notLost))-             $ st +             $ st              )         NT.ErrorEvent (NT.TransportError NT.EventEndPointFailed str) ->-          fail $ "Cloud Haskell fatal error: end point failed: " ++ str +          fatal $ "Cloud Haskell fatal error: end point failed: " ++ str         NT.ErrorEvent (NT.TransportError NT.EventTransportFailed str) ->-          fail $ "Cloud Haskell fatal error: transport failed: " ++ str +          fatal $ "Cloud Haskell fatal error: transport failed: " ++ str         NT.EndPointClosed ->-          return ()+          stopTracer (localTracer node) >> return ()         NT.ReceivedMulticast _ _ ->           -- If we received a multicast message, something went horribly wrong           -- and we just give up-          fail "Cloud Haskell fatal error: received unexpected multicast"+          fatal "Cloud Haskell fatal error: received unexpected multicast" +    fatal :: String -> IO ()+    fatal msg = stopTracer (localTracer node) >> fail msg+     invalidRequest :: NT.ConnectionId -> ConnectionState -> IO ()-    invalidRequest cid st = +    invalidRequest cid st = do       -- TODO: We should treat this as a fatal error on the part of the remote       -- node. That is, we should report the remote node as having died, and we       -- should close incoming connections (this requires a Transport layer       -- extension).+      traceEventFmtIO node "" [(TraceStr "[network] invalid request: "),+                               (Trace cid)]       go ( incomingAt cid ^= Nothing-         $ st +         $ st          )      state    = localState node@@ -435,8 +476,8 @@ -- Internal data types                                                        -- -------------------------------------------------------------------------------- -data NCState = NCState -  {  -- Mapping from remote processes to linked local processes +data NCState = NCState+  {  -- Mapping from remote processes to linked local processes     _links    :: !(Map Identifier (Set ProcessId))      -- Mapping from remote processes to monitoring local processes   , _monitors :: !(Map Identifier (Set (ProcessId, MonitorRef)))@@ -455,26 +496,66 @@                       , _registeredOnNodes = Map.empty                       } +-- | Thrown in response to the user invoking 'kill' (see Primitives.hs). This+-- type is deliberately not exported so it cannot be caught explicitly.+data ProcessKillException =+    ProcessKillException !ProcessId !String+  deriving (Typeable)++instance Exception ProcessKillException+instance Show ProcessKillException where+  show (ProcessKillException pid reason) =+    "killed-by=" ++ show pid ++ ",reason=" ++ reason+ --------------------------------------------------------------------------------+-- Tracing/Debugging                                                          --+--------------------------------------------------------------------------------++-- [Issue #104]++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++traceEventFmtIO :: LocalNode+                -> String+                -> [TraceArg]+                -> IO ()+traceEventFmtIO node fmt args =+  withLocalTracer node $ \t -> traceFormat t fmt args++withLocalTracer :: LocalNode -> (Tracer -> IO ()) -> IO ()+withLocalTracer node act = act (localTracer node)++-------------------------------------------------------------------------------- -- Core functionality                                                         -- --------------------------------------------------------------------------------  -- [Unified: Table 7] nodeController :: NC () nodeController = do-  node <- ask +  node <- ask   forever' $ do     msg  <- liftIO $ readChan (localCtrlChan node) -    -- [Unified: Table 7, rule nc_forward] +    -- [Unified: Table 7, rule nc_forward]     case destNid (ctrlMsgSignal msg) of-      Just nid' | nid' /= localNodeId node -> +      Just nid' | nid' /= localNodeId node ->         liftIO $ sendBinary node                             (ctrlMsgSender msg)-                            (NodeIdentifier nid') +                            (NodeIdentifier nid')                             WithImplicitReconnect                             msg-      _ -> +      _ ->         return ()      case msg of@@ -483,7 +564,7 @@       NCMsg (ProcessIdentifier from) (Monitor ref) ->         ncEffectMonitor from (monitorRefIdent ref) (Just ref)       NCMsg (ProcessIdentifier from) (Unlink them) ->-        ncEffectUnlink from them +        ncEffectUnlink from them       NCMsg (ProcessIdentifier from) (Unmonitor ref) ->         ncEffectUnmonitor from ref       NCMsg _from (Died ident reason) ->@@ -496,27 +577,33 @@         ncEffectWhereIs from label       NCMsg from (NamedSend label msg') ->         ncEffectNamedSend from label 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       unexpected ->         error $ "nodeController: unexpected message " ++ show unexpected  -- [Unified: Table 10]-ncEffectMonitor :: ProcessId        -- ^ Who's watching? +ncEffectMonitor :: ProcessId        -- ^ Who's watching?                 -> Identifier       -- ^ Who's being watched?                 -> Maybe MonitorRef -- ^ 'Nothing' to link                 -> NC () ncEffectMonitor from them mRef = do-  node <- ask -  shouldLink <- -    if not (isLocal node them) +  node <- ask+  shouldLink <-+    if not (isLocal node them)       then return True       else isValidLocalIdentifier them   case (shouldLink, isLocal node (ProcessIdentifier from)) of     (True, _) ->  -- [Unified: first rule]       case mRef of         Just ref -> modify' $ monitorsFor them ^: Set.insert (from, ref)-        Nothing  -> modify' $ linksFor them ^: Set.insert from +        Nothing  -> modify' $ linksFor them ^: Set.insert from     (False, True) -> -- [Unified: second rule]-      notifyDied from them DiedUnknownId mRef +      notifyDied from them DiedUnknownId mRef     (False, False) -> -- [Unified: third rule]       -- TODO: this is the right sender according to the Unified semantics,       -- but perhaps having 'them' as the sender would make more sense@@ -525,7 +612,7 @@                           (NodeIdentifier $ localNodeId node)                           (NodeIdentifier $ processNodeId from)                           WithImplicitReconnect-        NCMsg  +        NCMsg           { ctrlMsgSender = NodeIdentifier (localNodeId node)           , ctrlMsgSignal = Died them DiedUnknownId           }@@ -533,22 +620,22 @@ -- [Unified: Table 11] ncEffectUnlink :: ProcessId -> Identifier -> NC () ncEffectUnlink from them = do-  node <- ask -  when (isLocal node (ProcessIdentifier from)) $ +  node <- ask+  when (isLocal node (ProcessIdentifier from)) $     case them of-      ProcessIdentifier pid -> -        postAsMessage from $ DidUnlinkProcess pid -      NodeIdentifier nid -> +      ProcessIdentifier pid ->+        postAsMessage from $ DidUnlinkProcess pid+      NodeIdentifier nid ->         postAsMessage from $ DidUnlinkNode nid-      SendPortIdentifier cid -> -        postAsMessage from $ DidUnlinkPort cid +      SendPortIdentifier cid ->+        postAsMessage from $ DidUnlinkPort cid   modify' $ linksFor them ^: Set.delete from  -- [Unified: Table 11] ncEffectUnmonitor :: ProcessId -> MonitorRef -> NC () ncEffectUnmonitor from ref = do-  node <- ask -  when (isLocal node (ProcessIdentifier from)) $ +  node <- ask+  when (isLocal node (ProcessIdentifier from)) $     postAsMessage from $ DidUnmonitor ref   modify' $ monitorsFor (monitorRefIdent ref) ^: Set.delete (from, ref) @@ -556,6 +643,7 @@ ncEffectDied :: Identifier -> DiedReason -> NC () ncEffectDied ident reason = do   node <- ask+  traceNotifyDied node ident reason   (affectedLinks, unaffectedLinks) <- gets (splitNotif ident . (^. links))   (affectedMons,  unaffectedMons)  <- gets (splitNotif ident . (^. monitors)) @@ -563,41 +651,40 @@    let localOnly = case ident of NodeIdentifier _ -> True ; _ -> False -  forM_ (Map.toList affectedLinks) $ \(them, uss) -> +  forM_ (Map.toList affectedLinks) $ \(them, uss) ->     forM_ uss $ \us ->-      when (localOnly <= isLocal node (ProcessIdentifier us)) $ +      when (localOnly <= isLocal node (ProcessIdentifier us)) $         notifyDied us them reason Nothing    forM_ (Map.toList affectedMons) $ \(them, refs) ->     forM_ refs $ \(us, ref) ->       when (localOnly <= isLocal node (ProcessIdentifier us)) $-        notifyDied us them reason (Just ref)   +        notifyDied us them reason (Just ref)    modify' $ (links ^= unaffectedLinks) . (monitors ^= unaffectedMons)    modify' $ registeredHere ^: Map.filter (\pid -> not $ ident `impliesDeathOf` ProcessIdentifier pid)    remaining <- fmap Map.toList (gets (^. registeredOnNodes)) >>=-      mapM (\(pid,nidlist) -> +      mapM (\(pid,nidlist) ->         case ident `impliesDeathOf` ProcessIdentifier pid of-           True -> +           True ->               do forM_ nidlist $ \(nid,_) ->-                   when (not $ isLocal node (NodeIdentifier nid)) +                   when (not $ isLocal node (NodeIdentifier nid))                       (forwardNameDeath node nid)                  return Nothing            False -> return $ Just (pid,nidlist)  )   modify' $ registeredOnNodes ^= (Map.fromList (catMaybes remaining))-    where +    where        forwardNameDeath node nid =                    liftIO $ sendBinary node                              (NodeIdentifier $ localNodeId node)                              (NodeIdentifier $ nid)                              WithImplicitReconnect                              NCMsg-                             { ctrlMsgSender = NodeIdentifier (localNodeId node) +                             { ctrlMsgSender = NodeIdentifier (localNodeId node)                              , ctrlMsgSignal = Died ident reason-                             }              -+                             }  -- [Unified: Table 13] ncEffectSpawn :: ProcessId -> Closure (Process ()) -> SpawnRef -> NC ()@@ -612,9 +699,9 @@   pid' <- liftIO $ forkProcess node proc   liftIO $ sendMessage node                        (NodeIdentifier (localNodeId node))-                       (ProcessIdentifier pid) +                       (ProcessIdentifier pid)                        WithImplicitReconnect-                       (DidSpawn ref pid') +                       (DidSpawn ref pid')  -- Unified semantics does not explicitly describe how to implement 'register', -- but mentions it's "very similar to nsend" (Table 14)@@ -629,7 +716,7 @@            return $ isJust currentVal          Just thepid -> -- register request            do isvalidlocal <- isValidLocalIdentifier (ProcessIdentifier thepid)-              return $ (isNothing currentVal /= reregistration) && +              return $ (isNothing currentVal /= reregistration) &&                 (not (isLocal node (ProcessIdentifier thepid) ) || isvalidlocal )   if isLocal node (NodeIdentifier atnode)      then do when (isOk) $@@ -637,10 +724,10 @@                   updateRemote node currentVal mPid              liftIO $ sendMessage node                        (NodeIdentifier (localNodeId node))-                       (ProcessIdentifier from) +                       (ProcessIdentifier from)                        WithImplicitReconnect                        (RegisterReply label isOk)-     else let operation = +     else let operation =                  case reregistration of                     True -> flip decList                     False -> flip incList@@ -666,7 +753,7 @@             decList ((atag,1):xs) tag | atag == tag = xs             decList ((atag,n):xs) tag | atag == tag = (atag,n-1):xs             decList (x:xs) tag = x:decList xs tag-            forward node to reg = +            forward node to reg =               when (not $ isLocal node (NodeIdentifier to)) $                     liftIO $ sendBinary node                                         (ProcessIdentifier from)@@ -685,23 +772,84 @@   mPid <- gets (^. registeredHereFor label)   liftIO $ sendMessage node                        (NodeIdentifier (localNodeId node))-                       (ProcessIdentifier from) +                       (ProcessIdentifier from)                        WithImplicitReconnect                        (WhereIsReply label mPid)  -- [Unified: Table 14] ncEffectNamedSend :: Identifier -> String -> Message -> NC () ncEffectNamedSend from label msg = do-  node <- ask  +  node <- ask   mPid <- gets (^. registeredHereFor label)   -- If mPid is Nothing, we just ignore the named send (as per Table 14)-  forM_ mPid $ \pid -> +  forM_ mPid $ \pid ->     liftIO $ sendPayload node                          from-                         (ProcessIdentifier pid) +                         (ProcessIdentifier pid)                          NoImplicitReconnect-                         (messageToPayload msg) +                         (messageToPayload msg) +-- [Issue #69]+ncEffectKill :: ProcessId -> ProcessId -> String -> NC ()+ncEffectKill from to reason = do+  node <- ask+  when (isLocal node (ProcessIdentifier to)) $+    throwException to $ ProcessKillException from reason++-- [Issue #69]+ncEffectExit :: ProcessId -> ProcessId -> Message -> NC ()+ncEffectExit from to reason = do+  node <- ask+  when (isLocal node (ProcessIdentifier to)) $+    throwException to $ ProcessExitException from reason++-- [Issue #89]+ncEffectGetInfo :: ProcessId -> ProcessId -> NC ()+ncEffectGetInfo from pid =+  let lpid = processLocalId pid+      them = (ProcessIdentifier pid)+  in do+  node <- ask+  mProc <- liftIO $+            withMVar (localState node) $ return . (^. localProcessWithId lpid)+  case mProc of+    Nothing   -> dispatch (isLocal node (ProcessIdentifier from))+                          from node (ProcessInfoNone DiedUnknownId)+    Just _    -> do+      itsLinks    <- gets (^. linksFor    them)+      itsMons     <- gets (^. monitorsFor them)+      registered  <- gets (^. registeredHere)++      let reg = registeredNames registered+      dispatch (isLocal node (ProcessIdentifier from))+               from+               node+               ProcessInfo {+                   infoNode               = (processNodeId pid)+                 , infoRegisteredNames    = reg+                   -- we cannot populate this field+                 , infoMessageQueueLength = Nothing+                 , infoMonitors       = Set.toList itsMons+                 , infoLinks          = Set.toList itsLinks+                 }+  where dispatch :: (Serializable a, Show a)+                 => Bool+                 -> ProcessId+                 -> LocalNode+                 -> a+                 -> NC ()+        dispatch True  dest _    pInfo = postAsMessage dest $ pInfo+        dispatch False dest node pInfo = do+            liftIO $ sendMessage node+                                 (NodeIdentifier (localNodeId node))+                                 (ProcessIdentifier dest)+                                 WithImplicitReconnect+                                 pInfo++        registeredNames = Map.foldlWithKey (\ks k v -> if v == pid+                                                 then (k:ks)+                                                 else ks) []+ -------------------------------------------------------------------------------- -- Auxiliary                                                                  -- --------------------------------------------------------------------------------@@ -712,20 +860,20 @@            -> Maybe MonitorRef  -- ^ 'Nothing' for linking            -> NC () notifyDied dest src reason mRef = do-  node <- ask +  node <- ask   case (isLocal node (ProcessIdentifier dest), mRef, src) of     (True, Just ref, ProcessIdentifier pid) ->-      postAsMessage dest $ ProcessMonitorNotification ref pid reason +      postAsMessage dest $ ProcessMonitorNotification ref pid reason     (True, Just ref, NodeIdentifier nid) ->       postAsMessage dest $ NodeMonitorNotification ref nid reason     (True, Just ref, SendPortIdentifier cid) ->       postAsMessage dest $ PortMonitorNotification ref cid reason     (True, Nothing, ProcessIdentifier pid) ->-      throwException dest $ ProcessLinkException pid reason +      throwException dest $ ProcessLinkException pid reason     (True, Nothing, NodeIdentifier pid) ->-      throwException dest $ NodeLinkException pid reason +      throwException dest $ NodeLinkException pid reason     (True, Nothing, SendPortIdentifier pid) ->-      throwException dest $ PortLinkException pid reason +      throwException dest $ PortLinkException pid reason     (False, _, _) ->       -- The change in sender comes from [Unified: Table 10]       liftIO $ sendBinary node@@ -733,30 +881,33 @@                           (NodeIdentifier $ processNodeId dest)                           WithImplicitReconnect         NCMsg-          { ctrlMsgSender = NodeIdentifier (localNodeId node) +          { ctrlMsgSender = NodeIdentifier (localNodeId node)           , ctrlMsgSignal = Died src reason           }-      + -- | [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 (Died _ _)          = Nothing+destNid (Kill pid _)        = Just $ processNodeId pid+destNid (Exit pid _)        = Just $ processNodeId pid+destNid (GetInfo pid)       = Just $ processNodeId pid  -- | Check if a process is local to our own node-isLocal :: LocalNode -> Identifier -> Bool +isLocal :: LocalNode -> Identifier -> Bool isLocal nid ident = nodeOf ident == localNodeId nid --- | Lookup a local closure +-- | Lookup a local closure unClosure :: Typeable a => Closure a -> NC (Either String a) unClosure closure = do   rtable <- remoteTable <$> ask@@ -766,13 +917,13 @@ isValidLocalIdentifier :: Identifier -> NC Bool isValidLocalIdentifier ident = do   node <- ask-  liftIO . withMVar (localState node) $ \nSt -> +  liftIO . withMVar (localState node) $ \nSt ->     case ident of       NodeIdentifier nid ->         return $ nid == localNodeId node       ProcessIdentifier pid -> do         let mProc = nSt ^. localProcessWithId (processLocalId pid)-        return $ isJust mProc +        return $ isJust mProc       SendPortIdentifier cid -> do         let pid   = sendPortProcessId cid             mProc = nSt ^. localProcessWithId (processLocalId pid)@@ -787,24 +938,25 @@ --------------------------------------------------------------------------------  postAsMessage :: Serializable a => ProcessId -> a -> NC ()-postAsMessage pid = postMessage pid . createMessage  +postAsMessage pid = postMessage pid . createMessage  postMessage :: ProcessId -> Message -> NC ()-postMessage pid msg = withLocalProc pid $ \p -> enqueue (processQueue p) msg+postMessage pid msg = do+  withLocalProc pid $ \p -> enqueue (processQueue p) msg  throwException :: Exception e => ProcessId -> e -> NC ()-throwException pid e = withLocalProc pid $ \p -> +throwException pid e = withLocalProc pid $ \p ->   throwTo (processThread p) e -withLocalProc :: ProcessId -> (LocalProcess -> IO ()) -> NC () +withLocalProc :: ProcessId -> (LocalProcess -> IO ()) -> NC () withLocalProc pid p = do-  node <- ask -  liftIO $ 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 +    forM_ mProc p  -------------------------------------------------------------------------------- -- Accessors                                                                  --@@ -841,23 +993,23 @@ -- There is a hierarchy between identifiers: failure of a node implies failure -- of all processes on that node, and failure of a process implies failure of -- all typed channels to that process. In other words, if 'ident' refers to a--- node, then the /should trigger/ set will include +-- node, then the /should trigger/ set will include -- -- * the notifications for the node specifically--- * the notifications for processes on that node, and --- * the notifications for typed channels to processes on that node. +-- * the notifications for processes on that node, and+-- * the notifications for typed channels to processes on that node. -- -- Similarly, if 'ident' refers to a process, the /should trigger/ set will--- include +-- include ----- * the notifications for that process specifically and +-- * the notifications for that process specifically and -- * the notifications for typed channels to that process. -- -- See https://github.com/haskell/containers/issues/14 for the bang on _v. splitNotif :: Identifier            -> Map Identifier a            -> (Map Identifier a, Map Identifier a)-splitNotif ident = Map.partitionWithKey (\k !_v -> ident `impliesDeathOf` k) +splitNotif ident = Map.partitionWithKey (\k !_v -> ident `impliesDeathOf` k)  -------------------------------------------------------------------------------- -- Auxiliary                                                                  --

@@ -1,4 +1,4 @@-module Control.Distributed.Process.Serializable +module Control.Distributed.Process.Serializable   ( Serializable   , encodeFingerprint   , decodeFingerprint@@ -35,29 +35,29 @@  -- | Encode type representation as a bytestring encodeFingerprint :: Fingerprint -> ByteString-encodeFingerprint fp = +encodeFingerprint fp =   -- Since all CH nodes will run precisely the same binary, we don't have to   -- worry about cross-arch issues here (like endianness)-  BSI.unsafeCreate sizeOfFingerprint $ \p -> pokeByteOff p 0 fp +  BSI.unsafeCreate sizeOfFingerprint $ \p -> pokeByteOff p 0 fp --- | Decode a bytestring into a fingerprint. Throws an IO exception on failure +-- | Decode a bytestring into a fingerprint. Throws an IO exception on failure decodeFingerprint :: ByteString -> Fingerprint decodeFingerprint bs-  | BS.length bs /= sizeOfFingerprint = +  | BS.length bs /= sizeOfFingerprint =       throw $ userError "decodeFingerprint: Invalid length"   | otherwise = BSI.inlinePerformIO $ do       let (fp, offset, _) = BSI.toForeignPtr bs-      withForeignPtr fp $ \p -> peekByteOff p offset +      withForeignPtr fp $ \p -> peekByteOff p offset  -- | Size of a fingerprint sizeOfFingerprint :: Int sizeOfFingerprint = sizeOf (undefined :: Fingerprint) --- | The fingerprint of the typeRep of the argument +-- | The fingerprint of the typeRep of the argument fingerprint :: Typeable a => a -> Fingerprint fingerprint a = let TypeRep fp _ _ = typeOf a in fp  -- | Show fingerprint (for debugging purposes) showFingerprint :: Fingerprint -> ShowS-showFingerprint (Fingerprint hi lo) = +showFingerprint (Fingerprint hi lo) =   showString "(" . showHex hi . showString "," . showHex lo . showString ")"

@@ -1,15 +1,14 @@-module Main where +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 +import Control.Concurrent.MVar   ( MVar   , newEmptyMVar   , putMVar@@ -22,16 +21,17 @@ import Control.Applicative ((<$>), (<*>), pure, (<|>)) import qualified Network.Transport as NT (Transport, closeEndPoint) import Network.Socket (sClose)-import Network.Transport.TCP +import Network.Transport.TCP   ( createTransportExposeInternals   , TransportInternals(socketBetween)   , defaultTCPParameters   ) import Control.Distributed.Process-import Control.Distributed.Process.Internal.Types -  ( NodeId(nodeAddress) +import Control.Distributed.Process.Internal.Types+  ( NodeId(nodeAddress)   , LocalNode(localEndPoint)-  , RegisterReply(..)+  , ProcessExitException(..)+  , nullProcessId   ) import Control.Distributed.Process.Node import Control.Distributed.Process.Serializable (Serializable)@@ -51,7 +51,7 @@ --------------------------------------------------------------------------------  -- | Like fork, but throw exceptions in the child thread to the parent-forkTry :: IO () -> IO ThreadId +forkTry :: IO () -> IO ThreadId forkTry p = do   tid <- myThreadId   forkIO $ Ex.catch p (\e -> throwTo tid (e :: SomeException))@@ -71,7 +71,7 @@   WhereIsReply _ mPid <- expect   return mPid -data Add       = Add    ProcessId Double Double deriving (Typeable) +data Add       = Add    ProcessId Double Double deriving (Typeable) data Divide    = Divide ProcessId Double Double deriving (Typeable) data DivByZero = DivByZero deriving (Typeable) @@ -102,7 +102,7 @@ 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) +              -> Process (Maybe MonitorRef) monitorOrLink mOrL pid mSignal = do   result <- if mOrL then Just <$> monitor pid                     else link pid >> return Nothing@@ -116,11 +116,11 @@                    -> Bool            -- 'True' for monitor, 'False' for link                    -> Bool            -- Should we unmonitor?                    -> DiedReason      -- Expected cause of death-                   -> Maybe (MVar ()) -- Signal for 'monitor set up' +                   -> 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 +monitorTestProcess theirAddr mOrL un reason monitorSetup done =+  catch (do mRef <- monitorOrLink mOrL theirAddr monitorSetup             case (un, mRef) of               (True, Nothing) -> do                 unlink theirAddr@@ -137,13 +137,13 @@             True <- return $ pid == theirAddr && not mOrL && not un && reason == reason'             liftIO $ putMVar done ()         )-  + -------------------------------------------------------------------------------- -- The tests proper                                                           -- --------------------------------------------------------------------------------  -- | Basic ping test-testPing :: NT.Transport -> Assertion +testPing :: NT.Transport -> Assertion testPing transport = do   serverAddr <- newEmptyMVar   clientDone <- newEmptyMVar@@ -172,15 +172,15 @@    takeMVar clientDone --- | Monitor a process on an unreachable node -testMonitorUnreachable :: NT.Transport -> Bool -> Bool -> Assertion +-- | 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 +    addr <- forkProcess localNode . liftIO $ threadDelay 1000000     closeLocalNode localNode     putMVar deadProcess addr @@ -188,12 +188,12 @@     localNode <- newLocalNode transport initRemoteTable     theirAddr <- readMVar deadProcess     runProcess localNode $-      monitorTestProcess theirAddr mOrL un DiedDisconnect Nothing done +      monitorTestProcess theirAddr mOrL un DiedDisconnect Nothing done    takeMVar done  -- | Monitor a process which terminates normally-testMonitorNormalTermination :: NT.Transport -> Bool -> Bool -> Assertion +testMonitorNormalTermination :: NT.Transport -> Bool -> Bool -> Assertion testMonitorNormalTermination transport mOrL un = do   monitorSetup <- newEmptyMVar   monitoredProcess <- newEmptyMVar@@ -201,20 +201,20 @@    forkIO $ do     localNode <- newLocalNode transport initRemoteTable-    addr <- forkProcess localNode $ +    addr <- forkProcess localNode $       liftIO $ readMVar monitorSetup     putMVar monitoredProcess addr    forkIO $ do     localNode <- newLocalNode transport initRemoteTable     theirAddr <- readMVar monitoredProcess-    runProcess localNode $ +    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 :: NT.Transport -> Bool -> Bool -> Assertion testMonitorAbnormalTermination transport mOrL un = do   monitorSetup <- newEmptyMVar   monitoredProcess <- newEmptyMVar@@ -226,19 +226,19 @@     localNode <- newLocalNode transport initRemoteTable     addr <- forkProcess localNode . liftIO $ do       readMVar monitorSetup-      throwIO err +      throwIO err     putMVar monitoredProcess addr    forkIO $ do     localNode <- newLocalNode transport initRemoteTable     theirAddr <- readMVar monitoredProcess-    runProcess localNode $ +    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 :: NT.Transport -> Bool -> Bool -> Assertion testMonitorLocalDeadProcess transport mOrL un = do   processDead <- newEmptyMVar   processAddr <- newEmptyMVar@@ -258,7 +258,7 @@   takeMVar done  -- | Monitor a remote process that is already dead-testMonitorRemoteDeadProcess :: NT.Transport -> Bool -> Bool -> Assertion +testMonitorRemoteDeadProcess :: NT.Transport -> Bool -> Bool -> Assertion testMonitorRemoteDeadProcess transport mOrL un = do   processDead <- newEmptyMVar   processAddr <- newEmptyMVar@@ -279,7 +279,7 @@   takeMVar done  -- | Monitor a process that becomes disconnected-testMonitorDisconnect :: NT.Transport -> Bool -> Bool -> Assertion +testMonitorDisconnect :: NT.Transport -> Bool -> Bool -> Assertion testMonitorDisconnect transport mOrL un = do   processAddr <- newEmptyMVar   monitorSetup <- newEmptyMVar@@ -287,7 +287,7 @@    forkIO $ do     localNode <- newLocalNode transport initRemoteTable-    addr <- forkProcess localNode . liftIO $ threadDelay 1000000 +    addr <- forkProcess localNode . liftIO $ threadDelay 1000000     putMVar processAddr addr     readMVar monitorSetup     NT.closeEndPoint (localEndPoint localNode)@@ -297,18 +297,18 @@     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 :: NT.Transport -> Assertion testMath transport = do-  serverAddr <- newEmptyMVar +  serverAddr <- newEmptyMVar   clientDone <- newEmptyMVar    -- Server   forkIO $ do-    localNode <- newLocalNode transport initRemoteTable +    localNode <- newLocalNode transport initRemoteTable     addr <- forkProcess localNode math     putMVar serverAddr addr @@ -320,7 +320,7 @@     runProcess localNode $ do       pid <- getSelfPid       send mathServer (Add pid 1 2)-      3 <- expect :: Process Double  +      3 <- expect :: Process Double       send mathServer (Divide pid 8 2)       4 <- expect :: Process Double       send mathServer (Divide pid 8 0)@@ -330,15 +330,15 @@   takeMVar clientDone  -- | Send first message (i.e. connect) to an already terminated process--- (without monitoring); then send another message to a second process on +-- (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 :: NT.Transport -> Assertion testSendToTerminated transport = do   serverAddr1 <- newEmptyMVar   serverAddr2 <- newEmptyMVar   clientDone <- newEmptyMVar -  forkIO $ do +  forkIO $ do     terminated <- newEmptyMVar     localNode <- newLocalNode transport initRemoteTable     addr1 <- forkProcess localNode $ liftIO $ putMVar terminated ()@@ -356,34 +356,38 @@       send server1 "Hi"       send server2 (Pong pid)       Ping pid' <- expect-      True <- return $ pid' == server2 +      True <- return $ pid' == server2       liftIO $ putMVar clientDone ()    takeMVar clientDone  -- | Test (non-zero) timeout-testTimeout :: NT.Transport -> Assertion +testTimeout :: NT.Transport -> Assertion testTimeout transport = do   localNode <- newLocalNode transport initRemoteTable+  done <- newEmptyMVar+   runProcess localNode $ do     Nothing <- receiveTimeout 1000000 [match (\(Add _ _ _) -> return ())]-    return ()+    liftIO $ putMVar done () +  takeMVar done+ -- | Test zero timeout-testTimeout0 :: NT.Transport -> Assertion +testTimeout0 :: NT.Transport -> Assertion testTimeout0 transport = do   serverAddr <- newEmptyMVar   clientDone <- newEmptyMVar   messagesSent <- newEmptyMVar -  forkIO $ do +  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)] +      Just partner <- receiveTimeout 0 [match (\(Pong partner) -> return partner)]       self <- getSelfPid       send partner (Ping self)     putMVar serverAddr addr@@ -398,13 +402,13 @@       replicateM_ 10000 $ send server "Irrelevant message"       send server (Pong pid)       liftIO $ putMVar messagesSent ()-      Ping _ <- expect +      Ping _ <- expect       liftIO $ putMVar clientDone ()    takeMVar clientDone  -- | Test typed channels-testTypedChannels :: NT.Transport -> Assertion +testTypedChannels :: NT.Transport -> Assertion testTypedChannels transport = do   serverChannel <- newEmptyMVar :: IO (MVar (SendPort (SendPort Bool, Int)))   clientDone <- newEmptyMVar@@ -413,7 +417,7 @@     localNode <- newLocalNode transport initRemoteTable     forkProcess localNode $ do       (serverSendPort, rport) <- newChan-      liftIO $ putMVar serverChannel serverSendPort +      liftIO $ putMVar serverChannel serverSendPort       (clientSendPort, i) <- receiveChan rport       sendChan clientSendPort (even i)     return ()@@ -423,14 +427,14 @@     serverSendPort <- readMVar serverChannel     runProcess localNode $ do       (clientSendPort, rport) <- newChan-      sendChan serverSendPort (clientSendPort, 5) +      sendChan serverSendPort (clientSendPort, 5)       False <- receiveChan rport       liftIO $ putMVar clientDone ()-      -  takeMVar clientDone  +  takeMVar clientDone+ -- | Test merging receive ports-testMergeChannels :: NT.Transport -> Assertion +testMergeChannels :: NT.Transport -> Assertion testMergeChannels transport = do     localNode <- newLocalNode transport initRemoteTable     testFlat localNode True          "aaabbbccc"@@ -443,13 +447,13 @@     testBlocked localNode False   where     -- Single layer of merging-    testFlat :: LocalNode -> Bool -> String -> IO () +    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 +        rs  <- mapM charChannel "abc"+        m   <- mergePorts biased rs+        xs  <- replicateM 9 $ receiveChan m         True <- return $ xs == expected         liftIO $ putMVar done ()       takeMVar done@@ -462,7 +466,7 @@         rss  <- mapM (mapM charChannel) ["abc", "def", "ghi"]         ms   <- mapM (mergePorts biasedInner) rss         m    <- mergePorts biasedOuter ms-        xs   <- replicateM (9 * 3) $ receiveChan m +        xs   <- replicateM (9 * 3) $ receiveChan m         True <- return $ xs == expected         liftIO $ putMVar done ()       takeMVar done@@ -470,12 +474,12 @@     -- 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 +      vs <- replicateM 3 newEmptyMVar       done <- newEmptyMVar        forkProcess localNode $ do-        [sa, sb, sc] <- liftIO $ mapM readMVar vs -        mapM_ ((>> liftIO (threadDelay 10000)) . uncurry sendChan) +        [sa, sb, sc] <- liftIO $ mapM readMVar vs+        mapM_ ((>> liftIO (threadDelay 10000)) . uncurry sendChan)           [ -- a, b, c             (sa, 'a')           , (sb, 'b')@@ -504,8 +508,8 @@        forkProcess localNode $ do         (ss, rs) <- unzip <$> replicateM 3 newChan-        liftIO $ mapM_ (uncurry putMVar) $ zip vs ss -        m  <- mergePorts biased rs +        liftIO $ mapM_ (uncurry putMVar) $ zip vs ss+        m  <- mergePorts biased rs         xs <- replicateM (6 * 3) $ receiveChan m         True <- return $ xs == "abcacbbacbcacabcba"         liftIO $ putMVar done ()@@ -514,18 +518,19 @@      mergePorts :: Serializable a => Bool -> [ReceivePort a] -> Process (ReceivePort a)     mergePorts True  = mergePortsBiased-    mergePorts False = mergePortsRR +    mergePorts False = mergePortsRR      charChannel :: Char -> Process (ReceivePort Char)     charChannel c = do       (sport, rport) <- newChan-      replicateM_ 3 $ sendChan sport c +      replicateM_ 3 $ sendChan sport c       liftIO $ threadDelay 10000 -- Make sure messages have been sent       return rport -testTerminate :: NT.Transport -> Assertion +testTerminate :: NT.Transport -> Assertion testTerminate transport = do   localNode <- newLocalNode transport initRemoteTable+  done <- newEmptyMVar    pid <- forkProcess localNode $ do     liftIO $ threadDelay 100000@@ -534,12 +539,15 @@   runProcess localNode $ do     ref <- monitor pid     ProcessMonitorNotification ref' pid' (DiedException ex) <- expect-    True <- return $ ref == ref' && pid == pid' && ex == show ProcessTerminationException -    return ()+    True <- return $ ref == ref' && pid == pid' && ex == show ProcessTerminationException+    liftIO $ putMVar done () -testMonitorNode :: NT.Transport -> Assertion +  takeMVar done++testMonitorNode :: NT.Transport -> Assertion testMonitorNode transport = do   [node1, node2] <- replicateM 2 $ newLocalNode transport initRemoteTable+  done <- newEmptyMVar    closeLocalNode node1 @@ -547,9 +555,11 @@     ref <- monitorNode (localNodeId node1)     NodeMonitorNotification ref' nid DiedDisconnect <- expect     True <- return $ ref == ref' && nid == localNodeId node1-    return ()+    liftIO $ putMVar done () -testMonitorChannel :: NT.Transport -> Assertion +  takeMVar done++testMonitorChannel :: NT.Transport -> Assertion testMonitorChannel transport = do     [node1, node2] <- replicateM 2 $ newLocalNode transport initRemoteTable     gotNotification <- newEmptyMVar@@ -558,7 +568,7 @@       sport <- expect :: Process (SendPort ())       ref <- monitorPort sport       PortMonitorNotification ref' port' reason <- expect-      -- reason might be DiedUnknownId if the receive port is GCed before the +      -- 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 ()@@ -570,47 +580,54 @@      takeMVar gotNotification -testRegistry :: NT.Transport -> Assertion +testRegistry :: NT.Transport -> Assertion testRegistry transport = do   node <- newLocalNode transport initRemoteTable+  done <- newEmptyMVar -  pingServer <- forkProcess node ping +  pingServer <- forkProcess node ping    runProcess node $ do     register "ping" pingServer     Just pid <- whereis "ping"-    True <- return $ pingServer == pid +    True <- return $ pingServer == pid     us <- getSelfPid     nsend "ping" (Pong us)-    Ping pid' <- expect +    Ping pid' <- expect     True <- return $ pingServer == pid'-    return ()+    liftIO $ putMVar done () -testRemoteRegistry :: NT.Transport -> Assertion +  takeMVar done++testRemoteRegistry :: NT.Transport -> Assertion testRemoteRegistry transport = do   node1 <- newLocalNode transport initRemoteTable   node2 <- newLocalNode transport initRemoteTable+  done <- newEmptyMVar -  pingServer <- forkProcess node1 ping +  pingServer <- forkProcess node1 ping    runProcess node2 $ do     let nid1 = localNodeId node1     registerRemoteAsync nid1 "ping" pingServer-    receiveWait [ +    receiveWait [        matchIf (\(RegisterReply label' _) -> "ping" == label')                (\(RegisterReply _ _) -> return ()) ]      Just pid <- whereisRemote nid1 "ping"-    True <- return $ pingServer == pid +    True <- return $ pingServer == pid     us <- getSelfPid     nsendRemote nid1 "ping" (Pong us)-    Ping pid' <- expect +    Ping pid' <- expect     True <- return $ pingServer == pid'-    return ()+    liftIO $ putMVar done () -testSpawnLocal :: NT.Transport -> Assertion +  takeMVar done++testSpawnLocal :: NT.Transport -> Assertion testSpawnLocal transport = do   node <- newLocalNode transport initRemoteTable+  done <- newEmptyMVar    runProcess node $ do     us <- getSelfPid@@ -624,9 +641,12 @@       send us ()      send pid sport-    expect+    () <- expect+    liftIO $ putMVar done () -testReconnect :: NT.Transport -> TransportInternals -> Assertion +  takeMVar done++testReconnect :: NT.Transport -> TransportInternals -> Assertion testReconnect transport transportInternals = do   [node1, node2] <- replicateM 2 $ newLocalNode transport initRemoteTable   let nid1 = localNodeId node1@@ -651,40 +671,40 @@     send them "message 1" >> liftIO (threadDelay 100000)      -- Simulate network failure-    liftIO $ do +    liftIO $ do       sock <- socketBetween transportInternals (nodeAddress nid1) (nodeAddress nid2)       sClose sock       threadDelay 10000      -- Should not arrive-    send them "message 2" +    send them "message 2"      -- Should arrive     reconnect them-    send them "message 3" +    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" -      +    Just _  <- whereisRemote nid1 "a" +     -- Simulate network failure-    liftIO $ do +    liftIO $ do       sock <- socketBetween transportInternals (nodeAddress nid1) (nodeAddress nid2)       sClose sock       threadDelay 10000 -    -- This will happen due to implicit reconnect +    -- This will happen due to implicit reconnect     registerRemoteAsync nid1 "b" us     receiveWait [         matchIf (\(RegisterReply label' _) -> "b" == label')@@ -699,27 +719,27 @@     -- 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" +    Just _  <- whereisRemote nid1 "c"      liftIO $ putMVar registerTestOk ()    takeMVar registerTestOk --- | Test 'matchAny'. This repeats the 'testMath' but with a proxy server +-- | Test 'matchAny'. This repeats the 'testMath' but with a proxy server -- in between-testMatchAny :: NT.Transport -> Assertion +testMatchAny :: NT.Transport -> Assertion testMatchAny transport = do-  proxyAddr <- newEmptyMVar +  proxyAddr <- newEmptyMVar   clientDone <- newEmptyMVar    -- Math server   forkIO $ do-    localNode <- newLocalNode transport initRemoteTable +    localNode <- newLocalNode transport initRemoteTable     mathServer <- forkProcess localNode math-    proxyServer <- forkProcess localNode $ forever $ do +    proxyServer <- forkProcess localNode $ forever $ do       msg <- receiveWait [ matchAny return ]       forward msg mathServer-    putMVar proxyAddr proxyServer +    putMVar proxyAddr proxyServer    -- Client   forkIO $ do@@ -729,7 +749,7 @@     runProcess localNode $ do       pid <- getSelfPid       send mathServer (Add pid 1 2)-      3 <- expect :: Process Double  +      3 <- expect :: Process Double       send mathServer (Divide pid 8 2)       4 <- expect :: Process Double       send mathServer (Divide pid 8 0)@@ -738,12 +758,127 @@    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 :: NT.Transport -> Assertion testReceiveChanTimeout transport = do   done <- newEmptyMVar   sendPort <- newEmptyMVar-  +   forkTry $ do     localNode <- newLocalNode transport initRemoteTable     runProcess localNode $ do@@ -751,7 +886,7 @@       (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 +      -- 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@@ -762,7 +897,7 @@        -- Again, but now there is a message available       liftIO $ threadDelay 1000000-      Just False <- receiveChanTimeout 0 rp +      Just False <- receiveChanTimeout 0 rp        liftIO $ putMVar done () @@ -810,9 +945,9 @@       7 <- receiveChan rp2        -- Test Alternative instance-      +       sendChan spInt 3-      sendChan spBool True +      sendChan spBool True        let rp3 = (even <$> rpInt) <|> rpBool @@ -820,14 +955,14 @@       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 +                   if b                      then rpInt                      else return 7 @@ -838,14 +973,158 @@    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) = [ +tests (transport, transportInternals) = [     testGroup "Basic features" [         testCase "Ping"                (testPing                transport)-      , testCase "Math"                (testMath                transport) +      , testCase "Math"                (testMath                transport)       , testCase "Timeout"             (testTimeout             transport)       , testCase "Timeout0"            (testTimeout0            transport)-      , testCase "SendToTerminated"    (testSendToTerminated    transport) +      , testCase "SendToTerminated"    (testSendToTerminated    transport)       , testCase "TypedChannnels"      (testTypedChannels       transport)       , testCase "MergeChannels"       (testMergeChannels       transport)       , testCase "Terminate"           (testTerminate           transport)@@ -853,14 +1132,25 @@       , 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 +      -- 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)

@@ -6,7 +6,7 @@ import Control.Monad (join, replicateM, forever, replicateM_, void) import Control.Exception (IOException, throw) import Control.Concurrent (forkIO, threadDelay)-import Control.Concurrent.MVar +import Control.Concurrent.MVar   ( MVar   , newEmptyMVar   , readMVar@@ -18,12 +18,12 @@ import Control.Applicative ((<$>)) import System.Random (randomIO) import Network.Transport (Transport)-import Network.Transport.TCP +import Network.Transport.TCP   ( createTransportExposeInternals   , defaultTCPParameters   , TransportInternals(socketBetween)   )-import Network.Socket (sClose)  +import Network.Socket (sClose) import Control.Distributed.Process import Control.Distributed.Process.Closure import Control.Distributed.Process.Node@@ -46,7 +46,7 @@  factorial :: Int -> Process Int factorial 0 = return 1-factorial n = (n *) <$> factorial (n - 1) +factorial n = (n *) <$> factorial (n - 1)  addInt :: Int -> Int -> Int addInt x y = x + y@@ -95,7 +95,7 @@   ix <- randomIO   return (xs !! (ix `mod` length xs)) -remotableDecl [ +remotableDecl [     [d| dfib :: ([NodeId], SendPort Integer, Integer) -> Process () ;         dfib (_, reply, 0) = sendChan reply 0         dfib (_, reply, 1) = sendChan reply 1@@ -117,33 +117,33 @@ staticQuintuple = $(mkStatic 'quintuple)  factorialClosure :: Int -> Closure (Process Int)-factorialClosure = $(mkClosure 'factorial) +factorialClosure = $(mkClosure 'factorial)  addIntClosure :: Int -> Closure (Int -> Int)-addIntClosure = $(mkClosure 'addInt) +addIntClosure = $(mkClosure 'addInt)  putIntClosure :: Int -> Closure (MVar Int -> IO ())-putIntClosure = $(mkClosure 'putInt) +putIntClosure = $(mkClosure 'putInt)  sendPidClosure :: ProcessId -> Closure (Process ())-sendPidClosure = $(mkClosure 'sendPid) +sendPidClosure = $(mkClosure 'sendPid)  sendFac :: Int -> ProcessId -> Closure (Process ())-sendFac n pid = factorialClosure n `bindCP` cpSend $(mkStatic 'sdictInt) pid +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 +factorial' n = returnCP $(mkStatic 'sdictInt) n `bindCP` factorialOf  waitClosure :: Int -> Closure (Process ())-waitClosure = $(mkClosure 'wait) +waitClosure = $(mkClosure 'wait)  simulateNetworkFailure :: TransportInternals -> NodeId -> NodeId -> Process () simulateNetworkFailure transportInternals fr to = liftIO $ do   threadDelay 10000-  sock <- socketBetween transportInternals (nodeAddress fr) (nodeAddress to) +  sock <- socketBetween transportInternals (nodeAddress fr) (nodeAddress to)   sClose sock   threadDelay 10000 @@ -160,87 +160,87 @@     liftIO $ putMVar done ()   takeMVar done -testBind :: Transport -> RemoteTable -> Assertion +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 +    join . unClosure $ sendFac 6 us+    (720 :: Int) <- expect     liftIO $ putMVar done ()   takeMVar done -testSendPureClosure :: Transport -> RemoteTable -> Assertion +testSendPureClosure :: Transport -> RemoteTable -> Assertion testSendPureClosure transport rtable = do   serverAddr <- newEmptyMVar   serverDone <- newEmptyMVar -  forkIO $ do -    node <- newLocalNode transport rtable +  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 +      liftIO $ putMVar serverDone ()+    putMVar serverAddr addr    forkIO $ do-    node <- newLocalNode transport rtable +    node <- newLocalNode transport rtable     theirAddr <- readMVar serverAddr     runProcess node $ send theirAddr (addIntClosure 7)    takeMVar serverDone -testSendIOClosure :: Transport -> RemoteTable -> Assertion +testSendIOClosure :: Transport -> RemoteTable -> Assertion testSendIOClosure transport rtable = do   serverAddr <- newEmptyMVar   serverDone <- newEmptyMVar -  forkIO $ do -    node <- newLocalNode transport rtable +  forkIO $ do+    node <- newLocalNode transport rtable     addr <- forkProcess node $ do       cl <- expect       io <- unClosure cl :: Process (MVar Int -> IO ())-      liftIO $ do +      liftIO $ do         someMVar <- newEmptyMVar-        io someMVar +        io someMVar         5 <- readMVar someMVar-        putMVar serverDone () -    putMVar serverAddr addr +        putMVar serverDone ()+    putMVar serverAddr addr    forkIO $ do-    node <- newLocalNode transport rtable +    node <- newLocalNode transport rtable     theirAddr <- readMVar serverAddr-    runProcess node $ send theirAddr (putIntClosure 5) +    runProcess node $ send theirAddr (putIntClosure 5)    takeMVar serverDone -testSendProcClosure :: Transport -> RemoteTable -> Assertion +testSendProcClosure :: Transport -> RemoteTable -> Assertion testSendProcClosure transport rtable = do   serverAddr <- newEmptyMVar   clientDone <- newEmptyMVar -  forkIO $ do -    node <- newLocalNode transport rtable +  forkIO $ do+    node <- newLocalNode transport rtable     addr <- forkProcess node $ do       cl <- expect       pr <- unClosure cl :: Process (Int -> Process ())       pr 5-    putMVar serverAddr addr +    putMVar serverAddr addr    forkIO $ do-    node <- newLocalNode transport rtable +    node <- newLocalNode transport rtable     theirAddr <- readMVar serverAddr     runProcess node $ do       pid <- getSelfPid-      send theirAddr (cpSend $(mkStatic 'sdictInt) pid) +      send theirAddr (cpSend $(mkStatic 'sdictInt) pid)       5 <- expect :: Process Int       liftIO $ putMVar clientDone ()    takeMVar clientDone -testSpawn :: Transport -> RemoteTable -> Assertion +testSpawn :: Transport -> RemoteTable -> Assertion testSpawn transport rtable = do   serverNodeAddr <- newEmptyMVar   clientDone <- newEmptyMVar@@ -261,7 +261,7 @@    takeMVar clientDone -testCall :: Transport -> RemoteTable -> Assertion +testCall :: Transport -> RemoteTable -> Assertion testCall transport rtable = do   serverNodeAddr <- newEmptyMVar   clientDone <- newEmptyMVar@@ -279,7 +279,7 @@    takeMVar clientDone -testCallBind :: Transport -> RemoteTable -> Assertion +testCallBind :: Transport -> RemoteTable -> Assertion testCallBind transport rtable = do   serverNodeAddr <- newEmptyMVar   clientDone <- newEmptyMVar@@ -297,7 +297,7 @@    takeMVar clientDone -testSeq :: Transport -> RemoteTable -> Assertion +testSeq :: Transport -> RemoteTable -> Assertion testSeq transport rtable = do   node <- newLocalNode transport rtable   done <- newEmptyMVar@@ -315,7 +315,7 @@ -- 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 -> RemoteTable -> Assertion testSpawnSupervised transport rtable = do     [node1, node2]       <- replicateM 2 $ newLocalNode transport rtable     [superPid, childPid] <- replicateM 2 $ newEmptyMVar@@ -324,7 +324,7 @@     forkProcess node1 $ do       us <- getSelfPid       liftIO $ putMVar superPid us-      (child, _ref) <- spawnSupervised (localNodeId node2) (waitClosure 1000000) +      (child, _ref) <- spawnSupervised (localNodeId node2) (waitClosure 1000000)       liftIO $ do         putMVar childPid child         threadDelay 500000 -- Give the child a chance to link to us@@ -334,8 +334,8 @@       [super, child] <- liftIO $ mapM readMVar [superPid, childPid]       ref <- monitor child       ProcessMonitorNotification ref' pid' (DiedException e) <- expect-      True <- return $ ref' == ref -                    && pid' == child +      True <- return $ ref' == ref+                    && pid' == child                     && e == show (ProcessLinkException super (DiedException (show supervisorDeath)))       liftIO $ putMVar thirdProcessDone () @@ -344,19 +344,19 @@     supervisorDeath :: IOException     supervisorDeath = userError "Supervisor died" -testSpawnInvalid :: Transport -> RemoteTable -> Assertion +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 +    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'  +    True <- return $ ref' == ref && pid == pid'     liftIO $ putMVar done ()   takeMVar done -testClosureExpect :: Transport -> RemoteTable -> Assertion +testClosureExpect :: Transport -> RemoteTable -> Assertion testClosureExpect transport rtable = do   node <- newLocalNode transport rtable   done <- newEmptyMVar@@ -369,15 +369,15 @@     liftIO $ putMVar done ()   takeMVar done -testSpawnChannel :: Transport -> RemoteTable -> Assertion +testSpawnChannel :: Transport -> RemoteTable -> Assertion testSpawnChannel transport rtable = do   done <- newEmptyMVar   [node1, node2] <- replicateM 2 $ newLocalNode transport rtable    forkProcess node1 $ do-    pingServer <- spawnChannel +    pingServer <- spawnChannel                     (sdictSendPort sdictUnit)-                    (localNodeId node2)  +                    (localNodeId node2)                     ($(mkClosure 'typedPingServer) ())     (sendReply, receiveReply) <- newChan     sendChan pingServer sendReply@@ -386,7 +386,7 @@    takeMVar done -testTDict :: Transport -> RemoteTable -> Assertion +testTDict :: Transport -> RemoteTable -> Assertion testTDict transport rtable = do   done <- newEmptyMVar   [node1, node2] <- replicateM 2 $ newLocalNode transport rtable@@ -395,7 +395,7 @@     liftIO $ putMVar done ()   takeMVar done -testFib :: Transport -> RemoteTable -> Assertion +testFib :: Transport -> RemoteTable -> Assertion testFib transport rtable = do   nodes <- replicateM 4 $ newLocalNode transport rtable   done <- newEmptyMVar@@ -408,28 +408,28 @@    takeMVar done -testSpawnReconnect :: Transport -> RemoteTable -> TransportInternals -> Assertion +testSpawnReconnect :: Transport -> RemoteTable -> TransportInternals -> Assertion testSpawnReconnect transport rtable transportInternals = do-  [node1, node2] <- replicateM 2 $ newLocalNode transport rtable +  [node1, node2] <- replicateM 2 $ newLocalNode transport rtable   let nid1 = localNodeId node1       nid2 = localNodeId node2-  done <- newEmptyMVar -  iv <- newMVar (0 :: Int) +  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) +    _pid1 <- spawn nid1 ($(mkClosure 'signal) incr)     simulateNetworkFailure transportInternals nid2 nid1-    _pid2 <- spawn nid1 ($(mkClosure 'signal) incr) -    _pid3 <- spawn nid1 ($(mkClosure 'signal) incr) +    _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' +    True <- return $ count == 2 || count == 3 -- It depends on which message we get first in 'spawn'      liftIO $ putMVar done () @@ -437,10 +437,10 @@  -- | '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 -> RemoteTable -> Assertion testSpawnTerminate transport rtable = do-  slave  <- newLocalNode transport rtable -  master <- newLocalNode transport rtable +  slave  <- newLocalNode transport rtable+  master <- newLocalNode transport rtable   masterDone <- newEmptyMVar    runProcess master $ do@@ -476,5 +476,5 @@ main :: IO () main = do   Right transport <- createTransportExposeInternals "127.0.0.1" "8080" defaultTCPParameters-  let rtable = __remoteTable . __remoteTableDecl $ initRemoteTable +  let rtable = __remoteTable . __remoteTableDecl $ initRemoteTable   defaultMain (tests transport rtable)

@@ -0,0 +1,162 @@+{-# LANGUAGE DeriveDataTypeable        #-}+{-# OPTIONS_GHC -fno-warn-orphans      #-}+module Main where++import Control.Concurrent.MVar+  ( MVar+  , newEmptyMVar+  , putMVar+  , takeMVar+  )+import Control.Distributed.Process+import Control.Distributed.Process.Node+  ( forkProcess+  , newLocalNode+  , initRemoteTable+  , closeLocalNode+  , LocalNode)+import Data.Binary()+import Data.Typeable()+import Network.Transport.TCP+import Prelude hiding (catch, log)+import Test.Framework+  ( Test+  , defaultMain+  , testGroup+  )+import Test.HUnit (Assertion)+import Test.HUnit.Base (assertBool)+import Test.Framework.Providers.HUnit (testCase)++-- these utilities have been cribbed from distributed-process-platform+-- we should really find a way to share them...++-- | A mutable cell containing a test result.+type TestResult a = MVar a++delayedAssertion :: (Eq a) => String -> LocalNode -> a ->+                    (TestResult a -> Process ()) -> Assertion+delayedAssertion note localNode expected testProc = do+  result <- newEmptyMVar+  _ <- forkProcess localNode $ testProc result+  assertComplete note result expected++assertComplete :: (Eq a) => String -> MVar a -> a -> IO ()+assertComplete msg mv a = do+  b <- takeMVar mv+  assertBool msg (a == b)++stash :: TestResult a -> a -> Process ()+stash mvar x = liftIO $ putMVar mvar x++------++testLocalDeadProcessInfo :: TestResult (Maybe ProcessInfo) -> Process ()+testLocalDeadProcessInfo result = do+  pid <- spawnLocal $ do "finish" <- expect; return ()+  mref <- monitor pid+  send pid "finish"+  _ <- receiveWait [+      matchIf (\(ProcessMonitorNotification ref' pid' r) ->+                    ref' == mref && pid' == pid && r == DiedNormal)+              (\p -> return p)+    ]+  getProcessInfo pid >>= stash result++testLocalLiveProcessInfo :: TestResult Bool -> Process ()+testLocalLiveProcessInfo result = do+  self <- getSelfPid+  node <- getSelfNode+  register "foobar" self++  mon <- liftIO $ newEmptyMVar+  -- TODO: we can't get the mailbox's length+  -- mapM (send self) ["hello", "there", "mr", "process"]+  pid <- spawnLocal $ do+       link self+       mRef <- monitor self+       stash mon mRef+       "die" <- expect+       return ()++  monRef <- liftIO $ takeMVar mon++  mpInfo <- getProcessInfo self+  case mpInfo of+    Nothing -> stash result False+    Just p  -> verifyPInfo p pid monRef node+  where verifyPInfo :: ProcessInfo+                    -> ProcessId+                    -> MonitorRef+                    -> NodeId+                    -> Process ()+        verifyPInfo pInfo pid mref node =+          stash result $ infoNode pInfo     == node           &&+                         infoLinks pInfo    == [pid]          &&+                         infoMonitors pInfo == [(pid, mref)]  &&+--                         infoMessageQueueLength pInfo == Just 4 &&+                         infoRegisteredNames pInfo == ["foobar"]++testRemoteLiveProcessInfo :: LocalNode -> Assertion+testRemoteLiveProcessInfo node1 = do+  serverAddr <- liftIO $ newEmptyMVar :: IO (MVar ProcessId)+  liftIO $ launchRemote serverAddr+  serverPid <- liftIO $ takeMVar serverAddr+  withActiveRemote node1 $ \result -> do+    self <- getSelfPid+    link serverPid+    -- our send op shouldn't overtake link or monitor requests AFAICT+    -- so a little table tennis should get us synchronised properly+    send serverPid (self, "ping")+    "pong" <- expect+    pInfo <- getProcessInfo serverPid+    stash result $ pInfo /= Nothing+  where +    launchRemote :: MVar ProcessId -> IO ()+    launchRemote locMV = do+        node2 <- liftIO $ mkNode "8082"+        _ <- liftIO $ forkProcess node2 $ do+            self <- getSelfPid+            liftIO $ putMVar locMV self+            _ <- receiveWait [+                  match (\(pid, "ping") -> send pid "pong") +                ]+            "stop" <- expect+            return ()+        return ()++    withActiveRemote :: LocalNode+                     -> ((TestResult Bool -> Process ()) -> Assertion)+    withActiveRemote n = do+      a <- delayedAssertion "getProcessInfo remotePid failed" n True+      return a++tests :: LocalNode -> IO [Test]+tests node1 = do+  return [+    testGroup "Process Info" [+        testCase "testLocalDeadProcessInfo"+            (delayedAssertion+             "expected dead process-info to be ProcessInfoNone"+             node1 (Nothing) testLocalDeadProcessInfo)+      , testCase "testLocalLiveProcessInfo"+            (delayedAssertion+             "expected process-info to be correctly populated"+             node1 True testLocalLiveProcessInfo)+      , testCase "testRemoveLiveProcessInfo"+                 (testRemoteLiveProcessInfo node1)+    ] ]++mkNode :: String -> IO LocalNode+mkNode port = do+  Right (transport1, _) <- createTransportExposeInternals+                                    "127.0.0.1" port defaultTCPParameters+  newLocalNode transport1 initRemoteTable++main :: IO ()+main = do+  node1 <- mkNode "8081"+  testData <- tests node1+  defaultMain testData+  closeLocalNode node1+  return ()