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.0 +version: 0.2.1 cabal-version: >=1.8 build-type: Simple license: BSD3 @@ -56,8 +56,7 @@ Control.Distributed.Process.ManagedProcess.Server, Control.Distributed.Process.ManagedProcess.Server.Priority, Control.Distributed.Process.ManagedProcess.Server.Restricted, - Control.Distributed.Process.ManagedProcess.Timer - other-modules: + Control.Distributed.Process.ManagedProcess.Timer, Control.Distributed.Process.ManagedProcess.Internal.Types, Control.Distributed.Process.ManagedProcess.Internal.GenProcess 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 @@ -29,6 +29,7 @@ , GenProcess , peek , push + , enqueue , addUserTimer , removeUserTimer , act @@ -118,6 +119,8 @@ -- Priority Mailbox Handling -- -------------------------------------------------------------------------------- +type Safe = Bool + -- | 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 @@ -277,6 +280,12 @@ prioritise = (\_ m' -> return $ Just ((101 :: Int), m')) :: s -> Message -> Process (Maybe (Int, Message)) } ] m +-- | Enqueue a message to the back of the internal priority queue. +enqueue :: forall s . Message -> GenProcess s () +enqueue m = do + st <- processState + enqueueMessage st [] 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). @@ -416,6 +425,10 @@ | ProcessStop xr <- ac = return xr | ProcessStopping ps' xr <- ac = setProcessState ps' >> return xr | ProcessHibernate d' s' <- ac = (lift $ block d') >> recvQueueAux s' + | ProcessBecome pd' ps' <- ac = do + modifyState $ \st@ProcessState{..} -> st { procDef = pd', procState = ps' } + -- liftIO $ putStrLn "modified process def" + recvQueue | otherwise {- compiler foo -} = return $ ExitOther "IllegalState" recvQueueAux st = setProcessState st >> recvQueue @@ -459,19 +472,26 @@ (up, pf) <- gets $ liftA2 (,) (unhandledMessagePolicy . procDef) procFilters case pf of [] -> consumeMessage - _ -> filterMessage (filterNext up pf Nothing) + _ -> filterMessage (filterNext False up pf Nothing) consumeMessage = applyNext dequeue processApply filterMessage = applyNext peek - filterNext :: UnhandledMessagePolicy + filterNext :: Safe + -> UnhandledMessagePolicy -> [DispatchFilter s] -> Maybe (Filter s) -> Message -> GenProcess s (ProcessAction s) - filterNext mp' fs mf msg + filterNext isSafe mp' fs mf msg + | Just (FilterSafe s') <- mf = filterNext True mp' fs (Just $ FilterOk s') msg | Just (FilterSkip s') <- mf = setProcessState s' >> dequeue >> return ProcessSkip | Just (FilterStop s' r) <- mf = return $ ProcessStopping s' r + | isSafe + , Just (FilterOk s') <- mf + , [] <- fs = do setProcessState s' + act' <- processApply msg + dequeue >> return act' | Just (FilterOk s') <- mf , [] <- fs = setProcessState s' >> applyNext dequeue processApply | Nothing <- mf, [] <- fs = applyNext dequeue processApply @@ -479,12 +499,12 @@ , (f:fs') <- fs = do setProcessState s' act' <- lift $ dynHandleFilter s' f msg - filterNext mp' fs' act' msg + filterNext isSafe 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 + lift (dynHandleFilter s' f msg) >>= \a -> filterNext isSafe mp' fs' a msg applyNext :: (GenProcess s (Maybe Message)) -> (Message -> GenProcess s (ProcessAction s)) @@ -497,13 +517,13 @@ 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':extMatchers) ++ apiMatchers ++ infoMatchers + -- liftIO $ putStrLn $ "we have " ++ (show $ (length apiMatchers, length infoMatchers)) ++ " handlers" processApplyAux ms' pol pState msg processApplyAux [] p' s' m' = lift $ applyPolicy p' s' m' @@ -709,6 +729,7 @@ (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) + (ProcessBecome d' s') -> recvLoop d' s' recvDelay (ProcessActivity _) -> die $ "recvLoop.InvalidState - ProcessActivityNotSupported" where matchAux :: UnhandledMessagePolicy 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 @@ -318,6 +318,7 @@ | ProcessHibernate TimeInterval s -- ^ hibernate for /delay/ | ProcessStop ExitReason -- ^ stop the process, giving @ExitReason@ | ProcessStopping s ExitReason -- ^ stop the process with @ExitReason@, with updated state + | ProcessBecome (ProcessDefinition s) s -- ^ changes the current process definition -- | Returned from handlers for the synchronous 'call' protocol, encapsulates -- the reply data /and/ the action to take after sending the reply. A handler @@ -447,6 +448,7 @@ -- for internal use. For an API for working with filters, -- see "Control.Distributed.Process.ManagedProcess.Priority". data Filter s = FilterOk s + | FilterSafe s | forall m . (Show m) => FilterReject m s | FilterSkip s | FilterStop s ExitReason 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 @@ -32,6 +32,7 @@ , replyChan , reject , rejectWith + , become -- * Stateless actions , noReply_ , haltNoReply_ @@ -188,6 +189,9 @@ -- hibernate_ :: StatelessHandler s TimeInterval hibernate_ d = return . ProcessHibernate d + +become :: forall s . ProcessDefinition s -> s -> Action s +become def 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' 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 @@ -4,6 +4,7 @@ {-# LANGUAGE RecordWildCards #-} {-# LANGUAGE EmptyDataDecls #-} {-# LANGUAGE StandaloneDeriving #-} +{-# LANGUAGE AllowAmbiguousTypes #-} ----------------------------------------------------------------------------- -- | @@ -44,6 +45,9 @@ , ensureM , Filter() , DispatchFilter() + , safe + , apiSafe + , safely , Message() , evalAfter , currentTimeout @@ -57,6 +61,8 @@ , peek , push , addUserTimer + , act + , runAfter ) where import Control.Distributed.Process hiding (call, Message) @@ -77,6 +83,8 @@ , peek , push , evalAfter + , act + , runAfter ) import Control.Distributed.Process.ManagedProcess.Internal.Types import Control.Distributed.Process.Serializable @@ -106,6 +114,10 @@ , 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)) } + | HandleSafe + { + safeCheck :: s -> P.Message -> Process Bool + } -- ^ A safe wrapper {- check :: forall c s m . (Check c s m) @@ -125,6 +137,10 @@ c <- apiCheck s m' if c then return $ FilterOk s else apiHandler s m + | HandleSafe{..} <- h = FilterRaw $ \s m -> do + c <- safeCheck s m + let ctr = if c then FilterSafe else FilterOk + return $ Just $ ctr s where procUnless s _ _ True = return $ FilterOk s @@ -171,6 +187,33 @@ -> (s -> m -> Process (Filter s)) -> FilterHandler s info_ c h = info (const $ c) h + +apiSafe :: forall s m b . (Serializable m, Serializable b) + => (s -> m -> Maybe b -> Bool) + -> DispatchFilter s +apiSafe c = check $ HandleSafe (go c) + where + go c' s (i :: P.Message) = do + m <- unwrapMessage i :: Process (Maybe (Message m b)) + case m of + Just (CallMessage m' _) -> return $ c' s m' Nothing + Just (CastMessage m') -> return $ c' s m' Nothing + Just (ChanMessage m' _) -> return $ c' s m' Nothing + Nothing -> return False + +safe :: forall s m . (Serializable m) + => (s -> m -> Bool) + -> DispatchFilter s +safe c = check $ HandleSafe (go c) + where + go c' s (i :: P.Message) = do + m <- unwrapMessage i :: Process (Maybe m) + case m of + Just m' -> return $ c' s m' + Nothing -> return False + +safely :: forall s . (s -> P.Message -> Bool) -> DispatchFilter s +safely c = check $ HandleSafe $ \s m -> return (c s m) -- | Create a filter expression that will reject all messages of a specific type. reject :: forall s m r . (Show r) diff --git a/tests/TestPrioritisedProcess.hs b/tests/TestPrioritisedProcess.hs --- a/tests/TestPrioritisedProcess.hs +++ b/tests/TestPrioritisedProcess.hs @@ -29,6 +29,7 @@ import Data.Binary import Data.Either (rights) import Data.List (isInfixOf) +import Data.Maybe (isNothing) import Data.Typeable (Typeable) #if ! MIN_VERSION_base(4,6,0) @@ -270,6 +271,111 @@ stash result $ m == 25 kill pid "done" +testServerSwap :: TestResult Bool -> Process () +testServerSwap result = do + us <- getSelfPid + let def2 = statelessProcess { apiHandlers = [ handleCast (\s (i :: Int) -> send us (i, i+1) >> continue s) + , handleCall_ (\(i :: Int) -> return (i * 5)) + ] + , unhandledMessagePolicy = Drop -- otherwise `call` would fail + } + let def = statelessProcess + { apiHandlers = [ handleCall_ (\(m :: String) -> return m) ] + , infoHandlers = [ handleInfo (\s () -> become def2 s) ] + } `prioritised` [] + + pid <- spawnLocal $ pserve () (statelessInit Infinity) def + + m1 <- call pid "hello there" + let a1 = m1 == "hello there" + + send pid () --changeover + + m2 <- callTimeout pid "are you there?" (seconds 5) :: Process (Maybe String) + let a2 = isNothing m2 + + cast pid (45 :: Int) + res <- receiveWait [ matchIf (\(i :: Int) -> i == 45) (return . Left) + , match (\(_ :: Int, j :: Int) -> return $ Right j) ] + + let a3 = res == (Right 46) + + m4 <- call pid (20 :: Int) :: Process Int + let a4 = m4 == 100 + + stash result $ a1 && a2 && a3 && a4 + +testSafeExecutionContext :: TestResult Bool -> Process () +testSafeExecutionContext result = do + let t = (asTimeout $ seconds 5) + (sigSp, rp) <- newChan + (wp, lp) <- newChan + let def = statelessProcess + { apiHandlers = [ handleCall_ (\(m :: String) -> stranded rp wp Nothing >> return m) ] + , infoHandlers = [ handleInfo (\s (m :: String) -> stranded rp wp (Just m) >> continue s) ] + , exitHandlers = [ handleExit (\_ s (_ :: String) -> continue s) ] + } `prioritised` [] + + let spec = def { filters = [ + safe (\_ (_ :: String) -> True) + , apiSafe (\_ (_ :: String) (_ :: Maybe String) -> True) + ] + } + + pid <- spawnLocal $ pserve () (statelessInit Infinity) spec + send pid "hello" -- pid can't process this as it's stuck waiting on rp + + sleep $ seconds 3 + exit pid "ooops" -- now we force an exit signal once the receiveWait finishes + sendChan sigSp () -- and allow the receiveWait to complete + send pid "hi again" + + -- at this point, "hello" should still be in the backing queue/mailbox + sleep $ seconds 3 + + -- We should still be seeing "hello", since the 'safe' block saved us from + -- losing a message when we handled and swallowed the exit signal. + -- We should not see "hi again" until after "hello" has been processed + h <- receiveChanTimeout t lp + -- say $ "first response: " ++ (show h) + let a1 = h == (Just "hello") + + sleep $ seconds 3 + + -- now we should have "hi again" waiting in the mailbox... + sendChan sigSp () -- we must release the handler a second time... + h2 <- receiveChanTimeout t lp + -- say $ "second response: " ++ (show h2) + let a2 = h2 == (Just "hi again") + + void $ spawnLocal $ call pid "reply-please" >>= sendChan wp + + -- the call handler should be stuck waiting on rp + Nothing <- receiveChanTimeout (asTimeout $ seconds 2) lp + + -- now let's force an exit, then release the handler to see if it runs again... + exit pid "ooops2" + + sleep $ seconds 2 + sendChan sigSp () + + h3 <- receiveChanTimeout t lp +-- say $ "third response: " ++ (show h3) + let a3 = h3 == (Just "reply-please") + + stash result $ a1 && a2 && a3 + + where + + stranded :: ReceivePort () -> SendPort String -> Maybe String -> Process () + stranded gate chan str = do + -- say $ "stranded with " ++ (show str) + void $ receiveWait [ matchChan gate return ] + sleep $ seconds 1 + case str of + Nothing -> return () + Just s -> sendChan chan s + testExternalTimedOverflowHandling :: TestResult Bool -> Process () testExternalTimedOverflowHandling result = do (pid, cp) <- launchStmOverloadServer -- default 10k mailbox drain limit @@ -475,6 +581,12 @@ , testCase "Firing internal timeouts" (delayedAssertion "expected our info handler to run after the timeout" localNode True testUserTimerHandling) + , testCase "Creating 'Safe' Handlers" + (delayedAssertion "expected our handler to run on the old message" + localNode True testSafeExecutionContext) + , testCase "Swapping ProcessDefinitions at runtime" + (delayedAssertion "expected our handler to exist in the new handler list" + localNode True testServerSwap) ] ]