distributed-process-client-server 0.2.1 → 0.2.2

raw patch · 9 files changed

+344/−12 lines, 9 filesPVP ok

version bump matches the API change (PVP)

API changes (from Hackage documentation)

Files

• distributed-process-client-server.cabal +3/−1
• src/Control/Distributed/Process/ManagedProcess.hs +46/−4
• src/Control/Distributed/Process/ManagedProcess/Internal/GenProcess.hs +15/−1
• src/Control/Distributed/Process/ManagedProcess/Internal/PriorityQueue.hs +41/−0
• src/Control/Distributed/Process/ManagedProcess/Internal/Types.hs +2/−1
• src/Control/Distributed/Process/ManagedProcess/Server.hs +3/−0
• src/Control/Distributed/Process/ManagedProcess/Server/Gen.hs +181/−0
• src/Control/Distributed/Process/ManagedProcess/Server/Priority.hs +10/−1
• tests/TestPrioritisedProcess.hs +43/−4

@@ -1,5 +1,5 @@ name:           distributed-process-client-server-version:        0.2.1+version:        0.2.2 cabal-version:  >=1.8 build-type:     Simple license:        BSD3@@ -56,9 +56,11 @@                    Control.Distributed.Process.ManagedProcess.Server,                    Control.Distributed.Process.ManagedProcess.Server.Priority,                    Control.Distributed.Process.ManagedProcess.Server.Restricted,+                   Control.Distributed.Process.ManagedProcess.Server.Gen,                    Control.Distributed.Process.ManagedProcess.Timer,                    Control.Distributed.Process.ManagedProcess.Internal.Types,                    Control.Distributed.Process.ManagedProcess.Internal.GenProcess+  other-modules:   Control.Distributed.Process.ManagedProcess.Internal.PriorityQueue  test-suite ManagedProcessTests   type:            exitcode-stdio-1.0

@@ -368,6 +368,45 @@ -- abruptly. Once again, supervision hierarchies are a better way to ensure -- consistent cleanup occurs when valued resources are held by a process. --+-- [Filters, pre-processing, and safe handlers]+--+-- A prioritised process can take advantage of filters, which enable the server+-- to pre-process messages, reject them (based on the message itself, or the+-- server's state), and mark classes of message as requiring /safe/ handling.+--+-- Assuming a 'PrioritisedProcessDefinition' that holds its state as an 'Int',+-- here are some simple applications of filters:+--+-- >   let rejectUnchecked =+-- >        rejectApi Foo :: Int -> P.Message String String -> Process (Filter Int)+-- >+-- >     filters = [+-- >       store  (+1)+-- >     , ensure (>0)+-- >+-- >     , check $ api_ (\(s :: String) -> return $ "checked-" `isInfixOf` s) rejectUnchecked+-- >     , check $ info (\_ (_ :: MonitorRef, _ :: ProcessId) -> return False) $ reject Foo+-- >     , refuse ((> 10) :: Int -> Bool)+-- >     ]+--+-- We can store/update our state, ensure our state is in a valid condition,+-- check api and info messages, and refuse messages using simple predicates.+-- Messages cannot be modified by filters, not can reply data.+--+-- A 'safe' filter is a means to instruct the prioritised managed process loop+-- not to dequeue the current message from the internal priority queue until a+-- handler has successfully matched and run against it (without an exception,+-- either synchronous or asynchronous) to completion. Messages marked thus, will+-- remain in the priority queue even in the face of exit signals, which means that+-- if the server process code handles and swallows them, it will begin re-processing+-- the last message a second time.+--+-- It is important to recognise that the 'safe' filter does not act like a+-- transaction. There are no checkpoints, nor facilities for rolling back actions+-- on failure. If an exit signal terminates a handler for a message marked as+-- 'safe' and an exit handler catches and swallows it, the handler (and all prior+-- filters too) will be re-run in its entireity.+-- -- [Special Clients: Control Channels] -- -- For advanced users and those requiring very low latency, a prioritised@@ -461,7 +500,6 @@ -- 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: --@@ -597,21 +635,25 @@   , defaultProcessWithPriorities   , statelessProcess   , statelessInit-    -- * Server side callbacks-  , module Control.Distributed.Process.ManagedProcess.Server-    -- * Control channels+  -- * Control channels   , ControlChannel()   , ControlPort()   , newControlChan   , channelControlPort+    -- * Server side callbacks+  , module Control.Distributed.Process.ManagedProcess.Server     -- * Prioritised mailboxes   , module P+    -- * Low level and internal APIs & Process implementation+  , module Gen   ) where  import Control.Distributed.Process hiding (call, Message) import Control.Distributed.Process.ManagedProcess.Client import Control.Distributed.Process.ManagedProcess.Server+import qualified Control.Distributed.Process.ManagedProcess.Server.Restricted as R import qualified Control.Distributed.Process.ManagedProcess.Server.Priority as P hiding (reject)+import qualified Control.Distributed.Process.ManagedProcess.Internal.GenProcess as Gen import Control.Distributed.Process.ManagedProcess.Internal.GenProcess import Control.Distributed.Process.ManagedProcess.Internal.Types hiding (runProcess) import Control.Distributed.Process.Extras (ExitReason(..))

@@ -21,6 +21,7 @@   , processDefinition   , processFilters   , processUnhandledMsgPolicy+  , processQueue   , gets   , getAndModifyState   , modifyState@@ -30,8 +31,10 @@   , peek   , push   , enqueue+  , dequeue   , addUserTimer   , removeUserTimer+  , eval   , act   , runAfter   , evalAfter@@ -80,11 +83,12 @@   , matchRun   ) import Control.Distributed.Process.ManagedProcess.Internal.Types hiding (Message)-import qualified Control.Distributed.Process.Extras.Internal.Queue.PriorityQ as Q+import qualified Control.Distributed.Process.ManagedProcess.Internal.PriorityQueue as Q   ( empty   , dequeue   , enqueue   , peek+  , toList   ) import Control.Distributed.Process.Extras   ( ExitReason(..)@@ -201,6 +205,10 @@ processUnhandledMsgPolicy :: GenProcess s UnhandledMessagePolicy processUnhandledMsgPolicy = gets (unhandledMessagePolicy . procDef) +-- | Returns a /read only view/ on the internal priority queue.+processQueue :: GenProcess s [Message]+processQueue = gets internalQ >>= return . Q.toList+ -- | The @Timer@ for the system timeout. See @drainTimeout@. systemTimeout :: GenProcess s Timer systemTimeout = gets sysTimeout@@ -246,6 +254,10 @@ act = return . ProcessActivity {-# WARNING act "This interface is intended for internal use only" #-} +-- | Evaluate an expression in the 'GenProcess' monad.+eval :: forall s . GenProcess s (ProcessAction s) -> Action s+eval = return . ProcessExpression+ -- | Starts a timer and adds it as a /user timeout/. runAfter :: forall s m . (Serializable m) => TimeInterval -> m -> GenProcess s () runAfter d m = do@@ -418,6 +430,7 @@      nextAction :: ProcessAction s -> GenProcess s ExitReason     nextAction ac+      | ProcessExpression expr   <- ac = expr >>= nextAction       | ProcessActivity  act'    <- ac = act' >> recvQueue       | ProcessSkip              <- ac = recvQueue       | ProcessContinue  ps'     <- ac = recvQueueAux ps'@@ -731,6 +744,7 @@         (ProcessStopping s' r)    -> handleStop (LastKnown s') r >> return (r :: ExitReason)         (ProcessBecome   d' s')   -> recvLoop d' s' recvDelay         (ProcessActivity _)       -> die $ "recvLoop.InvalidState - ProcessActivityNotSupported"+        (ProcessExpression _)     -> die $ "recvLoop.InvalidState - ProcessExpressionNotSupported"   where     matchAux :: UnhandledMessagePolicy              -> s

@@ -0,0 +1,41 @@+{-# LANGUAGE BangPatterns       #-}+module Control.Distributed.Process.ManagedProcess.Internal.PriorityQueue where++-- NB: we might try this with a skewed binomial heap at some point,+-- but for now, we'll use this module from the fingertree package+import qualified Data.PriorityQueue.FingerTree as PQ+import qualified Data.Foldable as F (toList)+import Data.PriorityQueue.FingerTree (PQueue)++newtype PriorityQ k a = PriorityQ { q :: PQueue k a }++{-# INLINE empty #-}+empty :: Ord k => PriorityQ k v+empty = PriorityQ $ PQ.empty++{-# INLINE isEmpty #-}+isEmpty :: Ord k => PriorityQ k v -> Bool+isEmpty = PQ.null . q++{-# INLINE singleton #-}+singleton :: Ord k => k -> a -> PriorityQ k a+singleton !k !v = PriorityQ $ PQ.singleton k v++{-# INLINE enqueue #-}+enqueue :: Ord k => k -> v -> PriorityQ k v -> PriorityQ k v+enqueue !k !v p = PriorityQ (PQ.add k v $ q p)++{-# INLINE dequeue #-}+dequeue :: Ord k => PriorityQ k v -> Maybe (v, PriorityQ k v)+dequeue p = maybe Nothing (\(v, pq') -> Just (v, pq')) $+              case (PQ.minView (q p)) of+                Nothing     -> Nothing+                Just (v, q') -> Just (v, PriorityQ $ q')++{-# INLINE peek #-}+peek :: Ord k => PriorityQ k v -> Maybe v+peek p = maybe Nothing (\(v, _) -> Just v) $ dequeue p++{-# INLINE toList #-}+toList :: Ord k => PriorityQ k a -> [a]+toList = F.toList . q

@@ -90,7 +90,7 @@   , Routable(..)   , NFSerializable   )-import Control.Distributed.Process.Extras.Internal.Queue.PriorityQ+import Control.Distributed.Process.ManagedProcess.Internal.PriorityQueue   ( PriorityQ   ) import Control.Distributed.Process.Extras.Internal.Types@@ -313,6 +313,7 @@ data ProcessAction s =     ProcessSkip   | ProcessActivity  (GenProcess s ()) -- ^ run the given activity+  | ProcessExpression (GenProcess s (ProcessAction s)) -- ^ evaluate an expression   | ProcessContinue  s              -- ^ continue with (possibly new) state   | ProcessTimeout   Delay        s -- ^ timeout if no messages are received   | ProcessHibernate TimeInterval s -- ^ hibernate for /delay/

@@ -190,6 +190,9 @@ hibernate_ :: StatelessHandler s TimeInterval hibernate_ d = return . ProcessHibernate d +-- | The server loop will execute against the supplied 'ProcessDefinition', allowing+-- the process to change its behaviour (in terms of message handlers, exit handling,+-- termination, unhandled message policy, etc) become :: forall s . ProcessDefinition s -> s -> Action s become def st = return $ ProcessBecome def st 

@@ -0,0 +1,181 @@+{-# LANGUAGE ExistentialQuantification  #-}+{-# LANGUAGE ScopedTypeVariables        #-}+{-# LANGUAGE PatternGuards              #-}+{-# LANGUAGE RecordWildCards            #-}+{-# LANGUAGE EmptyDataDecls             #-}+{-# LANGUAGE StandaloneDeriving         #-}+{-# LANGUAGE AllowAmbiguousTypes        #-}++-----------------------------------------------------------------------------+-- |+-- 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 Server Portion of the /Managed Process/ API, as presented by the+-- 'GenProcess' monad. These functions are generally intended for internal+-- use, but the API is relatively stable and therefore they have been re-exported+-- here for general use. Note that if you modify a process' internal state+-- (especially that of the internal priority queue) then you are responsible for+-- any alteratoin that makes to the semantics of your processes behaviour.+--+-- See "Control.Distributed.Process.ManagedProcess.Internal.GenProcess"+-----------------------------------------------------------------------------+module Control.Distributed.Process.ManagedProcess.Server.Gen+  ( -- * Server actions+    reply+  , replyWith+  , noReply+  , continue+  , timeoutAfter+  , hibernate+  , stop+  , reject+  , rejectWith+  , become+  , haltNoReply+  , lift+  , Gen.recvLoop+  , Gen.precvLoop+  , Gen.currentTimeout+  , Gen.systemTimeout+  , Gen.drainTimeout+  , Gen.processState+  , Gen.processDefinition+  , Gen.processFilters+  , Gen.processUnhandledMsgPolicy+  , Gen.processQueue+  , Gen.gets+  , Gen.getAndModifyState+  , Gen.modifyState+  , Gen.setUserTimeout+  , Gen.setProcessState+  , GenProcess+  , Gen.peek+  , Gen.push+  , Gen.enqueue+  , Gen.dequeue+  , Gen.addUserTimer+  , Gen.removeUserTimer+  , Gen.eval+  , Gen.act+  , Gen.runAfter+  , Gen.evalAfter+  ) where++import Control.Distributed.Process.Extras+ ( ExitReason+ )+import Control.Distributed.Process.Extras.Time+ ( TimeInterval+ , Delay+ )+import Control.Distributed.Process.ManagedProcess.Internal.Types+ ( lift+ , ProcessAction(..)+ , GenProcess+ , ProcessReply(..)+ , ProcessDefinition+ )+import qualified Control.Distributed.Process.ManagedProcess.Internal.GenProcess as Gen+ ( recvLoop+ , precvLoop+ , currentTimeout+ , systemTimeout+ , drainTimeout+ , processState+ , processDefinition+ , processFilters+ , processUnhandledMsgPolicy+ , processQueue+ , gets+ , getAndModifyState+ , modifyState+ , setUserTimeout+ , setProcessState+ , GenProcess+ , peek+ , push+ , enqueue+ , dequeue+ , addUserTimer+ , removeUserTimer+ , eval+ , act+ , runAfter+ , evalAfter+ )+import Control.Distributed.Process.ManagedProcess.Internal.GenProcess+ ( processState+ )+import qualified Control.Distributed.Process.ManagedProcess.Server as Server+ ( replyWith+ , continue+ )+import Control.Distributed.Process.Serializable (Serializable)++-- | Reject the message we're currently handling.+reject :: forall r s . String -> GenProcess s (ProcessReply r s)+reject rs = processState >>= \st -> lift $ Server.continue st >>= return . ProcessReject rs++-- | Reject the message we're currently handling, giving an explicit reason.+rejectWith :: forall r m s . (Show r) => r -> GenProcess s (ProcessReply m s)+rejectWith rs = reject (show rs)++-- | Instructs the process to send a reply and continue running.+reply :: forall r s . (Serializable r) => r -> GenProcess s (ProcessReply r s)+reply r = processState >>= \s -> lift $ Server.continue s >>= Server.replyWith r++-- | Instructs the process to send a reply /and/ evaluate the 'ProcessAction'.+replyWith :: forall r s . (Serializable r)+         => r+         -> ProcessAction s+         -> GenProcess s (ProcessReply r s)+replyWith r s = return $ ProcessReply r s++-- | Instructs the process to skip sending a reply /and/ evaluate a 'ProcessAction'+noReply :: (Serializable r) => ProcessAction s -> GenProcess s (ProcessReply r s)+noReply = return . NoReply++-- | Halt process execution during a call handler, without paying any attention+-- to the expected return type.+haltNoReply :: forall s r . Serializable r => ExitReason -> GenProcess s (ProcessReply r s)+haltNoReply r = stop r >>= noReply++-- | Instructs the process to continue running and receiving messages.+continue :: GenProcess s (ProcessAction s)+continue = processState >>= 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.+--+-- Note that @timeoutAfter NoDelay@ will cause the timeout handler to execute+-- immediately if no messages are present in the process' mailbox.+--+timeoutAfter :: Delay -> GenProcess s (ProcessAction s)+timeoutAfter d = processState >>= \s -> return $ ProcessTimeout d s++-- | Instructs the process to /hibernate/ for the given 'TimeInterval'. Note+-- that no messages will be removed from the mailbox until after hibernation has+-- ceased. This is equivalent to calling @threadDelay@.+--+hibernate :: TimeInterval -> GenProcess s (ProcessAction s)+hibernate d = processState >>= \s -> return $ ProcessHibernate d s++-- | The server loop will execute against the supplied 'ProcessDefinition', allowing+-- the process to change its behaviour (in terms of message handlers, exit handling,+-- termination, unhandled message policy, etc)+become :: forall s . ProcessDefinition s -> GenProcess s (ProcessAction s)+become def = processState >>= \st -> return $ ProcessBecome def st++-- | 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 -> GenProcess s (ProcessAction s)+stop r = return $ ProcessStop r

@@ -188,6 +188,8 @@         -> FilterHandler s info_ c h = info (const $ c) h +-- | As 'safe', but as applied to api messages (i.e. those originating from+-- call as cast client interactions). apiSafe :: forall s m b . (Serializable m, Serializable b)      => (s -> m -> Maybe b -> Bool)      -> DispatchFilter s@@ -201,6 +203,13 @@         Just (ChanMessage m' _) -> return $ c' s m' Nothing         Nothing                 -> return False +-- | Given a check expression, if it evaluates to @True@ for some input,+-- then do not dequeue the message until after any matching handlers have+-- successfully run, or the the unhandled message policy is chosen if none match.+-- Thus, if an exit signal (async exception) terminates execution of a handler, and we+-- have an installed exit handler which allows the process to continue running,+-- we will retry the input in question since it has not been fully dequeued prior+-- to the exit signal arriving. safe :: forall s m . (Serializable m)      => (s -> m -> Bool)      -> DispatchFilter s@@ -212,6 +221,7 @@         Just m' -> return $ c' s m'         Nothing -> return False +-- | As 'safe', but matches on a raw message. safely :: forall s . (s -> P.Message -> Bool) -> DispatchFilter s safely c = check $ HandleSafe $ \s m -> return (c s m) @@ -251,7 +261,6 @@ 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))

@@ -12,13 +12,18 @@  ) import Control.Exception (SomeException) import Control.DeepSeq (NFData)-import Control.Distributed.Process hiding (call, send, catch, sendChan)+import Control.Distributed.Process hiding (call, send, catch, sendChan, wrapMessage) import Control.Distributed.Process.Node import Control.Distributed.Process.Extras hiding (__remoteTable, monitor) import Control.Distributed.Process.Async hiding (check)-import Control.Distributed.Process.ManagedProcess hiding (reject)+import Control.Distributed.Process.ManagedProcess hiding (reject, Message) import qualified Control.Distributed.Process.ManagedProcess.Server.Priority as P (Message)-import Control.Distributed.Process.ManagedProcess.Server.Priority+import Control.Distributed.Process.ManagedProcess.Server.Priority hiding (Message)+import qualified Control.Distributed.Process.ManagedProcess.Server.Gen as Gen+ ( dequeue+ , continue+ , lift+ ) import Control.Distributed.Process.SysTest.Utils import Control.Distributed.Process.Extras.Time import Control.Distributed.Process.Extras.Timer hiding (runAfter)@@ -29,7 +34,7 @@ import Data.Binary import Data.Either (rights) import Data.List (isInfixOf)-import Data.Maybe (isNothing)+import Data.Maybe (isNothing, isJust) import Data.Typeable (Typeable)  #if ! MIN_VERSION_base(4,6,0)@@ -241,6 +246,37 @@   pid <- spawnLocal $ pserve 0 (\c -> return $ InitOk c Infinity) p'   return (pid, cp) +testStupidInfiniteLoop :: TestResult Bool -> Process ()+testStupidInfiniteLoop result = do+  let def = statelessProcess {+                  apiHandlers = [+                    handleCast (\_ sp -> eval $ do q <- processQueue+                                                   m <- Gen.dequeue+                                                   Gen.lift $ sendChan sp (length q, m)+                                                   Gen.continue)+                  ]+                , infoHandlers = [+                    handleInfo (\_ (m :: String) -> eval $ do enqueue (wrapMessage m)+                                                              Gen.continue)+                  ]+                } :: ProcessDefinition ()++  let prio = def `prioritised` []+  pid <- spawnLocal $ pserve () (statelessInit Infinity) prio++  -- this message should create an infinite loop+  send pid "fooboo"++  (sp, rp) <- newChan :: Process (SendPort (Int, Maybe Message), ReceivePort (Int, Maybe Message))++  cast pid sp+  (i, m) <- receiveChan rp++  cast pid sp+  (i', m') <- receiveChan rp++  stash result $ (i == 1 && isJust m && i' == 0 && isNothing m')+ testFilteringBehavior :: TestResult Bool -> Process () testFilteringBehavior result = do   us <- getSelfPid@@ -587,6 +623,9 @@           , testCase "Swapping ProcessDefinitions at runtime"              (delayedAssertion "expected our handler to exist in the new handler list"               localNode True testServerSwap)+          , testCase "Accessing the internal process implementation"+             (delayedAssertion "it should allow us to modify the internal q"+              localNode True testStupidInfiniteLoop)          ]       ]