diff --git a/LICENCE b/LICENCE new file mode 100644 --- /dev/null +++ b/LICENCE @@ -0,0 +1,30 @@ +Copyright Tim Watson, 2012-2013. + +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 author 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.lhs b/Setup.lhs new file mode 100644 --- /dev/null +++ b/Setup.lhs @@ -0,0 +1,3 @@ +#!/usr/bin/env runhaskell +> import Distribution.Simple +> main = defaultMain diff --git a/distributed-process-execution.cabal b/distributed-process-execution.cabal new file mode 100644 --- /dev/null +++ b/distributed-process-execution.cabal @@ -0,0 +1,146 @@ +name: distributed-process-execution +version: 0.1.0 +cabal-version: >=1.8 +build-type: Simple +license: BSD3 +license-file: LICENCE +stability: experimental +Copyright: Tim Watson 2012 - 2013 +Author: Tim Watson +Maintainer: watson.timothy@gmail.com +Stability: experimental +Homepage: http://github.com/haskell-distributed/distributed-process-execution +Bug-Reports: http://github.com/haskell-distributed/distributed-process-execution/issues +synopsis: Execution Framework for The Cloud Haskell Application Platform +description: + The Execution Framework provides tools for load regulation, workload shedding and remote hand-off. + The currently implementation provides only a subset of the plumbing required, comprising tools + for event management, mailbox buffering and message routing. +category: Control +tested-with: GHC == 7.4.2 GHC == 7.6.2 +data-dir: "" + +source-repository head + type: git + location: https://github.com/haskell-distributed/distributed-process-execution + +flag perf + description: Build with profiling enabled + default: False + +library + build-depends: + base >= 4.4 && < 5, + data-accessor >= 0.2.2.3, + distributed-process >= 0.5.2 && < 0.6, + distributed-process-extras >= 0.1.1 && < 0.2, + distributed-process-supervisor >= 0.1.1 && < 0.2, + distributed-process-client-server >= 0.1.1 && < 0.2, + binary >= 0.6.3.0 && < 0.8, + deepseq >= 1.3.0.1 && < 1.4, + mtl, + containers >= 0.4 && < 0.6, + hashable >= 1.2.0.5 && < 1.3, + unordered-containers >= 0.2.3.0 && < 0.3, + fingertree < 0.2, + stm >= 2.4 && < 2.5, + time > 1.4 && < 1.5, + transformers + if impl(ghc <= 7.5) + Build-Depends: template-haskell == 2.7.0.0, + derive == 2.5.5, + uniplate == 1.6.12, + ghc-prim + extensions: CPP + hs-source-dirs: src + ghc-options: -Wall + exposed-modules: + Control.Distributed.Process.Execution, + Control.Distributed.Process.Execution.EventManager, + Control.Distributed.Process.Execution.Exchange, + Control.Distributed.Process.Execution.Mailbox + other-modules: + Control.Distributed.Process.Execution.Exchange.Broadcast, + Control.Distributed.Process.Execution.Exchange.Internal, + Control.Distributed.Process.Execution.Exchange.Router + + +test-suite ExchangeTests + type: exitcode-stdio-1.0 +-- x-uses-tf: true + build-depends: + base >= 4.4 && < 5, + ansi-terminal >= 0.5 && < 0.7, + containers, + hashable, + unordered-containers >= 0.2.3.0 && < 0.3, + distributed-process >= 0.5.2 && < 0.6, + distributed-process-execution, + distributed-process-extras >= 0.1.1 && < 0.2, + distributed-static, + bytestring, + data-accessor, + fingertree < 0.2, + network-transport >= 0.4 && < 0.5, + deepseq >= 1.3.0.1 && < 1.4, + mtl, + network-transport-tcp >= 0.4 && < 0.5, + binary >= 0.6.3.0 && < 0.8, + network >= 2.3 && < 2.7, + HUnit >= 1.2 && < 2, + stm >= 2.3 && < 2.5, + time > 1.4 && < 1.5, + test-framework >= 0.6 && < 0.9, + test-framework-hunit, + QuickCheck >= 2.4, + test-framework-quickcheck2, + transformers, + rematch >= 0.2.0.0, + ghc-prim + hs-source-dirs: + tests + ghc-options: -Wall -threaded -rtsopts -with-rtsopts=-N -fno-warn-unused-do-bind -eventlog + extensions: CPP + main-is: TestExchange.hs + other-modules: TestUtils + + +test-suite MailboxTests + type: exitcode-stdio-1.0 +-- x-uses-tf: true + build-depends: + base >= 4.4 && < 5, + ansi-terminal >= 0.5 && < 0.7, + containers, + hashable, + unordered-containers >= 0.2.3.0 && < 0.3, + distributed-process >= 0.5.2 && < 0.6, + distributed-process-execution, + distributed-process-extras >= 0.1.1 && < 0.2, + distributed-static, + bytestring, + data-accessor, + fingertree < 0.2, + network-transport >= 0.4 && < 0.5, + deepseq >= 1.3.0.1 && < 1.4, + mtl, + network-transport-tcp >= 0.4 && < 0.5, + binary >= 0.6.3.0 && < 0.8, + network >= 2.3 && < 2.7, + HUnit >= 1.2 && < 2, + stm >= 2.3 && < 2.5, + time > 1.4 && < 1.5, + test-framework >= 0.6 && < 0.9, + test-framework-hunit, + QuickCheck >= 2.4, + test-framework-quickcheck2, + transformers, + rematch >= 0.2.0.0, + ghc-prim + hs-source-dirs: + tests + ghc-options: -Wall -threaded -rtsopts -with-rtsopts=-N -fno-warn-unused-do-bind -eventlog + extensions: CPP + main-is: TestMailbox.hs + other-modules: MailboxTestFilters + TestUtils diff --git a/src/Control/Distributed/Process/Execution.hs b/src/Control/Distributed/Process/Execution.hs new file mode 100644 --- /dev/null +++ b/src/Control/Distributed/Process/Execution.hs @@ -0,0 +1,44 @@ +----------------------------------------------------------------------------- +-- | +-- Module : Control.Distributed.Process.Execution +-- Copyright : (c) Tim Watson 2013 - 2014 +-- License : BSD3 (see the file LICENSE) +-- +-- Maintainer : Tim Watson +-- Stability : experimental +-- Portability : non-portable (requires concurrency) +-- +-- [Inter-Process Traffic Management] +-- +-- The /Execution Framework/ provides tools for load regulation, workload +-- shedding and remote hand-off. The currently implementation provides only +-- a subset of the plumbing required, comprising tools for event management, +-- mailbox buffering and message routing. +-- +----------------------------------------------------------------------------- + +module Control.Distributed.Process.Execution + ( -- * Mailbox Buffering + module Control.Distributed.Process.Execution.Mailbox + -- * Message Exchanges + , module Control.Distributed.Process.Execution.Exchange + ) where + +import Control.Distributed.Process.Execution.Exchange hiding (startSupervised) +import Control.Distributed.Process.Execution.Mailbox hiding (startSupervised, post) + +{- + +Load regulation requires that we apply limits to various parts of the system. +The manner in which they're applied may vary, but the mechanisms are limited +to: + +1. rejecting the activity/request +2. accepting the activity immediately +3. blocking some or all requestors +4. blocking some (or all) activities +5. terminiating some (or all) activities + +-} + + diff --git a/src/Control/Distributed/Process/Execution/EventManager.hs b/src/Control/Distributed/Process/Execution/EventManager.hs new file mode 100644 --- /dev/null +++ b/src/Control/Distributed/Process/Execution/EventManager.hs @@ -0,0 +1,153 @@ +{-# LANGUAGE DeriveDataTypeable #-} +{-# LANGUAGE DeriveGeneric #-} +{-# LANGUAGE ExistentialQuantification #-} +{-# LANGUAGE RankNTypes #-} +{-# LANGUAGE RecordWildCards #-} +{-# LANGUAGE ScopedTypeVariables #-} +{-# LANGUAGE TemplateHaskell #-} +{-# LANGUAGE ImpredicativeTypes #-} + +----------------------------------------------------------------------------- +-- | +-- Module : Control.Distributed.Process.Execution.EventManager +-- Copyright : (c) Well-Typed / Tim Watson +-- License : BSD3 (see the file LICENSE) +-- +-- Maintainer : Tim Watson +-- Stability : experimental +-- Portability : non-portable (requires concurrency) +-- +-- [Overview] +-- +-- The /EventManager/ is a parallel/concurrent event handling tool, built on +-- top of the /Exchange API/. Arbitrary events are published to the event +-- manager using 'notify', and are broadcast simulataneously to a set of +-- registered /event handlers/. +-- +-- [Defining and Registering Event Handlers] +-- +-- Event handlers are defined as @Serializable m => s -> m -> Process s@, +-- i.e., an expression taking an initial state, an arbitrary @Serializable@ +-- event/message and performing an action in the @Process@ monad that evaluates +-- to a new state. +-- +-- See "Control.Distributed.Process.Execution.Exchange". +-- +----------------------------------------------------------------------------- + +module Control.Distributed.Process.Execution.EventManager + ( EventManager + , start + , startSupervised + , startSupervisedRef + , notify + , addHandler + , addMessageHandler + ) where + +import Control.Distributed.Process hiding (Message, link) +import qualified Control.Distributed.Process as P (Message) +import Control.Distributed.Process.Execution.Exchange + ( Exchange + , Message(..) + , post + , broadcastExchange + , broadcastExchangeT + , broadcastClient + ) +import qualified Control.Distributed.Process.Execution.Exchange as Exchange + ( startSupervised + ) +import Control.Distributed.Process.Extras.Internal.Primitives +import Control.Distributed.Process.Extras.Internal.Unsafe + ( InputStream + , matchInputStream + ) +import Control.Distributed.Process.Supervisor (SupervisorPid) +import Control.Distributed.Process.Serializable hiding (SerializableDict) +import Data.Binary +import Data.Typeable (Typeable) +import GHC.Generics + +{- notes + +Event manager is implemented over a simple BroadcastExchange. We eschew the +complexities of identifying handlers and allowing them to be removed/deleted +or monitored, since we avoid running them in the exchange process. Instead, +each handler runs as an independent process, leaving handler management up +to the user and allowing all the usual process managemnet techniques (e.g., +registration, supervision, etc) to be utilised instead. + +-} + +-- | Opaque handle to an Event Manager. +-- +newtype EventManager = EventManager { ex :: Exchange } + deriving (Typeable, Generic) +instance Binary EventManager where + +instance Resolvable EventManager where + resolve = resolve . ex + +-- | Start a new /Event Manager/ process and return an opaque handle +-- to it. +start :: Process EventManager +start = broadcastExchange >>= return . EventManager + +startSupervised :: SupervisorPid -> Process EventManager +startSupervised sPid = do + ex <- broadcastExchangeT >>= \t -> Exchange.startSupervised t sPid + return $ EventManager ex + +startSupervisedRef :: SupervisorPid -> Process (ProcessId, P.Message) +startSupervisedRef sPid = do + ex <- startSupervised sPid + Just pid <- resolve ex + return (pid, unsafeWrapMessage ex) + +-- | Broadcast an event to all registered handlers. +notify :: Serializable a => EventManager -> a -> Process () +notify em msg = post (ex em) msg + +-- | Add a new event handler. The handler runs in its own process, +-- which is spawned locally on behalf of the caller. +addHandler :: forall s a. Serializable a + => EventManager + -> (s -> a -> Process s) + -> Process s + -> Process ProcessId +addHandler m h s = + spawnLocal $ newHandler (ex m) (\s' m' -> handleMessage m' (h s')) s + +-- | As 'addHandler', but operates over a raw @Control.Distributed.Process.Message@. +addMessageHandler :: forall s. + EventManager + -> (s -> P.Message -> Process (Maybe s)) + -> Process s + -> Process ProcessId +addMessageHandler m h s = spawnLocal $ newHandler (ex m) h s + +newHandler :: forall s . + Exchange + -> (s -> P.Message -> Process (Maybe s)) + -> Process s + -> Process () +newHandler ex handler initState = do + linkTo ex + is <- broadcastClient ex + listen is handler =<< initState + +listen :: forall s . InputStream Message + -> (s -> P.Message -> Process (Maybe s)) + -> s + -> Process () +listen inStream handler state = do + receiveWait [ matchInputStream inStream ] >>= handleEvent inStream handler state + where + handleEvent is h s p = do + r <- h s (payload p) + let s2 = case r of + Nothing -> s + Just s' -> s' + listen is h s2 + diff --git a/src/Control/Distributed/Process/Execution/Exchange.hs b/src/Control/Distributed/Process/Execution/Exchange.hs new file mode 100644 --- /dev/null +++ b/src/Control/Distributed/Process/Execution/Exchange.hs @@ -0,0 +1,149 @@ +----------------------------------------------------------------------------- +-- | +-- Module : Control.Distributed.Process.Execution.Exchange +-- Copyright : (c) Tim Watson 2012 - 2014 +-- License : BSD3 (see the file LICENSE) +-- +-- Maintainer : Tim Watson +-- Stability : experimental +-- Portability : non-portable (requires concurrency) +-- +-- [Message Exchanges] +-- +-- The concept of a /message exchange/ is borrowed from the world of +-- messaging and enterprise integration. The /exchange/ acts like a kind of +-- mailbox, accepting inputs from /producers/ and forwarding these messages +-- to one or more /consumers/, depending on the implementation's semantics. +-- +-- This module provides some basic types of message exchange and exposes an API +-- for defining your own custom /exchange types/. +-- +-- [Broadcast Exchanges] +-- +-- The broadcast exchange type, started via 'broadcastExchange', forward their +-- inputs to all registered consumers (as the name suggests). This exchange type +-- is highly optimised for local (intra-node) traffic and provides two different +-- kinds of client binding, one which causes messages to be delivered directly +-- to the client's mailbox (viz 'bindToBroadcaster'), the other providing a +-- separate stream of messages that can be obtained using the @expect@ and +-- @receiveX@ family of messaging primitives (and thus composed with other forms +-- of input selection, such as typed channels and selective reads on the process +-- mailbox). +-- +-- /Important:/ When a @ProcessId@ is registered via 'bindToBroadcaster', only +-- the payload of the 'Message' (i.e., the underlying @Serializable@ datum) is +-- forwarded to the consumer, /not/ the whole 'Message' itself. +-- +-- [Router Exchanges] +-- +-- The /router/ API provides a means to selectively route messages to one or +-- more clients, depending on the content of the 'Message'. Two modes of binding +-- (and client selection) are provided out of the box, one of which matches the +-- message 'key', the second of which matches on a name and value from the +-- 'headers'. Alternative mechanisms for content based routing can be derived +-- by modifying the 'BindingSelector' expression passed to 'router' +-- +-- See 'messageKeyRouter' and 'headerContentRouter' for the built-in routing +-- exchanges, and 'router' for the extensible routing API. +-- +-- [Custom Exchange Types] +-- +-- Both the /broadcast/ and /router/ exchanges are implemented as custom +-- /exchange types/. The mechanism for defining custom exchange behaviours +-- such as these is very simple. Raw exchanges are started by evaluating +-- 'startExchange' with a specific 'ExchangeType' record. This type is +-- parameterised by the internal /state/ it holds, and defines two API callbacks +-- in its 'configureEx' and 'routeEx' fields. The former is evaluated whenever a +-- client process evaluates 'configureExchange', the latter whenever a client +-- evaluates 'post' or 'postMessage'. The 'configureEx' callback takes a raw +-- @Message@ (from "Control.Distributed.Process") and is responsible for +-- decoding the message and updating its own state (if required). It is via +-- this callback that custom exchange types can receive information about +-- clients and handle it in their own way. The 'routeEx' callback is evaluated +-- with the exchange type's own internal state and the 'Message' originally +-- sent to the exchange process (via 'post') and is responsible for delivering +-- the message to its clients in whatever way makes sense for that exchange +-- type. +-- +----------------------------------------------------------------------------- + +module Control.Distributed.Process.Execution.Exchange + ( -- * Fundamental API + Exchange() + , Message(..) + -- * Starting/Running an Exchange + , startExchange + , startSupervised + , startSupervisedRef + , runExchange + -- * Client Facing API + , post + , postMessage + , configureExchange + , createMessage + -- * Broadcast Exchange + , broadcastExchange + , broadcastExchangeT + , broadcastClient + , bindToBroadcaster + , BroadcastExchange + -- * Routing (Content Based) + , HeaderName + , Binding(..) + , Bindable + , BindingSelector + , RelayType(..) + -- * Starting a Router + , router + , supervisedRouter + -- * Routing (Publishing) API + , route + , routeMessage + -- * Routing via message/binding keys + , messageKeyRouter + , bindKey + -- * Routing via message headers + , headerContentRouter + , bindHeader + -- * Defining Custom Exchange Types + , ExchangeType(..) + , applyHandlers + ) where + +import Control.Distributed.Process.Execution.Exchange.Broadcast + ( broadcastExchange + , broadcastExchangeT + , broadcastClient + , bindToBroadcaster + , BroadcastExchange + ) +import Control.Distributed.Process.Execution.Exchange.Internal + ( Exchange() + , Message(..) + , ExchangeType(..) + , startExchange + , startSupervised + , startSupervisedRef + , runExchange + , post + , postMessage + , configureExchange + , createMessage + , applyHandlers + ) +import Control.Distributed.Process.Execution.Exchange.Router + ( HeaderName + , Binding(..) + , Bindable + , BindingSelector + , RelayType(..) + , router + , supervisedRouter + , route + , routeMessage + , messageKeyRouter + , bindKey + , headerContentRouter + , bindHeader + ) + diff --git a/src/Control/Distributed/Process/Execution/Exchange/Broadcast.hs b/src/Control/Distributed/Process/Execution/Exchange/Broadcast.hs new file mode 100644 --- /dev/null +++ b/src/Control/Distributed/Process/Execution/Exchange/Broadcast.hs @@ -0,0 +1,338 @@ +{-# LANGUAGE DeriveDataTypeable #-} +{-# LANGUAGE DeriveGeneric #-} +{-# LANGUAGE StandaloneDeriving #-} +{-# LANGUAGE ScopedTypeVariables #-} +{-# LANGUAGE PatternGuards #-} +{-# LANGUAGE RecordWildCards #-} +{-# LANGUAGE FlexibleInstances #-} +{-# LANGUAGE EmptyDataDecls #-} +{-# LANGUAGE TemplateHaskell #-} +{-# LANGUAGE ImpredicativeTypes #-} +{-# LANGUAGE UndecidableInstances #-} +{-# LANGUAGE MultiParamTypeClasses #-} + +-- | An exchange type that broadcasts all incomings 'Post' messages. +module Control.Distributed.Process.Execution.Exchange.Broadcast + ( + broadcastExchange + , broadcastExchangeT + , broadcastClient + , bindToBroadcaster + , BroadcastExchange + ) where + +import Control.Concurrent.STM (STM, atomically) +import Control.Concurrent.STM.TChan + ( TChan + , newBroadcastTChanIO + , dupTChan + , readTChan + , writeTChan + ) +import Control.DeepSeq (NFData) +import Control.Distributed.Process + ( Process + , MonitorRef + , ProcessMonitorNotification(..) + , ProcessId + , SendPort + , processNodeId + , getSelfPid + , getSelfNode + , liftIO + , newChan + , sendChan + , unsafeSend + , unsafeSendChan + , receiveWait + , match + , matchIf + , die + , handleMessage + , Match + ) +import qualified Control.Distributed.Process as P +import Control.Distributed.Process.Serializable() +import Control.Distributed.Process.Execution.Exchange.Internal + ( startExchange + , configureExchange + , Message(..) + , Exchange(..) + , ExchangeType(..) + , applyHandlers + ) +import Control.Distributed.Process.Extras.Internal.Types + ( Channel + , ServerDisconnected(..) + ) +import Control.Distributed.Process.Extras.Internal.Unsafe -- see [note: pcopy] + ( PCopy + , pCopy + , pUnwrap + , matchChanP + , InputStream(Null) + , newInputStream + ) +import Control.Monad (forM_, void) +import Data.Accessor + ( Accessor + , accessor + , (^:) + ) +import Data.Binary +import qualified Data.Foldable as Foldable (toList) +import Data.Map.Strict (Map) +import qualified Data.Map.Strict as Map +import Data.Typeable (Typeable) +import GHC.Generics + +-- newtype RoutingTable r = +-- RoutingTable { routes :: (Map String (Map ProcessId r)) } + +-- [note: BindSTM, BindPort and safety] +-- We keep these two /bind types/ separate, since only one of them +-- is truly serializable. The risk of unifying them is that at some +-- later time a maintainer might not realise that BindSTM cannot be +-- sent over the wire due to our use of PCopy. +-- + +data BindPort = BindPort { portClient :: !ProcessId + , portSend :: !(SendPort Message) + } deriving (Typeable, Generic) +instance Binary BindPort where +instance NFData BindPort where + +data BindSTM = + BindSTM { stmClient :: !ProcessId + , stmSend :: !(SendPort (PCopy (InputStream Message))) + } deriving (Typeable) +{- | forall r. (Routable r) => + BindR { client :: !ProcessId + , key :: !String + , chanC :: !r + } + deriving (Typeable, Generic) +-} + +data OutputStream = + WriteChan (SendPort Message) + | WriteSTM (Message -> STM ()) +-- | WriteP ProcessId + | NoWrite + deriving (Typeable) + +data Binding = Binding { outputStream :: !OutputStream + , inputStream :: !(InputStream Message) + } + | PidBinding !ProcessId + deriving (Typeable) + +data BindOk = BindOk + deriving (Typeable, Generic) +instance Binary BindOk where +instance NFData BindOk where + +data BindFail = BindFail !String + deriving (Typeable, Generic) +instance Binary BindFail where +instance NFData BindFail where + +data BindPlease = BindPlease + deriving (Typeable, Generic) +instance Binary BindPlease where +instance NFData BindPlease where + +type BroadcastClients = Map ProcessId Binding +data BroadcastEx = + BroadcastEx { _routingTable :: !BroadcastClients + , channel :: !(TChan Message) + } + +type BroadcastExchange = ExchangeType BroadcastEx + +-------------------------------------------------------------------------------- +-- Starting/Running the Exchange -- +-------------------------------------------------------------------------------- + +-- | Start a new /broadcast exchange/ and return a handle to the exchange. +broadcastExchange :: Process Exchange +broadcastExchange = broadcastExchangeT >>= startExchange + +-- | The 'ExchangeType' of a broadcast exchange. Can be combined with the +-- @startSupervisedRef@ and @startSupervised@ APIs. +-- +broadcastExchangeT :: Process BroadcastExchange +broadcastExchangeT = do + ch <- liftIO newBroadcastTChanIO + return $ ExchangeType { name = "BroadcastExchange" + , state = BroadcastEx Map.empty ch + , configureEx = apiConfigure + , routeEx = apiRoute + } + +-------------------------------------------------------------------------------- +-- Client Facing API -- +-------------------------------------------------------------------------------- + +-- | Create a binding to the given /broadcast exchange/ for the calling process +-- and return an 'InputStream' that can be used in the @expect@ and +-- @receiveWait@ family of messaging primitives. This form of client interaction +-- helps avoid cluttering the caller's mailbox with 'Message' data, since the +-- 'InputChannel' provides a separate input stream (in a similar fashion to +-- a typed channel). +-- Example: +-- +-- > is <- broadcastClient ex +-- > msg <- receiveWait [ matchInputStream is ] +-- > handleMessage (payload msg) +-- +broadcastClient :: Exchange -> Process (InputStream Message) +broadcastClient ex@Exchange{..} = do + myNode <- getSelfNode + us <- getSelfPid + if processNodeId pid == myNode -- see [note: pcopy] + then do (sp, rp) <- newChan + configureExchange ex $ pCopy (BindSTM us sp) + mRef <- P.monitor pid + P.finally (receiveWait [ matchChanP rp + , handleServerFailure mRef ]) + (P.unmonitor mRef) + else do (sp, rp) <- newChan :: Process (Channel Message) + configureExchange ex $ BindPort us sp + mRef <- P.monitor pid + P.finally (receiveWait [ + match (\(_ :: BindOk) -> return $ newInputStream $ Left rp) + , match (\(f :: BindFail) -> die f) + , handleServerFailure mRef + ]) + (P.unmonitor mRef) + +-- | Bind the calling process to the given /broadcast exchange/. For each +-- 'Message' the exchange receives, /only the payload will be sent/ +-- to the calling process' mailbox. +-- +-- Example: +-- +-- (producer) +-- > post ex "Hello" +-- +-- (consumer) +-- > bindToBroadcaster ex +-- > expect >>= liftIO . putStrLn +-- +bindToBroadcaster :: Exchange -> Process () +bindToBroadcaster ex@Exchange{..} = do + us <- getSelfPid + configureExchange ex $ (BindPlease, us) + +-------------------------------------------------------------------------------- +-- Exchage Definition/State & API Handlers -- +-------------------------------------------------------------------------------- + +apiRoute :: BroadcastEx -> Message -> Process BroadcastEx +apiRoute ex@BroadcastEx{..} msg = do + liftIO $ atomically $ writeTChan channel msg + forM_ (Foldable.toList _routingTable) $ routeToClient msg + return ex + where + routeToClient m (PidBinding p) = P.forward (payload m) p + routeToClient m b@(Binding _ _) = writeToStream (outputStream b) m + +-- TODO: implement unbind!!? + +apiConfigure :: BroadcastEx -> P.Message -> Process BroadcastEx +apiConfigure ex msg = do + -- for unsafe / non-serializable message passing hacks, see [note: pcopy] + applyHandlers ex msg $ [ \m -> handleMessage m (handleBindPort ex) + , \m -> handleBindSTM ex m + , \m -> handleMessage m (handleBindPlease ex) + , \m -> handleMessage m (handleMonitorSignal ex) + , (const $ return $ Just ex) + ] + where + handleBindPlease ex' (BindPlease, p) = do + case lookupBinding ex' p of + Nothing -> return $ (routingTable ^: Map.insert p (PidBinding p)) ex' + Just _ -> return ex' + + handleMonitorSignal bx (ProcessMonitorNotification _ p _) = + return $ (routingTable ^: Map.delete p) bx + + handleBindSTM ex'@BroadcastEx{..} msg' = do + bind' <- pUnwrap msg' :: Process (Maybe BindSTM) -- see [note: pcopy] + case bind' of + Nothing -> return Nothing + Just s -> do + let binding = lookupBinding ex' (stmClient s) + case binding of + Nothing -> createBinding ex' s >>= \ex'' -> handleBindSTM ex'' msg' + Just b -> sendBinding (stmSend s) b >> return (Just ex') + + createBinding bEx'@BroadcastEx{..} BindSTM{..} = do + void $ P.monitor stmClient + nch <- liftIO $ atomically $ dupTChan channel + let istr = newInputStream $ Right (readTChan nch) + let ostr = NoWrite -- we write to our own channel, not the broadcast + let bnd = Binding ostr istr + return $ (routingTable ^: Map.insert stmClient bnd) bEx' + + sendBinding sp' bs = unsafeSendChan sp' $ pCopy (inputStream bs) + + handleBindPort :: BroadcastEx -> BindPort -> Process BroadcastEx + handleBindPort x@BroadcastEx{..} BindPort{..} = do + let binding = lookupBinding x portClient + case binding of + Just _ -> unsafeSend portClient (BindFail "DuplicateBinding") >> return x + Nothing -> do + let istr = Null + let ostr = WriteChan portSend + let bound = Binding ostr istr + void $ P.monitor portClient + unsafeSend portClient BindOk + return $ (routingTable ^: Map.insert portClient bound) x + + lookupBinding BroadcastEx{..} k = Map.lookup k $ _routingTable + +{- [note: pcopy] + +We rely on risky techniques here, in order to allow for sharing useful +data that is not really serializable. For Cloud Haskell generally, this is +a bad idea, since we want message passing to work both locally and in a +distributed setting. In this case however, what we're really attempting is +an optimisation, since we only use unsafe PCopy based techniques when dealing +with exchange clients residing on our (local) node. + +The PCopy mechanism is defined in the (aptly named) "Unsafe" module. + +-} + +-- TODO: move handleServerFailure into Primitives.hs + +writeToStream :: OutputStream -> Message -> Process () +writeToStream (WriteChan sp) = sendChan sp -- see [note: safe remote send] +writeToStream (WriteSTM stm) = liftIO . atomically . stm +writeToStream NoWrite = const $ return () +{-# INLINE writeToStream #-} + +{- [note: safe remote send] + +Although we go to great lengths here to avoid serialization and/or copying +overheads, there are some activities for which we prefer to play it safe. +Chief among these is delivering messages to remote clients. Thankfully, our +unsafe @sendChan@ primitive will crash the caller/sender if there are any +encoding problems, however it is only because we /know/ for certain that +our recipient is remote, that we've chosen to write via a SendPort in the +first place! It makes sense therefore, to use the safe @sendChan@ operation +here, since for a remote call we /cannot/ avoid the overhead of serialization +anyway. + +-} + +handleServerFailure :: MonitorRef -> Match (InputStream Message) +handleServerFailure mRef = + matchIf (\(ProcessMonitorNotification r _ _) -> r == mRef) + (\(ProcessMonitorNotification _ _ d) -> die $ ServerDisconnected d) + +routingTable :: Accessor BroadcastEx BroadcastClients +routingTable = accessor _routingTable (\r e -> e { _routingTable = r }) + diff --git a/src/Control/Distributed/Process/Execution/Exchange/Internal.hs b/src/Control/Distributed/Process/Execution/Exchange/Internal.hs new file mode 100644 --- /dev/null +++ b/src/Control/Distributed/Process/Execution/Exchange/Internal.hs @@ -0,0 +1,276 @@ +{-# LANGUAGE DeriveDataTypeable #-} +{-# LANGUAGE DeriveGeneric #-} +{-# LANGUAGE StandaloneDeriving #-} +{-# LANGUAGE ScopedTypeVariables #-} +{-# LANGUAGE PatternGuards #-} +{-# LANGUAGE RecordWildCards #-} +{-# LANGUAGE FlexibleInstances #-} +{-# LANGUAGE EmptyDataDecls #-} +{-# LANGUAGE TemplateHaskell #-} +{-# LANGUAGE ImpredicativeTypes #-} +{-# LANGUAGE UndecidableInstances #-} +{-# LANGUAGE MultiParamTypeClasses #-} + +-- | Internal Exchange Implementation +module Control.Distributed.Process.Execution.Exchange.Internal + ( Exchange(..) + , Message(..) + , ExchangeType(..) + , startExchange + , startSupervised + , startSupervisedRef + , runExchange + , post + , postMessage + , configureExchange + , createMessage + , applyHandlers + ) where + +import Control.Concurrent.MVar (MVar, takeMVar, putMVar, newEmptyMVar) +import Control.DeepSeq (NFData) +import Control.Distributed.Process + ( Process + , ProcessMonitorNotification(..) + , ProcessId + , liftIO + , spawnLocal + , unsafeWrapMessage + ) +import qualified Control.Distributed.Process as P (Message, link) +import Control.Distributed.Process.Serializable hiding (SerializableDict) +import Control.Distributed.Process.Extras.Internal.Types + ( Resolvable(..) + ) +import Control.Distributed.Process.Extras.Internal.Primitives + ( Linkable(..) + ) +import Control.Distributed.Process.ManagedProcess + ( channelControlPort + , handleControlChan + , handleInfo + , handleRaw + , continue + , defaultProcess + , InitHandler + , InitResult(..) + , ProcessAction + , ProcessDefinition(..) + , ControlChannel + , ControlPort + ) +import qualified Control.Distributed.Process.ManagedProcess as MP + ( chanServe + ) +import Control.Distributed.Process.ManagedProcess.UnsafeClient + ( sendControlMessage + ) +import Control.Distributed.Process.Supervisor (SupervisorPid) +import Control.Distributed.Process.Extras.Time (Delay(Infinity)) +import Data.Binary +import Data.Typeable (Typeable) +import GHC.Generics +import Prelude hiding (drop) + +{- [design notes] + +Messages are sent to exchanges and forwarded to clients. An exchange +is parameterised by its routing mechanism, which is responsible for +maintaining its own client state and selecting the clients to which +messages are forwarded. + +-} + +-- | Opaque handle to an exchange. +-- +data Exchange = Exchange { pid :: !ProcessId + , cchan :: !(ControlPort ControlMessage) + , xType :: !String + } deriving (Typeable, Generic, Eq) +instance Binary Exchange where +instance Show Exchange where + show Exchange{..} = (xType ++ ":" ++ (show pid)) + +instance Resolvable Exchange where + resolve = return . Just . pid + +{- +instance Observable Exchange MonitorRef ProcessMonitorNotification where + observe = P.monitor . pid + unobserve = P.unmonitor + observableFrom ref (ProcessMonitorNotification ref' _ r) = + return $ if ref' == ref then Just r else Nothing +-} + +instance Linkable Exchange where + linkTo = P.link . pid + +-- we communicate with exchanges using control channels +sendCtrlMsg :: Exchange -> ControlMessage -> Process () +sendCtrlMsg Exchange{..} = sendControlMessage cchan + +-- | Messages sent to an exchange can optionally provide a routing +-- key and a list of (key, value) headers in addition to the underlying +-- payload. +data Message = + Message { key :: !String -- ^ a /routing key/ for the payload + , headers :: ![(String, String)] -- ^ arbitrary key-value headers + , payload :: !P.Message -- ^ the underlying @Message@ payload + } deriving (Typeable, Generic, Show) +instance Binary Message where +instance NFData Message where + +data ControlMessage = + Configure !P.Message + | Post !Message + deriving (Typeable, Generic) +instance Binary ControlMessage where +instance NFData ControlMessage where + +-- | Different exchange types are defined using record syntax. +-- The 'configureEx' and 'routeEx' API functions are called during the exchange +-- lifecycle when incoming traffic arrives. Configuration messages are +-- completely arbitrary types and the exchange type author is entirely +-- responsible for decoding them. Messages posted to the exchange (see the +-- 'Message' data type) are passed to the 'routeEx' API function along with the +-- exchange type's own internal state. Both API functions return a new +-- (potentially updated) state and run in the @Process@ monad. +-- +data ExchangeType s = + ExchangeType { name :: String + , state :: s + , configureEx :: s -> P.Message -> Process s + , routeEx :: s -> Message -> Process s + } + +-------------------------------------------------------------------------------- +-- Starting/Running an Exchange -- +-------------------------------------------------------------------------------- + +-- | Starts an /exchange process/ with the given 'ExchangeType'. +startExchange :: forall s. ExchangeType s -> Process Exchange +startExchange = doStart Nothing + +-- | Starts an exchange as part of a supervision tree. +-- +-- Example: +-- > childSpec = toChildStart $ startSupervisedRef exType +-- +startSupervisedRef :: forall s . ExchangeType s + -> SupervisorPid + -> Process (ProcessId, P.Message) +startSupervisedRef t s = do + ex <- startSupervised t s + return (pid ex, unsafeWrapMessage ex) + +-- | Starts an exchange as part of a supervision tree. +-- +-- Example: +-- > childSpec = toChildStart $ startSupervised exType +-- +startSupervised :: forall s . ExchangeType s + -> SupervisorPid + -> Process Exchange +startSupervised t s = doStart (Just s) t + +doStart :: Maybe SupervisorPid -> ExchangeType s -> Process Exchange +doStart mSp t = do + cchan <- liftIO $ newEmptyMVar + spawnLocal (maybeLink mSp >> runExchange t cchan) >>= \pid -> do + cc <- liftIO $ takeMVar cchan + return $ Exchange pid cc (name t) + where + maybeLink Nothing = return () + maybeLink (Just p') = P.link p' + +runExchange :: forall s. + ExchangeType s + -> MVar (ControlPort ControlMessage) + -> Process () +runExchange t tc = MP.chanServe t exInit (processDefinition t tc) + +exInit :: forall s. InitHandler (ExchangeType s) (ExchangeType s) +exInit t = return $ InitOk t Infinity + +-------------------------------------------------------------------------------- +-- Client Facing API -- +-------------------------------------------------------------------------------- + +-- | Posts an arbitrary 'Serializable' datum to an /exchange/. The raw datum is +-- wrapped in the 'Message' data type, with its 'key' set to @""@ and its +-- 'headers' to @[]@. +post :: Serializable a => Exchange -> a -> Process () +post ex msg = postMessage ex $ Message "" [] (unsafeWrapMessage msg) + +-- | Posts a 'Message' to an /exchange/. +postMessage :: Exchange -> Message -> Process () +postMessage ex msg = msg `seq` sendCtrlMsg ex $ Post msg + +-- | Sends an arbitrary 'Serializable' datum to an /exchange/, for use as a +-- configuration change - see 'configureEx' for details. +configureExchange :: Serializable m => Exchange -> m -> Process () +configureExchange e m = sendCtrlMsg e $ Configure (unsafeWrapMessage m) + +-- | Utility for creating a 'Message' datum from its 'key', 'headers' and +-- 'payload'. +createMessage :: Serializable m => String -> [(String, String)] -> m -> Message +createMessage k h m = Message k h $ unsafeWrapMessage m + +-- | Utility for custom exchange type authors - evaluates a set of primitive +-- message handlers from left to right, returning the first which evaluates +-- to @Just a@, or the initial @e@ value if all the handlers yield @Nothing@. +applyHandlers :: a + -> P.Message + -> [P.Message -> Process (Maybe a)] + -> Process a +applyHandlers e _ [] = return e +applyHandlers e m (f:fs) = do + r <- f m + case r of + Nothing -> applyHandlers e m fs + Just r' -> return r' + +-------------------------------------------------------------------------------- +-- Process Definition/State & API Handlers -- +-------------------------------------------------------------------------------- + +processDefinition :: forall s. + ExchangeType s + -> MVar (ControlPort ControlMessage) + -> ControlChannel ControlMessage + -> Process (ProcessDefinition (ExchangeType s)) +processDefinition _ tc cc = do + liftIO $ putMVar tc $ channelControlPort cc + return $ + defaultProcess { + apiHandlers = [ handleControlChan cc handleControlMessage ] + , infoHandlers = [ handleInfo handleMonitor + , handleRaw convertToCC + ] + } :: Process (ProcessDefinition (ExchangeType s)) + +handleMonitor :: forall s. + ExchangeType s + -> ProcessMonitorNotification + -> Process (ProcessAction (ExchangeType s)) +handleMonitor ex m = do + handleControlMessage ex (Configure (unsafeWrapMessage m)) + +convertToCC :: forall s. + ExchangeType s + -> P.Message + -> Process (ProcessAction (ExchangeType s)) +convertToCC ex msg = do + liftIO $ putStrLn "convert to cc" + handleControlMessage ex (Post $ Message "" [] msg) + +handleControlMessage :: forall s. + ExchangeType s + -> ControlMessage + -> Process (ProcessAction (ExchangeType s)) +handleControlMessage ex@ExchangeType{..} cm = + let action = case cm of + Configure msg -> configureEx state msg + Post msg -> routeEx state msg + in action >>= \s -> continue $ ex { state = s } + diff --git a/src/Control/Distributed/Process/Execution/Exchange/Router.hs b/src/Control/Distributed/Process/Execution/Exchange/Router.hs new file mode 100644 --- /dev/null +++ b/src/Control/Distributed/Process/Execution/Exchange/Router.hs @@ -0,0 +1,250 @@ +{-# LANGUAGE DeriveDataTypeable #-} +{-# LANGUAGE DeriveGeneric #-} +{-# LANGUAGE StandaloneDeriving #-} +{-# LANGUAGE ScopedTypeVariables #-} +{-# LANGUAGE PatternGuards #-} +{-# LANGUAGE RecordWildCards #-} +{-# LANGUAGE FlexibleInstances #-} +{-# LANGUAGE EmptyDataDecls #-} +{-# LANGUAGE TemplateHaskell #-} +{-# LANGUAGE ImpredicativeTypes #-} +{-# LANGUAGE UndecidableInstances #-} +{-# LANGUAGE MultiParamTypeClasses #-} + +-- | A simple API for /routing/, using a custom exchange type. +module Control.Distributed.Process.Execution.Exchange.Router + ( -- * Types + HeaderName + , Binding(..) + , Bindable + , BindingSelector + , RelayType(..) + -- * Starting a Router + , router + , supervisedRouter + , supervisedRouterRef + -- * Client (Publishing) API + , route + , routeMessage + -- * Routing via message/binding keys + , messageKeyRouter + , bindKey + -- * Routing via message headers + , headerContentRouter + , bindHeader + ) where + +import Control.DeepSeq (NFData) +import Control.Distributed.Process + ( Process + , ProcessMonitorNotification(..) + , ProcessId + , monitor + , handleMessage + , unsafeWrapMessage + ) +import qualified Control.Distributed.Process as P +import Control.Distributed.Process.Serializable (Serializable) +import Control.Distributed.Process.Execution.Exchange.Internal + ( startExchange + , startSupervised + , configureExchange + , Message(..) + , Exchange + , ExchangeType(..) + , post + , postMessage + , applyHandlers + ) +import Control.Distributed.Process.Extras.Internal.Primitives + ( deliver + , Resolvable(..) + ) +import Control.Distributed.Process.Supervisor (SupervisorPid) +import Data.Binary +import Data.Foldable (forM_) +import Data.Hashable +import Data.HashMap.Strict (HashMap) +import qualified Data.HashMap.Strict as Map +import Data.HashSet (HashSet) +import qualified Data.HashSet as Set +import Data.Typeable (Typeable) +import GHC.Generics + +type HeaderName = String + +-- | The binding key used by the built-in key and header based +-- routers. +data Binding = + BindKey { bindingKey :: !String } + | BindHeader { bindingKey :: !String + , headerName :: !HeaderName + } + | BindNone + deriving (Typeable, Generic, Eq, Show) +instance Binary Binding where +instance NFData Binding where +instance Hashable Binding where + +-- | Things that can be used as binding keys in a router. +class (Hashable k, Eq k, Serializable k) => Bindable k +instance (Hashable k, Eq k, Serializable k) => Bindable k + +-- | Used to convert a 'Message' into a 'Bindable' routing key. +type BindingSelector k = (Message -> Process k) + +-- | Given to a /router/ to indicate whether clients should +-- receive 'Message' payloads only, or the whole 'Message' object +-- itself. +data RelayType = PayloadOnly | WholeMessage + +data State k = State { bindings :: !(HashMap k (HashSet ProcessId)) + , selector :: !(BindingSelector k) + , relayType :: !RelayType + } + +type Router k = ExchangeType (State k) + +-------------------------------------------------------------------------------- +-- Starting/Running the Exchange -- +-------------------------------------------------------------------------------- + +-- | A router that matches on a 'Message' 'key'. To bind a client @Process@ to +-- such an exchange, use the 'bindKey' function. +messageKeyRouter :: RelayType -> Process Exchange +messageKeyRouter t = router t matchOnKey -- (return . BindKey . key) + where + matchOnKey :: Message -> Process Binding + matchOnKey m = return $ BindKey (key m) + +-- | A router that matches on a specific (named) header. To bind a client +-- @Process@ to such an exchange, use the 'bindHeader' function. +headerContentRouter :: RelayType -> HeaderName -> Process Exchange +headerContentRouter t n = router t (checkHeaders n) + where + checkHeaders hn Message{..} = do + case Map.lookup hn (Map.fromList headers) of + Nothing -> return BindNone + Just hv -> return $ BindHeader hn hv + +-- | Defines a /router/ exchange. The 'BindingSelector' is used to construct +-- a binding (i.e., an instance of the 'Bindable' type @k@) for each incoming +-- 'Message'. Such bindings are matched against bindings stored in the exchange. +-- Clients of a /router/ exchange are identified by a binding, mapped to +-- one or more 'ProcessId's. +-- +-- The format of the bindings, nature of their storage and mechanism for +-- submitting new bindings is implementation dependent (i.e., will vary by +-- exchange type). For example, the 'messageKeyRouter' and 'headerContentRouter' +-- implementations both use the 'Binding' data type, which can represent a +-- 'Message' key or a 'HeaderName' and content. As with all custom exchange +-- types, bindings should be submitted by evaluating 'configureExchange' with +-- a suitable data type. +-- +router :: (Bindable k) => RelayType -> BindingSelector k -> Process Exchange +router t s = routerT t s >>= startExchange + +supervisedRouterRef :: Bindable k + => RelayType + -> BindingSelector k + -> SupervisorPid + -> Process (ProcessId, P.Message) +supervisedRouterRef t sel spid = do + ex <- supervisedRouter t sel spid + Just pid <- resolve ex + return (pid, unsafeWrapMessage ex) + +-- | Defines a /router/ that can be used in a supervision tree. +supervisedRouter :: Bindable k + => RelayType + -> BindingSelector k + -> SupervisorPid + -> Process Exchange +supervisedRouter t sel spid = + routerT t sel >>= \t' -> startSupervised t' spid + +routerT :: Bindable k + => RelayType + -> BindingSelector k + -> Process (Router k) +routerT t s = do + return $ ExchangeType { name = "Router" + , state = State Map.empty s t + , configureEx = apiConfigure + , routeEx = apiRoute + } + +-------------------------------------------------------------------------------- +-- Client Facing API -- +-------------------------------------------------------------------------------- + +-- | Add a binding (for the calling process) to a 'messageKeyRouter' exchange. +bindKey :: String -> Exchange -> Process () +bindKey k ex = do + self <- P.getSelfPid + configureExchange ex (self, BindKey k) + +-- | Add a binding (for the calling process) to a 'headerContentRouter' exchange. +bindHeader :: HeaderName -> String -> Exchange -> Process () +bindHeader n v ex = do + self <- P.getSelfPid + configureExchange ex (self, BindHeader v n) + +-- | Send a 'Serializable' message to the supplied 'Exchange'. The given datum +-- will be converted to a 'Message', with the 'key' set to @""@ and the +-- 'headers' to @[]@. +-- +-- The routing behaviour will be dependent on the choice of 'BindingSelector' +-- given when initialising the /router/. +route :: Serializable m => Exchange -> m -> Process () +route = post + +-- | Send a 'Message' to the supplied 'Exchange'. +-- The routing behaviour will be dependent on the choice of 'BindingSelector' +-- given when initialising the /router/. +routeMessage :: Exchange -> Message -> Process () +routeMessage = postMessage + +-------------------------------------------------------------------------------- +-- Exchage Definition/State & API Handlers -- +-------------------------------------------------------------------------------- + +apiRoute :: forall k. Bindable k + => State k + -> Message + -> Process (State k) +apiRoute st@State{..} msg = do + binding <- selector msg + case Map.lookup binding bindings of + Nothing -> return st + Just bs -> forM_ bs (fwd relayType msg) >> return st + where + fwd WholeMessage m = deliver m + fwd PayloadOnly m = P.forward (payload m) + +-- TODO: implement 'unbind' ??? +-- TODO: apiConfigure currently leaks memory if clients die (we don't cleanup) + +apiConfigure :: forall k. Bindable k + => State k + -> P.Message + -> Process (State k) +apiConfigure st msg = do + applyHandlers st msg $ [ \m -> handleMessage m (createBinding st) + , \m -> handleMessage m (handleMonitorSignal st) + ] + where + createBinding s@State{..} (pid, bind) = do + case Map.lookup bind bindings of + Nothing -> do _ <- monitor pid + return $ s { bindings = newBind bind pid bindings } + Just ps -> return $ s { bindings = addBind bind pid bindings ps } + + newBind b p bs = Map.insert b (Set.singleton p) bs + addBind b' p' bs' ps = Map.insert b' (Set.insert p' ps) bs' + + handleMonitorSignal s@State{..} (ProcessMonitorNotification _ p _) = + let bs = bindings + bs' = Map.foldlWithKey' (\a k v -> Map.insert k (Set.delete p v) a) bs bs + in return $ s { bindings = bs' } + diff --git a/src/Control/Distributed/Process/Execution/Mailbox.hs b/src/Control/Distributed/Process/Execution/Mailbox.hs new file mode 100644 --- /dev/null +++ b/src/Control/Distributed/Process/Execution/Mailbox.hs @@ -0,0 +1,744 @@ +{-# LANGUAGE DeriveDataTypeable #-} +{-# LANGUAGE DeriveGeneric #-} +{-# LANGUAGE StandaloneDeriving #-} +{-# LANGUAGE ScopedTypeVariables #-} +{-# LANGUAGE PatternGuards #-} +{-# LANGUAGE RecordWildCards #-} +{-# LANGUAGE FlexibleInstances #-} +{-# LANGUAGE EmptyDataDecls #-} +{-# LANGUAGE TemplateHaskell #-} +{-# LANGUAGE ImpredicativeTypes #-} + +----------------------------------------------------------------------------- +-- | +-- Module : Control.Distributed.Process.Execution.Mailbox +-- Copyright : (c) Tim Watson 2012 - 2013 +-- License : BSD3 (see the file LICENSE) +-- +-- Maintainer : Tim Watson +-- Stability : experimental +-- Portability : non-portable (requires concurrency) +-- +-- Generic process that acts as an external mailbox and message buffer. +-- +-- [Overview] +-- +-- For use when rate limiting is not possible (or desired), this module +-- provides a /buffer process/ that receives mail via its 'post' API, buffers +-- the received messages and delivers them when its /owning process/ asks for +-- them. A mailbox has to be started with a maximum buffer size - the so called +-- /limit/ - and will discard messages once its internal storage reaches this +-- user defined threshold. +-- +-- The usual behaviour of the /buffer process/ is to accumulate messages in +-- its internal memory. When a client evaluates 'notify', the buffer will +-- send a 'NewMail' message to the (real) mailbox of its owning process as +-- soon as it has any message(s) ready to deliver. If the buffer already +-- contains undelivered mail, the 'NewMail' message will be dispatched +-- immediately. +-- +-- When the owning process wishes to receive mail, evaluating 'deliver' (from +-- any process) will cause the buffer to send its owner a 'Delivery' message +-- containing the accumulated messages and additional information about the +-- number of messages it is delivering, the number of messages dropped since +-- the last delivery and a handle for the mailbox (so that processes can have +-- multiple mailboxes if required, and distinguish between them). +-- +-- [Overflow Handling] +-- +-- A mailbox handles overflow - when the number of messages it is holding +-- reaches the limit - differently depending on the 'BufferType' selected +-- when it starts. The @Queue@ buffer will, once the limit is reached, drop +-- older messages first (i.e., the head of the queue) to make space for +-- newer ones. The @Ring@ buffer works similarly, but blocks new messages +-- so as to preserve existing ones instead. Finally, the @Stack@ buffer will +-- drop the last (i.e., most recently received) message to make room for new +-- mail. +-- +-- Mailboxes can be /resized/ by evaluating 'resize' with a new value for the +-- limit. If the new limit is older that the current/previous one, messages +-- are dropped as though the mailbox had previously seen a volume of mail +-- equal to the difference (in size) between the limits. In this situation, +-- the @Queue@ will drop as many older messages as neccessary to come within +-- the limit, whilst the other two buffer types will drop as many newer messages +-- as needed. +-- +-- [Ordering Guarantees] +-- +-- When messages are delivered to the owner, they arrive as a list of raw +-- @Message@ entries, given in descending age order (i.e., eldest first). +-- Whilst this approximates the FIFO ordering a process' mailbox would usually +-- offer, the @Stack@ buffer will appear to offer no ordering at all, since +-- it always deletes the most recent message(s). The @Queue@ and @Ring@ buffers +-- will maintain a more queue-like (i.e., FIFO) view of received messages, +-- with the obvious constraint the newer or older data might have been deleted. +-- +-- [Post API and Relaying] +-- +-- For messages to be properly handled by the mailbox, they can either be sent +-- via the 'post' API or directly to the 'Mailbox'. Messages sent directly to +-- the mailbox will still be handled via the internal buffers and subjected to +-- the mailbox limits. The 'post' API is really just a means to ensure that +-- the conversion from @Serializable a -> Message@ is done in the caller's +-- process and uses the safe @wrapMessage@ variant. +-- +-- [Acknowledgements] +-- +-- This API is based on the work of Erlang programmers Fred Hebert and +-- Geoff Cant, its design closely mirroring that of the the /pobox/ library +-- application. +-- +----------------------------------------------------------------------------- +module Control.Distributed.Process.Execution.Mailbox + ( + -- * Creating, Starting, Configuring and Running a Mailbox + Mailbox() + , startMailbox + , startSupervised + , startSupervisedMailbox + , createMailbox + , resize + , statistics + , monitor + , Limit + , BufferType(..) + , MailboxStats(..) + -- * Posting Mail + , post + -- * Obtaining Mail and Notifications + , notify + , deliver + , active + , NewMail(..) + , Delivery(..) + , FilterResult(..) + , acceptEverything + , acceptMatching + -- * Remote Table + , __remoteTable + ) where + +import Control.Concurrent.STM (atomically) +import Control.Concurrent.STM.TChan + ( TChan + , newBroadcastTChanIO + , dupTChan + , readTChan + , writeTChan + ) +import Control.Distributed.Process hiding (call, monitor) +import qualified Control.Distributed.Process as P (monitor) +import Control.Distributed.Process.Closure + ( remotable + , mkStaticClosure + ) +import Control.Distributed.Process.Serializable hiding (SerializableDict) +import Control.Distributed.Process.Extras.Internal.Types + ( ExitReason(..) + , Resolvable(..) + , Routable(..) + , Linkable(..) + ) +import Control.Distributed.Process.ManagedProcess + ( call + , sendControlMessage + , channelControlPort + , handleControlChan + , handleInfo + , handleRaw + , continue + , defaultProcess + , UnhandledMessagePolicy(..) + , InitHandler + , InitResult(..) + , ProcessAction + , ProcessDefinition(..) + , ControlChannel + , ControlPort + ) +import qualified Control.Distributed.Process.ManagedProcess as MP + ( chanServe + ) +import Control.Distributed.Process.ManagedProcess.Server + ( stop + ) +import Control.Distributed.Process.ManagedProcess.Server.Restricted as Restricted + ( getState + , Result + , RestrictedProcess + ) +import qualified Control.Distributed.Process.ManagedProcess.Server.Restricted as Restricted + ( handleCall + , reply + ) +import Control.Distributed.Process.Supervisor (SupervisorPid) +import Control.Distributed.Process.Extras.Time +import Control.Exception (SomeException) +import Data.Accessor + ( Accessor + , accessor + , (^:) + , (.>) + , (^=) + , (^.) + ) +import Data.Binary +import qualified Data.Foldable as Foldable +import Data.Sequence + ( Seq + , ViewL(EmptyL, (:<)) + , ViewR(EmptyR, (:>)) + , (<|) + , (|>) + ) +import qualified Data.Sequence as Seq +import Data.Typeable (Typeable) + +import GHC.Generics + +#if ! MIN_VERSION_base(4,6,0) +import Prelude hiding (catch, drop) +#else +import Prelude hiding (drop) +#endif + +-------------------------------------------------------------------------------- +-- Types -- +-------------------------------------------------------------------------------- + +-- external client/configuration API + +-- | Opaque handle to a mailbox. +-- +data Mailbox = Mailbox { pid :: !ProcessId + , cchan :: !(ControlPort ControlMessage) + } deriving (Typeable, Generic, Eq) +instance Binary Mailbox where +instance Show Mailbox where + show = ("Mailbox:" ++) . show . pid + +instance Linkable Mailbox where + linkTo = link . pid + +instance Resolvable Mailbox where + resolve = return . Just . pid + +instance Routable Mailbox where + sendTo = post + unsafeSendTo = post + +sendCtrlMsg :: Mailbox + -> ControlMessage + -> Process () +sendCtrlMsg Mailbox{..} = sendControlMessage cchan + +-- | Describes the different types of buffer. +-- +data BufferType = + Queue -- ^ FIFO buffer, limiter drops the eldest message (queue head) + | Stack -- ^ unordered buffer, limiter drops the newest (top) message + | Ring -- ^ FIFO buffer, limiter refuses (i.e., drops) new messages + deriving (Typeable, Eq, Show) + +-- TODO: re-implement this process in terms of a limiter expression, i.e., +-- +-- data Limit s = Accept s | Block s +-- +-- limit :: forall s. Closure (Message {- new mail -} -> Process (Limit s)) + +-- | Represents the maximum number of messages the internal buffer can hold. +-- +type Limit = Integer + +-- | A @Closure@ used to filter messages in /active/ mode. +-- +type Filter = Closure (Message -> Process FilterResult) + +-- | Marker message indicating to the owning process that mail has arrived. +-- +data NewMail = NewMail !Mailbox !Integer + deriving (Typeable, Generic, Show) +instance Binary NewMail where + +-- | Mail delivery. +-- +data Delivery = Delivery { box :: Mailbox -- ^ handle to the sending mailbox + , messages :: [Message] -- ^ list of raw messages + , count :: Integer -- ^ number of messages delivered + , totalDropped :: Integer -- ^ total dropped/skipped messages + } + deriving (Typeable, Generic) +instance Binary Delivery where + +-- TODO: keep running totals and send them with the stats... + +-- | Bundle of statistics data, available on request via +-- the 'mailboxStats' API call. +-- +data MailboxStats = + MailboxStats { pendingMessages :: Integer + , droppedMessages :: Integer + , currentLimit :: Limit + , owningProcess :: ProcessId + } deriving (Typeable, Generic, Show) +instance Binary MailboxStats where + +-- internal APIs + +data Post = Post !Message + deriving (Typeable, Generic) +instance Binary Post where + +data StatsReq = StatsReq + deriving (Typeable, Generic) +instance Binary StatsReq where + +data FilterResult = Keep | Skip | Send + deriving (Typeable, Generic) +instance Binary FilterResult + +data Mode = + Active !Filter -- ^ Send all buffered messages (or wait until one arrives) + | Notify -- ^ Send a notification once messages are ready to be received + | Passive -- ^ Accumulate messages in the buffer, dropping them if necessary + deriving (Typeable, Generic) +instance Binary Mode where +instance Show Mode where + show (Active _) = "Active" + show Notify = "Notify" + show Passive = "Passive" + +data ControlMessage = + Resize !Integer + | SetActiveMode !Mode + deriving (Typeable, Generic) +instance Binary ControlMessage where + +class Buffered a where + tag :: a -> BufferType + push :: Message -> a -> a + pop :: a -> Maybe (Message, a) + adjust :: Limit -> a -> a + drop :: Integer -> a -> a + +data BufferState = + BufferState { _mode :: Mode + , _bufferT :: BufferType + , _limit :: Limit + , _size :: Integer + , _dropped :: Integer + , _owner :: ProcessId + , ctrlChan :: ControlPort ControlMessage + } + +defaultState :: BufferType + -> Limit + -> ProcessId + -> ControlPort ControlMessage + -> BufferState +defaultState bufferT limit' pid cc = + BufferState { _mode = Passive + , _bufferT = bufferT + , _limit = limit' + , _size = 0 + , _dropped = 0 + , _owner = pid + , ctrlChan = cc + } + +data State = State { _buffer :: Seq Message + , _state :: BufferState + } + +instance Buffered State where + tag q = _bufferT $ _state q + + -- see note [buffer enqueue/dequeue semantics] + push m = (state .> size ^: (+1)) . (buffer ^: (m <|)) + + -- see note [buffer enqueue/dequeue semantics] + pop q = maybe Nothing + (\(s' :> a) -> Just (a, ( (buffer ^= s') + . (state .> size ^: (1-)) + $ q))) $ getR (q ^. buffer) + + adjust sz q = (state .> limit ^= sz) $ maybeDrop + where + maybeDrop + | size' <- (q ^. state ^. size), + size' > sz = (state .> size ^= sz) $ drop (size' - sz) q + | otherwise = q + + -- see note [buffer drop semantics] + drop n q + | n > 1 = drop (n - 1) $ drop 1 q + | isQueue q = dropR q + | otherwise = dropL q + where + dropR q' = maybe q' (\(s' :> _) -> dropOne q' s') $ getR (q' ^. buffer) + dropL q' = maybe q' (\(_ :< s') -> dropOne q' s') $ getL (q' ^. buffer) + dropOne q' s = ( (buffer ^= s) + . (state .> size ^: (\n' -> n' - 1)) + . (state .> dropped ^: (+1)) + $ q' ) + +{- note [buffer enqueue/dequeue semantics] +If we choose to add a message to the buffer, it is always +added to the left hand side of the sequence. This gives +FIFO (enqueue to tail) semantics for queues, LIFO (push +new head) semantics for stacks when dropping messages - note +that dequeueing will always take the eldest (RHS) message, +regardless of the buffer type - and queue-like semantics for +the ring buffer. + +We /always/ take the eldest message each time we dequeue, +in an attempt to maintain something approaching FIFO order +when processing the mailbox, for all data structures. Where +we do not achieve this is dropping messages, since the different +buffer types drop messages either on the right (eldest) or left +(youngest). + +-- note [buffer drop semantics] + +The "stack buffer", when full, only ever attempts to drop the +youngest (leftmost) message, such that it guarantees no ordering +at all, but that is enforced by the code calling 'drop' rather +than the data structure itself. The ring buffer behaves similarly, +since it rejects new messages altogether, which in practise means +dropping from the LHS. + +-} + +-------------------------------------------------------------------------------- +-- Starting/Running a Mailbox -- +-------------------------------------------------------------------------------- + +-- | Start a mailbox for the calling process. +-- +-- > create = getSelfPid >>= start +-- +createMailbox :: BufferType -> Limit -> Process Mailbox +createMailbox buffT maxSz = + getSelfPid >>= \self -> startMailbox self buffT maxSz + +-- | Start a mailbox for the supplied @ProcessId@. +-- +-- > start = spawnLocal $ run +-- +startMailbox :: ProcessId -> BufferType -> Limit -> Process Mailbox +startMailbox = doStartMailbox Nothing + +-- | As 'startMailbox', but suitable for use in supervisor child specs. +-- This variant is for use when you want to access to the underlying +-- 'Mailbox' handle in your supervised child refs. See supervisor's +-- @ChildRef@ data type for more information. +-- +-- Example: +-- > childSpec = toChildStart $ startSupervised pid bufferType mboxLimit +-- +-- See "Control.Distributed.Process.Supervisor" +-- +startSupervised :: ProcessId + -> BufferType + -> Limit + -> SupervisorPid + -> Process (ProcessId, Message) +startSupervised p b l s = do + mb <- startSupervisedMailbox p b l s + return (pid mb, unsafeWrapMessage mb) + +-- | As 'startMailbox', but suitable for use in supervisor child specs. +-- +-- Example: +-- > childSpec = toChildStart $ startSupervisedMailbox pid bufferType mboxLimit +-- +-- See "Control.Distributed.Process.Supervisor" +-- +startSupervisedMailbox :: ProcessId + -> BufferType + -> Limit + -> SupervisorPid + -> Process Mailbox +startSupervisedMailbox p b l s = doStartMailbox (Just s) p b l + +doStartMailbox :: Maybe SupervisorPid + -> ProcessId + -> BufferType + -> Limit + -> Process Mailbox +doStartMailbox mSp p b l = do + bchan <- liftIO $ newBroadcastTChanIO + rchan <- liftIO $ atomically $ dupTChan bchan + spawnLocal (maybeLink mSp >> runMailbox bchan p b l) >>= \pid -> do + cc <- liftIO $ atomically $ readTChan rchan + return $ Mailbox pid cc + where + maybeLink Nothing = return () + maybeLink (Just p') = link p' + +-- | Run the mailbox server loop. +-- +runMailbox :: TChan (ControlPort ControlMessage) + -> ProcessId + -> BufferType + -> Limit + -> Process () +runMailbox tc pid buffT maxSz = do + link pid + tc' <- liftIO $ atomically $ dupTChan tc + MP.chanServe (pid, buffT, maxSz) (mboxInit tc') (processDefinition pid tc) + +-------------------------------------------------------------------------------- +-- Mailbox Initialisation/Startup -- +-------------------------------------------------------------------------------- + +mboxInit :: TChan (ControlPort ControlMessage) + -> InitHandler (ProcessId, BufferType, Limit) State +mboxInit tc (pid, buffT, maxSz) = do + cc <- liftIO $ atomically $ readTChan tc + return $ InitOk (State Seq.empty $ defaultState buffT maxSz pid cc) Infinity + +-------------------------------------------------------------------------------- +-- Client Facing API -- +-------------------------------------------------------------------------------- + +-- | Monitor a mailbox. +-- +monitor :: Mailbox -> Process MonitorRef +monitor = P.monitor . pid + +-- | Instructs the mailbox to send a 'NewMail' signal as soon as any mail is +-- available for delivery. Once the signal is sent, it will not be resent, even +-- when further mail arrives, until 'notify' is called again. +-- +-- NB: signals are /only/ delivered to the mailbox's owning process. +-- +notify :: Mailbox -> Process () +notify mb = sendCtrlMsg mb $ SetActiveMode Notify + +-- | Instructs the mailbox to send a 'Delivery' as soon as any mail is +-- available, or immediately (if the buffer already contains data). +-- +-- NB: signals are /only/ delivered to the mailbox's owning process. +-- +active :: Mailbox -> Filter -> Process () +active mb f = sendCtrlMsg mb $ SetActiveMode $ Active f + +-- | Alters the mailbox's /limit/ - this might cause messages to be dropped! +-- +resize :: Mailbox -> Integer -> Process () +resize mb sz = sendCtrlMsg mb $ Resize sz + +-- | Posts a message to someone's mailbox. +-- +post :: Serializable a => Mailbox -> a -> Process () +post Mailbox{..} m = send pid (Post $ wrapMessage m) + +-- | Obtain statistics (from/to anywhere) about a mailbox. +-- +statistics :: Mailbox -> Process MailboxStats +statistics mb = call mb StatsReq + +-------------------------------------------------------------------------------- +-- PRIVATE Filter Implementation(s) -- +-------------------------------------------------------------------------------- + +everything :: Message -> Process FilterResult +everything _ = return Keep + +matching :: Closure (Message -> Process FilterResult) + -> Message + -> Process FilterResult +matching predicate msg = do + pred' <- unClosure predicate :: Process (Message -> Process FilterResult) + res <- handleMessage msg pred' + case res of + Nothing -> return Skip + Just fr -> return fr + +-------------------------------------------------------------------------------- +-- Process Definition/State & API Handlers -- +-------------------------------------------------------------------------------- + +processDefinition :: ProcessId + -> TChan (ControlPort ControlMessage) + -> ControlChannel ControlMessage + -> Process (ProcessDefinition State) +processDefinition pid tc cc = do + liftIO $ atomically $ writeTChan tc $ channelControlPort cc + return $ defaultProcess { apiHandlers = [ + handleControlChan cc handleControlMessages + , Restricted.handleCall handleGetStats + ] + , infoHandlers = [ handleInfo handlePost + , handleRaw handleRawInputs ] + , unhandledMessagePolicy = DeadLetter pid + } :: Process (ProcessDefinition State) + +handleControlMessages :: State + -> ControlMessage + -> Process (ProcessAction State) +handleControlMessages st cm + | (SetActiveMode new) <- cm = activateMode st new + | (Resize sz') <- cm = continue $ adjust sz' st + | otherwise = stop $ ExitOther "IllegalState" + where + activateMode :: State -> Mode -> Process (ProcessAction State) + activateMode st' new + | sz <- (st ^. state ^. size) + , sz == 0 = continue $ updated st' new + | otherwise = do + let updated' = updated st' new + case new of + Notify -> sendNotification updated' >> continue updated' + (Active _) -> sendMail updated' >>= continue + Passive -> {- shouldn't happen! -} die $ "IllegalState" + + updated s m = (state .> mode ^= m) s + +handleGetStats :: StatsReq -> RestrictedProcess State (Result MailboxStats) +handleGetStats _ = Restricted.reply . (^. stats) =<< getState + +handleRawInputs :: State -> Message -> Process (ProcessAction State) +handleRawInputs st msg = handlePost st (Post msg) + +handlePost :: State -> Post -> Process (ProcessAction State) +handlePost st (Post msg) = do + let st' = insert msg st + continue . (state .> mode ^= Passive) =<< forwardIfNecessary st' + where + forwardIfNecessary s + | Notify <- currentMode = sendNotification s >> return s + | Active _ <- currentMode = sendMail s + | otherwise = return s + + currentMode = st ^. state ^. mode + +-------------------------------------------------------------------------------- +-- Accessors, State/Stats Management & Utilities -- +-------------------------------------------------------------------------------- + +sendNotification :: State -> Process () +sendNotification st = do + pid <- getSelfPid + send ownerPid $ NewMail (Mailbox pid cchan) pending + where + ownerPid = st ^. state ^. owner + pending = st ^. state ^. size + cchan = ctrlChan (st ^. state) + +type Count = Integer +type Skipped = Integer + +sendMail :: State -> Process State +sendMail st = do + let Active f = st ^. state ^. mode + unCl <- catch (unClosure f >>= return . Just) + (\(_ :: SomeException) -> return Nothing) + case unCl of + Nothing -> return st -- TODO: Logging!? + Just f' -> do + (st', cnt, skipped, msgs) <- applyFilter f' st + us <- getSelfPid + send ownerPid $ Delivery { box = Mailbox us (ctrlChan $ st ^. state) + , messages = Foldable.toList msgs + , count = cnt + , totalDropped = skipped + droppedMsgs + } + return $ ( (state .> dropped ^= 0) + . (state .> size ^: ((cnt + skipped) -)) + $ st' ) + where + applyFilter f s = filterMessages f (s, 0, 0, Seq.empty) + + filterMessages :: (Message -> Process FilterResult) + -> (State, Count, Skipped, Seq Message) + -> Process (State, Count, Skipped, Seq Message) + filterMessages f accIn@(buff, cnt, drp, acc) = do + case pop buff of + Nothing -> return accIn + Just (m, buff') -> do + res <- f m + case res of + Keep -> filterMessages f (buff', cnt + 1, drp, acc |> m) + Skip -> filterMessages f (buff', cnt, drp + 1, acc) + Send -> return accIn + + ownerPid = st ^. state ^. owner + droppedMsgs = st ^. state ^. dropped + +insert :: Message -> State -> State +insert msg st@(State _ BufferState{..}) = + if _size /= _limit + then push msg st + else case _bufferT of + Ring -> (state .> dropped ^: (+1)) st + _ -> push msg $ drop 1 st + +isQueue :: State -> Bool +isQueue = (== Queue) . _bufferT . _state + +isStack :: State -> Bool +isStack = (== Stack) . _bufferT . _state + +getR :: Seq a -> Maybe (ViewR a) +getR s = + case Seq.viewr s of + EmptyR -> Nothing + a -> Just a + +getL :: Seq a -> Maybe (ViewL a) +getL s = + case Seq.viewl s of + EmptyL -> Nothing + a -> Just a + +mode :: Accessor BufferState Mode +mode = accessor _mode (\m st -> st { _mode = m }) + +bufferType :: Accessor BufferState BufferType +bufferType = accessor _bufferT (\t st -> st { _bufferT = t }) + +limit :: Accessor BufferState Limit +limit = accessor _limit (\l st -> st { _limit = l }) + +size :: Accessor BufferState Integer +size = accessor _size (\s st -> st { _size = s }) + +dropped :: Accessor BufferState Integer +dropped = accessor _dropped (\d st -> st { _dropped = d }) + +owner :: Accessor BufferState ProcessId +owner = accessor _owner (\o st -> st { _owner = o }) + +buffer :: Accessor State (Seq Message) +buffer = accessor _buffer (\b qb -> qb { _buffer = b }) + +state :: Accessor State BufferState +state = accessor _state (\s qb -> qb { _state = s }) + +stats :: Accessor State MailboxStats +stats = accessor getStats (\_ s -> s) -- TODO: use a READ ONLY accessor for this + where + getStats (State _ (BufferState _ _ lm sz dr op _)) = MailboxStats sz dr lm op + +$(remotable ['everything, 'matching]) + +-- | A /do-nothing/ filter that accepts all messages (i.e., returns @Keep@ +-- for any input). +acceptEverything :: Closure (Message -> Process FilterResult) +acceptEverything = $(mkStaticClosure 'everything) + +-- | A filter that takes a @Closure (Message -> Process FilterResult)@ holding +-- the filter function and applies it remotely (i.e., in the mailbox's own +-- managed process). +-- +acceptMatching :: Closure (Closure (Message -> Process FilterResult) + -> Message -> Process FilterResult) +acceptMatching = $(mkStaticClosure 'matching) + +-- | Instructs the mailbox to deliver all pending messages to the owner. +-- +deliver :: Mailbox -> Process () +deliver mb = active mb acceptEverything + diff --git a/tests/MailboxTestFilters.hs b/tests/MailboxTestFilters.hs new file mode 100644 --- /dev/null +++ b/tests/MailboxTestFilters.hs @@ -0,0 +1,43 @@ +{-# LANGUAGE TemplateHaskell #-} +{-# LANGUAGE ScopedTypeVariables #-} + +module MailboxTestFilters where + +import Control.Distributed.Process +import Control.Distributed.Process.Execution.Mailbox (FilterResult(..)) +import Control.Monad (forM) + +#if ! MIN_VERSION_base(4,6,0) +import Prelude hiding (catch, drop) +#else +import Prelude hiding (drop) +#endif +import Data.Maybe (catMaybes) +import Control.Distributed.Process.Closure (remotable, mkClosure, mkStaticClosure) + +filterInputs :: (String, Int, Bool) -> Message -> Process FilterResult +filterInputs (s, i, b) msg = do + rs <- forM [ \m -> handleMessageIf m (\s' -> s' == s) (\_ -> return Keep) + , \m -> handleMessageIf m (\i' -> i' == i) (\_ -> return Keep) + , \m -> handleMessageIf m (\b' -> b' == b) (\_ -> return Keep) + ] $ \h -> h msg + if (length (catMaybes rs) > 0) + then return Keep + else return Skip + +filterEvens :: Message -> Process FilterResult +filterEvens m = do + matched <- handleMessage m (\(i :: Int) -> do + if even i then return Keep else return Skip) + case matched of + Just fr -> return fr + _ -> return Skip + +$(remotable ['filterInputs, 'filterEvens]) + +intFilter :: Closure (Message -> Process FilterResult) +intFilter = $(mkStaticClosure 'filterEvens) + +myFilter :: (String, Int, Bool) -> Closure (Message -> Process FilterResult) +myFilter = $(mkClosure 'filterInputs) + diff --git a/tests/TestExchange.hs b/tests/TestExchange.hs new file mode 100644 --- /dev/null +++ b/tests/TestExchange.hs @@ -0,0 +1,186 @@ +{-# LANGUAGE BangPatterns #-} +{-# LANGUAGE TemplateHaskell #-} +{-# LANGUAGE ScopedTypeVariables #-} +{-# LANGUAGE MultiParamTypeClasses #-} + +module Main where + +import Control.Distributed.Process hiding (monitor) +import Control.Distributed.Process.Node +import Control.Distributed.Process.Execution.EventManager hiding (start) +import qualified Control.Distributed.Process.Extras +import Control.Distributed.Process.Execution.Exchange +import Control.Distributed.Process.Extras.Internal.Types +import Control.Distributed.Process.Extras.Internal.Primitives +import qualified Control.Distributed.Process.Execution.EventManager as EventManager + ( start + ) +import Control.Monad (void, forM, forever) +import Control.Rematch (equalTo) + +#if ! MIN_VERSION_base(4,6,0) +import Prelude hiding (catch, drop) +#else +import Prelude hiding (drop) +#endif +import qualified Network.Transport as NT +import Test.Framework as TF (testGroup, Test) +import Test.Framework.Providers.HUnit +import TestUtils + +testKeyBasedRouting :: TestResult Bool -> Process () +testKeyBasedRouting result = do + (sp, rp) <- newChan :: Process (Channel Int) + rex <- messageKeyRouter PayloadOnly + + -- Since the /router/ doesn't offer a syncrhonous start + -- option, we use spawnSignalled to get the same effect, + -- making it more likely (though it's not guaranteed) that + -- the spawned process will be bound to the routing exchange + -- prior to our evaluating 'routeMessage' below. + void $ spawnSignalled (bindKey "foobar" rex) $ const $ do + receiveWait [ match (\(s :: Int) -> sendChan sp s) ] + + routeMessage rex (createMessage "foobar" [] (123 :: Int)) + stash result . (== (123 :: Int)) =<< receiveChan rp + +testMultipleRoutes :: TestResult () -> Process () +testMultipleRoutes result = do + stash result () -- we don't rely on the test result for assertions... + (sp, rp) <- newChan + rex <- messageKeyRouter PayloadOnly + let recv = receiveWait [ + match (\(s :: String) -> getSelfPid >>= \us -> sendChan sp (us, Left s)) + , match (\(i :: Int) -> getSelfPid >>= \us -> sendChan sp (us, Right i)) + ] + + us <- getSelfPid + p1 <- spawnSignalled (link us >> bindKey "abc" rex) (const $ forever recv) + p2 <- spawnSignalled (link us >> bindKey "def" rex) (const $ forever recv) + p3 <- spawnSignalled (link us >> bindKey "abc" rex) (const $ forever recv) + + -- publish 2 messages with the routing-key set to 'abc' + routeMessage rex (createMessage "abc" [] "Hello") + routeMessage rex (createMessage "abc" [] (123 :: Int)) + + -- route another message with the 'abc' value a header (should be ignored) + routeMessage rex (createMessage "" [("abc", "abc")] "Goodbye") + + received <- forM (replicate (2 * 3) us) (const $ receiveChanTimeout 1000 rp) + + -- all bindings for 'abc' fired correctly + received `shouldContain` Just (p1, Left "Hello") + received `shouldContain` Just (p3, Left "Hello") + received `shouldContain` Just (p1, Right (123 :: Int)) + received `shouldContain` Just (p3, Right (123 :: Int)) + + -- however the bindings for 'def' never fired + received `shouldContain` Nothing + received `shouldNotContain` Just (p2, Left "Hello") + received `shouldNotContain` Just (p2, Right (123 :: Int)) + + -- none of the bindings should have examined the headers! + received `shouldNotContain` Just (p1, Left "Goodbye") + received `shouldNotContain` Just (p2, Left "Goodbye") + received `shouldNotContain` Just (p3, Left "Goodbye") + +testHeaderBasedRouting :: TestResult () -> Process () +testHeaderBasedRouting result = do + stash result () -- we don't rely on the test result for assertions... + (sp, rp) <- newChan + rex <- headerContentRouter PayloadOnly "x-name" + let recv = const $ forever $ receiveWait [ + match (\(s :: String) -> getSelfPid >>= \us -> sendChan sp (us, Left s)) + , match (\(i :: Int) -> getSelfPid >>= \us -> sendChan sp (us, Right i)) + ] + + us <- getSelfPid + p1 <- spawnSignalled (link us >> bindHeader "x-name" "yellow" rex) recv + p2 <- spawnSignalled (link us >> bindHeader "x-name" "red" rex) recv + _ <- spawnSignalled (link us >> bindHeader "x-type" "fast" rex) recv + + -- publish 2 messages with the routing-key set to 'abc' + routeMessage rex (createMessage "" [("x-name", "yellow")] "Hello") + routeMessage rex (createMessage "" [("x-name", "yellow")] (123 :: Int)) + routeMessage rex (createMessage "" [("x-name", "red")] (456 :: Int)) + routeMessage rex (createMessage "" [("x-name", "red")] (789 :: Int)) + routeMessage rex (createMessage "" [("x-type", "fast")] "Goodbye") + + -- route another message with the 'abc' value a header (should be ignored) + routeMessage rex (createMessage "" [("abc", "abc")] "FooBar") + + received <- forM (replicate 5 us) (const $ receiveChanTimeout 1000 rp) + + -- all bindings fired correctly + received `shouldContain` Just (p1, Left "Hello") + received `shouldContain` Just (p1, Right (123 :: Int)) + received `shouldContain` Just (p2, Right (456 :: Int)) + received `shouldContain` Just (p2, Right (789 :: Int)) + received `shouldContain` Nothing + + -- simple check that no other bindings have fired + length received `shouldBe` equalTo (5 :: Int) + +testSimpleEventHandling :: TestResult Bool -> Process () +testSimpleEventHandling result = do + (sp, rp) <- newChan + (sigStart, recvStart) <- newChan + em <- EventManager.start + Just pid <- resolve em + void $ monitor pid + + -- Note that in our init (state) function, we write a "start signal" + -- here; Without a start signal, the message sent to the event manager + -- (via notify) would race with the addHandler registration. + pid' <- addHandler em (myHandler sp) (sendChan sigStart ()) + link pid' + + () <- receiveChan recvStart + + notify em ("hello", "event", "manager") -- cast message + r <- receiveTimeout 100000000 [ + matchChan rp return + , match (\(ProcessMonitorNotification _ _ _) -> die "ServerDied") + ] + case r of + Just ("hello", "event", "manager") -> stash result True + _ -> stash result False + +myHandler :: SendPort (String, String, String) + -> () + -> (String, String, String) + -> Process () +myHandler sp s m@(_, _, _) = sendChan sp m >> return s + +myRemoteTable :: RemoteTable +myRemoteTable = + Control.Distributed.Process.Extras.__remoteTable initRemoteTable + +tests :: NT.Transport -> IO [Test] +tests transport = do + localNode <- newLocalNode transport myRemoteTable + return [ + testGroup "Event Manager" + [ + testCase "Simple Event Handlers" + (delayedAssertion "Expected the handler to run" + localNode True testSimpleEventHandling) + ] + + , testGroup "Router" + [ + testCase "Direct Key Routing" + (delayedAssertion "Expected the sole matching route to run" + localNode True testKeyBasedRouting) + , testCase "Key Based Selective Routing" + (delayedAssertion "Expected only the matching routes to run" + localNode () testMultipleRoutes) + , testCase "Header Based Selective Routing" + (delayedAssertion "Expected only the matching routes to run" + localNode () testHeaderBasedRouting) + ] + ] + +main :: IO () +main = testMain $ tests + diff --git a/tests/TestMailbox.hs b/tests/TestMailbox.hs new file mode 100644 --- /dev/null +++ b/tests/TestMailbox.hs @@ -0,0 +1,256 @@ +{-# LANGUAGE BangPatterns #-} +{-# LANGUAGE TemplateHaskell #-} +{-# LANGUAGE ScopedTypeVariables #-} + +module Main where + +import Control.Distributed.Process +import Control.Distributed.Process.Node +import qualified Control.Distributed.Process.Extras (__remoteTable) +import qualified Control.Distributed.Process.Execution.Mailbox (__remoteTable) +import Control.Distributed.Process.Execution.Mailbox +import Control.Distributed.Process.Extras.Time +import Control.Distributed.Process.Extras.Timer + +import Control.Rematch (equalTo) + +#if ! MIN_VERSION_base(4,6,0) +import Prelude hiding (catch, drop) +#else +import Prelude hiding (drop) +#endif + +import Data.Maybe (catMaybes) + +import Test.Framework as TF (testGroup, Test) +import Test.Framework.Providers.HUnit +import TestUtils +import qualified MailboxTestFilters (__remoteTable) +import MailboxTestFilters (myFilter, intFilter) + +import qualified Network.Transport as NT + +-- TODO: This whole test suite would be much better off using QuickCheck. +-- The test-framework driver however, doesn't have the API support we'd need +-- to wire in our tests, so we'll have to write a compatibility layer. +-- That should probably go into (or beneath) the C.D.P.P.Test module. + +allBuffersShouldRespectFIFOOrdering :: BufferType -> TestResult Bool -> Process () +allBuffersShouldRespectFIFOOrdering buffT result = do + let [a, b, c] = ["a", "b", "c"] + mbox <- createAndSend buffT [a, b, c] + active mbox acceptEverything + Just Delivery { messages = msgs } <- receiveTimeout (after 2 Seconds) + [ match return ] + let [ma', mb', mc'] = msgs + Just a' <- unwrapMessage ma' :: Process (Maybe String) + Just b' <- unwrapMessage mb' :: Process (Maybe String) + Just c' <- unwrapMessage mc' :: Process (Maybe String) + let values = [a', b', c'] + stash result $ values == [a, b, c] +-- if values /= [a, b, c] +-- then liftIO $ putStrLn $ "unexpected " ++ ((show buffT) ++ (" values: " ++ (show values))) +-- else return () + +resizeShouldRespectOrdering :: BufferType + -> TestResult [String] + -> Process () +resizeShouldRespectOrdering buffT result = do + let [a, b, c, d, e] = ["a", "b", "c", "d", "e"] + mbox <- createAndSend buffT [a, b, c, d, e] + resize mbox (3 :: Integer) + + active mbox acceptEverything + Just Delivery{ messages = msgs } <- receiveTimeout (after 2 Seconds) [ match return ] + + let [mc', md', me'] = msgs + Just c' <- unwrapMessage mc' :: Process (Maybe String) + Just d' <- unwrapMessage md' :: Process (Maybe String) + Just e' <- unwrapMessage me' :: Process (Maybe String) + let values = [c', d', e'] + stash result $ values + +bufferLimiting :: BufferType -> TestResult (Integer, [Maybe String]) -> Process () +bufferLimiting buffT result = do + let msgs = ["a", "b", "c", "d", "e", "f", "g"] + mbox <- createMailboxAndPost buffT 4 msgs + + MailboxStats{ pendingMessages = pending' + , droppedMessages = dropped' + , currentLimit = limit' } <- statistics mbox + pending' `shouldBe` equalTo 4 + dropped' `shouldBe` equalTo 3 + limit' `shouldBe` equalTo 4 + + active mbox acceptEverything + Just Delivery{ messages = recvd + , totalDropped = skipped } <- receiveTimeout (after 5 Seconds) + [ match return ] + seen <- mapM unwrapMessage recvd + stash result (skipped, seen) + +mailboxIsInitiallyPassive :: TestResult Bool -> Process () +mailboxIsInitiallyPassive result = do + mbox <- createMailbox Stack (6 :: Integer) + mapM_ (post mbox) ([1..5] :: [Int]) + Nothing <- receiveTimeout (after 3 Seconds) [ matchAny return ] + notify mbox + inbound <- receiveTimeout (after 3 Seconds) [ match return ] + case inbound of + Just (NewMail _ _) -> stash result True + Nothing -> stash result False + +complexMailboxFiltering :: (String, Int, Bool) + -> TestResult (String, Int, Bool) + -> Process () +complexMailboxFiltering inputs@(s', i', b') result = do + mbox <- createMailbox Stack (10 :: Integer) + post mbox s' + post mbox i' + post mbox b' + waitForMailboxReady mbox 3 + + active mbox $ myFilter inputs + Just Delivery{ messages = [m1, m2, m3] + , totalDropped = _ } <- receiveTimeout (after 5 Seconds) + [ match return ] + Just s <- unwrapMessage m1 :: Process (Maybe String) + Just i <- unwrapMessage m2 :: Process (Maybe Int) + Just b <- unwrapMessage m3 :: Process (Maybe Bool) + stash result $ (s, i, b) + +dropDuringFiltering :: TestResult Bool -> Process () +dropDuringFiltering result = do + let rng = [1..50] :: [Int] + mbox <- createMailbox Stack (50 :: Integer) + mapM_ (post mbox) rng + + waitForMailboxReady mbox 50 + active mbox $ intFilter + + Just Delivery{ messages = msgs } <- receiveTimeout (after 5 Seconds) + [ match return ] + seen <- mapM unwrapMessage msgs + stash result $ (catMaybes seen) == (filter even rng) + +mailboxHandleReUse :: TestResult Bool -> Process () +mailboxHandleReUse result = do + mbox <- createMailbox Queue (1 :: Limit) + post mbox "abc" + + notify mbox + Just (NewMail mbox' _) <- receiveTimeout (after 2 Seconds) + [ match return ] + deliver mbox' + _ <- expect :: Process Delivery + stash result True + +createAndSend :: BufferType -> [String] -> Process Mailbox +createAndSend buffT msgs = createMailboxAndPost buffT 10 msgs + +createMailboxAndPost :: BufferType -> Limit -> [String] -> Process Mailbox +createMailboxAndPost buffT maxSz msgs = do + (cc, cp) <- newChan + mbox <- createMailbox buffT maxSz + spawnLocal $ mapM_ (post mbox) msgs >> sendChan cc () + () <- receiveChan cp + waitForMailboxReady mbox $ min (toInteger (length msgs)) maxSz + return mbox + +waitForMailboxReady :: Mailbox -> Integer -> Process () +waitForMailboxReady mbox sz = do + sleep $ seconds 1 + notify mbox + m <- receiveWait [ + matchIf (\(NewMail mbox' sz') -> mbox == mbox' && sz' >= sz) + (\_ -> return True) + , match (\(NewMail _ _) -> return False) + , matchAny (\_ -> return False) + ] + case m of + True -> return () + False -> waitForMailboxReady mbox sz + +myRemoteTable :: RemoteTable +myRemoteTable = + Control.Distributed.Process.Execution.Mailbox.__remoteTable $ + Control.Distributed.Process.Extras.__remoteTable $ + MailboxTestFilters.__remoteTable initRemoteTable + +tests :: NT.Transport -> IO [Test] +tests transport = do + {- verboseCheckWithResult stdArgs -} + localNode <- newLocalNode transport myRemoteTable + return [ + testGroup "Dequeue/Pop Ordering" + [ + testCase "Queue Ordering" + (delayedAssertion + "Expected the Queue to offer FIFO ordering" + localNode True (allBuffersShouldRespectFIFOOrdering Queue)) + + , testCase "Stack Ordering" + (delayedAssertion + "Expected the Queue to offer FIFO ordering" + localNode True (allBuffersShouldRespectFIFOOrdering Stack)) + , testCase "Ring Ordering" + (delayedAssertion + "Expected the Queue to offer FIFO ordering" + localNode True (allBuffersShouldRespectFIFOOrdering Ring)) + ] + , testGroup "Resize & Ordering" + [ + testCase "Queue Drops Eldest" + (delayedAssertion + "expected c, d, e" + localNode ["c", "d", "e"] $ resizeShouldRespectOrdering Queue) + , testCase "Stack Drops Youngest" + (delayedAssertion + "expected a, b, c" + localNode ["a", "b", "c"] $ resizeShouldRespectOrdering Stack) + , testCase "Ring Drops Youngest" + (delayedAssertion + "expected a, b, c" + localNode ["a", "b", "c"] $ resizeShouldRespectOrdering Ring) + ] + , testGroup "Buffer Limits & Discarded Messages" + [ + testCase "Queue Drops Eldest and Enqueues New" + (delayedAssertion + "expected d, e, f, g" + localNode ((3 :: Integer), map Just ["d", "e", "f", "g"]) $ bufferLimiting Queue) + , testCase "Stack Drops Youngest And Pushes New" + (delayedAssertion + "expected a, b, c, g" + localNode ((3 :: Integer), map Just ["a", "b", "c", "g"]) $ bufferLimiting Stack) + , testCase "Ring Rejects New Entries" + (delayedAssertion + "expected a, b, c, d" + localNode ((3 :: Integer), map Just ["a", "b", "c", "d"]) $ bufferLimiting Ring) + ] + , testGroup "Notification, Activation and Delivery" + [ + testCase "Mailbox is initially Passive" + (delayedAssertion + "Expected the Mailbox to remain passive until told otherwise" + localNode True mailboxIsInitiallyPassive) + , testCase "Mailbox Notifications include usable control channel" + (delayedAssertion + "Expected traffic to be relayed directly to us" + localNode True mailboxHandleReUse) + , testCase "Complex Filtering Rules" + (delayedAssertion + "Expected the relevant filters to accept our data" + localNode inputs (complexMailboxFiltering inputs)) + , testCase "Filter out unwanted messages" + (delayedAssertion + "Expected only even numbers to be sent delivered" + localNode True dropDuringFiltering) + ] + ] + where + inputs = ("hello", 10 :: Int, True) + +main :: IO () +main = testMain $ tests + diff --git a/tests/TestUtils.hs b/tests/TestUtils.hs new file mode 100644 --- /dev/null +++ b/tests/TestUtils.hs @@ -0,0 +1,234 @@ +{-# LANGUAGE DeriveDataTypeable #-} +{-# LANGUAGE TemplateHaskell #-} +{-# LANGUAGE DeriveGeneric #-} + +module TestUtils + ( TestResult + -- ping ! + , Ping(Ping) + , ping + , shouldBe + , shouldMatch + , shouldContain + , shouldNotContain + , shouldExitWith + , expectThat + -- test process utilities + , TestProcessControl + , startTestProcess + , runTestProcess + , testProcessGo + , testProcessStop + , testProcessReport + , delayedAssertion + , assertComplete + , waitForExit + -- logging + , Logger() + , newLogger + , putLogMsg + , stopLogger + -- runners + , mkNode + , tryRunProcess + , testMain + , stash + ) where + +#if ! MIN_VERSION_base(4,6,0) +import Prelude hiding (catch) +#endif +import Control.Concurrent + ( ThreadId + , myThreadId + , forkIO + ) +import Control.Concurrent.STM + ( TQueue + , newTQueueIO + , readTQueue + , writeTQueue + ) +import Control.Concurrent.MVar + ( MVar + , newEmptyMVar + , takeMVar + , putMVar + ) + +import Control.Distributed.Process +import Control.Distributed.Process.Node +import Control.Distributed.Process.Serializable() +import Control.Distributed.Process.Extras.Time +import Control.Distributed.Process.Extras.Timer +import Control.Distributed.Process.Extras.Internal.Types +import Control.Exception (SomeException) +import qualified Control.Exception as Exception +import Control.Monad (forever) +import Control.Monad.STM (atomically) +import Control.Rematch hiding (match) +import Control.Rematch.Run +import Test.HUnit (Assertion, assertFailure) +import Test.HUnit.Base (assertBool) +import Test.Framework (Test, defaultMain) +import Control.DeepSeq + +import Network.Transport.TCP +import qualified Network.Transport as NT + +import Data.Binary +import Data.Typeable +import GHC.Generics + +--expect :: a -> Matcher a -> Process () +--expect a m = liftIO $ Rematch.expect a m + +expectThat :: a -> Matcher a -> Process () +expectThat a matcher = case res of + MatchSuccess -> return () + (MatchFailure msg) -> liftIO $ assertFailure msg + where res = runMatch matcher a + +shouldBe :: a -> Matcher a -> Process () +shouldBe = expectThat + +shouldContain :: (Show a, Eq a) => [a] -> a -> Process () +shouldContain xs x = expectThat xs $ hasItem (equalTo x) + +shouldNotContain :: (Show a, Eq a) => [a] -> a -> Process () +shouldNotContain xs x = expectThat xs $ isNot (hasItem (equalTo x)) + +shouldMatch :: a -> Matcher a -> Process () +shouldMatch = expectThat + +shouldExitWith :: (Addressable a) => a -> DiedReason -> Process () +shouldExitWith a r = do + _ <- resolve a + d <- receiveWait [ match (\(ProcessMonitorNotification _ _ r') -> return r') ] + d `shouldBe` equalTo r + +waitForExit :: MVar ExitReason + -> Process (Maybe ExitReason) +waitForExit exitReason = do + -- we *might* end up blocked here, so ensure the test doesn't jam up! + self <- getSelfPid + tref <- killAfter (within 10 Seconds) self "testcast timed out" + tr <- liftIO $ takeMVar exitReason + cancelTimer tref + case tr of + ExitNormal -> return Nothing + other -> return $ Just other + +mkNode :: String -> IO LocalNode +mkNode port = do + Right (transport1, _) <- createTransportExposeInternals + "127.0.0.1" port defaultTCPParameters + newLocalNode transport1 initRemoteTable + +-- | Run the supplied @testProc@ using an @MVar@ to collect and assert +-- against its result. Uses the supplied @note@ if the assertion fails. +delayedAssertion :: (Eq a) => String -> LocalNode -> a -> + (TestResult a -> Process ()) -> Assertion +delayedAssertion note localNode expected testProc = do + result <- newEmptyMVar + _ <- forkProcess localNode $ testProc result + assertComplete note result expected + +-- | Takes the value of @mv@ (using @takeMVar@) and asserts that it matches @a@ +assertComplete :: (Eq a) => String -> MVar a -> a -> IO () +assertComplete msg mv a = do + b <- takeMVar mv + assertBool msg (a == b) + +-- synchronised logging + +data Logger = Logger { _tid :: ThreadId, msgs :: TQueue String } + +-- | Create a new Logger. +-- Logger uses a 'TQueue' to receive and process messages on a worker thread. +newLogger :: IO Logger +newLogger = do + tid <- liftIO $ myThreadId + q <- liftIO $ newTQueueIO + _ <- forkIO $ logger q + return $ Logger tid q + where logger q' = forever $ do + msg <- atomically $ readTQueue q' + putStrLn msg + +-- | Send a message to the Logger +putLogMsg :: Logger -> String -> Process () +putLogMsg logger msg = liftIO $ atomically $ writeTQueue (msgs logger) msg + +-- | Stop the worker thread for the given Logger +stopLogger :: Logger -> IO () +stopLogger = (flip Exception.throwTo) Exception.ThreadKilled . _tid + +-- | Given a @builder@ function, make and run a test suite on a single transport +testMain :: (NT.Transport -> IO [Test]) -> IO () +testMain builder = do + Right (transport, _) <- createTransportExposeInternals + "127.0.0.1" "10501" defaultTCPParameters + testData <- builder transport + defaultMain testData + +-- | Runs a /test process/ around the supplied @proc@, which is executed +-- whenever the outer process loop receives a 'Go' signal. +runTestProcess :: Process () -> Process () +runTestProcess proc = do + ctl <- expect + case ctl of + Stop -> return () + Go -> proc >> runTestProcess proc + Report p -> receiveWait [matchAny (\m -> forward m p)] >> runTestProcess proc + +-- | Starts a test process on the local node. +startTestProcess :: Process () -> Process ProcessId +startTestProcess proc = + spawnLocal $ do + getSelfPid >>= register "test-process" + runTestProcess proc + +-- | Control signals used to manage /test processes/ +data TestProcessControl = Stop | Go | Report ProcessId + deriving (Typeable, Generic) + +instance Binary TestProcessControl where + +-- | A mutable cell containing a test result. +type TestResult a = MVar a + +-- | Stashes a value in our 'TestResult' using @putMVar@ +stash :: TestResult a -> a -> Process () +stash mvar x = liftIO $ putMVar mvar x + +-- | Tell a /test process/ to stop (i.e., 'terminate') +testProcessStop :: ProcessId -> Process () +testProcessStop pid = send pid Stop + +-- | Tell a /test process/ to continue executing +testProcessGo :: ProcessId -> Process () +testProcessGo pid = send pid Go + +-- | A simple @Ping@ signal +data Ping = Ping + deriving (Typeable, Generic, Eq, Show) + +instance Binary Ping where +instance NFData Ping where + +ping :: ProcessId -> Process () +ping pid = send pid Ping + + +tryRunProcess :: LocalNode -> Process () -> IO () +tryRunProcess node p = do + tid <- liftIO myThreadId + runProcess node $ catch p (\e -> liftIO $ Exception.throwTo tid (e::SomeException)) + +-- | Tell a /test process/ to send a report (message) +-- back to the calling process +testProcessReport :: ProcessId -> Process () +testProcessReport pid = do + self <- getSelfPid + send pid $ Report self