diff --git a/LICENCE b/LICENCE new file mode 100644 --- /dev/null +++ b/LICENCE @@ -0,0 +1,30 @@ +Copyright Tim Watson, 2017. + +All rights reserved. + +Redistribution and use in source and binary forms, with or without +modification, are permitted provided that the following conditions are met: + + * Redistributions of source code must retain the above copyright + notice, this list of conditions and the following disclaimer. + + * Redistributions in binary form must reproduce the above + copyright notice, this list of conditions and the following + disclaimer in the documentation and/or other materials provided + with the distribution. + + * Neither the name of the author nor the names of other + contributors may be used to endorse or promote products derived + from this software without specific prior written permission. + +THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS +"AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT +LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR +A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT +OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, +SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT +LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, +DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY +THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT +(INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE +OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. diff --git a/Setup.lhs b/Setup.lhs new file mode 100644 --- /dev/null +++ b/Setup.lhs @@ -0,0 +1,3 @@ +#!/usr/bin/env runhaskell +> import Distribution.Simple +> main = defaultMain diff --git a/distributed-process-fsm.cabal b/distributed-process-fsm.cabal new file mode 100644 --- /dev/null +++ b/distributed-process-fsm.cabal @@ -0,0 +1,85 @@ +name: distributed-process-fsm +version: 0.0.1 +cabal-version: >=1.8 +build-type: Simple +license: BSD3 +license-file: LICENCE +stability: experimental +Copyright: Tim Watson 2017 +Author: Tim Watson +Maintainer: Tim Watson +Stability: experimental +Homepage: http://github.com/haskell-distributed/distributed-process-fsm +Bug-Reports: http://github.com/haskell-distributed/distributed-process-fsm/issues +synopsis: The Cloud Haskell implementation of Erlang/OTP gen_statem +description: Cloud Haskell framework for building finite state machines around CSPs +category: Control +Tested-With: GHC==7.10.3 GHC==8.0.1 GHC==8.0.2 +data-dir: "" + +source-repository head + type: git + location: https://github.com/haskell-distributed/distributed-process-fsm + +library + build-depends: + base >= 4.8.2.0 && < 5, + distributed-process >= 0.6.6 && < 0.7, + distributed-process-extras >= 0.3.1 && < 0.4, + distributed-process-client-server >= 0.2.1 && < 0.3, + binary >= 0.6.3.0 && < 0.9, + deepseq >= 1.3.0.1 && < 1.6, + mtl, + containers >= 0.4 && < 0.6, + unordered-containers >= 0.2.3.0 && < 0.3, + stm >= 2.4 && < 2.5, + time > 1.4 && < 1.8, + transformers, + exceptions >= 0.5 + extensions: CPP + hs-source-dirs: src + ghc-options: -Wall + exposed-modules: + Control.Distributed.Process.FSM, + Control.Distributed.Process.FSM.Client, + Control.Distributed.Process.FSM.Internal.Types, + Control.Distributed.Process.FSM.Internal.Process + +test-suite FsmTests + type: exitcode-stdio-1.0 + -- x-uses-tf: true + build-depends: + base >= 4.4 && < 5, + ansi-terminal >= 0.5 && < 0.7, + network >= 2.3 && < 2.7, + network-transport >= 0.4 && < 0.5, + network-transport-tcp >= 0.4 && < 0.6, + distributed-process >= 0.6.6 && < 0.7, + distributed-process-fsm, + distributed-process-extras >= 0.3.1 && < 0.4, + distributed-process-systest >= 0.1.1 && < 0.2, + distributed-static, + binary >= 0.6.3.0 && < 0.9, + bytestring, + containers, + data-accessor, + deepseq >= 1.3.0.1 && < 1.5, + fingertree < 0.2, + hashable, + mtl, + stm >= 2.3 && < 2.5, + time, + transformers, + unordered-containers >= 0.2.3.0 && < 0.3, + test-framework >= 0.6 && < 0.9, + test-framework-hunit, + QuickCheck >= 2.4, + test-framework-quickcheck2, + HUnit >= 1.2 && < 2, + rematch >= 0.2.0.0, + ghc-prim + hs-source-dirs: + tests + ghc-options: -Wall -threaded -rtsopts -with-rtsopts=-N -fno-warn-unused-do-bind -eventlog + extensions: CPP + main-is: TestFSM.hs diff --git a/src/Control/Distributed/Process/FSM.hs b/src/Control/Distributed/Process/FSM.hs new file mode 100644 --- /dev/null +++ b/src/Control/Distributed/Process/FSM.hs @@ -0,0 +1,645 @@ +{-# LANGUAGE ExistentialQuantification #-} + +----------------------------------------------------------------------------- +-- | +-- Module : Control.Distributed.Process.FSM +-- Copyright : (c) Tim Watson 2017 +-- License : BSD3 (see the file LICENSE) +-- +-- Maintainer : Tim Watson +-- Stability : experimental +-- Portability : non-portable (requires concurrency) +-- +-- A /Managed Process/ API for building state machines. Losely based +-- on http://erlang.org/doc/man/gen_statem.html, but with a Haskell-ish +-- flavour. +-- +-- A state machine is defined by a "Step" that is executed whenever an input +-- arrives in the process mailbox. This "Step" will usually be produced by +-- evaluating the combinator-style API provided by this module, to link +-- events with actions and transitions that communicate responses back to +-- client processes, alter the state (or state data), and so on. +-- +-- [Overview and Examples] +-- +-- The "Step" that defines our state machines is parameterised by two types, +-- @s@ and @d@, which represent the state identity (e.g. state name) and +-- state data. These types are fixed, such that if you wish to alternate the +-- type or form state data for different state identities, you will need to +-- encode the relevant storage facilities into the state type @s@ (or choose +-- to align the two types and ensure they are adjusted in tandem yourself). +-- +-- The following example shows a simple pushbutton model for a toggling +-- push-button. You can push the button and it replies if it went on or off, +-- and you can ask for a count of how many times the switch has been pushed. +-- +-- We begin by defining the types we'll be working with. An algebraic data +-- type for the state (identity), an "Int" as the state data (for holding the +-- counter), and a request datum for issuing a /check/ on the number of times +-- the button has been set to @On@. Pushing the button will be represented +-- by unit. +-- +-- > data State = On | Off deriving (Eq, Show, Typeable, Generic) +-- > instance Binary State where +-- > +-- > data Check = Check deriving (Eq, Show, Typeable, Generic) +-- > instance Binary Check where +-- > +-- > type StateData = Integer +-- > type ButtonPush = () +-- +-- We define our starting state as a "Step", which yields the @Off@ state id +-- and an initial counter set to zero. +-- +-- > startState :: Step State Integer +-- > startState = yield Off initCount +-- > +-- > initCount :: StateData +-- > initCount = 0 +-- +-- To capture what happens when a specific event arrives in the mailbox, we +-- evaluate the 'event' function and specify the type it accepts. When combined +-- with 'await', this maps input messages of a specific type to actions and +-- state transitions. +-- +-- > await (event :: Event ButtonPush) actions +-- +-- Since our @startState@ is going to yield a specific state and data, we don't +-- want it to be evaluated each time we handle a message. The 'begin' function +-- ensures that its first argument is only evaluated once, on our first pass of +-- the state machine's instruction set (i.e. the "Step" structure that determines +-- its runtime execution). Thus we have: +-- +-- > begin startState $ await (event :: Event ButtonPush) actions +-- +-- In response to a button push, we want to set the state to it's opposite +-- setting. We behave differently in the two states, so we can't simply write +-- @if state == On then enter Off else enter On@, therefore we will use 'atState' +-- to execute an action only if the current state matches our input. +-- +-- > actions = atState On enter Off +-- +-- Now of course we need to choose between two states. We could use 'allState' +-- and switch on the state id in our code, but 'pick' provides an API that lets +-- us stay in the combinatorial pattern instead: +-- +-- > actions = ((pick (atState On enter Off)) +-- > (atState Off (set_ (+1) >> enter On))) +-- +-- The 'set_' function alters our state data (used to track the button click to +-- @On@ count), and the 'enter' function produces a "Transition". Transitions +-- can alter the sate id, state data, and/or determine what the server process +-- will do next (e.g. stop, timeout, etc).. +-- +-- > atState :: forall s d . (Eq s) => s -> FSM s d (Transition s d) -> Step s d +-- +-- Looking at the signature of 'atState', we can see that it takes a state id for +-- comparison, and an action in the "FSM" monad which evaluates to a "Transition". +-- Notice that the 'set_' action does not yield a transition. In fact, 'set' does, +-- but we need to throw it away by using monadic @>>@, so we can introduce the +-- 'enter' transition. +-- +-- Essentially this is all syntactic sugar. What happens here is that 'set_' +-- evaluates 'addTransition', which can be used to queue up multiple transitions. +-- So what we really want is @addTransition set (+1) >> addTransition enter On@, +-- however 'atState' wants an action that produces a "Transition", and finishing +-- our /sentence/ with @enter newState@ reads rather nicely, so we've opted for +-- @set (+1) >> enter On@ in the end. +-- +-- Only one options presents itself for replying to clients, and that is 'reply'. +-- Since an FSM process deals with control flow quite differently to an ordinary +-- managed process - taking input messages and passing them through the "Step" +-- definitions that operate as a simple state machine - we do not leverage the +-- usual @call@ APIs and instead utilise rpc channels to handle synchronous +-- client/server style interactions with a state machine process. +-- +-- Thus we treat replying as a separate "Step", and use 'join' to combine the +-- 'await' "Step" with 'reply' such that we have something akin to +-- +-- > (await (event :: Event ButtonPush) actions) `join` (reply currentState)) +-- +-- Putting this all together, we will now replace 'await', 'atState', 'join' and +-- so on, with their equivalent synonyms, provided as operators to make the +-- combinator pattern style look a bit more like an internal DSL. We end up with: +-- +-- > switchFsm :: Step State StateData +-- > switchFsm = startState +-- > ^. ((event :: Event ButtonPush) +-- > ~> ( (On ~@ enter Off)) +-- > .| (Off ~@ (set_ (+1) >> enter On)) +-- > ) |> (reply currentState)) +-- +-- Our client code will need to use the @call@ function from the Client module, +-- although it /is/ possible to interact synchronously with an FSM process (e.g. +-- in client/server mode) by hand, the implementation is very likely to change +-- in a future release and this isn't advised. +-- +-- To wire a synchronous @call@ up, we need to supply information about the +-- input and expected response types at the call site. These are used to determine +-- the type of channels used to communicate with the server. +-- +-- > pushButton :: ProcessId -> Process State +-- > pushButton pid = call pid (() :: ButtonPush) +-- +-- Starting a new switch server process is fairly simple: +-- +-- > pid <- start Off initCount switchFsm +-- +-- And we can interact with it using our defined function. +-- +-- > mSt <- pushButton pid +-- > mSt `shouldBe` equalTo On +-- > +-- > mSt' <- pushButton pid +-- > mSt' `shouldBe` equalTo Off +-- +-- However we haven't got a way to query the switched-on count yet. Let's add +-- that now. We will send our @Check@ datum to the server process, and expect +-- an @Int@ reply. +-- +-- > switchFsm :: Step State StateData +-- > switchFsm = startState +-- > ^. ((event :: Event ButtonPush) +-- > ~> ( (On ~@ (set_ (+1) >> enter Off)) -- on => off => on is possible with |> here... +-- > .| (Off ~@ (set_ (+1) >> enter On)) +-- > ) |> (reply currentState)) +-- > .| ((event :: Event Check) ~> reply stateData) +-- > +-- +-- Notice that we can still use the @(.|)@ operator - a synonym for 'pick' - +-- here, since we're picking between two branches based on the type of the event +-- received. The 'reply' function takes an action in the "FSM" monad, which must +-- evaluate to a "Serializable" type @r@, which is sent back to the client. +-- +-- We can now write our check function... +-- +-- > check :: ProcessId -> Process StateData +-- > check pid = call pid Check +-- +-- This is exactly the same approach that we took with @pushButton@. We can +-- leverage this in our code too, so after we've evaluated the @pushButton@ +-- twice (as above), we will see +-- +-- > mCk <- check pid +-- > mCk `shouldBe` equalTo (1 :: StateData) +-- +-- How do we terminate our FSM process when we're done with it? A process +-- built using this API will respond to exit signals in a matter befitting a +-- managed process, e.g. suitable for use in a supervision tree. +-- +-- We can handle exit signals by registering listeners for them, as though they +-- were incoming events. The type we match on must be the type of the /exit reason/ +-- (whatever that may be, whether it is "ExitReason" or some other type), not +-- the exception type being thrown. +-- +-- Let's play around with this in our button state machine. We will /catch/ an +-- exit where the reason is @ExitNormal@ and instead of stopping, we'll timeout +-- after three seconds and publish a @Reset@ event to ourselves. +-- +-- > data Reset = Reset deriving (Eq, Show, Typeable, Generic) +-- > instance Binary Reset where +-- > +-- > switchFsm = startState +-- > ^. ((event :: Event ButtonPush) +-- > ~> ( (On ~@ (set_ (+1) >> enter Off)) -- on => off => on is possible with |> here... +-- > .| (Off ~@ (set_ (+1) >> enter On)) +-- > ) |> (reply currentState)) +-- > .| ((event :: Event ExitReason) +-- > ~> ((== ExitNormal) ~? (\_ -> timeout (seconds 3) Reset))) +-- > .| ((event :: Event Check) ~> reply stateData) +-- > .| (event :: Event Reset) +-- > ~> (allState $ \Reset -> put initCount >> enter Off) +-- +-- Here 'put' works similarly to 'set_' and 'allState' applies the action/transition +-- regardless of the current state. The @condition ~? action@ operator, a synonym +-- for 'matching', will only match if the conditional expression evaluates to +-- @True@. Obviously if the "ExitReason" is something other than @ExitNormal@ +-- we will not timeout, and in fact we will not handle the exit signal at all. +-- +-- In order to participate properly in a supervision tree, a process should +-- respond to the @ExitShutdown@ "ExitReason" by executing a clean shutdown and +-- stopping normally. What happens if we try to handle this "ExitReason" +-- ourselves? +-- +-- > .| ((event :: Event Stop) +-- > ~> ( ((== ExitNormal) ~? (\_ -> timeout (seconds 3) Reset)) +-- > .| ((== ExitShutdown) ~? (\_ -> timeout (seconds 3) Reset)) +-- > .| ((const True) ~? stop) +-- > )) +-- +-- We've added an expression to always stop when the previous two branches fail, +-- so that even @ExitOther@ will lead to a normal shutdown. Let's test this... +-- +-- > exit pid ExitNormal +-- > sleep $ seconds 6 +-- > alive <- isProcessAlive pid +-- > alive `shouldBe` equalTo True +-- > +-- > exit pid ExitShutdown +-- > monitor pid >>= waitForDown +-- > alive' <- isProcessAlive pid +-- > alive' `shouldBe` equalTo False +-- +-- So we can see that our override of @ExitShutdown@ has failed, and this is +-- because any process implemented with the /managed process/ API will respond +-- to @ExitShutdown@ by executing its termination handlers and stopping normally. +-- +-- We can add a shutdown handler quite easily, by dealing with the @Stopping@ +-- event type, like so: +-- +-- > (event :: Event Stopping) ~> actions +-- +-- While we're discussing exit signals, let's briefly cover the /safe/ API we +-- have available to us for ensuring that if an exit signal interrupts one of +-- our actions/transitions before it completes, but we handle that exit signal +-- without terminating, that we can re-try handling the event again. +-- +-- The 'safeWait' function, and its operator synonym @(*>)@ do precisely this. +-- Let's write up an example and test it. +-- +-- > blockingFsm :: SendPort () -> Step State () +-- > blockingFsm sp = initState Off () +-- > ^. ((event :: Event ()) +-- > *> (allState $ \() -> (lift $ sleep (seconds 10) >> sendChan sp ()) >> resume)) +-- > .| ((event :: Event Stop) +-- > ~> ( ((== ExitNormal) ~? (\_ -> resume) ) +-- > .| ((== ExitShutdown) ~? const resume) +-- > )) +-- > +-- > 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 +-- > +-- +-- [Prioritising Events and Manipulating the Event Queue] +-- +-- We will review these capabilities by example. Our state machine will respond +-- to button clicks by postponing the events when its state id is @Off@. In the +-- other state (i.e. @On@), it will prioritise events passing a new state, and +-- respond to button clicks by pushing them onto a typed channel. In addition, +-- we handle @Event String@ by either putting the event at the back of the total +-- event queue, or putting a @()@ at the front/head of the queue. +-- +-- > genFSM :: SendPort () -> Step State () +-- > genFSM sp = initState Off () +-- > ^. ( (whenStateIs Off) +-- > |> ((event :: Event ()) ~> (always $ \() -> postpone)) +-- > ) +-- > .| ( (((pevent 100) :: Event State) ~> (always $ \state -> enter state)) +-- > .| ((event :: Event ()) ~> (always $ \() -> (lift $ sendChan sp ()) >> resume)) +-- > ) +-- > .| ( (event :: Event String) +-- > ~> ( (Off ~@ putBack) +-- > .| (On ~@ (nextEvent ())) +-- > ) +-- > ) +-- +-- Notice that we're able to apply filters/conditions on both state and event +-- types at the /top level/ of our DSL. +-- +-- Our test case will be a bit racy, since we'll be relying on having loaded up +-- a backlog of messages and using priorities to jump the queue. +-- +-- > republicationOfEvents :: Process () +-- > republicationOfEvents = do +-- > (sp, rp) <- newChan +-- > +-- > pid <- start Off () $ genFSM sp +-- > +-- > replicateM_ 15 $ send pid () +-- > +-- > Nothing <- receiveChanTimeout (asTimeout $ seconds 5) rp +-- > +-- > send pid On +-- > +-- > replicateM_ 15 $ receiveChan rp +-- > +-- > send pid "hello" -- triggers `nextEvent ()` +-- > +-- > res <- receiveChanTimeout (asTimeout $ seconds 5) rp :: Process (Maybe ()) +-- > res `shouldBe` equalTo (Just ()) +-- > +-- > send pid Off +-- > +-- > forM_ ([1..50] :: [Int]) $ \i -> send pid i +-- > send pid "yo" +-- > send pid On +-- > +-- > res' <- receiveChanTimeout (asTimeout $ seconds 20) rp :: Process (Maybe ()) +-- > res' `shouldBe` equalTo (Just ()) +-- > +-- > kill pid "thankyou byebye" +-- +-- Here, the difference between 'postpone' and 'putBack' is that 'postpone' will +-- ensure that the events given to it aren't re-processed until the state id +-- changes. Once the state change is detected, those postponed events are +-- set to be added to the front of the queue (ahead of other events) as soon +-- as the pass completes. +-- +----------------------------------------------------------------------------- +module Control.Distributed.Process.FSM + ( -- * Starting / Running an FSM Process + start + , run + -- * Defining FSM Steps, Actions, and Transitions + , initState + , yield + , event + , pevent + , enter + , resume + , reply + , postpone + , putBack + , nextEvent + , publishEvent + , timeout + , stop + , await + , safeWait + , whenStateIs + , pick + , begin + , join + , reverseJoin + , atState + , always + , allState + , matching + , set + , set_ + , put + -- * DSL-style API (operator sugar) + , (.|) + , (|>) + , (<|) + , (~>) + , (*>) + , (~@) + , (~?) + , (^.) + -- * Types and Utilities + , Event + , FSM + , lift + , liftIO + , stateData + , currentInput + , currentState + , currentMessage + , addTransition + , Step + , Transition + , State + ) where + +import Control.Distributed.Process (wrapMessage) +import Control.Distributed.Process.Extras (ExitReason) +import Control.Distributed.Process.Extras.Time + ( TimeInterval + ) +import Control.Distributed.Process.ManagedProcess + ( processState + , setProcessState + , runAfter + ) +import Control.Distributed.Process.ManagedProcess.Server.Priority (setPriority) +import Control.Distributed.Process.FSM.Internal.Process +import Control.Distributed.Process.FSM.Internal.Types +import Control.Distributed.Process.Serializable (Serializable) +import Prelude hiding ((*>)) + +-- | Fluent way to say 'yield' when you're building an initial state up (e.g. +-- whilst utilising 'begin'). +initState :: forall s d . s -> d -> Step s d +initState = yield + +-- | Given a state @s@ and state data @d@, set these for the current pass and +-- all subsequent passes. +yield :: forall s d . s -> d -> Step s d +yield = Yield + +-- | Creates an @Event m@ for some "Serializable" type @m@. When passed to +-- functions that follow the combinator pattern (such as 'await'), will ensure +-- that only messages of type @m@ are processed by the handling expression. +-- +event :: (Serializable m) => Event m +event = Wait + +-- | A /prioritised/ version of 'event'. The server will prioritise messages +-- matching the "Event" type @m@. +-- +-- See "Control.Distributed.Process.ManagedProcess.Server.Priority" for more +-- details about input prioritisation and prioritised process definitions. +pevent :: (Serializable m) => Int -> Event m +pevent = WaitP . setPriority + +-- | Evaluates to a "Transition" that instructs the process to enter the given +-- state @s@. All expressions following evaluation of 'enter' will see +-- 'currentState' containing the updated value, and any future events will be +-- processed in the new state. +-- +-- In addition, should any events/messages have been postponed in a previous +-- state, they will be immediately placed at the head of the queue (in front of +-- received messages) and processed once the current pass has been fully evaluated. +-- +enter :: forall s d . s -> FSM s d (Transition s d) +enter = return . Enter + +-- | Evaluates to a "Transition" that postpones the current event. +-- +-- Postponed events are placed in a temporary queue, where they remain until +-- the current state changes. +-- +postpone :: forall s d . FSM s d (Transition s d) +postpone = return Postpone + +-- | Evaluates to a "Transition" that places the current input event/message at +-- the back of the process mailbox. The message will be processed again in due +-- course, as the mailbox is processed in priority order. +-- +putBack :: forall s d . FSM s d (Transition s d) +putBack = return PutBack + +-- | Evaluates to a "Transition" that places the given "Serializable" message +-- at the head of the queue. Once the current pass is fully evaluated, the input +-- will be the next event to be processed unless it is trumped by another input +-- with a greater priority. +-- +nextEvent :: forall s d m . (Serializable m) => m -> FSM s d (Transition s d) +nextEvent m = return $ Push (wrapMessage m) + +-- | As 'nextEvent', but places the message at the back of the queue by default. +-- +-- Mailbox priority ordering will still take precedence over insertion order. +-- +publishEvent :: forall s d m . (Serializable m) => m -> FSM s d (Transition s d) +publishEvent m = return $ Enqueue (wrapMessage m) + +-- | Evaluates to a "Transition" that resumes evaluating the current step. +resume :: forall s d . FSM s d (Transition s d) +resume = return Remain + +-- | This /step/ will send a reply to a client process if (and only if) the +-- client provided a reply channel (in the form of @SendPort Message@) when +-- sending its event to the process. +-- +-- The expression used to produce the reply message must reside in the "FSM" monad. +-- The reply is /not/ sent immediately upon evaluating 'reply', however if the +-- sender supplied a reply channel, the reply is guaranteed to be sent prior to +-- evaluating the next pass. +-- +-- No attempt is made to ensure the receiving process is still alive or understands +-- the message - the onus is on the author to ensure the client and server +-- portions of the API understand each other with regard to types. +-- +-- No exception handling is applied when evaluating the supplied expression. +reply :: forall s d r . (Serializable r) => FSM s d r -> Step s d +reply = Reply + +-- | Given a "TimeInterval" and a "Serializable" event of type @m@, produces a +-- "Transition" that will ensure the event is re-queued after at least +-- @TimeInterval@ has expired. +-- +-- The same semantics as "System.Timeout" apply here. +-- +timeout :: Serializable m => TimeInterval -> m -> FSM s d (Transition s d) +timeout t m = return $ Eval $ runAfter t m + +-- | Produces a "Transition" that when evaluated, will cause the FSM server +-- process to stop with the supplied "ExitReason". +stop :: ExitReason -> FSM s d (Transition s d) +stop = return . Stop + +-- | Given a function from @d -> d@, apply it to the current state data. +-- +-- This expression functions as a "Transition" and is not applied immediately. +-- To /see/ state data changes in subsequent expressions during a single pass, +-- use 'yield' instead. +set :: forall s d . (d -> d) -> FSM s d (Transition s d) +set f = return $ Eval (processState >>= \s -> setProcessState $ s { stData = (f $ stData s) }) + +set_ :: forall s d . (d -> d) -> FSM s d () +set_ f = set f >>= addTransition + +-- | Set the current state data. +-- +-- This expression functions as a "Transition" and is not applied immediately. +-- To /see/ state data changes in subsequent expressions during a single pass, +-- use 'yield' instead. +put :: forall s d . d -> FSM s d () +put d = addTransition $ Eval $ do + processState >>= \s -> setProcessState $ s { stData = d } + +-- | Synonym for 'pick' +(.|) :: Step s d -> Step s d -> Step s d +(.|) = Alternate +infixr 9 .| + +-- | Pick one of the two "Step"s. Evaluates the LHS first, and proceeds to +-- evaluate the RHS only if the left does not produce a valid result. +pick :: Step s d -> Step s d -> Step s d +pick = Alternate + +-- | Synonym for 'begin' +(^.) :: Step s d -> Step s d -> Step s d +(^.) = Init +infixr 9 ^. + +-- | Provides a means to run a "Step" - the /LHS/ or first argument - only once +-- on initialisation. Subsequent passes will ignore the LHS and run the RHS only. +begin :: Step s d -> Step s d -> Step s d +begin = Init + +-- | Synonym for 'join'. +(|>) :: Step s d -> Step s d -> Step s d +(|>) = Sequence +infixr 9 |> + +-- | Join the first and second "Step" by running them sequentially from left to right. +join :: Step s d -> Step s d -> Step s d +join = Sequence + +-- | Inverse of "(|>)" +(<|) :: Step s d -> Step s d -> Step s d +(<|) = flip Sequence +-- infixl 9 <| + +-- | Join from right to left. +reverseJoin :: Step s d -> Step s d -> Step s d +reverseJoin = flip Sequence + +-- | Synonym for 'await' +(~>) :: forall s d m . (Serializable m) => Event m -> Step s d -> Step s d +(~>) = Await +infixr 9 ~> + +-- | For any event that matches the type @m@ of the first argument, evaluate +-- the "Step" given in the second argument. +await :: forall s d m . (Serializable m) => Event m -> Step s d -> Step s d +await = Await + +-- | Synonym for 'safeWait' +(*>) :: forall s d m . (Serializable m) => Event m -> Step s d -> Step s d +(*>) = SafeWait +infixr 9 *> + +-- | A /safe/ version of 'await'. The FSM will place a @check $ safe@ filter +-- around all messages matching the input type @m@ of the "Event" argument. +-- Should an exit signal interrupt the current pass, the input event will be +-- re-tried if an exit handler can be found for the exit-reason. +-- +-- In all other respects, this API behaves exactly like 'await' +safeWait :: forall s d m . (Serializable m) => Event m -> Step s d -> Step s d +safeWait = SafeWait + +-- | Synonym for 'atState' +(~@) :: forall s d . (Eq s) => s -> FSM s d (Transition s d) -> Step s d +(~@) = Perhaps +infixr 9 ~@ + +-- | Given a state @s@ and an expression that evaluates to a "Transition", +-- proceed with evaluation only if the 'currentState' is equal to @s@. +atState :: forall s d . (Eq s) => s -> FSM s d (Transition s d) -> Step s d +atState = Perhaps + +-- | Fluent way to say @atState s resume@. +whenStateIs :: forall s d . (Eq s) => s -> Step s d +whenStateIs s = s ~@ resume + +-- | Given an expression from a "Serializable" event @m@ to an expression in the +-- "FSM" monad that produces a "Transition", apply the expression to the current +-- input regardless of what our current state is set to. +allState :: forall s d m . (Serializable m) => (m -> FSM s d (Transition s d)) -> Step s d +allState = Always + +-- | Synonym for 'allState'. +always :: forall s d m . (Serializable m) => (m -> FSM s d (Transition s d)) -> Step s d +always = Always + +-- | Synonym for 'matching'. +(~?) :: forall s d m . (Serializable m) => (m -> Bool) -> (m -> FSM s d (Transition s d)) -> Step s d +(~?) = Matching + +-- | Given an expression from a "Serializable" input event @m@ to @Bool@, if the +-- expression evaluates to @True@ for the current input, pass the input on to the +-- expression given as the second argument. +matching :: forall s d m . (Serializable m) => (m -> Bool) -> (m -> FSM s d (Transition s d)) -> Step s d +matching = Matching diff --git a/src/Control/Distributed/Process/FSM/Client.hs b/src/Control/Distributed/Process/FSM/Client.hs new file mode 100644 --- /dev/null +++ b/src/Control/Distributed/Process/FSM/Client.hs @@ -0,0 +1,101 @@ +----------------------------------------------------------------------------- +-- | +-- Module : Control.Distributed.Process.FSM.Client +-- Copyright : (c) Tim Watson 2017 +-- License : BSD3 (see the file LICENSE) +-- +-- Maintainer : Tim Watson +-- Stability : experimental +-- Portability : non-portable (requires concurrency) +-- +-- The Client Portion of the /FSM/ API. +----------------------------------------------------------------------------- +module Control.Distributed.Process.FSM.Client + ( call + , callTimeout + , safeCall + ) where + +import Control.Distributed.Process + ( send + , wrapMessage + , newChan + , unwrapMessage + , receiveWait + , receiveTimeout + , monitor + , unmonitor + , die + , matchChan + , matchIf + , catchesExit + , handleMessageIf + , getSelfPid + , Message + , Process + , SendPort + , ReceivePort + , ProcessId + , ProcessMonitorNotification(..) + ) +import Control.Distributed.Process.Extras (ExitReason(ExitOther)) +import Control.Distributed.Process.Extras.Time (TimeInterval, asTimeout) +import Control.Distributed.Process.FSM.Internal.Types (baseErr) +import Control.Distributed.Process.Serializable (Serializable) +import Control.Monad.Catch (bracket) + +-- | Initiate a 'call' and if an exit signal arrives, return it as +-- @Left reason@, otherwise evaluate to @Right result@. +safeCall :: (Serializable m, Serializable r) + => ProcessId + -> m + -> Process (Either ExitReason r) +safeCall pid msg = do + us <- getSelfPid + (call pid msg >>= return . Right) + `catchesExit` [(\sid rsn -> handleMessageIf rsn (weFailed sid us) + (return . Left))] + where + weFailed a b (ExitOther _) = a == b + weFailed _ _ _ = False + +-- | As 'call' but times out if the response does not arrive without the +-- specified "TimeInterval". If the call times out, the caller's mailbox +-- is not affected (i.e. no message will arrive at a later time). +callTimeout :: (Serializable m, Serializable r) + => ProcessId + -> m + -> TimeInterval + -> Process (Maybe r) +callTimeout pid msg ti = bracket (monitor pid) unmonitor $ \mRef -> do + (sp, rp) <- newChan :: Process (SendPort Message, ReceivePort Message) + send pid (wrapMessage msg, sp) + msg' <- receiveTimeout (asTimeout ti) + [ matchChan rp return + , matchIf (\(ProcessMonitorNotification ref _ _) -> ref == mRef) + (\_ -> die $ ExitOther "ServerUnreachable") + ] :: Process (Maybe Message) + case msg' of + Nothing -> return Nothing + Just m -> do mR <- unwrapMessage m + case mR of + Just r -> return $ Just r + _ -> die $ ExitOther $ baseErr ++ ".Client:InvalidResponseType" + +-- | Make a synchronous /call/ to the FSM process at "ProcessId". If a +-- "Step" exists that upon receiving an event of type @m@ will eventually +-- reply to the caller, the reply will be the result of evaluating this +-- function. If not, or if the types do not match up, this function will +-- block indefinitely. +call :: (Serializable m, Serializable r) => ProcessId -> m -> Process r +call pid msg = bracket (monitor pid) unmonitor $ \mRef -> do + (sp, rp) <- newChan :: Process (SendPort Message, ReceivePort Message) + send pid (wrapMessage msg, sp) + msg' <- receiveWait [ matchChan rp return + , matchIf (\(ProcessMonitorNotification ref _ _) -> ref == mRef) + (\(ProcessMonitorNotification _ _ r) -> die $ ExitOther (show r)) + ] :: Process Message + mR <- unwrapMessage msg' + case mR of + Just r -> return r + _ -> die $ ExitOther $ baseErr ++ ".Client:InvalidResponseType" diff --git a/src/Control/Distributed/Process/FSM/Internal/Process.hs b/src/Control/Distributed/Process/FSM/Internal/Process.hs new file mode 100644 --- /dev/null +++ b/src/Control/Distributed/Process/FSM/Internal/Process.hs @@ -0,0 +1,168 @@ +{-# LANGUAGE ExistentialQuantification #-} +{-# LANGUAGE ScopedTypeVariables #-} +{-# LANGUAGE PatternGuards #-} +{-# LANGUAGE RecordWildCards #-} +----------------------------------------------------------------------------- +-- | +-- Module : Control.Distributed.Process.FSM.Internal.Process +-- Copyright : (c) Tim Watson 2017 +-- License : BSD3 (see the file LICENSE) +-- +-- Maintainer : Tim Watson +-- Stability : experimental +-- Portability : non-portable (requires concurrency) +-- +-- The /Managed Process/ implementation of an FSM process. +-- +-- See "Control.Distributed.Process.ManagedProcess". +----------------------------------------------------------------------------- +module Control.Distributed.Process.FSM.Internal.Process + ( start + , run + ) where + +import Control.Distributed.Process + ( Process + , ProcessId + , SendPort + , sendChan + , spawnLocal + , handleMessage + , wrapMessage + ) +import qualified Control.Distributed.Process as P + ( Message + ) +import Control.Distributed.Process.Extras (ExitReason) +import Control.Distributed.Process.Extras.Time (Delay(Infinity)) +import Control.Distributed.Process.FSM.Internal.Types hiding (liftIO) +import Control.Distributed.Process.ManagedProcess + ( ProcessDefinition(..) + , PrioritisedProcessDefinition(filters) + , Action + , ProcessAction + , InitHandler + , InitResult(..) + , DispatchFilter + , ExitState(..) + , defaultProcess + , prioritised + ) +import qualified Control.Distributed.Process.ManagedProcess as MP (pserve) +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 Control.Monad (void) +import Data.Maybe (isJust) +import qualified Data.Sequence as Q (empty) + +-- | Start an FSM process +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 an FSM process. NB: this is a /managed process listen-loop/ +-- and will not evaluate to its result until the server process stops. +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, Eq s) => InitHandler (s, d, Step s d) (State s d) +fsmInit (st, sd, prog) = + 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 = + (prioritised + defaultProcess + { + infoHandlers = [ handleInfo handleRpcRawInputs + , handleRaw handleAllRawInputs + ] + , exitHandlers = [ ExitSignalDispatcher (\s _ m -> handleExitReason s m) + ] + , shutdownHandler = handleShutdown + } (walkPFSM prog [])) { filters = (walkFSM prog []) } + +-- 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 + | 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 + +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) +handleRpcRawInputs st@State{..} (msg, port) = + handleInput msg $ st { stReply = (sendChan port), stTrans = Q.empty, stInput = Just msg } + +handleAllRawInputs :: forall s d. (Show s) => State s d + -> P.Message + -> Action (State s d) +handleAllRawInputs st@State{..} msg = + handleInput msg $ st { stReply = noOp, stTrans = Q.empty, stInput = Just msg } + +handleExitReason :: forall s d. (Show s) => State s d + -> P.Message + -> Process (Maybe (ProcessAction (State s d))) +handleExitReason st@State{..} msg = + let st' = st { stReply = noOp, stTrans = Q.empty, stInput = Just msg } + in tryHandleInput st' msg + +handleShutdown :: forall s d . ExitState (State s d) -> ExitReason -> Process () +handleShutdown es er + | (CleanShutdown s) <- es = shutdownAux s False + | (LastKnown s) <- es = shutdownAux s True + where + shutdownAux st@State{..} ef = + void $ tryHandleInput st (wrapMessage $ Stopping er ef) + +noOp :: P.Message -> Process () +noOp = const $ return () + +handleInput :: forall s d . (Show s) + => P.Message + -> State s d + -> Action (State s d) +handleInput msg st = do + res <- tryHandleInput st msg + case res of + Just act -> return act + Nothing -> continue st + +tryHandleInput :: forall s d. (Show s) => State s d + -> P.Message + -> Process (Maybe (ProcessAction (State s d))) +tryHandleInput st@State{..} msg = do + res <- apply st msg stProg + case res of + Just res' -> applyTransitions res' [] >>= return . Just + Nothing -> return Nothing diff --git a/src/Control/Distributed/Process/FSM/Internal/Types.hs b/src/Control/Distributed/Process/FSM/Internal/Types.hs new file mode 100644 --- /dev/null +++ b/src/Control/Distributed/Process/FSM/Internal/Types.hs @@ -0,0 +1,368 @@ +{-# LANGUAGE ExistentialQuantification #-} +{-# LANGUAGE ScopedTypeVariables #-} +{-# LANGUAGE PatternGuards #-} +{-# LANGUAGE RecordWildCards #-} +{-# LANGUAGE DeriveDataTypeable #-} +{-# LANGUAGE DeriveGeneric #-} +{-# LANGUAGE GeneralizedNewtypeDeriving #-} +{-# LANGUAGE MultiParamTypeClasses #-} +{-# LANGUAGE GADTs #-} +{-# LANGUAGE RankNTypes #-} +----------------------------------------------------------------------------- +-- | +-- Module : Control.Distributed.Process.FSM.Internal.Types +-- Copyright : (c) Tim Watson 2017 +-- License : BSD3 (see the file LICENSE) +-- +-- Maintainer : Tim Watson +-- Stability : experimental +-- Portability : non-portable (requires concurrency) +-- +-- Types and common functionality. +----------------------------------------------------------------------------- +module Control.Distributed.Process.FSM.Internal.Types + ( apply + , applyTransitions + , State(..) + , Transition(..) + , Event(..) + , Stopping(..) + , resolveEvent + , Step(..) + , FSM(..) + , runFSM + , lift + , liftIO + , currentState + , currentInput + , currentMessage + , stateData + , addTransition + , baseErr + , decodeToEvent + ) where + +import Control.Distributed.Process + ( Process + , unwrapMessage + , handleMessage + , handleMessageIf + , wrapMessage + , die + ) +import qualified Control.Distributed.Process as P + ( liftIO + , Message + ) +import Control.Distributed.Process.Extras (ExitReason(..)) +import Control.Distributed.Process.ManagedProcess + ( Action + , GenProcess + , continue + , stopWith + , setProcessState + , processState + ) +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 + ) +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) +import qualified Control.Monad.State.Strict as ST + ( MonadState + , StateT + , get + , modify + , lift + , runStateT + ) +import Data.Binary +import Data.Maybe (fromJust, isJust) +import Data.Sequence + ( Seq + , ViewR(..) + , (<|) + , (|>) + , viewr + ) +import qualified Data.Sequence as Q (null) +import Data.Typeable (Typeable, typeOf) +import Data.Tuple (swap, uncurry) +import GHC.Generics + +-- | The internal state of an FSM process. +data State s d = (Show s, Eq s) => + State { stName :: s + , stData :: d + , stProg :: Step s d -- original 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 + show State{..} = "State{stName=" ++ (show stName) + ++ ", stTrans=" ++ (show stTrans) ++ "}" + +-- | Represents a transition from one world state to another. Transitions can +-- be used to alter the process state, state data, to modify and/or interact with +-- the process mailbox, and to postpone processing of messages until state changes +-- take place. +-- +-- The fundmental state transactions are @Remain@, @Enter newState@, and +-- @Stop exitReason@. +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" + +-- | Represents an event arriving, parameterised by the type @m@ of the event. +-- Used in a combinatorial style to wire FSM steps, actions and transitions to +-- specific types of input event. +data Event m where + Wait :: (Serializable m) => Event m + WaitP :: (Serializable m) => Priority () -> Event m + Event :: (Serializable m) => m -> Event m + +-- | Event type wrapper passed to the FSM whenever we're shutting down. +data Stopping = Stopping { reason :: ExitReason -- ^ The "ExitReason" + , errored :: Bool -- ^ Was the shutdown triggered by an error + } deriving (Typeable, Generic, Show) +instance Binary Stopping where + +-- | Resolve an event into a priority setting, for insertion into a priority queue. +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@(WaitP _) = show $ "WaitP::" ++ (show $ typeOf ev) + show ev = show $ typeOf ev + +-- | Represents a step in a FSM definition +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 + Matching :: (Serializable m) => (m -> Bool) -> (m -> FSM s d (Transition s d)) -> Step s d + Sequence :: Step s d -> Step s d -> Step s d + Alternate :: Step s d -> Step s d -> Step s d + Reply :: (Serializable r) => FSM s d r -> Step s d + +instance forall s d . (Show s) => Show (Step s d) where + show st + | 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 _ _" + | Sequence a b <- st = "Sequence [" ++ (show a) ++ " |> " ++ (show b) ++ "]" + | Alternate a b <- st = "Alternate [" ++ (show a) ++ " .| " ++ (show b) ++ "]" + | Reply _ <- st = "Reply" + +-- | State monad transformer. +newtype FSM s d o = FSM { + unFSM :: ST.StateT (State s d) Process o + } + deriving ( Functor + , Monad + , ST.MonadState (State s d) + , MonadIO + , MonadFix + , Typeable + , Applicative + ) + +-- | Run an action in the @FSM@ monad. +runFSM :: State s d -> FSM s d o -> Process (o, State s d) +runFSM state proc = ST.runStateT (unFSM proc) state + +-- | Lift an action in the @Process@ monad to @FSM@. +lift :: Process a -> FSM s d a +lift p = FSM $ ST.lift p + +-- | Lift an IO action directly into @FSM@, @liftIO = lift . Process.LiftIO@. +liftIO :: IO a -> FSM s d a +liftIO = lift . P.liftIO + +-- | Fetch the state for the current pass. +currentState :: FSM s d s +currentState = ST.get >>= return . stName + +-- | Fetch the state data for the current pass. +stateData :: FSM s d d +stateData = ST.get >>= return . stData + +-- | Fetch the message that initiated the current pass. +currentMessage :: forall s d . FSM s d P.Message +currentMessage = ST.get >>= return . fromJust . stInput + +-- | Retrieve the 'currentMessage' and attempt to decode it to type @m@ +currentInput :: forall s d m . (Serializable m) => FSM s d (Maybe m) +currentInput = currentMessage >>= \m -> lift (unwrapMessage m :: Process (Maybe m)) + +-- | Add a "Transition" to be evaluated once the current pass completes. +addTransition :: Transition s d -> FSM s d () +addTransition t = ST.modify (\s -> fromJust $ enqueue s (Just t) ) + +{-# INLINE seqEnqueue #-} +seqEnqueue :: Seq a -> a -> Seq a +seqEnqueue s a = a <| s + +{-# INLINE seqPush #-} +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 getR #-} +getR :: Seq a -> Maybe (ViewR a) +getR s = + case (viewr s) of + EmptyR -> Nothing + a -> Just a + +enqueue :: State s d -> Maybe (Transition s d) -> Maybe (State s d) +enqueue st@State{..} trans + | isJust trans = Just $ st { stTrans = seqPush stTrans (fromJust trans) } + | otherwise = Nothing + +apply :: (Show s) => State s d -> P.Message -> Step s d -> Process (Maybe (State s d)) +apply st msg step + | Init is ns <- step = do + -- ensure we only `init` successfully once + -- P.liftIO $ putStrLn "Init _ _" + st' <- apply st msg is + case st' of + Just s -> apply (s { stProg = 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) + if isJust (ev) then apply st msg act' + else {- (P.liftIO $ putStrLn $ "Cannot decode " ++ (show (evt, msg))) >> -} + return Nothing + | Always fsm <- step = do + -- P.liftIO $ putStrLn "Always..." + runFSM st (handleMessage msg fsm) >>= mstash + | Perhaps eqn act' <- step = do + -- P.liftIO $ putStrLn $ "Perhaps " ++ (show eqn) ++ " in " ++ (show $ stName st) + if eqn == (stName st) then runFSM st act' >>= stash + else return Nothing + | Matching chk fsm <- step = do + -- P.liftIO $ putStrLn "Matching..." + 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) + 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 + else apply st msg al2 + | Reply rply <- step = do + let ev = Eval $ do fSt <- processState + s' <- MP.lift $ do (r, s) <- runFSM fSt rply + (stReply s) $ wrapMessage r + return s + setProcessState s' + -- (_, st') <- runFSM st (addTransition ev) + return $ enqueue st (Just ev) + | otherwise = error $ baseErr ++ ".Internal.Types.apply:InvalidStep" + where + mstash = return . uncurry enqueue . swap + stash (o, s) = return $ enqueue s (Just o) + +applyTransitions :: forall s d. (Show s) + => State s d + -> [GenProcess (State s d) ()] + -> Action (State s d) +applyTransitions st@State{..} evals + | Q.null stTrans, [] <- evals = continue $ st + | Q.null stTrans = act $ do setProcessState st + -- MP.liftIO $ putStrLn $ "ProcessState: " ++ (show stName) + mapM_ id evals + | (tr, st2) <- next + , 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 + , 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 + | (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 $ seqPop stTrans + in (t, st { stTrans = q }) + +-- | Base module name for error messages. +baseErr :: String +baseErr = "Control.Distributed.Process.FSM" + +-- | Given an "Event" for any "Serializable" type @m@ and a raw message, decode +-- the message and map it to either @Event m@ if the types are aligned, otherwise +-- @Nothing@. +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.... diff --git a/tests/TestFSM.hs b/tests/TestFSM.hs new file mode 100644 --- /dev/null +++ b/tests/TestFSM.hs @@ -0,0 +1,297 @@ +{-# LANGUAGE ExistentialQuantification #-} +{-# LANGUAGE ScopedTypeVariables #-} +{-# LANGUAGE DeriveDataTypeable #-} +{-# LANGUAGE DeriveGeneric #-} + +module Main where + +import Control.Distributed.Process hiding (call, send, sendChan) +import Control.Distributed.Process.UnsafePrimitives (send, sendChan) +import Control.Distributed.Process.Node +import Control.Distributed.Process.Extras + ( ExitReason(..) + , isProcessAlive + ) +import qualified Control.Distributed.Process.Extras (__remoteTable) +import Control.Distributed.Process.Extras.Time hiding (timeout) +import Control.Distributed.Process.Extras.Timer +import Control.Distributed.Process.FSM hiding (State, liftIO) +import Control.Distributed.Process.FSM.Client (call, callTimeout) +import Control.Distributed.Process.SysTest.Utils +import Control.Monad (replicateM_, forM_) +import Control.Rematch (equalTo) + +#if ! MIN_VERSION_base(4,6,0) +import Prelude hiding (catch, drop) +#else +import Prelude hiding (drop, (*>)) +#endif + +import Test.Framework as TF (defaultMain, testGroup, Test) +import Test.Framework.Providers.HUnit + +import Network.Transport.TCP +import qualified Network.Transport as NT + +-- import Control.Distributed.Process.Serializable (Serializable) +-- import Control.Monad (void) +import Data.Binary (Binary) +import Data.Maybe (fromJust) +import Data.Typeable (Typeable) +import GHC.Generics + +data State = On | Off deriving (Eq, Show, Typeable, Generic) +instance Binary State where + +data Reset = Reset deriving (Eq, Show, Typeable, Generic) +instance Binary Reset where + +data Check = Check deriving (Eq, Show, Typeable, Generic) +instance Binary Check where + +type StateData = Integer +type ButtonPush = () +type Stop = ExitReason + +initCount :: StateData +initCount = 0 + +startState :: Step State Integer +startState = initState Off initCount + +waitForDown :: MonitorRef -> Process DiedReason +waitForDown ref = + receiveWait [ matchIf (\(ProcessMonitorNotification ref' _ _) -> ref == ref') + (\(ProcessMonitorNotification _ _ dr) -> return dr) ] + +switchFsm :: Step State StateData +switchFsm = startState + ^. ((event :: Event ButtonPush) + ~> ( (On ~@ (set_ (+1) >> enter Off)) -- on => off => on is possible with |> here... + .| (Off ~@ (set_ (+1) >> enter On)) + ) |> (reply currentState)) + .| ((event :: Event Stop) + ~> ( ((== ExitNormal) ~? (\_ -> timeout (seconds 3) Reset)) + {- let's verify that we can't override a normal shutdown sequence... -} + .| ((== ExitShutdown) ~? (\_ -> timeout (seconds 3) Reset)) + .| ((const True) ~? stop) + )) + .| ((event :: Event Check) ~> reply stateData) + .| (event :: Event Reset) + ~> (allState $ \Reset -> put initCount >> enter Off) + +pushButton :: ProcessId -> Process State +pushButton pid = call pid (() :: ButtonPush) + +check :: ProcessId -> Process StateData +check pid = call pid Check + +switchFsmAlt :: Step State StateData +switchFsmAlt = + begin startState $ + pick (await (event :: Event ButtonPush) ((pick (atState On (set_ (+1) >> enter Off)) + (atState Off (set_ (+1) >> enter On))) `join` (reply currentState))) + (pick (await (event :: Event Stop) (pick (matching (== ExitNormal) (\_ -> timeout (seconds 3) Reset)) + (matching (const True) stop))) + (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) ~? (\_ -> 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)) + .| ( (whenStateIs Off) + |> ( ((event :: Event ()) + ~> (allState $ \s -> (lift $ sendChan off s) >> resume)) + .| (((event :: Event String) ~> (always $ \(_ :: String) -> resume)) + |> (reply (currentInput >>= return . fromJust :: FSM State () String))) + ) + ) + .| ( (On ~@ resume) -- equivalent to `whenStateIs On` + |> ((event :: Event ()) + ~> (allState $ \s -> (lift $ sendChan on s) >> resume)) + ) + +genFSM :: SendPort () -> Step State () +genFSM sp = initState Off () + ^. ( (whenStateIs Off) + |> ((event :: Event ()) ~> (always $ \() -> postpone)) + ) + .| ( (((pevent 100) :: Event State) ~> (always $ \state -> enter state)) + .| ((event :: Event ()) ~> (always $ \() -> (lift $ sendChan sp ()) >> resume)) + ) + .| ( (event :: Event String) + ~> ( (Off ~@ putBack) + .| (On ~@ (nextEvent ())) + ) + ) + +republicationOfEvents :: Process () +republicationOfEvents = do + (sp, rp) <- newChan + + pid <- start Off () $ genFSM sp + + replicateM_ 15 $ send pid () + + Nothing <- receiveChanTimeout (asTimeout $ seconds 5) rp + + send pid On + + replicateM_ 15 $ receiveChan rp + + send pid "hello" -- triggers `nextEvent ()` + + res <- receiveChanTimeout (asTimeout $ seconds 5) rp :: Process (Maybe ()) + res `shouldBe` equalTo (Just ()) + + send pid Off + + forM_ ([1..50] :: [Int]) $ \i -> send pid i + send pid "yo" + send pid On + + res' <- receiveChanTimeout (asTimeout $ seconds 20) rp :: Process (Maybe ()) + res' `shouldBe` equalTo (Just ()) + + kill pid "thankyou byebye" + +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 + + resp <- callTimeout pid "hello there" (seconds 3):: Process (Maybe String) + resp `shouldBe` equalTo (Nothing :: Maybe String) + + send pid Off + send pid () + Nothing <- receiveChanTimeout (asTimeout $ seconds 3) rpOn + () <- receiveChan rpOff + + res <- call pid "hello" :: Process String + res `shouldBe` equalTo "hello" + + 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" + monitor pid >>= waitForDown + alive' <- isProcessAlive pid + alive' `shouldBe` equalTo False + +notSoQuirkyDefinitions :: Process () +notSoQuirkyDefinitions = do + start Off initCount switchFsmAlt >>= walkingAnFsmTree + +quirkyOperators :: Process () +quirkyOperators = do + start Off initCount switchFsm >>= walkingAnFsmTree + +walkingAnFsmTree :: ProcessId -> Process () +walkingAnFsmTree pid = do + mSt <- pushButton pid + mSt `shouldBe` equalTo On + + mSt' <- pushButton pid + mSt' `shouldBe` equalTo Off + + mCk <- check pid + mCk `shouldBe` equalTo (2 :: StateData) + + -- verify that the process implementation turns exit signals into handlers... + exit pid ExitNormal + sleep $ seconds 6 + alive <- isProcessAlive pid + alive `shouldBe` equalTo True + + mCk2 <- check pid + mCk2 `shouldBe` equalTo (0 :: StateData) + + mrst' <- pushButton pid + mrst' `shouldBe` equalTo On + + exit pid ExitShutdown + monitor pid >>= waitForDown + alive' <- isProcessAlive pid + alive' `shouldBe` equalTo False + +myRemoteTable :: RemoteTable +myRemoteTable = + Control.Distributed.Process.Extras.__remoteTable $ initRemoteTable + +tests :: NT.Transport -> IO [Test] +tests transport = do + {- verboseCheckWithResult stdArgs -} + localNode <- newLocalNode transport myRemoteTable + return [ + testGroup "Language/DSL" + [ + testCase "Traversing an FSM definition (operators)" + (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) + , testCase "Traversing an FSM definition (event re-publication)" + (runProcess localNode republicationOfEvents) + ] + ] + +main :: IO () +main = testMain $ tests + +-- | Given a @builder@ function, make and run a test suite on a single transport +testMain :: (NT.Transport -> IO [Test]) -> IO () +testMain builder = do + Right (transport, _) <- createTransportExposeInternals + "127.0.0.1" "10501" defaultTCPParameters + testData <- builder transport + defaultMain testData