distributed-process-fsm (empty) → 0.0.1
raw patch · 8 files changed
+1697/−0 lines, 8 filesdep +HUnitdep +QuickCheckdep +ansi-terminalsetup-changed
Dependencies added: HUnit, QuickCheck, ansi-terminal, base, binary, bytestring, containers, data-accessor, deepseq, distributed-process, distributed-process-client-server, distributed-process-extras, distributed-process-fsm, distributed-process-systest, distributed-static, exceptions, fingertree, ghc-prim, hashable, mtl, network, network-transport, network-transport-tcp, rematch, stm, test-framework, test-framework-hunit, test-framework-quickcheck2, time, transformers, unordered-containers
Files
- LICENCE +30/−0
- Setup.lhs +3/−0
- distributed-process-fsm.cabal +85/−0
- src/Control/Distributed/Process/FSM.hs +645/−0
- src/Control/Distributed/Process/FSM/Client.hs +101/−0
- src/Control/Distributed/Process/FSM/Internal/Process.hs +168/−0
- src/Control/Distributed/Process/FSM/Internal/Types.hs +368/−0
- tests/TestFSM.hs +297/−0
+ LICENCE view
@@ -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.
+ Setup.lhs view
@@ -0,0 +1,3 @@+#!/usr/bin/env runhaskell+> import Distribution.Simple+> main = defaultMain
+ distributed-process-fsm.cabal view
@@ -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 <watson.timothy@gmail.com>+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
+ src/Control/Distributed/Process/FSM.hs view
@@ -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 <watson.timothy@gmail.com>+-- 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
+ src/Control/Distributed/Process/FSM/Client.hs view
@@ -0,0 +1,101 @@+-----------------------------------------------------------------------------+-- |+-- Module : Control.Distributed.Process.FSM.Client+-- Copyright : (c) Tim Watson 2017+-- License : BSD3 (see the file LICENSE)+--+-- Maintainer : Tim Watson <watson.timothy@gmail.com>+-- 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"
+ src/Control/Distributed/Process/FSM/Internal/Process.hs view
@@ -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 <watson.timothy@gmail.com>+-- 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
+ src/Control/Distributed/Process/FSM/Internal/Types.hs view
@@ -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 <watson.timothy@gmail.com>+-- 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....
+ tests/TestFSM.hs view
@@ -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