diff --git a/distributed-process-client-server.cabal b/distributed-process-client-server.cabal --- a/distributed-process-client-server.cabal +++ b/distributed-process-client-server.cabal @@ -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 diff --git a/src/Control/Distributed/Process/ManagedProcess.hs b/src/Control/Distributed/Process/ManagedProcess.hs --- a/src/Control/Distributed/Process/ManagedProcess.hs +++ b/src/Control/Distributed/Process/ManagedProcess.hs @@ -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(..)) diff --git a/src/Control/Distributed/Process/ManagedProcess/Internal/GenProcess.hs b/src/Control/Distributed/Process/ManagedProcess/Internal/GenProcess.hs --- a/src/Control/Distributed/Process/ManagedProcess/Internal/GenProcess.hs +++ b/src/Control/Distributed/Process/ManagedProcess/Internal/GenProcess.hs @@ -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 diff --git a/src/Control/Distributed/Process/ManagedProcess/Internal/PriorityQueue.hs b/src/Control/Distributed/Process/ManagedProcess/Internal/PriorityQueue.hs new file mode 100644 --- /dev/null +++ b/src/Control/Distributed/Process/ManagedProcess/Internal/PriorityQueue.hs @@ -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 diff --git a/src/Control/Distributed/Process/ManagedProcess/Internal/Types.hs b/src/Control/Distributed/Process/ManagedProcess/Internal/Types.hs --- a/src/Control/Distributed/Process/ManagedProcess/Internal/Types.hs +++ b/src/Control/Distributed/Process/ManagedProcess/Internal/Types.hs @@ -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/ diff --git a/src/Control/Distributed/Process/ManagedProcess/Server.hs b/src/Control/Distributed/Process/ManagedProcess/Server.hs --- a/src/Control/Distributed/Process/ManagedProcess/Server.hs +++ b/src/Control/Distributed/Process/ManagedProcess/Server.hs @@ -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 diff --git a/src/Control/Distributed/Process/ManagedProcess/Server/Gen.hs b/src/Control/Distributed/Process/ManagedProcess/Server/Gen.hs new file mode 100644 --- /dev/null +++ b/src/Control/Distributed/Process/ManagedProcess/Server/Gen.hs @@ -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 +-- 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 diff --git a/src/Control/Distributed/Process/ManagedProcess/Server/Priority.hs b/src/Control/Distributed/Process/ManagedProcess/Server/Priority.hs --- a/src/Control/Distributed/Process/ManagedProcess/Server/Priority.hs +++ b/src/Control/Distributed/Process/ManagedProcess/Server/Priority.hs @@ -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)) diff --git a/tests/TestPrioritisedProcess.hs b/tests/TestPrioritisedProcess.hs --- a/tests/TestPrioritisedProcess.hs +++ b/tests/TestPrioritisedProcess.hs @@ -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) ] ]