Test suite improvements, sequence breaks early

Author: hyperthunk

src/Control/Distributed/Process/FSM.hs

@@ -32,6 +32,9 @@ initState = Yield
 enter :: forall s d . s -> FSM s d (Transition s d)
 enter = return . Enter
 
+resume :: forall s d . FSM s d (Transition s d)
+resume = return Remain
+
 event :: (Serializable m) => Event m
 event = Wait
 
@@ -88,6 +91,13 @@ infixr 9 ~>
 await :: forall s d m . (Serializable m) => Event m -> Step s d -> Step s d
 await = Await
 
+(*>) :: forall s d m . (Serializable m) => Event m -> Step s d -> Step s d
+(*>) = SafeWait
+infixr 9 *>
+
+safeWait :: forall s d m . (Serializable m) => Event m -> Step s d -> Step s d
+safeWait = SafeWait
+
 (~@) :: forall s d . (Eq s) => s -> FSM s d (Transition s d) -> Step s d
 (~@) = Perhaps
 infixr 9 ~@

src/Control/Distributed/Process/FSM/Internal/Process.hs

@@ -33,6 +33,7 @@ import Control.Distributed.Process.ManagedProcess
  , Action
  , InitHandler
  , InitResult(..)
+ , DispatchFilter
  , defaultProcess
  , prioritised
  )
@@ -46,6 +47,7 @@ import Control.Distributed.Process.ManagedProcess.Server
 import Control.Distributed.Process.ManagedProcess.Internal.Types
  ( ExitSignalDispatcher(..)
  )
+import Data.Maybe (isJust)
 import qualified Data.Sequence as Q (empty)
 -- import Control.Distributed.Process.Serializable (Serializable)
 -- import Control.Monad (void)
@@ -57,14 +59,14 @@ start ::  forall s d . (Show s) => s -> d -> (Step s d) -> Process ProcessId
 start s d p = spawnLocal $ run s d p
 
 run :: forall s d . (Show s) => s -> d -> (Step s d) -> Process ()
-run s d p = MP.pserve (s, d, p) fsmInit processDefinition
+run s d p = MP.pserve (s, d, p) fsmInit (processDefinition p)
 
 fsmInit :: forall s d . (Show s) => InitHandler (s, d, Step s d) (State s d)
 fsmInit (st, sd, prog) =
-  return $ InitOk (State st sd prog Nothing (const $ return ()) Q.empty) Infinity
+  return $ InitOk (State st sd prog prog Nothing (const $ return ()) Q.empty) Infinity
 
-processDefinition :: forall s d . (Show s) => PrioritisedProcessDefinition (State s d)
-processDefinition =
+processDefinition :: forall s d . (Show s) => Step s d -> PrioritisedProcessDefinition (State s d)
+processDefinition prog =
   (prioritised
     defaultProcess
     {
@@ -73,7 +75,17 @@ processDefinition =
                      ]
     , exitHandlers = [ ExitSignalDispatcher (\s _ m -> handleAllRawInputs s m >>= return . Just)
                      ]
-    } []) { filters = [safely] }
+    } []) { filters = (walkFSM prog []) }
+
+-- we should probably make a Foldable (Step s d) for this
+walkFSM :: forall s d . Step s d -> [DispatchFilter (State s d)] -> [DispatchFilter (State s d)]
+walkFSM st acc
+  | SafeWait  evt act <- st = walkFSM act $ safely (\_ m -> isJust $ decodeToEvent evt m) : acc
+  | Await     _   act <- st = walkFSM act acc
+  | Sequence  ac1 ac2 <- st = walkFSM ac1 $ walkFSM ac2 acc
+  | Init      ac1 ac2 <- st = walkFSM ac1 $ walkFSM ac2 acc
+  | Alternate ac1 ac2 <- st = walkFSM ac1 $ walkFSM ac2 acc -- both branches need filter defs
+  | otherwise               = acc
 
 handleRpcRawInputs :: forall s d . (Show s) => State s d
                    -> (P.Message, SendPort P.Message)

src/Control/Distributed/Process/FSM/Internal/Types.hs

@@ -67,7 +67,8 @@ import Data.Tuple (swap, uncurry)
 
 data State s d = State { stName  :: s
                        , stData  :: d
-                       , stProg  :: Step s d
+                       , stProg  :: Step s d -- original program
+                       , stInstr :: Step s d -- current step in the program
                        , stInput :: Maybe P.Message
                        , stReply :: (P.Message -> Process ())
                        , stTrans :: Seq (Transition s d)
@@ -101,6 +102,7 @@ instance forall m . (Typeable m) => Show (Event m) where
 data Step s d where
   Init      :: Step s d -> Step s d -> Step s d
   Yield     :: s -> d -> Step s d
+  SafeWait  :: (Serializable m) => Event m -> Step s d -> Step s d
   Await     :: (Serializable m) => Event m -> Step s d -> Step s d
   Always    :: (Serializable m) => (m -> FSM s d (Transition s d)) -> Step s d
   Perhaps   :: (Eq s) => s -> FSM s d (Transition s d) -> Step s d
@@ -114,6 +116,7 @@ instance forall s d . (Show s) => Show (Step s d) where
     | Init      _ _ <- st = "Init"
     | Yield     _ _ <- st = "Yield"
     | Await     _ s <- st = "Await (_ " ++ (show s) ++ ")"
+    | SafeWait  _ s <- st = "SafeWait (_ " ++ (show s) ++ ")"
     | Always    _   <- st = "Always _"
     | Perhaps   s _ <- st = "Perhaps (" ++ (show s) ++ ")"
     | Matching  _ _ <- st = "Matching _ _"
@@ -189,11 +192,16 @@ apply st msg step
       P.liftIO $ putStrLn "Init _ _"
       st' <- apply st msg is
       case st' of
-        Just s  -> apply (s { stProg = ns }) msg ns
+        Just s  -> apply (s { stProg = ns, stInstr = ns }) msg ns
         Nothing -> die $ ExitOther $ baseErr ++ ":InitFailed"
   | Yield     sn  sd  <- step = do
       P.liftIO $ putStrLn "Yield s d"
       return $ Just $ st { stName = sn, stData = sd }
+  | SafeWait evt act' <- step = do
+      let ev = decodeToEvent evt msg
+      P.liftIO $ putStrLn $ (show evt) ++ " decoded: " ++ (show $ isJust ev)
+      if isJust (ev) then apply st msg act'
+                     else (P.liftIO $ putStrLn $ "Cannot decode " ++ (show (evt, msg))) >> return Nothing
   | Await     evt act' <- step = do
       let ev = decodeToEvent evt msg
       P.liftIO $ putStrLn $ (show evt) ++ " decoded: " ++ (show $ isJust ev)
@@ -211,8 +219,8 @@ apply st msg step
       runFSM st (handleMessageIf msg chk fsm) >>= mstash
   | Sequence  ac1 ac2 <- step = do s <- apply st msg ac1
                                    P.liftIO $ putStrLn $ "Seq LHS valid: " ++ (show $ isJust s)
-                                   let st' = if isJust s then fromJust s else st
-                                   apply st' msg ac2
+                                   if isJust s then apply (fromJust s) msg ac2
+                                               else return Nothing
   | Alternate al1 al2 <- step = do s <- apply st msg al1
                                    P.liftIO $ putStrLn $ "Alt LHS valid: " ++ (show $ isJust s)
                                    if isJust s then return s

stack-ghc-7.10.3.yaml

@@ -4,7 +4,7 @@ packages:
 - '.'
 - location:
     git: https://github.com/haskell-distributed/distributed-process-client-server.git
-    commit: 36b179ec804bfdf380286082255e8b1fae099414
+    commit: 5589d9ef5a50b86d489797f78c0118f75e53659e
   extra-dep: true
 
 extra-deps:

stack-ghc-8.0.1.yaml

@@ -4,7 +4,7 @@ packages:
 - '.'
 - location:
     git: https://github.com/haskell-distributed/distributed-process-client-server.git
-    commit: 36b179ec804bfdf380286082255e8b1fae099414
+    commit: 5589d9ef5a50b86d489797f78c0118f75e53659e
   extra-dep: true
 
 extra-deps:

stack.yaml

@@ -4,7 +4,7 @@ packages:
 - '.'
 - location:
     git: https://github.com/haskell-distributed/distributed-process-client-server.git
-    commit: 36b179ec804bfdf380286082255e8b1fae099414
+    commit: 5589d9ef5a50b86d489797f78c0118f75e53659e
   extra-dep: true
 
 extra-deps:

tests/TestFSM.hs

@@ -25,7 +25,7 @@ import Control.Rematch (equalTo)
 #if ! MIN_VERSION_base(4,6,0)
 import Prelude hiding (catch, drop)
 #else
-import Prelude hiding (drop)
+import Prelude hiding (drop, (*>))
 #endif
 
 import Test.Framework as TF (defaultMain, testGroup, Test)
@@ -89,14 +89,82 @@ switchFsmAlt =
                (pick (await (event :: Event Check) (reply stateData))
                      (await (event :: Event Reset) (always $ \Reset -> put initCount >> enter Off))))
 
+blockingFsm :: SendPort () -> Step State ()
+blockingFsm sp = initState Off ()
+            ^.  ((event :: Event ())
+                *> (allState $ \() -> (lift $ sleep (seconds 10) >> sendChan sp ()) >> resume))
+             .| ((event :: Event Stop)
+                ~> (  ((== ExitNormal)   ~? (\_ -> (liftIO $ putStrLn "resuming...") >> resume) )
+                      {- let's verify that we can't override
+                         a normal shutdown sequence... -}
+                   .| ((== ExitShutdown) ~? const resume)
+                   ))
+
+deepFSM :: SendPort () -> SendPort () -> Step State ()
+deepFSM on off = initState Off ()
+       ^. ((event :: Event State) ~> (allState $ \s -> enter s))
+       .| ( (Off ~@ resume)
+            |> ((event :: Event ())
+                ~> (allState $ \s -> (lift $ sendChan off s) >> resume))
+          )
+       .| ( (On ~@ resume)
+            |> ((event :: Event ())
+                ~> (allState $ \s -> (lift $ sendChan on s) >> resume))
+          )
+
+waitForDown :: MonitorRef -> Process DiedReason
+waitForDown ref =
+  receiveWait [ matchIf (\(ProcessMonitorNotification ref' _ _) -> ref == ref')
+                        (\(ProcessMonitorNotification _ _ dr) -> return dr) ]
+
+verifyOuterStateHandler :: Process ()
+verifyOuterStateHandler = do
+  (spOn, rpOn) <- newChan
+  (spOff, rpOff) <- newChan
+
+  pid <- start Off () $ deepFSM spOn spOff
+
+  send pid On
+  send pid ()
+  Nothing <- receiveChanTimeout (asTimeout $ seconds 3) rpOff
+  () <- receiveChan rpOn
+
+  send pid Off
+  send pid ()
+  Nothing <- receiveChanTimeout (asTimeout $ seconds 3) rpOn
+  () <- receiveChan rpOff
+
+  kill pid "bye bye"
+
+verifyMailboxHandling :: Process ()
+verifyMailboxHandling = do
+  (sp, rp) <- newChan :: Process (SendPort (), ReceivePort ())
+  pid <- start Off () (blockingFsm sp)
+
+  send pid ()
+  exit pid ExitNormal
+
+  sleep $ seconds 5
+  alive <- isProcessAlive pid
+  alive `shouldBe` equalTo True
+
+  -- we should resume after the ExitNormal handler runs, and get back into the ()
+  -- handler due to safeWait (*>) which adds a `safe` filter check for the given type
+  () <- receiveChan rp
+
+  exit pid ExitShutdown
+  monitor pid >>= waitForDown
+  alive' <- isProcessAlive pid
+  alive' `shouldBe` equalTo False
+
 verifyStopBehaviour :: Process ()
 verifyStopBehaviour = do
   pid <- start Off initCount switchFsm
   alive <- isProcessAlive pid
   alive `shouldBe` equalTo True
 
   exit pid $ ExitOther "foobar"
-  sleep $ seconds 5
+  monitor pid >>= waitForDown
   alive' <- isProcessAlive pid
   alive' `shouldBe` equalTo False
 
@@ -132,7 +200,7 @@ walkingAnFsmTree pid = do
   mrst' `shouldBe` equalTo On
 
   exit pid ExitShutdown
-  sleep $ seconds 5
+  monitor pid >>= waitForDown
   alive' <- isProcessAlive pid
   alive' `shouldBe` equalTo False
 
@@ -151,6 +219,12 @@ tests transport = do
            (runProcess localNode quirkyOperators)
         , testCase "Traversing an FSM definition (functions)"
            (runProcess localNode notSoQuirkyDefinitions)
+        , testCase "Traversing an FSM definition (exit handling)"
+           (runProcess localNode verifyStopBehaviour)
+        , testCase "Traversing an FSM definition (mailbox handling)"
+           (runProcess localNode verifyMailboxHandling)
+        , testCase "Traversing an FSM definition (nested definitions)"
+           (runProcess localNode verifyOuterStateHandler)
         ]
     ]