distributed-process (empty) → 0.2.0

raw patch · 20 files changed

+4314/−0 lines, 20 filesdep +ansi-terminaldep +basedep +binarysetup-changed

Dependencies added: ansi-terminal, base, binary, bytestring, containers, data-accessor, ghc-prim, mtl, network-transport, network-transport-tcp, old-locale, random, stm, template-haskell, time, transformers

Files

• LICENSE +31/−0
• Setup.hs +2/−0
• distributed-process.cabal +134/−0
• src/Control/Distributed/Process.hs +333/−0
• src/Control/Distributed/Process/Closure.hs +145/−0
• src/Control/Distributed/Process/Internal/CQueue.hs +105/−0
• src/Control/Distributed/Process/Internal/Closure.hs +133/−0
• src/Control/Distributed/Process/Internal/Closure/BuiltIn.hs +77/−0
• src/Control/Distributed/Process/Internal/Closure/Combinators.hs +182/−0
• src/Control/Distributed/Process/Internal/Closure/TH.hs +167/−0
• src/Control/Distributed/Process/Internal/Dynamic.hs +57/−0
• src/Control/Distributed/Process/Internal/MessageT.hs +130/−0
• src/Control/Distributed/Process/Internal/Node.hs +16/−0
• src/Control/Distributed/Process/Internal/Primitives.hs +496/−0
• src/Control/Distributed/Process/Internal/TypeRep.hs +21/−0
• src/Control/Distributed/Process/Internal/Types.hs +651/−0
• src/Control/Distributed/Process/Node.hs +691/−0
• src/Control/Distributed/Process/Serializable.hs +57/−0
• tests/TestCH.hs +600/−0
• tests/TestClosure.hs +286/−0

@@ -0,0 +1,31 @@+Copyright Well-Typed LLP, 2011-2012++All rights reserved.++Redistribution and use in source and binary forms, with or without+modification, are permitted provided that the following conditions are met:++    * Redistributions of source code must retain the above copyright+      notice, this list of conditions and the following disclaimer.++    * Redistributions in binary form must reproduce the above+      copyright notice, this list of conditions and the following+      disclaimer in the documentation and/or other materials provided+      with the distribution.++    * Neither the name of the owner nor the names of other+      contributors may be used to endorse or promote products derived+      from this software without specific prior written permission.++THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS+"AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT+LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR+A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT+OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,+SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT+LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,+DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY+THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT+(INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE+OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.+

@@ -0,0 +1,2 @@+import Distribution.Simple+main = defaultMain

@@ -0,0 +1,134 @@+Name:          distributed-process +Version:       0.2.0+Cabal-Version: >=1.8+Build-Type:    Simple+License:       BSD3 +License-File:  LICENSE+Copyright:     Well-Typed LLP+Author:        Duncan Coutts, Nicolas Wu, Edsko de Vries+Maintainer:    edsko@well-typed.com, dcoutts@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 +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.++               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 ++Source-Repository head+  Type:     git+  Location: https://github.com/haskell-distributed/distributed-process+  SubDir:   distributed-process++Library+  Build-Depends:     base >= 4.4 && < 5,+                     binary >= 0.5 && < 0.6,+                     network-transport >= 0.2 && < 0.3,+                     stm >= 2.3 && < 2.5,+                     transformers >= 0.2 && < 0.4,+                     mtl >= 2.0 && < 2.2,+                     data-accessor >= 0.2 && < 0.3,+                     bytestring >= 0.9 && < 0.10,+                     containers >= 0.4 && < 0.5,+                     old-locale >= 1.0 && < 1.1,+                     time >= 1.2 && < 1.3,+                     template-haskell >= 2.6 && < 2.8,+                     random >= 1.0 && < 1.1,+                     ghc-prim >= 0.2 && < 0.3+  Exposed-modules:   Control.Distributed.Process,+                     Control.Distributed.Process.Serializable,+                     Control.Distributed.Process.Closure,+                     Control.Distributed.Process.Node,+                     Control.Distributed.Process.Internal.Primitives,+                     Control.Distributed.Process.Internal.CQueue,+                     Control.Distributed.Process.Internal.Dynamic,+                     Control.Distributed.Process.Internal.Closure,+                     Control.Distributed.Process.Internal.TypeRep,+                     Control.Distributed.Process.Internal.MessageT,+                     Control.Distributed.Process.Internal.Types,+                     Control.Distributed.Process.Internal.Closure.BuiltIn,+                     Control.Distributed.Process.Internal.Closure.Combinators,+                     Control.Distributed.Process.Internal.Closure.TH,+                     Control.Distributed.Process.Internal.Node+  Extensions:        RankNTypes,+                     ScopedTypeVariables,+                     FlexibleInstances,+                     UndecidableInstances,+                     ExistentialQuantification,+                     GADTs,+                     GeneralizedNewtypeDeriving,+                     DeriveDataTypeable,+                     TemplateHaskell+  ghc-options:       -Wall+  HS-Source-Dirs:    src++Test-Suite TestCH+  Type:              exitcode-stdio-1.0+  Main-Is:           TestCH.hs+  Build-Depends:     base >= 4.4 && < 5,+                     binary >= 0.5 && < 0.6,+                     network-transport >= 0.2 && < 0.3,+                     network-transport-tcp >= 0.2 && < 0.3,+                     stm >= 2.3 && < 2.5,+                     transformers >= 0.2 && < 0.4,+                     mtl >= 2.0 && < 2.2,+                     data-accessor >= 0.2 && < 0.3,+                     bytestring >= 0.9 && < 0.10,+                     containers >= 0.4 && < 0.5,+                     old-locale >= 1.0 && < 1.1,+                     time >= 1.2 && < 1.3,+                     template-haskell >= 2.6 && < 2.8,+                     random >= 1.0 && < 1.1,+                     ghc-prim >= 0.2 && < 0.3,+                     ansi-terminal >= 0.5 && < 0.6+  Extensions:        RankNTypes,+                     ScopedTypeVariables,+                     FlexibleInstances,+                     UndecidableInstances,+                     ExistentialQuantification,+                     GADTs,+                     GeneralizedNewtypeDeriving,+                     DeriveDataTypeable,+                     TemplateHaskell+  ghc-options:       -Wall -threaded -rtsopts -with-rtsopts=-N -fno-warn-unused-do-bind +  HS-Source-Dirs:    tests src++Test-Suite TestClosure+  Type:              exitcode-stdio-1.0+  Main-Is:           TestClosure.hs+  Build-Depends:     base >= 4.4 && < 5,+                     binary >= 0.5 && < 0.6,+                     network-transport >= 0.2 && < 0.3,+                     network-transport-tcp >= 0.2 && < 0.3,+                     stm >= 2.3 && < 2.5,+                     transformers >= 0.2 && < 0.4,+                     mtl >= 2.0 && < 2.2,+                     data-accessor >= 0.2 && < 0.3,+                     bytestring >= 0.9 && < 0.10,+                     containers >= 0.4 && < 0.5,+                     old-locale >= 1.0 && < 1.1,+                     time >= 1.2 && < 1.3,+                     template-haskell >= 2.6 && < 2.8,+                     random >= 1.0 && < 1.1,+                     ghc-prim >= 0.2 && < 0.3,+                     ansi-terminal >= 0.5 && < 0.6+  Extensions:        RankNTypes,+                     ScopedTypeVariables,+                     FlexibleInstances,+                     UndecidableInstances,+                     ExistentialQuantification,+                     GADTs,+                     GeneralizedNewtypeDeriving,+                     DeriveDataTypeable,+                     TemplateHaskell+  ghc-options:       -Wall -threaded -rtsopts -with-rtsopts=-N -fno-warn-unused-do-bind +  HS-Source-Dirs:    tests src

@@ -0,0 +1,333 @@+-- | [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 +    ProcessId+  , NodeId+  , Process+  , SendPortId+  , processNodeId+  , sendPortProcessId+  , liftIO -- Reexported for convenience+    -- * Basic messaging+  , send +  , expect+    -- * Channels+  , ReceivePort+  , SendPort+  , sendPortId+  , newChan+  , sendChan+  , receiveChan+  , mergePortsBiased+  , mergePortsRR+    -- * Advanced messaging+  , Match+  , receiveWait+  , receiveTimeout+  , match+  , matchIf+  , matchUnknown+    -- * Process management+  , spawn+  , call+  , terminate+  , ProcessTerminationException(..)+  , SpawnRef+  , getSelfPid+  , getSelfNode+    -- * Monitoring and linking+  , link+  , linkNode+  , linkPort+  , unlink+  , unlinkNode+  , unlinkPort+  , monitor+  , monitorNode+  , monitorPort+  , unmonitor+  , ProcessLinkException(..)+  , NodeLinkException(..)+  , PortLinkException(..)+  , MonitorRef -- opaque+  , ProcessMonitorNotification(..)+  , NodeMonitorNotification(..)+  , PortMonitorNotification(..)+  , DiedReason(..)+    -- * Closures+  , Closure+  , Static+  , unClosure+  , RemoteTable+    -- * Logging+  , say+    -- * Registry+  , register+  , unregister+  , whereis+  , nsend+  , registerRemote+  , unregisterRemote+  , whereisRemote+  , whereisRemoteAsync+  , nsendRemote+  , WhereIsReply(..)+    -- * Auxiliary API+  , catch+  , expectTimeout+  , spawnAsync+  , spawnSupervised+  , spawnLink+  , spawnMonitor+  , DidSpawn(..)+  ) where++import Prelude hiding (catch)+import Data.Typeable (Typeable, typeOf)+import Control.Applicative ((<$>))+import Control.Exception (throw)+import Control.Monad.IO.Class (liftIO)+import Control.Distributed.Process.Internal.MessageT (getLocalNode)+import Control.Distributed.Process.Internal.Dynamic (fromDyn, dynTypeRep)+import Control.Distributed.Process.Internal.Types +  ( RemoteTable+  , NodeId(..)+  , ProcessId(..)+  , Process(..)+  , Closure(..)+  , Static(..)+  , MonitorRef(..)+  , ProcessMonitorNotification(..)+  , NodeMonitorNotification(..)+  , PortMonitorNotification(..)+  , ProcessLinkException(..)+  , NodeLinkException(..)+  , PortLinkException(..)+  , DiedReason(..)+  , SpawnRef(..)+  , DidSpawn(..)+  , Closure(..)+  , SendPort(..)+  , ReceivePort(..)+  , SerializableDict(..)+  , procMsg+  , LocalNode(..)+  , SendPortId(..)+  , WhereIsReply(..)+  )+import Control.Distributed.Process.Internal.Closure.BuiltIn +  ( linkClosure+  , unlinkClosure+  , sendClosure+  , expectClosure+  , serializableDictUnit+  )+import Control.Distributed.Process.Internal.Closure.Combinators (cpSeq, cpBind)+import Control.Distributed.Process.Internal.Primitives+  ( -- Basic messaging+    send +  , expect+    -- Channels+  , newChan+  , sendChan+  , receiveChan+  , mergePortsBiased+  , mergePortsRR+    -- Advanced messaging+  , Match+  , receiveWait+  , receiveTimeout+  , match+  , matchIf+  , matchUnknown+    -- Process management+  , terminate+  , ProcessTerminationException(..)+  , getSelfPid+  , getSelfNode+    -- Monitoring and linking+  , link+  , linkNode+  , linkPort+  , unlink+  , unlinkNode+  , unlinkPort+  , monitor+  , monitorNode+  , monitorPort+  , unmonitor+    -- Logging+  , say+    -- Registry+  , register+  , unregister+  , whereis+  , nsend+  , registerRemote+  , unregisterRemote+  , whereisRemote+  , whereisRemoteAsync+  , nsendRemote+    -- Auxiliary API+  , catch+  , expectTimeout+  , spawnAsync+  )+import Control.Distributed.Process.Internal.Closure (resolveClosure)++-- 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+--     that remote process.+--+-- 2.  'send' may block (when the system TCP buffers are full, while we are+--     trying to establish a connection to the remote endpoint, etc.) but its+--     return does not imply that the remote process received the message (much+--     less processed it)+--+-- 3.  Message delivery is reliable and ordered. That means that if process A+--     sends messages m1, m2, m3 to process B, B will either arrive all three+--     messages in order (m1, m2, m3) or a prefix thereof; messages will not be+--     'missing' (m1, m3) or reordered (m1, m3, m2)+--+-- In order to guarantee (3), we stipulate that+--+-- 3a. We do not garbage collect connections because Network.Transport provides+--     ordering guarantees only *per connection*.+--+-- 3b. Once a connection breaks, we have no way of knowing which messages+--     arrived and which did not; hence, once a connection fails, we assume the+--     remote process to be forever unreachable. Otherwise we might sent m1 and+--     m2, get notified of the broken connection, reconnect, send m3, but only+--     m1 and m3 arrive.+--+-- 3c. As a consequence of (3b) we should not reuse PIDs. If a process dies,+--     we consider it forever unreachable. Hence, new processes should get new+--     IDs or they too would be considered unreachable.+--+-- Main reference for Cloud Haskell is+--+-- [1] "Towards Haskell in the Cloud", Jeff Epstein, Andrew Black and Simon+--     Peyton-Jones.+--       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: +--+-- [3] "Programming Distributed Erlang Applications: Pitfalls and Recipes",+--     Hans Svensson and Lars-Ake Fredlund +--       http://man.lupaworld.com/content/develop/p37-svensson.pdf+-- [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+--     "If I send a message, is it guaranteed to reach the receiver?" of+--     the Erlang FAQ+--       http://www.erlang.org/faq/academic.html+-- [6] "Delivery of Messages", post on erlang-questions+--       http://erlang.org/pipermail/erlang-questions/2012-February/064767.html++--------------------------------------------------------------------------------+-- Primitives are defined in a separate module; here we only define derived   --+-- constructs                                                                 --+--------------------------------------------------------------------------------++-- | Spawn a process+spawn :: NodeId -> Closure (Process ()) -> Process ProcessId+spawn nid proc = do+  us   <- getSelfPid+  -- Since we throw an exception when the remote node dies, we could use+  -- linkNode instead. However, we don't have a way of "scoped" linking so if+  -- we call linkNode here, and unlinkNode after, then we might remove a link+  -- that was already set up+  mRef <- monitorNode nid+  sRef <- spawnAsync nid $ linkClosure us +                   `cpSeq` expectClosure serializableDictUnit+                   `cpSeq` unlinkClosure us+                   `cpSeq` proc+  mPid <- receiveWait +    [ matchIf (\(DidSpawn ref _) -> ref == sRef)+              (\(DidSpawn _ pid) -> return $ Just pid)+    , matchIf (\(NodeMonitorNotification ref _ _) -> ref == mRef)+              (\_ -> return Nothing)+    ]+  unmonitor mRef+  case mPid of+    Nothing  -> fail "spawn: remote node failed"+    Just pid -> send pid () >> return pid++-- | Spawn a process and link to it+--+-- Note that this is just the sequential composition of 'spawn' and 'link'. +-- (The "Unified" semantics that underlies Cloud Haskell does not even support+-- a synchronous link operation)+spawnLink :: NodeId -> Closure (Process ()) -> Process ProcessId+spawnLink nid proc = do+  pid <- spawn nid proc+  link pid+  return pid++-- | Like 'spawnLink', but monitor the spawned process+spawnMonitor :: NodeId -> Closure (Process ()) -> Process (ProcessId, MonitorRef)+spawnMonitor nid proc = do+  pid <- spawn nid proc+  ref <- monitor pid+  return (pid, ref)++-- | Run a process remotely and wait for it to reply+-- +-- We monitor the remote process; if it dies before it can send a reply, we die+-- too+call :: SerializableDict a -> NodeId -> Closure (Process a) -> Process a+call sdict@SerializableDict nid proc = do +  us <- getSelfPid+  (_, mRef) <- spawnMonitor nid (proc `cpBind` sendClosure sdict us)+  -- We are guaranteed to receive the reply before the monitor notification+  -- (if a reply is sent at all)+  -- NOTE: This might not be true if we switch to unreliable delivery.+  mResult <- receiveWait+    [ match (return . Right)+    , matchIf (\(ProcessMonitorNotification ref _ _) -> ref == mRef)+              (\(ProcessMonitorNotification _ _ reason) -> return (Left reason))+    ]+  case mResult of+    Right a  -> unmonitor mRef >> return a+    Left err -> fail $ "call: remote process died: " ++ show err ++-- | Spawn a child process, have the child link to the parent and the parent+-- monitor the child+spawnSupervised :: NodeId +                -> Closure (Process ()) +                -> Process (ProcessId, MonitorRef)+spawnSupervised nid proc = do+  us   <- getSelfPid+  them <- spawn nid (linkClosure us `cpSeq` proc) +  ref  <- monitor them+  return (them, ref)++-- | Deserialize a closure+unClosure :: forall a. Typeable a => Closure a -> Process a+unClosure (Closure (Static label) env) = do+    rtable <- remoteTable <$> procMsg getLocalNode +    case resolveClosure rtable label env of+      Nothing  -> throw . userError $ "Unregistered closure " ++ show label+      Just dyn -> return $ fromDyn dyn (throw (typeError dyn))+  where+    typeError dyn = userError $ "lookupStatic type error: " +                 ++ "cannot match " ++ show (dynTypeRep dyn) +                 ++ " against " ++ show (typeOf (undefined :: a))

@@ -0,0 +1,145 @@+-- | Implementation of 'Closure' that works around the absence of 'static'.+--+-- [Built-in closures]+--+-- We offer a number of standard commonly useful closures.+--+-- [Closure combinators]+--+-- Closures combinators allow to create closures from other closures. For+-- example, 'spawnSupervised' is defined as follows:+--+-- > spawnSupervised :: NodeId +-- >                 -> Closure (Process ()) +-- >                 -> Process (ProcessId, MonitorRef)+-- > spawnSupervised nid proc = do+-- >   us   <- getSelfPid+-- >   them <- spawn nid (linkClosure us `cpSeq` proc) +-- >   ref  <- monitor them+-- >   return (them, ref)+--+-- [User-defined closures]+--+-- Suppose we have a monomorphic function+--+-- > addInt :: Int -> Int -> Int+-- > addInt x y = x + y+--+-- Then the Template Haskell splice+--+-- > remotable ['addInt]+-- +-- creates a function +--+-- > $(mkClosure 'addInt) :: Int -> Closure (Int -> Int)+-- +-- which can be used to partially apply 'addInt' and turn it into a 'Closure',+-- which can be sent across the network. Closures can be deserialized with +--+-- > unClosure :: Typeable a => Closure a -> Process a+--+-- In general, given a monomorphic function @f :: a -> b@ the corresponding +-- function @$(mkClosure 'f)@ will have type @a -> Closure b@.+--+-- The call to 'remotable' will also generate a function+--+-- > __remoteTable :: RemoteTable -> RemoteTable+--+-- which can be used to construct the 'RemoteTable' used to initialize+-- Cloud Haskell. You should have (at most) one call to 'remotable' per module,+-- and compose all created functions when initializing Cloud Haskell:+--+-- > let rtable = M1.__remoteTable+-- >            . M2.__remoteTable+-- >            . ...+-- >            . Mn.__remoteTable+-- >            $ initRemoteTable +--+-- See Section 6, /Faking It/, of /Towards Haskell in the Cloud/ for more info. +--+-- [Serializable Dictionaries]+--+-- Some functions (such as 'sendClosure' or 'returnClosure') require an+-- explicit (reified) serializable dictionary. To create such a dictionary do+--+-- > serializableDictInt :: SerializableDict Int+-- > serializableDictInt = SerializableDict +-- +-- and then pass @'serializableDictInt@ to 'remotable'. This will fail if the+-- type is not serializable.+module Control.Distributed.Process.Closure +  ( -- * User-defined closures+    remotable+  , mkClosure+  , SerializableDict(..)+    -- * Built-in closures+  , linkClosure+  , unlinkClosure+  , sendClosure+  , returnClosure+  , expectClosure+    -- * Generic closure combinators+  , closureApply+  , closureConst+  , closureUnit+    -- * Arrow combinators for processes+  , CP+  , cpIntro+  , cpElim+  , cpId+  , cpComp+  , cpFirst+  , cpSwap+  , cpSecond+  , cpPair+  , cpCopy+  , cpFanOut+  , cpLeft+  , cpMirror+  , cpRight+  , cpEither+  , cpUntag+  , cpFanIn+  , cpApply+    -- * Derived combinators for processes+  , cpBind+  , cpSeq+  ) where ++import Control.Distributed.Process.Internal.Types (SerializableDict(..))+import Control.Distributed.Process.Internal.Closure.TH (remotable, mkClosure)+import Control.Distributed.Process.Internal.Closure.BuiltIn +  ( linkClosure+  , unlinkClosure+  , sendClosure+  , returnClosure+  , expectClosure+  )+import Control.Distributed.Process.Internal.Closure.Combinators +  ( -- Generic combinators+    closureApply+  , closureConst+  , closureUnit+    -- Arrow combinators for processes+  , CP+  , cpIntro+  , cpElim+  , cpId+  , cpComp+  , cpFirst+  , cpSwap+  , cpSecond+  , cpPair+  , cpCopy+  , cpFanOut+  , cpLeft+  , cpMirror+  , cpRight+  , cpEither+  , cpUntag+  , cpFanIn+  , cpApply+    -- Derived process operators+  , cpBind+  , cpSeq+  )

@@ -0,0 +1,105 @@+-- | Concurrent queue for single reader, single writer+module Control.Distributed.Process.Internal.CQueue +  ( CQueue+  , BlockSpec(..)+  , newCQueue+  , enqueue+  , dequeue+  ) where++import Control.Concurrent.MVar (MVar, newMVar, takeMVar, putMVar)+import Control.Concurrent.STM +  ( atomically+  , TChan+  , newTChan+  , writeTChan+  , readTChan+  , tryReadTChan+  )+import Control.Applicative ((<$>), (<*>))+import Control.Exception (mask, onException)+import System.Timeout (timeout)++-- We use a TCHan rather than a Chan so that we have a non-blocking read+data CQueue a = CQueue (MVar [a]) -- Arrived+                       (TChan a)  -- Incoming++newCQueue :: IO (CQueue a)+newCQueue = CQueue <$> newMVar [] <*> atomically newTChan++enqueue :: CQueue a -> a -> IO ()+enqueue (CQueue _arrived incoming) a = atomically $ writeTChan incoming a ++data BlockSpec = +    NonBlocking+  | Blocking+  | Timeout Int++-- | 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 [] 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'++    -- We reverse the accumulator on return only if we find a match+    checkArrived :: [a] -> [a] -> IO ([a], Maybe b)+    checkArrived acc []     = return (acc, Nothing)+    checkArrived acc (x:xs) = +      case check x of+        Just y  -> return (reverse acc ++ xs, Just y)+        Nothing -> checkArrived (x:acc) xs++    -- If we call checkBlocking there may or may not be a timeout+    checkBlocking :: [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 (x:acc)+        Just y  -> putMVar arrived (reverse acc) >> return y ++    -- checkNonBlocking is only called if there is no timeout+    checkNonBlocking :: [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 (x:acc)+          Just y  -> putMVar arrived (reverse acc) >> return (Just y)+        +    check :: a -> Maybe b+    check = checkMatches matches ++    checkMatches :: [a -> Maybe b] -> a -> Maybe b+    checkMatches []     _ = Nothing+    checkMatches (m:ms) a = case m a of Nothing -> checkMatches ms a+                                        Just b  -> Just b

@@ -0,0 +1,133 @@+{-# LANGUAGE MagicHash #-}+module Control.Distributed.Process.Internal.Closure +  ( -- * Runtime support+    initRemoteTable+  , resolveClosure+  ) where++import qualified Data.Map as Map (empty)+import Data.Accessor ((^.))+import Data.ByteString.Lazy (ByteString)+import Data.Binary (decode)+import Data.Typeable (TypeRep)+import Control.Applicative ((<$>))+import Control.Distributed.Process.Internal.Types+  ( RemoteTable(RemoteTable)+  , remoteTableLabel+  , remoteTableDict+  , StaticLabel(..)+  , ProcessId+  , Process+  , Closure(Closure)+  , Static(Static)+  , RuntimeSerializableSupport(..)+  )+import Control.Distributed.Process.Internal.Dynamic+  ( Dynamic(Dynamic)+  , toDyn+  , dynApp+  , dynApply+  , unsafeCoerce#+  )+import Control.Distributed.Process.Internal.TypeRep () -- Binary instances  +import qualified Control.Distributed.Process.Internal.Closure.BuiltIn as BuiltIn+  (remoteTable)++--------------------------------------------------------------------------------+-- Runtime support for closures                                               --+--------------------------------------------------------------------------------++-- | Initial (empty) remote-call meta data+initRemoteTable :: RemoteTable+initRemoteTable = BuiltIn.remoteTable (RemoteTable Map.empty Map.empty)++resolveClosure :: RemoteTable -> StaticLabel -> ByteString -> Maybe Dynamic+-- Built-in closures+resolveClosure rtable ClosureSend env = do+    rss <- rtable ^. remoteTableDict typ +    rssSend rss `dynApply` toDyn pid +  where+    (typ, pid) = decode env :: (TypeRep, ProcessId)+resolveClosure rtable ClosureReturn env = do+    rss <- rtable ^. remoteTableDict typ +    rssReturn rss `dynApply` toDyn arg +  where+    (typ, arg) = decode env :: (TypeRep, ByteString)+resolveClosure rtable ClosureExpect env = +    rssExpect <$> rtable ^. remoteTableDict typ+  where+    typ = decode env :: TypeRep+-- Generic closure combinators+resolveClosure rtable ClosureApply env = do +    f <- resolveClosure rtable labelf envf+    x <- resolveClosure rtable labelx envx+    f `dynApply` x+  where+    (labelf, envf, labelx, envx) = decode env +resolveClosure _rtable ClosureConst env = +  return $ Dynamic (decode env) (unsafeCoerce# const)+resolveClosure _rtable ClosureUnit _env =+  return $ toDyn ()+-- Arrow combinators+resolveClosure _rtable CpId env =+    return $ Dynamic (decode env) (unsafeCoerce# cpId)+  where+    cpId :: forall a. a -> Process a+    cpId = return+resolveClosure _rtable CpComp  env =+    return $ Dynamic (decode env) (unsafeCoerce# cpComp)+  where+    cpComp :: forall a b c. (a -> Process b) -> (b -> Process c) -> a -> Process c+    cpComp p q a = p a >>= q +resolveClosure _rtable CpFirst env =+    return $ Dynamic (decode env) (unsafeCoerce# cpFirst)+  where+    cpFirst :: forall a b c. (a -> Process b) -> (a, c) -> Process (b, c)+    cpFirst p (a, c) = do b <- p a ; return (b, c) +resolveClosure _rtable CpSwap env =+    return $ Dynamic (decode env) (unsafeCoerce# cpSwap) +  where+    cpSwap :: forall a b. (a, b) -> Process (b, a)+    cpSwap (a, b) = return (b, a) +resolveClosure _rtable CpCopy env =+    return $ Dynamic (decode env) (unsafeCoerce# cpCopy)+  where+    cpCopy :: forall a. a -> Process (a, a)+    cpCopy a = return (a, a) +resolveClosure _rtable CpLeft env =+    return $ Dynamic (decode env) (unsafeCoerce# cpLeft)+  where+    cpLeft :: forall a b c. (a -> Process b) -> Either a c -> Process (Either b c)+    cpLeft p (Left a)  = do b <- p a ; return (Left b) +    cpLeft _ (Right c) = return (Right c) +resolveClosure _rtable CpMirror env =+    return $ Dynamic (decode env) (unsafeCoerce# cpMirror)+  where+    cpMirror :: forall a b. Either a b -> Process (Either b a)+    cpMirror (Left a)  = return (Right a) +    cpMirror (Right b) = return (Left b)+resolveClosure _rtable CpUntag env =+    return $ Dynamic (decode env) (unsafeCoerce# cpUntag)+  where+    cpUntag :: forall a. Either a a -> Process a+    cpUntag (Left a)  = return a+    cpUntag (Right a) = return a+resolveClosure rtable CpApply env =+    return $ Dynamic typApply (unsafeCoerce# cpApply)+  where+    cpApply :: forall a b. (Closure (a -> Process b), a) -> Process b +    cpApply (Closure (Static flabel) fenv, x) = do+      let Just f = resolveClosure rtable flabel fenv+          Dynamic typResult val = f `dynApp` Dynamic typA (unsafeCoerce# x) +      if typResult == typProcB +        then unsafeCoerce# val+        else error $ "Type error in cpApply: "+                  ++ "mismatch between " ++ show typResult +                  ++ " and " ++ show typProcB++    (typApply, typA, typProcB) = decode env++-- User defined closures+resolveClosure rtable (UserStatic label) env = do+  val <- rtable ^. remoteTableLabel label +  dynApply val (toDyn env)

@@ -0,0 +1,77 @@+{-# LANGUAGE TemplateHaskell #-}+module Control.Distributed.Process.Internal.Closure.BuiltIn +  ( -- * Runtime support for the builtin closures+    remoteTable+    -- * Serialization dictionaries+  , serializableDictUnit+    -- * Closures+  , linkClosure+  , unlinkClosure+  , sendClosure+  , returnClosure+  , expectClosure+  ) where++import Data.Binary (encode)+import Data.Typeable (typeOf)+import Control.Distributed.Process.Internal.Primitives (link, unlink)+import Control.Distributed.Process.Internal.Types +  ( ProcessId+  , Closure+  , Process+  , RemoteTable+  , SerializableDict(..)+  , Closure(..)+  , Static(..)+  , StaticLabel(..)+  )+import Control.Distributed.Process.Internal.Closure.TH (remotable, mkClosure)+import Control.Distributed.Process.Internal.TypeRep () -- Binary instances++serializableDictUnit :: SerializableDict ()+serializableDictUnit = SerializableDict++remotable ['link, 'unlink, 'serializableDictUnit] ++remoteTable :: RemoteTable -> RemoteTable+remoteTable = __remoteTable++--------------------------------------------------------------------------------+-- TH generated closures                                                      --+--------------------------------------------------------------------------------++-- | Closure version of 'link'+linkClosure :: ProcessId -> Closure (Process ())+linkClosure = $(mkClosure 'link)++-- | Closure version of 'unlink'+unlinkClosure :: ProcessId -> Closure (Process ())+unlinkClosure = $(mkClosure 'unlink)++--------------------------------------------------------------------------------+-- Polymorphic closures                                                       --+--                                                                            --+-- TODO: These functions take a SerializableDict as argument. When we get     --+-- proper support for static, ideally this argument disappears completely;    --+-- but if not, it should turn into a static (SerializableDict a). We don't    --+-- require them to be "static" here because we don't have a pure 'unstatic'   --+-- function, and hence have no way of turning a static (SerializableDict a)   --+-- into an actual SerialziableDict a (we need that in order to pattern match  --+-- on the dictionary and bring the type class dictionary into scope, so that  --+-- we can call 'encode' (for instance in 'returnClosure').                    --+--------------------------------------------------------------------------------++-- | Closure version of 'send'+sendClosure :: forall a. SerializableDict a -> ProcessId -> Closure (a -> Process ())+sendClosure SerializableDict pid =+  Closure (Static ClosureSend) (encode (typeOf (undefined :: a), pid)) ++-- | Return any value+returnClosure :: forall a. SerializableDict a -> a -> Closure (Process a)+returnClosure SerializableDict val =+  Closure (Static ClosureReturn) (encode (typeOf (undefined :: a), encode val))++-- | Closure version of 'expect'+expectClosure :: forall a. SerializableDict a -> Closure (Process a)+expectClosure SerializableDict =+  Closure (Static ClosureExpect) (encode (typeOf (undefined :: a)))

@@ -0,0 +1,182 @@+module Control.Distributed.Process.Internal.Closure.Combinators +  ( -- * Generic combinators+    closureApply+  , closureConst+  , closureUnit+    -- * Arrow combinators for processes+  , CP+  , cpIntro+  , cpElim+  , cpId+  , cpComp+  , cpFirst+  , cpSwap+  , cpSecond+  , cpPair+  , cpCopy+  , cpFanOut+  , cpLeft+  , cpMirror+  , cpRight+  , cpEither+  , cpUntag+  , cpFanIn+  , cpApply+    -- * Derived process operators+  , cpBind+  , cpSeq+  ) where++import Prelude hiding (lookup)+import qualified Data.ByteString.Lazy as BS (empty)+import Data.Binary (encode)+import Data.Typeable (typeOf, Typeable)+import Control.Distributed.Process.Internal.Types+  ( Closure(..)+  , Static(..)+  , StaticLabel(..)+  , Process+  )+import Control.Distributed.Process.Internal.TypeRep () -- Binary instances++--------------------------------------------------------------------------------+-- Generic closure combinators                                                -- +--------------------------------------------------------------------------------++closureApply :: Closure (a -> b) -> Closure a -> Closure b+closureApply (Closure (Static labelf) envf) (Closure (Static labelx) envx) = +  Closure (Static ClosureApply) $ encode (labelf, envf, labelx, envx)++closureConst :: forall a b. (Typeable a, Typeable b) +          => Closure (a -> b -> a)+closureConst = Closure (Static ClosureConst) (encode $ typeOf aux)+  where+    aux :: a -> b -> a+    aux = undefined++closureUnit :: Closure ()+closureUnit = Closure (Static ClosureUnit) BS.empty++--------------------------------------------------------------------------------+-- Arrow combinators for processes                                            -- +--------------------------------------------------------------------------------++type CP a b = Closure (a -> Process b)++cpIntro :: (Typeable a, Typeable b)+        => Closure (Process b) -> CP a b +cpIntro = closureApply closureConst ++cpElim :: Typeable a +       => CP () a -> Closure (Process a)+cpElim = flip closureApply closureUnit ++cpId :: forall a. Typeable a +     => CP a a +cpId = Closure (Static CpId) (encode $ typeOf aux)+  where+    aux :: a -> Process a+    aux = undefined++cpComp :: forall a b c. (Typeable a, Typeable b, Typeable c) +       => CP a b -> CP b c -> CP a c+cpComp f g = comp `closureApply` f `closureApply` g +  where+    comp :: Closure ((a -> Process b) -> (b -> Process c) -> a -> Process c)+    comp = Closure (Static CpComp) (encode $ typeOf aux)+    +    aux :: (a -> Process b) -> (b -> Process c) -> a -> Process c+    aux = undefined++cpFirst :: forall a b c. (Typeable a, Typeable b, Typeable c)+        => CP a b -> CP (a, c) (b, c)+cpFirst = closureApply first+  where+    first :: Closure ((a -> Process b) -> (a, c) -> Process (b, c))+    first = Closure (Static CpFirst) (encode $ typeOf aux)++    aux :: (a -> Process b) -> (a, c) -> Process (b, c)+    aux = undefined++cpSwap :: forall a b. (Typeable a, Typeable b)+       => CP (a, b) (b, a)+cpSwap = Closure (Static CpSwap) (encode $ typeOf aux)+  where+    aux :: (a, b) -> Process (b, a)+    aux = undefined++cpSecond :: (Typeable a, Typeable b, Typeable c)+         => CP a b -> CP (c, a) (c, b)+cpSecond f = cpSwap `cpComp` cpFirst f `cpComp` cpSwap++cpPair :: (Typeable a, Typeable a', Typeable b, Typeable b')+        => CP a b -> CP a' b' -> CP (a, a') (b, b')+cpPair f g = cpFirst f `cpComp` cpSecond g++cpCopy :: forall a. Typeable a +       => CP a (a, a)+cpCopy = Closure (Static CpCopy) (encode $ typeOf aux)+  where+    aux :: a -> Process (a, a)+    aux = undefined++cpFanOut :: (Typeable a, Typeable b, Typeable c)+         => CP a b -> CP a c -> CP a (b, c)+cpFanOut f g = cpCopy `cpComp` (f `cpPair` g)         ++cpLeft :: forall a b c. (Typeable a, Typeable b, Typeable c)+       => CP a b -> CP (Either a c) (Either b c)+cpLeft = closureApply left+  where+    left :: Closure ((a -> Process b) -> Either a c -> Process (Either b c)) +    left = Closure (Static CpLeft) (encode $ typeOf aux)++    aux :: (a -> Process b) -> Either a c -> Process (Either b c)+    aux = undefined++cpMirror :: forall a b. (Typeable a, Typeable b)+         => CP (Either a b) (Either b a)+cpMirror = Closure (Static CpMirror) (encode $ typeOf aux)+  where+    aux :: Either a b -> Process (Either b a)+    aux = undefined++cpRight :: forall a b c. (Typeable a, Typeable b, Typeable c)+        => CP a b -> CP (Either c a) (Either c b)+cpRight f = cpMirror `cpComp` cpLeft f `cpComp` cpMirror ++cpEither :: (Typeable a, Typeable a', Typeable b, Typeable b')+         => CP a b -> CP a' b' -> CP (Either a a') (Either b b')+cpEither f g = cpLeft f `cpComp` cpRight g++cpUntag :: forall a. Typeable a+        => CP (Either a a) a+cpUntag = Closure (Static CpUntag) (encode $ typeOf aux)+  where+    aux :: Either a a -> Process a+    aux = undefined++cpFanIn :: (Typeable a, Typeable b, Typeable c) +        => CP a c -> CP b c -> CP (Either a b) c+cpFanIn f g = (f `cpEither` g) `cpComp` cpUntag ++cpApply :: forall a b. (Typeable a, Typeable b)+        => CP (CP a b, a) b+cpApply = Closure (Static CpApply) $ encode ( typeOf aux+                                            , typeOf (undefined :: a) +                                            , typeOf (undefined :: Process b)+                                            )+  where+    aux :: (Closure (a -> Process b), a) -> Process b+    aux = undefined++--------------------------------------------------------------------------------+-- Some derived operators for processes                                       -- +--------------------------------------------------------------------------------++cpBind :: (Typeable a, Typeable b) +       => Closure (Process a) -> Closure (a -> Process b) -> Closure (Process b)+cpBind x f = cpElim $ cpIntro x `cpComp` f++cpSeq :: Closure (Process ()) -> Closure (Process ()) -> Closure (Process ())+cpSeq p q = p `cpBind` cpIntro q

@@ -0,0 +1,167 @@+-- | Template Haskell support+--+-- (In a separate file for convenience)+module Control.Distributed.Process.Internal.Closure.TH +  ( -- * User-level API+    remotable+  , mkClosure+  ) where++import Prelude hiding (lookup)+import Data.ByteString.Lazy (ByteString)+import Data.Binary (encode, decode)+import Data.Accessor ((^=))+import Data.Typeable (typeOf)+import Control.Applicative ((<$>))+import Language.Haskell.TH +  ( -- Q monad and operations+    Q+  , reify+    -- Names+  , Name+  , mkName+  , nameBase+    -- Algebraic data types+  , Dec+  , Exp+  , Type(AppT, ArrowT)+  , Info(VarI)+    -- Lifted constructors+    -- .. Literals+  , stringL+    -- .. Patterns+  , normalB+  , clause+    -- .. Expressions+  , varE+  , litE+   -- .. Top-level declarations+  , funD+  , sigD+  )+import Control.Distributed.Process.Internal.Types+  ( RemoteTable+  , Closure(..)+  , Static(..)+  , StaticLabel(..)+  , Process+  , ProcessId+  , remoteTableLabel+  , remoteTableDict+  , SerializableDict(..)+  , RuntimeSerializableSupport(..)+  )+import Control.Distributed.Process.Internal.Dynamic (Dynamic, toDyn)+import Control.Distributed.Process.Internal.Primitives (send, expect)++--------------------------------------------------------------------------------+-- User-level API                                                             --+--------------------------------------------------------------------------------++-- | Create the closure, decoder, and metadata definitions for the given list+-- of functions+remotable :: [Name] -> Q [Dec] +remotable ns = do+  (closures, inserts) <- unzip <$> mapM generateDefs ns+  rtable <- createMetaData inserts +  return $ concat closures ++ rtable ++-- | Create a closure+-- +-- If @f :: a -> b@ then @mkClosure :: a -> Closure b@. Make sure to pass 'f'+-- as an argument to 'remotable' too.+mkClosure :: Name -> Q Exp+mkClosure = varE . closureName ++--------------------------------------------------------------------------------+-- Internal (Template Haskell)                                                --+--------------------------------------------------------------------------------++-- | Generate the code to add the metadata to the CH runtime+createMetaData :: [Q Exp] -> Q [Dec]+createMetaData is = +  [d| __remoteTable :: RemoteTable -> RemoteTable ;+      __remoteTable = $(compose is)+    |]++-- | Generate the necessary definitions for one function +--+-- Given an (f :: a -> b) in module M, create: +--  1. f__closure :: a -> Closure b,+--  2. registerLabel "M.f" (toDyn ((f . enc) :: ByteString -> b))+-- +-- Moreover, if b is of the form Process c, then additionally create+--  3. registerSender (Process c) (send :: ProcessId -> c -> Process ())+generateDefs :: Name -> Q ([Dec], Q Exp)+generateDefs n = do+  serializableDict <- [t| SerializableDict |]+  mType <- getType n+  case mType of+    Just (origName, ArrowT `AppT` arg `AppT` res) -> do+      (closure, label) <- generateClosure origName (return arg) (return res)+      let decoder = generateDecoder origName (return res)+          insert  = [| registerLabel $(stringE label) (toDyn $decoder) |]+      return (closure, insert)+    Just (origName, sdict `AppT` a) | sdict == serializableDict -> +      return ([], [| registerSerializableDict $(varE n) |])  +    _ -> +      fail $ "remotable: " ++ show n ++ " is not a function"+   +-- | Generate the closure creator (see 'generateDefs')+generateClosure :: Name -> Q Type -> Q Type -> Q ([Dec], String)+generateClosure n arg res = do+    closure <- sequence +      [ sigD (closureName n) [t| $arg -> Closure $res |]+      , sfnD (closureName n) [| Closure (Static (UserStatic ($(stringE label)))) . encode |]  +      ]+    return (closure, label)+  where+    label :: String +    label = show $ n++-- | Generate the decoder (see 'generateDefs')+generateDecoder :: Name -> Q Type -> Q Exp +generateDecoder n res = [| $(varE n) . decode :: ByteString -> $res |]++-- | The name for the function that generates the closure+closureName :: Name -> Name+closureName n = mkName $ nameBase n ++ "__closure"++registerLabel :: String -> Dynamic -> RemoteTable -> RemoteTable+registerLabel label dyn = remoteTableLabel label ^= Just dyn ++registerSerializableDict :: forall a. SerializableDict a -> RemoteTable -> RemoteTable+registerSerializableDict SerializableDict = +  let rss = RuntimeSerializableSupport {+                rssSend   = toDyn (send :: ProcessId -> a -> Process ()) +              , rssReturn = toDyn (return . decode :: ByteString -> Process a)  +              , rssExpect = toDyn (expect :: Process a)+              }+  in remoteTableDict (typeOf (undefined :: a)) ^= Just rss ++--------------------------------------------------------------------------------+-- Generic Template Haskell auxiliary functions                               --+--------------------------------------------------------------------------------++-- | Compose a set of expressions+compose :: [Q Exp] -> Q Exp+compose []     = [| id |]+compose [e]    = e +compose (e:es) = [| $e . $(compose es) |]++-- | Literal string as an expression+stringE :: String -> Q Exp+stringE = litE . stringL++-- | Look up the "original name" (module:name) and type of a top-level function+getType :: Name -> Q (Maybe (Name, Type))+getType name = do +  info <- reify name+  case info of +    VarI origName typ _ _ -> return $ Just (origName, typ)+    _                     -> return Nothing++-- | 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) []] +

@@ -0,0 +1,57 @@+-- | Reimplementation of Dynamic that supports dynBind+--+-- We don't have access to the internal representation of Dynamic, otherwise+-- we would not have to redefine it completely. Note that we use this only+-- internally, so the incompatibility with "our" Dynamic from the standard +-- Dynamic is not important.+{-# LANGUAGE MagicHash #-}+module Control.Distributed.Process.Internal.Dynamic +  ( Dynamic(..)+  , toDyn+  , fromDyn+  , fromDynamic+  , dynTypeRep+  , dynApply+  , dynApp+  , GHC.unsafeCoerce#+  ) where++import Data.Typeable (Typeable, TypeRep, typeOf, funResultTy)+import qualified GHC.Prim as GHC (Any, unsafeCoerce#)+import Data.Maybe (fromMaybe)++data Dynamic = Dynamic TypeRep GHC.Any++toDyn :: Typeable a => a -> Dynamic+toDyn x = Dynamic (typeOf x) (GHC.unsafeCoerce# x)++fromDyn :: Typeable a => Dynamic -> a -> a+fromDyn (Dynamic rep val) a =+  if rep == typeOf a +    then GHC.unsafeCoerce# val+    else a++fromDynamic :: forall a. Typeable a => Dynamic -> Maybe a+fromDynamic (Dynamic rep val) = +  if rep == typeOf (undefined :: a)+    then Just (GHC.unsafeCoerce# val)+    else Nothing++dynTypeRep :: Dynamic -> TypeRep+dynTypeRep (Dynamic t _) = t++instance Show Dynamic where+  show = show . dynTypeRep++dynApply :: Dynamic -> Dynamic -> Maybe Dynamic+dynApply (Dynamic t1 f) (Dynamic t2 x) =+  case funResultTy t1 t2 of+    Just t3 -> Just (Dynamic t3 (GHC.unsafeCoerce# f x))+    Nothing -> Nothing++dynApp :: Dynamic -> Dynamic -> Dynamic+dynApp f x = fromMaybe (error typeError) (dynApply f x)+  where+    typeError  = "Type error in dynamic application.\n" +              ++ "Can't apply function " ++ show f+              ++ " to argument " ++ show x

@@ -0,0 +1,130 @@+-- | Add message sending capability to a monad+-- +-- NOTE: Not thread-safe (you should not do concurrent sends within the same+-- monad).+module Control.Distributed.Process.Internal.MessageT +  ( MessageT+  , runMessageT+  , getLocalNode+  , sendPayload+  , sendBinary+  , sendMessage+  , payloadToMessage+  , createMessage+  ) where++import Data.Binary (Binary, encode)+import qualified Data.ByteString as BSS (ByteString)+import qualified Data.ByteString.Lazy as BSL (toChunks)+import Data.Map (Map)+import qualified Data.Map as Map (empty)+import Data.Accessor (Accessor, accessor, (^=), (^.))+import qualified Data.Accessor.Container as DAC (mapMaybe)+import Control.Category ((>>>))+import Control.Monad (unless, liftM)+import Control.Monad.State (gets, modify, evalStateT)+import Control.Monad.IO.Class (MonadIO, liftIO)+import Control.Concurrent.Chan (writeChan)+import Control.Distributed.Process.Internal.Types+  ( Identifier(..)+  , nodeOf+  , NodeId(nodeAddress) +  , LocalNode(localCtrlChan, localEndPoint)+  , NCMsg(NCMsg, ctrlMsgSender, ctrlMsgSignal)+  , DiedReason(DiedDisconnect)+  , ProcessSignal(Died)+  , MessageT(..)+  , MessageState(..)+  , createMessage+  , messageToPayload+  , payloadToMessage+  )+import Control.Distributed.Process.Serializable (Serializable)  +import qualified Network.Transport as NT +  ( EndPoint +  , Connection+  , connect+  , send+  , Reliability(ReliableOrdered)+  , defaultConnectHints+  )++--------------------------------------------------------------------------------+-- API                                                                        --+--------------------------------------------------------------------------------++runMessageT :: Monad m => LocalNode -> MessageT m a -> m a+runMessageT localNode m = +  evalStateT (unMessageT m) $ initMessageState localNode ++getLocalNode :: Monad m => MessageT m LocalNode+getLocalNode = gets messageLocalNode++sendPayload :: MonadIO m => Identifier -> [BSS.ByteString] -> MessageT m ()+sendPayload them payload = do+  mConn <- connTo them+  didSend <- case mConn of+    Just conn -> do+      didSend <- liftIO $ NT.send conn payload+      case didSend of+        Left _  -> return False+        Right _ -> return True +    Nothing -> return False+  unless didSend $ do+    node <- getLocalNode+    liftIO . writeChan (localCtrlChan node) $ NCMsg+      { ctrlMsgSender = them+      , ctrlMsgSignal = Died them DiedDisconnect+      }++sendBinary :: (MonadIO m, Binary a) => Identifier -> a -> MessageT m ()+sendBinary them = sendPayload them . BSL.toChunks . encode++sendMessage :: (MonadIO m, Serializable a) => Identifier -> a -> MessageT m ()+sendMessage them = sendPayload them . messageToPayload . createMessage++--------------------------------------------------------------------------------+-- Internal                                                                   --+--------------------------------------------------------------------------------++initMessageState :: LocalNode -> MessageState+initMessageState localNode = MessageState {+     messageLocalNode   = localNode +  , _messageConnections = Map.empty+  }++setupConnTo :: MonadIO m => Identifier -> MessageT m (Maybe NT.Connection)+setupConnTo them = do+    endPoint <- localEndPoint `liftM` getLocalNode +    mConn    <- liftIO $ NT.connect endPoint +                                    (nodeAddress . nodeOf $ them) +                                    NT.ReliableOrdered +                                    NT.defaultConnectHints+    case mConn of +      Right conn -> do+        didSend <- liftIO $ NT.send conn (BSL.toChunks . encode $ them)+        case didSend of+          Left _ ->+            return Nothing+          Right () -> do+            modify $ messageConnectionTo them ^= Just conn+            return $ Just conn+      Left _ ->+        return Nothing++connTo :: MonadIO m => Identifier -> MessageT m (Maybe NT.Connection)+connTo them = do+  mConn <- gets (^. messageConnectionTo them)+  case mConn of+    Just conn -> return $ Just conn+    Nothing   -> setupConnTo them++--------------------------------------------------------------------------------+-- Accessors                                                                  --+--------------------------------------------------------------------------------++messageConnections :: Accessor MessageState (Map Identifier NT.Connection)+messageConnections = accessor _messageConnections (\conns st -> st { _messageConnections = conns })++messageConnectionTo :: Identifier -> Accessor MessageState (Maybe NT.Connection)+messageConnectionTo them = messageConnections >>> DAC.mapMaybe them

@@ -0,0 +1,16 @@+module Control.Distributed.Process.Internal.Node +  ( runLocalProcess+  ) where++import Control.Monad.Reader (runReaderT)+import Control.Distributed.Process.Internal.Types +  ( LocalNode+  , LocalProcess+  , Process(unProcess)+  )+import Control.Distributed.Process.Internal.MessageT (runMessageT)  ++-- | Deconstructor for 'Process' (not exported to the public API) +runLocalProcess :: LocalNode -> Process a -> LocalProcess -> IO a+runLocalProcess node proc = runMessageT node . runReaderT (unProcess proc) +

@@ -0,0 +1,496 @@+-- | Cloud Haskell primitives+--+-- 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 +  ( -- * Basic messaging+    send +  , expect+    -- * Channels+  , newChan+  , sendChan+  , receiveChan+  , mergePortsBiased+  , mergePortsRR+    -- * Advanced messaging+  , Match+  , receiveWait+  , receiveTimeout+  , match+  , matchIf+  , matchUnknown+    -- * Process management+  , terminate+  , ProcessTerminationException(..)+  , getSelfPid+  , getSelfNode+    -- * Monitoring and linking+  , link+  , unlink+  , monitor+  , unmonitor+    -- * Logging+  , say+    -- * Registry+  , register+  , unregister+  , whereis+  , nsend+  , registerRemote+  , unregisterRemote+  , whereisRemote+  , whereisRemoteAsync+  , nsendRemote+    -- * Auxiliary API+  , catch+  , expectTimeout+  , spawnAsync+  , linkNode+  , linkPort+  , unlinkNode+  , unlinkPort+  , monitorNode+  , monitorPort+  ) where++import Prelude hiding (catch)+import Data.Binary (decode)+import Data.Typeable (Typeable)+import Data.Time.Clock (getCurrentTime)+import Data.Time.Format (formatTime)+import System.Locale (defaultTimeLocale)+import Control.Monad.Reader (ask)+import Control.Monad.IO.Class (MonadIO, liftIO)+import Control.Applicative ((<$>))+import Control.Exception (Exception, throw)+import qualified Control.Exception as Exception (catch)+import Control.Concurrent.MVar (modifyMVar)+import Control.Concurrent.Chan (writeChan)+import Control.Concurrent.STM +  ( STM+  , atomically+  , orElse+  , newTChan+  , readTChan+  , newTVar+  , readTVar+  , writeTVar+  )+import Control.Distributed.Process.Internal.CQueue (dequeue, BlockSpec(..))+import Control.Distributed.Process.Serializable (Serializable, fingerprint)+import Data.Accessor ((^.), (^:), (^=))+import Control.Distributed.Process.Internal.Types +  ( NodeId(..)+  , ProcessId(..)+  , LocalNode(..)+  , LocalProcess(..)+  , Process(..)+  , Closure(..)+  , Message(..)+  , MonitorRef(..)+  , SpawnRef(..)+  , NCMsg(..)+  , ProcessSignal(..)+  , monitorCounter +  , spawnCounter+  , Closure(..)+  , SendPort(..)+  , ReceivePort(..)+  , channelCounter+  , typedChannelWithId+  , TypedChannel(..)+  , SendPortId(..)+  , Identifier(..)+  , procMsg+  , DidUnmonitor(..)+  , DidUnlinkProcess(..)+  , DidUnlinkNode(..)+  , DidUnlinkPort(..)+  , WhereIsReply(..)+  , createMessage+  )+import Control.Distributed.Process.Internal.MessageT +  ( sendMessage+  , sendBinary+  , getLocalNode+  )  +import Control.Distributed.Process.Internal.Node (runLocalProcess)++--------------------------------------------------------------------------------+-- Basic messaging                                                            --+--------------------------------------------------------------------------------++-- | Send a message+send :: Serializable a => ProcessId -> a -> Process ()+-- This requires a lookup on every send. If we want to avoid that we need to+-- modify serializable to allow for stateful (IO) deserialization+send them msg = procMsg $ sendMessage (ProcessIdentifier them) msg ++-- | Wait for a message of a specific type+expect :: forall a. Serializable a => Process a+expect = receiveWait [match return] ++--------------------------------------------------------------------------------+-- Channels                                                                   --+--------------------------------------------------------------------------------++-- | Create a new typed channel+newChan :: Serializable a => Process (SendPort a, ReceivePort a)+newChan = do+  proc <- ask +  liftIO . modifyMVar (processState proc) $ \st -> do+    chan <- liftIO . atomically $ newTChan+    let lcid  = st ^. channelCounter+        cid   = SendPortId { sendPortProcessId = processId proc+                           , sendPortLocalId   = lcid+                           }+        sport = SendPort cid +        rport = ReceivePortSingle chan+        tch   = TypedChannel chan +    return ( (channelCounter ^: (+ 1))+           . (typedChannelWithId lcid ^= Just tch)+           $ st+           , (sport, rport)+           )++-- | Send a message on a typed channel+sendChan :: Serializable a => SendPort a -> a -> Process ()+sendChan (SendPort cid) msg = procMsg $ sendBinary (SendPortIdentifier cid) msg ++-- | Wait for a message on a typed channel+receiveChan :: Serializable a => ReceivePort a -> Process a+receiveChan = liftIO . atomically . receiveSTM +  where+    receiveSTM :: ReceivePort a -> STM a+    receiveSTM (ReceivePortSingle c) = +      readTChan c+    receiveSTM (ReceivePortBiased ps) =+      foldr1 orElse (map receiveSTM ps)+    receiveSTM (ReceivePortRR psVar) = do+      ps <- readTVar psVar+      a  <- foldr1 orElse (map receiveSTM ps)+      writeTVar psVar (rotate ps)+      return a++    rotate :: [a] -> [a]+    rotate []     = []+    rotate (x:xs) = xs ++ [x]++-- | 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)+mergePortsBiased = return . ReceivePortBiased ++-- | Like 'mergePortsBiased', but with a round-robin scheduler (rather than+-- left-biased)+mergePortsRR :: Serializable a => [ReceivePort a] -> Process (ReceivePort a)+mergePortsRR ps = liftIO . atomically $ ReceivePortRR <$> newTVar ps++--------------------------------------------------------------------------------+-- Advanced messaging                                                         -- +--------------------------------------------------------------------------------++-- | Opaque type used in 'receiveWait' and 'receiveTimeout'+newtype Match b = Match { unMatch :: Message -> Maybe (Process b) }++-- | Test the matches in order against each message in the queue+receiveWait :: [Match b] -> Process b+receiveWait ms = do+  queue <- processQueue <$> ask+  Just proc <- liftIO $ dequeue queue Blocking (map unMatch ms)+  proc++-- | Like 'receiveWait' but with a timeout.+-- +-- If the timeout is zero do a non-blocking check for matching messages. A+-- non-zero timeout is applied only when waiting for incoming messages (that is,+-- /after/ we have checked the messages that are already in the mailbox).+receiveTimeout :: Int -> [Match b] -> Process (Maybe b)+receiveTimeout t ms = do+  queue <- processQueue <$> ask+  let blockSpec = if t == 0 then NonBlocking else Timeout t+  mProc <- liftIO $ dequeue queue blockSpec (map unMatch ms)+  case mProc of+    Nothing   -> return Nothing+    Just proc -> Just <$> proc++-- | Match against any message of the right type+match :: forall a b. Serializable a => (a -> Process b) -> Match b+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 -> +  let decoded :: a+      decoded = decode . messageEncoding $ msg in+  if messageFingerprint msg == fingerprint (undefined :: a) && c decoded+    then Just $ p decoded +    else Nothing++-- | Remove any message from the queue+matchUnknown :: Process b -> Match b+matchUnknown = Match . const . Just++--------------------------------------------------------------------------------+-- Process management                                                         --+--------------------------------------------------------------------------------++-- | Thrown by 'terminate'+data ProcessTerminationException = ProcessTerminationException+  deriving (Show, Typeable)++instance Exception ProcessTerminationException++-- | Terminate (throws a ProcessTerminationException)+terminate :: Process a+terminate = liftIO $ throw ProcessTerminationException++-- | Our own process ID+getSelfPid :: Process ProcessId+getSelfPid = processId <$> ask ++-- | Get the node ID of our local node+getSelfNode :: Process NodeId+getSelfNode = localNodeId <$> procMsg getLocalNode++--------------------------------------------------------------------------------+-- Monitoring and linking                                                     --+--------------------------------------------------------------------------------++-- | Link to a remote process (asynchronous)+--+-- Note that 'link' provides unidirectional linking (see 'spawnSupervised').+-- Linking makes no distinction between normal and abnormal termination of+-- the remote process.+link :: ProcessId -> Process ()+link = sendCtrlMsg Nothing . Link . ProcessIdentifier++-- | Monitor another process (asynchronous)+monitor :: ProcessId -> Process MonitorRef +monitor = monitor' . ProcessIdentifier ++-- | Remove a link (synchronous)+unlink :: ProcessId -> Process ()+unlink pid = do+  unlinkAsync pid+  receiveWait [ matchIf (\(DidUnlinkProcess pid') -> pid' == pid) +                        (\_ -> return ()) +              ]++-- | Remove a node link (synchronous)+unlinkNode :: NodeId -> Process ()+unlinkNode nid = do+  unlinkNodeAsync nid+  receiveWait [ matchIf (\(DidUnlinkNode nid') -> nid' == nid)+                        (\_ -> return ())+              ]++-- | Remove a channel (send port) link (synchronous)+unlinkPort :: SendPort a -> Process ()+unlinkPort sport = do+  unlinkPortAsync sport+  receiveWait [ matchIf (\(DidUnlinkPort cid) -> cid == sendPortId sport)+                        (\_ -> return ())+              ]++-- | Remove a monitor (synchronous)+unmonitor :: MonitorRef -> Process ()+unmonitor ref = do+  unmonitorAsync ref+  receiveWait [ matchIf (\(DidUnmonitor ref') -> ref' == ref)+                        (\_ -> return ())+              ]++--------------------------------------------------------------------------------+-- Auxiliary API                                                              --+--------------------------------------------------------------------------------++-- | Catch exceptions within a process+catch :: Exception e => Process a -> (e -> Process a) -> Process a+catch p h = do+  node  <- procMsg getLocalNode+  lproc <- ask+  let run :: Process a -> IO a+      run proc = runLocalProcess node proc lproc +  liftIO $ Exception.catch (run p) (run . h) ++-- | Like 'expect' but with a timeout+expectTimeout :: forall a. Serializable a => Int -> Process (Maybe a)+expectTimeout timeout = receiveTimeout timeout [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+  spawnRef <- getSpawnRef+  sendCtrlMsg (Just nid) $ Spawn proc spawnRef+  return spawnRef++-- | Monitor a node+monitorNode :: NodeId -> Process MonitorRef+monitorNode = +  monitor' . NodeIdentifier++-- | Monitor a typed channel+monitorPort :: forall a. Serializable a => SendPort a -> Process MonitorRef+monitorPort (SendPort cid) = +  monitor' (SendPortIdentifier cid) ++-- | Remove a monitor (asynchronous)+unmonitorAsync :: MonitorRef -> Process ()+unmonitorAsync = +  sendCtrlMsg Nothing . Unmonitor++-- | Link to a node+linkNode :: NodeId -> Process ()+linkNode = link' . NodeIdentifier ++-- | Link to a channel (send port)+linkPort :: SendPort a -> Process ()+linkPort (SendPort cid) = +  link' (SendPortIdentifier cid)++-- | Remove a process link (asynchronous)+unlinkAsync :: ProcessId -> Process ()+unlinkAsync = +  sendCtrlMsg Nothing . Unlink . ProcessIdentifier++-- | Remove a node link (asynchronous)+unlinkNodeAsync :: NodeId -> Process ()+unlinkNodeAsync = +  sendCtrlMsg Nothing . Unlink . NodeIdentifier++-- | Remove a channel (send port) link (asynchronous)+unlinkPortAsync :: SendPort a -> Process ()+unlinkPortAsync (SendPort cid) = +  sendCtrlMsg Nothing . Unlink $ SendPortIdentifier cid++--------------------------------------------------------------------------------+-- Logging                                                                    --+--------------------------------------------------------------------------------++-- | Log a string+--+-- @say message@ sends a message (time, pid of the current process, message)+-- to the process registered as 'logger'.  By default, this process simply+-- sends the string to 'stderr'. Individual Cloud Haskell backends might+-- replace this with a different logger process, however.+say :: String -> Process ()+say string = do+  now <- liftIO getCurrentTime+  us  <- getSelfPid+  nsend "logger" (formatTime defaultTimeLocale "%c" now, us, string)++--------------------------------------------------------------------------------+-- Registry                                                                   --+--------------------------------------------------------------------------------++-- | Register a process with the local registry (asynchronous).+--+-- The process to be registered does not have to be local itself.+register :: String -> ProcessId -> Process ()+register label pid = +  sendCtrlMsg Nothing (Register label (Just pid))++-- | Register a process with a remote registry (asynchronous).+--+-- The process to be registered does not have to live on the same remote node.+registerRemote :: NodeId -> String -> ProcessId -> Process ()+registerRemote nid label pid = +  sendCtrlMsg (Just nid) (Register label (Just pid)) ++-- | Remove a process from the local registry (asynchronous).+unregister :: String -> Process ()+unregister label = +  sendCtrlMsg Nothing (Register label Nothing)++-- | Remove a process from a remote registry (asynchronous).+unregisterRemote :: NodeId -> String -> Process ()+unregisterRemote nid label =+  sendCtrlMsg (Just nid) (Register label Nothing)++-- | Query the local process registry (synchronous).+whereis :: String -> Process (Maybe ProcessId)+whereis label = do+  sendCtrlMsg Nothing (WhereIs label)+  receiveWait [ matchIf (\(WhereIsReply label' _) -> label == label')+                        (\(WhereIsReply _ mPid) -> return mPid)+              ]++-- | Query a remote process registry (synchronous)+whereisRemote :: NodeId -> String -> Process (Maybe ProcessId)+whereisRemote nid label = do+  whereisRemoteAsync nid label+  receiveWait [ matchIf (\(WhereIsReply label' _) -> label == label')+                        (\(WhereIsReply _ mPid) -> return mPid)+              ]++-- | Query a remote process registry (asynchronous)+--+-- Reply will come in the form of a 'WhereIsReply' message+whereisRemoteAsync :: NodeId -> String -> Process ()+whereisRemoteAsync nid label = +  sendCtrlMsg (Just nid) (WhereIs label)++-- | Named send to a process in the local registry (asynchronous) +nsend :: Serializable a => String -> a -> Process ()+nsend label msg = +  sendCtrlMsg Nothing (NamedSend label (createMessage msg))++-- | Named send to a process in a remote registry (asynchronous)+nsendRemote :: Serializable a => NodeId -> String -> a -> Process ()+nsendRemote nid label msg = +  sendCtrlMsg (Just nid) (NamedSend label (createMessage msg))++--------------------------------------------------------------------------------+-- Auxiliary functions                                                        --+--------------------------------------------------------------------------------++getMonitorRefFor :: Identifier -> Process MonitorRef+getMonitorRefFor ident = do+  proc <- ask+  liftIO $ modifyMVar (processState proc) $ \st -> do +    let counter = st ^. monitorCounter +    return ( monitorCounter ^: (+ 1) $ st+           , MonitorRef ident counter +           )++getSpawnRef :: Process SpawnRef+getSpawnRef = do+  proc <- ask+  liftIO $ modifyMVar (processState proc) $ \st -> do+    let counter = st ^. spawnCounter+    return ( spawnCounter ^: (+ 1) $ st+           , SpawnRef counter+           )++-- | Monitor a process/node/channel+monitor' :: Identifier -> Process MonitorRef+monitor' ident = do+  monitorRef <- getMonitorRefFor ident +  sendCtrlMsg Nothing $ Monitor monitorRef+  return monitorRef++-- | Link to a process/node/channel+link' :: Identifier -> Process ()+link' = sendCtrlMsg Nothing . Link++-- Send a control message+sendCtrlMsg :: Maybe NodeId  -- ^ Nothing for the local node+            -> ProcessSignal -- ^ Message to send +            -> Process ()+sendCtrlMsg mNid signal = do            +  us <- getSelfPid+  let msg = NCMsg { ctrlMsgSender = ProcessIdentifier us+                  , ctrlMsgSignal = signal+                  }+  case mNid of+    Nothing -> do+      ctrlChan <- localCtrlChan <$> procMsg getLocalNode +      liftIO $ writeChan ctrlChan msg +    Just nid ->+      procMsg $ sendBinary (NodeIdentifier nid) msg+

@@ -0,0 +1,21 @@+-- | 'Binary' instances for 'TypeRep'+{-# OPTIONS_GHC -fno-warn-orphans #-}+module Control.Distributed.Process.Internal.TypeRep () where++import Control.Applicative ((<$>), (<*>))+import Data.Binary (Binary(get, put))+import Data.Typeable.Internal (TypeRep(..), TyCon(..))+import GHC.Fingerprint.Type (Fingerprint(..))++instance Binary Fingerprint where+  put (Fingerprint hi lo) = put hi >> put lo+  get = Fingerprint <$> get <*> get ++instance Binary TypeRep where+  put (TypeRep fp tyCon ts) = put fp >> put tyCon >> put ts+  get = TypeRep <$> get <*> get <*> get++instance Binary TyCon where+  put (TyCon hash package modul name) = put hash >> put package >> put modul >> put name+  get = TyCon <$> get <*> get <*> get <*> get+

@@ -0,0 +1,651 @@+-- | Types used throughout the Cloud Haskell framework+--+-- 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.+{-# LANGUAGE MagicHash #-}+module Control.Distributed.Process.Internal.Types+  ( -- * Node and process identifiers +    NodeId(..)+  , LocalProcessId(..)+  , ProcessId(..)+  , Identifier(..)+  , nodeOf+    -- * Local nodes and processes+  , LocalNode(..)+  , LocalNodeState(..)+  , LocalProcess(..)+  , LocalProcessState(..)+  , Process(..)+  , procMsg+    -- * Typed channels+  , LocalSendPortId +  , SendPortId(..)+  , TypedChannel(..)+  , SendPort(..)+  , ReceivePort(..)+    -- * Closures+  , StaticLabel(..)+  , Static(..) +  , Closure(..)+  , RemoteTable(..)+  , SerializableDict(..)+  , RuntimeSerializableSupport(..)+    -- * Messages +  , Message(..)+  , createMessage+  , messageToPayload+  , payloadToMessage+    -- * Node controller user-visible data types +  , MonitorRef(..)+  , ProcessMonitorNotification(..)+  , NodeMonitorNotification(..)+  , PortMonitorNotification(..)+  , ProcessLinkException(..)+  , NodeLinkException(..)+  , PortLinkException(..)+  , DiedReason(..)+  , DidUnmonitor(..)+  , DidUnlinkProcess(..)+  , DidUnlinkNode(..)+  , DidUnlinkPort(..)+  , SpawnRef(..)+  , DidSpawn(..)+  , WhereIsReply(..)+    -- * Node controller internal data types +  , NCMsg(..)+  , ProcessSignal(..)+    -- * MessageT monad+  , MessageT(..)+  , MessageState(..)+    -- * Accessors+  , localProcesses+  , localPidCounter+  , localPidUnique+  , localProcessWithId+  , monitorCounter+  , spawnCounter+  , channelCounter+  , typedChannels+  , typedChannelWithId+  , remoteTableLabels+  , remoteTableDicts+  , remoteTableLabel+  , remoteTableDict+  ) where++import Data.Map (Map)+import Data.Int (Int32)+import Data.Typeable (Typeable, TypeRep)+import Data.Binary (Binary(put, get), putWord8, getWord8, encode)+import qualified Data.ByteString as BSS (ByteString, concat)+import qualified Data.ByteString.Lazy as BSL +  ( ByteString+  , toChunks+  , splitAt+  , fromChunks+  )+import Data.Accessor (Accessor, accessor)+import qualified Data.Accessor.Container as DAC (mapMaybe)+import Control.Category ((>>>))+import Control.Exception (Exception)+import Control.Concurrent (ThreadId)+import Control.Concurrent.MVar (MVar)+import Control.Concurrent.Chan (Chan)+import Control.Concurrent.STM (TChan, TVar)+import qualified Network.Transport as NT (EndPoint, EndPointAddress, Connection)+import Control.Applicative (Applicative, (<$>), (<*>))+import Control.Monad.Reader (MonadReader(..), ReaderT)+import Control.Monad.IO.Class (MonadIO)+import Control.Monad.State (MonadState, StateT)+import qualified Control.Monad.Trans.Class as Trans (lift)+import Control.Distributed.Process.Serializable +  ( Fingerprint+  , Serializable+  , fingerprint+  , encodeFingerprint+  , sizeOfFingerprint+  , decodeFingerprint+  , showFingerprint+  )+import Control.Distributed.Process.Internal.CQueue (CQueue)+import Control.Distributed.Process.Internal.Dynamic (Dynamic) ++--------------------------------------------------------------------------------+-- Node and process identifiers                                               --+--------------------------------------------------------------------------------++-- | Node identifier +newtype NodeId = NodeId { nodeAddress :: NT.EndPointAddress }+  deriving (Eq, Ord, Binary)++instance Show NodeId where+  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 +  { lpidUnique  :: Int32+  , lpidCounter :: Int32+  }+  deriving (Eq, Ord, Typeable)++instance Show LocalProcessId where+  show = show . lpidCounter ++-- | Process identifier+data ProcessId = ProcessId +  { -- | The ID of the node the process is running on+    processNodeId  :: NodeId+    -- | Node-local identifier for the process+  , processLocalId :: LocalProcessId +  }+  deriving (Eq, Ord, Typeable)++instance Show ProcessId where+  show (ProcessId (NodeId addr) (LocalProcessId _ lid)) +    = "pid://" ++ show addr ++ ":" ++ show lid++-- | Union of all kinds of identifiers +data Identifier = +    NodeIdentifier NodeId+  | ProcessIdentifier ProcessId +  | SendPortIdentifier SendPortId+  deriving (Eq, Ord)++instance Show Identifier where+  show (NodeIdentifier nid)     = show nid+  show (ProcessIdentifier pid)  = show pid+  show (SendPortIdentifier cid) = show cid++nodeOf :: Identifier -> NodeId+nodeOf (NodeIdentifier nid)     = nid+nodeOf (ProcessIdentifier pid)  = processNodeId pid+nodeOf (SendPortIdentifier cid) = processNodeId (sendPortProcessId cid)++--------------------------------------------------------------------------------+-- Local nodes and processes                                                  --+--------------------------------------------------------------------------------++-- | Local nodes+data LocalNode = LocalNode +  { -- | 'NodeId' of the node+    localNodeId :: NodeId+    -- | The network endpoint associated with this node +  , localEndPoint :: NT.EndPoint +    -- | Local node state +  , localState :: MVar LocalNodeState+    -- | Channel for the node controller+  , localCtrlChan :: Chan NCMsg+    -- | Runtime lookup table for supporting closures+    -- TODO: this should be part of the CH state, not the local endpoint state+  , remoteTable :: RemoteTable +  }++-- | Local node state+data LocalNodeState = LocalNodeState +  { _localProcesses  :: Map LocalProcessId LocalProcess+  , _localPidCounter :: Int32+  , _localPidUnique  :: Int32+  }++-- | Processes running on our local node+data LocalProcess = LocalProcess +  { processQueue  :: CQueue Message +  , processId     :: ProcessId+  , processState  :: MVar LocalProcessState+  , processThread :: ThreadId+  }++-- | Local process state+data LocalProcessState = LocalProcessState+  { _monitorCounter :: Int32+  , _spawnCounter   :: Int32+  , _channelCounter :: Int32+  , _typedChannels  :: Map LocalSendPortId TypedChannel +  }++-- | The Cloud Haskell 'Process' type+newtype Process a = Process { +    unProcess :: ReaderT LocalProcess (MessageT IO) a +  }+  deriving (Functor, Monad, MonadIO, MonadReader LocalProcess, Typeable, Applicative)++procMsg :: MessageT IO a -> Process a+procMsg = Process . Trans.lift ++--------------------------------------------------------------------------------+-- Typed channels                                                             --+--------------------------------------------------------------------------------++type LocalSendPortId = Int32++-- | A send port is identified by a SendPortId.+--+-- You cannot send directly to a SendPortId; instead, use 'newChan'+-- to create a SendPort.+data SendPortId = SendPortId {+    -- | The ID of the process that will receive messages sent on this port+    sendPortProcessId :: ProcessId+    -- | Process-local ID of the channel+  , sendPortLocalId   :: LocalSendPortId+  }+  deriving (Eq, Ord)++instance Show SendPortId where+  show (SendPortId (ProcessId (NodeId addr) (LocalProcessId _ plid)) clid)  +    = "cid://" ++ show addr ++ ":" ++ show plid ++ ":" ++ show clid++data TypedChannel = forall a. Serializable a => TypedChannel (TChan a)++-- | The send send of a typed channel (serializable)+newtype SendPort a = SendPort { +    -- | The (unique) ID of this send port+    sendPortId :: SendPortId +  }+  deriving (Typeable, Binary, Show, Eq, Ord)++-- | The receive end of a typed channel (not serializable)+data ReceivePort a = +    -- | A single receive port+    ReceivePortSingle (TChan a)+    -- | A left-biased combination of receive ports +  | ReceivePortBiased [ReceivePort a] +    -- | A round-robin combination of receive ports+  | ReceivePortRR (TVar [ReceivePort a]) ++--------------------------------------------------------------------------------+-- Closures                                                                   --+--------------------------------------------------------------------------------++data StaticLabel = +    UserStatic String+  -- Built-in closures +  | ClosureReturn+  | ClosureSend +  | ClosureExpect+  -- Generic closure combinators+  | ClosureApply+  | ClosureConst+  | ClosureUnit+  -- Arrow combinators for processes+  | CpId+  | CpComp +  | CpFirst+  | CpSwap+  | CpCopy+  | CpLeft+  | CpMirror+  | CpUntag+  | CpApply+  deriving (Typeable, Show)++-- | A static value is one that is bound at top-level.+newtype Static a = Static StaticLabel +  deriving (Typeable, Show)++-- | A closure is a static value and an encoded environment+data Closure a = Closure (Static (BSL.ByteString -> a)) BSL.ByteString+  deriving (Typeable, Show)++-- | Used to fake 'static' (see paper)+data RemoteTable = RemoteTable {+    -- | If the user creates a closure of type @a -> Closure b@ from a function+    -- @f : a -> b@, then '_remoteTableLabels' should have an entry for "f"+    -- of type @ByteString -> b@ (basically, @f . encode@)+    _remoteTableLabels  :: Map String Dynamic +    -- | Runtime counterpart to SerializableDict +  , _remoteTableDicts   :: Map TypeRep RuntimeSerializableSupport +  }++-- | Reification of 'Serializable' (see "Control.Distributed.Process.Closure")+data SerializableDict a where+    SerializableDict :: Serializable a => SerializableDict a+  deriving (Typeable)++-- | Runtime support for implementing "polymorphic" functions with a +-- Serializable qualifier (sendClosure, returnClosure, ..). +--+-- We don't attempt to keep this minimal, but instead just add functions as+-- convenient. This will be replaced anyway once 'static' has been implemented.+data RuntimeSerializableSupport = RuntimeSerializableSupport {+   rssSend   :: Dynamic+ , rssReturn :: Dynamic+ , rssExpect :: Dynamic+ }++--------------------------------------------------------------------------------+-- Messages                                                                   --+--------------------------------------------------------------------------------++-- | Messages consist of their typeRep fingerprint and their encoding+data Message = Message +  { messageFingerprint :: Fingerprint +  , messageEncoding    :: BSL.ByteString+  }++instance Show Message where+  show (Message fp enc) = show enc ++ " :: " ++ showFingerprint fp [] ++-- | Turn any serialiable term into a message+createMessage :: Serializable a => a -> Message+createMessage a = Message (fingerprint a) (encode a)++-- | Serialize a message+messageToPayload :: Message -> [BSS.ByteString]+messageToPayload (Message fp enc) = encodeFingerprint fp : BSL.toChunks enc++-- | Deserialize a message+payloadToMessage :: [BSS.ByteString] -> Message+payloadToMessage payload = Message fp msg+  where+    (encFp, msg) = BSL.splitAt (fromIntegral sizeOfFingerprint) +                 $ BSL.fromChunks payload +    fp = decodeFingerprint . BSS.concat . BSL.toChunks $ encFp++--------------------------------------------------------------------------------+-- Node controller user-visible data types                                    --+--------------------------------------------------------------------------------++-- | 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+  , monitorRefCounter :: Int32+  }+  deriving (Eq, Ord, Show)++-- | Message sent by process monitors+data ProcessMonitorNotification = +    ProcessMonitorNotification MonitorRef ProcessId DiedReason+  deriving (Typeable, Show)++-- | Message sent by node monitors+data NodeMonitorNotification = +    NodeMonitorNotification MonitorRef NodeId DiedReason+  deriving (Typeable, Show)++-- | Message sent by channel (port) monitors+data PortMonitorNotification = +    PortMonitorNotification MonitorRef SendPortId DiedReason+  deriving (Typeable, Show)++-- | Exceptions thrown when a linked process dies+data ProcessLinkException = +    ProcessLinkException ProcessId DiedReason+  deriving (Typeable, Show)++-- | Exception thrown when a linked node dies+data NodeLinkException = +    NodeLinkException NodeId DiedReason+  deriving (Typeable, Show)++-- | Exception thrown when a linked channel (port) dies+data PortLinkException = +    PortLinkException SendPortId DiedReason+  deriving (Typeable, Show)++instance Exception ProcessLinkException+instance Exception NodeLinkException+instance Exception PortLinkException++-- | Why did a process die?+data DiedReason = +    -- | Normal termination+    DiedNormal+    -- | The process exited with an exception+    -- (provided as 'String' because 'Exception' does not implement 'Binary')+  | DiedException String+    -- | We got disconnected from the process node+  | DiedDisconnect+    -- | The process node died+  | DiedNodeDown+    -- | Invalid (process/node/channel) identifier +  | DiedUnknownId+  deriving (Show, Eq)++-- | (Asynchronous) reply from unmonitor+newtype DidUnmonitor = DidUnmonitor MonitorRef+  deriving (Typeable, Binary)++-- | (Asynchronous) reply from unlink+newtype DidUnlinkProcess = DidUnlinkProcess ProcessId+  deriving (Typeable, Binary)++-- | (Asynchronous) reply from unlinkNode+newtype DidUnlinkNode = DidUnlinkNode NodeId+  deriving (Typeable, Binary)++-- | (Asynchronous) reply from unlinkPort+newtype DidUnlinkPort = DidUnlinkPort SendPortId +  deriving (Typeable, Binary)++-- | 'SpawnRef' are used to return pids of spawned processes+newtype SpawnRef = SpawnRef Int32+  deriving (Show, Binary, Typeable, Eq)++-- | (Asynchronius) reply from 'spawn'+data DidSpawn = DidSpawn SpawnRef ProcessId+  deriving (Show, Typeable)++-- | (Asynchronous) reply from 'whereis'+data WhereIsReply = WhereIsReply String (Maybe ProcessId)+  deriving (Show, Typeable)++--------------------------------------------------------------------------------+-- Node controller internal data types                                        --+--------------------------------------------------------------------------------++-- | Messages to the node controller+data NCMsg = NCMsg +  { ctrlMsgSender :: Identifier +  , ctrlMsgSignal :: ProcessSignal+  }+  deriving Show++-- | Signals to the node controller (see 'NCMsg')+data ProcessSignal =+    Link Identifier +  | Unlink Identifier +  | Monitor MonitorRef+  | Unmonitor MonitorRef+  | Died Identifier DiedReason+  | Spawn (Closure (Process ())) SpawnRef +  | WhereIs String+  | Register String (Maybe ProcessId) -- Nothing to unregister+  | NamedSend String Message+  deriving Show++--------------------------------------------------------------------------------+-- Binary instances                                                           --+--------------------------------------------------------------------------------++instance Binary LocalProcessId where+  put lpid = put (lpidUnique lpid) >> put (lpidCounter lpid)+  get      = LocalProcessId <$> get <*> get++instance Binary ProcessId where+  put pid = put (processNodeId pid) >> put (processLocalId pid)+  get     = ProcessId <$> get <*> get++instance Binary ProcessMonitorNotification where+  put (ProcessMonitorNotification ref pid reason) = put ref >> put pid >> put reason+  get = ProcessMonitorNotification <$> get <*> get <*> get++instance Binary NodeMonitorNotification where+  put (NodeMonitorNotification ref pid reason) = put ref >> put pid >> put reason+  get = NodeMonitorNotification <$> get <*> get <*> get++instance Binary PortMonitorNotification where+  put (PortMonitorNotification ref pid reason) = put ref >> put pid >> put reason+  get = PortMonitorNotification <$> get <*> get <*> get++instance Binary NCMsg where+  put msg = put (ctrlMsgSender msg) >> put (ctrlMsgSignal msg)+  get     = NCMsg <$> get <*> get++instance Binary MonitorRef where+  put ref = put (monitorRefIdent ref) >> put (monitorRefCounter ref)+  get     = MonitorRef <$> get <*> get++instance Binary ProcessSignal where+  put (Link pid)            = putWord8 0 >> put pid+  put (Unlink pid)          = putWord8 1 >> put pid+  put (Monitor ref)         = putWord8 2 >> put ref+  put (Unmonitor ref)       = putWord8 3 >> put ref +  put (Died who reason)     = putWord8 4 >> put who >> put reason+  put (Spawn proc ref)      = putWord8 5 >> put proc >> put ref+  put (WhereIs label)       = putWord8 6 >> put label+  put (Register label pid)  = putWord8 7 >> put label >> put pid+  put (NamedSend label msg) = putWord8 8 >> put label >> put (messageToPayload msg) +  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+      8 -> NamedSend <$> get <*> (payloadToMessage <$> get)+      _ -> fail "ProcessSignal.get: invalid"++instance Binary DiedReason where+  put DiedNormal        = putWord8 0+  put (DiedException e) = putWord8 1 >> put e +  put DiedDisconnect    = putWord8 2+  put DiedNodeDown      = putWord8 3+  put DiedUnknownId     = putWord8 4+  get = do+    header <- getWord8+    case header of+      0 -> return DiedNormal+      1 -> DiedException <$> get+      2 -> return DiedDisconnect+      3 -> return DiedNodeDown+      4 -> return DiedUnknownId +      _ -> fail "DiedReason.get: invalid"++instance Binary (Closure a) where+  put (Closure (Static label) env) = put label >> put env+  get = Closure <$> (Static <$> get) <*> get ++instance Binary DidSpawn where+  put (DidSpawn ref pid) = put ref >> put pid+  get = DidSpawn <$> get <*> get++instance Binary SendPortId where+  put cid = put (sendPortProcessId cid) >> put (sendPortLocalId cid)+  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 +    case header of+      0 -> ProcessIdentifier <$> get+      1 -> NodeIdentifier <$> get+      2 -> SendPortIdentifier <$> get +      _ -> fail "Identifier.get: invalid"++instance Binary StaticLabel where+  put (UserStatic string) = putWord8 0 >> put string+  put ClosureReturn = putWord8 1+  put ClosureSend   = putWord8 2+  put ClosureExpect = putWord8 3+  put ClosureApply  = putWord8 4+  put ClosureConst  = putWord8 5+  put ClosureUnit   = putWord8 6+  put CpId          = putWord8 7+  put CpComp        = putWord8 8+  put CpFirst       = putWord8 9+  put CpSwap        = putWord8 10+  put CpCopy        = putWord8 11 +  put CpLeft        = putWord8 12+  put CpMirror      = putWord8 13+  put CpUntag       = putWord8 14+  put CpApply       = putWord8 15+  get = do+    header <- getWord8+    case header of+      0  -> UserStatic <$> get+      1  -> return ClosureReturn+      2  -> return ClosureSend +      3  -> return ClosureExpect +      4  -> return ClosureApply+      5  -> return ClosureConst+      6  -> return ClosureUnit+      7  -> return CpId+      8  -> return CpComp +      9  -> return CpFirst+      10 -> return CpSwap+      11 -> return CpCopy+      12 -> return CpLeft+      13 -> return CpMirror+      14 -> return CpUntag+      15 -> return CpApply+      _  -> fail "StaticLabel.get: invalid"++instance Binary WhereIsReply where+  put (WhereIsReply label mPid) = put label >> put mPid+  get = WhereIsReply <$> get <*> get++--------------------------------------------------------------------------------+-- Accessors                                                                  --+--------------------------------------------------------------------------------++localProcesses :: Accessor LocalNodeState (Map LocalProcessId LocalProcess)+localProcesses = accessor _localProcesses (\procs st -> st { _localProcesses = procs })++localPidCounter :: Accessor LocalNodeState Int32+localPidCounter = accessor _localPidCounter (\ctr st -> st { _localPidCounter = ctr })++localPidUnique :: Accessor LocalNodeState Int32+localPidUnique = accessor _localPidUnique (\unq st -> st { _localPidUnique = unq })++localProcessWithId :: LocalProcessId -> Accessor LocalNodeState (Maybe LocalProcess)+localProcessWithId lpid = localProcesses >>> DAC.mapMaybe lpid++monitorCounter :: Accessor LocalProcessState Int32+monitorCounter = accessor _monitorCounter (\cnt st -> st { _monitorCounter = cnt })++spawnCounter :: Accessor LocalProcessState Int32+spawnCounter = accessor _spawnCounter (\cnt st -> st { _spawnCounter = cnt })++channelCounter :: Accessor LocalProcessState LocalSendPortId+channelCounter = accessor _channelCounter (\cnt st -> st { _channelCounter = cnt })++typedChannels :: Accessor LocalProcessState (Map LocalSendPortId TypedChannel)+typedChannels = accessor _typedChannels (\cs st -> st { _typedChannels = cs })++typedChannelWithId :: LocalSendPortId -> Accessor LocalProcessState (Maybe TypedChannel)+typedChannelWithId cid = typedChannels >>> DAC.mapMaybe cid++remoteTableLabels :: Accessor RemoteTable (Map String Dynamic)+remoteTableLabels = accessor _remoteTableLabels (\ls tbl -> tbl { _remoteTableLabels = ls })++remoteTableDicts :: Accessor RemoteTable (Map TypeRep RuntimeSerializableSupport)+remoteTableDicts = accessor _remoteTableDicts (\ds tbl -> tbl { _remoteTableDicts = ds })++remoteTableLabel :: String -> Accessor RemoteTable (Maybe Dynamic)+remoteTableLabel label = remoteTableLabels >>> DAC.mapMaybe label++remoteTableDict :: TypeRep -> Accessor RemoteTable (Maybe RuntimeSerializableSupport)+remoteTableDict label = remoteTableDicts >>> DAC.mapMaybe label++--------------------------------------------------------------------------------+-- MessageT monad                                                             --+--------------------------------------------------------------------------------++newtype MessageT m a = MessageT { unMessageT :: StateT MessageState m a }+  deriving (Functor, Monad, MonadIO, MonadState MessageState, Applicative)++data MessageState = MessageState { +     messageLocalNode   :: LocalNode+  , _messageConnections :: Map Identifier NT.Connection +  }

@@ -0,0 +1,691 @@+-- | Local nodes+module Control.Distributed.Process.Node +  ( LocalNode+  , newLocalNode+  , closeLocalNode+  , forkProcess+  , runProcess+  , initRemoteTable+  , localNodeId+  ) where++import Prelude hiding (catch)+import System.IO (fixIO, hPutStrLn, stderr)+import qualified Data.ByteString.Lazy as BSL (fromChunks)+import Data.Binary (decode)+import Data.Map (Map)+import qualified Data.Map as Map +  ( empty+  , lookup+  , insert+  , delete+  , toList+  , partitionWithKey+  )+import qualified Data.List as List (delete)+import Data.Set (Set)+import qualified Data.Set as Set (empty, insert, delete, member)+import Data.Foldable (forM_)+import Data.Maybe (isJust, fromMaybe)+import Data.Typeable (Typeable)+import Control.Category ((>>>))+import Control.Applicative ((<$>))+import Control.Monad (void, when, forever)+import Control.Monad.IO.Class (MonadIO, liftIO)+import Control.Monad.State (MonadState, StateT, evalStateT, gets, modify)+import qualified Control.Monad.Trans.Class as Trans (lift)+import Control.Exception (throwIO, SomeException, Exception, throwTo)+import qualified Control.Exception as Exception (catch)+import Control.Concurrent (forkIO)+import Control.Concurrent.MVar +  ( newMVar +  , withMVar+  , modifyMVar+  , modifyMVar_+  , newEmptyMVar+  , putMVar+  , takeMVar+  )+import Control.Concurrent.Chan (newChan, writeChan, readChan)+import Control.Concurrent.STM (atomically, writeTChan)+import Control.Distributed.Process.Internal.CQueue (enqueue, newCQueue)+import qualified Network.Transport as NT +  ( Transport+  , EndPoint+  , newEndPoint+  , receive+  , Event(..)+  , EventErrorCode(..)+  , TransportError(..)+  , address+  , closeEndPoint+  , ConnectionId+  )+import Data.Accessor (Accessor, accessor, (^.), (^=), (^:))+import qualified Data.Accessor.Container as DAC (mapDefault, mapMaybe)+import System.Random (randomIO)+import Control.Distributed.Process.Internal.Types +  ( RemoteTable+  , NodeId(..)+  , LocalProcessId(..)+  , ProcessId(..)+  , LocalNode(..)+  , LocalNodeState(..)+  , LocalProcess(..)+  , LocalProcessState(..)+  , Process(..)+  , DiedReason(..)+  , NCMsg(..)+  , ProcessSignal(..)+  , localPidCounter+  , localPidUnique+  , localProcessWithId+  , MonitorRef(..)+  , ProcessMonitorNotification(..)+  , NodeMonitorNotification(..)+  , PortMonitorNotification(..)+  , ProcessLinkException(..)+  , NodeLinkException(..)+  , PortLinkException(..)+  , DidUnmonitor(..)+  , DidUnlinkProcess(..)+  , DidUnlinkNode(..)+  , DidUnlinkPort(..)+  , SpawnRef+  , DidSpawn(..)+  , Closure(..)+  , Static(..)+  , Message+  , MessageT+  , TypedChannel(..)+  , Identifier(..)+  , nodeOf+  , SendPortId(..)+  , typedChannelWithId+  , WhereIsReply(..)+  , messageToPayload+  )+import Control.Distributed.Process.Serializable (Serializable)+import Control.Distributed.Process.Internal.MessageT +  ( sendBinary+  , sendMessage+  , sendPayload+  , getLocalNode+  , runMessageT+  , payloadToMessage+  , createMessage+  )+import Control.Distributed.Process.Internal.Dynamic (fromDynamic)+import Control.Distributed.Process.Internal.Closure +  ( initRemoteTable+  , resolveClosure+  )+import Control.Distributed.Process.Internal.Node (runLocalProcess)+import Control.Distributed.Process.Internal.Primitives (expect, register)++--------------------------------------------------------------------------------+-- Initialization                                                             --+--------------------------------------------------------------------------------++-- | Initialize a new local node. +-- +-- Note that proper Cloud Haskell initialization and configuration is still +-- to do.+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 +        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 +    { _localProcesses      = Map.empty+    , _localPidCounter     = 0 +    , _localPidUnique      = unq +    }+  ctrlChan <- newChan+  let node = LocalNode { localNodeId   = NodeId $ NT.address endPoint+                       , localEndPoint = endPoint+                       , localState    = state+                       , localCtrlChan = ctrlChan+                       , remoteTable   = rtable+                       }+  void . forkIO $ runNodeController node +  void . forkIO $ handleIncomingMessages node+  return 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 +  runProcess node $ register "logger" logger++-- | Force-close a local node+--+-- TODO: for now we just close the associated endpoint+closeLocalNode :: LocalNode -> IO ()+closeLocalNode node =+  NT.closeEndPoint (localEndPoint node)++-- | Run a process on a local node and wait for it to finish+runProcess :: LocalNode -> Process () -> IO ()+runProcess node proc = do+  done <- newEmptyMVar+  void $ forkProcess node (proc >> liftIO (putMVar done ()))+  takeMVar done++-- | Spawn a new process on a local node+forkProcess :: LocalNode -> Process () -> IO ProcessId+forkProcess node proc = modifyMVar (localState node) $ \st -> do+  let lpid  = LocalProcessId { lpidCounter = st ^. localPidCounter+                             , lpidUnique  = st ^. localPidUnique+                             }+  let pid   = ProcessId { processNodeId  = localNodeId node +                        , processLocalId = lpid+                        }+  pst <- newMVar LocalProcessState { _monitorCounter = 0+                                   , _spawnCounter   = 0+                                   , _channelCounter = 0+                                   , _typedChannels  = Map.empty+                                   }+  queue <- newCQueue+  (_, lproc) <- fixIO $ \ ~(tid, _) -> do+    let lproc = LocalProcess { processQueue  = queue+                             , processId     = pid+                             , processState  = pst +                             , processThread = tid+                             }+    tid' <- forkIO $ do+      reason <- Exception.catch +        (runLocalProcess node proc lproc >> return DiedNormal)+        (return . DiedException . (show :: SomeException -> String))+      -- [Unified: Table 4, rules termination and exiting]+      modifyMVar_ (localState node) $ +        return . (localProcessWithId lpid ^= Nothing)+      writeChan (localCtrlChan node) NCMsg +        { ctrlMsgSender = ProcessIdentifier pid +        , ctrlMsgSignal = Died (ProcessIdentifier pid) reason +        }+    return (tid', lproc)++  if lpidCounter lpid == maxBound+    then do+      newUnique <- randomIO+      return ( (localProcessWithId lpid ^= Just lproc)+             . (localPidCounter ^= 0)+             . (localPidUnique ^= newUnique)+             $ st+             , pid+             )+    else+      return ( (localProcessWithId lpid ^= Just lproc)+             . (localPidCounter ^: (+ 1))+             $ st+             , pid +             )++handleIncomingMessages :: LocalNode -> IO ()+handleIncomingMessages node = go [] Map.empty Map.empty Set.empty+  where+    go :: [NT.ConnectionId] -- ^ Connections whose purpose we don't yet know +       -> Map NT.ConnectionId LocalProcess -- ^ Connections to local processes+       -> Map NT.ConnectionId TypedChannel -- ^ Connections to typed channels+       -> Set NT.ConnectionId              -- ^ Connections to our controller+       -> IO () +    go uninitConns procs chans ctrls = do+      event <- NT.receive endpoint+      case event of+        NT.ConnectionOpened cid _rel _theirAddr ->+          -- TODO: Check if _rel is ReliableOrdered, and if not, treat as+          -- (**) below.+          go (cid : uninitConns) procs chans ctrls +        NT.Received cid payload -> +          case ( Map.lookup cid procs +               , Map.lookup cid chans+               , cid `Set.member` ctrls+               , cid `elem` uninitConns+               ) of+            (Just proc, _, _, _) -> do+              let msg = payloadToMessage payload+              enqueue (processQueue proc) msg+              go uninitConns procs chans ctrls +            (_, Just (TypedChannel chan), _, _) -> do+              atomically $ writeTChan chan . decode . BSL.fromChunks $ payload+              go uninitConns procs chans ctrls +            (_, _, True, _) -> do+              let ctrlMsg = decode . BSL.fromChunks $ payload+              writeChan ctrlChan ctrlMsg+              go uninitConns procs chans ctrls +            (_, _, _, True) -> +              case decode (BSL.fromChunks payload) of+                ProcessIdentifier pid -> do+                  let lpid = processLocalId pid+                  mProc <- withMVar state $ return . (^. localProcessWithId lpid) +                  case mProc of+                    Just proc -> +                      go (List.delete cid uninitConns) +                         (Map.insert cid proc procs)+                         chans+                         ctrls+                    Nothing ->+                      -- Request for an unknown process. +                      --+                      -- 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). (**)+                      go (List.delete cid uninitConns) procs chans ctrls+                SendPortIdentifier chId -> do+                  let lcid = sendPortLocalId chId+                      lpid = processLocalId (sendPortProcessId chId)+                  mProc <- withMVar state $ return . (^. localProcessWithId lpid)+                  case mProc of+                    Just proc -> do+                      mChannel <- withMVar (processState proc) $ return . (^. typedChannelWithId lcid)+                      case mChannel of+                        Just channel ->+                          go (List.delete cid uninitConns)+                             procs+                             (Map.insert cid channel chans)+                             ctrls+                        Nothing ->+                          -- Unknown typed channel+                          -- TODO (**) above+                          go (List.delete cid uninitConns) procs chans ctrls+                    Nothing ->+                      -- Unknown process+                      -- TODO (**) above+                      go (List.delete cid uninitConns) procs chans ctrls+                NodeIdentifier _ ->+                  go (List.delete cid uninitConns)+                     procs+                     chans+                     (Set.insert cid ctrls)+            _ ->+              -- Unexpected message+              -- TODO (**) above +              go uninitConns procs chans ctrls+        NT.ConnectionClosed cid -> +          go (List.delete cid uninitConns) +             (Map.delete cid procs)+             (Map.delete cid chans)+             (Set.delete cid ctrls)+        NT.ErrorEvent (NT.TransportError (NT.EventConnectionLost (Just theirAddr) _) _) -> do +          -- [Unified table 9, rule node_disconnect]+          let nid = NodeIdentifier $ NodeId theirAddr+          writeChan ctrlChan NCMsg +            { ctrlMsgSender = nid+            , ctrlMsgSignal = Died nid DiedDisconnect+            }+        NT.ErrorEvent (NT.TransportError (NT.EventConnectionLost Nothing _) _) ->+          -- TODO: We should treat an asymetrical connection loss (incoming+          -- connection broken, but outgoing connection still potentially ok)+          -- as a fatal error on the part of the remote node (like (**), above)+          fail "handleIncomingMessages: TODO"+        NT.ErrorEvent (NT.TransportError NT.EventEndPointFailed str) ->+          fail $ "Cloud Haskell fatal error: end point failed: " ++ str +        NT.ErrorEvent (NT.TransportError NT.EventTransportFailed str) ->+          fail $ "Cloud Haskell fatal error: transport failed: " ++ str +        NT.EndPointClosed ->+          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"+    +    state    = localState node+    endpoint = localEndPoint node+    ctrlChan = localCtrlChan node++--------------------------------------------------------------------------------+-- Top-level access to the node controller                                    --+--------------------------------------------------------------------------------++runNodeController :: LocalNode -> IO ()+runNodeController node =+  runMessageT node (evalStateT (unNC nodeController) initNCState)++--------------------------------------------------------------------------------+-- Internal data types                                                        --+--------------------------------------------------------------------------------++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))+     -- Process registry+  , _registry :: Map String ProcessId+  }++newtype NC a = NC { unNC :: StateT NCState (MessageT IO) a }+  deriving (Functor, Monad, MonadIO, MonadState NCState)++ncMsg :: MessageT IO a -> NC a+ncMsg = NC . Trans.lift++initNCState :: NCState+initNCState = NCState { _links    = Map.empty+                      , _monitors = Map.empty+                      , _registry = Map.empty+                      }++--------------------------------------------------------------------------------+-- Core functionality                                                         --+--------------------------------------------------------------------------------++-- [Unified: Table 7]+nodeController :: NC ()+nodeController = do+  node <- ncMsg getLocalNode +  forever $ do+    msg  <- liftIO $ readChan (localCtrlChan node)++    -- [Unified: Table 7, rule nc_forward] +    case destNid (ctrlMsgSignal msg) of+      Just nid' | nid' /= localNodeId node -> +        ncMsg $ sendBinary (NodeIdentifier nid') msg+      _ -> +        return ()++    case msg of+      NCMsg (ProcessIdentifier from) (Link them) ->+        ncEffectMonitor from them Nothing+      NCMsg (ProcessIdentifier from) (Monitor ref) ->+        ncEffectMonitor from (monitorRefIdent ref) (Just ref)+      NCMsg (ProcessIdentifier from) (Unlink them) ->+        ncEffectUnlink from them +      NCMsg (ProcessIdentifier from) (Unmonitor ref) ->+        ncEffectUnmonitor from ref+      NCMsg _from (Died ident reason) ->+        ncEffectDied ident reason+      NCMsg (ProcessIdentifier from) (Spawn proc ref) ->+        ncEffectSpawn from proc ref+      NCMsg _from (Register label pid) ->+        ncEffectRegister label pid+      NCMsg (ProcessIdentifier from) (WhereIs label) ->+        ncEffectWhereIs from label+      NCMsg from (NamedSend label msg') ->+        ncEffectNamedSend from label msg'+      unexpected ->+        error $ "nodeController: unexpected message " ++ show unexpected++-- [Unified: Table 10]+ncEffectMonitor :: ProcessId        -- ^ Who's watching? +                -> Identifier       -- ^ Who's being watched?+                -> Maybe MonitorRef -- ^ 'Nothing' to link+                -> NC ()+ncEffectMonitor from them mRef = do+  node <- ncMsg getLocalNode +  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 +    (False, True) -> -- [Unified: second rule]+      notifyDied from them DiedUnknownId mRef +    (False, False) -> -- [Unified: third rule]+      ncMsg $ sendBinary (NodeIdentifier $ processNodeId from) NCMsg +        { ctrlMsgSender = NodeIdentifier (localNodeId node)+        , ctrlMsgSignal = Died them DiedUnknownId+        }++-- [Unified: Table 11]+ncEffectUnlink :: ProcessId -> Identifier -> NC ()+ncEffectUnlink from them = do+  node <- ncMsg getLocalNode +  when (isLocal node (ProcessIdentifier from)) $ +    case them of+      ProcessIdentifier pid -> +        postAsMessage from $ DidUnlinkProcess pid +      NodeIdentifier nid -> +        postAsMessage from $ DidUnlinkNode nid+      SendPortIdentifier cid -> +        postAsMessage from $ DidUnlinkPort cid +  modify $ linksFor them ^: Set.delete from++-- [Unified: Table 11]+ncEffectUnmonitor :: ProcessId -> MonitorRef -> NC ()+ncEffectUnmonitor from ref = do+  node <- ncMsg getLocalNode +  when (isLocal node (ProcessIdentifier from)) $ +    postAsMessage from $ DidUnmonitor ref+  modify $ monitorsFor (monitorRefIdent ref) ^: Set.delete (from, ref)++-- [Unified: Table 12]+ncEffectDied :: Identifier -> DiedReason -> NC ()+ncEffectDied ident reason = do+  node <- ncMsg getLocalNode+  (affectedLinks, unaffectedLinks) <- gets (splitNotif ident . (^. links))+  (affectedMons,  unaffectedMons)  <- gets (splitNotif ident . (^. monitors))++  let localOnly = case ident of NodeIdentifier _ -> True ; _ -> False++  forM_ (Map.toList affectedLinks) $ \(them, uss) -> +    forM_ uss $ \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)++  modify $ (links ^= unaffectedLinks) . (monitors ^= unaffectedMons)++-- [Unified: Table 13]+ncEffectSpawn :: ProcessId -> Closure (Process ()) -> SpawnRef -> NC ()+ncEffectSpawn pid cProc ref = do+  mProc <- unClosure cProc+  -- If the closure does not exist, we spawn a process that throws an exception+  -- This allows the remote node to find out what's happening+  let proc = fromMaybe (fail $ "Error: unknown closure " ++ show cProc) mProc+  node <- ncMsg getLocalNode+  pid' <- liftIO $ forkProcess node proc+  ncMsg $ sendMessage (ProcessIdentifier pid) (DidSpawn ref pid') ++-- Unified semantics does not explicitly describe how to implement 'register',+-- but mentions it's "very similar to nsend" (Table 14)+ncEffectRegister :: String -> Maybe ProcessId -> NC ()+ncEffectRegister label mPid = +  modify $ registryFor label ^= mPid+  -- An acknowledgement is not necessary. If we want a synchronous register,+  -- it suffices to send a whereis requiry immediately after the register+  -- (that may not suffice if we do decide for unreliable messaging instead)++-- Unified semantics does not explicitly describe 'whereis'+ncEffectWhereIs :: ProcessId -> String -> NC ()+ncEffectWhereIs from label = do+  mPid <- gets (^. registryFor label)+  ncMsg $ sendMessage (ProcessIdentifier from) (WhereIsReply label mPid)++-- [Unified: Table 14]+ncEffectNamedSend :: Identifier -> String -> Message -> NC ()+ncEffectNamedSend _from label msg = do+  mPid <- gets (^. registryFor label)+  -- If mPid is Nothing, we just ignore the named send (as per Table 14)+  -- TODO: messages don't carry a "from", but if they do, we should set it+  -- to be the 'from' of the original named send, not this node controller+  forM_ mPid $ \pid -> +    ncMsg $ sendPayload (ProcessIdentifier pid) (messageToPayload msg) ++--------------------------------------------------------------------------------+-- Auxiliary                                                                  --+--------------------------------------------------------------------------------++notifyDied :: ProcessId         -- ^ Who to notify?+           -> Identifier        -- ^ Who died?+           -> DiedReason        -- ^ How did they die?+           -> Maybe MonitorRef  -- ^ 'Nothing' for linking+           -> NC ()+notifyDied dest src reason mRef = do+  node <- ncMsg getLocalNode +  case (isLocal node (ProcessIdentifier dest), mRef, src) of+    (True, Just ref, ProcessIdentifier pid) ->+      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 +    (True, Nothing, NodeIdentifier pid) ->+      throwException dest $ NodeLinkException pid reason +    (True, Nothing, SendPortIdentifier pid) ->+      throwException dest $ PortLinkException pid reason +    (False, _, _) ->+      -- TODO: why the change in sender? How does that affect 'reconnect' semantics?+      -- (see [Unified: Table 10]+      ncMsg $ sendBinary (NodeIdentifier $ processNodeId dest) NCMsg+        { 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+-- 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 ++-- | Check if a process is local to our own node+isLocal :: LocalNode -> Identifier -> Bool +isLocal nid ident = nodeOf ident == localNodeId nid++-- | Lookup a local closure +unClosure :: Typeable a => Closure a -> NC (Maybe a)+unClosure (Closure (Static label) env) = do+  rtable <- remoteTable <$> ncMsg getLocalNode+  return (resolveClosure rtable label env >>= fromDynamic)++-- | Check if an identifier refers to a valid local object+isValidLocalIdentifier :: Identifier -> NC Bool+isValidLocalIdentifier ident = do+  node <- ncMsg getLocalNode+  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 +      SendPortIdentifier cid -> do+        let pid   = sendPortProcessId cid+            mProc = nSt ^. localProcessWithId (processLocalId pid)+        case mProc of+          Nothing -> return False+          Just proc -> withMVar (processState proc) $ \pSt -> do+            let mCh = pSt ^. typedChannelWithId (sendPortLocalId cid)+            return $ isJust mCh++--------------------------------------------------------------------------------+-- Messages to local processes                                                --+--------------------------------------------------------------------------------++postAsMessage :: Serializable a => ProcessId -> a -> NC ()+postAsMessage pid = postMessage pid . createMessage  ++postMessage :: ProcessId -> Message -> NC ()+postMessage pid msg = withLocalProc pid $ \p -> enqueue (processQueue p) msg++throwException :: Exception e => ProcessId -> e -> NC ()+throwException pid e = withLocalProc pid $ \p -> +  throwTo (processThread p) e++withLocalProc :: ProcessId -> (LocalProcess -> IO ()) -> NC () +withLocalProc pid p = do+  node <- ncMsg getLocalNode +  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 ++--------------------------------------------------------------------------------+-- Accessors                                                                  --+--------------------------------------------------------------------------------++links :: Accessor NCState (Map Identifier (Set ProcessId))+links = accessor _links (\ls st -> st { _links = ls })++monitors :: Accessor NCState (Map Identifier (Set (ProcessId, MonitorRef)))+monitors = accessor _monitors (\ms st -> st { _monitors = ms })++registry :: Accessor NCState (Map String ProcessId)+registry = accessor _registry (\ry st -> st { _registry = ry })++linksFor :: Identifier -> Accessor NCState (Set ProcessId)+linksFor ident = links >>> DAC.mapDefault Set.empty ident++monitorsFor :: Identifier -> Accessor NCState (Set (ProcessId, MonitorRef))+monitorsFor ident = monitors >>> DAC.mapDefault Set.empty ident++registryFor :: String -> Accessor NCState (Maybe ProcessId)+registryFor ident = registry >>> DAC.mapMaybe ident++-- | @splitNotif ident@ splits a notifications map into those+-- notifications that should trigger when 'ident' fails and those links that+-- should not.+--+-- 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 +--+-- * the notifications for the node specifically+-- * 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 +--+-- * the notifications for that process specifically and +-- * the notifications for typed channels to that process.+splitNotif :: Identifier+           -> Map Identifier a+           -> (Map Identifier a, Map Identifier a)+splitNotif ident = Map.partitionWithKey (const . impliesDeathOf ident)+  where+    -- | Does the death of one entity (node, project, channel) imply the death+    -- of another?+    impliesDeathOf :: Identifier -- ^ Who died +                   -> Identifier -- ^ Who's being watched +                   -> Bool       -- ^ Does this death implies the death of the watchee? +    NodeIdentifier nid `impliesDeathOf` NodeIdentifier nid' = +      nid' == nid+    NodeIdentifier nid `impliesDeathOf` ProcessIdentifier pid =+      processNodeId pid == nid+    NodeIdentifier nid `impliesDeathOf` SendPortIdentifier cid =+      processNodeId (sendPortProcessId cid) == nid+    ProcessIdentifier pid `impliesDeathOf` ProcessIdentifier pid' =+      pid' == pid+    ProcessIdentifier pid `impliesDeathOf` SendPortIdentifier cid =+      sendPortProcessId cid == pid+    SendPortIdentifier cid `impliesDeathOf` SendPortIdentifier cid' =+      cid' == cid+    _ `impliesDeathOf` _ =+      False

@@ -0,0 +1,57 @@+module Control.Distributed.Process.Serializable +  ( Serializable+  , encodeFingerprint+  , decodeFingerprint+  , fingerprint+  , sizeOfFingerprint+  , Fingerprint+  , showFingerprint+  ) where++import Data.Binary (Binary)+import Data.Typeable (Typeable(..))+import Data.Typeable.Internal (TypeRep(TypeRep))+import Numeric (showHex)+import Control.Exception (throw)+import GHC.Fingerprint.Type (Fingerprint(..))+import Data.ByteString (ByteString)+import qualified Data.ByteString as BS+import qualified Data.ByteString.Internal as BSI ( unsafeCreate+                                                 , inlinePerformIO+                                                 , toForeignPtr+                                                 )+import Foreign.Storable (pokeByteOff, peekByteOff, sizeOf)+import Foreign.ForeignPtr (withForeignPtr)++-- | Objects that can be sent across the network+class (Binary a, Typeable a) => Serializable a+instance (Binary a, Typeable a) => Serializable a++-- | Encode type representation as a bytestring+encodeFingerprint :: Fingerprint -> ByteString+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 ++-- | Decode a bytestring into a fingerprint. Throws an IO exception on failure +decodeFingerprint :: ByteString -> Fingerprint+decodeFingerprint bs+  | 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 ++-- | Size of a fingerprint+sizeOfFingerprint :: Int+sizeOfFingerprint = sizeOf (undefined :: Fingerprint)++-- | 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) = +  showString "(" . showHex hi . showString "," . showHex lo . showString ")"

@@ -0,0 +1,600 @@+module Main where ++import Prelude hiding (catch)+import Data.Binary (Binary(..))+import Data.Typeable (Typeable)+import Data.Foldable (forM_)+import Control.Concurrent (forkIO, threadDelay)+import Control.Concurrent.MVar +  ( MVar+  , newEmptyMVar+  , putMVar+  , takeMVar+  , readMVar+  )+import Control.Monad (replicateM_, replicateM)+import Control.Exception (throwIO)+import Control.Applicative ((<$>), (<*>))+import qualified Network.Transport as NT (Transport, closeEndPoint)+import Network.Transport.TCP (createTransport, defaultTCPParameters)+import Control.Distributed.Process+import Control.Distributed.Process.Internal.Types (LocalNode(localEndPoint))+import Control.Distributed.Process.Node+import Control.Distributed.Process.Serializable (Serializable)+import TestAuxiliary++newtype Ping = Ping ProcessId+  deriving (Typeable, Binary, Show)++newtype Pong = Pong ProcessId+  deriving (Typeable, Binary, Show)++-- | The ping server from the paper+ping :: Process ()+ping = do+  Pong partner <- expect+  self <- getSelfPid+  send partner (Ping self)+  ping++-- | Basic ping test+testPing :: NT.Transport -> IO ()+testPing transport = do+  serverAddr <- newEmptyMVar+  clientDone <- newEmptyMVar++  -- Server+  forkIO $ do+    localNode <- newLocalNode transport initRemoteTable+    addr <- forkProcess localNode ping+    putMVar serverAddr addr++  -- Client+  forkIO $ do+    localNode <- newLocalNode transport initRemoteTable+    pingServer <- readMVar serverAddr++    let numPings = 10000++    runProcess localNode $ do+      pid <- getSelfPid+      replicateM_ numPings $ do+        send pingServer (Pong pid)+        Ping _ <- expect+        return ()++    putMVar clientDone ()++  takeMVar clientDone++-- | Monitor or link to a remote node+monitorOrLink :: Bool            -- ^ 'True' for monitor, 'False' for link+              -> ProcessId       -- Process to monitor/link to+              -> Maybe (MVar ()) -- MVar to signal on once the monitor has been set up+              -> Process (Maybe MonitorRef) +monitorOrLink mOrL pid mSignal = do+  result <- if mOrL then Just <$> monitor pid+                    else link pid >> return Nothing+  -- Monitor is asynchronous, which usually does not matter but if we want a+  -- *specific* signal then it does. Therefore we wait an arbitrary delay and+  -- hope that this means the monitor has been set up+  forM_ mSignal $ \signal -> liftIO . forkIO $ threadDelay 100000 >> putMVar signal ()+  return result++monitorTestProcess :: ProcessId       -- Process to monitor/link to+                   -> Bool            -- 'True' for monitor, 'False' for link+                   -> Bool            -- Should we unmonitor?+                   -> DiedReason      -- Expected cause of death+                   -> Maybe (MVar ()) -- Signal for 'monitor set up' +                   -> MVar ()         -- Signal for successful termination+                   -> Process ()+monitorTestProcess theirAddr mOrL un reason monitorSetup done = +  catch (do mRef <- monitorOrLink mOrL theirAddr monitorSetup +            case (un, mRef) of+              (True, Nothing) -> do+                unlink theirAddr+                liftIO $ putMVar done ()+              (True, Just ref) -> do+                unmonitor ref+                liftIO $ putMVar done ()+              (False, ref) -> do+                ProcessMonitorNotification ref' pid reason' <- expect+                True <- return $ Just ref' == ref && pid == theirAddr && mOrL && reason == reason'+                liftIO $ putMVar done ()+        )+        (\(ProcessLinkException pid reason') -> do+            True <- return $ pid == theirAddr && not mOrL && not un && reason == reason'+            liftIO $ putMVar done ()+        )+  ++-- | Monitor a process on an unreachable node +testMonitorUnreachable :: NT.Transport -> Bool -> Bool -> IO ()+testMonitorUnreachable transport mOrL un = do+  deadProcess <- newEmptyMVar+  done <- newEmptyMVar++  forkIO $ do+    localNode <- newLocalNode transport initRemoteTable+    addr <- forkProcess localNode . liftIO $ threadDelay 1000000 +    closeLocalNode localNode+    putMVar deadProcess addr++  forkIO $ do+    localNode <- newLocalNode transport initRemoteTable+    theirAddr <- readMVar deadProcess+    runProcess localNode $+      monitorTestProcess theirAddr mOrL un DiedDisconnect Nothing done ++  takeMVar done++-- | Monitor a process which terminates normally+testMonitorNormalTermination :: NT.Transport -> Bool -> Bool -> IO ()+testMonitorNormalTermination transport mOrL un = do+  monitorSetup <- newEmptyMVar+  monitoredProcess <- newEmptyMVar+  done <- newEmptyMVar++  forkIO $ do+    localNode <- newLocalNode transport initRemoteTable+    addr <- forkProcess localNode $ +      liftIO $ readMVar monitorSetup+    putMVar monitoredProcess addr++  forkIO $ do+    localNode <- newLocalNode transport initRemoteTable+    theirAddr <- readMVar monitoredProcess+    runProcess localNode $ +      monitorTestProcess theirAddr mOrL un DiedNormal (Just monitorSetup) done++  takeMVar done++-- | Monitor a process which terminates abnormally+testMonitorAbnormalTermination :: NT.Transport -> Bool -> Bool -> IO ()+testMonitorAbnormalTermination transport mOrL un = do+  monitorSetup <- newEmptyMVar+  monitoredProcess <- newEmptyMVar+  done <- newEmptyMVar++  let err = userError "Abnormal termination"++  forkIO $ do+    localNode <- newLocalNode transport initRemoteTable+    addr <- forkProcess localNode . liftIO $ do+      readMVar monitorSetup+      throwIO err +    putMVar monitoredProcess addr++  forkIO $ do+    localNode <- newLocalNode transport initRemoteTable+    theirAddr <- readMVar monitoredProcess+    runProcess localNode $ +      monitorTestProcess theirAddr mOrL un (DiedException (show err)) (Just monitorSetup) done++  takeMVar done+    +-- | Monitor a local process that is already dead+testMonitorLocalDeadProcess :: NT.Transport -> Bool -> Bool -> IO ()+testMonitorLocalDeadProcess transport mOrL un = do+  processDead <- newEmptyMVar+  processAddr <- newEmptyMVar+  localNode <- newLocalNode transport initRemoteTable+  done <- newEmptyMVar++  forkIO $ do+    addr <- forkProcess localNode . liftIO $ putMVar processDead ()+    putMVar processAddr addr++  forkIO $ do+    theirAddr <- readMVar processAddr+    readMVar processDead+    runProcess localNode $ do+      monitorTestProcess theirAddr mOrL un DiedUnknownId Nothing done++  takeMVar done++-- | Monitor a remote process that is already dead+testMonitorRemoteDeadProcess :: NT.Transport -> Bool -> Bool -> IO ()+testMonitorRemoteDeadProcess transport mOrL un = do+  processDead <- newEmptyMVar+  processAddr <- newEmptyMVar+  done <- newEmptyMVar++  forkIO $ do+    localNode <- newLocalNode transport initRemoteTable+    addr <- forkProcess localNode . liftIO $ putMVar processDead ()+    putMVar processAddr addr++  forkIO $ do+    localNode <- newLocalNode transport initRemoteTable+    theirAddr <- readMVar processAddr+    readMVar processDead+    runProcess localNode $ do+      monitorTestProcess theirAddr mOrL un DiedUnknownId Nothing done++  takeMVar done++-- | Monitor a process that becomes disconnected+testMonitorDisconnect :: NT.Transport -> Bool -> Bool -> IO ()+testMonitorDisconnect transport mOrL un = do+  processAddr <- newEmptyMVar+  monitorSetup <- newEmptyMVar+  done <- newEmptyMVar++  forkIO $ do+    localNode <- newLocalNode transport initRemoteTable+    addr <- forkProcess localNode . liftIO $ threadDelay 1000000 +    putMVar processAddr addr+    readMVar monitorSetup+    NT.closeEndPoint (localEndPoint localNode)++  forkIO $ do+    localNode <- newLocalNode transport initRemoteTable+    theirAddr <- readMVar processAddr+    runProcess localNode $ do+      monitorTestProcess theirAddr mOrL un DiedDisconnect (Just monitorSetup) done+  +  takeMVar done++data Add       = Add    ProcessId Double Double deriving (Typeable) +data Divide    = Divide ProcessId Double Double deriving (Typeable)+data DivByZero = DivByZero deriving (Typeable)++instance Binary Add where+  put (Add pid x y) = put pid >> put x >> put y+  get = Add <$> get <*> get <*> get++instance Binary Divide where+  put (Divide pid x y) = put pid >> put x >> put y+  get = Divide <$> get <*> get <*> get++instance Binary DivByZero where+  put DivByZero = return ()+  get = return DivByZero++math :: Process ()+math = do+  receiveWait+    [ match (\(Add pid x y) -> send pid (x + y))+    , matchIf (\(Divide _   _ y) -> y /= 0)+              (\(Divide pid x y) -> send pid (x / y))+    , match (\(Divide pid _ _) -> send pid DivByZero)+    ]+  math++-- | Test the math server (i.e., receiveWait)+testMath :: NT.Transport -> IO ()+testMath transport = do+  serverAddr <- newEmptyMVar +  clientDone <- newEmptyMVar++  -- Server+  forkIO $ do+    localNode <- newLocalNode transport initRemoteTable +    addr <- forkProcess localNode math+    putMVar serverAddr addr++  -- Client+  forkIO $ do+    localNode <- newLocalNode transport initRemoteTable+    mathServer <- readMVar serverAddr++    runProcess localNode $ do+      pid <- getSelfPid+      send mathServer (Add pid 1 2)+      3 <- expect :: Process Double  +      send mathServer (Divide pid 8 2)+      4 <- expect :: Process Double+      send mathServer (Divide pid 8 0)+      DivByZero <- expect+      liftIO $ putMVar clientDone ()++  takeMVar clientDone++-- | Send first message (i.e. connect) to an already terminated process+-- (without monitoring); then send another message to a second process on +-- the same remote node (we're checking that the remote node did not die)+testSendToTerminated :: NT.Transport -> IO ()+testSendToTerminated transport = do+  serverAddr1 <- newEmptyMVar+  serverAddr2 <- newEmptyMVar+  clientDone <- newEmptyMVar++  forkIO $ do +    terminated <- newEmptyMVar+    localNode <- newLocalNode transport initRemoteTable+    addr1 <- forkProcess localNode $ liftIO $ putMVar terminated ()+    addr2 <- forkProcess localNode $ ping+    readMVar terminated+    putMVar serverAddr1 addr1+    putMVar serverAddr2 addr2++  forkIO $ do+    localNode <- newLocalNode transport initRemoteTable+    server1 <- readMVar serverAddr1+    server2 <- readMVar serverAddr2+    runProcess localNode $ do+      pid <- getSelfPid+      send server1 "Hi"+      send server2 (Pong pid)+      Ping pid' <- expect+      True <- return $ pid' == server2 +      liftIO $ putMVar clientDone ()++  takeMVar clientDone++-- | Test (non-zero) timeout+testTimeout :: NT.Transport -> IO ()+testTimeout transport = do+  localNode <- newLocalNode transport initRemoteTable+  runProcess localNode $ do+    Nothing <- receiveTimeout 1000000 [match (\(Add _ _ _) -> return ())]+    return ()++-- | Test zero timeout+testTimeout0 :: NT.Transport -> IO ()+testTimeout0 transport = do+  serverAddr <- newEmptyMVar+  clientDone <- newEmptyMVar+  messagesSent <- newEmptyMVar++  forkIO $ do +    localNode <- newLocalNode transport initRemoteTable+    addr <- forkProcess localNode $ do+      liftIO $ readMVar messagesSent >> threadDelay 1000000+      -- Variation on the venerable ping server which uses a zero timeout+      -- Since we wait for all messages to be sent before doing this receive,+      -- we should nevertheless find the right message immediately+      Just partner <- receiveTimeout 0 [match (\(Pong partner) -> return partner)] +      self <- getSelfPid+      send partner (Ping self)+    putMVar serverAddr addr++  forkIO $ do+    localNode <- newLocalNode transport initRemoteTable+    server <- readMVar serverAddr+    runProcess localNode $ do+      pid <- getSelfPid+      -- Send a bunch of messages. A large number of messages that the server+      -- is not interested in, and then a single message that it wants+      replicateM_ 10000 $ send server "Irrelevant message"+      send server (Pong pid)+      liftIO $ putMVar messagesSent ()+      Ping _ <- expect +      liftIO $ putMVar clientDone ()++  takeMVar clientDone++-- | Test typed channels+testTypedChannels :: NT.Transport -> IO ()+testTypedChannels transport = do+  serverChannel <- newEmptyMVar :: IO (MVar (SendPort (SendPort Bool, Int)))+  clientDone <- newEmptyMVar++  forkIO $ do+    localNode <- newLocalNode transport initRemoteTable+    forkProcess localNode $ do+      (serverSendPort, rport) <- newChan+      liftIO $ putMVar serverChannel serverSendPort +      (clientSendPort, i) <- receiveChan rport+      sendChan clientSendPort (even i)+    return ()++  forkIO $ do+    localNode <- newLocalNode transport initRemoteTable+    serverSendPort <- readMVar serverChannel+    runProcess localNode $ do+      (clientSendPort, rport) <- newChan+      sendChan serverSendPort (clientSendPort, 5) +      False <- receiveChan rport+      liftIO $ putMVar clientDone ()+      +  takeMVar clientDone ++-- | Test merging receive ports+testMergeChannels :: NT.Transport -> IO ()+testMergeChannels transport = do+    localNode <- newLocalNode transport initRemoteTable+    testFlat localNode True          "aaabbbccc"+    testFlat localNode False         "abcabcabc"+    testNested localNode True True   "aaabbbcccdddeeefffggghhhiii"+    testNested localNode True False  "adgadgadgbehbehbehcficficfi"+    testNested localNode False True  "abcabcabcdefdefdefghighighi"+    testNested localNode False False "adgbehcfiadgbehcfiadgbehcfi"+    testBlocked localNode True+    testBlocked localNode False+  where+    -- Single layer of merging+    testFlat :: LocalNode -> Bool -> String -> IO () +    testFlat localNode biased expected = +      runProcess localNode $ do+        rs  <- mapM charChannel "abc" +        m   <- mergePorts biased rs +        xs  <- replicateM 9 $ receiveChan m +        True <- return $ xs == expected+        return ()++    -- Two layers of merging+    testNested :: LocalNode -> Bool -> Bool -> String -> IO ()+    testNested localNode biasedInner biasedOuter expected = +      runProcess localNode $ do+        rss  <- mapM (mapM charChannel) ["abc", "def", "ghi"]+        ms   <- mapM (mergePorts biasedInner) rss+        m    <- mergePorts biasedOuter ms+        xs   <- replicateM (9 * 3) $ receiveChan m +        True <- return $ xs == expected+        return ()++    -- 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 ++      forkProcess localNode $ do+        [sa, sb, sc] <- liftIO $ mapM readMVar vs +        mapM_ ((>> liftIO (threadDelay 10000)) . uncurry sendChan) +          [ -- a, b, c+            (sa, 'a')+          , (sb, 'b')+          , (sc, 'c')+            -- a, c, b+          , (sa, 'a')+          , (sc, 'c')+          , (sb, 'b')+            -- b, a, c+          , (sb, 'b')+          , (sa, 'a')+          , (sc, 'c')+            -- b, c, a+          , (sb, 'b')+          , (sc, 'c')+          , (sa, 'a')+            -- c, a, b+          , (sc, 'c')+          , (sa, 'a')+          , (sb, 'b')+            -- c, b, a+          , (sc, 'c')+          , (sb, 'b')+          , (sa, 'a')+          ]++      runProcess localNode $ do+        (ss, rs) <- unzip <$> replicateM 3 newChan+        liftIO $ mapM_ (uncurry putMVar) $ zip vs ss +        m  <- mergePorts biased rs +        xs <- replicateM (6 * 3) $ receiveChan m+        True <- return $ xs == "abcacbbacbcacabcba"+        return ()++    mergePorts :: Serializable a => Bool -> [ReceivePort a] -> Process (ReceivePort a)+    mergePorts True  = mergePortsBiased+    mergePorts False = mergePortsRR ++    charChannel :: Char -> Process (ReceivePort Char)+    charChannel c = do+      (sport, rport) <- newChan+      replicateM_ 3 $ sendChan sport c +      liftIO $ threadDelay 10000 -- Make sure messages have been sent+      return rport++testTerminate :: NT.Transport -> IO ()+testTerminate transport = do+  localNode <- newLocalNode transport initRemoteTable++  pid <- forkProcess localNode $ do+    liftIO $ threadDelay 100000+    terminate++  runProcess localNode $ do+    ref <- monitor pid+    ProcessMonitorNotification ref' pid' (DiedException ex) <- expect+    True <- return $ ref == ref' && pid == pid' && ex == show ProcessTerminationException +    return ()++testMonitorNode :: NT.Transport -> IO ()+testMonitorNode transport = do+  [node1, node2] <- replicateM 2 $ newLocalNode transport initRemoteTable++  closeLocalNode node1++  runProcess node2 $ do+    ref <- monitorNode (localNodeId node1)+    NodeMonitorNotification ref' nid DiedDisconnect <- expect+    True <- return $ ref == ref' && nid == localNodeId node1+    return ()++testMonitorChannel :: NT.Transport -> IO ()+testMonitorChannel transport = do+    [node1, node2] <- replicateM 2 $ newLocalNode transport initRemoteTable+    gotNotification <- newEmptyMVar++    pid <- forkProcess node1 $ do+      sport <- expect :: Process (SendPort ())+      ref <- monitorPort sport+      PortMonitorNotification ref' port' DiedNormal <- expect+      return $ ref' == ref && port' == sendPortId sport +      liftIO $ putMVar gotNotification ()++    runProcess node2 $ do+      (sport, _) <- newChan :: Process (SendPort (), ReceivePort ())+      send pid sport+      liftIO $ threadDelay 100000++    takeMVar gotNotification++testRegistry :: NT.Transport -> IO ()+testRegistry transport = do+  node <- newLocalNode transport initRemoteTable++  pingServer <- forkProcess node ping ++  runProcess node $ do+    register "ping" pingServer+    Just pid <- whereis "ping"+    True <- return $ pingServer == pid +    us <- getSelfPid+    nsend "ping" (Pong us)+    Ping pid' <- expect +    True <- return $ pingServer == pid'+    return ()+    +testRemoteRegistry :: NT.Transport -> IO ()+testRemoteRegistry transport = do+  node1 <- newLocalNode transport initRemoteTable+  node2 <- newLocalNode transport initRemoteTable++  pingServer <- forkProcess node1 ping ++  runProcess node2 $ do+    let nid1 = localNodeId node1+    registerRemote nid1 "ping" pingServer+    Just pid <- whereisRemote nid1 "ping"+    True <- return $ pingServer == pid +    us <- getSelfPid+    nsendRemote nid1 "ping" (Pong us)+    Ping pid' <- expect +    True <- return $ pingServer == pid'+    return ()++main :: IO ()+main = do+  Right transport <- createTransport "127.0.0.1" "8080" defaultTCPParameters+  runTests +    [ ("Ping",             testPing             transport)+    , ("Math",             testMath             transport) +    , ("Timeout",          testTimeout          transport)+    , ("Timeout0",         testTimeout0         transport)+    , ("SendToTerminated", testSendToTerminated transport) +    , ("TypedChannnels",   testTypedChannels    transport)+    , ("MergeChannels",    testMergeChannels    transport)+    , ("Terminate",        testTerminate        transport)+    , ("Registry",         testRegistry         transport)+    , ("RemoteRegistry",   testRemoteRegistry   transport)+      -- Monitoring processes+      --+      -- The "missing" combinations in the list below don't make much sense, as+      -- we cannot guarantee that the monitor reply or link exception will not +      -- happen before the unmonitor or unlink+    , ("MonitorUnreachable",           testMonitorUnreachable         transport True  False)+    , ("MonitorNormalTermination",     testMonitorNormalTermination   transport True  False)+    , ("MonitorAbnormalTermination",   testMonitorAbnormalTermination transport True  False)+    , ("MonitorLocalDeadProcess",      testMonitorLocalDeadProcess    transport True  False)+    , ("MonitorRemoteDeadProcess",     testMonitorRemoteDeadProcess   transport True  False)+    , ("MonitorDisconnect",            testMonitorDisconnect          transport True  False)+    , ("LinkUnreachable",              testMonitorUnreachable         transport False False)+    , ("LinkNormalTermination",        testMonitorNormalTermination   transport False False)+    , ("LinkAbnormalTermination",      testMonitorAbnormalTermination transport False False)+    , ("LinkLocalDeadProcess",         testMonitorLocalDeadProcess    transport False False)+    , ("LinkRemoteDeadProcess",        testMonitorRemoteDeadProcess   transport False False)+    , ("LinkDisconnect",               testMonitorDisconnect          transport False False)+    , ("UnmonitorNormalTermination",   testMonitorNormalTermination   transport True  True)+    , ("UnmonitorAbnormalTermination", testMonitorAbnormalTermination transport True  True)+    , ("UnmonitorDisconnect",          testMonitorDisconnect          transport True  True)+    , ("UnlinkNormalTermination",      testMonitorNormalTermination   transport False True)+    , ("UnlinkAbnormalTermination",    testMonitorAbnormalTermination transport False True)+    , ("UnlinkDisconnect",             testMonitorDisconnect          transport False True)+      -- Monitoring nodes and channels+    , ("MonitorNode",                  testMonitorNode                transport)+    , ("MonitorChannel",               testMonitorChannel             transport)+    ]

@@ -0,0 +1,286 @@+module Main where++import qualified Data.ByteString.Lazy as BSL (empty)+import Control.Monad (join, replicateM)+import Control.Exception (IOException, throw)+import Control.Concurrent (forkIO, threadDelay)+import Control.Concurrent.MVar (MVar, newEmptyMVar, readMVar, takeMVar, putMVar)+import Control.Applicative ((<$>))+import Network.Transport (Transport)+import Network.Transport.TCP (createTransport, defaultTCPParameters)+import Control.Distributed.Process+import Control.Distributed.Process.Closure+import Control.Distributed.Process.Node+import Control.Distributed.Process.Internal.Types (Closure(..), Static(..), StaticLabel(..))+import TestAuxiliary++serializableDictInt :: SerializableDict Int+serializableDictInt = SerializableDict++addInt :: Int -> Int -> Int+addInt x y = x + y++putInt :: Int -> MVar Int -> IO ()+putInt = flip putMVar++sendInt :: ProcessId -> Int -> Process ()+sendInt = send++sendPid :: ProcessId -> Process ()+sendPid toPid = do+  fromPid <- getSelfPid+  send toPid fromPid++factorial :: Int -> Process Int+factorial 0 = return 1+factorial n = (n *) <$> factorial (n - 1) ++factorialOf :: () -> Int -> Process Int+factorialOf () = factorial++returnForTestApply :: (Closure (Int -> Process Int), Int) -> (Closure (Int -> Process Int), Int)+returnForTestApply = id ++wait :: Int -> Process ()+wait = liftIO . threadDelay++$(remotable ['addInt, 'putInt, 'sendInt, 'sendPid, 'factorial, 'factorialOf, 'returnForTestApply, 'wait, 'serializableDictInt])++testSendPureClosure :: Transport -> RemoteTable -> IO ()+testSendPureClosure transport rtable = do+  serverAddr <- newEmptyMVar+  serverDone <- newEmptyMVar++  forkIO $ do +    node <- newLocalNode transport rtable +    addr <- forkProcess node $ do+      cl <- expect+      fn <- unClosure cl :: Process (Int -> Int)+      11 <- return $ fn 6+      liftIO $ putMVar serverDone () +    putMVar serverAddr addr ++  forkIO $ do+    node <- newLocalNode transport rtable +    theirAddr <- readMVar serverAddr+    runProcess node $ send theirAddr ($(mkClosure 'addInt) 5) ++  takeMVar serverDone++testSendIOClosure :: Transport -> RemoteTable -> IO ()+testSendIOClosure transport rtable = do+  serverAddr <- newEmptyMVar+  serverDone <- newEmptyMVar++  forkIO $ do +    node <- newLocalNode transport rtable +    addr <- forkProcess node $ do+      cl <- expect+      io <- unClosure cl :: Process (MVar Int -> IO ())+      liftIO $ do +        someMVar <- newEmptyMVar+        io someMVar +        5 <- readMVar someMVar+        putMVar serverDone () +    putMVar serverAddr addr ++  forkIO $ do+    node <- newLocalNode transport rtable +    theirAddr <- readMVar serverAddr+    runProcess node $ send theirAddr ($(mkClosure 'putInt) 5) ++  takeMVar serverDone++testSendProcClosure :: Transport -> RemoteTable -> IO ()+testSendProcClosure transport rtable = do+  serverAddr <- newEmptyMVar+  clientDone <- newEmptyMVar++  forkIO $ do +    node <- newLocalNode transport rtable +    addr <- forkProcess node $ do+      cl <- expect+      pr <- unClosure cl :: Process (Int -> Process ())+      pr 5+    putMVar serverAddr addr ++  forkIO $ do+    node <- newLocalNode transport rtable +    theirAddr <- readMVar serverAddr+    runProcess node $ do+      pid <- getSelfPid+      send theirAddr ($(mkClosure 'sendInt) pid) +      5 <- expect :: Process Int+      liftIO $ putMVar clientDone ()++  takeMVar clientDone++testSpawn :: Transport -> RemoteTable -> IO ()+testSpawn transport rtable = do+  serverNodeAddr <- newEmptyMVar+  clientDone <- newEmptyMVar++  forkIO $ do+    node <- newLocalNode transport rtable+    putMVar serverNodeAddr (localNodeId node)++  forkIO $ do+    node <- newLocalNode transport rtable+    nid <- readMVar serverNodeAddr+    runProcess node $ do+      pid   <- getSelfPid+      pid'  <- spawn nid ($(mkClosure 'sendPid) pid)+      pid'' <- expect+      True <- return $ pid' == pid''+      liftIO $ putMVar clientDone ()++  takeMVar clientDone++testCall :: Transport -> RemoteTable -> IO ()+testCall transport rtable = do+  serverNodeAddr <- newEmptyMVar+  clientDone <- newEmptyMVar++  forkIO $ do+    node <- newLocalNode transport rtable+    putMVar serverNodeAddr (localNodeId node)++  forkIO $ do+    node <- newLocalNode transport rtable+    nid <- readMVar serverNodeAddr+    runProcess node $ do+      (120 :: Int) <- call serializableDictInt nid ($(mkClosure 'factorial) 5)+      liftIO $ putMVar clientDone ()++  takeMVar clientDone++sendFac :: Int -> ProcessId -> Closure (Process ())+sendFac n pid = $(mkClosure 'factorial) n `cpBind` $(mkClosure 'sendInt) pid++factorial' :: Int -> Closure (Process Int)+factorial' n = returnClosure serializableDictInt n `cpBind` $(mkClosure 'factorialOf) ()++testBind :: Transport -> RemoteTable -> IO ()+testBind transport rtable = do+  node <- newLocalNode transport rtable+  runProcess node $ do+    us <- getSelfPid+    join . unClosure $ sendFac 6 us +    (720 :: Int) <- expect +    return ()++testCallBind :: Transport -> RemoteTable -> IO ()+testCallBind transport rtable = do+  serverNodeAddr <- newEmptyMVar+  clientDone <- newEmptyMVar++  forkIO $ do+    node <- newLocalNode transport rtable+    putMVar serverNodeAddr (localNodeId node)++  forkIO $ do+    node <- newLocalNode transport rtable+    nid <- readMVar serverNodeAddr+    runProcess node $ do+      (120 :: Int) <- call serializableDictInt nid (factorial' 5)+      liftIO $ putMVar clientDone ()++  takeMVar clientDone++testSeq :: Transport -> RemoteTable -> IO ()+testSeq transport rtable = do+  node <- newLocalNode transport rtable+  runProcess node $ do+    us <- getSelfPid+    join . unClosure $ sendFac 5 us `cpSeq` sendFac 6 us+    120 :: Int <- expect+    720 :: Int <- expect+    return ()++testApply :: Transport -> RemoteTable -> IO ()+testApply transport rtable = do+    node <- newLocalNode transport rtable+    runProcess node $ do+      (120 :: Int) <- join (unClosure bar)+      return ()+  where+    bar :: Closure (Process Int)+    bar = closureApply cpApply foo++    foo :: Closure (Closure (Int -> Process Int), Int)+    foo = $(mkClosure 'returnForTestApply) ($(mkClosure 'factorialOf) (), 5) ++-- Test 'spawnSupervised'+--+-- Set up a supervisor, spawn a child, then have a third process monitor the+-- child. The supervisor then throws an exception, the child dies because it+-- was linked to the supervisor, and the third process notices that the child+-- dies.+testSpawnSupervised :: Transport -> RemoteTable -> IO ()+testSpawnSupervised transport rtable = do+    [node1, node2]       <- replicateM 2 $ newLocalNode transport rtable+    [superPid, childPid] <- replicateM 2 $ newEmptyMVar+    thirdProcessDone     <- newEmptyMVar++    forkProcess node1 $ do+      us <- getSelfPid+      liftIO $ putMVar superPid us+      (child, _ref) <- spawnSupervised (localNodeId node2) ($(mkClosure 'wait) 1000000) +      liftIO $ do+        putMVar childPid child+        threadDelay 500000 -- Give the child a chance to link to us+        throw supervisorDeath++    forkProcess node2 $ do+      [super, child] <- liftIO $ mapM readMVar [superPid, childPid]+      ref <- monitor child+      ProcessMonitorNotification ref' pid' (DiedException e) <- expect+      True <- return $ ref' == ref +                    && pid' == child +                    && e == show (ProcessLinkException super (DiedException (show supervisorDeath)))+      liftIO $ putMVar thirdProcessDone ()++    takeMVar thirdProcessDone+  where+    supervisorDeath :: IOException+    supervisorDeath = userError "Supervisor died"++testSpawnInvalid :: Transport -> RemoteTable -> IO ()+testSpawnInvalid transport rtable = do+  node <- newLocalNode transport rtable+  runProcess node $ do+    (pid, ref) <- spawnMonitor (localNodeId node) (Closure (Static (UserStatic "ThisDoesNotExist")) BSL.empty)+    ProcessMonitorNotification ref' pid' _reason <- expect +    -- Depending on the exact interleaving, reason might be NoProc or the exception thrown by the absence of the static closure+    True <- return $ ref' == ref && pid == pid'  +    return ()++testClosureExpect :: Transport -> RemoteTable -> IO ()+testClosureExpect transport rtable = do+  node <- newLocalNode transport rtable+  runProcess node $ do+    nodeId <- getSelfNode+    us     <- getSelfPid+    them   <- spawn nodeId $ expectClosure serializableDictInt `cpBind` $(mkClosure 'sendInt) us+    send them (1234 :: Int)+    (1234 :: Int) <- expect+    return ()++main :: IO ()+main = do+  Right transport <- createTransport "127.0.0.1" "8080" defaultTCPParameters+  let rtable = __remoteTable initRemoteTable +  runTests +    [ ("SendPureClosure", testSendPureClosure transport rtable)+    , ("SendIOClosure",   testSendIOClosure   transport rtable)+    , ("SendProcClosure", testSendProcClosure transport rtable)+    , ("Spawn",           testSpawn           transport rtable)+    , ("Call",            testCall            transport rtable)+    , ("Bind",            testBind            transport rtable)+    , ("CallBind",        testCallBind        transport rtable)+    , ("Seq",             testSeq             transport rtable)+    , ("Apply",           testApply           transport rtable)+    , ("SpawnSupervised", testSpawnSupervised transport rtable)+    , ("SpawnInvalid",    testSpawnInvalid    transport rtable)+    , ("ClosureExpect",   testClosureExpect   transport rtable)+    ]