distributed-process-client-server (empty) → 0.1.1
raw patch · 13 files changed
+3262/−0 lines, 13 filesdep +HUnitdep +ansi-terminaldep +basesetup-changed
Dependencies added: HUnit, ansi-terminal, base, binary, containers, data-accessor, deepseq, derive, distributed-process, distributed-process-async, distributed-process-client-server, distributed-process-extras, distributed-process-tests, fingertree, ghc-prim, hashable, mtl, network, network-transport, network-transport-tcp, rematch, stm, template-haskell, test-framework, test-framework-hunit, time, transformers, uniplate, unordered-containers
Files
- LICENCE +30/−0
- Setup.lhs +3/−0
- distributed-process-client-server.cabal +129/−0
- src/Control/Distributed/Process/ManagedProcess.hs +525/−0
- src/Control/Distributed/Process/ManagedProcess/Client.hs +164/−0
- src/Control/Distributed/Process/ManagedProcess/Internal/GenProcess.hs +328/−0
- src/Control/Distributed/Process/ManagedProcess/Internal/Types.hs +415/−0
- src/Control/Distributed/Process/ManagedProcess/Server.hs +600/−0
- src/Control/Distributed/Process/ManagedProcess/Server/Priority.hs +95/−0
- src/Control/Distributed/Process/ManagedProcess/Server/Restricted.hs +275/−0
- src/Control/Distributed/Process/ManagedProcess/UnsafeClient.hs +181/−0
- tests/TestManagedProcess.hs +296/−0
- tests/TestPrioritisedProcess.hs +221/−0
+ LICENCE view
@@ -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.
+ Setup.lhs view
@@ -0,0 +1,3 @@+#!/usr/bin/env runhaskell+> import Distribution.Simple+> main = defaultMain
+ distributed-process-client-server.cabal view
@@ -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
+ src/Control/Distributed/Process/ManagedProcess.hs view
@@ -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 <watson.timothy@gmail.com>+-- 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 - <http://www.erlang.org/doc/man/gen_server.html>.+--+-- 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+
+ src/Control/Distributed/Process/ManagedProcess/Client.hs view
@@ -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 <watson.timothy@gmail.com>+-- 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) ]+
+ src/Control/Distributed/Process/ManagedProcess/Internal/GenProcess.hs view
@@ -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)+
+ src/Control/Distributed/Process/ManagedProcess/Internal/Types.hs view
@@ -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)+
+ src/Control/Distributed/Process/ManagedProcess/Server.hs view
@@ -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 <watson.timothy@gmail.com>+-- 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+
+ src/Control/Distributed/Process/ManagedProcess/Server/Priority.hs view
@@ -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+
+ src/Control/Distributed/Process/ManagedProcess/Server/Restricted.hs view
@@ -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 <watson.timothy@gmail.com>+-- 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+
+ src/Control/Distributed/Process/ManagedProcess/UnsafeClient.hs view
@@ -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 <watson.timothy@gmail.com>+-- 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) ]+
+ tests/TestManagedProcess.hs view
@@ -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+
+ tests/TestPrioritisedProcess.hs view
@@ -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++