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-client-server.cabal b/distributed-process-client-server.cabal new file mode 100644 --- /dev/null +++ b/distributed-process-client-server.cabal @@ -0,0 +1,129 @@ +name: distributed-process-client-server +version: 0.1.1 +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-client-server +Bug-Reports: http://github.com/haskell-distributed/distributed-process-client-server/issues +synopsis: The Cloud Haskell Application Platform +description: Modelled after Erlang OTP's gen_server, this framework provides similar + facilities for Cloud Haskell, grouping essential practices for client/server + development into a set of modules and standards designed to help you build + concurrent, distributed applications with relative ease. +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-client-server + +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-async >= 0.2.0 && < 0.3, + 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.ManagedProcess, + Control.Distributed.Process.ManagedProcess.Client, + Control.Distributed.Process.ManagedProcess.UnsafeClient, + Control.Distributed.Process.ManagedProcess.Server, + Control.Distributed.Process.ManagedProcess.Server.Priority, + Control.Distributed.Process.ManagedProcess.Server.Restricted + other-modules: + Control.Distributed.Process.ManagedProcess.Internal.Types, + Control.Distributed.Process.ManagedProcess.Internal.GenProcess + +test-suite ManagedProcessTests + type: exitcode-stdio-1.0 + x-uses-tf: true + build-depends: + base >= 4.4 && < 5, + ansi-terminal >= 0.5 && < 0.6, + containers, + distributed-process >= 0.5.2 && < 0.6, + distributed-process-extras >= 0.1.1 && < 0.2, + distributed-process-async >= 0.2.0 && < 0.3, + distributed-process-client-server, + distributed-process-tests >= 0.4.1 && < 0.5, + network-transport >= 0.4 && < 0.5, + mtl, + fingertree < 0.2, + network-transport-tcp >= 0.4 && < 0.5, + binary >= 0.6.3.0 && < 0.8, + deepseq >= 1.3.0.1 && < 1.4, + network >= 2.3 && < 2.6, + HUnit >= 1.2 && < 2, + stm >= 2.3 && < 2.5, + test-framework >= 0.6 && < 0.9, + test-framework-hunit, + 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 + extensions: CPP + main-is: TestManagedProcess.hs + +test-suite PrioritisedProcessTests + type: exitcode-stdio-1.0 + x-uses-tf: true + build-depends: + base >= 4.4 && < 5, + ansi-terminal >= 0.5 && < 0.6, + containers, + distributed-process >= 0.5.2 && < 0.6, + distributed-process-extras >= 0.1.1 && < 0.2, + distributed-process-async >= 0.2.0 && < 0.3, + distributed-process-client-server, + distributed-process-tests >= 0.4.1 && < 0.5, + network-transport >= 0.4 && < 0.5, + mtl, + fingertree < 0.2, + network-transport-tcp >= 0.4 && < 0.5, + binary >= 0.6.3.0 && < 0.8, + deepseq >= 1.3.0.1 && < 1.4, + network >= 2.3 && < 2.6, + HUnit >= 1.2 && < 2, + stm >= 2.3 && < 2.5, + test-framework >= 0.6 && < 0.9, + test-framework-hunit, + 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 + extensions: CPP + main-is: TestPrioritisedProcess.hs diff --git a/src/Control/Distributed/Process/ManagedProcess.hs b/src/Control/Distributed/Process/ManagedProcess.hs new file mode 100644 --- /dev/null +++ b/src/Control/Distributed/Process/ManagedProcess.hs @@ -0,0 +1,525 @@ +{-# LANGUAGE DeriveDataTypeable #-} +{-# LANGUAGE ExistentialQuantification #-} +{-# LANGUAGE ScopedTypeVariables #-} +{-# LANGUAGE TemplateHaskell #-} +{-# LANGUAGE RecordWildCards #-} + +----------------------------------------------------------------------------- +-- | +-- Module : Control.Distributed.Process.ManagedProcess +-- Copyright : (c) Tim Watson 2012 +-- License : BSD3 (see the file LICENSE) +-- +-- Maintainer : Tim Watson +-- Stability : experimental +-- Portability : non-portable (requires concurrency) +-- +-- This module provides a high(er) level API for building complex @Process@ +-- implementations by abstracting out the management of the process' mailbox, +-- reply/response handling, timeouts, process hiberation, error handling +-- and shutdown/stop procedures. It is modelled along similar lines to OTP's +-- gen_server API - . +-- +-- In particular, a /managed process/ will interoperate cleanly with the +-- supervisor API in distributed-process-supervision. +-- +-- [API Overview] +-- +-- Once started, a /managed process/ will consume messages from its mailbox and +-- pass them on to user defined /handlers/ based on the types received (mapped +-- to those accepted by the handlers) and optionally by also evaluating user +-- supplied predicates to determine which handler(s) should run. +-- Each handler returns a 'ProcessAction' which specifies how we should proceed. +-- If none of the handlers is able to process a message (because their types are +-- incompatible), then the 'unhandledMessagePolicy' will be applied. +-- +-- The 'ProcessAction' type defines the ways in which our process can respond +-- to its inputs, whether by continuing to read incoming messages, setting an +-- optional timeout, sleeping for a while or stopping. The optional timeout +-- behaves a little differently to the other process actions. If no messages +-- are received within the specified time span, a user defined 'timeoutHandler' +-- will be called in order to determine the next action. +-- +-- The 'ProcessDefinition' type also defines a @shutdownHandler@, +-- which is called whenever the process exits, whether because a callback has +-- returned 'stop' as the next action, or as the result of unhandled exit signal +-- or similar asynchronous exceptions thrown in (or to) the process itself. +-- +-- The other handlers are split into two groups: /apiHandlers/ and /infoHandlers/. +-- The former contains handlers for the 'cast' and 'call' protocols, whilst the +-- latter contains handlers that deal with input messages which are not sent +-- via these API calls (i.e., messages sent using bare 'send' or signals put +-- into the process mailbox by the node controller, such as +-- 'ProcessMonitorNotification' and the like). +-- +-- [The Cast/Call Protocol] +-- +-- Deliberate interactions with a /managed process/ usually fall into one of +-- two categories. A 'cast' interaction involves a client sending a message +-- asynchronously and the server handling this input. No reply is sent to +-- the client. On the other hand, a 'call' is a /remote procedure call/, +-- where the client sends a message and waits for a reply from the server. +-- +-- All expressions given to @apiHandlers@ have to conform to the /cast|call/ +-- protocol. The protocol (messaging) implementation is hidden from the user; +-- API functions for creating user defined @apiHandlers@ are given instead, +-- which take expressions (i.e., a function or lambda expression) and create the +-- appropriate @Dispatcher@ for handling the cast (or call). +-- +-- These cast/call protocols are for dealing with /expected/ inputs. They +-- will usually form the explicit public API for the process, and be exposed by +-- providing module level functions that defer to the cast/call API, giving +-- the author an opportunity to enforce the correct types. For +-- example: +-- +-- @ +-- {- Ask the server to add two numbers -} +-- add :: ProcessId -> Double -> Double -> Double +-- add pid x y = call pid (Add x y) +-- @ +-- +-- Note here that the return type from the call is /inferred/ and will not be +-- enforced by the type system. If the server sent a different type back in +-- the reply, then the caller might be blocked indefinitely! In fact, the +-- result of mis-matching the expected return type (in the client facing API) +-- with the actual type returned by the server is more severe in practise. +-- The underlying types that implement the /call/ protocol carry information +-- about the expected return type. If there is a mismatch between the input and +-- output types that the client API uses and those which the server declares it +-- can handle, then the message will be considered unroutable - no handler will +-- be executed against it and the unhandled message policy will be applied. You +-- should, therefore, take great care to align these types since the default +-- unhandled message policy is to terminate the server! That might seem pretty +-- extreme, but you can alter the unhandled message policy and/or use the +-- various overloaded versions of the call API in order to detect errors on the +-- server such as this. +-- +-- The cost of potential type mismatches between the client and server is the +-- main disadvantage of this looser coupling between them. This mechanism does +-- however, allow servers to handle a variety of messages without specifying the +-- entire protocol to be supported in excruciating detail. +-- +-- [Handling Unexpected/Info Messages] +-- +-- An explicit protocol for communicating with the process can be +-- configured using 'cast' and 'call', but it is not possible to prevent +-- other kinds of messages from being sent to the process mailbox. When +-- any message arrives for which there are no handlers able to process +-- its content, the 'UnhandledMessagePolicy' will be applied. Sometimes +-- it is desireable to process incoming messages which aren't part of the +-- protocol, rather than let the policy deal with them. This is particularly +-- true when incoming messages are important to the process, but their point +-- of origin is outside the author's control. Handling /signals/ such as +-- 'ProcessMonitorNotification' is a typical example of this: +-- +-- > handleInfo_ (\(ProcessMonitorNotification _ _ r) -> say $ show r >> continue_) +-- +-- [Handling Process State] +-- +-- The 'ProcessDefinition' is parameterised by the type of state it maintains. +-- A process that has no state will have the type @ProcessDefinition ()@ and can +-- be bootstrapped by evaluating 'statelessProcess'. +-- +-- All call/cast handlers come in two flavours, those which take the process +-- state as an input and those which do not. Handlers that ignore the process +-- state have to return a function that takes the state and returns the required +-- action. Versions of the various action generating functions ending in an +-- underscore are provided to simplify this: +-- +-- @ +-- statelessProcess { +-- apiHandlers = [ +-- handleCall_ (\\(n :: Int) -> return (n * 2)) +-- , handleCastIf_ (\\(c :: String, _ :: Delay) -> c == \"timeout\") +-- (\\(\"timeout\", (d :: Delay)) -> timeoutAfter_ d) +-- ] +-- , timeoutHandler = \\_ _ -> stop $ ExitOther \"timeout\" +-- } +-- @ +-- +-- [Avoiding Side Effects] +-- +-- If you wish to only write side-effect free code in your server definition, +-- then there is an explicit API for doing so. Instead of using the handlers +-- definition functions in this module, import the /pure/ server module instead, +-- which provides a StateT based monad for building referentially transparent +-- callbacks. +-- +-- See "Control.Distributed.Process.ManagedProcess.Server.Restricted" for +-- details and API documentation. +-- +-- [Handling Errors] +-- +-- Error handling appears in several contexts and process definitions can +-- hook into these with relative ease. Only process failures as a result of +-- asynchronous exceptions are supported by the API, which provides several +-- scopes for error handling. +-- +-- Catching exceptions inside handler functions is no different to ordinary +-- exception handling in monadic code. +-- +-- @ +-- handleCall (\\x y -> +-- catch (hereBeDragons x y) +-- (\\(e :: SmaugTheTerribleException) -> +-- return (Left (show e)))) +-- @ +-- +-- The caveats mentioned in "Control.Distributed.Process.Extras" about +-- exit signal handling obviously apply here as well. +-- +-- [Structured Exit Handling] +-- +-- Because "Control.Distributed.Process.ProcessExitException" is a ubiquitous +-- signalling mechanism in Cloud Haskell, it is treated unlike other +-- asynchronous exceptions. The 'ProcessDefinition' 'exitHandlers' field +-- accepts a list of handlers that, for a specific exit reason, can decide +-- how the process should respond. If none of these handlers matches the +-- type of @reason@ then the process will exit with @DiedException why@. In +-- addition, a private /exit handler/ is installed for exit signals where +-- @reason :: ExitReason@, which is a form of /exit signal/ used explicitly +-- by the supervision APIs. This behaviour, which cannot be overriden, is to +-- gracefully shut down the process, calling the @shutdownHandler@ as usual, +-- before stopping with @reason@ given as the final outcome. +-- +-- /Example: handling custom data is @ProcessExitException@/ +-- +-- > handleExit (\state from (sigExit :: SomeExitData) -> continue s) +-- +-- Under some circumstances, handling exit signals is perfectly legitimate. +-- Handling of /other/ forms of asynchronous exception (e.g., exceptions not +-- generated by an /exit/ signal) is not supported by this API. Cloud Haskell's +-- primitives for exception handling /will/ work normally in managed process +-- callbacks however. +-- +-- If any asynchronous exception goes unhandled, the process will immediately +-- exit without running the @shutdownHandler@. It is very important to note +-- that in Cloud Haskell, link failures generate asynchronous exceptions in +-- the target and these will NOT be caught by the API and will therefore +-- cause the process to exit /without running the termination handler/ +-- callback. If your termination handler is set up to do important work +-- (such as resource cleanup) then you should avoid linking you process +-- and use monitors instead. +-- +-- [Prioritised Mailboxes] +-- +-- Many processes need to prioritise certain classes of message over others, +-- so two subsets of the API are given to supporting those cases. +-- +-- A 'PrioritisedProcessDefintion' combines the usual 'ProcessDefintion' - +-- containing the cast/call API, error, termination and info handlers - with a +-- list of 'Priority' entries, which are used at runtime to prioritise the +-- server's inputs. Note that it is only messages which are prioritised; The +-- server's various handlers are still evaluated in insertion order. +-- +-- Prioritisation does not guarantee that a prioritised message/type will be +-- processed before other traffic - indeed doing so in a multi-threaded runtime +-- would be very hard - but in the absence of races between multiple processes, +-- if two messages are both present in the process' own mailbox, they will be +-- applied to the ProcessDefinition's handler's in priority order. This is +-- achieved by draining the real mailbox into a priority queue and processing +-- each message in turn. +-- +-- A prioritised process must be configured with a 'Priority' list to be of +-- any use. Creating a prioritised process without any priorities would be a +-- big waste of computational resources, and it is worth thinking carefully +-- about whether or not prioritisation is truly necessary in your design before +-- choosing to use it. +-- +-- Using a prioritised process is as simple as calling 'pserve' instead of +-- 'serve', and passing an initialised 'PrioritisedProcessDefinition'. +-- +-- [Control Channels] +-- +-- For advanced users and those requiring very low latency, a prioritised +-- process definition might not be suitable, since it performs considerable +-- work /behind the scenes/. There are also designs that need to segregate a +-- process' /control plane/ from other kinds of traffic it is expected to +-- receive. For such use cases, a /control channel/ may prove a better choice, +-- since typed channels are already prioritised during the mailbox scans that +-- the base @receiveWait@ and @receiveTimeout@ primitives from +-- distribute-process provides. +-- +-- In order to utilise a /control channel/ in a server, it must be passed to the +-- corresponding 'handleControlChan' function (or its stateless variant). The +-- control channel is created by evaluating 'newControlChan', in the same way +-- that we create regular typed channels. +-- +-- In order for clients to communicate with a server via its control channel +-- however, they must pass a handle to a 'ControlPort', which can be obtained by +-- evaluating 'channelControlPort' on the 'ControlChannel'. A 'ControlPort' is +-- @Serializable@, so they can alternatively be sent to other processes. +-- +-- /Control channel/ traffic will only be prioritised over other traffic if the +-- handlers using it are present before others (e.g., @handleInfo, handleCast@, +-- etc) in the process definition. It is not possible to combine prioritised +-- processes with /control channels/. Attempting to do so will satisfy the +-- compiler, but crash with a runtime error once you attempt to evaluate the +-- prioritised server loop (i.e., 'pserve'). +-- +-- Since the primary purpose of control channels is to simplify and optimise +-- client-server communication over a single channel, this module provides an +-- alternate server loop in the form of 'chanServe'. Instead of passing an +-- initialised 'ProcessDefinition', this API takes an expression from a +-- 'ControlChannel' to 'ProcessDefinition', operating in the 'Process' monad. +-- Providing the opaque reference in this fashion is useful, since the type of +-- messages the control channel carries will not correlate directly to the +-- inter-process traffic we use internally. +-- +-- Although control channels are intended for use as a single control plane +-- (via 'chanServe'), it /is/ possible to use them as a more strictly typed +-- communications backbone, since they do enforce absolute type safety in client +-- code, being bound to a particular type on creation. For rpc (i.e., 'call') +-- interaction however, it is not possible to have the server reply to a control +-- channel, since they're a /one way pipe/. It is possible to alleviate this +-- situation by passing a request type than contains a typed channel bound to +-- the expected reply type, enabling client and server to match on both the input +-- and output types as specifically as possible. Note that this still does not +-- guarantee an agreement on types between all parties at runtime however. +-- +-- An example of how to do this follows: +-- +-- > data Request = Request String (SendPort String) +-- > deriving (Typeable, Generic) +-- > instance Binary Request where +-- > +-- > -- note that our initial caller needs an mvar to obtain the control port... +-- > echoServer :: MVar (ControlPort Request) -> Process () +-- > echoServer mv = do +-- > cc <- newControlChan :: Process (ControlChannel Request) +-- > liftIO $ putMVar mv $ channelControlPort cc +-- > let s = statelessProcess { +-- > apiHandlers = [ +-- > handleControlChan_ cc (\(Request m sp) -> sendChan sp m >> continue_) +-- > ] +-- > } +-- > serve () (statelessInit Infinity) s +-- > +-- > echoClient :: String -> ControlPort Request -> Process String +-- > echoClient str cp = do +-- > (sp, rp) <- newChan +-- > sendControlMessage cp $ Request str sp +-- > receiveChan rp +-- +-- [Performance Considerations] +-- +-- The various server loops are fairly optimised, but there /is/ a definite +-- cost associated with scanning the mailbox to match on protocol messages, +-- plus additional costs in space and time due to mapping over all available +-- /info handlers/ for non-protocol (i.e., neither /call/ nor /cast/) messages. +-- These are exacerbated significantly when using prioritisation, whilst using +-- a single control channel is very fast and carries little overhead. +-- +-- From the client perspective, it's important to remember that the /call/ +-- protocol will wait for a reply in most cases, triggering a full O(n) scan of +-- the caller's mailbox. If the mailbox is extremely full and calls are +-- regularly made, this may have a significant impact on the caller. The +-- @callChan@ family of client API functions can alleviate this, by using (and +-- matching on) a private typed channel instead, but the server must be written +-- to accomodate this. Similar gains can be had using a /control channel/ and +-- providing a typed reply channel in the request data, however the 'call' +-- mechanism does not support this notion, so not only are we unable +-- to use the various /reply/ functions, client code should also consider +-- monitoring the server's pid and handling server failures whilst waiting on +-- +----------------------------------------------------------------------------- + +module Control.Distributed.Process.ManagedProcess + ( -- * Starting/Running server processes + InitResult(..) + , InitHandler + , serve + , pserve + , chanServe + , runProcess + , prioritised + -- * Client interactions + , module Control.Distributed.Process.ManagedProcess.Client + -- * Defining server processes + , ProcessDefinition(..) + , PrioritisedProcessDefinition(..) + , RecvTimeoutPolicy(..) + , Priority(..) + , DispatchPriority() + , Dispatcher() + , DeferredDispatcher() + , ShutdownHandler + , TimeoutHandler + , ProcessAction(..) + , ProcessReply + , Condition + , CallHandler + , CastHandler + , UnhandledMessagePolicy(..) + , CallRef + , ControlChannel() + , ControlPort() + , defaultProcess + , defaultProcessWithPriorities + , statelessProcess + , statelessInit + -- * Server side callbacks + , handleCall + , handleCallIf + , handleCallFrom + , handleCallFromIf + , handleCast + , handleCastIf + , handleInfo + , handleRaw + , handleRpcChan + , handleRpcChanIf + , action + , handleDispatch + , handleExit + -- * Stateless callbacks + , handleCall_ + , handleCallFrom_ + , handleCallIf_ + , handleCallFromIf_ + , handleCast_ + , handleCastIf_ + , handleRpcChan_ + , handleRpcChanIf_ + -- * Control channels + , newControlChan + , channelControlPort + , handleControlChan + , handleControlChan_ + -- * Prioritised mailboxes + , module Control.Distributed.Process.ManagedProcess.Server.Priority + -- * Constructing handler results + , condition + , state + , input + , reply + , replyWith + , noReply + , noReply_ + , haltNoReply_ + , continue + , continue_ + , timeoutAfter + , timeoutAfter_ + , hibernate + , hibernate_ + , stop + , stopWith + , stop_ + , replyTo + , replyChan + ) where + +import Control.Distributed.Process hiding (call, Message) +import Control.Distributed.Process.ManagedProcess.Client +import Control.Distributed.Process.ManagedProcess.Server +import Control.Distributed.Process.ManagedProcess.Server.Priority +import Control.Distributed.Process.ManagedProcess.Internal.GenProcess +import Control.Distributed.Process.ManagedProcess.Internal.Types +import Control.Distributed.Process.Extras (ExitReason(..)) +import Control.Distributed.Process.Extras.Time +import Control.Distributed.Process.Serializable +import Prelude hiding (init) + +-- TODO: automatic registration + +-- | Starts the /message handling loop/ for a managed process configured with +-- the supplied process definition, after calling the init handler with its +-- initial arguments. Note that this function does not return until the server +-- exits. +serve :: a + -> InitHandler a s + -> ProcessDefinition s + -> Process () +serve argv init def = runProcess (recvLoop def) argv init + +-- | Starts the /message handling loop/ for a prioritised managed process, +-- configured with the supplied process definition, after calling the init +-- handler with its initial arguments. Note that this function does not return +-- until the server exits. +pserve :: a + -> InitHandler a s + -> PrioritisedProcessDefinition s + -> Process () +pserve argv init def = runProcess (precvLoop def) argv init + +-- | Starts the /message handling loop/ for a managed process, configured with +-- a typed /control channel/. The caller supplied expression is evaluated with +-- an opaque reference to the channel, which must be passed when calling +-- @handleControlChan@. The meaning and behaviour of the init handler and +-- initial arguments are the same as those given to 'serve'. Note that this +-- function does not return until the server exits. +-- +chanServe :: (Serializable b) + => a + -> InitHandler a s + -> (ControlChannel b -> Process (ProcessDefinition s)) + -> Process () +chanServe argv init mkDef = do + pDef <- mkDef . ControlChannel =<< newChan + runProcess (recvLoop pDef) argv init + +-- | Wraps any /process loop/ and ensures that it adheres to the +-- managed process start/stop semantics, i.e., evaluating the +-- @InitHandler@ with an initial state and delay will either +-- @die@ due to @InitStop@, exit silently (due to @InitIgnore@) +-- or evaluate the process' @loop@. The supplied @loop@ must evaluate +-- to @ExitNormal@, otherwise the calling processing will @die@ with +-- whatever @ExitReason@ is given. +-- +runProcess :: (s -> Delay -> Process ExitReason) + -> a + -> InitHandler a s + -> Process () +runProcess loop args init = do + ir <- init args + case ir of + InitOk s d -> loop s d >>= checkExitType + InitStop s -> die $ ExitOther s + InitIgnore -> return () + where + checkExitType :: ExitReason -> Process () + checkExitType ExitNormal = return () + checkExitType other = die other + +-- | A default 'ProcessDefinition', with no api, info or exit handler. +-- The default 'timeoutHandler' simply continues, the 'shutdownHandler' +-- is a no-op and the 'unhandledMessagePolicy' is @Terminate@. +defaultProcess :: ProcessDefinition s +defaultProcess = ProcessDefinition { + apiHandlers = [] + , infoHandlers = [] + , exitHandlers = [] + , timeoutHandler = \s _ -> continue s + , shutdownHandler = \_ _ -> return () + , unhandledMessagePolicy = Terminate + } :: ProcessDefinition s + +-- | Turns a standard 'ProcessDefinition' into a 'PrioritisedProcessDefinition', +-- by virtue of the supplied list of 'DispatchPriority' expressions. +-- +prioritised :: ProcessDefinition s + -> [DispatchPriority s] + -> PrioritisedProcessDefinition s +prioritised def ps = PrioritisedProcessDefinition def ps defaultRecvTimeoutPolicy + +-- | Sets the default 'recvTimeoutPolicy', which gives up after 10k reads. +defaultRecvTimeoutPolicy :: RecvTimeoutPolicy +defaultRecvTimeoutPolicy = RecvCounter 10000 + +-- | Creates a default 'PrioritisedProcessDefinition' from a list of +-- 'DispatchPriority'. See 'defaultProcess' for the underlying definition. +defaultProcessWithPriorities :: [DispatchPriority s] -> PrioritisedProcessDefinition s +defaultProcessWithPriorities dps = prioritised defaultProcess dps + +-- | A basic, stateless 'ProcessDefinition'. See 'defaultProcess' for the +-- default field values. +statelessProcess :: ProcessDefinition () +statelessProcess = defaultProcess :: ProcessDefinition () + +-- | A default, state /unaware/ 'InitHandler' that can be used with +-- 'statelessProcess'. This simply returns @InitOk@ with the empty +-- state (i.e., unit) and the given 'Delay'. +statelessInit :: Delay -> InitHandler () () +statelessInit d () = return $ InitOk () d + diff --git a/src/Control/Distributed/Process/ManagedProcess/Client.hs b/src/Control/Distributed/Process/ManagedProcess/Client.hs new file mode 100644 --- /dev/null +++ b/src/Control/Distributed/Process/ManagedProcess/Client.hs @@ -0,0 +1,164 @@ +{-# LANGUAGE ExistentialQuantification #-} +{-# LANGUAGE ScopedTypeVariables #-} + +----------------------------------------------------------------------------- +-- | +-- Module : Control.Distributed.Process.ManagedProcess.Client +-- Copyright : (c) Tim Watson 2012 - 2013 +-- License : BSD3 (see the file LICENSE) +-- +-- Maintainer : Tim Watson +-- Stability : experimental +-- Portability : non-portable (requires concurrency) +-- +-- The Client Portion of the /Managed Process/ API. +----------------------------------------------------------------------------- + +module Control.Distributed.Process.ManagedProcess.Client + ( -- * API for client interactions with the process + sendControlMessage + , shutdown + , call + , safeCall + , tryCall + , callTimeout + , flushPendingCalls + , callAsync + , cast + , callChan + , syncCallChan + , syncSafeCallChan + ) where + +import Control.Distributed.Process hiding (call) +import Control.Distributed.Process.Serializable +import Control.Distributed.Process.Async hiding (check) +import Control.Distributed.Process.ManagedProcess.Internal.Types +import qualified Control.Distributed.Process.ManagedProcess.Internal.Types as T +import Control.Distributed.Process.Extras hiding (monitor, sendChan) +import Control.Distributed.Process.Extras.Time +import Data.Maybe (fromJust) + +import Prelude hiding (init) + +-- | Send a control message over a 'ControlPort'. +-- +sendControlMessage :: Serializable m => ControlPort m -> m -> Process () +sendControlMessage cp m = sendChan (unPort cp) (CastMessage m) + +-- | Send a signal instructing the process to terminate. The /receive loop/ which +-- manages the process mailbox will prioritise @Shutdown@ signals higher than +-- any other incoming messages, but the server might be busy (i.e., still in the +-- process of excuting a handler) at the time of sending however, so the caller +-- should not make any assumptions about the timeliness with which the shutdown +-- signal will be handled. If responsiveness is important, a better approach +-- might be to send an /exit signal/ with 'Shutdown' as the reason. An exit +-- signal will interrupt any operation currently underway and force the running +-- process to clean up and terminate. +shutdown :: ProcessId -> Process () +shutdown pid = cast pid Shutdown + +-- | Make a synchronous call - will block until a reply is received. +-- The calling process will exit with 'ExitReason' if the calls fails. +call :: forall s a b . (Addressable s, Serializable a, Serializable b) + => s -> a -> Process b +call sid msg = initCall sid msg >>= waitResponse Nothing >>= decodeResult + where decodeResult (Just (Right r)) = return r + decodeResult (Just (Left err)) = die err + decodeResult Nothing {- the impossible happened -} = terminate + +-- | Safe version of 'call' that returns information about the error +-- if the operation fails. If an error occurs then the explanation will be +-- will be stashed away as @(ExitOther String)@. +safeCall :: forall s a b . (Addressable s, Serializable a, Serializable b) + => s -> a -> Process (Either ExitReason b) +safeCall s m = initCall s m >>= waitResponse Nothing >>= return . fromJust + +-- | Version of 'safeCall' that returns 'Nothing' if the operation fails. If +-- you need information about *why* a call has failed then you should use +-- 'safeCall' or combine @catchExit@ and @call@ instead. +tryCall :: forall s a b . (Addressable s, Serializable a, Serializable b) + => s -> a -> Process (Maybe b) +tryCall s m = initCall s m >>= waitResponse Nothing >>= decodeResult + where decodeResult (Just (Right r)) = return $ Just r + decodeResult _ = return Nothing + +-- | Make a synchronous call, but timeout and return @Nothing@ if a reply +-- is not received within the specified time interval. +-- +-- If the result of the call is a failure (or the call was cancelled) then +-- the calling process will exit, with the 'ExitReason' given as the reason. +-- If the call times out however, the semantics on the server side are +-- undefined, i.e., the server may or may not successfully process the +-- request and may (or may not) send a response at a later time. From the +-- callers perspective, this is somewhat troublesome, since the call result +-- cannot be decoded directly. In this case, the 'flushPendingCalls' API /may/ +-- be used to attempt to receive the message later on, however this makes +-- /no attempt whatsoever/ to guarantee /which/ call response will in fact +-- be returned to the caller. In those semantics are unsuited to your +-- application, you might choose to @exit@ or @die@ in case of a timeout, +-- or alternatively, use the 'callAsync' API and associated @waitTimeout@ +-- function (in the /Async API/), which takes a re-usable handle on which +-- to wait (with timeouts) multiple times. +-- +callTimeout :: forall s a b . (Addressable s, Serializable a, Serializable b) + => s -> a -> TimeInterval -> Process (Maybe b) +callTimeout s m d = initCall s m >>= waitResponse (Just d) >>= decodeResult + where decodeResult :: (Serializable b) + => Maybe (Either ExitReason b) + -> Process (Maybe b) + decodeResult Nothing = return Nothing + decodeResult (Just (Right result)) = return $ Just result + decodeResult (Just (Left reason)) = die reason + +flushPendingCalls :: forall b . (Serializable b) + => TimeInterval + -> (b -> Process b) + -> Process (Maybe b) +flushPendingCalls d proc = do + receiveTimeout (asTimeout d) [ + match (\(CallResponse (m :: b) _) -> proc m) + ] + +-- | Invokes 'call' /out of band/, and returns an /async handle/. +-- +callAsync :: forall s a b . (Addressable s, Serializable a, Serializable b) + => s -> a -> Process (Async b) +callAsync server msg = async $ task $ call server msg + +-- | Sends a /cast/ message to the server identified by @server@. The server +-- will not send a response. Like Cloud Haskell's 'send' primitive, cast is +-- fully asynchronous and /never fails/ - therefore 'cast'ing to a non-existent +-- (e.g., dead) server process will not generate an error. +-- +cast :: forall a m . (Addressable a, Serializable m) + => a -> m -> Process () +cast server msg = sendTo server ((CastMessage msg) :: T.Message m ()) + +-- | Sends a /channel/ message to the server and returns a @ReceivePort@ on +-- which the reponse can be delivered, if the server so chooses (i.e., the +-- might ignore the request or crash). +callChan :: forall s a b . (Addressable s, Serializable a, Serializable b) + => s -> a -> Process (ReceivePort b) +callChan server msg = do + (sp, rp) <- newChan + sendTo server ((ChanMessage msg sp) :: T.Message a b) + return rp + +-- | A synchronous version of 'callChan'. +syncCallChan :: forall s a b . (Addressable s, Serializable a, Serializable b) + => s -> a -> Process b +syncCallChan server msg = do + r <- syncSafeCallChan server msg + case r of + Left e -> die e + Right r' -> return r' + +-- | A safe version of 'syncCallChan', which returns @Left ExitReason@ if the +-- call fails. +syncSafeCallChan :: forall s a b . (Addressable s, Serializable a, Serializable b) + => s -> a -> Process (Either ExitReason b) +syncSafeCallChan server msg = do + rp <- callChan server msg + awaitResponse server [ matchChan rp (return . Right) ] + diff --git a/src/Control/Distributed/Process/ManagedProcess/Internal/GenProcess.hs b/src/Control/Distributed/Process/ManagedProcess/Internal/GenProcess.hs new file mode 100644 --- /dev/null +++ b/src/Control/Distributed/Process/ManagedProcess/Internal/GenProcess.hs @@ -0,0 +1,328 @@ +{-# LANGUAGE ExistentialQuantification #-} +{-# LANGUAGE ScopedTypeVariables #-} +{-# LANGUAGE ViewPatterns #-} +{-# LANGUAGE PatternGuards #-} + +-- | This is the @Process@ implementation of a /managed process/ +module Control.Distributed.Process.ManagedProcess.Internal.GenProcess + (recvLoop, precvLoop) where + +import Control.Applicative ((<$>)) +import Control.Concurrent (threadDelay) +import Control.Concurrent.STM hiding (check) +import Control.Distributed.Process hiding (call, Message) +import qualified Control.Distributed.Process as P (Message) +import Control.Distributed.Process.ManagedProcess.Server +import Control.Distributed.Process.ManagedProcess.Internal.Types +import Control.Distributed.Process.Extras.Internal.Queue.PriorityQ + ( PriorityQ + , enqueue + , dequeue + ) +import qualified Control.Distributed.Process.Extras.Internal.Queue.PriorityQ as PriorityQ + ( empty + ) +import Control.Distributed.Process.Extras + ( ExitReason(..) + , Shutdown(..) + ) +import qualified Control.Distributed.Process.Extras.SystemLog as Log +import Control.Distributed.Process.Extras.Time +import Control.Distributed.Process.Extras.Timer + ( cancelTimer + , runAfter + , TimerRef + ) +import Control.Monad (void) +import Prelude hiding (init) + +-------------------------------------------------------------------------------- +-- Priority Mailbox Handling -- +-------------------------------------------------------------------------------- + +type Queue = PriorityQ Int P.Message +type TimeoutSpec = (Delay, Maybe (TimerRef, (STM ()))) +data TimeoutAction s = Stop s ExitReason | Go Delay s + +precvLoop :: PrioritisedProcessDefinition s -> s -> Delay -> Process ExitReason +precvLoop ppDef pState recvDelay = do + void $ verify $ processDef ppDef + tref <- startTimer recvDelay + recvQueue ppDef pState tref $ PriorityQ.empty + where + verify pDef = mapM_ disallowCC $ apiHandlers pDef + + disallowCC (DispatchCC _ _) = die $ ExitOther "IllegalControlChannel" + disallowCC _ = return () + +recvQueue :: PrioritisedProcessDefinition s + -> s + -> TimeoutSpec + -> Queue + -> Process ExitReason +recvQueue p s t q = + let pDef = processDef p + ps = priorities p + in do (ac, d, q') <- catchExit (processNext pDef ps s t q) + (\_ (r :: ExitReason) -> + return (ProcessStop r, Infinity, q)) + nextAction ac d q' + where + nextAction ac d q' + | ProcessContinue s' <- ac = recvQueueAux p (priorities p) s' d q' + | ProcessTimeout t' s' <- ac = recvQueueAux p (priorities p) s' t' q' + | ProcessHibernate d' s' <- ac = block d' >> recvQueueAux p (priorities p) s' d q' + | ProcessStop r <- ac = (shutdownHandler $ processDef p) s r >> return r + | ProcessStopping s' r <- ac = (shutdownHandler $ processDef p) s' r >> return r + | otherwise {- compiler foo -} = die "IllegalState" + + recvQueueAux ppDef prioritizers pState delay queue = + let ex = (trapExit:(exitHandlers $ processDef ppDef)) + eh = map (\d' -> (dispatchExit d') pState) ex + in (do t' <- startTimer delay + mq <- drainMessageQueue pState prioritizers queue + recvQueue ppDef pState t' mq) + `catchExit` + (\pid (reason :: ExitReason) -> do + let pd = processDef ppDef + let ps = pState + let pq = queue + let em = unsafeWrapMessage reason + (a, d, q') <- findExitHandlerOrStop pd ps pq eh pid em + nextAction a d q') + + findExitHandlerOrStop :: ProcessDefinition s + -> s + -> Queue + -> [ProcessId -> P.Message -> Process (Maybe (ProcessAction s))] + -> ProcessId + -> P.Message + -> Process (ProcessAction s, Delay, Queue) + findExitHandlerOrStop _ _ pq [] _ er = do + mEr <- unwrapMessage er :: Process (Maybe ExitReason) + case mEr of + Nothing -> die "InvalidExitHandler" -- TODO: better error message? + Just er' -> return (ProcessStop er', Infinity, pq) + findExitHandlerOrStop pd ps pq (eh:ehs) pid er = do + mAct <- eh pid er + case mAct of + Nothing -> findExitHandlerOrStop pd ps pq ehs pid er + Just pa -> return (pa, Infinity, pq) + + processNext def ps' pState tSpec queue = + let ex = (trapExit:(exitHandlers def)) + h = timeoutHandler def in do + -- as a side effect, this check will cancel the timer + timedOut <- checkTimer pState tSpec h + case timedOut of + Stop s' r -> return $ (ProcessStopping s' r, (fst tSpec), queue) + Go t' s' -> do + -- checkTimer could've run our timeoutHandler, which changes "s" + case (dequeue queue) of + Nothing -> do + -- if the internal queue is empty, we fall back to reading the + -- actual mailbox, however if /that/ times out, then we need + -- to let the timeout handler kick in again and make a decision + drainOrTimeout s' t' queue ps' h + Just (m', q') -> do + act <- catchesExit (processApply def s' m') + (map (\d' -> (dispatchExit d') s') ex) + return (act, t', q') + + processApply def pState msg = + let pol = unhandledMessagePolicy def + apiMatchers = map (dynHandleMessage pol pState) (apiHandlers def) + infoMatchers = map (dynHandleMessage pol pState) (infoHandlers def) + shutdown' = dynHandleMessage pol pState shutdownHandler' + ms' = (shutdown':apiMatchers) ++ infoMatchers + in processApplyAux ms' pol pState msg + + processApplyAux [] p' s' m' = applyPolicy p' s' m' + processApplyAux (h:hs) p' s' m' = do + attempt <- h m' + case attempt of + Nothing -> processApplyAux hs p' s' m' + Just act -> return act + + drainOrTimeout pState delay queue ps' h = do + let matches = [ matchMessage return ] + recv = case delay of + Infinity -> receiveWait matches >>= return . Just + NoDelay -> receiveTimeout 0 matches + Delay i -> receiveTimeout (asTimeout i) matches in do + r <- recv + case r of + Nothing -> h pState delay >>= \act -> return $ (act, delay, queue) + Just m -> do + queue' <- enqueueMessage pState ps' m queue + -- Returning @ProcessContinue@ simply causes the main loop to go + -- into 'recvQueueAux', which ends up in 'drainMessageQueue'. + -- In other words, we continue draining the /real/ mailbox. + return $ (ProcessContinue pState, delay, queue') + +drainMessageQueue :: s -> [DispatchPriority s] -> Queue -> Process Queue +drainMessageQueue pState priorities' queue = do + m <- receiveTimeout 0 [ matchMessage return ] + case m of + Nothing -> return queue + Just m' -> do + queue' <- enqueueMessage pState priorities' m' queue + drainMessageQueue pState priorities' queue' + +enqueueMessage :: s + -> [DispatchPriority s] + -> P.Message + -> Queue + -> Process Queue +enqueueMessage _ [] m' q = return $ enqueue (-1 :: Int) m' q +enqueueMessage s (p:ps) m' q = let checkPrio = prioritise p s in do + checkPrio m' >>= maybeEnqueue s m' q ps + where + maybeEnqueue :: s + -> P.Message + -> Queue + -> [DispatchPriority s] + -> Maybe (Int, P.Message) + -> Process Queue + maybeEnqueue s' msg q' ps' Nothing = enqueueMessage s' ps' msg q' + maybeEnqueue _ _ q' _ (Just (i, m)) = return $ enqueue (i * (-1 :: Int)) m q' + +-------------------------------------------------------------------------------- +-- Ordinary/Blocking Mailbox Handling -- +-------------------------------------------------------------------------------- + +recvLoop :: ProcessDefinition s -> s -> Delay -> Process ExitReason +recvLoop pDef pState recvDelay = + let p = unhandledMessagePolicy pDef + handleTimeout = timeoutHandler pDef + handleStop = shutdownHandler pDef + shutdown' = matchDispatch p pState shutdownHandler' + matchers = map (matchDispatch p pState) (apiHandlers pDef) + ex' = (trapExit:(exitHandlers pDef)) + ms' = (shutdown':matchers) ++ matchAux p pState (infoHandlers pDef) + in do + ac <- catchesExit (processReceive ms' handleTimeout pState recvDelay) + (map (\d' -> (dispatchExit d') pState) ex') + case ac of + (ProcessContinue s') -> recvLoop pDef s' recvDelay + (ProcessTimeout t' s') -> recvLoop pDef s' t' + (ProcessHibernate d' s') -> block d' >> recvLoop pDef s' recvDelay + (ProcessStop r) -> handleStop pState r >> return (r :: ExitReason) + (ProcessStopping s' r) -> handleStop s' r >> return (r :: ExitReason) + where + matchAux :: UnhandledMessagePolicy + -> s + -> [DeferredDispatcher s] + -> [Match (ProcessAction s)] + matchAux p ps ds = [matchAny (auxHandler (applyPolicy p ps) ps ds)] + + auxHandler :: (P.Message -> Process (ProcessAction s)) + -> s + -> [DeferredDispatcher s] + -> P.Message + -> Process (ProcessAction s) + auxHandler policy _ [] msg = policy msg + auxHandler policy st (d:ds :: [DeferredDispatcher s]) msg + | length ds > 0 = let dh = dispatchInfo d in do + -- NB: we *do not* want to terminate/dead-letter messages until + -- we've exhausted all the possible info handlers + m <- dh st msg + case m of + Nothing -> auxHandler policy st ds msg + Just act -> return act + -- but here we *do* let the policy kick in + | otherwise = let dh = dispatchInfo d in do + m <- dh st msg + case m of + Nothing -> policy msg + Just act -> return act + + processReceive :: [Match (ProcessAction s)] + -> TimeoutHandler s + -> s + -> Delay + -> Process (ProcessAction s) + processReceive ms handleTimeout st d = do + next <- recv ms d + case next of + Nothing -> handleTimeout st d + Just pa -> return pa + + recv :: [Match (ProcessAction s)] + -> Delay + -> Process (Maybe (ProcessAction s)) + recv matches d' = + case d' of + Infinity -> receiveWait matches >>= return . Just + NoDelay -> receiveTimeout 0 matches + Delay t' -> receiveTimeout (asTimeout t') matches + +-------------------------------------------------------------------------------- +-- Simulated Receive Timeouts -- +-------------------------------------------------------------------------------- + +startTimer :: Delay -> Process TimeoutSpec +startTimer d + | Delay t <- d = do sig <- liftIO $ newEmptyTMVarIO + tref <- runAfter t $ liftIO $ atomically $ putTMVar sig () + return (d, Just (tref, (readTMVar sig))) + | otherwise = return (d, Nothing) + +checkTimer :: s + -> TimeoutSpec + -> TimeoutHandler s + -> Process (TimeoutAction s) +checkTimer pState spec handler = let delay = fst spec in do + timedOut <- pollTimer spec -- this will cancel the timer + case timedOut of + False -> go spec pState + True -> do + act <- handler pState delay + case act of + ProcessTimeout t' s' -> return $ Go t' s' + ProcessStop r -> return $ Stop pState r + ProcessStopping s' r -> return $ Stop s' r + ProcessHibernate d' s' -> block d' >> go spec s' + ProcessContinue s' -> go spec s' + where + go d s = return $ Go (fst d) s + +pollTimer :: TimeoutSpec -> Process Bool +pollTimer (_, Nothing ) = return False +pollTimer (_, Just (tref, sig)) = do + cancelTimer tref -- cancelling a dead/completed timer is a no-op + gotSignal <- liftIO $ atomically $ pollSTM sig + return $ maybe False (const True) gotSignal + where + pollSTM :: (STM ()) -> STM (Maybe ()) + pollSTM sig' = (Just <$> sig') `orElse` return Nothing + +-------------------------------------------------------------------------------- +-- Utilities -- +-------------------------------------------------------------------------------- + +-- an explicit 'cast' giving 'Shutdown' will stop the server gracefully +shutdownHandler' :: Dispatcher s +shutdownHandler' = handleCast (\_ Shutdown -> stop $ ExitNormal) + +-- @(ProcessExitException from ExitShutdown)@ will stop the server gracefully +trapExit :: ExitSignalDispatcher s +trapExit = handleExit (\_ _ (r :: ExitReason) -> stop r) + +block :: TimeInterval -> Process () +block i = liftIO $ threadDelay (asTimeout i) + +applyPolicy :: UnhandledMessagePolicy + -> s + -> P.Message + -> Process (ProcessAction s) +applyPolicy p s m = + case p of + Terminate -> stop $ ExitOther "UnhandledInput" + DeadLetter pid -> forward m pid >> continue s + Drop -> continue s + Log -> logIt >> continue s + where + logIt = + Log.report Log.info Log.logChannel $ "Unhandled Gen Input Message: " ++ (show m) + diff --git a/src/Control/Distributed/Process/ManagedProcess/Internal/Types.hs b/src/Control/Distributed/Process/ManagedProcess/Internal/Types.hs new file mode 100644 --- /dev/null +++ b/src/Control/Distributed/Process/ManagedProcess/Internal/Types.hs @@ -0,0 +1,415 @@ +{-# LANGUAGE DeriveDataTypeable #-} +{-# LANGUAGE ExistentialQuantification #-} +{-# LANGUAGE StandaloneDeriving #-} +{-# LANGUAGE TemplateHaskell #-} +{-# LANGUAGE DeriveGeneric #-} +{-# LANGUAGE StandaloneDeriving #-} +{-# LANGUAGE ScopedTypeVariables #-} +{-# LANGUAGE GeneralizedNewtypeDeriving #-} + +-- | Types used throughout the ManagedProcess framework +module Control.Distributed.Process.ManagedProcess.Internal.Types + ( -- * Exported data types + InitResult(..) + , Condition(..) + , ProcessAction(..) + , ProcessReply(..) + , CallHandler + , CastHandler + , DeferredCallHandler + , StatelessCallHandler + , InfoHandler + , ChannelHandler + , StatelessChannelHandler + , InitHandler + , ShutdownHandler + , TimeoutHandler + , UnhandledMessagePolicy(..) + , ProcessDefinition(..) + , Priority(..) + , DispatchPriority(..) + , PrioritisedProcessDefinition(..) + , RecvTimeoutPolicy(..) + , ControlChannel(..) + , newControlChan + , ControlPort(..) + , channelControlPort + , Dispatcher(..) + , DeferredDispatcher(..) + , ExitSignalDispatcher(..) + , MessageMatcher(..) + , DynMessageHandler(..) + , Message(..) + , CallResponse(..) + , CallId + , CallRef(..) + , makeRef + , initCall + , unsafeInitCall + , waitResponse + ) where + +import Control.Distributed.Process hiding (Message) +import qualified Control.Distributed.Process as P (Message) +import Control.Distributed.Process.Serializable +import Control.Distributed.Process.Extras + ( Recipient(..) + , ExitReason(..) + , Addressable + , Resolvable(..) + , Routable(..) + , NFSerializable + ) +import Control.Distributed.Process.Extras.Internal.Types + ( resolveOrDie + ) +import Control.Distributed.Process.Extras.Time +import Control.DeepSeq (NFData) +import Data.Binary hiding (decode) +import Data.Typeable (Typeable) + +import Prelude hiding (init) + +import GHC.Generics + +-------------------------------------------------------------------------------- +-- API -- +-------------------------------------------------------------------------------- + +type CallId = MonitorRef + +newtype CallRef a = CallRef { unCaller :: (Recipient, CallId) } + deriving (Eq, Show, Typeable, Generic) +instance Serializable a => Binary (CallRef a) where +instance NFData a => NFData (CallRef a) where + +makeRef :: forall a . (Serializable a) => Recipient -> CallId -> CallRef a +makeRef r c = CallRef (r, c) + +instance Resolvable (CallRef a) where + resolve (CallRef (r, _)) = resolve r + +instance Routable (CallRef a) where + sendTo (CallRef (client, tag)) msg = sendTo client (CallResponse msg tag) + unsafeSendTo (CallRef (c, tag)) msg = unsafeSendTo c (CallResponse msg tag) + +data Message a b = + CastMessage a + | CallMessage a (CallRef b) + | ChanMessage a (SendPort b) + deriving (Typeable, Generic) + +instance (Serializable a, Serializable b) => Binary (Message a b) where +instance (NFSerializable a, NFSerializable b) => NFData (Message a b) where +deriving instance (Eq a, Eq b) => Eq (Message a b) +deriving instance (Show a, Show b) => Show (Message a b) + +data CallResponse a = CallResponse a CallId + deriving (Typeable, Generic) + +instance Serializable a => Binary (CallResponse a) +instance NFSerializable a => NFData (CallResponse a) +deriving instance Eq a => Eq (CallResponse a) +deriving instance Show a => Show (CallResponse a) + +-- | Return type for and 'InitHandler' expression. +data InitResult s = + InitOk s Delay {- + ^ a successful initialisation, initial state and timeout -} + | InitStop String {- + ^ failed initialisation and the reason, this will result in an error -} + | InitIgnore {- + ^ the process has decided not to continue starting - this is not an error -} + deriving (Typeable) + +-- | The action taken by a process after a handler has run and its updated state. +-- See 'continue' +-- 'timeoutAfter' +-- 'hibernate' +-- 'stop' +-- 'stopWith' +-- +data ProcessAction s = + ProcessContinue s -- ^ continue with (possibly new) state + | ProcessTimeout Delay s -- ^ timeout if no messages are received + | ProcessHibernate TimeInterval s -- ^ hibernate for /delay/ + | ProcessStop ExitReason -- ^ stop the process, giving @ExitReason@ + | ProcessStopping s ExitReason -- ^ stop the process with @ExitReason@, with updated state + +-- | Returned from handlers for the synchronous 'call' protocol, encapsulates +-- the reply data /and/ the action to take after sending the reply. A handler +-- can return @NoReply@ if they wish to ignore the call. +data ProcessReply r s = + ProcessReply r (ProcessAction s) + | NoReply (ProcessAction s) + +-- | Wraps a predicate that is used to determine whether or not a handler +-- is valid based on some combination of the current process state, the +-- type and/or value of the input message or both. +data Condition s m = + Condition (s -> m -> Bool) -- ^ predicated on the process state /and/ the message + | State (s -> Bool) -- ^ predicated on the process state only + | Input (m -> Bool) -- ^ predicated on the input message only + +-- | An expression used to handle a /call/ message. +type CallHandler s a b = s -> a -> Process (ProcessReply b s) + +-- | An expression used to handle a /call/ message where the reply is deferred +-- via the 'CallRef'. +type DeferredCallHandler s a b = s -> CallRef b -> a -> Process (ProcessReply b s) + +-- | An expression used to handle a /call/ message in a stateless process. +type StatelessCallHandler a b = a -> CallRef b -> Process (ProcessReply b ()) + +-- | An expression used to handle a /cast/ message. +type CastHandler s a = s -> a -> Process (ProcessAction s) + +-- | An expression used to handle an /info/ message. +type InfoHandler s a = s -> a -> Process (ProcessAction s) + +-- | An expression used to handle a /channel/ message. +type ChannelHandler s a b = s -> SendPort b -> a -> Process (ProcessAction s) + +-- | An expression used to handle a /channel/ message in a stateless process. +type StatelessChannelHandler a b = SendPort b -> a -> Process (ProcessAction ()) + +-- | An expression used to initialise a process with its state. +type InitHandler a s = a -> Process (InitResult s) + +-- | An expression used to handle process termination. +type ShutdownHandler s = s -> ExitReason -> Process () + +-- | An expression used to handle process timeouts. +type TimeoutHandler s = s -> Delay -> Process (ProcessAction s) + +-- dispatching to implementation callbacks + +-- TODO: Now that we've got matchSTM available, we can have two kinds of CC. +-- The easiest approach would be to add an StmControlChannel newtype, since +-- that can't be Serializable (and will have to rely on PCopy for delivery). +-- Rather than write stmChanServe in terms of creating that channel object +-- ourselves (which is necessary for the TypedChannel based approach we +-- currently offer), I think it should accept the (STM a) "read" action and +-- leave the PCopy based delivery nonsense to the user, since we don't want +-- to /encourage/ that sort of thing outside of this codebase. + +{- + +data InputChannelDispatcher = + InputChannelDispatcher { chan :: InputChannel s + , dispatch :: s -> Message a b -> Process (ProcessAction s) + } + +instance MessageMatcher Dispatcher where + matchDispatch _ _ (DispatchInputChannelDispatcher c d) = matchInputChan (d s) +-} + +-- | Provides a means for servers to listen on a separate, typed /control/ +-- channel, thereby segregating the channel from their regular +-- (and potentially busy) mailbox. +newtype ControlChannel m = + ControlChannel { + unControl :: (SendPort (Message m ()), ReceivePort (Message m ())) + } + +-- | Creates a new 'ControlChannel'. +newControlChan :: (Serializable m) => Process (ControlChannel m) +newControlChan = newChan >>= return . ControlChannel + +-- | The writable end of a 'ControlChannel'. +-- +newtype ControlPort m = + ControlPort { + unPort :: SendPort (Message m ()) + } deriving (Show) +deriving instance (Serializable m) => Binary (ControlPort m) +instance Eq (ControlPort m) where + a == b = unPort a == unPort b + +-- | Obtain an opaque expression for communicating with a 'ControlChannel'. +-- +channelControlPort :: (Serializable m) + => ControlChannel m + -> ControlPort m +channelControlPort cc = ControlPort $ fst $ unControl cc + +-- | Provides dispatch from cast and call messages to a typed handler. +data Dispatcher s = + forall a b . (Serializable a, Serializable b) => + Dispatch + { + dispatch :: s -> Message a b -> Process (ProcessAction s) + } + | forall a b . (Serializable a, Serializable b) => + DispatchIf + { + dispatch :: s -> Message a b -> Process (ProcessAction s) + , dispatchIf :: s -> Message a b -> Bool + } + | forall a b . (Serializable a, Serializable b) => + DispatchCC -- control channel dispatch + { + channel :: ReceivePort (Message a b) + , dispatch :: s -> Message a b -> Process (ProcessAction s) + } + +-- | Provides dispatch for any input, returns 'Nothing' for unhandled messages. +data DeferredDispatcher s = + DeferredDispatcher + { + dispatchInfo :: s + -> P.Message + -> Process (Maybe (ProcessAction s)) + } + +-- | Provides dispatch for any exit signal - returns 'Nothing' for unhandled exceptions +data ExitSignalDispatcher s = + ExitSignalDispatcher + { + dispatchExit :: s + -> ProcessId + -> P.Message + -> Process (Maybe (ProcessAction s)) + } + +class MessageMatcher d where + matchDispatch :: UnhandledMessagePolicy -> s -> d s -> Match (ProcessAction s) + +instance MessageMatcher Dispatcher where + matchDispatch _ s (Dispatch d) = match (d s) + matchDispatch _ s (DispatchIf d cond) = matchIf (cond s) (d s) + matchDispatch _ s (DispatchCC c d) = matchChan c (d s) + +class DynMessageHandler d where + dynHandleMessage :: UnhandledMessagePolicy + -> s + -> d s + -> P.Message + -> Process (Maybe (ProcessAction s)) + +instance DynMessageHandler Dispatcher where + dynHandleMessage _ s (Dispatch d) msg = handleMessage msg (d s) + dynHandleMessage _ s (DispatchIf d c) msg = handleMessageIf msg (c s) (d s) + dynHandleMessage _ _ (DispatchCC _ _) _ = error "ThisCanNeverHappen" + +instance DynMessageHandler DeferredDispatcher where + dynHandleMessage _ s (DeferredDispatcher d) = d s + +newtype Priority a = Priority { getPrio :: Int } + +data DispatchPriority s = + PrioritiseCall + { + prioritise :: s -> P.Message -> Process (Maybe (Int, P.Message)) + } + | PrioritiseCast + { + prioritise :: s -> P.Message -> Process (Maybe (Int, P.Message)) + } + | PrioritiseInfo + { + prioritise :: s -> P.Message -> Process (Maybe (Int, P.Message)) + } + +-- | For a 'PrioritisedProcessDefinition', this policy determines for how long +-- the /receive loop/ should continue draining the process' mailbox before +-- processing its received mail (in priority order). +-- +-- If a prioritised /managed process/ is receiving a lot of messages (into its +-- /real/ mailbox), the server might never get around to actually processing its +-- inputs. This (mandatory) policy provides a guarantee that eventually (i.e., +-- after a specified number of received messages or time interval), the server +-- will stop removing messages from its mailbox and process those it has already +-- received. +-- +data RecvTimeoutPolicy = RecvCounter Int | RecvTimer TimeInterval + deriving (Typeable) + +-- | A @ProcessDefinition@ decorated with @DispatchPriority@ for certain +-- input domains. +data PrioritisedProcessDefinition s = + PrioritisedProcessDefinition + { + processDef :: ProcessDefinition s + , priorities :: [DispatchPriority s] + , recvTimeout :: RecvTimeoutPolicy + } + +-- | Policy for handling unexpected messages, i.e., messages which are not +-- sent using the 'call' or 'cast' APIs, and which are not handled by any of the +-- 'handleInfo' handlers. +data UnhandledMessagePolicy = + Terminate -- ^ stop immediately, giving @ExitOther "UnhandledInput"@ as the reason + | DeadLetter ProcessId -- ^ forward the message to the given recipient + | Log -- ^ log messages, then behave identically to @Drop@ + | Drop -- ^ dequeue and then drop/ignore the message + +-- | Stores the functions that determine runtime behaviour in response to +-- incoming messages and a policy for responding to unhandled messages. +data ProcessDefinition s = ProcessDefinition { + apiHandlers :: [Dispatcher s] -- ^ functions that handle call/cast messages + , infoHandlers :: [DeferredDispatcher s] -- ^ functions that handle non call/cast messages + , exitHandlers :: [ExitSignalDispatcher s] -- ^ functions that handle exit signals + , timeoutHandler :: TimeoutHandler s -- ^ a function that handles timeouts + , shutdownHandler :: ShutdownHandler s -- ^ a function that is run just before the process exits + , unhandledMessagePolicy :: UnhandledMessagePolicy -- ^ how to deal with unhandled messages + } + +-- note [rpc calls] +-- One problem with using plain expect/receive primitives to perform a +-- synchronous (round trip) call is that a reply matching the expected type +-- could come from anywhere! The Call.hs module uses a unique integer tag to +-- distinguish between inputs but this is easy to forge, and forces all callers +-- to maintain a tag pool, which is quite onerous. +-- +-- Here, we use a private (internal) tag based on a 'MonitorRef', which is +-- guaranteed to be unique per calling process (in the absence of mallicious +-- peers). This is handled throughout the roundtrip, such that the reply will +-- either contain the CallId (i.e., the ame 'MonitorRef' with which we're +-- tracking the server process) or we'll see the server die. +-- +-- Of course, the downside to all this is that the monitoring and receiving +-- clutters up your mailbox, and if your mailbox is extremely full, could +-- incur delays in delivery. The callAsync function provides a neat +-- work-around for that, relying on the insulation provided by Async. + +-- TODO: Generify this /call/ API and use it in Call.hs to avoid tagging + +-- TODO: the code below should be moved elsewhere. Maybe to Client.hs? +initCall :: forall s a b . (Addressable s, Serializable a, Serializable b) + => s -> a -> Process (CallRef b) +initCall sid msg = do + pid <- resolveOrDie sid "initCall: unresolveable address " + mRef <- monitor pid + self <- getSelfPid + let cRef = makeRef (Pid self) mRef in do + sendTo pid (CallMessage msg cRef :: Message a b) + return cRef + +unsafeInitCall :: forall s a b . (Addressable s, + NFSerializable a, NFSerializable b) + => s -> a -> Process (CallRef b) +unsafeInitCall sid msg = do + pid <- resolveOrDie sid "unsafeInitCall: unresolveable address " + mRef <- monitor pid + self <- getSelfPid + let cRef = makeRef (Pid self) mRef in do + unsafeSendTo pid (CallMessage msg cRef :: Message a b) + return cRef + +waitResponse :: forall b. (Serializable b) + => Maybe TimeInterval + -> CallRef b + -> Process (Maybe (Either ExitReason b)) +waitResponse mTimeout cRef = + let (_, mRef) = unCaller cRef + matchers = [ matchIf (\((CallResponse _ ref) :: CallResponse b) -> ref == mRef) + (\((CallResponse m _) :: CallResponse b) -> return (Right m)) + , matchIf (\(ProcessMonitorNotification ref _ _) -> ref == mRef) + (\(ProcessMonitorNotification _ _ r) -> return (Left (err r))) + ] + err r = ExitOther $ show r in + case mTimeout of + (Just ti) -> finally (receiveTimeout (asTimeout ti) matchers) (unmonitor mRef) + Nothing -> finally (receiveWait matchers >>= return . Just) (unmonitor mRef) + diff --git a/src/Control/Distributed/Process/ManagedProcess/Server.hs b/src/Control/Distributed/Process/ManagedProcess/Server.hs new file mode 100644 --- /dev/null +++ b/src/Control/Distributed/Process/ManagedProcess/Server.hs @@ -0,0 +1,600 @@ +{-# LANGUAGE ExistentialQuantification #-} +{-# LANGUAGE ScopedTypeVariables #-} + +----------------------------------------------------------------------------- +-- | +-- Module : Control.Distributed.Process.ManagedProcess.Server +-- Copyright : (c) Tim Watson 2012 - 2013 +-- License : BSD3 (see the file LICENSE) +-- +-- Maintainer : Tim Watson +-- Stability : experimental +-- Portability : non-portable (requires concurrency) +-- +-- The Server Portion of the /Managed Process/ API. +----------------------------------------------------------------------------- + +module Control.Distributed.Process.ManagedProcess.Server + ( -- * Server actions + condition + , state + , input + , reply + , replyWith + , noReply + , continue + , timeoutAfter + , hibernate + , stop + , stopWith + , replyTo + , replyChan + -- * Stateless actions + , noReply_ + , haltNoReply_ + , continue_ + , timeoutAfter_ + , hibernate_ + , stop_ + -- * Server handler/callback creation + , handleCall + , handleCallIf + , handleCallFrom + , handleCallFromIf + , handleRpcChan + , handleRpcChanIf + , handleCast + , handleCastIf + , handleInfo + , handleRaw + , handleDispatch + , handleDispatchIf + , handleExit + , handleExitIf + -- * Stateless handlers + , action + , handleCall_ + , handleCallIf_ + , handleCallFrom_ + , handleCallFromIf_ + , handleRpcChan_ + , handleRpcChanIf_ + , handleCast_ + , handleCastIf_ + -- * Working with Control Channels + , handleControlChan + , handleControlChan_ + ) where + +import Control.Distributed.Process hiding (call, Message) +import qualified Control.Distributed.Process as P (Message) +import Control.Distributed.Process.Serializable +import Control.Distributed.Process.ManagedProcess.Internal.Types +import Control.Distributed.Process.Extras + ( ExitReason(..) + , Routable(..) + ) +import Control.Distributed.Process.Extras.Time +import Prelude hiding (init) + +-------------------------------------------------------------------------------- +-- Producing ProcessAction and ProcessReply from inside handler expressions -- +-------------------------------------------------------------------------------- + +-- | Creates a 'Condition' from a function that takes a process state @a@ and +-- an input message @b@ and returns a 'Bool' indicating whether the associated +-- handler should run. +-- +condition :: forall a b. (Serializable a, Serializable b) + => (a -> b -> Bool) + -> Condition a b +condition = Condition + +-- | Create a 'Condition' from a function that takes a process state @a@ and +-- returns a 'Bool' indicating whether the associated handler should run. +-- +state :: forall s m. (Serializable m) => (s -> Bool) -> Condition s m +state = State + +-- | Creates a 'Condition' from a function that takes an input message @m@ and +-- returns a 'Bool' indicating whether the associated handler should run. +-- +input :: forall s m. (Serializable m) => (m -> Bool) -> Condition s m +input = Input + +-- | Instructs the process to send a reply and continue running. +reply :: (Serializable r) => r -> s -> Process (ProcessReply r s) +reply r s = continue s >>= replyWith r + +-- | Instructs the process to send a reply /and/ evaluate the 'ProcessAction'. +replyWith :: (Serializable r) + => r + -> ProcessAction s + -> Process (ProcessReply r s) +replyWith r s = return $ ProcessReply r s + +-- | Instructs the process to skip sending a reply /and/ evaluate a 'ProcessAction' +noReply :: (Serializable r) => ProcessAction s -> Process (ProcessReply r s) +noReply = return . NoReply + +-- | Continue without giving a reply to the caller - equivalent to 'continue', +-- but usable in a callback passed to the 'handleCall' family of functions. +noReply_ :: forall s r . (Serializable r) => s -> Process (ProcessReply r s) +noReply_ s = continue s >>= noReply + +-- | Halt process execution during a call handler, without paying any attention +-- to the expected return type. +haltNoReply_ :: Serializable r => ExitReason -> Process (ProcessReply r s) +haltNoReply_ r = stop r >>= noReply + +-- | Instructs the process to continue running and receiving messages. +continue :: s -> Process (ProcessAction s) +continue = return . ProcessContinue + +-- | Version of 'continue' that can be used in handlers that ignore process state. +-- +continue_ :: (s -> Process (ProcessAction s)) +continue_ = return . ProcessContinue + +-- | Instructs the process loop to wait for incoming messages until 'Delay' +-- is exceeded. If no messages are handled during this period, the /timeout/ +-- handler will be called. Note that this alters the process timeout permanently +-- such that the given @Delay@ will remain in use until changed. +timeoutAfter :: Delay -> s -> Process (ProcessAction s) +timeoutAfter d s = return $ ProcessTimeout d s + +-- | Version of 'timeoutAfter' that can be used in handlers that ignore process state. +-- +-- > action (\(TimeoutPlease duration) -> timeoutAfter_ duration) +-- +timeoutAfter_ :: Delay -> (s -> Process (ProcessAction s)) +timeoutAfter_ d = return . ProcessTimeout d + +-- | Instructs the process to /hibernate/ for the given 'TimeInterval'. Note +-- that no messages will be removed from the mailbox until after hibernation has +-- ceased. This is equivalent to calling @threadDelay@. +-- +hibernate :: TimeInterval -> s -> Process (ProcessAction s) +hibernate d s = return $ ProcessHibernate d s + +-- | Version of 'hibernate' that can be used in handlers that ignore process state. +-- +-- > action (\(HibernatePlease delay) -> hibernate_ delay) +-- +hibernate_ :: TimeInterval -> (s -> Process (ProcessAction s)) +hibernate_ d = return . ProcessHibernate d + +-- | Instructs the process to terminate, giving the supplied reason. If a valid +-- 'shutdownHandler' is installed, it will be called with the 'ExitReason' +-- returned from this call, along with the process state. +stop :: ExitReason -> Process (ProcessAction s) +stop r = return $ ProcessStop r + +-- | As 'stop', but provides an updated state for the shutdown handler. +stopWith :: s -> ExitReason -> Process (ProcessAction s) +stopWith s r = return $ ProcessStopping s r + +-- | Version of 'stop' that can be used in handlers that ignore process state. +-- +-- > action (\ClientError -> stop_ ExitNormal) +-- +stop_ :: ExitReason -> (s -> Process (ProcessAction s)) +stop_ r _ = stop r + +-- | Sends a reply explicitly to a caller. +-- +-- > replyTo = sendTo +-- +replyTo :: (Serializable m) => CallRef m -> m -> Process () +replyTo = sendTo + +-- | Sends a reply to a 'SendPort' (for use in 'handleRpcChan' et al). +-- +-- > replyChan = sendChan +-- +replyChan :: (Serializable m) => SendPort m -> m -> Process () +replyChan = sendChan + +-------------------------------------------------------------------------------- +-- Wrapping handler expressions in Dispatcher and DeferredDispatcher -- +-------------------------------------------------------------------------------- + +-- | Constructs a 'call' handler from a function in the 'Process' monad. +-- The handler expression returns the reply, and the action will be +-- set to 'continue'. +-- +-- > handleCall_ = handleCallIf_ $ input (const True) +-- +handleCall_ :: (Serializable a, Serializable b) + => (a -> Process b) + -> Dispatcher s +handleCall_ = handleCallIf_ $ input (const True) + +-- | Constructs a 'call' handler from an ordinary function in the 'Process' +-- monad. This variant ignores the state argument present in 'handleCall' and +-- 'handleCallIf' and is therefore useful in a stateless server. Messges are +-- only dispatched to the handler if the supplied condition evaluates to @True@ +-- +-- See 'handleCall' +handleCallIf_ :: forall s a b . (Serializable a, Serializable b) + => Condition s a -- ^ predicate that must be satisfied for the handler to run + -> (a -> Process b) -- ^ a function from an input message to a reply + -> Dispatcher s +handleCallIf_ cond handler + = DispatchIf { + dispatch = doHandle handler + , dispatchIf = checkCall cond + } + where doHandle :: (Serializable a, Serializable b) + => (a -> Process b) + -> s + -> Message a b + -> Process (ProcessAction s) + doHandle h s (CallMessage p c) = (h p) >>= mkCallReply c s + doHandle _ _ _ = die "CALL_HANDLER_TYPE_MISMATCH" -- note [Message type] + + -- handling 'reply-to' in the main process loop is awkward at best, + -- so we handle it here instead and return the 'action' to the loop + mkCallReply :: (Serializable b) + => CallRef b + -> s + -> b + -> Process (ProcessAction s) + mkCallReply c s m = + let (c', t) = unCaller c + in sendTo c' (CallResponse m t) >> continue s + +-- | Constructs a 'call' handler from a function in the 'Process' monad. +-- > handleCall = handleCallIf (const True) +-- +handleCall :: (Serializable a, Serializable b) + => (s -> a -> Process (ProcessReply b s)) + -> Dispatcher s +handleCall = handleCallIf $ state (const True) + +-- | Constructs a 'call' handler from an ordinary function in the 'Process' +-- monad. Given a function @f :: (s -> a -> Process (ProcessReply b s))@, +-- the expression @handleCall f@ will yield a 'Dispatcher' for inclusion +-- in a 'Behaviour' specification for the /GenProcess/. Messages are only +-- dispatched to the handler if the supplied condition evaluates to @True@. +-- +handleCallIf :: forall s a b . (Serializable a, Serializable b) + => Condition s a -- ^ predicate that must be satisfied for the handler to run + -> (s -> a -> Process (ProcessReply b s)) + -- ^ a reply yielding function over the process state and input message + -> Dispatcher s +handleCallIf cond handler + = DispatchIf { + dispatch = doHandle handler + , dispatchIf = checkCall cond + } + where doHandle :: (Serializable a, Serializable b) + => (s -> a -> Process (ProcessReply b s)) + -> s + -> Message a b + -> Process (ProcessAction s) + doHandle h s (CallMessage p c) = (h s p) >>= mkReply c + doHandle _ _ _ = die "CALL_HANDLER_TYPE_MISMATCH" -- note [Message type] + +-- | A variant of 'handleCallFrom_' that ignores the state argument. +-- +handleCallFrom_ :: forall s a b . (Serializable a, Serializable b) + => (CallRef b -> a -> Process (ProcessReply b s)) + -> Dispatcher s +handleCallFrom_ = handleCallFromIf_ $ input (const True) + +-- | A variant of 'handleCallFromIf' that ignores the state argument. +-- +handleCallFromIf_ :: forall s a b . (Serializable a, Serializable b) + => (Condition s a) + -> (CallRef b -> a -> Process (ProcessReply b s)) + -> Dispatcher s +handleCallFromIf_ c h = + DispatchIf { + dispatch = doHandle h + , dispatchIf = checkCall c + } + where doHandle :: (Serializable a, Serializable b) + => (CallRef b -> a -> Process (ProcessReply b s)) + -> s + -> Message a b + -> Process (ProcessAction s) + doHandle h' _ (CallMessage p c') = (h' c' p) >>= mkReply c' + doHandle _ _ _ = die "CALL_HANDLER_TYPE_MISMATCH" -- note [Message type] + +-- | As 'handleCall' but passes the 'CallRef' to the handler function. +-- This can be useful if you wish to /reply later/ to the caller by, e.g., +-- spawning a process to do some work and have it @replyTo caller response@ +-- out of band. In this case the callback can pass the 'CallRef' to the +-- worker (or stash it away itself) and return 'noReply'. +-- +handleCallFrom :: forall s a b . (Serializable a, Serializable b) + => (s -> CallRef b -> a -> Process (ProcessReply b s)) + -> Dispatcher s +handleCallFrom = handleCallFromIf $ state (const True) + +-- | As 'handleCallFrom' but only runs the handler if the supplied 'Condition' +-- evaluates to @True@. +-- +handleCallFromIf :: forall s a b . (Serializable a, Serializable b) + => Condition s a -- ^ predicate that must be satisfied for the handler to run + -> (s -> CallRef b -> a -> Process (ProcessReply b s)) + -- ^ a reply yielding function over the process state, sender and input message + -> Dispatcher s +handleCallFromIf cond handler + = DispatchIf { + dispatch = doHandle handler + , dispatchIf = checkCall cond + } + where doHandle :: (Serializable a, Serializable b) + => (s -> CallRef b -> a -> Process (ProcessReply b s)) + -> s + -> Message a b + -> Process (ProcessAction s) + doHandle h s (CallMessage p c) = (h s c p) >>= mkReply c + doHandle _ _ _ = die "CALL_HANDLER_TYPE_MISMATCH" -- note [Message type] + +-- | Creates a handler for a /typed channel/ RPC style interaction. The +-- handler takes a @SendPort b@ to reply to, the initial input and evaluates +-- to a 'ProcessAction'. It is the handler code's responsibility to send the +-- reply to the @SendPort@. +-- +handleRpcChan :: forall s a b . (Serializable a, Serializable b) + => (s -> SendPort b -> a -> Process (ProcessAction s)) + -> Dispatcher s +handleRpcChan = handleRpcChanIf $ input (const True) + +-- | As 'handleRpcChan', but only evaluates the handler if the supplied +-- condition is met. +-- +handleRpcChanIf :: forall s a b . (Serializable a, Serializable b) + => Condition s a + -> (s -> SendPort b -> a -> Process (ProcessAction s)) + -> Dispatcher s +handleRpcChanIf c h + = DispatchIf { + dispatch = doHandle h + , dispatchIf = checkRpc c + } + where doHandle :: (Serializable a, Serializable b) + => (s -> SendPort b -> a -> Process (ProcessAction s)) + -> s + -> Message a b + -> Process (ProcessAction s) + doHandle h' s (ChanMessage p c') = h' s c' p + doHandle _ _ _ = die "RPC_HANDLER_TYPE_MISMATCH" -- node [Message type] + +-- | A variant of 'handleRpcChan' that ignores the state argument. +-- +handleRpcChan_ :: forall a b . (Serializable a, Serializable b) + => (SendPort b -> a -> Process (ProcessAction ())) + -> Dispatcher () +handleRpcChan_ h = handleRpcChan (\() -> h) + +-- | A variant of 'handleRpcChanIf' that ignores the state argument. +-- +handleRpcChanIf_ :: forall a b . (Serializable a, Serializable b) + => Condition () a + -> (SendPort b -> a -> Process (ProcessAction ())) + -> Dispatcher () +handleRpcChanIf_ c h = handleRpcChanIf c (\() -> h) + +-- | Constructs a 'cast' handler from an ordinary function in the 'Process' +-- monad. +-- > handleCast = handleCastIf (const True) +-- +handleCast :: (Serializable a) + => (s -> a -> Process (ProcessAction s)) + -> Dispatcher s +handleCast = handleCastIf $ input (const True) + +-- | Constructs a 'cast' handler from an ordinary function in the 'Process' +-- monad. Given a function @f :: (s -> a -> Process (ProcessAction s))@, +-- the expression @handleCall f@ will yield a 'Dispatcher' for inclusion +-- in a 'Behaviour' specification for the /GenProcess/. +-- +handleCastIf :: forall s a . (Serializable a) + => Condition s a -- ^ predicate that must be satisfied for the handler to run + -> (s -> a -> Process (ProcessAction s)) + -- ^ an action yielding function over the process state and input message + -> Dispatcher s +handleCastIf cond h + = DispatchIf { + dispatch = (\s ((CastMessage p) :: Message a ()) -> h s p) + , dispatchIf = checkCast cond + } + +-- | Constructs a /control channel/ handler from a function in the +-- 'Process' monad. The handler expression returns no reply, and the +-- /control message/ is treated in the same fashion as a 'cast'. +-- +-- > handleControlChan = handleControlChanIf $ input (const True) +-- +handleControlChan :: forall s a . (Serializable a) + => ControlChannel a -- ^ the receiving end of the control channel + -> (s -> a -> Process (ProcessAction s)) + -- ^ an action yielding function over the process state and input message + -> Dispatcher s +handleControlChan chan h + = DispatchCC { channel = snd $ unControl chan + , dispatch = (\s ((CastMessage p) :: Message a ()) -> h s p) + } + +-- | Version of 'handleControlChan' that ignores the server state. +-- +handleControlChan_ :: forall s a. (Serializable a) + => ControlChannel a + -> (a -> (s -> Process (ProcessAction s))) + -> Dispatcher s +handleControlChan_ chan h + = DispatchCC { channel = snd $ unControl chan + , dispatch = (\s ((CastMessage p) :: Message a ()) -> h p $ s) + } + +-- | Version of 'handleCast' that ignores the server state. +-- +handleCast_ :: (Serializable a) + => (a -> (s -> Process (ProcessAction s))) -> Dispatcher s +handleCast_ = handleCastIf_ $ input (const True) + +-- | Version of 'handleCastIf' that ignores the server state. +-- +handleCastIf_ :: forall s a . (Serializable a) + => Condition s a -- ^ predicate that must be satisfied for the handler to run + -> (a -> (s -> Process (ProcessAction s))) + -- ^ a function from the input message to a /stateless action/, cf 'continue_' + -> Dispatcher s +handleCastIf_ cond h + = DispatchIf { dispatch = (\s ((CastMessage p) :: Message a ()) -> h p $ s) + , dispatchIf = checkCast cond + } + +-- | Constructs an /action/ handler. Like 'handleDispatch' this can handle both +-- 'cast' and 'call' messages, but you won't know which you're dealing with. +-- This can be useful where certain inputs require a definite action, such as +-- stopping the server, without concern for the state (e.g., when stopping we +-- need only decide to stop, as the terminate handler can deal with state +-- cleanup etc). For example: +-- +-- @action (\MyCriticalSignal -> stop_ ExitNormal)@ +-- +action :: forall s a . (Serializable a) + => (a -> (s -> Process (ProcessAction s))) + -- ^ a function from the input message to a /stateless action/, cf 'continue_' + -> Dispatcher s +action h = handleDispatch perform + where perform :: (s -> a -> Process (ProcessAction s)) + perform s a = let f = h a in f s + +-- | Constructs a handler for both /call/ and /cast/ messages. +-- @handleDispatch = handleDispatchIf (const True)@ +-- +handleDispatch :: forall s a . (Serializable a) + => (s -> a -> Process (ProcessAction s)) + -> Dispatcher s +handleDispatch = handleDispatchIf $ input (const True) + +-- | Constructs a handler for both /call/ and /cast/ messages. Messages are only +-- dispatched to the handler if the supplied condition evaluates to @True@. +-- Handlers defined in this way have no access to the call context (if one +-- exists) and cannot therefore reply to calls. +-- +handleDispatchIf :: forall s a . (Serializable a) + => Condition s a + -> (s -> a -> Process (ProcessAction s)) + -> Dispatcher s +handleDispatchIf cond handler = DispatchIf { + dispatch = doHandle handler + , dispatchIf = check cond + } + where doHandle :: (Serializable a) + => (s -> a -> Process (ProcessAction s)) + -> s + -> Message a () + -> Process (ProcessAction s) + doHandle h s msg = + case msg of + (CallMessage p _) -> (h s p) + (CastMessage p) -> (h s p) + (ChanMessage p _) -> (h s p) + +-- | Creates a generic input handler (i.e., for received messages that are /not/ +-- sent using the 'cast' or 'call' APIs) from an ordinary function in the +-- 'Process' monad. +handleInfo :: forall s a. (Serializable a) + => (s -> a -> Process (ProcessAction s)) + -> DeferredDispatcher s +handleInfo h = DeferredDispatcher { dispatchInfo = doHandleInfo h } + where + doHandleInfo :: forall s2 a2. (Serializable a2) + => (s2 -> a2 -> Process (ProcessAction s2)) + -> s2 + -> P.Message + -> Process (Maybe (ProcessAction s2)) + doHandleInfo h' s msg = handleMessage msg (h' s) + +-- | Handle completely /raw/ input messages. +-- +handleRaw :: forall s. (s -> P.Message -> Process (ProcessAction s)) + -> DeferredDispatcher s +handleRaw h = DeferredDispatcher { dispatchInfo = doHandle h } + where + doHandle h' s msg = h' s msg >>= return . Just + +-- | Creates an /exit handler/ scoped to the execution of any and all the +-- registered call, cast and info handlers for the process. +handleExit :: forall s a. (Serializable a) + => (s -> ProcessId -> a -> Process (ProcessAction s)) + -> ExitSignalDispatcher s +handleExit h = ExitSignalDispatcher { dispatchExit = doHandleExit h } + where + doHandleExit :: (s -> ProcessId -> a -> Process (ProcessAction s)) + -> s + -> ProcessId + -> P.Message + -> Process (Maybe (ProcessAction s)) + doHandleExit h' s p msg = handleMessage msg (h' s p) + +handleExitIf :: forall s a . (Serializable a) + => (s -> a -> Bool) + -> (s -> ProcessId -> a -> Process (ProcessAction s)) + -> ExitSignalDispatcher s +handleExitIf c h = ExitSignalDispatcher { dispatchExit = doHandleExit c h } + where + doHandleExit :: (s -> a -> Bool) + -> (s -> ProcessId -> a -> Process (ProcessAction s)) + -> s + -> ProcessId + -> P.Message + -> Process (Maybe (ProcessAction s)) + doHandleExit c' h' s p msg = handleMessageIf msg (c' s) (h' s p) + +-- handling 'reply-to' in the main process loop is awkward at best, +-- so we handle it here instead and return the 'action' to the loop +mkReply :: (Serializable b) + => CallRef b + -> ProcessReply b s + -> Process (ProcessAction s) +mkReply _ (NoReply a) = return a +mkReply c (ProcessReply r' a) = sendTo c r' >> return a + +-- these functions are the inverse of 'condition', 'state' and 'input' + +check :: forall s m a . (Serializable m) + => Condition s m + -> s + -> Message m a + -> Bool +check (Condition c) st msg = c st $ decode msg +check (State c) st _ = c st +check (Input c) _ msg = c $ decode msg + +checkRpc :: forall s m a . (Serializable m) + => Condition s m + -> s + -> Message m a + -> Bool +checkRpc cond st msg@(ChanMessage _ _) = check cond st msg +checkRpc _ _ _ = False + +checkCall :: forall s m a . (Serializable m) + => Condition s m + -> s + -> Message m a + -> Bool +checkCall cond st msg@(CallMessage _ _) = check cond st msg +checkCall _ _ _ = False + +checkCast :: forall s m . (Serializable m) + => Condition s m + -> s + -> Message m () + -> Bool +checkCast cond st msg@(CastMessage _) = check cond st msg +checkCast _ _ _ = False + +decode :: Message a b -> a +decode (CallMessage a _) = a +decode (CastMessage a) = a +decode (ChanMessage a _) = a + diff --git a/src/Control/Distributed/Process/ManagedProcess/Server/Priority.hs b/src/Control/Distributed/Process/ManagedProcess/Server/Priority.hs new file mode 100644 --- /dev/null +++ b/src/Control/Distributed/Process/ManagedProcess/Server/Priority.hs @@ -0,0 +1,95 @@ +{-# LANGUAGE ExistentialQuantification #-} +{-# LANGUAGE ScopedTypeVariables #-} +{-# LANGUAGE PatternGuards #-} + +module Control.Distributed.Process.ManagedProcess.Server.Priority + ( prioritiseCall + , prioritiseCall_ + , prioritiseCast + , prioritiseCast_ + , prioritiseInfo + , prioritiseInfo_ + , setPriority + ) where + +import Control.Distributed.Process hiding (call, Message) +import qualified Control.Distributed.Process as P (Message) +import Control.Distributed.Process.ManagedProcess.Internal.Types +import Control.Distributed.Process.Serializable +import Prelude hiding (init) + +setPriority :: Int -> Priority m +setPriority = Priority + +prioritiseCall_ :: forall s a b . (Serializable a, Serializable b) + => (a -> Priority b) + -> DispatchPriority s +prioritiseCall_ h = prioritiseCall (\_ -> h) + +prioritiseCall :: forall s a b . (Serializable a, Serializable b) + => (s -> a -> Priority b) + -> DispatchPriority s +prioritiseCall h = PrioritiseCall (\s -> unCall $ h s) + where + unCall :: (a -> Priority b) -> P.Message -> Process (Maybe (Int, P.Message)) + unCall h' m = unwrapMessage m >>= return . matchPrioritise m h' + + matchPrioritise :: P.Message + -> (a -> Priority b) + -> Maybe (Message a b) + -> Maybe (Int, P.Message) + matchPrioritise msg p msgIn + | (Just a@(CallMessage m _)) <- msgIn + , True <- isEncoded msg = Just (getPrio $ p m, wrapMessage a) + | (Just (CallMessage m _)) <- msgIn + , False <- isEncoded msg = Just (getPrio $ p m, msg) + | otherwise = Nothing + +prioritiseCast_ :: forall s a . (Serializable a) + => (a -> Priority ()) + -> DispatchPriority s +prioritiseCast_ h = prioritiseCast (\_ -> h) + +prioritiseCast :: forall s a . (Serializable a) + => (s -> a -> Priority ()) + -> DispatchPriority s +prioritiseCast h = PrioritiseCast (\s -> unCast $ h s) + where + unCast :: (a -> Priority ()) -> P.Message -> Process (Maybe (Int, P.Message)) + unCast h' m = unwrapMessage m >>= return . matchPrioritise m h' + + matchPrioritise :: P.Message + -> (a -> Priority ()) + -> Maybe (Message a ()) + -> Maybe (Int, P.Message) + matchPrioritise msg p msgIn + | (Just a@(CastMessage m)) <- msgIn + , True <- isEncoded msg = Just (getPrio $ p m, wrapMessage a) + | (Just (CastMessage m)) <- msgIn + , False <- isEncoded msg = Just (getPrio $ p m, msg) + | otherwise = Nothing + +prioritiseInfo_ :: forall s a . (Serializable a) + => (a -> Priority ()) + -> DispatchPriority s +prioritiseInfo_ h = prioritiseInfo (\_ -> h) + +prioritiseInfo :: forall s a . (Serializable a) + => (s -> a -> Priority ()) + -> DispatchPriority s +prioritiseInfo h = PrioritiseInfo (\s -> unMsg $ h s) + where + unMsg :: (a -> Priority ()) -> P.Message -> Process (Maybe (Int, P.Message)) + unMsg h' m = unwrapMessage m >>= return . matchPrioritise m h' + + matchPrioritise :: P.Message + -> (a -> Priority ()) + -> Maybe a + -> Maybe (Int, P.Message) + matchPrioritise msg p msgIn + | (Just m') <- msgIn + , True <- isEncoded msg = Just (getPrio $ p m', wrapMessage m') + | (Just m') <- msgIn + , False <- isEncoded msg = Just (getPrio $ p m', msg) + | otherwise = Nothing + diff --git a/src/Control/Distributed/Process/ManagedProcess/Server/Restricted.hs b/src/Control/Distributed/Process/ManagedProcess/Server/Restricted.hs new file mode 100644 --- /dev/null +++ b/src/Control/Distributed/Process/ManagedProcess/Server/Restricted.hs @@ -0,0 +1,275 @@ +{-# LANGUAGE ExistentialQuantification #-} +{-# LANGUAGE ScopedTypeVariables #-} +{-# LANGUAGE DeriveDataTypeable #-} +{-# LANGUAGE GeneralizedNewtypeDeriving #-} + +----------------------------------------------------------------------------- +-- | +-- Module : Control.Distributed.Process.ManagedProcess.Server.Restricted +-- Copyright : (c) Tim Watson 2012 - 2013 +-- License : BSD3 (see the file LICENSE) +-- +-- Maintainer : Tim Watson +-- Stability : experimental +-- Portability : non-portable (requires concurrency) +-- +-- A /safe/ variant of the Server Portion of the /Managed Process/ API. Most +-- of these operations have the same names as similar operations in the impure +-- @Server@ module (re-exported by the primary API in @ManagedProcess@). To +-- remove the ambiguity, some combination of either qualification and/or the +-- @hiding@ clause will be required. +-- +-- [Restricted Server Callbacks] +-- +-- The idea behind this module is to provide /safe/ callbacks, i.e., server +-- code that is free from side effects. This safety is enforced by the type +-- system via the @RestrictedProcess@ monad. A StateT interface is provided +-- for code running in the @RestrictedProcess@ monad, so that server side +-- state can be managed safely without resorting to IO (or code running in +-- the @Process@ monad). +-- +----------------------------------------------------------------------------- + +module Control.Distributed.Process.ManagedProcess.Server.Restricted + ( -- * Exported Types + RestrictedProcess + , Result(..) + , RestrictedAction(..) + -- * Creating call/cast protocol handlers + , handleCall + , handleCallIf + , handleCast + , handleCastIf + , handleInfo + , handleExit + , handleTimeout + -- * Handling Process State + , putState + , getState + , modifyState + -- * Handling responses/transitions + , reply + , noReply + , haltNoReply + , continue + , timeoutAfter + , hibernate + , stop + -- * Utilities + , say + ) where + +import Control.Applicative (Applicative) +import Control.Distributed.Process hiding (call, say) +import qualified Control.Distributed.Process as P (say) +import Control.Distributed.Process.Extras + (ExitReason(..)) +import Control.Distributed.Process.ManagedProcess.Internal.Types +import qualified Control.Distributed.Process.ManagedProcess.Server as Server +import Control.Distributed.Process.Extras.Time +import Control.Distributed.Process.Serializable +import Prelude hiding (init) + +import Control.Monad.IO.Class (MonadIO) +import qualified Control.Monad.State as ST + ( MonadState + , MonadTrans + , StateT + , get + , lift + , modify + , put + , runStateT + ) + +import Data.Typeable + +-- | Restricted (i.e., pure, free from side effects) execution +-- environment for call/cast/info handlers to execute in. +-- +newtype RestrictedProcess s a = RestrictedProcess { + unRestricted :: ST.StateT s Process a + } + deriving (Functor, Monad, ST.MonadState s, MonadIO, Typeable, Applicative) + +-- | The result of a 'call' handler's execution. +data Result a = + Reply a -- ^ reply with the given term + | Timeout Delay a -- ^ reply with the given term and enter timeout + | Hibernate TimeInterval a -- ^ reply with the given term and hibernate + | Stop ExitReason -- ^ stop the process with the given reason + deriving (Typeable) + +-- | The result of a safe 'cast' handler's execution. +data RestrictedAction = + RestrictedContinue -- ^ continue executing + | RestrictedTimeout Delay -- ^ timeout if no messages are received + | RestrictedHibernate TimeInterval -- ^ hibernate (i.e., sleep) + | RestrictedStop ExitReason -- ^ stop/terminate the server process + +-------------------------------------------------------------------------------- +-- Handling state in RestrictedProcess execution environments -- +-------------------------------------------------------------------------------- + +-- | Log a trace message using the underlying Process's @say@ +say :: String -> RestrictedProcess s () +say msg = lift . P.say $ msg + +-- | Get the current process state +getState :: RestrictedProcess s s +getState = ST.get + +-- | Put a new process state state +putState :: s -> RestrictedProcess s () +putState = ST.put + +-- | Apply the given expression to the current process state +modifyState :: (s -> s) -> RestrictedProcess s () +modifyState = ST.modify + +-------------------------------------------------------------------------------- +-- Generating replies and state transitions inside RestrictedProcess -- +-------------------------------------------------------------------------------- + +-- | Instructs the process to send a reply and continue running. +reply :: forall s r . (Serializable r) => r -> RestrictedProcess s (Result r) +reply = return . Reply + +-- | Continue without giving a reply to the caller - equivalent to 'continue', +-- but usable in a callback passed to the 'handleCall' family of functions. +noReply :: forall s r . (Serializable r) + => Result r + -> RestrictedProcess s (Result r) +noReply r = return r + +-- | Halt process execution during a call handler, without paying any attention +-- to the expected return type. +haltNoReply :: forall s r . (Serializable r) + => ExitReason + -> RestrictedProcess s (Result r) +haltNoReply r = noReply (Stop r) + +-- | Instructs the process to continue running and receiving messages. +continue :: forall s . RestrictedProcess s RestrictedAction +continue = return RestrictedContinue + +-- | Instructs the process loop to wait for incoming messages until 'Delay' +-- is exceeded. If no messages are handled during this period, the /timeout/ +-- handler will be called. Note that this alters the process timeout permanently +-- such that the given @Delay@ will remain in use until changed. +timeoutAfter :: forall s. Delay -> RestrictedProcess s RestrictedAction +timeoutAfter d = return $ RestrictedTimeout d + +-- | Instructs the process to /hibernate/ for the given 'TimeInterval'. Note +-- that no messages will be removed from the mailbox until after hibernation has +-- ceased. This is equivalent to evaluating @liftIO . threadDelay@. +-- +hibernate :: forall s. TimeInterval -> RestrictedProcess s RestrictedAction +hibernate d = return $ RestrictedHibernate d + +-- | Instructs the process to terminate, giving the supplied reason. If a valid +-- 'shutdownHandler' is installed, it will be called with the 'ExitReason' +-- returned from this call, along with the process state. +stop :: forall s. ExitReason -> RestrictedProcess s RestrictedAction +stop r = return $ RestrictedStop r + +-------------------------------------------------------------------------------- +-- Wrapping handler expressions in Dispatcher and DeferredDispatcher -- +-------------------------------------------------------------------------------- + +-- | A version of "Control.Distributed.Process.ManagedProcess.Server.handleCall" +-- that takes a handler which executes in 'RestrictedProcess'. +-- +handleCall :: forall s a b . (Serializable a, Serializable b) + => (a -> RestrictedProcess s (Result b)) + -> Dispatcher s +handleCall = handleCallIf $ Server.state (const True) + +-- | A version of "Control.Distributed.Process.ManagedProcess.Server.handleCallIf" +-- that takes a handler which executes in 'RestrictedProcess'. +-- +handleCallIf :: forall s a b . (Serializable a, Serializable b) + => (Condition s a) + -> (a -> RestrictedProcess s (Result b)) + -> Dispatcher s +handleCallIf cond h = Server.handleCallIf cond (wrapCall h) + +-- | A version of "Control.Distributed.Process.ManagedProcess.Server.handleCast" +-- that takes a handler which executes in 'RestrictedProcess'. +-- +handleCast :: forall s a . (Serializable a) + => (a -> RestrictedProcess s RestrictedAction) + -> Dispatcher s +handleCast = handleCastIf (Server.state (const True)) + +-- | A version of "Control.Distributed.Process.ManagedProcess.Server.handleCastIf" +-- that takes a handler which executes in 'RestrictedProcess'. +-- +handleCastIf :: forall s a . (Serializable a) + => Condition s a -- ^ predicate that must be satisfied for the handler to run + -> (a -> RestrictedProcess s RestrictedAction) + -- ^ an action yielding function over the process state and input message + -> Dispatcher s +handleCastIf cond h = Server.handleCastIf cond (wrapHandler h) + +-- | A version of "Control.Distributed.Process.ManagedProcess.Server.handleInfo" +-- that takes a handler which executes in 'RestrictedProcess'. +-- +handleInfo :: forall s a. (Serializable a) + => (a -> RestrictedProcess s RestrictedAction) + -> DeferredDispatcher s +-- cast and info look the same to a restricted process +handleInfo h = Server.handleInfo (wrapHandler h) + +handleExit :: forall s a. (Serializable a) + => (a -> RestrictedProcess s RestrictedAction) + -> ExitSignalDispatcher s +handleExit h = Server.handleExit $ \s _ a -> (wrapHandler h) s a + +handleTimeout :: forall s . (Delay -> RestrictedProcess s RestrictedAction) + -> TimeoutHandler s +handleTimeout h = \s d -> do + (r, s') <- runRestricted s (h d) + case r of + RestrictedContinue -> Server.continue s' + (RestrictedTimeout i) -> Server.timeoutAfter i s' + (RestrictedHibernate i) -> Server.hibernate i s' + (RestrictedStop r') -> Server.stop r' + +-------------------------------------------------------------------------------- +-- Implementation -- +-------------------------------------------------------------------------------- + +wrapHandler :: forall s a . (Serializable a) + => (a -> RestrictedProcess s RestrictedAction) + -> s + -> a + -> Process (ProcessAction s) +wrapHandler h s a = do + (r, s') <- runRestricted s (h a) + case r of + RestrictedContinue -> Server.continue s' + (RestrictedTimeout i) -> Server.timeoutAfter i s' + (RestrictedHibernate i) -> Server.hibernate i s' + (RestrictedStop r') -> Server.stop r' + +wrapCall :: forall s a b . (Serializable a, Serializable b) + => (a -> RestrictedProcess s (Result b)) + -> s + -> a + -> Process (ProcessReply b s) +wrapCall h s a = do + (r, s') <- runRestricted s (h a) + case r of + (Reply r') -> Server.reply r' s' + (Timeout i r') -> Server.timeoutAfter i s' >>= Server.replyWith r' + (Hibernate i r') -> Server.hibernate i s' >>= Server.replyWith r' + (Stop r'' ) -> Server.stop r'' >>= Server.noReply + +runRestricted :: s -> RestrictedProcess s a -> Process (a, s) +runRestricted state proc = ST.runStateT (unRestricted proc) state + +-- | TODO MonadTrans instance? lift :: (Monad m) => m a -> t m a +lift :: Process a -> RestrictedProcess s a +lift p = RestrictedProcess $ ST.lift p + diff --git a/src/Control/Distributed/Process/ManagedProcess/UnsafeClient.hs b/src/Control/Distributed/Process/ManagedProcess/UnsafeClient.hs new file mode 100644 --- /dev/null +++ b/src/Control/Distributed/Process/ManagedProcess/UnsafeClient.hs @@ -0,0 +1,181 @@ +{-# LANGUAGE ExistentialQuantification #-} +{-# LANGUAGE ScopedTypeVariables #-} + +----------------------------------------------------------------------------- +-- | +-- Module : Control.Distributed.Process.ManagedProcess.UnsafeClient +-- Copyright : (c) Tim Watson 2012 - 2013 +-- License : BSD3 (see the file LICENSE) +-- +-- Maintainer : Tim Watson +-- Stability : experimental +-- Portability : non-portable (requires concurrency) +-- +-- Unsafe variant of the /Managed Process Client API/. This module implements +-- the client portion of a Managed Process using the unsafe variants of cloud +-- haskell's messaging primitives. It relies on the -extras implementation of +-- @UnsafePrimitives@, which forces evaluation for types that provide an +-- @NFData@ instance. Direct use of the underlying unsafe primitives (from +-- the distributed-process library) without @NFData@ instances is unsupported. +-- +-- IMPORTANT NOTE: As per the platform documentation, it is not possible to +-- /guarantee/ that an @NFData@ instance will force evaluation in the same way +-- that a @Binary@ instance would (when encoding to a byte string). Please read +-- the unsafe primitives documentation carefully and make sure you know what +-- you're doing. You have been warned. +-- +-- See "Control.Distributed.Process.Extras". +-- See "Control.Distributed.Process.Extras.UnsafePrimitives". +-- See "Control.Distributed.Process.UnsafePrimitives". +----------------------------------------------------------------------------- + +-- TODO: This module is basically cut+paste duplicaton of the /safe/ Client - fix +-- Caveats... we've got to support two different type constraints, somehow, so +-- that the correct implementation gets used depending on whether or not we're +-- passing NFData or just Binary instances... + +module Control.Distributed.Process.ManagedProcess.UnsafeClient + ( -- * Unsafe variants of the Client API + sendControlMessage + , shutdown + , call + , safeCall + , tryCall + , callTimeout + , flushPendingCalls + , callAsync + , cast + , callChan + , syncCallChan + , syncSafeCallChan + ) where + +import Control.Distributed.Process + ( Process + , ProcessId + , ReceivePort + , newChan + , matchChan + , match + , die + , terminate + , receiveTimeout + , unsafeSendChan + ) +import Control.Distributed.Process.Async + ( Async + , async + , task + ) +import Control.Distributed.Process.Extras + ( awaitResponse + , Addressable + , Routable(..) + , NFSerializable + , ExitReason + , Shutdown(..) + ) +import Control.Distributed.Process.ManagedProcess.Internal.Types + ( Message(CastMessage, ChanMessage) + , CallResponse(..) + , ControlPort(..) + , unsafeInitCall + , waitResponse + ) +import Control.Distributed.Process.Extras.Time + ( TimeInterval + , asTimeout + ) +import Control.Distributed.Process.Serializable hiding (SerializableDict) +import Data.Maybe (fromJust) + +-- | Send a control message over a 'ControlPort'. This version of +-- @shutdown@ uses /unsafe primitives/. +-- +sendControlMessage :: Serializable m => ControlPort m -> m -> Process () +sendControlMessage cp m = unsafeSendChan (unPort cp) (CastMessage m) + +-- | Send a signal instructing the process to terminate. This version of +-- @shutdown@ uses /unsafe primitives/. +shutdown :: ProcessId -> Process () +shutdown pid = cast pid Shutdown + +-- | Make a synchronous call - uses /unsafe primitives/. +call :: forall s a b . (Addressable s, NFSerializable a, NFSerializable b) + => s -> a -> Process b +call sid msg = unsafeInitCall sid msg >>= waitResponse Nothing >>= decodeResult + where decodeResult (Just (Right r)) = return r + decodeResult (Just (Left err)) = die err + decodeResult Nothing {- the impossible happened -} = terminate + +-- | Safe version of 'call' that returns information about the error +-- if the operation fails - uses /unsafe primitives/. +safeCall :: forall s a b . (Addressable s, NFSerializable a, NFSerializable b) + => s -> a -> Process (Either ExitReason b) +safeCall s m = unsafeInitCall s m >>= waitResponse Nothing >>= return . fromJust + +-- | Version of 'safeCall' that returns 'Nothing' if the operation fails. +-- Uses /unsafe primitives/. +tryCall :: forall s a b . (Addressable s, NFSerializable a, NFSerializable b) + => s -> a -> Process (Maybe b) +tryCall s m = unsafeInitCall s m >>= waitResponse Nothing >>= decodeResult + where decodeResult (Just (Right r)) = return $ Just r + decodeResult _ = return Nothing + +-- | Make a synchronous call, but timeout and return @Nothing@ if a reply +-- is not received within the specified time interval - uses /unsafe primitives/. +-- +callTimeout :: forall s a b . (Addressable s, NFSerializable a, NFSerializable b) + => s -> a -> TimeInterval -> Process (Maybe b) +callTimeout s m d = unsafeInitCall s m >>= waitResponse (Just d) >>= decodeResult + where decodeResult :: (NFSerializable b) + => Maybe (Either ExitReason b) + -> Process (Maybe b) + decodeResult Nothing = return Nothing + decodeResult (Just (Right result)) = return $ Just result + decodeResult (Just (Left reason)) = die reason + +flushPendingCalls :: forall b . (NFSerializable b) + => TimeInterval + -> (b -> Process b) + -> Process (Maybe b) +flushPendingCalls d proc = do + receiveTimeout (asTimeout d) [ + match (\(CallResponse (m :: b) _) -> proc m) + ] + +-- | Invokes 'call' /out of band/, and returns an "async handle." +-- Uses /unsafe primitives/. +-- +callAsync :: forall s a b . (Addressable s, NFSerializable a, NFSerializable b) + => s -> a -> Process (Async b) +callAsync server msg = async $ task $ call server msg + +-- | Sends a /cast/ message to the server identified by @server@ - uses /unsafe primitives/. +-- +cast :: forall a m . (Addressable a, NFSerializable m) + => a -> m -> Process () +cast server msg = unsafeSendTo server ((CastMessage msg) :: Message m ()) + +-- | Sends a /channel/ message to the server and returns a @ReceivePort@ - uses /unsafe primitives/. +callChan :: forall s a b . (Addressable s, NFSerializable a, NFSerializable b) + => s -> a -> Process (ReceivePort b) +callChan server msg = do + (sp, rp) <- newChan + unsafeSendTo server ((ChanMessage msg sp) :: Message a b) + return rp + +syncCallChan :: forall s a b . (Addressable s, NFSerializable a, NFSerializable b) + => s -> a -> Process b +syncCallChan server msg = do + r <- syncSafeCallChan server msg + case r of + Left e -> die e + Right r' -> return r' + +syncSafeCallChan :: forall s a b . (Addressable s, NFSerializable a, NFSerializable b) + => s -> a -> Process (Either ExitReason b) +syncSafeCallChan server msg = do + rp <- callChan server msg + awaitResponse server [ matchChan rp (return . Right) ] + diff --git a/tests/TestManagedProcess.hs b/tests/TestManagedProcess.hs new file mode 100644 --- /dev/null +++ b/tests/TestManagedProcess.hs @@ -0,0 +1,296 @@ +{-# LANGUAGE ScopedTypeVariables #-} +{-# LANGUAGE DeriveDataTypeable #-} +{-# LANGUAGE BangPatterns #-} +{-# LANGUAGE TemplateHaskell #-} + +module Main where + +import Control.Concurrent.MVar +import Control.Exception (SomeException) +import Control.Distributed.Process hiding (call) +import Control.Distributed.Process.Node +import Control.Distributed.Process.Extras hiding (__remoteTable, monitor, send, nsend) +import Control.Distributed.Process.ManagedProcess +import Control.Distributed.Process.Tests.Internal.Utils +import Control.Distributed.Process.Extras.Time +import Control.Distributed.Process.Serializable() + +import MathsDemo +import Counter +import qualified SafeCounter as SafeCounter + +#if ! MIN_VERSION_base(4,6,0) +import Prelude hiding (catch) +#endif + +import Test.Framework (Test, testGroup) +import Test.Framework.Providers.HUnit (testCase) +import TestUtils +import ManagedProcessCommon + +import qualified Network.Transport as NT +import Control.Monad (void) + +-- utilities + +server :: Process (ProcessId, (MVar ExitReason)) +server = mkServer Terminate + +mkServer :: UnhandledMessagePolicy + -> Process (ProcessId, (MVar ExitReason)) +mkServer policy = + let s = standardTestServer policy + in do + exitReason <- liftIO $ newEmptyMVar + pid <- spawnLocal $ do + catch ((serve () (statelessInit Infinity) s >> stash exitReason ExitNormal) + `catchesExit` [ + (\_ msg -> do + mEx <- unwrapMessage msg :: Process (Maybe ExitReason) + case mEx of + Nothing -> return Nothing + Just r -> stash exitReason r >>= return . Just + ) + ]) + (\(e :: SomeException) -> stash exitReason $ ExitOther (show e)) + return (pid, exitReason) + +explodingServer :: ProcessId + -> Process (ProcessId, MVar ExitReason) +explodingServer pid = + let srv = explodingTestProcess pid + in do + exitReason <- liftIO $ newEmptyMVar + spid <- spawnLocal $ do + catch (serve () (statelessInit Infinity) srv >> stash exitReason ExitNormal) + (\(e :: SomeException) -> stash exitReason $ ExitOther (show e)) + return (spid, exitReason) + +testCallReturnTypeMismatchHandling :: TestResult Bool -> Process () +testCallReturnTypeMismatchHandling result = + let procDef = statelessProcess { + apiHandlers = [ + handleCall (\s (m :: String) -> reply m s) + ] + , unhandledMessagePolicy = Terminate + } in do + pid <- spawnLocal $ serve () (statelessInit Infinity) procDef + res <- safeCall pid "hello buddy" :: Process (Either ExitReason ()) + case res of + Left (ExitOther _) -> stash result True + _ -> stash result False + +testChannelBasedService :: TestResult Bool -> Process () +testChannelBasedService result = + let procDef = statelessProcess { + apiHandlers = [ + handleRpcChan (\s p (m :: String) -> + replyChan p m >> continue s) + ] + } in do + pid <- spawnLocal $ serve () (statelessInit Infinity) procDef + echo <- syncCallChan pid "hello" + stash result (echo == "hello") + kill pid "done" + +-- MathDemo tests + +testAdd :: ProcessId -> TestResult Double -> Process () +testAdd pid result = add pid 10 10 >>= stash result + +testBadAdd :: ProcessId -> TestResult (Either ExitReason Int) -> Process () +testBadAdd pid result = safeCall pid (Add 10 10) >>= stash result + +testDivByZero :: ProcessId -> TestResult (Either DivByZero Double) -> Process () +testDivByZero pid result = divide pid 125 0 >>= stash result + +-- SafeCounter tests + +testSafeCounterCurrentState :: ProcessId -> TestResult Int -> Process () +testSafeCounterCurrentState pid result = + SafeCounter.getCount pid >>= stash result + +testSafeCounterIncrement :: ProcessId -> TestResult Int -> Process () +testSafeCounterIncrement pid result = do + 5 <- SafeCounter.getCount pid + SafeCounter.resetCount pid + 1 <- SafeCounter.incCount pid + 2 <- SafeCounter.incCount pid + SafeCounter.getCount pid >>= stash result + +-- Counter tests + +testCounterCurrentState :: TestResult Int -> Process () +testCounterCurrentState result = do + pid <- Counter.startCounter 5 + getCount pid >>= stash result + +testCounterIncrement :: TestResult Bool -> Process () +testCounterIncrement result = do + pid <- Counter.startCounter 1 + n <- getCount pid + 2 <- incCount pid + 3 <- incCount pid + getCount pid >>= \n' -> stash result (n' == (n + 2)) + +testCounterExceedsLimit :: TestResult Bool -> Process () +testCounterExceedsLimit result = do + pid <- Counter.startCounter 1 + mref <- monitor pid + + -- exceed the limit + 9 `times` (void $ incCount pid) + + -- this time we should fail + _ <- (incCount pid) + `catchExit` \_ (_ :: ExitReason) -> return 0 + + r <- receiveWait [ + matchIf (\(ProcessMonitorNotification ref _ _) -> ref == mref) + (\(ProcessMonitorNotification _ _ r') -> return r') + ] + stash result (r /= DiedNormal) + +tests :: NT.Transport -> IO [Test] +tests transport = do + localNode <- newLocalNode transport initRemoteTable + mpid <- newEmptyMVar + _ <- forkProcess localNode $ launchMathServer >>= stash mpid + pid <- takeMVar mpid + scpid <- newEmptyMVar + _ <- forkProcess localNode $ SafeCounter.startCounter 5 >>= stash scpid + safeCounter <- takeMVar scpid + return [ + testGroup "basic server functionality" [ + testCase "basic call with explicit server reply" + (delayedAssertion + "expected a response from the server" + localNode (Just "foo") (testBasicCall $ wrap server)) + , testCase "basic (unsafe) call with explicit server reply" + (delayedAssertion + "expected a response from the server" + localNode (Just "foo") (testUnsafeBasicCall $ wrap server)) + , testCase "basic call with implicit server reply" + (delayedAssertion + "expected n * 2 back from the server" + localNode (Just 4) (testBasicCall_ $ wrap server)) + , testCase "basic (unsafe) call with implicit server reply" + (delayedAssertion + "expected n * 2 back from the server" + localNode (Just 4) (testUnsafeBasicCall_ $ wrap server)) + , testCase "basic cast with manual send and explicit server continue" + (delayedAssertion + "expected pong back from the server" + localNode (Just "pong") (testBasicCast $ wrap server)) + , testCase "basic (unsafe) cast with manual send and explicit server continue" + (delayedAssertion + "expected pong back from the server" + localNode (Just "pong") (testUnsafeBasicCast $ wrap server)) + , testCase "cast and explicit server timeout" + (delayedAssertion + "expected the server to stop after the timeout" + localNode (Just $ ExitOther "timeout") (testControlledTimeout $ wrap server)) + , testCase "(unsafe) cast and explicit server timeout" + (delayedAssertion + "expected the server to stop after the timeout" + localNode (Just $ ExitOther "timeout") (testUnsafeControlledTimeout $ wrap server)) + , testCase "unhandled input when policy = Terminate" + (delayedAssertion + "expected the server to stop upon receiving unhandled input" + localNode (Just $ ExitOther "UnhandledInput") + (testTerminatePolicy $ wrap server)) + , testCase "(unsafe) unhandled input when policy = Terminate" + (delayedAssertion + "expected the server to stop upon receiving unhandled input" + localNode (Just $ ExitOther "UnhandledInput") + (testUnsafeTerminatePolicy $ wrap server)) + , testCase "unhandled input when policy = Drop" + (delayedAssertion + "expected the server to ignore unhandled input and exit normally" + localNode Nothing (testDropPolicy $ wrap (mkServer Drop))) + , testCase "(unsafe) unhandled input when policy = Drop" + (delayedAssertion + "expected the server to ignore unhandled input and exit normally" + localNode Nothing (testUnsafeDropPolicy $ wrap (mkServer Drop))) + , testCase "unhandled input when policy = DeadLetter" + (delayedAssertion + "expected the server to forward unhandled messages" + localNode (Just ("UNSOLICITED_MAIL", 500 :: Int)) + (testDeadLetterPolicy $ \p -> mkServer (DeadLetter p))) + , testCase "(unsafe) unhandled input when policy = DeadLetter" + (delayedAssertion + "expected the server to forward unhandled messages" + localNode (Just ("UNSOLICITED_MAIL", 500 :: Int)) + (testUnsafeDeadLetterPolicy $ \p -> mkServer (DeadLetter p))) + , testCase "incoming messages are ignored whilst hibernating" + (delayedAssertion + "expected the server to remain in hibernation" + localNode True (testHibernation $ wrap server)) + , testCase "(unsafe) incoming messages are ignored whilst hibernating" + (delayedAssertion + "expected the server to remain in hibernation" + localNode True (testUnsafeHibernation $ wrap server)) + , testCase "long running call cancellation" + (delayedAssertion "expected to get AsyncCancelled" + localNode True (testKillMidCall $ wrap server)) + , testCase "(unsafe) long running call cancellation" + (delayedAssertion "expected to get AsyncCancelled" + localNode True (testUnsafeKillMidCall $ wrap server)) + , testCase "simple exit handling" + (delayedAssertion "expected handler to catch exception and continue" + localNode Nothing (testSimpleErrorHandling $ explodingServer)) + , testCase "(unsafe) simple exit handling" + (delayedAssertion "expected handler to catch exception and continue" + localNode Nothing (testUnsafeSimpleErrorHandling $ explodingServer)) + , testCase "alternative exit handlers" + (delayedAssertion "expected handler to catch exception and continue" + localNode Nothing (testAlternativeErrorHandling $ explodingServer)) + , testCase "(unsafe) alternative exit handlers" + (delayedAssertion "expected handler to catch exception and continue" + localNode Nothing (testUnsafeAlternativeErrorHandling $ explodingServer)) + ] + , testGroup "math server examples" [ + testCase "error (Left) returned from x / 0" + (delayedAssertion + "expected the server to return DivByZero" + localNode (Left DivByZero) (testDivByZero pid)) + , testCase "10 + 10 = 20" + (delayedAssertion + "expected the server to return DivByZero" + localNode 20 (testAdd pid)) + , testCase "10 + 10 does not evaluate to 10 :: Int at all!" + (delayedAssertion + "expected the server to return ExitOther..." + localNode + (Left $ ExitOther $ "DiedException \"exit-from=" ++ (show pid) ++ "\"") + (testBadAdd pid)) + ] + , testGroup "counter server examples" [ + testCase "initial counter state = 5" + (delayedAssertion + "expected the server to return the initial state of 5" + localNode 5 testCounterCurrentState) + , testCase "increment counter twice" + (delayedAssertion + "expected the server to return the incremented state as 7" + localNode True testCounterIncrement) + , testCase "exceed counter limits" + (delayedAssertion + "expected the server to terminate once the limit was exceeded" + localNode True testCounterExceedsLimit) + ] + , testGroup "safe counter examples" [ + testCase "initial counter state = 5" + (delayedAssertion + "expected the server to return the initial state of 5" + localNode 5 (testSafeCounterCurrentState safeCounter)) + , testCase "increment counter twice" + (delayedAssertion + "expected the server to return the incremented state as 7" + localNode 2 (testSafeCounterIncrement safeCounter)) + ] + ] + +main :: IO () +main = testMain $ tests + diff --git a/tests/TestPrioritisedProcess.hs b/tests/TestPrioritisedProcess.hs new file mode 100644 --- /dev/null +++ b/tests/TestPrioritisedProcess.hs @@ -0,0 +1,221 @@ +{-# LANGUAGE ScopedTypeVariables #-} +{-# LANGUAGE DeriveDataTypeable #-} +{-# LANGUAGE BangPatterns #-} +{-# LANGUAGE DeriveGeneric #-} + +-- NB: this module contains tests for the GenProcess /and/ GenServer API. + +module Main where + +import Control.Concurrent.MVar +import Control.Exception (SomeException) +import Control.DeepSeq (NFData) +import Control.Distributed.Process hiding (call, send) +import Control.Distributed.Process.Node +import Control.Distributed.Process.Extras hiding (__remoteTable) +import Control.Distributed.Process.Async +import Control.Distributed.Process.ManagedProcess +import Control.Distributed.Process.Tests.Internal.Utils +import Control.Distributed.Process.Extras.Time +import Control.Distributed.Process.Extras.Timer +import Control.Distributed.Process.Serializable() + +import Data.Binary +import Data.Either (rights) +import Data.Typeable (Typeable) + +#if ! MIN_VERSION_base(4,6,0) +import Prelude hiding (catch) +#endif + +import Test.Framework (Test, testGroup) +import Test.Framework.Providers.HUnit (testCase) +import TestUtils +import ManagedProcessCommon + +import qualified Network.Transport as NT + +import GHC.Generics (Generic) + +-- utilities + +server :: Process (ProcessId, (MVar ExitReason)) +server = mkServer Terminate + +mkServer :: UnhandledMessagePolicy + -> Process (ProcessId, (MVar ExitReason)) +mkServer policy = + let s = standardTestServer policy + p = s `prioritised` ([] :: [DispatchPriority ()]) + in do + exitReason <- liftIO $ newEmptyMVar + pid <- spawnLocal $ do + catch ((pserve () (statelessInit Infinity) p >> stash exitReason ExitNormal) + `catchesExit` [ + (\_ msg -> do + mEx <- unwrapMessage msg :: Process (Maybe ExitReason) + case mEx of + Nothing -> return Nothing + Just r -> stash exitReason r >>= return . Just + ) + ]) + (\(e :: SomeException) -> stash exitReason $ ExitOther (show e)) + return (pid, exitReason) + +explodingServer :: ProcessId + -> Process (ProcessId, MVar ExitReason) +explodingServer pid = + let srv = explodingTestProcess pid + pSrv = srv `prioritised` ([] :: [DispatchPriority s]) + in do + exitReason <- liftIO $ newEmptyMVar + spid <- spawnLocal $ do + catch (pserve () (statelessInit Infinity) pSrv >> stash exitReason ExitNormal) + (\(e :: SomeException) -> stash exitReason $ ExitOther (show e)) + return (spid, exitReason) + +data GetState = GetState + deriving (Typeable, Generic, Show, Eq) +instance Binary GetState where +instance NFData GetState where + +data MyAlarmSignal = MyAlarmSignal + deriving (Typeable, Generic, Show, Eq) +instance Binary MyAlarmSignal where +instance NFData MyAlarmSignal where + +mkPrioritisedServer :: Process ProcessId +mkPrioritisedServer = + let p = procDef `prioritised` ([ + prioritiseInfo_ (\MyAlarmSignal -> setPriority 10) + , prioritiseCast_ (\(_ :: String) -> setPriority 2) + , prioritiseCall_ (\(cmd :: String) -> (setPriority (length cmd)) :: Priority ()) + ] :: [DispatchPriority [Either MyAlarmSignal String]] + ) :: PrioritisedProcessDefinition [(Either MyAlarmSignal String)] + in spawnLocal $ pserve () (initWait Infinity) p + where + initWait :: Delay + -> InitHandler () [Either MyAlarmSignal String] + initWait d () = do + () <- expect + return $ InitOk [] d + + procDef :: ProcessDefinition [(Either MyAlarmSignal String)] + procDef = + defaultProcess { + apiHandlers = [ + handleCall (\s GetState -> reply (reverse s) s) + , handleCall (\s (cmd :: String) -> reply () ((Right cmd):s)) + , handleCast (\s (cmd :: String) -> continue ((Right cmd):s)) + ] + , infoHandlers = [ + handleInfo (\s (sig :: MyAlarmSignal) -> continue ((Left sig):s)) + ] + , unhandledMessagePolicy = Drop + , timeoutHandler = \_ _ -> stop $ ExitOther "timeout" + } :: ProcessDefinition [(Either MyAlarmSignal String)] + +-- test cases + +testInfoPrioritisation :: TestResult Bool -> Process () +testInfoPrioritisation result = do + pid <- mkPrioritisedServer + -- the server (pid) is configured to wait for () during its init + -- so we can fill up its mailbox with String messages, and verify + -- that the alarm signal (which is prioritised *above* these) + -- actually gets processed first despite the delivery order + cast pid "hello" + cast pid "prioritised" + cast pid "world" + -- note that these have to be a "bare send" + send pid MyAlarmSignal + -- tell the server it can move out of init and start processing messages + send pid () + st <- call pid GetState :: Process [Either MyAlarmSignal String] + -- the result of GetState is a list of messages in reverse insertion order + case head st of + Left MyAlarmSignal -> stash result True + _ -> stash result False + +testCallPrioritisation :: TestResult Bool -> Process () +testCallPrioritisation result = do + pid <- mkPrioritisedServer + asyncRefs <- (mapM (callAsync pid) + ["first", "the longest", "commands", "we do prioritise"]) + :: Process [Async ()] + -- NB: This sleep is really important - the `init' function is waiting + -- (selectively) on the () signal to go, and if it receives this *before* + -- the async worker has had a chance to deliver the longest string message, + -- our test will fail. Such races are /normal/ given that the async worker + -- runs in another process and delivery order between multiple processes + -- is undefined (and in practise, paritally depenendent on the scheduler) + sleep $ seconds 1 + send pid () + mapM wait asyncRefs :: Process [AsyncResult ()] + st <- call pid GetState :: Process [Either MyAlarmSignal String] + let ms = rights st + stash result $ ms == ["we do prioritise", "the longest", "commands", "first"] + +tests :: NT.Transport -> IO [Test] +tests transport = do + localNode <- newLocalNode transport initRemoteTable + return [ + testGroup "basic server functionality matches un-prioritised processes" [ + testCase "basic call with explicit server reply" + (delayedAssertion + "expected a response from the server" + localNode (Just "foo") (testBasicCall $ wrap server)) + , testCase "basic call with implicit server reply" + (delayedAssertion + "expected n * 2 back from the server" + localNode (Just 4) (testBasicCall_ $ wrap server)) + , testCase "basic cast with manual send and explicit server continue" + (delayedAssertion + "expected pong back from the server" + localNode (Just "pong") (testBasicCast $ wrap server)) + , testCase "cast and explicit server timeout" + (delayedAssertion + "expected the server to stop after the timeout" + localNode (Just $ ExitOther "timeout") (testControlledTimeout $ wrap server)) + , testCase "unhandled input when policy = Terminate" + (delayedAssertion + "expected the server to stop upon receiving unhandled input" + localNode (Just $ ExitOther "UnhandledInput") + (testTerminatePolicy $ wrap server)) + , testCase "unhandled input when policy = Drop" + (delayedAssertion + "expected the server to ignore unhandled input and exit normally" + localNode Nothing (testDropPolicy $ wrap (mkServer Drop))) + , testCase "unhandled input when policy = DeadLetter" + (delayedAssertion + "expected the server to forward unhandled messages" + localNode (Just ("UNSOLICITED_MAIL", 500 :: Int)) + (testDeadLetterPolicy $ \p -> mkServer (DeadLetter p))) + , testCase "incoming messages are ignored whilst hibernating" + (delayedAssertion + "expected the server to remain in hibernation" + localNode True (testHibernation $ wrap server)) + , testCase "long running call cancellation" + (delayedAssertion "expected to get AsyncCancelled" + localNode True (testKillMidCall $ wrap server)) + , testCase "simple exit handling" + (delayedAssertion "expected handler to catch exception and continue" + localNode Nothing (testSimpleErrorHandling $ explodingServer)) + , testCase "alternative exit handlers" + (delayedAssertion "expected handler to catch exception and continue" + localNode Nothing (testAlternativeErrorHandling $ explodingServer)) + ] + , testGroup "Prioritised Mailbox Handling" [ + testCase "Info Message Prioritisation" + (delayedAssertion "expected the info handler to be prioritised" + localNode True testInfoPrioritisation) + , testCase "Call Message Prioritisation" + (delayedAssertion "expected the longest strings to be prioritised" + localNode True testCallPrioritisation) + ] + ] + +main :: IO () +main = testMain $ tests + +