diff --git a/LICENSE b/LICENSE new file mode 100644 --- /dev/null +++ b/LICENSE @@ -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. + diff --git a/Setup.hs b/Setup.hs new file mode 100644 --- /dev/null +++ b/Setup.hs @@ -0,0 +1,2 @@ +import Distribution.Simple +main = defaultMain diff --git a/distributed-process.cabal b/distributed-process.cabal new file mode 100644 --- /dev/null +++ b/distributed-process.cabal @@ -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 + (), + 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 diff --git a/src/Control/Distributed/Process.hs b/src/Control/Distributed/Process.hs new file mode 100644 --- /dev/null +++ b/src/Control/Distributed/Process.hs @@ -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 +-- (), +-- 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)) diff --git a/src/Control/Distributed/Process/Closure.hs b/src/Control/Distributed/Process/Closure.hs new file mode 100644 --- /dev/null +++ b/src/Control/Distributed/Process/Closure.hs @@ -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 + ) diff --git a/src/Control/Distributed/Process/Internal/CQueue.hs b/src/Control/Distributed/Process/Internal/CQueue.hs new file mode 100644 --- /dev/null +++ b/src/Control/Distributed/Process/Internal/CQueue.hs @@ -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 diff --git a/src/Control/Distributed/Process/Internal/Closure.hs b/src/Control/Distributed/Process/Internal/Closure.hs new file mode 100644 --- /dev/null +++ b/src/Control/Distributed/Process/Internal/Closure.hs @@ -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) diff --git a/src/Control/Distributed/Process/Internal/Closure/BuiltIn.hs b/src/Control/Distributed/Process/Internal/Closure/BuiltIn.hs new file mode 100644 --- /dev/null +++ b/src/Control/Distributed/Process/Internal/Closure/BuiltIn.hs @@ -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))) diff --git a/src/Control/Distributed/Process/Internal/Closure/Combinators.hs b/src/Control/Distributed/Process/Internal/Closure/Combinators.hs new file mode 100644 --- /dev/null +++ b/src/Control/Distributed/Process/Internal/Closure/Combinators.hs @@ -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 diff --git a/src/Control/Distributed/Process/Internal/Closure/TH.hs b/src/Control/Distributed/Process/Internal/Closure/TH.hs new file mode 100644 --- /dev/null +++ b/src/Control/Distributed/Process/Internal/Closure/TH.hs @@ -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) []] + diff --git a/src/Control/Distributed/Process/Internal/Dynamic.hs b/src/Control/Distributed/Process/Internal/Dynamic.hs new file mode 100644 --- /dev/null +++ b/src/Control/Distributed/Process/Internal/Dynamic.hs @@ -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 diff --git a/src/Control/Distributed/Process/Internal/MessageT.hs b/src/Control/Distributed/Process/Internal/MessageT.hs new file mode 100644 --- /dev/null +++ b/src/Control/Distributed/Process/Internal/MessageT.hs @@ -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 diff --git a/src/Control/Distributed/Process/Internal/Node.hs b/src/Control/Distributed/Process/Internal/Node.hs new file mode 100644 --- /dev/null +++ b/src/Control/Distributed/Process/Internal/Node.hs @@ -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) + diff --git a/src/Control/Distributed/Process/Internal/Primitives.hs b/src/Control/Distributed/Process/Internal/Primitives.hs new file mode 100644 --- /dev/null +++ b/src/Control/Distributed/Process/Internal/Primitives.hs @@ -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 + diff --git a/src/Control/Distributed/Process/Internal/TypeRep.hs b/src/Control/Distributed/Process/Internal/TypeRep.hs new file mode 100644 --- /dev/null +++ b/src/Control/Distributed/Process/Internal/TypeRep.hs @@ -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 + diff --git a/src/Control/Distributed/Process/Internal/Types.hs b/src/Control/Distributed/Process/Internal/Types.hs new file mode 100644 --- /dev/null +++ b/src/Control/Distributed/Process/Internal/Types.hs @@ -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 + } diff --git a/src/Control/Distributed/Process/Node.hs b/src/Control/Distributed/Process/Node.hs new file mode 100644 --- /dev/null +++ b/src/Control/Distributed/Process/Node.hs @@ -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 diff --git a/src/Control/Distributed/Process/Serializable.hs b/src/Control/Distributed/Process/Serializable.hs new file mode 100644 --- /dev/null +++ b/src/Control/Distributed/Process/Serializable.hs @@ -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 ")" diff --git a/tests/TestCH.hs b/tests/TestCH.hs new file mode 100644 --- /dev/null +++ b/tests/TestCH.hs @@ -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) + ] diff --git a/tests/TestClosure.hs b/tests/TestClosure.hs new file mode 100644 --- /dev/null +++ b/tests/TestClosure.hs @@ -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) + ]