distributed-process-client-server 0.1.3.2 → 0.2.0

raw patch · 17 files changed

+3238/−694 lines, 17 filesdep +distributed-process-systestdep +exceptionsdep −data-accessordep −distributed-process-testsdep ~basedep ~deepseqdep ~distributed-processPVP ok

version bump matches the API change (PVP)

Dependencies added: distributed-process-systest, exceptions

Dependencies removed: data-accessor, distributed-process-tests

Dependency ranges changed: base, deepseq, distributed-process, distributed-process-async, distributed-process-extras

API changes (from Hackage documentation)

Files

• distributed-process-client-server.cabal +36/−26
• src/Control/Distributed/Process/ManagedProcess.hs +326/−126
• src/Control/Distributed/Process/ManagedProcess/Client.hs +102/−10
• src/Control/Distributed/Process/ManagedProcess/Internal/GenProcess.hs +661/−199
• src/Control/Distributed/Process/ManagedProcess/Internal/Types.hs +355/−89
• src/Control/Distributed/Process/ManagedProcess/Server.hs +179/−115
• src/Control/Distributed/Process/ManagedProcess/Server/Priority.hs +242/−12
• src/Control/Distributed/Process/ManagedProcess/Server/Restricted.hs +10/−15
• src/Control/Distributed/Process/ManagedProcess/Timer.hs +187/−0
• src/Control/Distributed/Process/ManagedProcess/UnsafeClient.hs +21/−5
• tests/Counter.hs +94/−0
• tests/ManagedProcessCommon.hs +372/−0
• tests/MathsDemo.hs +59/−0
• tests/SafeCounter.hs +104/−0
• tests/TestManagedProcess.hs +165/−82
• tests/TestPrioritisedProcess.hs +276/−15
• tests/TestUtils.hs +49/−0

@@ -1,13 +1,13 @@ name:           distributed-process-client-server-version:        0.1.3.2+version:        0.2.0 cabal-version:  >=1.8 build-type:     Simple license:        BSD3 license-file:   LICENCE stability:      experimental-Copyright:      Tim Watson 2012 - 2013+Copyright:      Tim Watson 2012 - 2017 Author:         Tim Watson-Maintainer:     Facundo Domínguez <facundo.dominguez@tweag.io>+Maintainer:     Tim Watson <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@@ -17,7 +17,7 @@                 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+Tested-With:    GHC==7.10.3 GHC==8.0.1 GHC==8.0.2 data-dir:       ""  source-repository head@@ -26,21 +26,21 @@  library   build-depends:-                   base >= 4.4 && < 5,-                   data-accessor >= 0.2.2.3,-                   distributed-process >= 0.5.2 && < 0.7,-                   distributed-process-extras >= 0.2.0 && < 0.3,-                   distributed-process-async >= 0.2.1 && < 0.3,-                   binary >= 0.6.3.0 && < 0.8,-                   deepseq >= 1.3.0.1 && < 1.5,+                   base >= 4.8.2.0 && < 5,+                   distributed-process >= 0.6.6 && < 0.7,+                   distributed-process-extras >= 0.3.1 && < 0.4,+                   distributed-process-async >= 0.2.4 && < 0.3,+                   binary >= 0.6.3.0 && < 0.9,+                   deepseq >= 1.3.0.1 && < 1.6,                    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.6,-                   transformers+                   time > 1.4 && < 1.8,+                   transformers,+                   exceptions >= 0.5   if impl(ghc <= 7.5)     Build-Depends:   template-haskell == 2.7.0.0,                      derive == 2.5.5,@@ -55,7 +55,8 @@                    Control.Distributed.Process.ManagedProcess.UnsafeClient,                    Control.Distributed.Process.ManagedProcess.Server,                    Control.Distributed.Process.ManagedProcess.Server.Priority,-                   Control.Distributed.Process.ManagedProcess.Server.Restricted+                   Control.Distributed.Process.ManagedProcess.Server.Restricted,+                   Control.Distributed.Process.ManagedProcess.Timer   other-modules:                    Control.Distributed.Process.ManagedProcess.Internal.Types,                    Control.Distributed.Process.ManagedProcess.Internal.GenProcess@@ -67,16 +68,16 @@                    base >= 4.4 && < 5,                    ansi-terminal >= 0.5 && < 0.7,                    containers,-                   distributed-process >= 0.5.2 && < 0.7,-                   distributed-process-extras >= 0.2.0 && < 0.3,-                   distributed-process-async >= 0.2.1 && < 0.3,+                   distributed-process >= 0.6.6 && < 0.7,+                   distributed-process-extras >= 0.3.1 && < 0.4,+                   distributed-process-async >= 0.2.4 && < 0.3,                    distributed-process-client-server,-                   distributed-process-tests >= 0.4.2 && < 0.5,+                   distributed-process-systest >= 0.1.1 && < 0.2,                    network-transport >= 0.4 && < 0.5,                    mtl,                    fingertree < 0.2,                    network-transport-tcp >= 0.4 && < 0.6,-                   binary >= 0.6.3.0 && < 0.8,+                   binary >= 0.6.3.0 && < 0.9,                    deepseq >= 1.3.0.1 && < 1.5,                    network >= 2.3 && < 2.7,                    HUnit >= 1.2 && < 2,@@ -85,7 +86,13 @@                    test-framework-hunit,                    transformers,                    rematch >= 0.2.0.0,-                   ghc-prim+                   ghc-prim,+                   exceptions >= 0.5+  other-modules:   Counter,+                   ManagedProcessCommon,+                   MathsDemo,+                   SafeCounter,+                   TestUtils   hs-source-dirs:                    tests   ghc-options:     -Wall -threaded -rtsopts -with-rtsopts=-N -fno-warn-unused-do-bind@@ -99,16 +106,16 @@                    base >= 4.4 && < 5,                    ansi-terminal >= 0.5 && < 0.7,                    containers,-                   distributed-process >= 0.5.2 && < 0.7,-                   distributed-process-extras >= 0.2.0 && < 0.3,-                   distributed-process-async >= 0.2.1 && < 0.3,+                   distributed-process >= 0.6.6 && < 0.7,+                   distributed-process-extras >= 0.3.1 && < 0.4,+                   distributed-process-async >= 0.2.4 && < 0.3,                    distributed-process-client-server,-                   distributed-process-tests >= 0.4.2 && < 0.5,+                   distributed-process-systest >= 0.1.1 && < 0.2,                    network-transport >= 0.4 && < 0.5,                    mtl,                    fingertree < 0.2,                    network-transport-tcp >= 0.4 && < 0.6,-                   binary >= 0.6.3.0 && < 0.8,+                   binary >= 0.6.3.0 && < 0.9,                    deepseq >= 1.3.0.1 && < 1.5,                    network >= 2.3 && < 2.7,                    HUnit >= 1.2 && < 2,@@ -117,7 +124,10 @@                    test-framework-hunit,                    transformers,                    rematch >= 0.2.0.0,-                   ghc-prim+                   ghc-prim,+                   exceptions >= 0.5+  other-modules:   ManagedProcessCommon,+                   TestUtils   hs-source-dirs:                    tests   ghc-options:     -Wall -threaded -rtsopts -with-rtsopts=-N -fno-warn-unused-do-bind

@@ -1,13 +1,10 @@-{-# LANGUAGE DeriveDataTypeable         #-} {-# LANGUAGE ExistentialQuantification  #-} {-# LANGUAGE ScopedTypeVariables        #-}-{-# LANGUAGE TemplateHaskell            #-}-{-# LANGUAGE RecordWildCards            #-}  ----------------------------------------------------------------------------- -- | -- Module      :  Control.Distributed.Process.ManagedProcess--- Copyright   :  (c) Tim Watson 2012+-- Copyright   :  (c) Tim Watson 2012 - 2017 -- License     :  BSD3 (see the file LICENSE) -- -- Maintainer  :  Tim Watson <watson.timothy@gmail.com>@@ -23,7 +20,7 @@ -- In particular, a /managed process/ will interoperate cleanly with the -- supervisor API in distributed-process-supervision. ----- [API Overview]+-- [API Overview For The Impatient] -- -- 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@@ -35,8 +32,8 @@ -- -- 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+-- 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. --@@ -45,32 +42,129 @@ -- 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 handlers are split into groups: /apiHandlers/, /infoHandlers/, and+-- /extHandlers/. ----- [The Cast/Call Protocol]+-- [Seriously, TL;DR] ----- Deliberate interactions with a /managed process/ usually fall into one of+-- Use 'serve' for a process that sits reading its mailbox and generally behaves+-- as you'd expect. Use 'pserve' and 'PrioritisedProcessDefinition' for a server+-- that manages its mailbox more comprehensively and handles errors a bit differently.+-- Both use the same client API.+--+-- DO NOT mask in handler code, unless you can guarantee it won't be long+-- running and absolutely won't block kill signals from a supervisor.+--+-- Do look at the various API offerings, as there are several, at different+-- levels of abstraction.+--+-- [Managed Process Mailboxes]+--+-- Managed processes come in two flavours, with different runtime characteristics+-- and (to some extent) semantics. These flavours are differentiated by the way+-- in which they handle the server process mailbox - all client interactions+-- remain the same.+--+-- The /vanilla/ managed process mailbox, provided by the 'serve' API, is roughly+-- akin to a tail recursive /listen/ function that calls a list of passed in+-- matchers. We might naively implement it roughly like this:+--+-- >+-- > loop :: stateT -> [(stateT -> Message -> Maybe stateT)] -> Process ()+-- > loop state handlers = do+-- >   st2 <- receiveWait $ map (\d -> handleMessage (d state)) handlers+-- >   case st2 of+-- >     Nothing -> {- we're done serving -} return ()+-- >     Just s2 -> loop s2 handlers+-- >+--+-- Obviously all the details have been ellided, but this is the essential premise+-- behind a /managed process loop/. The process keeps reading from its mailbox+-- indefinitely, until either a handler instructs it to stop, or an asynchronous+-- exception (or exit signal - in the form of an async @ProcessExitException@)+-- terminates it. This kind of mailbox has fairly intuitive runtime characteristics+-- compared to a /plain server process/ (i.e. one implemented without the use of+-- this library): messages will pile up in its mailbox whilst handlers are+-- running, and each handler will be checked against the mailbox based on the+-- type of messages it recognises. We can potentially end up scanning a very+-- large mailbox trying to match each handler, which can be a performance+-- bottleneck depending on expected traffic patterns.+--+-- For most simple server processes, this technique works well and is easy to+-- reason about a use. See the sections on error and exit handling later on for+-- more details about 'serve' based managed processes.+--+-- [Prioritised Mailboxes]+--+-- A prioritised mailbox serves two purposes. The first of these is to allow a+-- managed process author to specify that certain classes of message should be+-- prioritised by the server loop. This is achieved by draining the /real/+-- process mailbox into an internal priority queue, and running the server's+-- handlers repeatedly over its contents, which are dequeued in priority order.+-- The obvious consequence of this approach leads to the second purpose (or the+-- accidental side effect, depending on your point of view) of a prioritised+-- mailbox, which is that we avoid scanning a large mailbox when searching for+-- messages that match the handlers we anticipate running most frequently (or+-- those messages that we deem most important).+--+-- There are several consequences to this approach. One is that we do quite a bit+-- more work to manage the process mailbox behind the scenes, therefore we have+-- additional space overhead to consider (although we are also reducing the size+-- of the mailbox, so there is some counter balance here). The other is that if+-- we do not see the anticipated traffic patterns at runtime, then we might+-- spend more time attempting to prioritise infrequent messages than we would+-- have done simply receiving them! We do however, gain a degree of safety with+-- regards message loss that the 'serve' based /vanilla/ mailbox cannot offer.+-- See the sections on error and exit handling later on for more details about+-- these.+--+-- A Prioritised 'pserve' loop maintains its internal state - including the user+-- defined /server state/ - in an @IORef@, ensuring it is held consistently+-- between executions, even in the face of unhandled exceptions.+--+-- [Defining Prioritised Process Definitions]+--+-- 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 the order in which they+-- are specified in the 'ProcessDefinition'.+--+-- 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 handlers in priority order.+--+-- A prioritised process should probably be configured with a 'Priority' list to+-- be useful. Creating a prioritised process without any priorities could be a+-- potential 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'.+--+-- [The Cast and Call Protocols]+--+-- Deliberate interactions with a /managed process/ usually falls 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/+-- All expressions given to @apiHandlers@ have to conform to the /cast or 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+-- These cast and 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:+-- providing module level functions that defer to the cast or call client API,+-- giving the process author an opportunity to enforce the correct input and+-- response types. For example: -- -- @ -- {- Ask the server to add two numbers -}@@ -97,7 +191,8 @@ -- 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.+-- entire protocol to be supported in excruciating detail. For that, we would+-- want /session types/, which are beyond the scope of this library. -- -- [Handling Unexpected/Info Messages] --@@ -106,7 +201,7 @@ -- 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+-- it is desirable 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@@ -140,7 +235,7 @@ -- [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+-- then there is an explicit API for doing so. Instead of using the handler -- definition functions in this module, import the /pure/ server module instead, -- which provides a StateT based monad for building referentially transparent -- callbacks.@@ -151,12 +246,8 @@ -- [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.+-- hook into these with relative ease. Catching exceptions inside handle+-- functions is no different to ordinary exception handling in monadic code. -- -- @ --   handleCall (\\x y ->@@ -166,7 +257,9 @@ -- @ -- -- The caveats mentioned in "Control.Distributed.Process.Extras" about--- exit signal handling obviously apply here as well.+-- exit signal handling are very important here - it is strongly advised that+-- you do not catch exceptions of type @ProcessExitException@ unless you plan+-- to re-throw them again. -- -- [Structured Exit Handling] --@@ -175,11 +268,11 @@ -- 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+-- 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,+-- @(reason :: ExitReason) == ExitShutdown@, which is an of /exit signal/ used+-- explicitly by 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@/@@ -190,47 +283,93 @@ -- 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.+-- callbacks, but you are strongly advised against swallowing exceptions in+-- general, or masking, unless you have carefully considered the consequences. ----- 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.+-- [Different Mailbox Types and Exceptions: Message Loss] ----- [Prioritised Mailboxes]+-- Neither the /vanilla/ nor the /prioritised/ mailbox implementations will+-- allow you to handle arbitrary asynchronous exceptions outside of your handler+-- code. The way in which the two mailboxes handle unexpected asynchronous+-- exceptions differs significantly however. The first consideration pertains to+-- potential message loss. ----- Many processes need to prioritise certain classes of message over others,--- so two subsets of the API are given to supporting those cases.+-- Consider a plain Cloud Haskell expression such as the following: ----- 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.+-- @+--   catch (receiveWait [ match (\(m :: SomeType) -> doSomething m) ])+--         (\(e :: SomeCustomAsyncException) -> handleExFrom e pid)+-- @ ----- 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.+-- It is entirely possible that @receiveWait@ will succeed in matching a message+-- of type @SomeType@ from the mailbox and removing it, to be handed to the+-- supplied expression @doSomething@. Should an asynchronous exception arrive+-- at this moment in time, though the handler might run and allow the server to+-- recover, the message will be permanently lost. ----- 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.+-- The mailbox exposed by 'serve' operates in exactly this way, and as such it+-- is advisible to avoid swallowing asynchronous exceptions, since doing so can+-- introduce the possibility of unexpected message loss. ----- Using a prioritised process is as simple as calling 'pserve' instead of--- 'serve', and passing an initialised 'PrioritisedProcessDefinition'.+-- The prioritised mailbox exposed by 'pserve' on the other hand, does not suffer+-- this scenario. Whilst the mailbox is drained into the internal priority queue,+-- asynchronous exceptions are masked, and only once the queue has been updated+-- are they removed. In addition, it is possible to @peek@ at the priority queue+-- without removing a message, thereby ensuring that should the handler fail or+-- an asynchronous exception arrive whilst processing the message, we can resume+-- handling our message immediately upon recovering from the exception. This+-- behaviour allows the process to guarantee against message loss, whilst avoiding+-- masking within handlers, which is generally bad form (and can potentially lead+-- to zombie processes, when supervised servers refuse to respond to @kill@+-- signals whilst stuck in a long running handler). ----- [Control Channels]+-- Also note that a process' internal state is subject to the same semantics,+-- such that the arrival of an asynchronous exception (including exit signals!)+-- can lead to handlers (especially exit and shutdown handlers) running with+-- a stale version of their state. For this reason - since we cannot guarantee+-- an up to date state in the presence of these semantics - a shutdown handler+-- for a 'serve' loop will always have its state passed as @LastKnown stateT@. --+-- [Different Mailbox Types and Exceptions: Error Recovery And Shutdown]+--+-- If any asynchronous exception goes unhandled by a /vanilla/ process, the+-- server will immediately exit without running the user supplied @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 'serve'+-- 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. If your code absolutely must run its+-- termination handlers in the face of any unhandled (async) exception, consider+-- using a prioritised mailbox, which handles this. Alternatively, consider+-- arranging your processes in a supervision tree, and using a shutdown strategy+-- to ensure that siblings terminate cleanly (based off a supervisor's ordered+-- shutdown signal) in order to ensure cleanup code can run reliably.+--+-- As mentioned above, a prioritised mailbox behaves differently in the face+-- of unhandled asynchronous exceptions. Whilst 'pserve' still offers no means+-- for handling arbitrary async exceptions outside your handlers - and you should+-- avoid handling them within, to the maximum extent possible - it does execute+-- its receiving process in such a way that any unhandled exception will be+-- caught and rethrown. Because of this, and the fact that a prioritised process+-- manages its internal state in an @IORef@, shutdown handlers are guaranteed+-- to run even in the face of async exceptions. These are run with the latest+-- version of the server state available, given as @CleanShutdown stateT@ when+-- the process is terminating normally (i.e. for reasons @ExitNormal@ or+-- @ExitShutdown@), and @LastKnown stateT@ when an exception terminated the+-- server process abruptly. The latter acknowledges that we cannot guarantee+-- the exception did not interrupt us after the last handler ran and returned an+-- updated state, but prior to storing the update.+--+-- Although shutdown handlers are run even in the face of unhandled exceptions+-- (and prior to re-throwing, when there is one present), they are not run in a+-- masked state. In fact, exceptions are explicitly unmasked prior to executing+-- a handler, therefore it is possible for a shutdown handler to terminate+-- abruptly. Once again, supervision hierarchies are a better way to ensure+-- consistent cleanup occurs when valued resources are held by a process.+--+-- [Special Clients: 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@@ -301,6 +440,98 @@ -- >   sendControlMessage cp $ Request str sp -- >   receiveChan rp --+-- [Communicating with the outside world: External (STM) Input Channels]+--+-- Both client and server APIs provide a mechanism for interacting with a running+-- server process via STM. This is primarily intended for code that runs outside+-- of Cloud Haskell's /Process/ monad, but can also be used as a channel for+-- sending and/or receiving non-serializable data to or from a managed process.+-- Obviously if you attempt to do this across a remote boundary, things will go+-- spectacularly wrong. The APIs provided do not attempt to restrain this, or+-- to impose any particular scheme on the programmer, therefore you're on your+-- own when it comes to writing the /STM/ code for reading and writing data+-- between client and server.+--+-- For code running inside the /Process/ monad and passing Serializable thunks,+-- there is no real advantage to this approach, and indeed there are several+-- serious disadvantages - none of Cloud Haskell's ordering guarantees will hold+-- when passing data to and from server processes in this fashion, nor are there+-- any guarantees the runtime system can make with regards interleaving between+-- messages passed across Cloud Haskell's communication fabric vs. data shared+-- via STM. This is true even when client(s) and server(s) reside on the same+-- local node.+--+--+-- A server wishing to receive data via STM can do so using the @handleExternal@+-- API. By way of example, here is a simple echo server implemented using STM:+--+-- > demoExternal = do+-- >   inChan <- liftIO newTQueueIO+-- >   replyQ <- liftIO newTQueueIO+-- >   let procDef = statelessProcess {+-- >                     apiHandlers = [+-- >                       handleExternal+-- >                         (readTQueue inChan)+-- >                         (\s (m :: String) -> do+-- >                             liftIO $ atomically $ writeTQueue replyQ m+-- >                             continue s)+-- >                     ]+-- >                     }+-- >   let txt = "hello 2-way stm foo"+-- >   pid <- spawnLocal $ serve () (statelessInit Infinity) procDef+-- >   echoTxt <- liftIO $ do+-- >     -- firstly we write something that the server can receive+-- >     atomically $ writeTQueue inChan txt+-- >     -- then sit and wait for it to write something back to us+-- >     atomically $ readTQueue replyQ+-- >+-- >   say (show $ echoTxt == txt)+--+-- For request/reply channels such as this, a convenience based on the call API+-- is also provided, which allows the server author to write an ordinary call+-- handler, and the client author to utilise an API that monitors the server and+-- does the usual stuff you'd expect an RPC style client to do. Here is another+-- example of this in use, demonstrating the @callSTM@ and @handleCallExternal@+-- APIs in practise.+--+-- > data StmServer = StmServer { serverPid  :: ProcessId+-- >                            , writerChan :: TQueue String+-- >                            , readerChan :: TQueue String+-- >                            }+-- >+-- > instance Resolvable StmServer where+-- >   resolve = return . Just . serverPid+-- >+-- > echoStm :: StmServer -> String -> Process (Either ExitReason String)+-- > echoStm StmServer{..} = callSTM serverPid+-- >                                 (writeTQueue writerChan)+-- >                                 (readTQueue  readerChan)+-- >+-- > launchEchoServer :: CallHandler () String String -> Process StmServer+-- > launchEchoServer handler = do+-- >   (inQ, replyQ) <- liftIO $ do+-- >     cIn <- newTQueueIO+-- >     cOut <- newTQueueIO+-- >     return (cIn, cOut)+-- >+-- >   let procDef = statelessProcess {+-- >                   apiHandlers = [+-- >                     handleCallExternal+-- >                       (readTQueue inQ)+-- >                       (writeTQueue replyQ)+-- >                       handler+-- >                   ]+-- >                 }+-- >+-- >   pid <- spawnLocal $ serve () (statelessInit Infinity) procDef+-- >   return $ StmServer pid inQ replyQ+-- >+-- > testExternalCall :: TestResult Bool -> Process ()+-- > testExternalCall result = do+-- >   let txt = "hello stm-call foo"+-- >   srv <- launchEchoServer (\st (msg :: String) -> reply msg st)+-- >   echoStm srv txt >>= stash result . (== Right txt)+-- -- [Performance Considerations] -- -- The various server loops are fairly optimised, but there /is/ a definite@@ -339,83 +570,50 @@   , ProcessDefinition(..)   , PrioritisedProcessDefinition(..)   , RecvTimeoutPolicy(..)-  , Priority(..)+  , Priority()   , DispatchPriority()-  , Dispatcher()-  , DeferredDispatcher()   , ShutdownHandler   , TimeoutHandler-  , ProcessAction(..)-  , ProcessReply   , Condition+  , Action+  , ProcessAction()+  , Reply+  , ProcessReply()+  , ActionHandler   , CallHandler   , CastHandler+  , StatelessHandler+  , DeferredCallHandler+  , StatelessCallHandler+  , InfoHandler+  , ChannelHandler+  , StatelessChannelHandler   , UnhandledMessagePolicy(..)   , CallRef-  , ControlChannel()-  , ControlPort()+  , ExitState(..)+  , isCleanShutdown+  , exitState   , 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_+  , module Control.Distributed.Process.ManagedProcess.Server     -- * Control channels+  , ControlChannel()+  , ControlPort()   , 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+  , module P   ) 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 qualified Control.Distributed.Process.ManagedProcess.Server.Priority as P hiding (reject) import Control.Distributed.Process.ManagedProcess.Internal.GenProcess-import Control.Distributed.Process.ManagedProcess.Internal.Types+import Control.Distributed.Process.ManagedProcess.Internal.Types hiding (runProcess) import Control.Distributed.Process.Extras (ExitReason(..)) import Control.Distributed.Process.Extras.Time import Control.Distributed.Process.Serializable@@ -479,8 +677,9 @@     InitIgnore -> return ()   where     checkExitType :: ExitReason -> Process ()-    checkExitType ExitNormal = return ()-    checkExitType other      = die other+    checkExitType ExitNormal   = return ()+    checkExitType ExitShutdown = return ()+    checkExitType other        = die other  -- | A default 'ProcessDefinition', with no api, info or exit handler. -- The default 'timeoutHandler' simply continues, the 'shutdownHandler'@@ -489,6 +688,7 @@ defaultProcess = ProcessDefinition {     apiHandlers      = []   , infoHandlers     = []+  , externHandlers   = []   , exitHandlers     = []   , timeoutHandler   = \s _ -> continue s   , shutdownHandler  = \_ _ -> return ()@@ -501,16 +701,17 @@ prioritised :: ProcessDefinition s             -> [DispatchPriority s]             -> PrioritisedProcessDefinition s-prioritised def ps = PrioritisedProcessDefinition def ps defaultRecvTimeoutPolicy+prioritised def ps =+  PrioritisedProcessDefinition def ps [] defaultRecvTimeoutPolicy  -- | Sets the default 'recvTimeoutPolicy', which gives up after 10k reads. defaultRecvTimeoutPolicy :: RecvTimeoutPolicy-defaultRecvTimeoutPolicy = RecvCounter 10000+defaultRecvTimeoutPolicy = RecvMaxBacklog 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+defaultProcessWithPriorities = prioritised defaultProcess  -- | A basic, stateless 'ProcessDefinition'. See 'defaultProcess' for the -- default field values.@@ -522,4 +723,3 @@ -- state (i.e., unit) and the given 'Delay'. statelessInit :: Delay -> InitHandler () () statelessInit d () = return $ InitOk () d-

@@ -4,7 +4,7 @@ ----------------------------------------------------------------------------- -- | -- Module      :  Control.Distributed.Process.ManagedProcess.Client--- Copyright   :  (c) Tim Watson 2012 - 2013+-- Copyright   :  (c) Tim Watson 2012 - 2017 -- License     :  BSD3 (see the file LICENSE) -- -- Maintainer  :  Tim Watson <watson.timothy@gmail.com>@@ -28,15 +28,19 @@   , callChan   , syncCallChan   , syncSafeCallChan+  , callSTM   ) where -import Control.Distributed.Process hiding (call)+import Control.Concurrent.STM (atomically, STM)+import Control.Distributed.Process hiding (call, finally) import Control.Distributed.Process.Serializable import Control.Distributed.Process.Async hiding (check)-import Control.Distributed.Process.ManagedProcess.Internal.Types+import Control.Distributed.Process.ManagedProcess.Internal.Types hiding (liftIO) import qualified Control.Distributed.Process.ManagedProcess.Internal.Types as T+import Control.Distributed.Process.Extras.Internal.Types (resolveOrDie) import Control.Distributed.Process.Extras hiding (monitor, sendChan) import Control.Distributed.Process.Extras.Time+import Control.Monad.Catch (finally) import Data.Maybe (fromJust)  import Prelude hiding (init)@@ -60,6 +64,9 @@  -- | Make a synchronous call - will block until a reply is received. -- The calling process will exit with 'ExitReason' if the calls fails.+--+-- __NOTE: this function does not catch exceptions!__+-- call :: forall s a b . (Addressable s, Serializable a, Serializable b)                  => s -> a -> Process b call sid msg = initCall sid msg >>= waitResponse Nothing >>= decodeResult@@ -68,15 +75,44 @@         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)@.+-- if the operation fails. If the calling process dies (that is, forces itself+-- to exit such that an exit signal arises with @ExitOther String@) then+-- evaluation will return @Left exitReason@ and the explanation will be+-- stashed away as @(ExitOther String)@.+--+-- __NOTE: this function does not catch exceptions!__+--+-- The /safety/ of the name, comes from carefully handling situations in which+-- the server dies while we're waiting for a reply. Notably, exit signals from+-- other processes, kill signals, and both synchronous and asynchronous+-- exceptions can still terminate the caller abruptly. To avoid this consider+-- masking or evaluating within your own exception handling code.+-- 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+safeCall s m = do+  us <- getSelfPid+  (fmap fromJust (initCall s m >>= waitResponse Nothing) :: Process (Either ExitReason b))+    `catchesExit` [(\pid msg -> handleMessageIf msg (weFailed pid us)+                                                    (return . Left))] +  where++    weFailed a b (ExitOther _) = a == b+    weFailed _ _ _             = False+ -- | 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.+--+-- __NOTE: this function does not catch exceptions!__+--+-- In fact, this API handles fewer exceptions than it's relative, "safeCall".+-- Notably, exit signals, kill signals, and both synchronous and asynchronous+-- exceptions can still terminate the caller abruptly. To avoid this consider+-- masking or evaluating within your own exception handling code (as mentioned+-- above).+-- 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@@ -92,7 +128,7 @@ -- 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/+-- 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@@ -111,11 +147,12 @@         decodeResult (Just (Right result)) = return $ Just result         decodeResult (Just (Left reason))  = die reason +-- | Attempt to flush out any pending call responses.  flushPendingCalls :: forall b . (Serializable b)                   => TimeInterval                   -> (b -> Process b)                   -> Process (Maybe b)-flushPendingCalls d proc = do+flushPendingCalls d proc =   receiveTimeout (asTimeout d) [       match (\(CallResponse (m :: b) _) -> proc m)     ]@@ -133,7 +170,7 @@ -- cast :: forall a m . (Addressable a, Serializable m)                  => a -> m -> Process ()-cast server msg = sendTo server ((CastMessage msg) :: T.Message m ())+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@@ -142,7 +179,7 @@          => s -> a -> Process (ReceivePort b) callChan server msg = do   (sp, rp) <- newChan-  sendTo server ((ChanMessage msg sp) :: T.Message a b)+  sendTo server (ChanMessage msg sp :: T.Message a b)   return rp  -- | A synchronous version of 'callChan'.@@ -162,3 +199,58 @@   rp <- callChan server msg   awaitResponse server [ matchChan rp (return . Right) ] +-- | Manages an rpc-style interaction with a server process, using @STM@ actions+-- to read/write data. The server process is monitored for the duration of the+-- /call/. The stm write expression is passed the input, and the read expression+-- is evaluated and the result given as @Right b@ or @Left ExitReason@ if a+-- monitor signal is detected whilst waiting.+--+-- Note that the caller will exit (with @ExitOther String@) if the server+-- address is un-resolvable.+--+-- A note about scheduling and timing guarantees (or lack thereof): It is not+-- possibly to guarantee the contents of @ExitReason@ in cases where this API+-- fails due to server exits/crashes. We establish a monitor prior to evaluating+-- the stm writer action, however @monitor@ is asychronous and we've no way to+-- know whether or not the scheduler will allow monitor establishment to proceed+-- first, or the stm transaction. As a result, assuming that your server process+-- can die/fail/exit on evaluating the read end of the STM write we perform here+-- (and we assume this is very likely, since we apply no safety rules and do not+-- even worry about serializing thunks passed from the client's thread), it is+-- just as likely that in the case of failure you will see a reason such as+-- @ExitOther "DiedUnknownId"@ due to the server process crashing before the node+-- controller can establish a monitor.+--+-- As unpleasant as this is, there's little we can do about it without making+-- false assumptions about the runtime. Cloud Haskell's semantics guarantee us+-- only that we will see /some/ monitor signal in the event of a failure here.+-- To provide a more robust error handling, you can catch/trap failures in the+-- server process and return a wrapper reponse datum here instead. This will+-- /still/ be subject to the failure modes described above in cases where the+-- server process exits abnormally, but that will at least allow the caller to+-- differentiate between expected and exceptional failure conditions.+--+callSTM :: forall s a b . (Addressable s)+         => s+         -> (a -> STM ())+         -> STM b+         -> a+         -> Process (Either ExitReason b)+callSTM server writeAction readAction input = do+    -- NB: we must establish the monitor before writing, to ensure we have+    -- a valid ref such that server failure gets reported properly+    pid <- resolveOrDie server "callSTM: unresolveable address "+    mRef <- monitor pid++    liftIO $ atomically $ writeAction input++    finally (receiveWait [ matchRef mRef+                         , matchSTM readAction (return . Right)+                         ])+            (unmonitor mRef)++  where+    matchRef :: MonitorRef -> Match (Either ExitReason b)+    matchRef r = matchIf (\(ProcessMonitorNotification r' _ _) -> r == r')+                         (\(ProcessMonitorNotification _ _ d) ->+                            return (Left (ExitOther (show d))))

@@ -1,26 +1,89 @@ {-# LANGUAGE ExistentialQuantification  #-} {-# LANGUAGE ScopedTypeVariables        #-}-{-# LANGUAGE ViewPatterns               #-} {-# LANGUAGE PatternGuards              #-}+{-# LANGUAGE BangPatterns               #-}+{-# LANGUAGE RecordWildCards            #-}+{-# LANGUAGE TupleSections              #-}+{-# LANGUAGE DeriveDataTypeable         #-}+{-# LANGUAGE DeriveGeneric              #-}+{-# LANGUAGE GeneralizedNewtypeDeriving #-}+{-# LANGUAGE MultiParamTypeClasses      #-}+{-# LANGUAGE Rank2Types                 #-}  -- | This is the @Process@ implementation of a /managed process/ module Control.Distributed.Process.ManagedProcess.Internal.GenProcess-  (recvLoop, precvLoop) where+  ( recvLoop+  , precvLoop+  , currentTimeout+  , systemTimeout+  , drainTimeout+  , processState+  , processDefinition+  , processFilters+  , processUnhandledMsgPolicy+  , gets+  , getAndModifyState+  , modifyState+  , setUserTimeout+  , setProcessState+  , GenProcess+  , peek+  , push+  , addUserTimer+  , removeUserTimer+  , act+  , runAfter+  , evalAfter+  ) 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.Applicative (liftA2)+import Control.Distributed.Process+  ( match+  , matchAny+  , matchMessage+  , handleMessage+  , handleMessageIf+  , receiveTimeout+  , receiveWait+  , forward+  , catchesExit+  , catchExit+  , die+  , unsafeWrapMessage+  , Process+  , ProcessId+  , Match+  )+import qualified Control.Distributed.Process as P+  ( liftIO+  )+import Control.Distributed.Process.Internal.Types+  ( Message(..)+  , ProcessExitException(..)+  ) import Control.Distributed.Process.ManagedProcess.Server-import Control.Distributed.Process.ManagedProcess.Internal.Types-import Control.Distributed.Process.Extras.Internal.Queue.PriorityQ-  ( PriorityQ-  , enqueue-  , dequeue+  ( handleCast+  , handleExitIf+  , stop+  , continue   )-import qualified Control.Distributed.Process.Extras.Internal.Queue.PriorityQ as PriorityQ+import Control.Distributed.Process.ManagedProcess.Timer+  ( Timer(timerDelay)+  , TimerKey+  , TimedOut(..)+  , delayTimer+  , startTimer+  , stopTimer+  , matchTimeout+  , matchKey+  , matchRun+  )+import Control.Distributed.Process.ManagedProcess.Internal.Types hiding (Message)+import qualified Control.Distributed.Process.Extras.Internal.Queue.PriorityQ as Q   ( empty+  , dequeue+  , enqueue+  , peek   ) import Control.Distributed.Process.Extras   ( ExitReason(..)@@ -28,187 +91,625 @@   ) 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.Distributed.Process.Serializable (Serializable) import Control.Monad (void)-import Prelude hiding (init)+import Control.Monad.Catch+  ( mask_+  , catch+  , throwM+  , mask+  , SomeException+  )+import qualified Control.Monad.State.Strict as ST+  ( get+  )+import Data.IORef (newIORef, atomicModifyIORef')+import Data.Maybe (fromJust)+import qualified Data.Map.Strict as Map+  ( size+  , insert+  , delete+  , lookup+  , empty+  , foldrWithKey+  )  -------------------------------------------------------------------------------- -- Priority Mailbox Handling                                                  -- -------------------------------------------------------------------------------- -type Queue = PriorityQ Int P.Message-type TimeoutSpec = (Delay, Maybe (TimerRef, (STM ())))-data TimeoutAction s = Stop s ExitReason | Go Delay s+-- | Evaluate the given function over the @ProcessState s@ for the caller, and+-- return the result.+gets :: forall s a . (ProcessState s -> a) -> GenProcess s a+gets f = ST.get >>= \(s :: State s) -> liftIO $ do+  atomicModifyIORef' s $ \(s' :: ProcessState s) -> (s', f s' :: a) -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+-- | Modify our state.+modifyState :: (ProcessState s -> ProcessState s) -> GenProcess s ()+modifyState f =+  ST.get >>= \s -> liftIO $ mask_ $ do+    atomicModifyIORef' s $ \s' -> (f s', ())++-- | Modify our state and return a value (potentially from it).+getAndModifyState :: (ProcessState s -> (ProcessState s, a))+                  -> GenProcess s a+getAndModifyState f =+  ST.get >>= \s -> liftIO $ mask_ $ do+    atomicModifyIORef' s $ \s' -> f s'++-- | Set the current process state.+setProcessState :: s -> GenProcess s ()+setProcessState st' =+  modifyState $ \st@ProcessState{..} -> st { procState = st' }++-- | Set the mailbox draining timer.+setDrainTimeout :: Timer -> GenProcess s ()+setDrainTimeout t = modifyState $ \st@ProcessState{..} -> st { sysTimeout = t }++-- | Set the user timeout applied whilst a prioritised process loop is in+-- a blocking receive.+setUserTimeout :: Delay -> GenProcess s ()+setUserTimeout d =+  modifyState $ \st@ProcessState{..} -> st { usrTimeout = d }++-- | Add a /user timer/, bound to the given datum.+addUserTimer :: Timer -> Message -> GenProcess s TimerKey+addUserTimer t m =+  getAndModifyState $ \st@ProcessState{..} ->+    let sz = Map.size usrTimers+        tk = sz + 1+    in (st { usrTimers = (Map.insert tk (t, m) usrTimers) }, tk)++-- | Remove a /user timer/, for the given key.+removeUserTimer :: TimerKey -> GenProcess s ()+removeUserTimer i =+  modifyState $ \st@ProcessState{..} -> st { usrTimers = (Map.delete i usrTimers) }++-- | Consume the timer with the given @TimerKey@. The timer is removed from the+-- @ProcessState@ and given to the supplied expression, whose evaluation is given+-- back to the caller.+consumeTimer :: forall s a . TimerKey -> (Message -> GenProcess s a) -> GenProcess s a+consumeTimer k f = do+  mt <- gets usrTimers+  let tm = Map.lookup k mt+  let ut = Map.delete k mt+  modifyState $ \st@ProcessState{..} -> st { usrTimers = ut }+  case tm of+    Nothing     -> lift $ die $ "GenProcess.consumeTimer - InvalidTimerKey"+    Just (_, m) -> f m++-- | The @ProcessDefinition@ for the current loop.+processDefinition :: GenProcess s (ProcessDefinition s)+processDefinition = gets procDef++-- | The list of prioritisers for the current loop.+processPriorities :: GenProcess s ([DispatchPriority s])+processPriorities = gets procPrio++-- | The list of filters for the current loop.+processFilters :: GenProcess s ([DispatchFilter s])+processFilters = gets procFilters++-- | Evaluates to the user defined state for the currently executing server loop.+processState :: GenProcess s s+processState = gets procState++-- | Evaluates to the @UnhandledMessagePolicy@ for the current loop.+processUnhandledMsgPolicy :: GenProcess s UnhandledMessagePolicy+processUnhandledMsgPolicy = gets (unhandledMessagePolicy . procDef)++-- | The @Timer@ for the system timeout. See @drainTimeout@.+systemTimeout :: GenProcess s Timer+systemTimeout = gets sysTimeout++-- | The policy for the system timeout. This is used to determine how the loop+-- should limit the time spent draining the /real/ process mailbox into our+-- internal priority queue.+timeoutPolicy :: GenProcess s RecvTimeoutPolicy+timeoutPolicy = gets timeoutSpec++-- | The @Delay@ for the @drainTimeout@.+drainTimeout :: GenProcess s Delay+drainTimeout = gets (timerDelay . sysTimeout)++-- | The current (user supplied) timeout.+currentTimeout :: GenProcess s Delay+currentTimeout = gets usrTimeout++-- | Update and store the internal priority queue.+updateQueue :: (Queue -> Queue) -> GenProcess s ()+updateQueue f =+  modifyState $ \st@ProcessState{..} -> st { internalQ = f internalQ }++-- | Evaluate any matching /info handler/ with the supplied datum after waiting+-- for at least @TimeInterval@. The process state (for the resulting @Action s@)+-- is also given and the process loop will go on as per @Server.continue@.+--+-- Informally, evaluating this expression (such that the @Action@ is given as the+-- result of a handler or filter) will ensure that the supplied message (datum)+-- is availble for processing no sooner than @TimeInterval@.+--+-- Currently, this expression creates an @Action@ that triggers immediate+-- evaluation in the process loop before continuing with the given state. The+-- process loop stores a /user timeout/ for the given time interval, which is+-- trigerred like a wait/drain timeout. This implementation is subject to change.+evalAfter :: forall s m . (Serializable m) => TimeInterval -> m -> s -> Action s+evalAfter d m s = act $ runAfter d m >> setProcessState s++-- | Produce an @Action s@ that, if it is the result of a handler, will cause the+-- server loop to evaluate the supplied expression. This is given in the @GenProcess@+-- monad, which is intended for internal use only.+act :: forall s . GenProcess s () -> Action s+act = return . ProcessActivity+{-# WARNING act "This interface is intended for internal use only" #-}++-- | Starts a timer and adds it as a /user timeout/.+runAfter :: forall s m . (Serializable m) => TimeInterval -> m -> GenProcess s ()+runAfter d m = do+  t <- lift $ startTimer (Delay d)+  void $ addUserTimer t (unsafeWrapMessage m)+{-# WARNING runAfter "This interface is intended for internal use only" #-}++--------------------------------------------------------------------------------+-- Internal Priority Queue                                                    --+--------------------------------------------------------------------------------++-- | Dequeue a message from the internal priority queue.+dequeue :: GenProcess s (Maybe Message)+dequeue = getAndModifyState $ \st -> do+            let pq = internalQ st+            case Q.dequeue pq of+              Nothing      -> (st, Nothing)+              Just (m, q') -> (st { internalQ = q' }, Just m)++-- | Peek at the next available message in the internal priority queue, without+-- removing it.+peek :: GenProcess s (Maybe Message)+peek = getAndModifyState $ \st -> do+         let pq = internalQ st+         (st, Q.peek pq)++-- | Push a message to the head of the internal priority queue.+push :: forall s . Message -> GenProcess s ()+push m = do+  st <- processState+  enqueueMessage st [ PrioritiseInfo {+    prioritise = (\_ m' ->+      return $ Just ((101 :: Int), m')) :: s -> Message -> Process (Maybe (Int, Message)) } ] m++-- | Enqueue a message in the internal priority queue. The given message will be+-- evaluated by all the supplied prioritisers, and if none match it, then it will+-- be assigned the lowest possible priority (i.e. put at the back of the queue).+enqueueMessage :: forall s . s+               -> [DispatchPriority s]+               -> Message+               -> GenProcess s ()+enqueueMessage s [] m' =+  enqueueMessage s [ PrioritiseInfo {+    prioritise = (\_ m ->+      return $ Just ((-1 :: Int), m)) :: s -> Message -> Process (Maybe (Int, Message)) } ] m'+enqueueMessage s (p:ps) m' = let checkPrio = prioritise p s in do+    (lift $ checkPrio m') >>= doEnqueue s ps m'   where-    verify pDef = mapM_ disallowCC $ apiHandlers pDef+    doEnqueue :: s+              -> [DispatchPriority s]+              -> Message+              -> Maybe (Int, Message)+              -> GenProcess s ()+    doEnqueue s' ps' msg Nothing       = enqueueMessage s' ps' msg+    doEnqueue _  _   _   (Just (i, m)) = updateQueue (Q.enqueue (i * (-1 :: Int)) m) -    disallowCC (DispatchCC _ _) = die $ ExitOther "IllegalControlChannel"-    disallowCC _                = return ()+--------------------------------------------------------------------------------+-- Process Loop Implementations                                               --+-------------------------------------------------------------------------------- -recvQueue :: PrioritisedProcessDefinition s+-- | Maps handlers to a dynamic action that can take place outside of a+-- expect/recieve block. This is used by the prioritised process loop.+class DynMessageHandler d where+  dynHandleMessage :: UnhandledMessagePolicy+                   -> s+                   -> d s+                   -> 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)++instance DynMessageHandler ExternDispatcher where+  dynHandleMessage _ s (DispatchCC  _ d)     msg = handleMessage msg (d s)+  dynHandleMessage _ s (DispatchSTM _ d _ _) msg = handleMessage msg (d s)++instance DynMessageHandler DeferredDispatcher where+  dynHandleMessage _ s (DeferredDispatcher d) = d s++-- | Maps filters to an action that can take place outside of a+-- expect/recieve block.+class DynFilterHandler d where+  dynHandleFilter :: s+                  -> d s+                  -> Message+                  -> Process (Maybe (Filter s))++instance DynFilterHandler DispatchFilter where+  dynHandleFilter s (FilterApi d)   msg = handleMessage msg (d s)+  dynHandleFilter s (FilterAny d)   msg = handleMessage msg (d s)+  dynHandleFilter s (FilterRaw d)   msg = d s msg+  dynHandleFilter s (FilterState d) _   = d s++-- | Prioritised process loop.+--+-- Evaluating this function will cause the caller to enter a server loop,+-- constantly reading messages from its mailbox (and/or other supplied control+-- planes) and passing these to handler functions in the supplied process+-- definition. Only when it is determined that the server process should+-- terminate - either by the handlers deciding to stop the process, or by an+-- unhandled exit signal or other form of failure condition (e.g. synchronous or+-- asynchronous exceptions).+--+-- ensureIOManagerIsRunning before evaluating this loop...+--+precvLoop :: PrioritisedProcessDefinition s           -> s-          -> TimeoutSpec-          -> Queue+          -> Delay           -> 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'+precvLoop ppDef pState recvDelay = do+  st <- P.liftIO $ newIORef $ ProcessState { timeoutSpec = recvTimeout ppDef+                                           , sysTimeout  = delayTimer Infinity+                                           , usrTimeout  = recvDelay+                                           , internalQ   = Q.empty+                                           , procState   = pState+                                           , procDef     = processDef ppDef+                                           , procPrio    = priorities ppDef+                                           , procFilters = filters ppDef+                                           , usrTimers   = Map.empty+                                           }++  mask $ \restore -> do+    res <- catch (fmap Right $ restore $ loop st)+                 (\(e :: SomeException) -> return $ Left e)++    -- res could be (Left ex), so we restore process state & def from our IORef+    ps <- P.liftIO $ atomicModifyIORef' st $ \s' -> (s', s')+    let st' = procState ps+        pd = procDef ps+        sh = shutdownHandler pd+    case res of+      Right (exitReason, _) -> do+        restore $ sh (CleanShutdown st') exitReason+        return exitReason+      Left ex -> do+        -- we'll attempt to run the exit handler with the original state+        restore $ sh (LastKnown st') (ExitOther $ show ex)+        throwM ex   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"+    loop st' = catchExit (runProcess st' recvQueue)+                         (\_ (r :: ExitReason) -> return (r, st')) -    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')+recvQueue :: GenProcess s ExitReason+recvQueue = do+  pd <- processDefinition+  let ex = trapExit:(exitHandlers $ pd)+  let exHandlers = map (\d' -> (dispatchExit d')) ex -    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)+  catch (drainMailbox >> processNext >>= nextAction)+        (\(e :: ProcessExitException) ->+            handleExit exHandlers e >>= nextAction)+  where -    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')+    handleExit :: [(s -> ProcessId -> Message -> Process (Maybe (ProcessAction s)))]+               -> ProcessExitException+               -> GenProcess s (ProcessAction s)+    handleExit []     ex                                 = throwM ex+    handleExit (h:hs) ex@(ProcessExitException pid msg) = do+      r <- processState >>= \s -> lift $ h s pid msg+      case r of+        Nothing -> handleExit hs ex+        Just p  -> return p -    processApply def pState msg =+    nextAction :: ProcessAction s -> GenProcess s ExitReason+    nextAction ac+      | ProcessActivity  act'    <- ac = act' >> recvQueue+      | ProcessSkip              <- ac = recvQueue+      | ProcessContinue  ps'     <- ac = recvQueueAux ps'+      | ProcessTimeout   d   ps' <- ac = setUserTimeout d >> recvQueueAux ps'+      | ProcessStop      xr      <- ac = return xr+      | ProcessStopping  ps' xr  <- ac = setProcessState ps' >> return xr+      | ProcessHibernate d' s'   <- ac = (lift $ block d') >> recvQueueAux s'+      | otherwise {- compiler foo -}   = return $ ExitOther "IllegalState"++    recvQueueAux st = setProcessState st >> recvQueue++      -- TODO: at some point we should re-implement our state monad in terms of+      -- mkWeakIORef instead of a full IORef. At that point, we can implement hiberation+      -- in the following terms:+      -- 1. the user defines (at some level, perhaps outside of this API) some+      --    means for writing a process' state to a backing store+      --    NB: this could be /persistent/, or a file, or database, etc...+      -- 2. when we enter hibernation, we do the following:+      --    (a) write the process state to the chosen backing store+      --    (b) evaluate yield (telling the RTS we're willing to give up our time slice)+      --    (c) enter a blocking receiveWait with no state on our stack...+      --        [NB] presumably at this point our state will be eligible for GC+      --    (d) when we finally receive a message, reboot the process thus:+      --        (i)   read our state back from the given backing store+      --        (ii)  call a user defined function to rebuild the state if custom+      --              actions need to be taken (e.g. they might've stored something+      --              like an STM TVar and need to request a new one from some+      --              well known service or registry - alt. they might want to+      --              /replay/ actions to rebuild their state as an FSM might)+      --        (iii) re-enter the recv loop and immediately processNext+      --+      -- This will give roughly the same semantics as erlang's hibernate/3, although+      -- the RTS does GC globally rather than per-thread, but that might change in+      -- some future release (who knows!?).+      --+      -- Also, this gives us the ability to migrate process state across remote+      -- boundaries. Not only can a process be moved in this way, if we generalise+      -- the mechanism to move a serialised closure, we can migrate the whole process+      -- and its state as well. The main difference here (with ordinary use of+      -- @Closure@ et al for moving processes around, is that we do not insist+      -- on the process state being serializable, simply that they provide a+      -- function to read+write the state, and a (state -> state) function to be+      -- called during rehydration if custom actions need to be taken.+      --++    processNext :: GenProcess s (ProcessAction s)+    processNext = do+      (up, pf) <- gets $ liftA2 (,) (unhandledMessagePolicy . procDef) procFilters+      case pf of+        [] -> consumeMessage+        _  -> filterMessage  (filterNext up pf Nothing)++    consumeMessage = applyNext dequeue processApply+    filterMessage = applyNext peek++    filterNext :: UnhandledMessagePolicy+               -> [DispatchFilter s]+               -> Maybe (Filter s)+               -> Message+               -> GenProcess s (ProcessAction s)+    filterNext mp' fs mf msg+      | Just (FilterSkip s')   <- mf = setProcessState s' >> dequeue >> return ProcessSkip+      | Just (FilterStop s' r) <- mf = return $ ProcessStopping s' r+      | Just (FilterOk s')     <- mf+      , []                     <- fs = setProcessState s' >> applyNext dequeue processApply+      | Nothing <- mf, []      <- fs = applyNext dequeue processApply+      | Just (FilterOk s')     <- mf+      , (f:fs')                <- fs = do+          setProcessState s'+          act' <- lift $ dynHandleFilter s' f msg+          filterNext mp' fs' act' msg+      | Just (FilterReject _ s') <- mf = do+          setProcessState s' >> dequeue >>= lift . applyPolicy mp' s' . fromJust+      | Nothing <- mf {- filter didn't apply to the input type -}+      , (f:fs') <- fs = processState >>= \s' -> do+          lift (dynHandleFilter s' f msg) >>= \a -> filterNext mp' fs' a msg++    applyNext :: (GenProcess s (Maybe Message))+              -> (Message -> GenProcess s (ProcessAction s))+              -> GenProcess s (ProcessAction s)+    applyNext queueOp handler = do+      next <- queueOp+      case next of+        Nothing  -> drainOrTimeout+        Just msg -> handler msg++    processApply msg = do+      (def, pState) <- gets $ liftA2 (,) procDef procState+       let pol          = unhandledMessagePolicy def           apiMatchers  = map (dynHandleMessage pol pState) (apiHandlers def)           infoMatchers = map (dynHandleMessage pol pState) (infoHandlers def)+          extMatchers  = map (dynHandleMessage pol pState) (externHandlers def)           shutdown'    = dynHandleMessage pol pState shutdownHandler'-          ms'          = (shutdown':apiMatchers) ++ infoMatchers-      in processApplyAux ms' pol pState msg+          ms'          = (shutdown':extMatchers) ++ apiMatchers ++ infoMatchers+      processApplyAux ms' pol pState msg -    processApplyAux []     p' s' m' = applyPolicy p' s' m'+    processApplyAux []     p' s' m' = lift $ 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+     attempt <- lift $ 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+    drainMailbox :: GenProcess s ()+    drainMailbox = do+      -- see note [timer handling whilst draining the process' mailbox]+      ps <- processState+      pd <- processDefinition+      pp <- processPriorities+      ut <- gets usrTimers+      let ts = Map.foldrWithKey (\k (t, _) ms -> ms ++ matchKey k t) [] ut+      let ms = ts ++ (matchAny (return . Right) : (mkMatchers ps pd))+      timerAcc <- timeoutPolicy >>= \spec -> case spec of+                                               RecvTimer      _   -> return Nothing+                                               RecvMaxBacklog cnt -> return $ Just cnt+      -- see note [handling async exceptions during non-blocking reads]+      -- Also note that we only use the system timeout here, dropping into the+      -- user timeout only if we end up in a blocking read on the mailbox.+      --+      mask_ $ do+        tt <- maybeStartTimer+        drainAux ps pp timerAcc (ms ++ matchTimeout tt)+        (lift $ stopTimer tt) >>= setDrainTimeout++    drainAux :: s+             -> [DispatchPriority s]+             -> Limit+             -> [Match (Either TimedOut Message)]+             -> GenProcess s ()+    drainAux ps' pp' maxbq ms = do+      (cnt, m) <- scanMailbox maxbq ms+      case m of+        Nothing               -> return ()+        Just (Right m')       -> do enqueueMessage ps' pp' m'+                                    drainAux ps' pp' cnt ms+        Just (Left TimedOut)  -> return ()+        Just (Left (Yield i)) ->+          -- we saw a user defined timer fire, and will have an associated message...+          -- this is a bit complex, we have to enqueue the message and remove the timer+          -- the latter part of which is handled for us by consumeTimer+          consumeTimer i push >> drainAux ps' pp' cnt ms++    maybeStartTimer :: GenProcess s Timer+    maybeStartTimer = do+      tp <- timeoutPolicy+      t <- case tp of+             RecvTimer d -> (lift $ startTimer $ Delay d)+             _           -> return $ delayTimer Infinity+      setDrainTimeout t+      return t++    scanMailbox :: Limit+                -> [Match (Either TimedOut Message)]+                -> GenProcess s (Limit, Maybe (Either TimedOut Message))+    scanMailbox lim ms+      | Just 0 <- lim = return (lim, Just $ Left TimedOut)+      | Just c <- lim = do {- non-blocking read on our mailbox, any external inputs,+                              plus whatever match specs the TimeoutManager gives -}+                        lift $ fmap (Just (c - 1), ) (receiveTimeout 0 ms)+      | otherwise     = lift $ fmap (lim, ) (receiveTimeout 0 ms)++    -- see note [timer handling whilst draining the process' mailbox]+    drainOrTimeout :: GenProcess s (ProcessAction s)+    drainOrTimeout = do+      pd <- processDefinition+      ps <- processState+      ud <- currentTimeout+      mr <- mkMatchRunners+      let ump     = unhandledMessagePolicy pd+          hto     = timeoutHandler pd+          matches = mr ++ ((matchMessage return):map (matchExtern ump ps) (externHandlers pd))+          recv    = case ud of+                      Infinity -> lift $ fmap Just (receiveWait matches)+                      NoDelay  -> lift $ receiveTimeout 0 matches+                      Delay i  -> lift $ receiveTimeout (asTimeout i) matches++      -- see note [masking async exceptions during recv]+      mask $ \restore -> recv >>= \r ->         case r of-          Nothing -> h pState delay >>= \act -> return $ (act, delay, queue)+          Nothing -> restore $ lift $ hto ps ud           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')+            pp <- processPriorities+            enqueueMessage ps pp m+            -- Returning @ProcessSkip@ simply causes us to go back into+            -- listening mode until we hit RecvTimeoutPolicy+            restore $ return ProcessSkip -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'+    mkMatchRunners :: GenProcess s [Match Message]+    mkMatchRunners = do+      ut <- gets usrTimers+      fn <- mkRunner+      let ms = Map.foldrWithKey (\k (t, _) ms' -> ms' ++ matchRun fn k t) [] ut+      return ms -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'+    mkRunner :: GenProcess s (TimerKey -> Process Message)+    mkRunner = do+      st <- ST.get+      let fn = \k -> do (m, _) <- runProcess st (consumeTimer k return)+                        return m+      return fn +    mkMatchers :: s+                -> ProcessDefinition s+                -> [Match (Either TimedOut Message)]+    mkMatchers st df =+      map (matchMapExtern (unhandledMessagePolicy df) st toRight)+          (externHandlers df)++    toRight :: Message -> Either TimedOut Message+    toRight = Right++-- note [handling async exceptions during non-blocking reads]+-- Our golden rule is that if we've dequeued any kind of Message at all+-- from the process mailbox (or input channels), we must not /lose/ it+-- if an asynchronous exception arrives. We therefore mask  when we perform a+-- non-blocking scan on the mailbox, and whilst we enqueue messages.+--+-- If an initial scan of the mailbox yields no data, we fall back to making+-- a blocking read; See note [masking async exceptions during recv].+--+-- Once messages have been safely moved from the mailbox to our priority queue,+-- we restore the masking state whilst running handlers.+--++-- note [timer handling whilst draining the process' mailbox]+-- To prevent a DOS vector - and quite a likely accidental one at that - we do not+-- sit draining the mailbox indefinitely, since continuous reading would thus+-- leave us unable to process any inputs and we'd eventually run out of memory.+-- Instead, the PrioritisedProcessDefinition holds a RecvTimeoutPolicy which can+-- hold either a max-messages-processed limit or a timeout value. Using whichever+-- policy is provided, drainMessageQueue will stop attempting to receive new mail+-- either once the message count limit is exceeded or the timer expires, at which+-- point we go back to processNext.++-- note [masking async exceptions during recv]+-- Reading the process' mailbox is mask'ed anyway, however this only+-- covers dequeue on the underlying CQueue, such that either before+-- the dequeue takes place, or after (during evaluation of the result,+-- or execution of the discovered @Match@ for the message), we can still+-- be terminated by an asynchronous exception. This is wrong, from the+-- perspective of a managed process, since in the case of an exit signal+-- we might handle the exception, at which point we've dequeued and+-- subsequently lost a message.+--+-- Masking recv then, prevents this from happening, and is relatively+-- safe, because we know the following (having written all the handlers+-- explicitly ourselves):+--+-- 1. each handler does nothing more than return the underlying message+-- 2. in the most complex case, we have @Left . unsafeWrapMessage@ or+--    @fmap Right readSTM thing@ inside of @matchSTM@+-- 3. We should not, therefore, introduce any uninterruptible behaviour+-- 4. We cannot, however, be certain that this holds true for decoding+--    (and subsequent calls into Binary and/or Bytestrings), so at best+--    we can mask, but not uninterruptibleMask+--+-- NB: According to /qnikst/, atomicModifyIORef' does not require us to+-- use uninterruptibleMask anyway, so this is fine...+--+ -------------------------------------------------------------------------------- -- Ordinary/Blocking Mailbox Handling                                         -- -------------------------------------------------------------------------------- +-- TODO: wrap recvLoop in the same exception handling as precvLoop+--       notably, we need to ensure the shutdownHandler runs even in the face+--       of exceptions, and it would be useful/good IMO to pass an IORef for+--       the state, so we can have a decent LastKnown value for it++-- | Managed process loop.+--+-- Evaluating this function will cause the caller to enter a server loop,+-- constantly reading messages from its mailbox (and/or other supplied control+-- planes) and passing these to handler functions in the supplied process+-- definition. Only when it is determined that the server process should+-- terminate - either by the handlers deciding to stop the process, or by an+-- unhandled exit signal or other form of failure condition (e.g. synchronous or+-- asynchronous exceptions).+-- 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)+      extMatchers   = map (matchDispatch p pState) (externHandlers pDef)+      matchers      = extMatchers ++ (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)+        ProcessSkip               -> recvLoop pDef pState recvDelay -- TODO: handle differently...+        (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 (LastKnown pState) r >> return (r :: ExitReason)+        (ProcessStopping s' r)    -> handleStop (LastKnown s') r >> return (r :: ExitReason)+        (ProcessActivity _)       -> die $ "recvLoop.InvalidState - ProcessActivityNotSupported"   where     matchAux :: UnhandledMessagePolicy              -> s@@ -216,10 +717,10 @@              -> [Match (ProcessAction s)]     matchAux p ps ds = [matchAny (auxHandler (applyPolicy p ps) ps ds)] -    auxHandler :: (P.Message -> Process (ProcessAction s))+    auxHandler :: (Message -> Process (ProcessAction s))                -> s                -> [DeferredDispatcher s]-               -> P.Message+               -> Message                -> Process (ProcessAction s)     auxHandler policy _  [] msg = policy msg     auxHandler policy st (d:ds :: [DeferredDispatcher s]) msg@@ -228,14 +729,14 @@         -- 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+          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+          Nothing   -> policy msg+          Just act' -> return act'      processReceive :: [Match (ProcessAction s)]                    -> TimeoutHandler s@@ -258,46 +759,6 @@         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                                                                  -- -------------------------------------------------------------------------------- @@ -307,14 +768,16 @@  -- @(ProcessExitException from ExitShutdown)@ will stop the server gracefully trapExit :: ExitSignalDispatcher s-trapExit = handleExit (\_ _ (r :: ExitReason) -> stop r)+trapExit = handleExitIf (\_ e -> e == ExitShutdown)+                        (\_ _ (r :: ExitReason) -> stop r)  block :: TimeInterval -> Process ()-block i = liftIO $ threadDelay (asTimeout i)+block i =+  void $ receiveTimeout (asTimeout i) [ match (\(_ :: TimedOut) -> return ()) ]  applyPolicy :: UnhandledMessagePolicy             -> s-            -> P.Message+            -> Message             -> Process (ProcessAction s) applyPolicy p s m =   case p of@@ -325,4 +788,3 @@   where     logIt =       Log.report Log.info Log.logChannel $ "Unhandled Gen Input Message: " ++ (show m)-

@@ -1,21 +1,38 @@ {-# LANGUAGE DeriveDataTypeable         #-} {-# LANGUAGE ExistentialQuantification  #-}-{-# LANGUAGE StandaloneDeriving         #-}-{-# LANGUAGE TemplateHaskell            #-} {-# LANGUAGE DeriveGeneric              #-} {-# LANGUAGE StandaloneDeriving         #-} {-# LANGUAGE ScopedTypeVariables        #-} {-# LANGUAGE GeneralizedNewtypeDeriving #-}+{-# LANGUAGE LiberalTypeSynonyms        #-}+{-# LANGUAGE Rank2Types                 #-}+{-# LANGUAGE UndecidableInstances       #-}+{-# LANGUAGE FlexibleInstances          #-}+{-# LANGUAGE TypeFamilies               #-}+{-# LANGUAGE MultiParamTypeClasses      #-}+{-# LANGUAGE FunctionalDependencies     #-}  -- | Types used throughout the ManagedProcess framework module Control.Distributed.Process.ManagedProcess.Internal.Types   ( -- * Exported data types     InitResult(..)+  , GenProcess()+  , runProcess+  , lift+  , liftIO+  , ProcessState(..)+  , State+  , Queue+  , Limit   , Condition(..)   , ProcessAction(..)   , ProcessReply(..)+  , Action+  , Reply+  , ActionHandler   , CallHandler   , CastHandler+  , StatelessHandler   , DeferredCallHandler   , StatelessCallHandler   , InfoHandler@@ -23,11 +40,17 @@   , StatelessChannelHandler   , InitHandler   , ShutdownHandler+  , ExitState(..)+  , isCleanShutdown+  , exitState   , TimeoutHandler   , UnhandledMessagePolicy(..)   , ProcessDefinition(..)   , Priority(..)   , DispatchPriority(..)+  , DispatchFilter(..)+  , Filter(..)+--   , Check(..)   , PrioritisedProcessDefinition(..)   , RecvTimeoutPolicy(..)   , ControlChannel(..)@@ -35,22 +58,29 @@   , ControlPort(..)   , channelControlPort   , Dispatcher(..)+  , ExternDispatcher(..)   , DeferredDispatcher(..)   , ExitSignalDispatcher(..)   , MessageMatcher(..)-  , DynMessageHandler(..)+  , ExternMatcher(..)   , Message(..)   , CallResponse(..)   , CallId   , CallRef(..)+  , CallRejected(..)   , makeRef+  , caller+  , rejectToCaller+  , recipient+  , tag   , initCall   , unsafeInitCall   , waitResponse   ) where -import Control.Distributed.Process hiding (Message)-import qualified Control.Distributed.Process as P (Message)+import Control.Concurrent.STM (STM)+import Control.Distributed.Process hiding (Message, mask, finally, liftIO)+import qualified Control.Distributed.Process as P (Message, liftIO) import Control.Distributed.Process.Serializable import Control.Distributed.Process.Extras   ( Recipient(..)@@ -60,45 +90,91 @@   , Routable(..)   , NFSerializable   )+import Control.Distributed.Process.Extras.Internal.Queue.PriorityQ+  ( PriorityQ+  ) import Control.Distributed.Process.Extras.Internal.Types   ( resolveOrDie   ) import Control.Distributed.Process.Extras.Time+import Control.Distributed.Process.ManagedProcess.Timer (Timer, TimerKey) import Control.DeepSeq (NFData(..))+import Control.Monad.Fix (MonadFix)+import Control.Monad.Catch+  ( catch+  , throwM+  , uninterruptibleMask+  , mask+  , finally+  , MonadThrow+  , MonadCatch+  , MonadMask+  )+import qualified Control.Monad.Catch as Catch+  ( catch+  , throwM+  )+import Control.Monad.IO.Class (MonadIO)+import qualified Control.Monad.State.Strict as ST+  ( MonadState+  , StateT+  , get+  , lift+  , runStateT+  ) import Data.Binary hiding (decode)+import Data.Map.Strict (Map) import Data.Typeable (Typeable)-+import Data.IORef (IORef) import Prelude hiding (init)- import GHC.Generics  -------------------------------------------------------------------------------- -- API                                                                        -- -------------------------------------------------------------------------------- +-- | wrapper for a @MonitorRef@ type CallId = MonitorRef +-- | Wraps a consumer of the call API 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 rnf (CallRef x) = rnf x `seq` () -makeRef :: forall a . (Serializable a) => Recipient -> CallId -> CallRef a-makeRef r c = CallRef (r, c)+-- | Retrieve the @Recipient@ for a @CallRef@.+recipient :: CallRef a -> Recipient+recipient = fst . unCaller -instance Resolvable (CallRef a) where-  resolve (CallRef (r, _)) = resolve r+-- | Retrieve the @CallId@ for a @CallRef@.+tag :: CallRef a -> CallId+tag = snd . unCaller -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)+instance Binary (CallRef a) where+instance NFData (CallRef a) where rnf (CallRef x) = rnf x `seq` () +-- | Creates a @CallRef@ for the given @Recipient@ and @CallId@+makeRef :: Recipient -> CallId -> CallRef a+makeRef r c = CallRef (r, c)++-- | @Message@ type used internally by the call, cast, and rpcChan APIs. data Message a b =     CastMessage a   | CallMessage a (CallRef b)   | ChanMessage a (SendPort b)   deriving (Typeable, Generic) +-- | Retrieve the @Recipient@ from a @Message@. If the supplied message is+-- a /cast/ or /chan/ message will evaluate to @Nothing@, otherwise @Just ref@.+caller :: forall a b . Message a b -> Maybe Recipient+caller (CallMessage _ ref) = Just $ recipient ref+caller _                   = Nothing++-- | Reject a /call/ message with the supplied string. Sends @CallRejected@ to+-- the recipient if the input is a @CallMessage@, otherwise has no side effects.+rejectToCaller :: forall a b .+                  Message a b -> String -> Process ()+rejectToCaller (CallMessage _ ref) m = sendTo ref (CallRejected m (tag ref))+rejectToCaller _                   _ = return ()+ instance (Serializable a, Serializable b) => Binary (Message a b) where instance (NFSerializable a, NFSerializable b) => NFData (Message a b) where   rnf (CastMessage a) = rnf a `seq` ()@@ -107,6 +183,7 @@ deriving instance (Eq a, Eq b) => Eq (Message a b) deriving instance (Show a, Show b) => Show (Message a b) +-- | Response type for the call API data CallResponse a = CallResponse a CallId   deriving (Typeable, Generic) @@ -116,6 +193,20 @@ deriving instance Eq a => Eq (CallResponse a) deriving instance Show a => Show (CallResponse a) +-- | Sent to a consumer of the /call/ API when a server filter expression+-- explicitly rejects an incoming call message.+data CallRejected = CallRejected String CallId+  deriving (Typeable, Generic, Show, Eq)+instance Binary CallRejected where+instance NFData CallRejected where++instance Resolvable (CallRef a) where+  resolve (CallRef (r, _)) = resolve r++instance Routable (CallRef a) where+  sendTo (CallRef (c, _)) = sendTo c+  unsafeSendTo (CallRef (c, _)) = unsafeSendTo c+ -- | Return type for and 'InitHandler' expression. data InitResult s =     InitOk s Delay {-@@ -126,15 +217,103 @@         ^ the process has decided not to continue starting - this is not an error -}   deriving (Typeable) +-- | Represent a max-backlog from RecvTimeoutPolicy+type Limit = Maybe Int++-- | Internal priority queue, used by prioritised processes.+type Queue = PriorityQ Int P.Message++-- | Map from @TimerKey@ to @(Timer, Message)@.+type TimerMap = Map TimerKey (Timer, P.Message)++-- | Internal state of a prioritised process loop.+data ProcessState s = ProcessState { timeoutSpec :: RecvTimeoutPolicy+                                   , procDef     :: ProcessDefinition s+                                   , procPrio    :: [DispatchPriority s]+                                   , procFilters :: [DispatchFilter s]+                                   , usrTimeout  :: Delay+                                   , sysTimeout  :: Timer+                                   , usrTimers   :: TimerMap+                                   , internalQ   :: Queue+                                   , procState   :: s+                                   }++-- | Prioritised process state, held as an @IORef@.+type State s = IORef (ProcessState s)++-- | StateT based monad for prioritised process loops.+newtype GenProcess s a = GenProcess {+   unManaged :: ST.StateT (State s) Process a+ }+ deriving ( Functor+          , Monad+          , ST.MonadState (State s)+          , MonadIO+          , MonadFix+          , Typeable+          , Applicative+          )++instance forall s . MonadThrow (GenProcess s) where+  throwM = lift . Catch.throwM++instance forall s . MonadCatch (GenProcess s) where+  catch p h = do+    pSt <- ST.get+    -- we can throw away our state since it is always accessed via an IORef+    (a, _) <- lift $ Catch.catch (runProcess pSt p) (runProcess pSt . h)+    return a++instance forall s . MonadMask (GenProcess s) where+  mask p = do+      pSt <- ST.get+      lift $ mask $ \restore -> do+        (a, _) <- runProcess pSt (p (liftRestore restore))+        return a+    where+      liftRestore restoreP = \p2 -> do+        ourSTate <- ST.get+        (a', _) <- lift $ restoreP $ runProcess ourSTate p2+        return a'++  uninterruptibleMask p = do+      pSt <- ST.get+      (a, _) <- lift $ uninterruptibleMask $ \restore ->+        runProcess pSt (p (liftRestore restore))+      return a+    where+      liftRestore restoreP = \p2 -> do+        ourSTate <- ST.get+        (a', _) <- lift $ restoreP $ runProcess ourSTate p2+        return a'++-- | Run an action in the @GenProcess@ monad.+runProcess :: State s -> GenProcess s a -> Process (a, State s)+runProcess state proc = ST.runStateT (unManaged proc) state++-- | Lift an action in the @Process@ monad to @GenProcess@.+lift :: Process a -> GenProcess s a+lift p = GenProcess $ ST.lift p++-- | Lift an IO action directly into @GenProcess@, @liftIO = lift . Process.LiftIO@.+liftIO :: IO a -> GenProcess s a+liftIO = lift . P.liftIO+ -- | The action taken by a process after a handler has run and its updated state.--- See 'continue'---     'timeoutAfter'---     'hibernate'---     'stop'---     'stopWith'+-- See "Control.Distributed.Process.ManagedProcess.Server.continue"+--     "Control.Distributed.Process.ManagedProcess.Server.timeoutAfter"+--     "Control.Distributed.Process.ManagedProcess.Server.hibernate"+--     "Control.Distributed.Process.ManagedProcess.Server.stop"+--     "Control.Distributed.Process.ManagedProcess.Server.stopWith" --+-- Also see "Control.Distributed.Process.Management.Priority.act" and+-- "Control.Distributed.Process.ManagedProcess.Priority.runAfter".+--+-- And other actions. This type should not be used directly. data ProcessAction s =-    ProcessContinue  s              -- ^ continue with (possibly new) state+    ProcessSkip+  | ProcessActivity  (GenProcess s ()) -- ^ run the given activity+  | 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@@@ -145,6 +324,7 @@ -- can return @NoReply@ if they wish to ignore the call. data ProcessReply r s =     ProcessReply r (ProcessAction s)+  | ProcessReject String (ProcessAction s)  -- TODO: can we use a functional dependency here?   | NoReply (ProcessAction s)  -- | Wraps a predicate that is used to determine whether or not a handler@@ -155,59 +335,86 @@   | 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)+class Check c s m | s m -> c where+  -- data Checker c :: * -> * -> *+  -- apply :: s -> m -> Checker c s m -> Bool+  apply :: s -> m -> c -> Bool --- | An expression used to handle a /call/ message in a stateless process.-type StatelessCallHandler a b = a -> CallRef b -> Process (ProcessReply b ())+instance Check (Condition s m) s m where+  -- data Checker (Condition s m) s m = CheckCond (Condition s m)+  apply s m (Condition f) = f s m+  apply s _ (State f)     = f s+  apply _ m (Input f)     = f m --- | An expression used to handle a /cast/ message.-type CastHandler s a = s -> a -> Process (ProcessAction s)+instance Check (s -> m -> Bool) s m where+   -- data Checker (s -> m -> Bool) s m = CheckF (s -> m -> Bool)+   apply s m f = f s m+-} --- | An expression used to handle an /info/ message.-type InfoHandler s a = s -> a -> Process (ProcessAction s)+-- | Informs a /shutdown handler/ of whether it is running due to a clean+-- shutdown, or in response to an unhandled exception.+data ExitState s = CleanShutdown s -- ^ given when an ordered shutdown is underway+                 | LastKnown s     {-+                  ^ given due to an unhandled exception, passing the last known state -} --- | An expression used to handle a /channel/ message.-type ChannelHandler s a b = s -> SendPort b -> a -> Process (ProcessAction s)+-- | @True@ if the @ExitState@ is @CleanShutdown@, otherwise @False@.+isCleanShutdown :: ExitState s -> Bool+isCleanShutdown (CleanShutdown _) = True+isCleanShutdown _                 = False --- | An expression used to handle a /channel/ message in a stateless process.-type StatelessChannelHandler a b = SendPort b -> a -> Process (ProcessAction ())+-- | Evaluates to the @s@ state datum in the given @ExitState@.+exitState :: ExitState s -> s+exitState (CleanShutdown s) = s+exitState (LastKnown s)     = s --- | An expression used to initialise a process with its state.-type InitHandler a s = a -> Process (InitResult s)+-- | An action (server state transition) in the @Process@ monad+type Action s = Process (ProcessAction s) --- | An expression used to handle process termination.-type ShutdownHandler s = s -> ExitReason -> Process ()+-- | An action (server state transition) causing a reply to a  caller, in the+-- @Process@ monad+type Reply b s = Process (ProcessReply b s) --- | An expression used to handle process timeouts.-type TimeoutHandler s = s -> Delay -> Process (ProcessAction s)+-- | An expression used to handle a message+type ActionHandler s a = s -> a -> Action s --- dispatching to implementation callbacks+-- | An expression used to handle a message and providing a reply+type CallHandler s a b = s -> a -> Reply b s --- 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.+-- | An expression used to ignore server state during handling+type StatelessHandler s a = a -> (s -> Action s) -{-+-- | An expression used to handle a /call/ message where the reply is deferred+-- via the 'CallRef'+type DeferredCallHandler s a b = CallRef b -> CallHandler s a b -data InputChannelDispatcher =-  InputChannelDispatcher { chan :: InputChannel s-                         , dispatch :: s -> Message a b -> Process (ProcessAction s)-                         }+-- | An expression used to handle a /call/ message ignoring server state+type StatelessCallHandler s a b = CallRef b -> a -> Reply b s -instance MessageMatcher Dispatcher where-  matchDispatch _ _ (DispatchInputChannelDispatcher c d) = matchInputChan (d s)--}+-- | An expression used to handle a /cast/ message+type CastHandler s a = ActionHandler s a +-- | An expression used to handle an /info/ message+type InfoHandler s a = ActionHandler s a++-- | An expression used to handle a /channel/ message+type ChannelHandler s a b = SendPort b -> ActionHandler s a++-- | An expression used to handle a /channel/ message in a stateless process+type StatelessChannelHandler s a b = SendPort b -> StatelessHandler s a++-- | 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 = ExitState s -> ExitReason -> Process ()++-- | An expression used to handle process timeouts+type TimeoutHandler s = ActionHandler s Delay++-- dispatching to implementation callbacks+ -- | Provides a means for servers to listen on a separate, typed /control/ -- channel, thereby segregating the channel from their regular -- (and potentially busy) mailbox.@@ -218,7 +425,7 @@  -- | Creates a new 'ControlChannel'. newControlChan :: (Serializable m) => Process (ControlChannel m)-newControlChan = newChan >>= return . ControlChannel+newControlChan = fmap ControlChannel newChan  -- | The writable end of a 'ControlChannel'. --@@ -232,11 +439,39 @@  -- | Obtain an opaque expression for communicating with a 'ControlChannel'. ---channelControlPort :: (Serializable m)-                   => ControlChannel m+channelControlPort :: ControlChannel m                    -> ControlPort m channelControlPort cc = ControlPort $ fst $ unControl cc +-- | Given as the result of evaluating a "DispatchFilter". This type is intended+-- for internal use. For an API for working with filters,+-- see "Control.Distributed.Process.ManagedProcess.Priority".+data Filter s = FilterOk s+              | forall m . (Show m) => FilterReject m s+              | FilterSkip s+              | FilterStop s ExitReason++-- | Provides dispatch from a variety of inputs to a typed filter handler.+data DispatchFilter s =+    forall a b . (Serializable a, Serializable b) =>+    FilterApi+    {+      apiFilter :: s -> Message a b -> Process (Filter s)+    }+  | forall a . (Serializable a) =>+    FilterAny+    {+      anyFilter :: s -> a -> Process (Filter s)+    }+  | FilterRaw+    {+      rawFilter :: s -> P.Message -> Process (Maybe (Filter s))+    }+  | FilterState+    {+      stateFilter :: s -> Process (Maybe (Filter s))+    }+ -- | Provides dispatch from cast and call messages to a typed handler. data Dispatcher s =     forall a b . (Serializable a, Serializable b) =>@@ -250,12 +485,23 @@       dispatch   :: s -> Message a b -> Process (ProcessAction s)     , dispatchIf :: s -> Message a b -> Bool     }-  | forall a b . (Serializable a, Serializable b) =>++-- | Provides dispatch for channels and STM actions+data ExternDispatcher s =+    forall a b . (Serializable a, Serializable b) =>     DispatchCC  -- control channel dispatch     {-      channel  :: ReceivePort (Message a b)-    , dispatch :: s -> Message a b -> Process (ProcessAction s)+      channel      :: ReceivePort (Message a b)+    , dispatchChan :: s -> Message a b -> Process (ProcessAction s)     }+  | forall a . (Serializable a) =>+    DispatchSTM -- arbitrary STM actions+    {+      stmAction   :: STM a+    , dispatchStm :: s -> a -> Process (ProcessAction s)+    , matchStm    :: Match P.Message+    , matchAnyStm :: forall m . (P.Message -> m) -> Match m+    }  -- | Provides dispatch for any input, returns 'Nothing' for unhandled messages. data DeferredDispatcher s =@@ -276,31 +522,37 @@                  -> Process (Maybe (ProcessAction s))   } +-- | Defines the means of dispatching inbound messages to a handler 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)+  matchDispatch _ s (Dispatch         d)      = match   (d s)+  matchDispatch _ s (DispatchIf       d cond) = matchIf (cond s) (d s) -class DynMessageHandler d where-  dynHandleMessage :: UnhandledMessagePolicy-                   -> s-                   -> d s-                   -> P.Message-                   -> Process (Maybe (ProcessAction s))+instance MessageMatcher ExternDispatcher where+  matchDispatch _ s (DispatchCC  c d)     = matchChan c (d s)+  matchDispatch _ s (DispatchSTM c d _ _) = matchSTM  c (d 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"+-- | Defines the means of dispatching messages from external channels (e.g.+-- those defined in terms of "ControlChannel", and STM actions) to a handler.+class ExternMatcher d where+  matchExtern :: UnhandledMessagePolicy -> s -> d s -> Match P.Message -instance DynMessageHandler DeferredDispatcher where-  dynHandleMessage _ s (DeferredDispatcher d) = d s+  matchMapExtern :: forall m s . UnhandledMessagePolicy+                 -> s -> (P.Message -> m) -> d s -> Match m +instance ExternMatcher ExternDispatcher where+  matchExtern _ _ (DispatchCC  c _)     = matchChan c (return . unsafeWrapMessage)+  matchExtern _ _ (DispatchSTM _ _ m _) = m++  matchMapExtern _ _ f (DispatchCC c _)      = matchChan c (return . f . unsafeWrapMessage)+  matchMapExtern _ _ f (DispatchSTM _ _ _ p) = p f++-- | Priority of a message, encoded as an @Int@ newtype Priority a = Priority { getPrio :: Int } +-- | Dispatcher for prioritised handlers data DispatchPriority s =     PrioritiseCall     {@@ -326,7 +578,7 @@ -- will stop removing messages from its mailbox and process those it has already -- received. ---data RecvTimeoutPolicy = RecvCounter Int | RecvTimer TimeInterval+data RecvTimeoutPolicy = RecvMaxBacklog Int | RecvTimer TimeInterval   deriving (Typeable)  -- | A @ProcessDefinition@ decorated with @DispatchPriority@ for certain@@ -336,6 +588,7 @@   {     processDef  :: ProcessDefinition s   , priorities  :: [DispatchPriority s]+  , filters     :: [DispatchFilter s]   , recvTimeout :: RecvTimeoutPolicy   } @@ -347,13 +600,15 @@   | 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+  deriving (Show, Eq)  -- | 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+    apiHandlers    :: [Dispatcher s]       -- ^ functions that handle call/cast messages+  , infoHandlers   :: [DeferredDispatcher s] -- ^ functions that handle non call/cast messages+  , externHandlers :: [ExternDispatcher s] -- ^ functions that handle control channel and STM inputs+  , 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@@ -380,6 +635,10 @@ -- 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?++-- | The send part of the /call/ client-server interaction. The resulting+-- "CallRef" can be used to identify the corrolary response message (if one is+-- sent by the server), and is unique to this /call-reply/ pair. initCall :: forall s a b . (Addressable s, Serializable a, Serializable b)          => s -> a -> Process (CallRef b) initCall sid msg = do@@ -390,8 +649,11 @@     sendTo pid (CallMessage msg cRef :: Message a b)     return cRef -unsafeInitCall :: forall s a b . (Addressable s,-                                  NFSerializable a, NFSerializable b)+-- | Version of @initCall@ that utilises "unsafeSendTo".+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 "@@ -401,6 +663,9 @@     unsafeSendTo pid (CallMessage msg cRef  :: Message a b)     return cRef +-- | Wait on the server's response after an "initCall" has been previously been sent.+--+-- This function does /not/ trap asynchronous exceptions. waitResponse :: forall b. (Serializable b)              => Maybe TimeInterval              -> CallRef b@@ -409,11 +674,12 @@   let (_, mRef) = unCaller cRef       matchers  = [ matchIf (\((CallResponse _ ref) :: CallResponse b) -> ref == mRef)                             (\((CallResponse m _) :: CallResponse b) -> return (Right m))+                  , matchIf (\((CallRejected _ ref)) -> ref == mRef)+                            (\(CallRejected s _) -> return (Left $ ExitOther $ s))                   , 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)-+      Nothing   -> finally (fmap Just (receiveWait matchers)) (unmonitor mRef)

@@ -1,10 +1,11 @@ {-# LANGUAGE ExistentialQuantification  #-} {-# LANGUAGE ScopedTypeVariables        #-}+{-# LANGUAGE PatternGuards              #-}  ----------------------------------------------------------------------------- -- | -- Module      :  Control.Distributed.Process.ManagedProcess.Server--- Copyright   :  (c) Tim Watson 2012 - 2013+-- Copyright   :  (c) Tim Watson 2012 - 2017 -- License     :  BSD3 (see the file LICENSE) -- -- Maintainer  :  Tim Watson <watson.timothy@gmail.com>@@ -29,6 +30,8 @@   , stopWith   , replyTo   , replyChan+  , reject+  , rejectWith     -- * Stateless actions   , noReply_   , haltNoReply_@@ -64,12 +67,17 @@     -- * Working with Control Channels   , handleControlChan   , handleControlChan_+    -- * Working with external/STM actions+  , handleExternal+  , handleExternal_+  , handleCallExternal   ) where +import Control.Concurrent.STM (STM, atomically) 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.ManagedProcess.Internal.Types hiding (liftIO, lift) import Control.Distributed.Process.Extras   ( ExitReason(..)   , Routable(..)@@ -81,6 +89,11 @@ -- Producing ProcessAction and ProcessReply from inside handler expressions   -- -------------------------------------------------------------------------------- +-- note [Message type]: Since we own both client and server portions of the+-- codebase, we know for certain which types will be passed to which kinds+-- of handler, so the catch-all cases that @die $ "THIS_CAN_NEVER_HAPPEN"@ and+-- such, are relatively sane despite appearances!+ -- | 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.@@ -102,52 +115,64 @@ input :: forall s m. (Serializable m) => (m -> Bool) -> Condition s m input = Input +-- | Reject the message we're currently handling.+reject :: forall r s . s -> String -> Reply r s+reject st rs = continue st >>= return . ProcessReject rs++-- | Reject the message we're currently handling, giving an explicit reason.+rejectWith :: forall r m s . (Show r) => s -> r -> Reply m s+rejectWith st rs = reject st (show rs)+ -- | Instructs the process to send a reply and continue running.-reply :: (Serializable r) => r -> s -> Process (ProcessReply r s)+reply :: (Serializable r) => r -> s -> Reply 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)+          -> Reply 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 :: (Serializable r) => ProcessAction s -> Reply 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_ :: forall s r . (Serializable r) => s -> Reply 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_ :: Serializable r => ExitReason -> Reply r s haltNoReply_ r = stop r >>= noReply  -- | Instructs the process to continue running and receiving messages.-continue :: s -> Process (ProcessAction s)+continue :: s -> Action s continue = return . ProcessContinue  -- | Version of 'continue' that can be used in handlers that ignore process state. ---continue_ :: (s -> Process (ProcessAction s))+continue_ :: (s -> Action 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)+--+-- Note that @timeoutAfter NoDelay@ will cause the timeout handler to execute+-- immediately if no messages are present in the process' mailbox.+--+timeoutAfter :: Delay -> s -> Action 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_ :: StatelessHandler s Delay timeoutAfter_ d = return . ProcessTimeout d  -- | Instructs the process to /hibernate/ for the given 'TimeInterval'. Note@@ -161,24 +186,24 @@ -- -- > action (\(HibernatePlease delay) -> hibernate_ delay) ---hibernate_ :: TimeInterval -> (s -> Process (ProcessAction s))+hibernate_ :: StatelessHandler s TimeInterval 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 :: ExitReason -> Action 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 -> ExitReason -> Action 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_ :: StatelessHandler s ExitReason stop_ r _ = stop r  -- | Sends a reply explicitly to a caller.@@ -222,17 +247,10 @@     -> Dispatcher s handleCallIf_ cond handler   = DispatchIf {-      dispatch   = doHandle handler+      dispatch   = \s (CallMessage p c) -> handler p >>= mkCallReply c s     , 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]-+  where         -- 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)@@ -248,59 +266,45 @@ -- > handleCall = handleCallIf (const True) -- handleCall :: (Serializable a, Serializable b)-           => (s -> a -> Process (ProcessReply b s))+           => CallHandler s a b            -> 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+-- 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))+    -> CallHandler s a b         -- ^ a reply yielding function over the process state and input message     -> Dispatcher s handleCallIf cond handler-  = DispatchIf {-      dispatch   = doHandle handler+  = DispatchIf+    { dispatch   = \s (CallMessage p c) -> handler s p >>= mkReply c     , 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))+                => StatelessCallHandler s a b                 -> 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))+                  => Condition s a+                  -> StatelessCallHandler s a b                   -> Dispatcher s-handleCallFromIf_ c h =+handleCallFromIf_ cond handler =   DispatchIf {-      dispatch   = doHandle h-    , dispatchIf = checkCall c+      dispatch   = \_ (CallMessage p c) -> handler c p >>= mkReply c+    , dispatchIf = checkCall cond     }-  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.,@@ -309,7 +313,7 @@ -- 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))+           => DeferredCallHandler s a b            -> Dispatcher s handleCallFrom = handleCallFromIf $ state (const True) @@ -318,21 +322,14 @@ -- 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))+    -> DeferredCallHandler s a b         -- ^ a reply yielding function over the process state, sender and input message     -> Dispatcher s handleCallFromIf cond handler   = DispatchIf {-      dispatch   = doHandle handler+      dispatch   = \s (CallMessage p c) -> handler c s p >>= mkReply c     , 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@@ -340,7 +337,7 @@ -- reply to the @SendPort@. -- handleRpcChan :: forall s a b . (Serializable a, Serializable b)-              => (s -> SendPort b -> a -> Process (ProcessAction s))+              => ChannelHandler s a b               -> Dispatcher s handleRpcChan = handleRpcChanIf $ input (const True) @@ -349,42 +346,39 @@ -- handleRpcChanIf :: forall s a b . (Serializable a, Serializable b)                 => Condition s a-                -> (s -> SendPort b -> a -> Process (ProcessAction s))+                -> ChannelHandler s a b                 -> Dispatcher s-handleRpcChanIf c h+handleRpcChanIf cond handler   = DispatchIf {-      dispatch   = doHandle h-    , dispatchIf = checkRpc c+      dispatch   = \s (ChanMessage p c) -> handler c s p+    , dispatchIf = checkRpc cond     }-  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)+handleRpcChan_ :: forall s a b . (Serializable a, Serializable b)+                  => StatelessChannelHandler s a b+                     -- (SendPort b -> a -> (s -> Action s))+                  -> Dispatcher s+handleRpcChan_ = handleRpcChanIf_ $ input (const True)  -- | 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)+handleRpcChanIf_ :: forall s a b . (Serializable a, Serializable b)+                 => Condition s a+                 -> StatelessChannelHandler s a b+                 -> Dispatcher s+handleRpcChanIf_ c h+  = DispatchIf { dispatch   = \s ((ChanMessage m p) :: Message a b) -> h p m s+               , dispatchIf = checkRpc c+               }  -- | Constructs a 'cast' handler from an ordinary function in the 'Process' -- monad. -- > handleCast = handleCastIf (const True) -- handleCast :: (Serializable a)-           => (s -> a -> Process (ProcessAction s))+           => CastHandler s a            -> Dispatcher s handleCast = handleCastIf $ input (const True) @@ -395,15 +389,79 @@ -- handleCastIf :: forall s a . (Serializable a)     => Condition s a -- ^ predicate that must be satisfied for the handler to run-    -> (s -> a -> Process (ProcessAction s))+    -> CastHandler s a        -- ^ 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)+      dispatch   = \s ((CastMessage p) :: Message a ()) -> h s p     , dispatchIf = checkCast cond     } +-- | Creates a generic input handler for @STM@ actions, from an ordinary+-- function in the 'Process' monad. The @STM a@ action tells the server how+-- to read inputs, which when presented are passed to the handler in the same+-- manner as @handleInfo@ messages would be.+--+-- Note that messages sent to the server's mailbox will never match this+-- handler, only data arriving via the @STM a@ action will.+--+-- Notably, this kind of handler can be used to pass non-serialisable data to+-- a server process. In such situations, the programmer is responsible for+-- managing the underlying @STM@ infrastructure, and the server simply composes+-- the @STM a@ action with the other reads on its mailbox, using the underlying+-- @matchSTM@ API from distributed-process.+--+-- NB: this function cannot be used with a prioristised process definition.+--+handleExternal :: forall s a . (Serializable a)+               => STM a+               -> ActionHandler s a+               -> ExternDispatcher s+handleExternal a h =+  let matchMsg'   = matchSTM a (\(m :: r) -> return $ unsafeWrapMessage m)+      matchAny' f = matchSTM a (\(m :: r) -> return $ f (unsafeWrapMessage m)) in+  DispatchSTM+    { stmAction   = a+    , dispatchStm = h+    , matchStm    = matchMsg'+    , matchAnyStm = matchAny'+    }++-- | Version of @handleExternal@ that ignores state.+handleExternal_ :: forall s a . (Serializable a)+                => STM a+                -> StatelessHandler s a+                -> ExternDispatcher s+handleExternal_ a h = handleExternal a (flip h)++-- | Handle @call@ style API interactions using arbitrary /STM/ actions.+--+-- The usual @CallHandler@ is preceded by an stm action that, when evaluated,+-- yields a value, and a second expression that is used to send a reply back+-- to the /caller/. The corrolary client API is /callSTM/.+--+handleCallExternal :: forall s r w . (Serializable r)+                   => STM r+                   -> (w -> STM ())+                   -> CallHandler s r w+                   -> ExternDispatcher s+handleCallExternal reader writer handler =+  let matchMsg'   = matchSTM reader (\(m :: r) -> return $ unsafeWrapMessage m)+      matchAny' f = matchSTM reader (\(m :: r) -> return $ f $ unsafeWrapMessage m) in+  DispatchSTM+    { stmAction   = reader+    , dispatchStm = doStmReply handler+    , matchStm    = matchMsg'+    , matchAnyStm = matchAny'+    }+  where+    doStmReply d s m = d s m >>= doXfmReply writer++    doXfmReply _ (NoReply a)         = return a+    doXfmReply _ (ProcessReject _ a) = return a+    doXfmReply w (ProcessReply r' a) = liftIO (atomically $ w r') >> return a+ -- | 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'.@@ -412,40 +470,41 @@ -- handleControlChan :: forall s a . (Serializable a)     => ControlChannel a -- ^ the receiving end of the control channel-    -> (s -> a -> Process (ProcessAction s))+    -> ActionHandler s a        -- ^ an action yielding function over the process state and input message-    -> Dispatcher s+    -> ExternDispatcher s handleControlChan chan h-  = DispatchCC { channel  = snd $ unControl chan-               , dispatch = (\s ((CastMessage p) :: Message a ()) -> h s p)+  = DispatchCC { channel      = snd $ unControl chan+               , dispatchChan = \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+           -> StatelessHandler s a+           -> ExternDispatcher s handleControlChan_ chan h-  = DispatchCC { channel    = snd $ unControl chan-               , dispatch   = (\s ((CastMessage p) :: Message a ()) -> h p $ s)+  = DispatchCC { channel      = snd $ unControl chan+               , dispatchChan = \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+            => StatelessHandler s a+            -> 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)))+    -> StatelessHandler s a         -- ^ 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 { dispatch   = \s ((CastMessage p) :: Message a ()) -> h p $ s                , dispatchIf = checkCast cond                } @@ -459,18 +518,18 @@ -- @action (\MyCriticalSignal -> stop_ ExitNormal)@ -- action :: forall s a . (Serializable a)-    => (a -> (s -> Process (ProcessAction s)))+    => StatelessHandler s a           -- ^ 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))+  where perform :: ActionHandler s a         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))+               => ActionHandler s a                -> Dispatcher s handleDispatch = handleDispatchIf $ input (const True) @@ -481,33 +540,33 @@ -- handleDispatchIf :: forall s a . (Serializable a)                  => Condition s a-                 -> (s -> a -> Process (ProcessAction s))+                 -> ActionHandler s a                  -> Dispatcher s handleDispatchIf cond handler = DispatchIf {       dispatch = doHandle handler     , dispatchIf = check cond     }   where doHandle :: (Serializable a)-                 => (s -> a -> Process (ProcessAction s))+                 => ActionHandler s a                  -> 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)+                (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))+           => ActionHandler s a            -> DeferredDispatcher s handleInfo h = DeferredDispatcher { dispatchInfo = doHandleInfo h }   where     doHandleInfo :: forall s2 a2. (Serializable a2)-                             => (s2 -> a2 -> Process (ProcessAction s2))+                             => ActionHandler s2 a2                              -> s2                              -> P.Message                              -> Process (Maybe (ProcessAction s2))@@ -515,39 +574,40 @@  -- | Handle completely /raw/ input messages. ---handleRaw :: forall s. (s -> P.Message -> Process (ProcessAction s))+handleRaw :: forall s. ActionHandler s P.Message           -> DeferredDispatcher s handleRaw h = DeferredDispatcher { dispatchInfo = doHandle h }   where-    doHandle h' s msg = h' s msg >>= return . Just+    doHandle h' s msg = fmap Just (h' s msg)  -- | 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))+           => (ProcessId -> ActionHandler s a)            -> ExitSignalDispatcher s handleExit h = ExitSignalDispatcher { dispatchExit = doHandleExit h }   where-    doHandleExit :: (s -> ProcessId -> a -> Process (ProcessAction s))+    doHandleExit :: (ProcessId -> ActionHandler s a)                  -> s                  -> ProcessId                  -> P.Message                  -> Process (Maybe (ProcessAction s))-    doHandleExit h' s p msg = handleMessage msg (h' s p)+    doHandleExit h' s p msg = handleMessage msg (h' p s) +-- | Conditional version of @handleExit@ handleExitIf :: forall s a . (Serializable a)              => (s -> a -> Bool)-             -> (s -> ProcessId -> a -> Process (ProcessAction s))+             -> (ProcessId -> ActionHandler s a)              -> ExitSignalDispatcher s handleExitIf c h = ExitSignalDispatcher { dispatchExit = doHandleExit c h }   where     doHandleExit :: (s -> a -> Bool)-                 -> (s -> ProcessId -> a -> Process (ProcessAction s))+                 -> (ProcessId -> ActionHandler s a)                  -> s                  -> ProcessId                  -> P.Message                  -> Process (Maybe (ProcessAction s))-    doHandleExit c' h' s p msg = handleMessageIf msg (c' s) (h' s p)+    doHandleExit c' h' s p msg = handleMessageIf msg (c' s) (h' p s)  -- 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@@ -555,8 +615,13 @@         => CallRef b         -> ProcessReply b s         -> Process (ProcessAction s)-mkReply _ (NoReply a)         = return a-mkReply c (ProcessReply r' a) = sendTo c r' >> return a+mkReply cRef act+  | (NoReply a)          <- act  = return a+  | (CallRef (_, tg'))   <- cRef+  , (ProcessReply  r' a) <- act  = sendTo cRef (CallResponse r' tg') >> return a+  | (CallRef (_, ct'))   <- cRef+  , (ProcessReject r' a) <- act  = sendTo cRef (CallRejected r' ct') >> return a+  | otherwise                    = die $ ExitOther "mkReply.InvalidState"  -- these functions are the inverse of 'condition', 'state' and 'input' @@ -597,4 +662,3 @@ decode (CallMessage a _) = a decode (CastMessage a)   = a decode (ChanMessage a _) = a-

@@ -1,38 +1,265 @@ {-# LANGUAGE ExistentialQuantification  #-} {-# LANGUAGE ScopedTypeVariables        #-} {-# LANGUAGE PatternGuards              #-}+{-# LANGUAGE RecordWildCards            #-}+{-# LANGUAGE EmptyDataDecls             #-}+{-# LANGUAGE StandaloneDeriving         #-} +-----------------------------------------------------------------------------+-- |+-- Module      :  Control.Distributed.Process.ManagedProcess.Server.Priority+-- Copyright   :  (c) Tim Watson 2012 - 2017+-- License     :  BSD3 (see the file LICENSE)+--+-- Maintainer  :  Tim Watson <watson.timothy@gmail.com>+-- Stability   :  experimental+-- Portability :  non-portable (requires concurrency)+--+-- The Prioritised Server portion of the /Managed Process/ API.+----------------------------------------------------------------------------- module Control.Distributed.Process.ManagedProcess.Server.Priority-  ( prioritiseCall+  ( -- * Prioritising API Handlers+    prioritiseCall   , prioritiseCall_   , prioritiseCast   , prioritiseCast_   , prioritiseInfo   , prioritiseInfo_   , setPriority+    -- * Creating Filters+  , check+  , raw+  , raw_+  , api+  , api_+  , info+  , info_+  , refuse+  , reject+  , rejectApi+  , store+  , storeM+  , crash+  , ensure+  , ensureM+  , Filter()+  , DispatchFilter()+  , Message()+  , evalAfter+  , currentTimeout+  , processState+  , processDefinition+  , processFilters+  , processUnhandledMsgPolicy+  , setUserTimeout+  , setProcessState+  , GenProcess+  , peek+  , push+  , addUserTimer   ) where  import Control.Distributed.Process hiding (call, Message) import qualified Control.Distributed.Process as P (Message)+import Control.Distributed.Process.Extras+  ( ExitReason(..)+  )+import Control.Distributed.Process.ManagedProcess.Internal.GenProcess+  ( addUserTimer+  , currentTimeout+  , processState+  , processDefinition+  , processFilters+  , processUnhandledMsgPolicy+  , setUserTimeout+  , setProcessState+  , GenProcess+  , peek+  , push+  , evalAfter+  ) import Control.Distributed.Process.ManagedProcess.Internal.Types import Control.Distributed.Process.Serializable import Prelude hiding (init) +-- | Sent to a caller in cases where the server is rejecting an API input and+-- a @Recipient@ is available (i.e. a /call/ message handling filter).+data RejectedByServer = RejectedByServer deriving (Show)++-- | Represents a pair of expressions that can be used to define a @DispatchFilter@.+data FilterHandler s =+    forall m . (Serializable m) =>+    HandlePure+    {+      pureCheck :: s -> m -> Process Bool+    , handler :: s -> m -> Process (Filter s)+    } -- ^ A pure handler, usable where the target handler is based on @handleInfo@+  | forall m b . (Serializable m, Serializable b) =>+    HandleApi+    {+      apiCheck :: s -> m -> Process Bool+    , apiHandler :: s -> Message m b -> Process (Filter s)+    } -- ^ An API handler, usable where the target handler is based on @handle{Call, Cast, RpcChan}@+  | HandleRaw+    {+      rawCheck :: s -> P.Message -> Process Bool+    , rawHandler :: s -> P.Message -> Process (Maybe (Filter s))+    } -- ^ A raw handler, usable where the target handler is based on @handleRaw@+  | HandleState { stateHandler :: s -> Process (Maybe (Filter s)) }++{-+check :: forall c s m . (Check c s m)+      => c -> (s -> Process (Filter s)) -> s -> m -> Process (Filter s)+-}++-- | Create a filter from a @FilterHandler@.+check :: forall s . FilterHandler s -> DispatchFilter s+check h+  | HandlePure{..}  <- h = FilterAny $ \s m -> pureCheck s m >>= procUnless s m handler+  | HandleRaw{..}   <- h = FilterRaw $ \s m -> do+      c <- rawCheck s m+      if c then return $ Just $ FilterOk s+           else rawHandler s m+  | HandleState{..} <- h = FilterState stateHandler+  | HandleApi{..}   <- h = FilterApi $ \s m@(CallMessage m' _) -> do+      c <- apiCheck s m'+      if c then return $ FilterOk s+           else apiHandler s m++  where+    procUnless s _ _ True  = return $ FilterOk s+    procUnless s m h' False = h' s m++-- | A raw filter (targetting raw messages).+raw :: forall s .+       (s -> P.Message -> Process Bool)+    -> (s -> P.Message -> Process (Maybe (Filter s)))+    -> FilterHandler s+raw = HandleRaw++-- | A raw filter that ignores the server state in its condition expression.+raw_ :: forall s .+        (P.Message -> Process Bool)+     -> (s -> P.Message -> Process (Maybe (Filter s)))+     -> FilterHandler s+raw_ c h = raw (const $ c) h++-- | An API filter (targetting /call/, /cast/, and /chan/ messages).+api :: forall s m b . (Serializable m, Serializable b)+    => (s -> m -> Process Bool)+    -> (s -> Message m b -> Process (Filter s))+    -> FilterHandler s+api = HandleApi++-- | An API filter that ignores the server state in its condition expression.+api_ :: forall m b s . (Serializable m, Serializable b)+     => (m -> Process Bool)+     -> (s -> Message m b -> Process (Filter s))+     -> FilterHandler s+api_ c h = api (const $ c) h++-- | An info filter (targetting info messages of a specific type)+info :: forall s m . (Serializable m)+        => (s -> m -> Process Bool)+        -> (s -> m -> Process (Filter s))+        -> FilterHandler s+info = HandlePure++-- | An info filter that ignores the server state in its condition expression.+info_ :: forall s m . (Serializable m)+        => (m -> Process Bool)+        -> (s -> m -> Process (Filter s))+        -> FilterHandler s+info_ c h = info (const $ c) h++-- | Create a filter expression that will reject all messages of a specific type.+reject :: forall s m r . (Show r)+       => r -> s -> m -> Process (Filter s)+reject r = \s _ -> do return $ FilterReject (show r) s++-- | Create a filter expression that will crash (i.e. stop) the server.+crash :: forall s . s -> ExitReason -> Process (Filter s)+crash s r = return $ FilterStop s r++-- | A version of @reject@ that deals with API messages (i.e. /call/, /cast/, etc)+-- and in the case of a /call/ interaction, will reject the messages and reply to+-- the sender accordingly (with @CallRejected@).+rejectApi :: forall s m b r . (Show r, Serializable m, Serializable b)+          => r -> s -> Message m b -> Process (Filter s)+rejectApi r = \s m -> do let r' = show r+                         rejectToCaller m r'+                         return $ FilterSkip s++-- | Modify the server state every time a message is recieved.+store :: (s -> s) -> DispatchFilter s+store f = FilterState $ return . Just . FilterOk . f++-- | Motify the server state when messages of a certain type arrive...+storeM :: forall s m . (Serializable m)+       => (s -> m -> Process s)+       -> DispatchFilter s+storeM proc = check $ HandlePure (\_ _ -> return True)+                                 (\s m -> proc s m >>= return . FilterOk)++-- | Refuse messages for which the given expression evaluates to @True@.+refuse :: forall s m . (Serializable m)+       => (m -> Bool)+       -> DispatchFilter s+refuse c = check $ info (const $ \m -> return $ c m) (reject RejectedByServer)++{-++apiCheck :: forall s m r . (Serializable m, Serializable r)+      => (s -> Message m r -> Bool)+      -> (s -> Message m r -> Process (Filter s))+      -> DispatchFilter s+apiCheck c h = checkM (\s m -> return $ c s m) h+-}++-- | Ensure that the server state is consistent with the given expression each+-- time a message arrives/is processed. If the expression evaluates to @True@+-- then the filter will evaluate to "FilterOk", otherwise "FilterStop" (which+-- will cause the server loop to stop with @ExitOther filterFail@).+ensure :: forall s . (s -> Bool) -> DispatchFilter s+ensure c =+  check $ HandleState { stateHandler = (\s -> if c s+                                                then return $ Just $ FilterOk s+                                                else return $ Just $ FilterStop s filterFail)+                      }+-- | As @ensure@ but runs in the @Process@ monad, and matches only inputs of type @m@.+ensureM :: forall s m . (Serializable m) => (s -> m -> Process Bool) -> DispatchFilter s+ensureM c =+  check $ HandlePure { pureCheck = c+                     , handler = (\s _ -> return $ FilterStop s filterFail) :: s -> m -> Process (Filter s)+                     }++-- TODO: add the type rep for a more descriptive failure message++filterFail :: ExitReason+filterFail = ExitOther "Control.Distributed.Process.ManagedProcess.Priority:FilterFailed"++-- | Sets an explicit priority from 1..100. Values > 100 are rounded to 100,+-- and values < 1 are set to 0. setPriority :: Int -> Priority m-setPriority = Priority+setPriority n+  | n < 1     = Priority 0+  | n > 100   = Priority 100+  | otherwise = Priority n +-- | Prioritise a call handler, ignoring the server's state prioritiseCall_ :: forall s a b . (Serializable a, Serializable b)                 => (a -> Priority b)                 -> DispatchPriority s-prioritiseCall_ h = prioritiseCall (\_ -> h)+prioritiseCall_ h = prioritiseCall (const h) +-- | Prioritise a call handler prioritiseCall :: forall s a b . (Serializable a, Serializable b)                => (s -> a -> Priority b)                -> DispatchPriority s-prioritiseCall h = PrioritiseCall (\s -> unCall $ h s)+prioritiseCall h = PrioritiseCall (unCall . h)   where     unCall :: (a -> Priority b) -> P.Message -> Process (Maybe (Int, P.Message))-    unCall h' m = unwrapMessage m >>= return . matchPrioritise m h'+    unCall h' m = fmap (matchPrioritise m h') (unwrapMessage m)      matchPrioritise :: P.Message                     -> (a -> Priority b)@@ -45,18 +272,20 @@       , False <- isEncoded msg = Just (getPrio $ p m, msg)       | otherwise              = Nothing +-- | Prioritise a cast handler, ignoring the server's state prioritiseCast_ :: forall s a . (Serializable a)                 => (a -> Priority ())                 -> DispatchPriority s-prioritiseCast_ h = prioritiseCast (\_ -> h)+prioritiseCast_ h = prioritiseCast (const h) +-- | Prioritise a cast handler prioritiseCast :: forall s a . (Serializable a)                => (s -> a -> Priority ())                -> DispatchPriority s-prioritiseCast h = PrioritiseCast (\s -> unCast $ h s)+prioritiseCast h = PrioritiseCast (unCast . h)   where     unCast :: (a -> Priority ()) -> P.Message -> Process (Maybe (Int, P.Message))-    unCast h' m = unwrapMessage m >>= return . matchPrioritise m h'+    unCast h' m = fmap (matchPrioritise m h') (unwrapMessage m)      matchPrioritise :: P.Message                     -> (a -> Priority ())@@ -69,18 +298,20 @@       , False <- isEncoded msg = Just (getPrio $ p m, msg)       | otherwise              = Nothing +-- | Prioritise an info handler, ignoring the server's state prioritiseInfo_ :: forall s a . (Serializable a)                 => (a -> Priority ())                 -> DispatchPriority s-prioritiseInfo_ h = prioritiseInfo (\_ -> h)+prioritiseInfo_ h = prioritiseInfo (const h) +-- | Prioritise an info handler prioritiseInfo :: forall s a . (Serializable a)                => (s -> a -> Priority ())                -> DispatchPriority s-prioritiseInfo h = PrioritiseInfo (\s -> unMsg $ h s)+prioritiseInfo h = PrioritiseInfo (unMsg . h)   where     unMsg :: (a -> Priority ()) -> P.Message -> Process (Maybe (Int, P.Message))-    unMsg h' m = unwrapMessage m >>= return . matchPrioritise m h'+    unMsg h' m = fmap (matchPrioritise m h') (unwrapMessage m)      matchPrioritise :: P.Message                     -> (a -> Priority ())@@ -92,4 +323,3 @@       | (Just m') <- msgIn       , False <- isEncoded msg = Just (getPrio $ p m', msg)       | otherwise              = Nothing-

@@ -6,7 +6,7 @@ ----------------------------------------------------------------------------- -- | -- Module      :  Control.Distributed.Process.ManagedProcess.Server.Restricted--- Copyright   :  (c) Tim Watson 2012 - 2013+-- Copyright   :  (c) Tim Watson 2012 - 2017 -- License     :  BSD3 (see the file LICENSE) -- -- Maintainer  :  Tim Watson <watson.timothy@gmail.com>@@ -59,12 +59,11 @@   , 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 Control.Distributed.Process.ManagedProcess.Internal.Types hiding (lift) import qualified Control.Distributed.Process.ManagedProcess.Server as Server import Control.Distributed.Process.Extras.Time import Control.Distributed.Process.Serializable@@ -73,7 +72,6 @@ import Control.Monad.IO.Class (MonadIO) import qualified Control.Monad.State as ST   ( MonadState-  , MonadTrans   , StateT   , get   , lift@@ -113,7 +111,7 @@  -- | Log a trace message using the underlying Process's @say@ say :: String -> RestrictedProcess s ()-say msg = lift . P.say $ msg+say = lift . P.say  -- | Get the current process state getState :: RestrictedProcess s s@@ -140,7 +138,7 @@ noReply :: forall s r . (Serializable r)            => Result r            -> RestrictedProcess s (Result r)-noReply r = return r+noReply = return  -- | Halt process execution during a call handler, without paying any attention -- to the expected return type.@@ -189,7 +187,7 @@ -- that takes a handler which executes in 'RestrictedProcess'. -- handleCallIf :: forall s a b . (Serializable a, Serializable b)-             => (Condition s a)+             => Condition s a              -> (a -> RestrictedProcess s (Result b))              -> Dispatcher s handleCallIf cond h = Server.handleCallIf cond (wrapCall h)@@ -221,11 +219,13 @@ -- cast and info look the same to a restricted process handleInfo h = Server.handleInfo (wrapHandler h) +-- | Handle exit signals handleExit :: forall s a. (Serializable a)            => (a -> RestrictedProcess s RestrictedAction)            -> ExitSignalDispatcher s-handleExit h = Server.handleExit $ \s _ a -> (wrapHandler h) s a+handleExit h = Server.handleExit $ \_ s a -> wrapHandler h s a +-- | Handle timeouts handleTimeout :: forall s . (Delay -> RestrictedProcess s RestrictedAction)                          -> TimeoutHandler s handleTimeout h = \s d -> do@@ -242,9 +242,7 @@  wrapHandler :: forall s a . (Serializable a)             => (a -> RestrictedProcess s RestrictedAction)-            -> s-            -> a-            -> Process (ProcessAction s)+            -> ActionHandler s a wrapHandler h s a = do   (r, s') <- runRestricted s (h a)   case r of@@ -255,9 +253,7 @@  wrapCall :: forall s a b . (Serializable a, Serializable b)             => (a -> RestrictedProcess s (Result b))-            -> s-            -> a-            -> Process (ProcessReply b s)+            -> CallHandler s a b wrapCall h s a = do   (r, s') <- runRestricted s (h a)   case r of@@ -272,4 +268,3 @@ -- | TODO MonadTrans instance? lift :: (Monad m) => m a -> t m a lift :: Process a -> RestrictedProcess s a lift p = RestrictedProcess $ ST.lift p-

@@ -0,0 +1,187 @@+{-# LANGUAGE ScopedTypeVariables        #-}+{-# LANGUAGE PatternGuards              #-}+{-# LANGUAGE BangPatterns               #-}+{-# LANGUAGE RecordWildCards            #-}+{-# LANGUAGE DeriveDataTypeable         #-}+{-# LANGUAGE DeriveGeneric              #-}++-----------------------------------------------------------------------------+-- |+-- Module      :  Control.Distributed.Process.ManagedProcess.Timer+-- Copyright   :  (c) Tim Watson 2017+-- License     :  BSD3 (see the file LICENSE)+--+-- Maintainer  :  Tim Watson <watson.timothy@gmail.com>+-- Stability   :  experimental+-- Portability :  non-portable (requires concurrency)+--+-- This module provides a wrap around a simple 'Timer' that can be started,+-- stopped, reset, cleared, and read. A convenient function is provided for+-- creating a @Match@ expression for the timer.+--+-- [Notes]+--+-- The timers defined in this module are based on a @TVar Bool@. When the+-- client program is @-threaded@ (i.e. @rtsSupportsBoundThreads == True@), then+-- the timers are set using @registerDelay@, which is very efficient and relies+-- only no the RTS IO Manager. When we're not @-threaded@, we fall back to using+-- "Control.Distributed.Process.Extras.Timer" to set the @TVar@, which has much+-- the same effect, but requires us to spawn a process to handle setting the+-- @TVar@ - a process which could theoretically die before setting the variable.+--+module Control.Distributed.Process.ManagedProcess.Timer+  ( Timer(timerDelay)+  , TimerKey+  , delayTimer+  , startTimer+  , stopTimer+  , resetTimer+  , clearTimer+  , matchTimeout+  , matchKey+  , matchRun+  , isActive+  , readTimer+  , TimedOut(..)+  ) where++import Control.Concurrent (rtsSupportsBoundThreads)+import Control.Concurrent.STM hiding (check)+import Control.Distributed.Process+  ( matchSTM+  , Process+  , ProcessId+  , Match+  , Message+  , liftIO+  )+import qualified Control.Distributed.Process as P+  ( liftIO+  )+import Control.Distributed.Process.Extras.Time (asTimeout, Delay(..))+import Control.Distributed.Process.Extras.Timer+  ( cancelTimer+  , runAfter+  , TimerRef+  )+import Data.Binary (Binary)+import Data.Maybe (isJust, fromJust)+import Data.Typeable (Typeable)+import GHC.Conc (registerDelay)+import GHC.Generics++--------------------------------------------------------------------------------+-- Timeout Management                                                         --+--------------------------------------------------------------------------------++-- | A key for storing timers in prioritised process backing state.+type TimerKey = Int++-- | Used during STM reads on Timers and to implement blocking. Since timers+-- can be associated with a "TimerKey", the second constructor for this type+-- yields a key indicating whic "Timer" it refers to. Note that the user is+-- responsible for establishing and maintaining the mapping between @Timer@s+-- and their keys.+data TimedOut = TimedOut | Yield TimerKey+  deriving (Eq, Show, Typeable, Generic)+instance Binary TimedOut where++-- | We hold timers in 2 states, each described by a Delay.+-- isActive = isJust . mtSignal+-- the TimerRef is optional since we only use the Timer module from extras+-- when we're unable to registerDelay (i.e. not running under -threaded)+data Timer = Timer { timerDelay :: Delay+                   , mtPidRef   :: Maybe TimerRef+                   , mtSignal   :: Maybe (TVar Bool)+                   }++-- | @True@ if a @Timer@ is currently active.+isActive :: Timer -> Bool+isActive = isJust . mtSignal++-- | Creates a default @Timer@ which is inactive.+delayTimer :: Delay -> Timer+delayTimer d = Timer d noPid noTVar+  where+    noPid  = Nothing :: Maybe ProcessId+    noTVar = Nothing :: Maybe (TVar Bool)++-- | Starts a @Timer@+-- Will use the GHC @registerDelay@ API if @rtsSupportsBoundThreads == True@+startTimer :: Delay -> Process Timer+startTimer d+  | Delay t <- d = establishTimer t+  | otherwise    = return $ delayTimer d+  where+    establishTimer t'+      | rtsSupportsBoundThreads = do sig <- liftIO $ registerDelay (asTimeout t')+                                     return Timer { timerDelay = d+                                                  , mtPidRef = Nothing+                                                  , mtSignal = Just sig+                                                  }+      | otherwise = do+          tSig  <- liftIO $ newTVarIO False+          -- NB: runAfter spawns a process, which is defined in terms of+          -- expectTimeout (asTimeout t) :: Process (Maybe CancelTimer)+          --+          tRef <- runAfter t' $ P.liftIO $ atomically $ writeTVar tSig True+          return Timer { timerDelay  = d+                       , mtPidRef = Just tRef+                       , mtSignal = Just tSig+                       }++-- | Stops a previously started @Timer@. Has no effect if the @Timer@ is inactive.+stopTimer :: Timer -> Process Timer+stopTimer t@Timer{..} = do+  clearTimer mtPidRef+  return t { mtPidRef = Nothing+           , mtSignal = Nothing+           }++-- | Clears and restarts a @Timer@.+resetTimer :: Timer -> Delay -> Process Timer+resetTimer Timer{..} d = clearTimer mtPidRef >> startTimer d++-- | Clears/cancels a running timer. Has no effect if the @Timer@ is inactive.+clearTimer :: Maybe TimerRef -> Process ()+clearTimer ref+  | isJust ref = cancelTimer (fromJust ref)+  | otherwise  = return ()++-- | Creates a @Match@ for a given timer, for use with Cloud Haskell's messaging+-- primitives for selective receives.+matchTimeout :: Timer -> [Match (Either TimedOut Message)]+matchTimeout t@Timer{..}+    | isActive t = [ matchSTM (readTimer $ fromJust mtSignal)+                              (return . Left) ]+    | otherwise  = []++-- | Create a match expression for a given @Timer@. When the timer expires+-- (i.e. the "TVar Bool" is set to @True@), the "Match" will return @Yield i@,+-- where @i@ is the given "TimerKey".+matchKey :: TimerKey -> Timer -> [Match (Either TimedOut Message)]+matchKey i t@Timer{..}+  | isActive t = [matchSTM (readTVar (fromJust mtSignal) >>= \expired ->+                               if expired then return (Yield i) else retry)+                           (return . Left)]+  | otherwise  = []++-- | As "matchKey", but instead of a returning @Yield i@, the generated "Match"+-- handler evaluates the first argument - and expression from "TimerKey" to+-- @Process Message@ - to determine its result.+matchRun :: (TimerKey -> Process Message)+         -> TimerKey+         -> Timer+         -> [Match Message]+matchRun f k t@Timer{..}+  | isActive t = [matchSTM (readTVar (fromJust mtSignal) >>= \expired ->+                               if expired then return k else retry) f]+  | otherwise  = []++-- | Reads a given @TVar Bool@ for a timer, and returns @STM TimedOut@ once the+-- variable is set to true. Will @retry@ in the meanwhile.+readTimer :: TVar Bool -> STM TimedOut+readTimer t = do+   expired <- readTVar t+   if expired then return TimedOut+              else retry

@@ -1,10 +1,11 @@ {-# LANGUAGE ExistentialQuantification  #-} {-# LANGUAGE ScopedTypeVariables        #-}+{-# LANGUAGE LiberalTypeSynonyms        #-}  ----------------------------------------------------------------------------- -- | -- Module      :  Control.Distributed.Process.ManagedProcess.UnsafeClient--- Copyright   :  (c) Tim Watson 2012 - 2013+-- Copyright   :  (c) Tim Watson 2012 - 2017 -- License     :  BSD3 (see the file LICENSE) -- -- Maintainer  :  Tim Watson <watson.timothy@gmail.com>@@ -61,6 +62,9 @@   , terminate   , receiveTimeout   , unsafeSendChan+  , getSelfPid+  , catchesExit+  , handleMessageIf   ) import Control.Distributed.Process.Async   ( Async@@ -72,7 +76,7 @@   , Addressable   , Routable(..)   , NFSerializable-  , ExitReason+  , ExitReason(..)   , Shutdown(..)   ) import Control.Distributed.Process.ManagedProcess.Internal.Types@@ -112,8 +116,17 @@ -- 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+safeCall s m = do+  us <- getSelfPid+  (fmap fromJust (unsafeInitCall s m >>= waitResponse Nothing) :: Process (Either ExitReason b))+    `catchesExit` [\pid msg -> handleMessageIf msg (weFailed pid us)+                                                   (return . Left)] +  where++    weFailed a b (ExitOther _) = a == b+    weFailed _ _ _             = False+ -- | Version of 'safeCall' that returns 'Nothing' if the operation fails. --  Uses /unsafe primitives/. tryCall :: forall s a b . (Addressable s, NFSerializable a, NFSerializable b)@@ -135,11 +148,12 @@         decodeResult (Just (Right result)) = return $ Just result         decodeResult (Just (Left reason))  = die reason +-- | Block for @TimeInterval@ waiting for any matching @CallResponse@ flushPendingCalls :: forall b . (NFSerializable b)                   => TimeInterval                   -> (b -> Process b)                   -> Process (Maybe b)-flushPendingCalls d proc = do+flushPendingCalls d proc =   receiveTimeout (asTimeout d) [       match (\(CallResponse (m :: b) _) -> proc m)     ]@@ -165,6 +179,7 @@   unsafeSendTo server ((ChanMessage msg sp) :: Message a b)   return rp +-- | A synchronous version of 'callChan'. syncCallChan :: forall s a b . (Addressable s, NFSerializable a, NFSerializable b)          => s -> a -> Process b syncCallChan server msg = do@@ -173,9 +188,10 @@     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, NFSerializable a, NFSerializable b)             => s -> a -> Process (Either ExitReason b) syncSafeCallChan server msg = do   rp <- callChan server msg   awaitResponse server [ matchChan rp (return . Right) ]-

@@ -0,0 +1,94 @@+{-# LANGUAGE DeriveDataTypeable   #-}+{-# LANGUAGE DeriveGeneric        #-}+{-# LANGUAGE TemplateHaskell      #-}+{-# LANGUAGE ScopedTypeVariables  #-}+{-# LANGUAGE BangPatterns         #-}++module Counter+  ( startCounter,+    getCount,+    incCount,+    resetCount,+    wait,+    waitTimeout+  ) where++import Control.Distributed.Process hiding (call)+import Control.Distributed.Process.Async+import Control.Distributed.Process.Extras+import Control.Distributed.Process.Extras.Time+import Control.Distributed.Process.ManagedProcess+import Data.Binary+import Data.Typeable (Typeable)++import GHC.Generics++--------------------------------------------------------------------------------+-- Types                                                                      --+--------------------------------------------------------------------------------++-- Call and Cast request types. Response types are unnecessary as the GenProcess+-- API uses the Async API, which in turn guarantees that an async handle can+-- /only/ give back a reply for that *specific* request through the use of an+-- anonymous middle-man (as the sender and receiver in our case).++data Increment = Increment+  deriving (Typeable, Generic, Eq, Show)+instance Binary Increment where++data Fetch = Fetch+  deriving (Typeable, Generic, Eq, Show)+instance Binary Fetch where++data Reset = Reset+  deriving (Typeable, Generic, Eq, Show)+instance Binary Reset where++type State = Int++--------------------------------------------------------------------------------+-- API                                                                        --+--------------------------------------------------------------------------------++-- | Increment count+incCount :: ProcessId -> Process Int+incCount sid = call sid Increment++-- | Get the current count - this is replicating what 'call' actually does+getCount :: ProcessId -> Process Int+getCount sid = call sid Fetch++-- | Reset the current count+resetCount :: ProcessId -> Process ()+resetCount sid = cast sid Reset++-- | Start a counter server+startCounter :: Int -> Process ProcessId+startCounter startCount =+  let server = serverDefinition+  in spawnLocal $ serve startCount init' server+  where init' :: InitHandler Int Int+        init' count = return $ InitOk count Infinity++--------------------------------------------------------------------------------+-- Implementation                                                             --+--------------------------------------------------------------------------------++serverDefinition :: ProcessDefinition State+serverDefinition = defaultProcess {+     apiHandlers = [+          handleCallIf (condition (\count Increment -> count >= 10))-- invariant+                       (\_ (_ :: Increment) -> haltMaxCount)++        , handleCall handleIncrement+        , handleCall (\count Fetch -> reply count count)+        , handleCast (\_ Reset -> continue 0)+        ]+    } :: ProcessDefinition State++haltMaxCount :: Reply Int State+haltMaxCount = haltNoReply_ (ExitOther "Count > 10")++handleIncrement :: CallHandler State Increment Int +handleIncrement count Increment =+    let next = count + 1 in continue next >>= replyWith next

@@ -0,0 +1,372 @@+{-# LANGUAGE ScopedTypeVariables #-}+{-# LANGUAGE FlexibleInstances   #-}+{-# LANGUAGE RecordWildCards     #-}++module ManagedProcessCommon where++import Control.Concurrent.MVar (MVar)+import Control.Concurrent.STM.TQueue+ ( newTQueueIO+ , readTQueue+ , writeTQueue+ , TQueue+ )+import Control.Distributed.Process hiding (call, send)+import Control.Distributed.Process.Extras hiding (monitor)+import qualified Control.Distributed.Process as P+import Control.Distributed.Process.SysTest.Utils+import Control.Distributed.Process.Extras.Time+import Control.Distributed.Process.Extras.Timer+import Control.Distributed.Process.Async+import Control.Distributed.Process.ManagedProcess+import qualified Control.Distributed.Process.ManagedProcess.UnsafeClient as Unsafe+import Control.Distributed.Process.Serializable()++#if ! MIN_VERSION_base(4,6,0)+import Prelude hiding (catch)+#endif++import TestUtils++type Launcher a = a -> Process (ProcessId, MVar ExitReason)++explodingTestProcess :: ProcessId -> ProcessDefinition ()+explodingTestProcess pid =+  statelessProcess {+    apiHandlers = [+       handleCall_ (\(s :: String) ->+                     (die s) :: Process String)+     , handleCast  (\_ (i :: Int) ->+                     getSelfPid >>= \p -> die (p, i))+     ]+  , exitHandlers = [+       handleExit  (\_ s (m :: String) -> do send pid (m :: String)+                                             continue s)+     , handleExit  (\_ s m@((_ :: ProcessId),+                            (_ :: Int)) -> P.send pid m >> continue s)+     ]+  }++standardTestServer :: UnhandledMessagePolicy -> ProcessDefinition ()+standardTestServer policy =+  statelessProcess {+        apiHandlers = [+              -- note: state is passed here, as a 'stateless' process is+              -- in fact process definition whose state is ()++              handleCastIf  (input (\msg -> msg == "stop"))+                            (\_ _ -> stop ExitNormal)++            , handleCall    (\s' (m :: String) -> reply m s')+            , handleCall_   (\(n :: Int) -> return (n * 2))    -- "stateless"++            , handleCall    (\s' (_ :: Delay) -> (reject s' "invalid-call") :: Reply () ())++            , handleCast    (\s' ("ping", pid :: ProcessId) ->+                                 send pid "pong" >> continue s')+            , handleCastIf_ (input (\(c :: String, _ :: Delay) -> c == "timeout"))+                            (\("timeout", d) -> timeoutAfter_ d)++            , handleCast_   (\("hibernate", d :: TimeInterval) -> hibernate_ d)+          ]+      , unhandledMessagePolicy = policy+      , timeoutHandler         = \_ _ -> stop $ ExitOther "timeout"+    }++wrap :: (Process (ProcessId, MVar ExitReason)) -> Launcher a+wrap it = \_ -> do it++data StmServer = StmServer { serverPid  :: ProcessId+                           , writerChan :: TQueue String+                           , readerChan :: TQueue String+                           }++instance Resolvable StmServer where+  resolve = return . Just . serverPid++echoStm :: StmServer -> String -> Process (Either ExitReason String)+echoStm StmServer{..} = callSTM serverPid+                                (writeTQueue writerChan)+                                (readTQueue  readerChan)++launchEchoServer :: CallHandler () String String -> Process StmServer+launchEchoServer handler = do+  (inQ, replyQ) <- liftIO $ do+    cIn <- newTQueueIO+    cOut <- newTQueueIO+    return (cIn, cOut)++  let procDef = statelessProcess {+                  externHandlers = [+                    handleCallExternal+                      (readTQueue inQ)+                      (writeTQueue replyQ)+                      handler+                  ]+                }++  pid <- spawnLocal $ serve () (statelessInit Infinity) procDef+  return $ StmServer pid inQ replyQ++-- common test cases++testBasicCall :: Launcher () -> TestResult (Maybe String) -> Process ()+testBasicCall launch result = do+  (pid, _) <- launch ()+  callTimeout pid "foo" (within 5 Seconds) >>= stash result++testUnsafeBasicCall :: Launcher () -> TestResult (Maybe String) -> Process ()+testUnsafeBasicCall launch result = do+  (pid, _) <- launch ()+  Unsafe.callTimeout pid "foo" (within 5 Seconds) >>= stash result++testBasicCall_ :: Launcher () -> TestResult (Maybe Int) -> Process ()+testBasicCall_ launch result = do+  (pid, _) <- launch ()+  callTimeout pid (2 :: Int) (within 5 Seconds) >>= stash result++testUnsafeBasicCall_ :: Launcher () -> TestResult (Maybe Int) -> Process ()+testUnsafeBasicCall_ launch result = do+  (pid, _) <- launch ()+  Unsafe.callTimeout pid (2 :: Int) (within 5 Seconds) >>= stash result++testBasicCast :: Launcher () -> TestResult (Maybe String) -> Process ()+testBasicCast launch result = do+  self <- getSelfPid+  (pid, _) <- launch ()+  cast pid ("ping", self)+  expectTimeout (after 3 Seconds) >>= stash result++testUnsafeBasicCast :: Launcher () -> TestResult (Maybe String) -> Process ()+testUnsafeBasicCast launch result = do+  self <- getSelfPid+  (pid, _) <- launch ()+  Unsafe.cast pid ("ping", self)+  expectTimeout (after 3 Seconds) >>= stash result++testControlledTimeout :: Launcher () -> TestResult (Maybe ExitReason) -> Process ()+testControlledTimeout launch result = do+  (pid, exitReason) <- launch ()+  cast pid ("timeout", Delay $ within 1 Seconds)+  waitForExit exitReason >>= stash result++testUnsafeControlledTimeout :: Launcher () -> TestResult (Maybe ExitReason) -> Process ()+testUnsafeControlledTimeout launch result = do+  (pid, exitReason) <- launch ()+  Unsafe.cast pid ("timeout", Delay $ within 1 Seconds)+  waitForExit exitReason >>= stash result++testTerminatePolicy :: Launcher () -> TestResult (Maybe ExitReason) -> Process ()+testTerminatePolicy launch result = do+  (pid, exitReason) <- launch ()+  send pid ("UNSOLICITED_MAIL", 500 :: Int)+  waitForExit exitReason >>= stash result++testUnsafeTerminatePolicy :: Launcher () -> TestResult (Maybe ExitReason) -> Process ()+testUnsafeTerminatePolicy launch result = do+  (pid, exitReason) <- launch ()+  send pid ("UNSOLICITED_MAIL", 500 :: Int)+  waitForExit exitReason >>= stash result++testDropPolicy :: Launcher () -> TestResult (Maybe ExitReason) -> Process ()+testDropPolicy launch result = do+  (pid, exitReason) <- launch ()++  send pid ("UNSOLICITED_MAIL", 500 :: Int)++  sleep $ milliSeconds 250+  mref <- monitor pid++  cast pid "stop"++  r <- receiveTimeout (after 10 Seconds) [+      matchIf (\(ProcessMonitorNotification ref _ _) -> ref == mref)+              (\(ProcessMonitorNotification _ _ r) ->+                case r of+                  DiedUnknownId -> stash result Nothing+                  _ -> waitForExit exitReason >>= stash result)+    ]+  case r of+    Nothing -> stash result Nothing+    _       -> return ()++testUnsafeDropPolicy :: Launcher () -> TestResult (Maybe ExitReason) -> Process ()+testUnsafeDropPolicy launch result = do+  (pid, exitReason) <- launch ()++  send pid ("UNSOLICITED_MAIL", 500 :: Int)++  sleep $ milliSeconds 250+  mref <- monitor pid++  Unsafe.cast pid "stop"++  r <- receiveTimeout (after 10 Seconds) [+      matchIf (\(ProcessMonitorNotification ref _ _) -> ref == mref)+              (\(ProcessMonitorNotification _ _ r) ->+                case r of+                  DiedUnknownId -> stash result Nothing+                  _ -> waitForExit exitReason >>= stash result)+    ]+  case r of+    Nothing -> stash result Nothing+    _       -> return ()++testDeadLetterPolicy :: Launcher ProcessId+                     -> TestResult (Maybe (String, Int))+                     -> Process ()+testDeadLetterPolicy launch result = do+  self <- getSelfPid+  (pid, _) <- launch self++  send pid ("UNSOLICITED_MAIL", 500 :: Int)+  cast pid "stop"++  receiveTimeout+    (after 5 Seconds)+    [ match (\m@(_ :: String, _ :: Int) -> return m) ] >>= stash result++testUnsafeDeadLetterPolicy :: Launcher ProcessId+                     -> TestResult (Maybe (String, Int))+                     -> Process ()+testUnsafeDeadLetterPolicy launch result = do+  self <- getSelfPid+  (pid, _) <- launch self++  send pid ("UNSOLICITED_MAIL", 500 :: Int)+  Unsafe.cast pid "stop"++  receiveTimeout+    (after 5 Seconds)+    [ match (\m@(_ :: String, _ :: Int) -> return m) ] >>= stash result++testHibernation :: Launcher () -> TestResult Bool -> Process ()+testHibernation launch result = do+  (pid, _) <- launch ()+  mref <- monitor pid++  cast pid ("hibernate", (within 3 Seconds))+  cast pid "stop"++  -- the process mustn't stop whilst it's supposed to be hibernating+  r <- receiveTimeout (after 2 Seconds) [+      matchIf (\(ProcessMonitorNotification ref _ _) -> ref == mref)+              (\_ -> return ())+    ]+  case r of+    Nothing -> kill pid "done" >> stash result True+    Just _  -> stash result False++testUnsafeHibernation :: Launcher () -> TestResult Bool -> Process ()+testUnsafeHibernation launch result = do+  (pid, _) <- launch ()+  mref <- monitor pid++  Unsafe.cast pid ("hibernate", (within 3 Seconds))+  Unsafe.cast pid "stop"++  -- the process mustn't stop whilst it's supposed to be hibernating+  r <- receiveTimeout (after 2 Seconds) [+      matchIf (\(ProcessMonitorNotification ref _ _) -> ref == mref)+              (\_ -> return ())+    ]+  case r of+    Nothing -> kill pid "done" >> stash result True+    Just _  -> stash result False++testKillMidCall :: Launcher () -> TestResult Bool -> Process ()+testKillMidCall launch result = do+  (pid, _) <- launch ()+  cast pid ("hibernate", (within 3 Seconds))+  callAsync pid "hello-world" >>= cancelWait >>= unpack result pid+  where unpack :: TestResult Bool -> ProcessId -> AsyncResult () -> Process ()+        unpack res sid AsyncCancelled = kill sid "stop" >> stash res True+        unpack res sid _              = kill sid "stop" >> stash res False++testUnsafeKillMidCall :: Launcher () -> TestResult Bool -> Process ()+testUnsafeKillMidCall launch result = do+  (pid, _) <- launch ()+  Unsafe.cast pid ("hibernate", (within 3 Seconds))+  Unsafe.callAsync pid "hello-world" >>= cancelWait >>= unpack result pid+  where unpack :: TestResult Bool -> ProcessId -> AsyncResult () -> Process ()+        unpack res sid AsyncCancelled = kill sid "stop" >> stash res True+        unpack res sid _              = kill sid "stop" >> stash res False++testSimpleErrorHandling :: Launcher ProcessId+                        -> TestResult (Maybe ExitReason)+                        -> Process ()+testSimpleErrorHandling launch result = do+  self <- getSelfPid+  (pid, exitReason) <- launch self+  register "SUT" pid+  sleep $ seconds 2++  -- this should be *altered* because of the exit handler+  Nothing <- callTimeout pid "foobar" (within 1 Seconds) :: Process (Maybe String)++  Right _ <- awaitResponse pid [+      matchIf (\(s :: String) -> s == "foobar")+              (\s -> return (Right s) :: Process (Either ExitReason String))+    ]++  shutdown pid+  waitForExit exitReason >>= stash result++testUnsafeSimpleErrorHandling :: Launcher ProcessId+                        -> TestResult (Maybe ExitReason)+                        -> Process ()+testUnsafeSimpleErrorHandling launch result = do+  self <- getSelfPid+  (pid, exitReason) <- launch self++  -- this should be *altered* because of the exit handler+  Nothing <- Unsafe.callTimeout pid "foobar" (within 1 Seconds) :: Process (Maybe String)+  "foobar" <- expect++  Unsafe.shutdown pid+  waitForExit exitReason >>= stash result++testAlternativeErrorHandling :: Launcher ProcessId+                             -> TestResult (Maybe ExitReason)+                             -> Process ()+testAlternativeErrorHandling launch result = do+  self <- getSelfPid+  (pid, exitReason) <- launch self++  -- this should be ignored/altered because of the second exit handler+  cast pid (42 :: Int)+  (Just True) <- receiveTimeout (after 2 Seconds) [+        matchIf (\((p :: ProcessId), (i :: Int)) -> p == pid && i == 42)+                (\_ -> return True)+      ]++  shutdown pid+  waitForExit exitReason >>= stash result++testUnsafeAlternativeErrorHandling :: Launcher ProcessId+                             -> TestResult (Maybe ExitReason)+                             -> Process ()+testUnsafeAlternativeErrorHandling launch result = do+  self <- getSelfPid+  (pid, exitReason) <- launch self++  -- this should be ignored/altered because of the second exit handler+  Unsafe.cast pid (42 :: Int)+  (Just True) <- receiveTimeout (after 2 Seconds) [+        matchIf (\((p :: ProcessId), (i :: Int)) -> p == pid && i == 42)+                (\_ -> return True)+      ]++  Unsafe.shutdown pid+  waitForExit exitReason >>= stash result++testServerRejectsMessage :: Launcher ProcessId+                         -> TestResult ExitReason+                         -> Process ()+testServerRejectsMessage launch result = do+  self <- getSelfPid+  (pid, _) <- launch self++  -- server is configured to reject (m :: Delay)+  Left res <- safeCall pid Infinity :: Process (Either ExitReason ())+  stash result res

@@ -0,0 +1,59 @@+{-# LANGUAGE DeriveDataTypeable #-}++module MathsDemo+  ( add+  , divide+  , launchMathServer+  , DivByZero(..)+  , Add(..)+  ) where++import Control.Distributed.Process hiding (call)+import Control.Distributed.Process.Extras+import Control.Distributed.Process.Extras.Time+import Control.Distributed.Process.ManagedProcess++import Data.Binary (Binary(..))+import Data.Typeable (Typeable)++data Add       = Add    Double Double deriving (Typeable)+data Divide    = Divide Double Double deriving (Typeable)+data DivByZero = DivByZero deriving (Typeable, Eq)++instance Binary Add where+  put (Add x y) = put x >> put y+  get = Add <$> get <*> get++instance Binary Divide where+  put (Divide x y) = put x >> put y+  get = Divide <$> get <*> get++instance Binary DivByZero where+  put DivByZero = return ()+  get = return DivByZero++-- public API++add :: ProcessId -> Double -> Double -> Process Double+add sid x y = call sid (Add x y)++divide :: ProcessId -> Double -> Double+          -> Process (Either DivByZero Double)+divide sid x y = call sid (Divide x y )++launchMathServer :: Process ProcessId+launchMathServer =+  let server = statelessProcess {+      apiHandlers = [+          handleCall_   (\(Add    x y) -> return (x + y))+        , handleCallIf_ (input (\(Divide _ y) -> y /= 0)) handleDivide+        , handleCall_   (\(Divide _ _) -> divByZero)+        , action        (\("stop") -> stop_ ExitNormal)+        ]+    }+  in spawnLocal $ serve () (statelessInit Infinity) server+  where handleDivide :: Divide -> Process (Either DivByZero Double)+        handleDivide (Divide x y) = return $ Right $ x / y++        divByZero :: Process (Either DivByZero Double)+        divByZero = return $ Left DivByZero

@@ -0,0 +1,104 @@+{-# LANGUAGE DeriveDataTypeable   #-}+{-# LANGUAGE TemplateHaskell      #-}+{-# LANGUAGE ScopedTypeVariables  #-}+{-# LANGUAGE BangPatterns         #-}+{-# LANGUAGE DeriveGeneric        #-}++module SafeCounter+  ( startCounter,+    getCount,+    getCountAsync,+    incCount,+    resetCount,+    wait,+    waitTimeout,+    Fetch(..),+    Increment(..),+    Reset(..)+  ) where++import Control.Distributed.Process hiding (call, say)+import Control.Distributed.Process.Extras+import Control.Distributed.Process.Async+import Control.Distributed.Process.ManagedProcess+  ( ProcessDefinition(..)+  , InitHandler+  , InitResult(..)+  , defaultProcess+  , condition+  )+import qualified Control.Distributed.Process.ManagedProcess as ManagedProcess (serve)+import Control.Distributed.Process.ManagedProcess.Client+import Control.Distributed.Process.ManagedProcess.Server.Restricted+import Control.Distributed.Process.Extras.Time+import Control.Distributed.Process.Serializable+import Data.Binary+import Data.Typeable (Typeable)+import GHC.Generics++--------------------------------------------------------------------------------+-- Types                                                                      --+--------------------------------------------------------------------------------++data Increment = Increment+  deriving (Show, Typeable, Generic)+instance Binary Increment where++data Fetch = Fetch+  deriving (Show, Typeable, Generic)+instance Binary Fetch where++data Reset = Reset deriving (Show, Typeable, Generic)+instance Binary Reset where++--------------------------------------------------------------------------------+-- API                                                                        --+--------------------------------------------------------------------------------++-- | Increment count+incCount :: ProcessId -> Process Int+incCount sid = call sid Increment++-- | Get the current count+getCount :: ProcessId -> Process Int+getCount sid = call sid Fetch++-- | Get the current count asynchronously+getCountAsync :: ProcessId -> Process (Async Int)+getCountAsync sid = callAsync sid Fetch++-- | Reset the current count+resetCount :: ProcessId -> Process ()+resetCount sid = cast sid Reset++-- | Start a counter server+startCounter :: Int -> Process ProcessId+startCounter startCount =+  let server = serverDefinition+  in spawnLocal $ ManagedProcess.serve startCount init' server+  where init' :: InitHandler Int Int+        init' count = return $ InitOk count Infinity++--------------------------------------------------------------------------------+-- Implementation                                                             --+--------------------------------------------------------------------------------++serverDefinition :: ProcessDefinition Int+serverDefinition = defaultProcess {+   apiHandlers = [+        handleCallIf+          (condition (\count Increment -> count >= 10)) -- invariant+          (\Increment -> halt :: RestrictedProcess Int (Result Int))++      , handleCall handleIncrement+      , handleCall (\Fetch -> getState >>= reply)+      , handleCast (\Reset -> putState (0 :: Int) >> continue)+      ]+   } :: ProcessDefinition Int++halt :: forall s r . Serializable r => RestrictedProcess s (Result r)+halt = haltNoReply (ExitOther "Count > 10")++handleIncrement :: Increment -> RestrictedProcess Int (Result Int)+handleIncrement _ = modifyState (+1) >> getState >>= reply+

@@ -1,17 +1,21 @@ {-# LANGUAGE ScopedTypeVariables #-}-{-# LANGUAGE DeriveDataTypeable  #-}-{-# LANGUAGE BangPatterns        #-}-{-# LANGUAGE TemplateHaskell     #-}+{-# LANGUAGE RecordWildCards     #-}  module Main where +import Control.Concurrent.STM (atomically)+import Control.Concurrent.STM.TQueue+ ( newTQueueIO+ , readTQueue+ , writeTQueue+ ) import Control.Concurrent.MVar import Control.Exception (SomeException)-import Control.Distributed.Process hiding (call)+import Control.Distributed.Process hiding (call, catch) 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.SysTest.Utils import Control.Distributed.Process.Extras.Time import Control.Distributed.Process.Serializable() @@ -30,26 +34,27 @@  import qualified Network.Transport as NT import Control.Monad (void)+import Control.Monad.Catch (catch)  -- utilities -server :: Process (ProcessId, (MVar ExitReason))+server :: Process (ProcessId, MVar ExitReason) server = mkServer Terminate  mkServer :: UnhandledMessagePolicy-         -> Process (ProcessId, (MVar ExitReason))+         -> Process (ProcessId, MVar ExitReason) mkServer policy =   let s = standardTestServer policy   in do-    exitReason <- liftIO $ newEmptyMVar-    pid <- spawnLocal $ do+    exitReason <- liftIO newEmptyMVar+    pid <- spawnLocal $        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+                        Just r  -> fmap Just (stash exitReason r)                     )                  ])               (\(e :: SomeException) -> stash exitReason $ ExitOther (show e))@@ -60,8 +65,8 @@ explodingServer pid =   let srv = explodingTestProcess pid   in do-    exitReason <- liftIO $ newEmptyMVar-    spid <- spawnLocal $ do+    exitReason <- liftIO newEmptyMVar+    spid <- spawnLocal $        catch  (serve () (statelessInit Infinity) srv >> stash exitReason ExitNormal)               (\(e :: SomeException) -> stash exitReason $ ExitOther (show e))     return (spid, exitReason)@@ -84,7 +89,7 @@ testChannelBasedService result =   let procDef = statelessProcess {                     apiHandlers = [-                      handleRpcChan (\s p (m :: String) ->+                      handleRpcChan (\p s (m :: String) ->                                    replyChan p m >> continue s)                     ]                     } in do@@ -93,16 +98,71 @@     stash result (echo == "hello")     kill pid "done" +testExternalService :: TestResult Bool -> Process ()+testExternalService result = do+  inChan <- liftIO newTQueueIO+  replyQ <- liftIO newTQueueIO+  let procDef = statelessProcess {+                    externHandlers = [+                      handleExternal+                        (readTQueue inChan)+                        (\s (m :: String) -> do+                            liftIO $ atomically $ writeTQueue replyQ m+                            continue s)+                    ]+                    }+  let txt = "hello 2-way stm foo"+  pid <- spawnLocal $ serve () (statelessInit Infinity) procDef+  echoTxt <- liftIO $ do+    -- firstly we write something that the server can receive+    atomically $ writeTQueue inChan txt+    -- then sit and wait for it to write something back to us+    atomically $ readTQueue replyQ++  stash result (echoTxt == txt)+  kill pid "done"++testExternalCall :: TestResult Bool -> Process ()+testExternalCall result = do+  let txt = "hello stm-call foo"+  srv <- launchEchoServer (\st (msg :: String) -> reply msg st)+  echoStm srv txt >>= stash result . (== Right txt)+  killProc srv "done"++testExternalCallHaltingServer :: TestResult Bool -> Process ()+testExternalCallHaltingServer result = do+  let msg = "foo bar baz"+  srv <- launchEchoServer (\_ (_ :: String) -> haltNoReply_ ExitNormal)+  echoReply <- echoStm srv msg+  case echoReply of+    -- sadly, we cannot guarantee that our monitor will be set up fast+    -- enough, as per the documentation!+    Left (ExitOther reason) -> stash result $ reason `elem` [ "DiedUnknownId"+                                                            , "DiedNormal"+                                                            ]+    (Left ExitNormal)       -> stash result False+    (Left ExitShutdown)     -> stash result False+    (Right _)               -> stash result False+ -- MathDemo tests -testAdd :: ProcessId -> TestResult Double -> Process ()-testAdd pid result = add pid 10 10 >>= stash result+testAdd :: TestResult Double -> Process ()+testAdd result = do+  pid <- launchMathServer+  add pid 10 10 >>= stash result+  kill pid "done" -testBadAdd :: ProcessId -> TestResult (Either ExitReason Int) -> Process ()-testBadAdd pid result = safeCall pid (Add 10 10) >>= stash result+testBadAdd :: TestResult Bool -> Process ()+testBadAdd result = do+  pid <- launchMathServer+  res <- safeCall pid (Add 10 10) :: Process (Either ExitReason Int)+  stash result (res == (Left $ ExitOther $ "DiedException \"exit-from=" ++ (show pid) ++ "\"")) -testDivByZero :: ProcessId -> TestResult (Either DivByZero Double) -> Process ()-testDivByZero pid result = divide pid 125 0 >>= stash result+testDivByZero :: TestResult (Either DivByZero Double) -> Process ()+testDivByZero result = do+  pid <- launchMathServer+  divide pid 125 0 >>= stash result+  kill pid "done"  -- SafeCounter tests @@ -154,14 +214,11 @@ 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" [+        testGroup "Basic Client/Server Functionality" [             testCase "basic call with explicit server reply"             (delayedAssertion              "expected a response from the server"@@ -186,84 +243,111 @@             (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"+          , testCase "basic channel based rpc"             (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"+             "expected response back from the server"+             localNode True testChannelBasedService)+             ]+           , testGroup "Unhandled Message Policies" [+              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))+          ]+        , testGroup "Server Exit Handling" [+              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 "Advanced Server Interactions" [+            testCase "taking arbitrary STM actions"             (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"+             "expected the server to read the STM queue and reply using STM"+             localNode True testExternalService)+          , testCase "using callSTM to manage non-CH interactions"             (delayedAssertion-             "expected the server to remain in hibernation"-             localNode True (testHibernation $ wrap server))-          , testCase "(unsafe) incoming messages are ignored whilst hibernating"+             "expected the server to reply back via the TQueue"+             localNode True testExternalCall)+          , testCase "getting error data back from callSTM"             (delayedAssertion-             "expected the server to remain in hibernation"-             localNode True (testUnsafeHibernation $ wrap server))+             "expected the server to exit with ExitNormal"+             localNode True testExternalCallHaltingServer)           , 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))+            localNode True (testUnsafeKillMidCall $ wrap server))+          , testCase "server rejects call"+              (delayedAssertion "expected server to send CallRejected"+              localNode (ExitOther "invalid-call") (testServerRejectsMessage $ wrap server))+          , testCase "invalid return type handling"+            (delayedAssertion+             "expected response to fail on runtime type verification"+             localNode True testCallReturnTypeMismatchHandling)+          , 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))           ]         , testGroup "math server examples" [             testCase "error (Left) returned from x / 0"               (delayedAssertion                "expected the server to return DivByZero"-               localNode (Left DivByZero) (testDivByZero pid))+               localNode (Left DivByZero) testDivByZero)           , testCase "10 + 10 = 20"               (delayedAssertion                "expected the server to return DivByZero"-               localNode 20 (testAdd pid))+               localNode 20 testAdd)           , 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))+             localNode True testBadAdd)           ]         , testGroup "counter server examples" [             testCase "initial counter state = 5"@@ -293,4 +377,3 @@  main :: IO () main = testMain $ tests-

@@ -1,27 +1,34 @@ {-# 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.Concurrent.STM.TQueue+ ( newTQueueIO+ , readTQueue+ , writeTQueue+ ) import Control.Exception (SomeException) import Control.DeepSeq (NFData)-import Control.Distributed.Process hiding (call, send)+import Control.Distributed.Process hiding (call, send, catch, sendChan) 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 hiding (__remoteTable, monitor)+import Control.Distributed.Process.Async hiding (check)+import Control.Distributed.Process.ManagedProcess hiding (reject)+import qualified Control.Distributed.Process.ManagedProcess.Server.Priority as P (Message)+import Control.Distributed.Process.ManagedProcess.Server.Priority+import Control.Distributed.Process.SysTest.Utils import Control.Distributed.Process.Extras.Time-import Control.Distributed.Process.Extras.Timer+import Control.Distributed.Process.Extras.Timer hiding (runAfter) import Control.Distributed.Process.Serializable()+import Control.Monad+import Control.Monad.Catch (catch)  import Data.Binary import Data.Either (rights)+import Data.List (isInfixOf) import Data.Typeable (Typeable)  #if ! MIN_VERSION_base(4,6,0)@@ -68,10 +75,12 @@   let srv = explodingTestProcess pid       pSrv = srv `prioritised` ([] :: [DispatchPriority s])   in do-    exitReason <- liftIO $ newEmptyMVar+    exitReason <- liftIO newEmptyMVar     spid <- spawnLocal $ do        catch  (pserve () (statelessInit Infinity) pSrv >> stash exitReason ExitNormal)-              (\(e :: SomeException) -> stash exitReason $ ExitOther (show e))+              (\(e :: SomeException) -> do+                -- say "died in handler..."+                stash exitReason $ ExitOther (show e))     return (spid, exitReason)  data GetState = GetState@@ -115,8 +124,211 @@           , timeoutHandler         = \_ _ -> stop $ ExitOther "timeout"           } :: ProcessDefinition [(Either MyAlarmSignal String)] --- test cases+mkOverflowHandlingServer :: (PrioritisedProcessDefinition Int ->+                             PrioritisedProcessDefinition Int)+                         -> Process ProcessId+mkOverflowHandlingServer modIt =+  let p = procDef `prioritised` ([+               prioritiseCall_ (\GetState -> setPriority 99 :: Priority Int)+             , prioritiseCast_ (\(_ :: String) -> setPriority 1)+             ] :: [DispatchPriority Int]+          ) :: PrioritisedProcessDefinition Int+  in spawnLocal $ pserve () (initWait Infinity) (modIt p)+  where+    initWait :: Delay+             -> InitHandler () Int+    initWait d () = return $ InitOk 0 d +    procDef :: ProcessDefinition Int+    procDef =+      defaultProcess {+            apiHandlers = [+               handleCall (\s GetState -> reply s s)+             , handleCast (\s (_ :: String) -> continue $ s + 1)+            ]+          } :: ProcessDefinition Int++launchStmServer :: CallHandler () String String -> Process StmServer+launchStmServer handler = do+  (inQ, replyQ) <- liftIO $ do+    cIn <- newTQueueIO+    cOut <- newTQueueIO+    return (cIn, cOut)++  let procDef = statelessProcess {+                  externHandlers = [+                    handleCallExternal+                      (readTQueue inQ)+                      (writeTQueue replyQ)+                      handler+                  ]+                , apiHandlers = [+                    action (\() -> stop_ ExitNormal)+                  ]+                }++  let p = procDef `prioritised` ([+               prioritiseCast_ (\() -> setPriority 99 :: Priority ())+             , prioritiseCast_ (\(_ :: String) -> setPriority 100)+             ] :: [DispatchPriority ()]+          ) :: PrioritisedProcessDefinition ()++  pid <- spawnLocal $ pserve () (statelessInit Infinity) p+  return $ StmServer pid inQ replyQ++launchStmOverloadServer :: Process (ProcessId, ControlPort String)+launchStmOverloadServer = do+  cc <- newControlChan :: Process (ControlChannel String)+  let cp = channelControlPort cc++  let procDef = statelessProcess {+                  externHandlers = [+                    handleControlChan_ cc (\(_ :: String) -> continue_)+                  ]+                , apiHandlers = [+                    handleCast (\s sp -> sendChan sp () >> continue s)+                  ]+                }++  let p = procDef `prioritised` ([+               prioritiseCast_ (\() -> setPriority 99 :: Priority ())+             ] :: [DispatchPriority ()]+          ) :: PrioritisedProcessDefinition ()++  pid <- spawnLocal $ pserve () (statelessInit Infinity) p+  return (pid, cp)++data Foo = Foo deriving (Show)++launchFilteredServer :: ProcessId -> Process (ProcessId, ControlPort (SendPort Int))+launchFilteredServer us = do+  cc <- newControlChan :: Process (ControlChannel (SendPort Int))+  let cp = channelControlPort cc++  let procDef = defaultProcess {+                  externHandlers = [+                    handleControlChan cc (\s (p :: SendPort Int) -> sendChan p s >> continue s)+                  ]+                , apiHandlers = [+                    handleCast (\s sp -> sendChan sp () >> continue s)+                  , handleCall_ (\(s :: String) -> return s)+                  , handleCall_ (\(i :: Int) -> return i)+                  ]+                , unhandledMessagePolicy = DeadLetter us+                } :: ProcessDefinition Int++  let p = procDef `prioritised` ([+               prioritiseCast_ (\() -> setPriority 1 :: Priority ())+             , prioritiseCall_ (\(_ :: String) -> setPriority 100 :: Priority String)+             ] :: [DispatchPriority Int]+          ) :: PrioritisedProcessDefinition Int++  let rejectUnchecked =+        rejectApi Foo :: Int -> P.Message String String -> Process (Filter Int)++  let p' = p {+    filters = [+      store  (+1)+    , ensure (>0)  -- a bit pointless, but we're just checking the API++    , check $ api_ (\(s :: String) -> return $ "checked-" `isInfixOf` s) rejectUnchecked+    , check $ info (\_ (_ :: MonitorRef, _ :: ProcessId) -> return False) $ reject Foo+    , refuse ((> 10) :: Int -> Bool)+    ]+  }++  pid <- spawnLocal $ pserve 0 (\c -> return $ InitOk c Infinity) p'+  return (pid, cp)++testFilteringBehavior :: TestResult Bool -> Process ()+testFilteringBehavior result = do+  us <- getSelfPid+  (sp, rp) <- newChan+  (pid, cp) <- launchFilteredServer us+  mRef <- monitor pid++  sendControlMessage cp sp++  r <- receiveChan rp :: Process Int+  when (r > 1) $ stash result False >> die "we're done..."++  Left _ <- safeCall pid "bad-input" :: Process (Either ExitReason String)++  send pid (mRef, us)  -- server doesn't like this, dead letters it...+  -- back to us+  void $ receiveWait [ matchIf (\(m, p) -> m == mRef && p == us) return ]++  sendControlMessage cp sp++  r2 <- receiveChan rp :: Process Int+  when (r2 < 3) $ stash result False >> die "we're done again..."++  -- server also doesn't like this, and sends it right back (via \DeadLetter us/)+  send pid (25 :: Int)++  m <- receiveWait [ matchIf (== 25) return ] :: Process Int+  stash result $ m == 25+  kill pid "done"++testExternalTimedOverflowHandling :: TestResult Bool -> Process ()+testExternalTimedOverflowHandling result = do+  (pid, cp) <- launchStmOverloadServer -- default 10k mailbox drain limit+  wrk <- spawnLocal $ mapM_ (sendControlMessage cp . show) ([1..500000] :: [Int])++  sleep $ milliSeconds 250 -- give the worker time to start spamming the server...++  (sp, rp) <- newChan+  cast pid sp -- tell the server we're expecting a reply++  -- it might take "a while" for us to get through the first 10k messages+  -- from our chatty friend wrk, before we finally get our control message seen+  -- by the reader/listener loop, and in fact timing wise we don't even know when+  -- our message will arrive, since we're racing with wrk to communicate with+  -- the server. It's important therefore to give sufficient time for the right+  -- conditions to occur so that our message is finally received and processed,+  -- yet we don't want to lock up the build for 10-20 mins either. This value+  -- of 30 seconds seems like a reasonable compromise.+  answer <- receiveChanTimeout (asTimeout $ seconds 30) rp++  stash result $ answer == Just ()+  kill wrk "done"+  kill pid "done"++testExternalCall :: TestResult Bool -> Process ()+testExternalCall result = do+  let txt = "hello stm-call foo"+  srv <- launchStmServer (\st (msg :: String) -> reply msg st)+  echoStm srv txt >>= stash result . (== Right txt)+  killProc srv "done"++testTimedOverflowHandling :: TestResult Bool -> Process ()+testTimedOverflowHandling result = do+  pid <- mkOverflowHandlingServer (\s -> s { recvTimeout = RecvTimer $ within 3 Seconds })+  wrk <- spawnLocal $ mapM_ (cast pid . show) ([1..500000] :: [Int])++  sleep $ seconds 1 -- give the worker time to start spamming us...+  cast pid "abc" -- just getting in line here...++  st <- call pid GetState :: Process Int+  -- the result of GetState is a list of messages in reverse insertion order+  stash result $ st > 0+  kill wrk "done"+  kill pid "done"++testOverflowHandling :: TestResult Bool -> Process ()+testOverflowHandling result = do+  pid <- mkOverflowHandlingServer (\s -> s { recvTimeout = RecvMaxBacklog 100 })+  wrk <- spawnLocal $ mapM_ (cast pid . show) ([1..50000] :: [Int])++  sleep $ seconds 1+  cast pid "abc" -- just getting in line here...++  st <- call pid GetState :: Process Int+  -- the result of GetState is a list of messages in reverse insertion order+  stash result $ st > 0+  kill wrk "done"+  kill pid "done"+ testInfoPrioritisation :: TestResult Bool -> Process () testInfoPrioritisation result = do   pid <- mkPrioritisedServer@@ -137,6 +349,33 @@     Left MyAlarmSignal -> stash result True     _ -> stash result False +testUserTimerHandling :: TestResult Bool -> Process ()+testUserTimerHandling result = do+  us <- getSelfPid+  let p = (procDef us) `prioritised` ([+               prioritiseInfo_ (\MyAlarmSignal -> setPriority 100)+             ] :: [DispatchPriority ()]+          ) :: PrioritisedProcessDefinition ()+  pid <- spawnLocal $ pserve () (statelessInit Infinity) p+  cast pid ()+  expect >>= stash result . (== MyAlarmSignal)+  kill pid "goodbye..."++  where++    procDef :: ProcessId -> ProcessDefinition ()+    procDef us =+      statelessProcess {+            apiHandlers = [+              handleCast (\s () -> evalAfter (seconds 5) MyAlarmSignal s)+            ]+          , infoHandlers = [+               handleInfo (\s (sig :: MyAlarmSignal) -> send us sig >> continue s)+            ]+          , unhandledMessagePolicy = Drop+          } :: ProcessDefinition ()++ testCallPrioritisation :: TestResult Bool -> Process () testCallPrioritisation result = do   pid <- mkPrioritisedServer@@ -151,7 +390,7 @@   -- is undefined (and in practise, paritally depenendent on the scheduler)   sleep $ seconds 1   send pid ()-  mapM wait asyncRefs :: Process [AsyncResult ()]+  _ <- 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"]@@ -198,6 +437,9 @@           , testCase "long running call cancellation"             (delayedAssertion "expected to get AsyncCancelled"              localNode True (testKillMidCall $ wrap server))+          , testCase "server rejects call"+             (delayedAssertion "expected server to send CallRejected"+              localNode (ExitOther "invalid-call") (testServerRejectsMessage $ wrap server))           , testCase "simple exit handling"             (delayedAssertion "expected handler to catch exception and continue"              localNode Nothing (testSimpleErrorHandling $ explodingServer))@@ -212,10 +454,29 @@           , testCase "Call Message Prioritisation"             (delayedAssertion "expected the longest strings to be prioritised"              localNode True testCallPrioritisation)+          , testCase "Size-Based Mailbox Overload Management"+            (delayedAssertion "expected the server loop to stop reading the mailbox"+             localNode True testOverflowHandling)+          , testCase "Timeout-Based Mailbox Overload Management"+            (delayedAssertion "expected the server loop to stop reading the mailbox"+             localNode True testTimedOverflowHandling)           ]+       , testGroup "Advanced Server Interactions" [+            testCase "using callSTM to manage non-CH interactions"+            (delayedAssertion+             "expected the server to reply back via the TQueue"+             localNode True testExternalCall)+          , testCase "Timeout-Based Overload Management with Control Channels"+            (delayedAssertion "expected the server loop to reply"+             localNode True testExternalTimedOverflowHandling)+          , testCase "Complex pre/before filters"+             (delayedAssertion "expected verifiable filter actions"+              localNode True testFilteringBehavior)+          , testCase "Firing internal timeouts"+             (delayedAssertion "expected our info handler to run after the timeout"+              localNode True testUserTimerHandling)+         ]       ]  main :: IO () main = testMain $ tests--

@@ -0,0 +1,49 @@+{-# LANGUAGE DeriveDataTypeable        #-}+{-# LANGUAGE TemplateHaskell           #-}++module TestUtils+  ( testMain+  , mkNode+  , waitForExit+  ) where++import Control.Concurrent.MVar+  ( MVar+  , takeMVar+  )++import Control.Distributed.Process+import Control.Distributed.Process.Node+import Control.Distributed.Process.Extras+import Control.Distributed.Process.Extras.Time+import Control.Distributed.Process.Extras.Timer+import Test.Framework (Test, defaultMain)++import Network.Transport.TCP+import qualified Network.Transport as NT++waitForExit :: MVar ExitReason+            -> Process (Maybe ExitReason)+waitForExit exitReason = do+    -- we *might* end up blocked here, so ensure the test doesn't jam up!+  self <- getSelfPid+  tref <- killAfter (within 10 Seconds) self "testcast timed out"+  tr <- liftIO $ takeMVar exitReason+  cancelTimer tref+  case tr of+    ExitNormal -> return Nothing+    other      -> return $ Just other++mkNode :: String -> IO LocalNode+mkNode port = do+  Right (transport1, _) <- createTransportExposeInternals+                                    "127.0.0.1" port defaultTCPParameters+  newLocalNode transport1 initRemoteTable++-- | Given a @builder@ function, make and run a test suite on a single transport+testMain :: (NT.Transport -> IO [Test]) -> IO ()+testMain builder = do+  Right (transport, _) <- createTransportExposeInternals+                                    "127.0.0.1" "0" defaultTCPParameters+  testData <- builder transport+  defaultMain testData