Happy eyeballs algorithm using ZIO

While researching structured concurrency for the article on Project Loom & Fibers, one of the most valuable resource were blogs (1, 2) by Nathaniel J. Smith, covering the design of the Python library Trio.

In one of his talks Nathaniel goes through the implementation of the “Happy Eyeballs” algorithm, which is a great example on how structured concurrency and high-level concurrency libraries help in creating understandable, concurrent code.

In the talk, Nathaniel mentions that the Trio implementation is not only much shorter than twisted one (one of the most popular python concurrency libraries), but also correct, as it fixes a bug present in the other implementation.

Living in the Scala world (and prompted by Kasper), a natural question arises: what would this look like in Scala? After all, Scala is the language to tame concurrency problems! :)

Spoiler: the implementation is even shorter than in Python. Plus, it uses no mutable state and is type safe!

What is Happy Eyeballs?

When doing a DNS lookup, a hostname can resolve to multiple IP addresses. We might have a couple of servers for load-balancing, fail-over, or both IPv4 and IPv6 addresses.

Which address to connect to? We could try connecting one-by-one, but if the first ones don’t work, that would cause a significant delay for the user. We could connect to all at once and pick the fastest, but that puts unnecessary load on the servers.

We need to hit a middle ground, and that’s what Happy Eyeballs, formally specified in RFC8305 does (your browser is probably using this as well!).

The algorithm is simple: we try to connect to the first address. If we don’t get a response after 250ms, we concurrently start connecting to the second one (while still waiting for the first). If after 500ms we still don’t have a connected socket, we start connecting to the third one, etc.

Once any of the sockets connect, we interrupt all other processes that we started (as they are no longer necessary) and return the one that has succeeded.

As a result, we’ll get a connected socket within reasonable time (or failure if none of the addresses work), without creating unnecessary traffic.

Scala implementation using ZIO

To implement this algorithm in Scala, we’ll use the ZIO concurrency library. ZIO exposes a set of operations, which allow describing concurrently running processes. These descriptions are immutable data structures; smaller descriptions can be combined in various ways to create more complex ones.

Importantly, these process descriptions are lazy: they only get evaluated on demand, typically when the whole program is fully described.

We’ll divide our work into two parts. First, we’ll create a generic implementation of the Happy Eyeballs algorithm, which describes how to start evaluating given tasks, until one of them completes successfully.

Second, we’ll apply the above implementation to a list of tasks, where each one describes opening a socket to one of the addresses resolved for a given domain name.

Generic Happy Eyeballs

We don’t have to limit ourselves to opening sockets; the algorithm is general, and can be applied to any list of tasks, which can be started with a delay. Hence, our algorithm will take two parameters.

First is a list of descriptions of how to run individual tasks. Note that these descriptions are lazily evaluated, so we can safely create them beforehand, without having to worry that the computations already start running.

Second, we’ll need the delay with which subsequent tasks should be started:

def happyEyeballs[R, T](
  tasks: List[ZIO[R, Throwable, T]], 
  delay: Duration): ZIO[R with Clock, Throwable, T] = ...

The ZIO datatype takes three type parameters:

• R: the requirements on the environment, that the computation needs to be run in. Here, the individual tasks can run in any environment (hence we require them to be of type ZIO[R, ..., ...]. However, the result will use the clock to delay subsequent computations. That’s why the resulting computation needs to access the Clock. That’s expressed in the return type: ZIO[R with Clock, ..., ...].
• E: the type of errors that are thrown by the computation. Here we take the easy way out and specify that any kind of exceptions can be thrown. We could make this generic as well, if needed.
• T: the type of the resulting value. Each candidate computation from tasks results in a value of type T (in our case, this will be a connected socket). The result of the whole computation also will be an instance of T, corresponding to the value produced by the candidate task that succeeds.

Let’s start our implementation by looking at the list of tasks. There are three possibilities: the list can be empty, have one element, or more elements. The first two cases are easy. If there’s no elements, we return an error. If there’s only one task, then we have no other option as returning that single task:

def happyEyeballs[R, T](
  tasks: List[ZIO[R, Throwable, T]], 
  delay: Duration): ZIO[R with Clock, Throwable, T] = tasks match {
  case Nil         => IO.fail(new IllegalStateException("no tasks"))
  case task :: Nil => task
  case task :: otherTasks => ...
}

What if there are more tasks? We’ll now need to run two processes in parallel. The first process is task: the head of our tasks list. The second is a process which:

1. waits up to delay, or until the first task fails
2. starts the next task from the list

We don’t just need to run these two in parallel. We need to race them, that is, create a description of a process which runs two computations, and returns the result of the first one that completes. The other should be interrupted (cancelled) so that it doesn’t do unnecessary work. Luckily, race is one of the built-in combinators.

But we’re not done yet. We also need a signal which will allow starting subsequent computations as soon as the first one fails, even before the delay passes. To do that, we can use a single-element queue, which will initially be empty and get filled when the task fails.

Hence, we create the queue (creating the queue is also wrapped with a ZIO process description, and we need to use flatMap to combine two process descriptions):

Queue.bounded[Unit](1).flatMap { taskFailed =>
   ...
}

Next, we create a description of a process, which waits for the completion of the first of two other ones: either sleeping for the given delay, or taking an element from our signalling queue. The .take method on the queue returns a process description, which is suspended until an element is available:

val sleepOrFailed = ZIO.sleep(delay).race(taskFailed.take)

We also describe a process, which runs the initial task, and signals the queue on failure, whatever the actual error is:

val taskWithSignalOnFailed = task.onError(_ => taskFailed.offer(()))

Finally, we once again use race to describe a process which runs the initial task, and if that fails or the timeout passes, recursively tries subsequent tasks:

taskWithSignalOnFailed.race(
  sleepOrFailed *> happyEyeballs(otherTasks, delay))

The *> operator sequences two process descriptions, running the second after the first completes successfully, and returning the result of the second one.

Putting all of this together, here’s the full implementation of Happy Eyeballs in ZIO:

def happyEyeballs[R, T](
  tasks: List[ZIO[R, Throwable, T]], 
  delay: Duration): ZIO[R with Clock, Throwable, T] = tasks match {
  case Nil         => IO.fail(new IllegalStateException("no tasks"))
  case task :: Nil => task
  case task :: otherTasks => 
    Queue.bounded[Unit](1).flatMap { taskFailed =>
      val taskWithSignalOnFailed = 
        task.onError(_ => taskFailed.offer(()))
      val sleepOrFailed = ZIO.sleep(delay).race(taskFailed.take)

      taskWithSignalOnFailed.race(
        sleepOrFailed *> happyEyeballs(otherTasks, delay))
  }
}

Equally important as to what the code contains, is what is missing: note that at no point we have explicitly used the concepts of forking, fibers, threads or launching in the background. We’ve only used higher-level operations, which make sure that the control flow follows the structure of the code.

Applying to sockets

Now that we have a generic implementation of the algorithm, we can apply it to our socket problem. To do that, we’ll need to first resolve the domain name to a list of addresses, and then connect to one of them.

Since we are in Java, we have to use Java APIs. While ZIO is non-blocking, and is based on asynchronous operations, the networking APIs for resolving names and connecting sockets are thread-blocking (at least until Project Loom makes them fiber-suspending!).

That’s why we need to isolate them and run on a dedicated thread pool. Additionally, we need to be able to interrupt these tasks using the brutal, but effective method of Thread.interrupt (interrupting asynchronous ZIO processes is done by the ZIO scheduler, and happens whenever the proccess encounters a yield point: flatMap, a sleep or an asynchronous call).

Luckily again, ZIO provides a way to lift a blocking effect into a ZIO process description, using the effectBlocking method. For example:

effectBlocking(InetAddress.getAllByName("debian.org").toList)

is a value of type ZIO[Blocking, Throwable, List[InetAddress], that is, requires the Blocking thread pool to be provided in the environment, when evaluating the effect; throws arbitrary exceptions; and returns a list of resolved addresses for the given domain name.

Once we have that list, we create a list of process descriptions, where each one opens a socket on port 443:

.map { addresses =>
  addresses.map(address => new Socket(address, 443)))
}

We now get a value of type:

ZIO[Blocking, Throwable, List[ZIO[Blocking, Throwable, Socket]]

The nested ZIO types might look scary at first, but they describe exactly what the value is: a description of a process, which computes a list of process descriptions, each of which can produce a connected socket.

Update 6/2/2020: Leo Noel spotted a bug in the original implementation; if two sockets are established at the same time (before interruption is handled), both would be left open, but only one returned. Hence, we need to ensure that if multiple sockets have been opened, we close any of the additional ones.

To make sure that we keep only one socket open, and close other sockets that have been successfully opened, we need to modify the tasks which are passed to our happyEyeballs implementation.

The tasks will additionally enqueue any successfully computed values using the .onExit method:

def releasableHappyEyeballs[R, T](
    tasks: List[ZIO[R, Throwable, T]],
    delay: Duration,
    releaseExtra: T => ZIO[R, Nothing, Unit]
): ZIO[R with Clock, Throwable, T] =
  for {
    successful <- Queue.bounded[T](tasks.size)
    enqueueingTasks = tasks.map { task =>
      task.onExit {
        case Success(value) => successful.offer(value)
        case Failure(_)     => ZIO.unit
      }
    }
    ...

Then, after running the algorithm, we’ll release (call socket.close()) all values but one:

    ...
    _ <- happyEyeballs(enqueueingTasks, delay)
    // there has to be at least one, otherwise the above would fail
    first :: other <- successful.takeAll
    _ <- ZIO.foreach(other)(releaseExtra)
  } yield first
}

Finally, we can run our modified algorithm, sequencing the effect of creating the list of socket-connecting tasks, with the effect of connecting one of the sockets:

.flatMap(tasks => releasableHappyEyeballs(tasks, 
  250.milliseconds, closeSocket))

The whole application, with the resulting process description being evaluated on a default thread pool, and with printing the result is:

object TestSocket extends App {
  override def run(args: List[String]): ZIO[ZEnv, Nothing, Int] = {
    effectBlocking(InetAddress.getAllByName("debian.org").toList)
      .map { addresses =>
        addresses.map(a => effectBlocking(new Socket(a, 443)))
      }
      .flatMap(tasks => releasableHappyEyeballs(tasks, 
        250.milliseconds, closeSocket))
      .tap(v => putStrLn(s"Connected: ${v.getInetAddress}"))
      .tapError(error => putStrLn(s"ERROR: $error"))
      .fold(_ => 1, _ => 0)
  }
  def closeSocket(s: Socket): ZIO[Blocking, Nothing, Unit] = 
    effectBlocking(s.close()).catchAll(_ => ZIO.unit)
}

Summing up

Although only about 10 lines of code, our implementation of the Happy Eyeballs algorithm is not only concise, but more importantly correct. That’s possible because we’ve been using high-level concurrency constructs, staying true to assumptions of structured concurrency: that the textual representation of the code should correspond to its control flow; no thread or fibers are leaked.

We’ve leveraged the following capabilities of Scala and the ZIO library:

• lazily-evaluated, immutable descriptions of processes, represented as values: the ZIO data structure
• safe and composable interruption of computations which are no longer required (or should!) complete
• lifting Java’s blocking calls into a description of an interruptible process
• racing or sequencing descriptions of two processes

The code is available on GitHub in the adamw/zio-structured repository, and should be runnable both from the command-line (using the sbt build tool), or after importing into an IDE (such as IntelliJ).

The code contains three classes:

• HappyEyeballs: implementation of the algorithm
• ReleasableHappyEyeballs: a layer on top of HappyEyeballs to release any extra values that have been successfully computed
• TestPrintln: runs the algorithm using processes which output messages to the console, simulate delays (sleep) and fail/succeed
• TestSocket: running the algorithm to connect a socket

Try it out and explore Scala on your own!