DEVS state machine

00_readme

core.clj - DEVS state machine implementation. This ns provides construction functions as well as the generic evolver. A state machine is simply data about the allowed states, allowed inputs, and actions to trigger on transitions.

core-test.clj - Tests for DEVS

echo.clj - A sample usage. Illustrates construction and application of the state machine.

core-test.clj

(ns devs.core-test
  (:require [devs.core :refer :all]
            [expectations :refer :all])
  (:use clojure.pprint))

;;; in-state? returns a predicate that returns the whole machine on a match, nil otherwise
(expect {:state :initial} ((in-state? :initial) {:state :initial}))
(expect nil               ((in-state? :initial) {:state :anything-else}))
(expect {:state :initial :extra-key 'with-data} ((in-state? :initial) {:state :initial :extra-key 'with-data}))

;;; side-effect returns a function that calls a function. whatever the function returns becomes the new state
(expect 'foo              ((side-effect (fn [_] 'foo)) {:state :anything}))
(expect 'foo              ((side-effect (fn [_ extra] extra) 'foo) {:state :anything}))
(expect {:state :next :extra 'foo} ((side-effect (fn [state]
                                                   (merge state
                                                          {:state :next
                                                           :extra 'foo})))
                                    {:state :anything}))

;;; new-state returns a function that, when applied, will update the state machine's state
(given ((new-state :listening) {:state :initial :extra 'foo :state-alphabet #{:initial :listening}})
       (expect 
        :state :listening
        :extra 'foo))

(expect Exception ((new-state :listening) {:state-alphabet #{:one :two :three}}))


;;; on-event attaches a new transition to the transition map
(expect true (contains? (on-event {:input-alphabet #{:read :write}} :read identity) :transitions))
(expect true (seq? (get-in 
                    (on-event {:input-alphabet #{:read :write}} :read identity)
                    [:transitions :read])))
(expect false (seq? (get-in
                     (on-event {:input-alphabet #{:read :write}} :read identity)
                     [:transitions :write])))

(let [m1 {:input-alphabet #{:evt-a :evt-b :evt-c}
          :state-alphabet #{:waiting-for-a :waiting-for-b :waiting-for-c :waiting-for-godot}
          :state          :waiting-for-a}
      
      m1 (on-event m1 :evt-a
                   (in-state? :waiting-for-a)
                   (new-state :waiting-for-b))
      m1 (on-event m1 :evt-b
                   (in-state? :waiting-for-b)
                   (new-state :waiting-for-c))
      m1 (on-event m1 :evt-c
                   (in-state? :waiting-for-c)
                   (new-state :waiting-for-godot))
      m1 (on-event m1 :godot
                   (new-state :waiting-for-godot))]

  (expect :waiting-for-b (:state (evolve m1 :evt-a)))
  (expect :waiting-for-c (:state (evolve (evolve m1 :evt-a) :evt-b)))
  (expect :waiting-for-godot (:state (evolve (evolve (evolve m1 :evt-a) :evt-b) :evt-c)))
  (expect :waiting-for-godot (:state (evolve m1 :godot)))
  (expect :waiting-for-godot (:state (evolve (evolve (evolve m1 :godot) :godot) :godot)))
  (expect :waiting-for-godot (:state (evolve (assoc m1 :state :waiting-c) :godot)))
  
  )

;;; internal events for automatic state transitions
(let [m1 {:input-alphabet #{:evt-a :evt-b :evt-c}
          :state-alphabet #{:waiting-for-a :waiting-for-b :waiting-for-c :waiting-for-godot}
          :state          :waiting-for-a}

      m1 (on-event m1 :evt-a (in-state? :waiting-for-a) (generate-event :evt-b) (new-state :waiting-for-b))
      m1 (on-event m1 :evt-b (in-state? :waiting-for-b) (new-state :waiting-for-c) (generate-event :evt-c))
      m1 (on-event m1 :evt-c (in-state? :waiting-for-c) (new-state :waiting-for-godot))
      m1 (on-event m1 :godot (new-state :waiting-for-godot))]

  (expect :waiting-for-godot (:state (evolve m1 :evt-a)))
  
  )

(let [m2 {:input-alphabet #{:ping}
          :state-alphabet #{:receiving :sending}
          :state          :receiving}

      m2 (on-event m2 :ping     (in-state? :receiving) (new-state :sending))
      m2 (outputs  m2 :sending  (generate :pong)       (new-state :receiving))

      m2'  (evolve m2 :ping)
      m2'' (evolve m2' :ping)]

  (expect '(:pong) (:output m2'))
  (expect '(:pong :pong) (:output m2''))
  )

(let [m2 {:input-alphabet #{:ping}
          :state-alphabet #{:receiving :sending}
          :state          :receiving
          :condition      false}

      m2 (on-event m2 :ping     (in-state? :receiving)
                   (guard #(:condition %) (new-state :sending)))

      m2'  (evolve m2 :ping)

      m2'' (evolve (assoc m2 :condition true) :ping)]

  (expect :receiving (:state m2'))
  (expect :sending   (:state m2''))
  )

core.clj

(ns devs.core)

;;;;;;;;;; DEVS - Discrete Event System Specification

(defn alternatives
  [state fs]
  (if (empty? fs)
    state
    (if-let [ret ((first fs) state)]
      ret
      (recur state (rest fs)))))

(defn sequential
  [state fs]
  (if (empty? fs)
    state
    (if-let [ret ((first fs) state)]
      (recur ret (rest fs)))))

(defn lift-sequential
  [fs]
  (fn [machine] (sequential machine (seq fs))))

(defn in-state?
  "in-state? returns a predicate. Later, when that predicate is applied to a state machine,
   it ensures that the machine is in the labeled state. If so, the predicate returns the
   entire machine. If not, it returns nil."
  [label]
  (fn [current]
    (if (= (:state current) label)
      current
      nil)))

(defn side-effect
  "side-effect returns a thunk to be applied later. When the thunk is applied to a state machine,
   it calls the original function as (apply f state args). The side effecting function MUST
   return a new value for the state machine if processing is to continue. The side effecting
   function MAY return nil to abort processing."
  [f & args]
  (fn [current]
    (apply f current args)))

(defn guard
  "guard returns a thunk to be applied later. When the thunk is applied,
   it evaluates (apply guard-fn state). If the result is true, it continues
   processing with true-fn. If false and false-fn is provided, then false-fn
   is evaluated."
  ([guard-fn true-fn] (guard guard-fn true-fn (constantly nil)))
  ([guard-fn true-fn false-fn]
     (fn [current]
       ((if (guard-fn current) true-fn false-fn) current))))

(defn new-state
  [next-label]
  (fn [current]
    (when-not (some #{next-label} (:state-alphabet current))
      (throw (ex-info "Unrecognized state symbol" {:state-machine current :next-label next-label})))
    (assoc current :state next-label)))

(defn on-event
  [machine evt & forms]
  (update-in machine [:transitions evt] conj forms))

(defn generate-event
  [next-state]
  (fn [current]
    (update-in current [:internal-events] conj next-state)))

;; TODO - reuse alternative/sequential mechanism for output function
(defn outputs
  "Append an incomplete output function to the state machine's output generator. Example:

   (outputs machine :waiting (generate :read))

   The state machine can generate output after every input, whether the input was external or
   internally initiated. The output is consed onto the list (:output machine), so the history
   of all outputs is available."
  [machine state & forms]
  (update-in machine [:output-function state] conj forms))

;;; TODO - allow _output_ to be a thunk. Call it when needed.
(defn generate
  [output]
  (fn [current]
    (update-in current [:output] conj output)))

(defn- apply-alternatives
  [machine alts]
    (if-not alts machine)
    (if-let [ret (alternatives machine (map lift-sequential alts))]
      ret
      machine))

(defn select-future
  [machine input]
  (apply-alternatives machine (get-in machine [:transitions input])))

(defn write-outputs
  [machine]
  (apply-alternatives machine (get-in machine [:output-function (:state machine)])))

(defn evolve
  [machine input]
  (when-not (some #{input} (:input-alphabet machine))
    (ex-info "Unrecognized input symbol" {:state-machine machine :input input}))
  (let [next-state  (update-in machine [:input-history] conj input)
        next-state  (apply-alternatives next-state (get-in next-state [:transitions input]))
        next-state  (or next-state machine)
        output-fs   (get-in next-state [:output-function (:state next-state)])
        next-state  (apply-alternatives next-state output-fs)]
    (let [deferred (seq (:internal-events next-state))]
      (if-not deferred
        next-state
        (recur (assoc-in next-state [:internal-events] (rest deferred)) (first deferred))))))

echo.clj

(ns planck-server-nio.echo
  (:require [devs.core :refer :all]
            [planck-server-nio.log :refer (error warn info debug)]
            [planck-server-nio.socket :as s]
            [planck-server-nio.selector :as sel]
            [planck-server-nio.threads :as t])
  (:import  (java.nio ByteBuffer)))


;;;;;;;;;; NIO Buffer manipulation
(defn new-buffer [sz] (ByteBuffer/allocateDirect sz))
(defn prepare-for-write [b] (.flip (.duplicate b)))

;;;;;;;;;; Echo Server protocol
(defn read-buffer-from-socket
  [state]
  (let [socket        (:socket state)
        buffer        (new-buffer 8192)
        bytes-in      (.read socket buffer)]
    (debug {:message "Read buffer from socket" :socket socket :bytes-read bytes-in})
    (cond (neg? bytes-in)  ((generate-event :close) state)
          (zero? bytes-in) state
          :else            (update-in state [:buffer-queue] conj (prepare-for-write buffer)))))

(defn write-buffer-to-socket
  [state]
  (debug {:message "Write buffer to socket" :socket (:socket state)})
  (let [socket        (:socket state)
        buffers       (seq (:buffer-queue state))
        buffer        (first buffers)
        bytes-out     (.write socket buffer)]
    (debug {:bytes-written bytes-out})
    (if (.hasRemaining buffer)
      ((generate :write) state)
      (if (empty (rest buffers))
        (dissoc ((generate-event :empty-buffers) state) :buffer-queue)
        (assoc-in state [:buffer-queue] (rest buffers))))))

(defn all-data-drained?
  [state]
  (empty? (:buffer-queue state)))

(defn close-socket
  [state]
  (if-let [socket (:socket state)]
    (.close socket))
  (dissoc state :socket))

(defn echo-server-state-machine
  "Construct a DEVS data structure for the echo server protocol."
  [socket]
  (-> {:input-alphabet         #{:read :write :empty :close :empty-buffers}
       :output-alphabet        #{:read :write}
       :state-alphabet         #{:reading :writing :draining :closed}
       :state                  :reading
       :socket                 socket}
      (on-event :read
                (in-state? :reading)
                (side-effect read-buffer-from-socket)
                (new-state :writing))
      (on-event :read
                (in-state? :writing)
                (side-effect read-buffer-from-socket)
                (new-state :writing))
      (on-event :write
                (in-state? :writing)
                (side-effect write-buffer-to-socket)
                (new-state :writing))
      (on-event :write
                (in-state? :draining)
                (guard all-data-drained?
                       (generate-event :empty-buffers)
                       write-buffer-to-socket)
                (new-state :draining))
      (on-event :empty-buffers
                (in-state? :writing)
                (new-state :reading))
      (on-event :empty-buffers
                (in-state? :draining)
                (side-effect close-socket)
                (new-state :closed))
      (on-event :close
                (in-state? :reading)
                (side-effect close-socket)
                (new-state :closed))
      (on-event :close
                (in-state? :writing)
                (new-state :draining))
      (outputs  :reading  (generate [:read]))
      (outputs  :writing  (generate [:read :write]))
      (outputs  :draining (generate [:write]))
      (outputs  :closed   (generate []))))

;;; TODO - using agents + records is complecting state & dispatch. reconsider.

;;;;;;;;;; Gluing Selectors to DEVS
(defn expect-events
  [selector sel-key state]
  (sel/expect!  selector sel-key (into #{} (first (:output state))))
  state)

(defrecord ClientConnection [state-machine]
  sel/SelectionEvent
  (handle-event [this selector sel-key]
    (let [input-event   (cond 
                         (.isWritable sel-key) :write
                         (.isReadable sel-key) :read)
          next-state    (evolve state-machine input-event)]
      (debug {:next-state next-state})
      (if (= (:state next-state) :closed)
        (.cancel sel-key)
        (->ClientConnection (expect-events selector sel-key next-state))))))

(defrecord Listening []
  sel/SelectionEvent
  (handle-event [this selector sel-key]
    (debug {:message "Listening/handle-event" :selector selector :sel-key sel-key})
    (when (sel/acceptable? sel-key)
      (do
        (let [client-socket (.accept (.channel sel-key))]
          (info {:message "Connected" :client (.getRemoteAddress client-socket)})
          (.configureBlocking client-socket false)
          (sel/register! selector client-socket #{:read} (->ClientConnection (echo-server-state-machine client-socket)))
          (sel/expect! selector sel-key #{:accept}))))
    this))

;;;;;;;;;; Startup
(defn start-echo-server [port]
  (let [socket  (s/server-socket (s/inet-address port))]
    (when-not socket (throw (ex-info "Unable to bind to socket" {:port port})))
    (let [selector (sel/selector)]
      (sel/register! selector socket #{:accept} (->Listening))
      (reify t/Interruptible
        (stop [this]
          (t/stop selector)
          (s/close socket))))))