Single Flight Actors

Origin

Earlier, I was working on an IOS widget to display the shuttle bus arrival times in NUS. It was later scrapped when I realised that widgets have a limited number of refreshes per day, so it wasn’t possible to provide real time data.

Prior to the widget, I had already built a proxy layer on top of the official NUS bus API. I was even considerate enough to implement a caching layer and a single flight mechanism to not overwhelm the provider should my widget ever go “viral” and get users. Little did I know that it would’ve been abandoned so quickly :(

Gleam

Gleam was made to bring type safety to the BEAM family of programming languages. I’ve written a little Elixir at a previous stint but didn’t really like it’s syntax and lack of types. Also, Gleam looks a lot more like Rust so it’s much more readible.

pstt, you can learn more about BEAM here

Why did I decide to use Gleam for a proxy? Just because I could of course.

Also because I saw alistair’s cool dashboard of the UK rail system built with Gleam, along with OpenAI’s release of symphony which was in Elixir. Yes I am a clear victim of shiny object syndrome.

Single Flight

Third party resources are common in most software systems, but what happens when duplicate requests are made?

You’d most likely be fine if you had little users, or if the API was free. But what your users were in the millions? Some would say caching, but not if all the requests came in at the same time.

Funny to talk about scale for my inexistent bus widget

A single flight mechanism ensures that, for a given request key, only one request is sent at any given time. It is a deduplication system to prevent identical work from being repeated.

Please consult the ascii diagram below:

Without single flight
---------------------
Req A ───┐
Req B ───┼───> expensive_work()
Req C ───┘

Same work runs 3 times

With single flight
------------------
Req A ───┐
         │   acquires lock
         ├──> [ leader ] ───> expensive_work()
Req B ───┤        │
         │        ├─── wait
Req C ───┘        └─── wait

After work completes:
[ leader ] ───> stores result / broadcasts result
                    │
                    ├───> gets result
                    └───> gets result

Work runs once, all callers share the same result 

This can bring many benefits, such as:

Reduced cost
If your API is paid per-use, you wouldn’t be wasting money on duplicate calls.

Resource savings
Even if the API was free, less requests means less connections, less memory and cpu usage.

Bye bye thundering herds
Requests with the same key often come in the same time, your upstream service would be grateful if you didn’t hammer them with duplicate requests every now and then

Actors

Citing the documentation for actors:

An Actor is a process like any other BEAM process and can be used to hold state, execute code, and communicate with other processes by sending and receiving messages.

So actors are processes, that communicate with other actors through messages, while each of them holds their isolated state. Instead of synchronising assess through global state, they inform one another of state updates, mutating their individual state to match the desired global state.

The documentation linked above has an example using actors to simulate a stack which can receive messages from other actors to push and pop elements.

Implementation

We first need to define what kind of messages our actors can act on. In the case of a single flight mechanism, we can either:

1. make a request, or
2. complete a request, and broadcast the result

pub type Message(k, v) {
  Request(key: k, work: fn(k) -> v, caller: Subject(v))
  Done(key: k, result: v)
}

Here we have a sum type for Message, which takes generics k and v representing the function’s input and output.

The Request variant is for when we want to make a call to say, some upstream service. Subject is the actor’s “mailbox”, where it sends and receives messages from other actors. In order to get the response for our request, we need to let the single flight actor know how to get back to the requester.

The Done variant is for when the API call is done, and we want to inform the requesters of the result.

type State(k, v) {
  State(
    self: Subject(Message(k, v)),
    in_flight: Dict(k, List(Subject(v))) 
  )
}

State represents the internal state of an actor. self is it’s mailbox, handling messages of type Message(k, v).

in_flight keeps track of the requests that are have been made and are waiting for a response, it uses Dict(k, List(Subject(v))) which may look like a mouthful at first, but it’s just:

• Dict a dictionary (or, hashmap)
• mapping each key of type k
• to a list (or, array) of subjects with type Subject(v), meaning their mailboxes are expecting a value of type v which is the response value from the upstream call

Now that we’ve defined the messages, and the state of the actor, we can now look at how the actor responses to messages as they show up in their mailboxes.

Behold:

fn handle_message(
  state: State(k, v),
  message: Message(k, v),
) -> actor.Next(State(k, v), Message(k, v)) {
  // When the actor receives a message
  case message {
    // If it's of type Request
    Request(key, work, caller) ->
      // Check if there's a request in flight for the given key
      case dict.get(state.in_flight, key) {
        // Duplicate request found 
        Ok(waiters) -> {
          // Append the caller's subject to the list
          let callers = [caller, ..waiters]
          let in_flight = dict.insert(state.in_flight, key, callers) 
          
          actor.continue(
            State(
              self: state.self,
              in_flight: in_flight
            )
          )
        }

        // It's the first time we're making this request 
        Error(Nil) -> {         
          // Spawn a worker process to make the request
          process.spawn(fn() {          
            let result = work(key)
            let message = Done(key: key, result: result)
            
            // Send the value back when it's done 
            actor.send(state.self, message) 
          })
          
          // Append the caller's subject to the list 
          let in_flight = dict.insert(state.in_flight, key, [caller])
          
          actor.continue(
            State(
              self: state.self,
              in_flight: in_flight
            )
          )
        }
      }


    // If it's of type Done 
    Done(key, result) -> {
      // Retrieve the list of callers we need to broadcast the result to 
      case dict.get(state.in_flight, key) {
        Ok(waiters) ->
          list.each(waiters, fn(waiter) { 
          // Broadcast the result
            process.send(waiter, result) 
          })
          
        // This should be an invalid state,
        // but reachable if the process crashes before 
        // inserting the key into the dictionary
        Error(Nil) -> Nil
      }
      
      // Delete the request key from the dictionary
      let in_flight = dict.delete(state.in_flight, key)       
      
      actor.continue(
        State(
          self: state.self,
          in_flight: in_flight 
        ),
      )
    }
  }
}

If the code was too much too digest, here’s a sequence diagram:

And voila, you have used actors to build a single flight mechanism in Gleam! I have it published here on Github.

Reminder to not try and handroll important mechanisms such as this one when you’re working in a production system, unless you have a really good reason to.