Event-driven finite state machine for a distributed trading system

One problem I had when building my distributed trading system is managing states asynchronously from multiple triggers. For example, when the alpha engine says buy, it needs confirmation from the position engine to see if it is safe to enter a new position. A few years ago I would have handled this with a whole bunch of flags to switch states. In fact, that's what I did.

This time I could chain one check after another imperatively, or via callbacks. But the underlying constraints are that these triggers:

1. are resource-intensive to generate,
2. might need to compose many of them,
3. are not sequential and don't have one-to-one dependency, and
4. most importantly, they live in separate programs on different machines.

So instead of writing code, I sat on the problem for a few days. Eventually something clicked: I was looking at an event-driven finite state machine (FSM), a well-worn pattern for tracking state transitions. Intimidating term, but my first implementation was just if-else statements and it still qualified.

Once the problem was an FSM, each of my system's components only needed to push signals to a central module and pull states back from it. No component has to know what to call next, or poll anything to see if the stars are aligned. The entire module came out to less than 200 lines of Clojure and took no more than a few hours to write.

The responsibilities of the FSM module are to:

1. listen to all the signals,
2. figure out all the transitions, and
3. publish the latest states for the rest of the system.

The rest of this post walks through those three jobs.

Handling asynchronous events

I use RabbitMQ as the message transport layer between my system's modules. All I need to do here is associate an appropriate message handler with each triggering input for the FSM. Here's one of the event handlers, using the Clojure RabbitMQ library, Langohr. The rest of this part is just standard RabbitMQ publish/subscribe stuff.

(defn- event-message-handler [ch {:keys [headers delivery-tag redelivery?]} ^bytes payload]
  (let [{:keys [message-type user-id instrument quantity]} (read-payload payload)]
    (when (= :position-event message-type)
      (-> (get-cached-states user-id)       ;; fetch current states for this user
          (update-position-state instrument quantity)  ;; evaluate the next state
          (cache-states user-id))))         ;; store the new states
  (lbc/ack ch delivery-tag))

This is called when a position event is received with information such as user, instrument, and quantity. The handler threads that information through: fetch the current states for that user, evaluate the next state given the input, cache the new states.

State transitions

Below is one of my system's state transition diagrams.

There are 4 states represented by 4 colours, with 4 triggers signalling state transitions. The program is expected to handle up to hundreds of independent states concurrently, with event triggers coming in a couple of times per second.

As I was saying, my first implementation was just a set of if-else methods. For example, an engage trigger would call the engaging method to determine the next state given the implicit input engage and the current state.

(defn engaging
  [current]
  (condp = current
    "white" "yellow"
    "yellow" "white"
    "green" "red"
    "red" "green"))

There were a handful of these boilerplate methods. So after I deployed my system I came back to refactor them. I'd been meaning to give core.logic a try for a while, so this seemed like a good place to start using it.

Before we can ask the logic solver a question, we need to define relations. Here I define a transition relation to specify all the state transition definitions conveniently in one place.

(defrel transition from input to)
(facts transition [[nil :open :green]
                   [nil :close :white]
                   [:white :engage :yellow]
                   [:white :disengage :white]
                   [:white :open :green]
                   [:white :close :white]
                   [:yellow :engage :white]
                   [:yellow :disengage :white]
                   [:yellow :open :green]
                   [:yellow :close :yellow]
                   [:green :engage :red]
                   [:green :disengage :red]
                   [:green :open :green]
                   [:green :close :yellow]
                   [:red :engage :green]
                   [:red :disengage :red]
                   [:red :open :red]
                   [:red :close :white]])

And the event handler methods are just wrappers for a one-liner logic expression asking the question: given the current state, cur-state, and input trigger, input, what state can q take to satisfy this constraint?

(defn next-state
  "Solver for next state"
  [input cur-state]
  (first (run 1 [q] (transition cur-state input q))))

(def colour-clicked (partial next-state :engage))
(def colour-deactivate (partial next-state :disengage))
(defn next-position-colour [cur open?]
  (if open?
    (next-state :open cur)
    (next-state :close cur)))

Not the most illustrative core.logic example, but it does the job. Getting started with core.logic is surprisingly easy. I went through the Primer and tutorial and got this working in one try.

State caching and sharing

With the transitions taken care of, states are cached and served on Redis for the other parts of the system. I use Redis for this because it is fast and easy. Values are stored in edn format instead of something more popular like JSON, to maintain data structure through the wire.

(def pool         (car/make-conn-pool))
(def spec-server1 (car/make-conn-spec))
(defmacro with-car [& body] `(car/with-conn pool spec-server1 ~@body))

(defn get-cached-states
  "Generate edn from database."
  [id]
  (edn/read-string (with-car (car/get (str "states:" id)))))

(defn cache-states [m id]
  (with-car (car/set (str "states:" id) (str m))))

This is my first time using edn in production. All inter-process messages in this trading system are edn formatted. It works seamlessly with Clojure: str to write, clojure.edn/read-string to read. My trade broker interface is written in Java, and it uses edn-java to parse and unparse complex Clojure data structures, such as nested maps with keywords. I find coupling edn with Redis a fantastic choice: it's almost like working with Clojure's native concurrency data structures, such as atom, but external programs can access the data too.

Simple and quick

I haven't done any benchmarks. All I can say is that this setup handles my simplistic use case with barely any load on the server, so I'm happy with it.

I'd handled this with flags before, and a few days of thinking made that whole mess disappear into a couple hundred lines. Seeing through to the underlying problem and solving it with a well-worn pattern, that's the biggest satisfaction here.