distributed-process-client-server 0.2.0 → 0.2.1

raw patch · 6 files changed

+190/−9 lines, 6 filesPVP ok

version bump matches the API change (PVP)

API changes (from Hackage documentation)

Files

• distributed-process-client-server.cabal +2/−3
• src/Control/Distributed/Process/ManagedProcess/Internal/GenProcess.hs +27/−6
• src/Control/Distributed/Process/ManagedProcess/Internal/Types.hs +2/−0
• src/Control/Distributed/Process/ManagedProcess/Server.hs +4/−0
• src/Control/Distributed/Process/ManagedProcess/Server/Priority.hs +43/−0
• tests/TestPrioritisedProcess.hs +112/−0

@@ -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 

@@ -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

@@ -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

@@ -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'

@@ -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)

@@ -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)          ]       ]