Prioritise event listeners and allow for queue modifications

Author: hyperthunk

src/Control/Distributed/Process/FSM.hs

@@ -13,6 +13,7 @@
 
 module Control.Distributed.Process.FSM where
 
+import Control.Distributed.Process (wrapMessage)
 import Control.Distributed.Process.Extras (ExitReason)
 import Control.Distributed.Process.Extras.Time
  ( TimeInterval
@@ -21,23 +22,40 @@ import Control.Distributed.Process.ManagedProcess
  ( processState
  , setProcessState
  , runAfter
+ , Priority
  )
+import Control.Distributed.Process.ManagedProcess.Server.Priority (setPriority)
 import qualified Control.Distributed.Process.ManagedProcess.Internal.Types as MP (liftIO)
 import Control.Distributed.Process.FSM.Internal.Types
 import Control.Distributed.Process.Serializable (Serializable)
 
 initState :: forall s d . s -> d -> Step s d
 initState = Yield
 
+event :: (Serializable m) => Event m
+event = Wait
+
+pevent :: (Serializable m) => Int -> Event m
+pevent = WaitP . setPriority
+
 enter :: forall s d . s -> FSM s d (Transition s d)
 enter = return . Enter
 
+postpone :: forall s d . FSM s d (Transition s d)
+postpone = return Postpone
+
+putBack :: forall s d . FSM s d (Transition s d)
+putBack = return PutBack
+
+nextEvent :: forall s d m . (Serializable m) => m -> FSM s d (Transition s d)
+nextEvent m = return $ Push (wrapMessage m)
+
+publishEvent :: forall s d m . (Serializable m) => m -> FSM s d (Transition s d)
+publishEvent m = return $ Enqueue (wrapMessage m)
+
 resume :: forall s d . FSM s d (Transition s d)
 resume = return Remain
 
-event :: (Serializable m) => Event m
-event = Wait
-
 reply :: forall s d r . (Serializable r) => FSM s d r -> Step s d
 reply = Reply
 
@@ -105,6 +123,9 @@ infixr 9 ~@
 atState :: forall s d . (Eq s) => s -> FSM s d (Transition s d) -> Step s d
 atState = Perhaps
 
+whenStateIs :: forall s d . (Eq s) => s -> Step s d
+whenStateIs s = s ~@ resume
+
 allState :: forall s d m . (Serializable m) => (m -> FSM s d (Transition s d)) -> Step s d
 allState = Always
 

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

@@ -21,6 +21,7 @@ import Control.Distributed.Process
  , sendChan
  , spawnLocal
  , liftIO
+ , handleMessage
  )
 import qualified Control.Distributed.Process as P
  ( Message
@@ -38,14 +39,17 @@ import Control.Distributed.Process.ManagedProcess
  , prioritised
  )
 import qualified Control.Distributed.Process.ManagedProcess as MP (pserve)
-import Control.Distributed.Process.ManagedProcess.Server.Priority (safely)
+import Control.Distributed.Process.ManagedProcess.Server.Priority
+ ( safely
+ )
 import Control.Distributed.Process.ManagedProcess.Server
  ( handleRaw
  , handleInfo
  , continue
  )
 import Control.Distributed.Process.ManagedProcess.Internal.Types
  ( ExitSignalDispatcher(..)
+ , DispatchPriority(PrioritiseInfo)
  )
 import Data.Maybe (isJust)
 import qualified Data.Sequence as Q (empty)
@@ -55,15 +59,16 @@ import qualified Data.Sequence as Q (empty)
 -- import Data.Typeable (Typeable)
 -- import GHC.Generics
 
-start ::  forall s d . (Show s) => s -> d -> (Step s d) -> Process ProcessId
+start ::  forall s d . (Show s, Eq 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 :: forall s d . (Show s, Eq s) => s -> d -> (Step s d) -> Process ()
 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 :: forall s d . (Show s, Eq s) => InitHandler (s, d, Step s d) (State s d)
 fsmInit (st, sd, prog) =
-  return $ InitOk (State st sd prog prog Nothing (const $ return ()) Q.empty) Infinity
+  let st' = State st sd prog Nothing (const $ return ()) Q.empty Q.empty
+  in return $ InitOk st' Infinity
 
 processDefinition :: forall s d . (Show s) => Step s d -> PrioritisedProcessDefinition (State s d)
 processDefinition prog =
@@ -75,10 +80,12 @@ processDefinition prog =
                      ]
     , exitHandlers = [ ExitSignalDispatcher (\s _ m -> handleAllRawInputs s m >>= return . Just)
                      ]
-    } []) { filters = (walkFSM prog []) }
+    } (walkPFSM prog [])) { 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)]
+-- we should probably make a Foldable (Step s d) for these
+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
@@ -87,6 +94,20 @@ walkFSM st acc
   | Alternate ac1 ac2 <- st = walkFSM ac1 $ walkFSM ac2 acc -- both branches need filter defs
   | otherwise               = acc
 
+walkPFSM :: forall s d . Step s d
+         -> [DispatchPriority (State s d)]
+         -> [DispatchPriority (State s d)]
+walkPFSM st acc
+  | SafeWait  evt act <- st  = walkPFSM act (checkPrio evt acc)
+  | Await     evt act <- st  = walkPFSM act (checkPrio evt acc)
+  | Sequence  ac1 ac2 <- st  = walkPFSM ac1 $ walkPFSM ac2 acc
+  | Init      ac1 ac2 <- st  = walkPFSM ac1 $ walkPFSM ac2 acc
+  | Alternate ac1 ac2 <- st  = walkPFSM ac1 $ walkPFSM ac2 acc -- both branches need filter defs
+  | otherwise                = acc
+  where
+    checkPrio ev acc = (mkPrio ev):acc
+    mkPrio ev' = PrioritiseInfo $ \s m -> handleMessage m (resolveEvent ev' m s)
+
 handleRpcRawInputs :: forall s d . (Show s) => State s d
                    -> (P.Message, SendPort P.Message)
                    -> Action (State s d)
@@ -108,7 +129,7 @@ handleInput :: forall s d . (Show s)
             -> Action (State s d)
 handleInput msg st@State{..} = do
   liftIO $ putStrLn $ "handleInput: " ++ (show stName)
-  liftIO $ putStrLn $ "apply " ++ (show stProg)
+  liftIO $ putStrLn $ "apply: " ++ (show stProg)
   res <- apply st msg stProg
   liftIO $ putStrLn $ "got a result: " ++ (show res)
   case res of

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

@@ -35,11 +35,15 @@ import Control.Distributed.Process.ManagedProcess
  , setProcessState
  , processState
  )
-import qualified Control.Distributed.Process.ManagedProcess.Internal.Types as MP (lift, liftIO)
-import Control.Distributed.Process.ManagedProcess.Server.Priority
- ( push
- , act
+import qualified Control.Distributed.Process.ManagedProcess.Internal.GenProcess as Gen (enqueue, push)
+import Control.Distributed.Process.ManagedProcess.Internal.Types
+ ( Priority(..)
  )
+import qualified Control.Distributed.Process.ManagedProcess.Internal.Types as MP
+ ( lift
+ , liftIO
+ )
+import Control.Distributed.Process.ManagedProcess.Server.Priority (act)
 import Control.Distributed.Process.Serializable (Serializable)
 import Control.Monad.Fix (MonadFix)
 import Control.Monad.IO.Class (MonadIO)
@@ -65,13 +69,14 @@ import Data.Typeable (Typeable, typeOf)
 import Data.Tuple (swap, uncurry)
 -- import GHC.Generics
 
-data State s d = State { stName  :: s
+data State s d = (Show s, Eq s) =>
+                 State { stName  :: s
                        , stData  :: 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)
+                       , stQueue :: Seq P.Message
                        }
 
 instance forall s d . (Show s) => Show (State s d) where
@@ -80,24 +85,42 @@ instance forall s d . (Show s) => Show (State s d) where
 
 data Transition s d = Remain
                     | PutBack
+                    | Push P.Message
+                    | Enqueue P.Message
+                    | Postpone
                     | Enter s
                     | Stop ExitReason
                     | Eval (GenProcess (State s d) ())
 
 instance forall s d . (Show s) => Show (Transition s d) where
   show Remain    = "Remain"
   show PutBack   = "PutBack"
+  show Postpone  = "Postpone"
+  show (Push m)  = "Push " ++ (show m)
+  show (Enqueue m) = "Enqueue " ++ (show m)
   show (Enter s) = "Enter " ++ (show s)
   show (Stop er) = "Stop " ++ (show er)
   show (Eval _)  = "Eval"
 
 data Event m where
-  Wait  :: (Serializable m) => Event m
-  Event :: (Serializable m) => m -> Event m
+  Wait    :: (Serializable m) => Event m
+  WaitP   :: (Serializable m) => Priority () -> Event m
+  Event   :: (Serializable m) => m -> Event m
+
+resolveEvent :: forall s d m . (Serializable m)
+             => Event m
+             -> P.Message
+             -> State s d
+             -> m
+             -> Process (Int, P.Message)
+resolveEvent ev m _ _
+  | WaitP p <- ev       = return (getPrio p, m)
+  | otherwise           = return (0, m)
 
 instance forall m . (Typeable m) => Show (Event m) where
-  show ev@Wait = show $ "Wait::" ++ (show $ typeOf ev)
-  show ev      = show $ typeOf ev
+  show ev@Wait      = show $ "Wait::" ++ (show $ typeOf ev)
+  show ev@(WaitP _) = show $ "WaitP::" ++ (show $ typeOf ev)
+  show ev           = show $ typeOf ev
 
 data Step s d where
   Init      :: Step s d -> Step s d -> Step s d
@@ -171,9 +194,13 @@ seqEnqueue s a = a <| s
 seqPush :: Seq a -> a -> Seq a
 seqPush s a = s |> a
 
+{-# INLINE seqPop #-}
+seqPop :: Seq a -> Maybe (a, Seq a)
+seqPop s = maybe Nothing (\(s' :> a) -> Just (a, s')) $ getR s
+
 {-# INLINE seqDequeue #-}
 seqDequeue :: Seq a -> Maybe (a, Seq a)
-seqDequeue s = maybe Nothing (\(s' :> a) -> Just (a, s')) $ getR s
+seqDequeue = seqPop
 
 {-# INLINE peek #-}
 peek :: Seq a -> Maybe a
@@ -198,7 +225,7 @@ apply st msg step
       P.liftIO $ putStrLn "Init _ _"
       st' <- apply st msg is
       case st' of
-        Just s  -> apply (s { stProg = ns, stInstr = ns }) msg ns
+        Just s  -> apply (s { stProg = ns }) msg ns
         Nothing -> die $ ExitOther $ baseErr ++ ":InitFailed"
   | Yield     sn  sd  <- step = do
       P.liftIO $ putStrLn "Yield s d"
@@ -255,28 +282,38 @@ applyTransitions st@State{..} evals
                               MP.liftIO $ putStrLn $ "ProcessState: " ++ (show stName)
                               mapM_ id evals
   | (tr, st2) <- next
-  , Enter s   <- tr = let st' = st2 { stName = s }
-                      in do P.liftIO $ putStrLn $ "NEWSTATE: " ++ (show st')
-                            applyTransitions st' evals
+  , PutBack   <- tr = applyTransitions st2 ((Gen.enqueue $ fromJust stInput) : evals)
+  | isJust stInput
+  , input     <- fromJust stInput
+  , (tr, st2) <- next
+  , Postpone  <- tr = applyTransitions (st2 { stQueue = seqEnqueue stQueue input }) evals
   | (tr, st2) <- next
-  , PutBack   <- tr = applyTransitions st2 ((push $ fromJust stInput) : evals)
-  {- let act' = setProcessState $ fsmSt { fsmName = stName, fsmData = stData }
-                                     push stInput -}
+  , Enqueue m <- tr = applyTransitions st2 ((Gen.enqueue m):evals)
+  | (tr, st2) <- next
+  , Push m    <- tr = applyTransitions st2 ((Gen.push m):evals)
   | (tr, st2) <- next
   , Eval proc <- tr = applyTransitions st2 (proc:evals)
   | (tr, st2) <- next
   , Remain    <- tr = applyTransitions st2 evals
   | (tr, _)   <- next
   , Stop  er  <- tr = stopWith st er
-  | otherwise = error $ baseErr ++ ".Internal.Process.applyTransitions:InvalidState"
+  | (tr, st2) <- next
+  , Enter s   <- tr =
+      if s == stName then applyTransitions st2 evals
+                     else do let st' = st2 { stName = s }
+                             let evals' = if Q.null stQueue then evals
+                                                            else (mapM_ Gen.push stQueue) : evals
+                             applyTransitions st' evals'
+  | otherwise = error $ baseErr ++ ".Internal.Process.applyTransitions:InvalidTransition"
   where
     -- don't call if Q.null!
-    next = let (t, q) = fromJust $ seqDequeue stTrans
+    next = let (t, q) = fromJust $ seqPop stTrans
            in (t, st { stTrans = q })
 
 baseErr :: String
 baseErr = "Control.Distributed.Process.FSM"
 
 decodeToEvent :: Serializable m => Event m -> P.Message -> Maybe (Event m)
 decodeToEvent Wait         msg = unwrapMessage msg >>= fmap Event
+decodeToEvent (WaitP _)    msg = unwrapMessage msg >>= fmap Event
 decodeToEvent ev@(Event _) _   = Just ev  -- it's a bit odd that we'd end up here....

stack-ghc-7.10.3.yaml

@@ -4,7 +4,7 @@ packages:
 - '.'
 - location:
     git: https://github.com/haskell-distributed/distributed-process-client-server.git
-    commit: 5589d9ef5a50b86d489797f78c0118f75e53659e
+    commit: ceaf6b48a6cc90e2bc726025ce3a28b40899440c
   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: 5589d9ef5a50b86d489797f78c0118f75e53659e
+    commit: ceaf6b48a6cc90e2bc726025ce3a28b40899440c
   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: 5589d9ef5a50b86d489797f78c0118f75e53659e
+    commit: ceaf6b48a6cc90e2bc726025ce3a28b40899440c
   extra-dep: true
 
 extra-deps: