How we built inter-thread messaging from scratch

Inter-thread messaging is a fundamental part of any asynchronous system. It is the component responsible for the transportation of data between threads. Messaging forms the infrastructure, scaffolding multi-threaded applications, and just like real-world transport infrastructure, we want it to be inexpensive, fast, reliable, and clean. For QuestDB, we wrote our own messaging system, and this post is about how it works and how fast it is.

Architecture

Borrowing heavily from world-famous Disruptor our messaging revolves around multiple threads accessing shared circular data structure. We call it RingQueue. Semantically RingQueue provides unbounded, index-based, random access to its elements. It does not coordinate concurrent access nor does it provide guarantees on thread safety. Coordination and thread-safety is a concern of Sequences. Sequences are responsible for providing indices that can be used to access RingQueue concurrently and safely.

To help sequences do their magic they have to be shaped into a graph. We start with syntax to chain sequences together:

a.then(b).then(c).then(d)

The result is a trivial sequence graph:

a -> b -> c -> d

To branch we use helper class FanOut:

a.then(FanOut.to(b).and(c)).then(d)

The result is this sequence graph:

     +--> B -->+
A -->|         |--> D
     +--> C -->+

These two pieces of syntax are flexible enough to create any desired flow. This example shows that FanOut can have chain of sequences and other FanOuts:

a.then(FanOut.to(FanOut.to(b).and(c)).and(d.then(e)).then(f)

It is quite a mouthful but it creates this nice little graph:

        +--> B -->+
    +-> |         |
    |   +--> C -->+
A-->|             |--> F
    |             |
    +-> D -> E -->+

FanOut can also be used as a placeholder in a chain to allow threads to subscribe/unsubscribe on the fly. Dynamic subscription is then simply adding a new sequence to FanOut:

// You can add as many sequences into fan out as you like.
// Sequences can be added either up front or subscribe/unsubscribe on the fly.
FanOut fanOut = new FanOut();

// ordinary producer sequence
Sequence seqProducer = new SPSequence(queue.getCapacity());
// daisy chain producer and fan out and loop back producer
seqProducer.then(fanOut).then(seqProducer);

// meanwhile in another thread ....
...

// Add individual consumer sequences later as needed.
// This is thread safe non-blocking operation that can be performed from any thread.
// It is important to use current producer position as consumer starting point when subscribing on the fly.
Sequence consumer1 = fanOut.addAndGet(new SCSequence(seqProducer.current()));

// do something useful with consumer1 sequence
...

// remove sequence from fanOut to unsubscribe
fanOut.remove(consumer1);

Typical graph must contain single producer sequence and one or more consumer sequences. It will also have to be circular, e.g. to start and end with producer sequence. Graph has to be circular because we use circular underlying data structure, RingQueue. Without loop-back producer would be liable to overwrite queue elements before consumers had a chance to read them. Worse still, queue elements can be written to and read from concurrently. We don't want that to happen, right?

To help create practical sequence graph we implemented 4 types of sequences we can play with. These sequences are better understood as combination of their types and properties. SP - single producer, MP - multiple producer, SC - single consumer and MC - multiple consumer. Multi- sequences allow concurrent access and they guarantee that no two threads can retrieve same index. It is this property adds extra fun dimension to sequence graphs. Consider this graph:

A -> B -> A

or in Java notations:

A.then(B).then(A)

When "B" is an instance of MCSequence() we have a self-balancing worker pool. When "A" is MPSequence(), we have many-to-many pub-sub system. Cool, eh?

Single- sequences are faster but they are not thread-safe. They should be preferred for single-threaded consumer models.

Lets take a look at how threads interact with sequences. This is a typical example of publisher:

// loop until there is work to do
// consumer thread may be able to rely on producer to
// publish "special" message to indicate end of stream.
while (true) {

  // Non-blocking call. Method returns immediately either with zero-based
  // ring queue index or negative long indicating one of following:
  // -1 = queue is empty
  // -2 = there was a contest for queue index and this thread has lost
  long cursor = sequence.next();
  if (cursor < 0) {
    // negative cursor is an error
    // thread has a choice of things to do:
    // - busy spin
    // - yield/park
    // - work on something else
    LockSupport.parkNanos(1);
    continue;
  }

  // write to queue
  try {
    queue.get(cursor).value;
  } finally {
    // releasing cursor promptly is important
    sequence.done(cursor);
  }
}

Sequence.next() return values are:

-1 Queue is unavailable. It is either full or empty, depending on whether it is producer or consumer sequence

-2 Temporary race condition. Sequence failed CAS and delegated decision to your application.

Consumer sequence interaction is almost identical. The only difference would be consumer reading queue item instead of writing it.

Performance of single-threaded sequences can benefit further from batching. Batching relies on receiving range of indices from sequence and calling done() at end of batch rather than for every queue item. This is what consumer code might look like (producer code is the same):

while (running) {
  long cursor = sequence.next();

  if (cursor < 0) {
    LockSupport.parkNanos(1);
    continue;
  }

  // get max index sequence can reach
  long available = sequence.available();

  // look thru queue elements without using sequence
  while (cursor < available) {
    queue.get(cursor++);
  }

  // calling done() only once per batch can yield significant performance benefit
  sequence.done(available - 1);
}

Multi-threaded sequence do not support batches.

Performance

I used Shipilev's project that already had Disruptor benchmark and I added QuestDB implementation of the same pipeline.

Benchmark source on GitHub

2 CPU MBP 2015

Benchmark          (slicesK)  (threads)  (workMult)  Mode  Cnt    Score    Error  Units
Disruptor.run            500          2          10    ss   50   10.043 ±  0.158  ms/op
Disruptor.run           1000          2          10    ss   50   19.944 ±  0.285  ms/op
Disruptor.run           5000          2          10    ss   50  133.082 ±  6.032  ms/op
QuestdbFanOut.run        500          2          10    ss   50   13.027 ±  0.180  ms/op
QuestdbFanOut.run       1000          2          10    ss   50   26.329 ±  0.327  ms/op
QuestdbFanOut.run       5000          2          10    ss   50  141.686 ±  4.129  ms/op
QuestdbWorker.run        500          2          10    ss   50   29.470 ±  0.976  ms/op
QuestdbWorker.run       1000          2          10    ss   50   62.205 ±  3.278  ms/op
QuestdbWorker.run       5000          2          10    ss   50  321.697 ± 12.031  ms/op

4 CPU x5960 @ 4.2Ghz

Benchmark          (slicesK)  (threads)  (workMult)  Mode  Cnt    Score    Error  Units
Disruptor.run            500          4          10    ss   50    6.892 ±  0.654  ms/op
Disruptor.run           1000          4          10    ss   50   10.143 ±  0.623  ms/op
Disruptor.run           5000          4          10    ss   50   54.084 ±  4.164  ms/op
QuestdbFanOut.run        500          4          10    ss   50    6.364 ±  0.197  ms/op
QuestdbFanOut.run       1000          4          10    ss   50   11.454 ±  0.754  ms/op
QuestdbFanOut.run       5000          4          10    ss   50   50.928 ±  3.264  ms/op
QuestdbWorker.run        500          4          10    ss   50   14.240 ±  1.341  ms/op
QuestdbWorker.run       1000          4          10    ss   50   27.246 ±  2.777  ms/op
QuestdbWorker.run       5000          4          10    ss   50  142.207 ± 15.157  ms/op

Disruptor and QuestDB perform essentially the same.

How to get it

Our messaging system is on Maven central as a part of QuestDB. Don't worry about package size though, QuestDB jar is around 3.6MB and has no dependencies. Jump to the GitHub release page for version reference.