What is a distributed commit protocol?

A commit protocol is an algorithm used for atomically committing a transaction. Atomicity implies that either all the changes (writes / updates) in the transaction will be committed taking the database to a new state or everything will be rolled-back without changing any state in the system.

Distributed commit protocols are widely used in distributed databases but are generally applicable to all kinds of distributed systems. For example, a distributed OLTP database that supports ACID transactions will have to use such a protocol to atomically commit or rollback a transaction on multiple participating nodes, leaving the database in a consistent state at the end with all committed changes being durable.

Two Phase Commit

As the name suggests, 2PC executes in two separate phases. The general topology of the system has a coordinator node where the transaction originates and two or more participating nodes. Coordinator node is also a participating node. For the sake of this discussion, let’s refer to transaction as a database transaction that comprises of one or more SQL statements (e.g INSERT, UPDATE, DELETE) that attempt to change the state of table / database.

Take this user flow as an example

SQL> INSERT INTO ...
SQL> INSERT INTO ...
SQL> DELETE FROM ...
SQL> UPDATE TABLE T ...
SQL> COMMIT

For a distributed transaction that uses 2PC, COMMIT statement triggers the execution of the protocol

Phase 1 – Request Vote

• Coordinator requests vote from all the participants. The votes will be used to decide if the transaction can be safely committed or needs to be aborted.
• Each participant sends one of the following response types to coordinator
  • YES – Participant ready to commit the transaction locally
  • NO – Participant wants to abort the transaction. Usually the case when something has gone wrong during the local execution of transaction at the participant

Phase 2 – Commit or Abort

Coordinator uses the responses received to decide the outcome of protocol.

All participants voted YES

• Coordinator sends COMMIT message to all participants. Each participants commits the transaction locally, release locks/resources etc and sends an acknowledgement to coordinator.
• Coordinator commits the transaction locally (updates metadata etc) and sends the response to user.

At least one participant voted NO

• Since the transaction has to be either atomically committed at all sites or no state change should happen, a single NO vote determines the outcome of the algorithm
• Coordinator sends ABORT message to all participants asking them to abort the local transaction.
• This will result in each participant rolling back the local changes (e.g using undo log) and restore the local state of data to a point where transaction never really happened.
• Coordinator receives acknowledgement from all participants nodes and aborts the transaction locally.

Note that coordinator can also be a participant in the protocol and regardless of whether it is a pure coordinator or both coordinator and participant, it can still decide to ABORT the transaction (for whatever problem it encounters locally) even if it had received YES votes from all the participants.

The above mentioned steps in both phases highlight the good scenario when everything in the system goes well without failures or crashes. Let’s now discuss few failure scenarios to understand critical problems associated with two-phase commit.

Failure Scenario 1: Coordinator crashes after completion of phase 1 but before starting phase 2

In this case, coordinator crashes after receiving votes from phase 1 but before starting phase 2. Consider a participant that had responded YES in first phase. It can’t simply assume positive outcome and go ahead with committing the transaction because it doesn’t know if rest of the participants had agreed to commit in phase 1. 

If the participants can communicate with each other, then they can find out the response of each other. If at least one of the participants had voted NO, it is pretty clear that coordinator would have decided to abort the transaction. So, the participants can make progress on this in-flight transaction by aborting the local changes.

However, if all the participants had voted YES, it is still not enough for them to go ahead with committing the transaction since they don’t know the final outcome decided by the coordinator. The protocol is effectively blocked on coordinator recovering from the failure and continuing with phase 2.

Failure Scenario 2: Coordinator starts phase 2, crashes before sending phase 2 outcome to a subset of nodes. Some of the participant nodes that received the phase 2 outcome act upon it and crash.

This is an extension of the previous scenario. Recovery becomes tricky if all the alive participant nodes had voted YES in phase 1. It is invalid for any of them to go ahead and commit the transaction because what if the crashed participant had voted NO and had aborted the transaction before crashing. When it comes back up, the data in the cluster would be inconsistent.

Similarly it is invalid to go ahead and abort the transaction if the outcome was to commit and the crashed participant had already committed the transaction.

Similar to the previous scenario, the protocol gets blocked due to the ambiguity on the outcome of the protocol in the event of failure

• When the failed coordinator recovers, then it can help participants progress by telling them to commit/abort the transaction depending on the outcome decided before it crashed.
• If the coordinator never recovers and we bring up a backup coordinator then it will still have to wait to get a response from the crashed participant node to determine the final outcome of the blocked protocol before it allows the remaining participants to proceed with commit or abort.

2PC protocol chooses to block and trades off liveness for safety to maintain consistency and correctness in the cluster in the event of failure. As a result, the transaction that was in the middle of commit by 2PC gets stalled incurring higher latency.

Additionally, any other concurrent conflicting (touching overlapping data) transactions on the participant nodes get stalled too since their completion is dependent on the completion of the transaction that is waiting on two-phase commit protocol to finish.

As an example, consider transactions T1 and T2 that want to update row R. T1 and T2 are conflicting transactions as they are interested in updating same data. Let’s say that database starts executing T1. As part of that, it would have acquired a lock (aka row level locking) on R to prevent any concurrent conflicting transaction (e.g T2) from modifying the row at the same time.

Only after T1 commits, can T2 proceed with updating row R. If T1 is stalled due to the above mentioned blocking scenario in two-phase commit, T2 and any other conflicting transaction will be impacted as well resulting in reduced throughput (transactions per sec) of the system in addition to higher latency per transaction.

In this post, I will give insight into a very important aspect of GC logging that engineers should be aware of when trying to understand the logs and looking for potential issues.

Stop-the-world GC operations can lead to long JVM pauses as application threads are suspended while JVM does garbage collection.

Let’s look at a sample snippet from GC log indicating GC activity and the corresponding JVM pause:

[GC pause (G1 Evacuation Pause) (young)
Desired survivor size 54525952 bytes, new threshold 8 (max 15)
- age   1:   10867224 bytes,   10867224 total
- age   2:    5046712 bytes,   15913936 total
- age   3:   17522776 bytes,   33436712 total
- age   4:    2174200 bytes,   35610912 total
- age   5:    1407704 bytes,   37018616 total
- age   6:   11410824 bytes,   48429440 total
- age   7:    3577488 bytes,   52006928 total
- age   8:   15038040 bytes,   67044968 total
- age   9:   17797760 bytes,   84842728 total
 2064.416: [G1Ergonomics (CSet Construction) start choosing CSet, _pending_cards: 11711, predicted base time: 24.56 ms, remaining time: 175.44 ms, target pause time: 200.00 ms]
 2064.416: [G1Ergonomics (CSet Construction) add young regions to CSet, eden: 215 regions, survivors: 26 regions, predicted young region time: 159.57 ms] 
.............
.............
.............
.............
.............
[Eden: 1720.0M(608.0M)->0.0B(9720.0M) Survivors: 208.0M->104.0M Heap: 3161.5M(16.0G)->1373.6M(16.0G)]
[Times: user=0.62 sys=0.12, real=0.05 secs] 
Total time for which application threads were stopped: 0.0514601 seconds, Stopping threads took: 0.0000901 seconds

Looking at the last two lines tells us how long the GC activity took and the time application threads were stopped for.

[Times: user=0.62 sys=0.12, real=0.05 secs]
Total time for which application threads were stopped: 0.0514601 seconds, Stopping threads took: 0.0000901 seconds

As indicated by the log, the pause lasted 0.05secs. Like this we can see other GC activity in the log to find out significant pauses.

Until very recently, I typically had one major purpose when looking at GC logs:

Find out the substantial pauses (as indicated by total time for which application threads were stopped) and then try to understand from code, heap dumps etc why our application is experiencing long GC pause.

However, another equally important metric that I used to ignore is the time taken to stop the threads and whether the pause was due to GC or not. Let me give some important background before diving into this metric.

JVM Safepoints

JVM has a concept of safepoint. Safepoint is a point in execution of a thread where state of the thread is well defined. All the object references used by the thread are accounted for by the JVM. Ensuring that a thread has entered a safepoint stage allows JVM to run some bookkeeping and other operations that might manipulate heap and stack space. Therefore, once the thread exits safepoint stage, any operation(s) run by JVM in the meantime wouldn’t have changed the thread’s view of memory.

However, if JVM ran any maintenance operation without ensuring that thread(s) have reached a safe stage, there is a possibility of corrupting the thread’s view of memory. As an example, JVM might collect a live object if it was not accounted for as of that time. By knowing the state of thread (what is there in registers, stack, heap etc), safepoints help JVM avoid such problems.

What JVM operations require threads to enter safepoint?

Few operations that require threads to enter safepoint stage:

• Stop the world GC operations
• Heap dump
• Flushing the code cache
• Method de-optimization

How does a thread reach safepoint stage?

JVM doesn’t force a thread to enter safepoint stage. Instead, it lets threads poll a safepoint flag at some points in the byte-code. Specifically, the flag indicates whether the thread should enter safepoint or not. Accordingly, the thread suspends itself if the flag indicates so. The JVM implementation inserts “safepoint polls” at appropriate places. It is up to the JVM to ensure that following are highly optimized to minimize the impact on application performance:

• Number of safepoint polls.
• The cost of polling/checking the flag.

Few examples of states of a threads when it has already reached safepoint stage:

• Thread waiting on monitor/lock/blocking IO is in a safepoint state. Before the transition to blocked happened (to wait for I/O or lock), the thread was brought to safepoint state.
• A thread executing native code via JNI – even though the thread is running, before the execution was transferred to JNI land, the thread was brought to a safepoint state.

Time taken by threads to reach safepoint

Now that we have some reasonable context on safepoints, let’s get back to their relationship with GC logs. Since JVM doesn’t force the threads to enter safepoint, we don’t know the number of instructions the thread executes before it hits a safepoint poll and enters safepoint. Thus it is possible (in theory at least) that threads might taken their own sweet time to reach safepoint.

One thing that readers should remember is that using the JVM option XX:+PrintGCApplicationStoppedTime will report pause times in GC logs along with the time taken to pause the threads. Note that these two metrics will be reported in the GC logs regardless of the fact that JVM operation that triggered the pause was GC or not.

Going back to the GC log snippet shown earlier:

Total time for which application threads were stopped: 0.0514601 seconds, Stopping threads took: 0.0000901 seconds

We should interpret it as follows: the threads were stopped for 0.05secs out of which 0.00009secs were taken to stop the threads.

If threads are taking a long time to reach safepoint, the “Stopping threads took” metric will be higher and consequently “total time for which application threads were stopped” will also be higher since it subsumes the former.

If a thread is taking a long time to reach safepoint, it will derail the JVM — since the threads that have reached the safepoint will be suspended forever until (1) offending thread reaches safepoint and (2) JVM executes the operation that triggered the safepoint in first place and finally (3) JVM lets threads exit safepoint and resume execution.

Some major factors that can increase the time taken by threads to reach safepoint:

• High degree of page faults – during a page fault, a thread is suspended by the OS until the requested page is retrieved from the backing store, entered into the page table etc. The suspended thread(s) will not be able to check for safepoint polls until they resume execution. So if the JVM has really initiated a safepoint, all the other threads that have reached safepoint will have to wait until this thread reaches safepoint and JVM completes the safepoint operation.
• High ratio of number of threads to cores — increases the CPU contention and excessive context switching.
• Counted loops – Loops where the counter variable is of INT type are treated as counted loops by the JVM and it does not insert a safepoint poll in the loop to speed up the execution of such loops. So if an enormous computation is being done in an uncounted loop, the calling thread will not be able to check for safepoint poll until loop execution is over.

Takeaways for examining GC logs

Make sure we are using at least the following JVM options:

• -XX:+PrintGCDetails
• -XX:+PrintGCTimeStamps
• -XX:+PrintGCApplicationStoppedTime

At a high level, we should look for a couple of patterns:

GC activity and corresponding pauses

[GC pause (G1 Evacuation Pause) (young)
Desired survivor size 54525952 bytes, new threshold 8 (max 15)
- age   1:   10867224 bytes,   10867224 total
- age   2:    5046712 bytes,   15913936 total
............
............
............
[Times: user=0.62 sys=0.12, real=0.05 secs] 
Total time for which application threads were stopped: 0.0514601 seconds, Stopping threads took: 0.0000901 seconds

More specifically, this pattern will have info on GC activity and then corresponding pause time followed by next GC activity and pause time. Something like:

Look at both “total time taken for which the application threads were stopped” and “stopping threads took”.

Pauses without GC activity

By now the readers should know that apart from GC, there are other JVM internal operations as well that will require threads to reach a safepoint. Since we are using -XX:+PrintGCApplicationStoppedTime option, such pauses will also be present in the GC log. Between each successive pause, there will be no GC activity and that is how we need to identify such pauses from the log. The following snippet shows an example:

Total time for which application threads were stopped: 46.7196326 seconds, Stopping threads took: 46.6686622 seconds
Total time for which application threads were stopped: 43.5977492 seconds, Stopping threads took: 43.4309684 seconds
Total time for which application threads were stopped: 30.3403226 seconds, Stopping threads took: 30.2303471 seconds
Total time for which application threads were stopped: 5.6498432 seconds, Stopping threads took: 5.5128792 seconds
Total time for which application threads were stopped: 1.2957247 seconds, Stopping threads took: 1.2336227 seconds

As shown above, between each successive pause, there is no GC activity. So such pauses are not GC related.

Couple of things that we should think about investigating:

• Firstly, we need to find out what is causing the JVM to initiate safepoint. The following JVM options can be helpful. They will print the details of the safepoint (including reason, timings etc).
  • -XX:+PrintSafepointStatistics
  • -XX:PrintSafepointStatisticsCount=1
  • Note that information for these options is logged to stdout(not GC log).
• Secondly, we need to see if threads took a long time to reach safepoint. This is true for the example shown above:
  • In one case, it took 46secs to stop the threads whereas the total time for which the threads were stopped was just a little more than 46secs.
  • Clearly, whatever safepoint operation led to the above pause lasted for negligible period.
  • We need to investigate why threads took a long time to reach safepoint.

Few additional options that can be useful:

• -XX:+SafepointTimeout
• XX:SafepointTimeoutDelay=100

Using these options will help us debug the case where the threads are taking longer than usual to reach safepoint. This will timeout the safepoint sync if a thread fails to reach the safepoint in 100ms. It will also print information where exactly thread is spending time (e.g page fault trap) that is preventing it from reaching safepoint.

Pessimistic locking protocol

Let’s first discuss the opposite of optimistic locking to setup the context. Pessimistic locking is the main locking paradigm used for guaranteeing mutual exclusion for a given piece of code subject to execution by reader and writer threads concurrently.

It is called as pessimistic because each thread acquires an explicit lock before executing the critical section code and releases the lock once it finishes the critical section execution

• Thread acquires the lock
  • If another thread holds the lock, the calling thread is blocked. Once the thread currently owning the lock releases it, thread(s) waiting on the lock are woken up and become eligible for acquiring the lock. The thread that actually gets to acquire the lock is more dependent on OS scheduling and/or the scheduling (if any) enforced by the lock library.
• Thread executes the critical section
• Thread releases the lock

Taking example of readers-writer lock, pessimistic locking guarantees the following

• If a writer thread owns an exclusive lock, no other reader/writer threads will be able to acquire the lock for the resource and read/update data.
• If reader thread(s) own a shared lock, no other writer thread will be able to acquire the lock for the resource and update data.

Optimistic locking protocol

The most important aspect of optimistic locking technique is that readers don’t take any locks. The high level idea for the reader is to go ahead and read data followed by some verification logic to check that the read was consistent or not. In other words,

• If the data was read while it was being mutated, then it was not a consistent read.
• If the data was read while it was not being mutated, then the reader got a consistent read.

A common way to implement this is using a monotonically increasing version number. The shared resource is associated with a version number (typically 8 bytes).

• The value starts at 0
• An even value of the version number indicates no one is currently updating the resource.
• An odd value of the version number indicates a writer is currently in the process of updating the data.

Protocol for the reader

1. Read the version number — store it as V1.
2. Read the data.
3. Read the version number — store it as V2.

If V1 and V2 are even and equal, we got a consistent read else we know someone was changing the data. Otherwise, reader got an inconsistent read since some writer was concurrently making changes.

This could have happened even before reader executed step (1) or the writer could have started making the changes after step (1). In any case, the read happened while the data was potentially being changed. So the reader should discard the data read in step (2) and retry the operation.

Protocol for the writer

1. Read the version number.
2. If it is odd (implying a concurrent writer is already in progress), bail out or retry.
3. If even, atomically increment the version number by 1 to make it odd:
   1. If you are using Java, we can use AtomicLong for the version number.
   2. In any case, atomic increment will be done using CAS (compare and swap).
   3. If the atomic increment fails (implying a concurrent writer succeeded in bumping up the version), bail out or retry.
4. Make changes.
5. Increment the version number by 1 to make it even.

How does the reader read data?

Since readers don’t use any locks, it is possible that writer might be actively changing the data while reader is trying to read it. The 3 step protocol mentioned above lets the reader detect whether the read was consistent or not but there is still a possibility that reader sees a core-dump (crash due to segmentation fault) while it is trying to read the data in step 2 and follows an invalid pointer.

There are two potential ways to solve this:

• Reader uses memcpy() – it implies that the data region that optimistic locking is being used over is simply copied by the reader in step 2 and not interpreted. Later if the version is matched, the reader knows it was a consistent read and reader can go ahead and interpret/understand the contents. If the version mismatches, the reader can just throw away the copied data and reattempt.
• Reader is aware of the write pattern – this avoids memcpy but is very implementation specific and really depends on the data structure (and it’s format) that is being protected by optimistic locking. For example, if the writer is changing the shared data in a strict sequence of steps then may be the writer can use store barriers to indicate its progress and/or a strong indicator of when it is not safe for the reader to proceed further.

There could be more ways to solve this and I encourage readers to share their thoughts in comments.

Suitable scenarios and Retry mechanism

Before we discuss the retry mechanism, let’s first try to understand the cases where optimistic locking is suitable. There are two scenarios where we should consider using optimistic locking:

• Writers hold locks for a short period of time. In other words, the critical section is small and can be executed fairly quickly.
• There is less contention/conflict.

High contention will increase the rate of read and write operations that fail (due to version mismatch or atomic increment failures) and have to be reattempted (if the implementation chooses to).

Reattempt for reader is to simply invoke the read operation again which will execute the 3 steps mentioned above for reader protocol.

Writer can typically use a spin-lock. In this case, the write protocol is executed in an infinite while loop. The specific implementation can choose to have a small sleep time between each attempt.

Using spin lock works since the assumptions are (1) critical section is small and (2) contention is generally less. Also, all modern CPU architectures are multi-core. Using spin-lock on a single processor system does not make any sense.

If the critical section is long and takes a significant time to execute then the writers issuing reattempts within spin locks will only be wasting CPU cycles by spinning for a long period of time. Similarly if the contention is high, the writers will again end up spinning a lot trying to race against each other at step 2 or step 3 of their protocol and making reattempts.

It is not necessary to use spin-lock for writers. It is perfectly fine to fail the operation and let the caller of the API decide if the operation needs to be reattempted.

Advantages and Use Cases

Optimistic locking avoids the overhead of managing explicit locks. If the conflicts are rare (and especially if it’s a read heavy workload), optimistic locking will help improve performance on multi-core architectures since the reads are executed without any lock/unlock and writes use a very lightweight version based lock.

On the other hand, if there are frequent conflicts then optimistic locking is unlikely to provide any benefits since the cost of constantly retrying the operation (and this actually depends on the context and the nature of write operation) might negate the benefit of avoiding the overhead of lock/unlock operations.

We can leverage optimistic locking on pretty much all data structures that we would use pessimistic locking on — in-memory caches, shared memory regions, lightweight database transactions are few of the common uses cases of optimistic locking.

Both volatile and synchronized are used in multi-threaded programs to get some degree of thread safety depending on the operations performed by different threads. Consider the following piece of code

class VolatileDemo {
  private int data;
}

int get() {
  return data;
}

void set(int val) {
  data = val;
}

Suppose get() is invoked by multiple threads and set() is called by only only thread. Since there is only one writer, we don’t have to worry about race conditions. However, considering a modern multi-core architecture, it is likely that the reader and writer threads could be running on different cores. Thus the most recent value of data might be sitting in CPU cache of the core that writer thread was running on. This can prevent reader threads from seeing the result of latest set() operation. This is a memory visibility problem from concurrency point of view where updates made to a shared object by one thread may not be visible to other threads.

How does volatile solve visibility problem?

For the scenario described above, declaring “data” field as volatile will help. Every time a volatile variable is updated, JVM executes a “store barrier” instruction effectively flushing the CPU caches and ensuring that updates are written to main memory. Reads of a volatile variable are also done on physical memory thus guaranteeing that readers will always see the updated value.

Using volatile for a field in Java also instructs the compiler not to optimize away the variable.

Some common uses of volatile

Let’s go through some common coding patterns in Java where volatile can be useful

Using volatile for a status change flag

Let’s say there is a thread that does some piece of work until it has been told to stop. There might be a top level thread that created a bunch of worker threads and is responsible for telling the worker threads to stop. The worker thread code might look like:

private volatile boolean done = false;

// some other thread tells this thread to stop
void stop() {
  done = true
}

void work() {
  // if done is not volatile, the worker thread may not see
  // the value of done as true if the read is done from processor
  // cache
  while (!done) {
     // do something
  }
}

In this case, declaring “done” as volatile allows the worker thread to see the update. This also highlights a situation where cache coherence implemented by the hardware wouldn’t have helped the worker thread in seeing the change in value to “done” through cache invalidation (and cache miss) if the compiler had optimized away the loop thinking “done” will never be set to true.

while (!false) {
   // do something
}

Using volatile in callbacks

Consider we are writing some test that uses a library to register one or more callback functions. The test code essentially implements an interface of callback functions and performs some actions by calling the APIs of library — as a result of which some callbacks will be invoked by the library into the test code.

class Test {
  private volatile int callbackOneStat;
  private volatile int callbackTwoStat

  @Test
  void test() {
    library.registerCallbacks(new CallbackObject());
    library.performActions();
    // guarantees callbacks are invoked
    library.waitForActionsToComplete(); 
    assertEquals(10, callbackOneStat);
    assertEquals(20, callbackTwoStat);
  }

  private class CallbackObject implements Callback {

    @Override
    void callbackOne(Action action, int value) {
      callbackOneStat = value;
    }

    @Override
    void callbackTwo(Action action, int value) {
      callbackTwoStat = value;
    }
  }
}

Since the callbacks are invoked by a thread from library, declaring the stats as volatile guarantees that the thread running test code will see the change in value.

Using volatile as an execution indicator

With the new Java Memory Model, volatile also prevents any instruction reordering (by compiler or CPU). More specifically:

• The non-volatile writes and reads that happen before a volatile write as per the program order cannot be reordered to happen after the volatile write.
  • This implies that once a write to volatile variable V is completed by thread A, the writes to all variables visible to A up until the write of V are also made visible in memory. Thus another thread B sees the updated value of not only the volatile field V, but also of the other non volatile fields (that were visible to thread A when volatile write to V was completed)
• Similarly, non-volatile reads and writes that happen after volatile read as per the program order cannot be reordered to happen before the volatile read.

This guarantee (also referred to as happens-before relationship in Java) extends the usage of volatile field as an indicator of what operations have indeed completed. A commonly used scenario is to check if certain piece of code is initialized — where the initialization of member variables and data structures is done in one thread and the actual work is done in other thread but is dependent on initialization having properly completed.

class VolatileDemo {
  private Map<String, String> hashMap; 
  private List<Reader> readers;
  private volatile boolean initDone = false;

  // some other thread does this
  void init() {
    // initialize and/or populate hashMap
    // initialize/create readers
    initDone = true;
  }

  // different thread does the work
  void work() {
     while (!initDone) {
       // sleep
     }
     // use the initialized map and readers
  }
}

Java Memory Model will not allow any of the initialization statements to be reordered after the write to volatile variable “initDone”. Therefore if the thread running the work() function is actually seeing the value of initDone as true, it is guaranteed that map and readers are initialized and the worker thread is seeing the initialized data structures.

Since the initialization of map (including the initialization of objects in the map) and initialization of readers definitely happens before the volatile write in initializer thread and read of map and readers definitely happens in worker thread after the volatile read, we can conclude that all initialized data is visible in worker thread.

Old Java Memory Model only ensured that volatile reads and writes cannot be reordered with other volatile reads and writes and there were absolutely no guarantees w.r.t operations on non-volatile variables. This prevented the use of volatile in above cases.

When is volatile enough?

The usage of volatile as a synchronization construct is fairly limited. It provides a subset of properties provided by synchronized. With synchronized method or block, we get:

• Mutual exclusion and thus no race conditions for any kind of operations on shared data
• Memory visibility and happens before relationship – a monitor unlock by a thread is guaranteed to happen before a monitor lock (on the same lock object) by another thread.

Now in case of volatile, we only get the second property. Volatile doesn’t provide mutual exclusion and therefore if two threads are writing to shared data and the change to data is dependent on the current value of the data or any other global shared state of the program, using volatile is not enough for ensuring correct behavior and consistency of data. Synchronization should be used in such cases

Another example where volatile is not suitable would be implementing a thread safe counter which follows the read-modify-write pattern since the increment operator is actually doing multiple operations underneath. The correct way to implement such a thread safe counter is to use AtomicInteger, AtomicLong provided by Java concurrent library.

Volatile is enough when:

Multiple readers and single writer scenario:

• Declaring volatile will do the job no matter what kind of updates are being made to the shared data.

Multiple readers and multiple writers scenario:

• Update to shared data should not be dependent on the current value of the data and any other state of the program.
• If the nature of updates to shared data do not have the above invariant, then synchronization should be used

We should carefully assess the nature of updates to shared data and if it can be updated by multiple threads (mostly true in all general cases) and then decide if volatile can be used instead of synchronization.

Introduction to Netty

Netty is an open source framework for building a high performance networking (messaging or RPC or transport) component of a distributed system. It is built on the design principles of asynchronous programming and helps developers by abstracting away significant design, implementation and maintenance effort required for having a high performance networking layer in a system. Non-blocking, asynchronous and event driven I/O are the basic building blocks of Netty.

Why use Netty when we have Java NIO?

Let’s refresh our memory of I/O capabilities we get with JDK. The following code is a simple server written using Java I/O (old IO libraries)

public class SimpleBlockingServer {

  private final int port;

  private SimpleBlockingServer(final int port) {
    this.port = port;
  }

  public static void main(String[] args) throws Exception {
    int port = Integer.parseInt(args[0]);
    new SimpleBlockingServer(port).start();
  }

  private void start() throws Exception {
    // bind the server to specified port
    // ServerSocket will listen for new connections
    // and accept them
    final ServerSocket serverSocket = new ServerSocket(port);
    // infinite loop to handle incoming connections
    while (true) {
      // listen on the server socket for new connections and accept
      // will block until connection is available
      // socket represents a newly accepted connection
      final Socket socket = serverSocket.accept();
      System.out.println("Accepted connection from socket: " + socket);
      // create a separate thread for each incoming connection
      final RequestHandler requestHandler = new RequestHandler(socket);
      new Thread(requestHandler).start();
    }
  }
}

The full code can be seen here. The main problem with the above code is that I/O methods provided by java.net (sometimes referred to as old Java I/O) underneath use blocking system calls.

• The accept() call will wait forever until a new connection is accepted. This implies that the above thread can’t process any other already accepted connection having some activity (for e.g, data available to read). To do this multiplexing between connections, the server simply creates a separate thread for each connection otherwise the code will not be able to handle more than one connection.
• Creating a separate thread for each connection will not scale beyond a point when the server should be able to handle several hundred thousands of concurrent connections without any noticeable degradation in performance. With thread per connection approach, it is very likely that OS will spend much more time in context switching than doing the actual work.
• Similarly, thread per connection approach is not the wisest approach to utilize system resources at peak workloads. What we need is a mechanism to handle several concurrent connections in fewer threads.

Java NIO (Non blocking IO) solved this problem by introducing Selector – the holy grail of high performance network I/O in Java. Selector can be used to register a set of non blocking sockets/descriptors that are monitored for any activity/events. This allows a single thread to handle multiple concurrent connections. A sample implementation can be seen here where a single thread is multiplexing between multiple non-blocking registered sockets.

Now a sophisticated implementation using Selector will require some sort of custom dispatch code (in accordance with reactor pattern) where events happening on the descriptors registered with Selector are handed off to appropriate handler thread which further uses non blocking read/write socket operations. Writing and maintaining this specialized multi-threaded networking code can be quite cumbersome especially for large scale distributed systems.

This is where Netty comes into picture. Java NIO is the foundation of non-blocking IO libraries in Netty. Netty goes one step ahead and leverages the concept of callbacks, futures, promises, event based programming paradigm to deliver a fully asynchronous network I/O library. Developers no longer have to directly work with Selector as Netty abstracts away all of that including multi-threading.

In fact, selector has its own performance disadvantages and is not the most efficient option available today. More performant variants of select()/poll() system calls are available in the form of epoll (Linux only) and kqueue (BSD only) for highly scalable multiplexing. The good news is that Netty provides wrappers over all of them.

Distributed In-Memory Partitioned Store

As part of my weekend project, I am implementing a distributed system (to run some experiments for learning purposes). The following diagram depicts the architecture as a 3-node cluster. Each node will host some data. Since the focus of this post is RPC, I have tried to show how the nodes will talk to each other (for sending/receiving messages, heartbeat pings etc) via the RPC client and service running on each endpoint — to keep diagram clean, all M*N arrows are not shown between endpoints.

Now to complement this post with actual deployable code, I developed a version of the above architecture with a simple demo service to demonstrate RPC request-response between endpoints. I call this as V0 design

• The in-memory store is on a single node. Let’s call it data node
• The demo service runs on all other nodes and sends/receives messages to/from data node.
  • It is fine to run the demo service on data node as well. In that case, the request/response will happen via local RPC endpoint.

V0 Design

• In-Memory Store – A simple hash table storing Key-Value mappings on data node
• Async thread pool – The requests received by RPC server are processed asynchronously — explained in detail further in the post
• Demo service – Multi-threaded application that issues get/put requests via the local RPC client service
• Netty Channel – This represents a connection between two endpoints.

Protocol

The messaging format is specified using Google Protocol buffers. The following code shows the format of RPCRequest and how it is composed of a get or put request.

message GetRequest {
  required string key = 1;
}

message PutRequest {
  required string key = 1;
  required string value = 2;
}

message RPCRequest {
  required int64 sequenceNum = 1;
  oneof KVStoreServerRequest {
    GetRequest getRequest = 2;
    PutRequest putRequest = 3;
  }
}

RPC Server

Channel and Event Loop

Channel represents a connection (at least for current discussion) and supports asynchronous and non-blocking I/O operations. A channel is assigned to a single event loop in netty. An event loop is a single threaded event executor responsible for multiplexing amongst the channels it is managing. An event loop group is a collection of such single threaded event executors. As mentioned earlier in the post, the event loop (and group) come in multiple flavors depending on what they use internally — selector, epoll or kqueue.

I have used NioEventLoopGroup which uses Java NIO Selector internally. We need to have two event loop groups for RPC server.

• One event loop group responsible for accepting connections at the server. For each accepted connection, a child channel (SocketChannel) is created and assigned to the second event loop group.
• The second event loop group assigns the accepted connection (represented by SocketChannel) to one of its event loop for the life of connection. All the event activity on the connection will be managed by that particular event loop.

Channel Initializer, Channel Handler and Channel Pipeline

Channel handler is powerful concept in Netty that allows us to write custom code for handling inbound and outbound events as they happen and propagate across the channel. Each channel has its own pipeline where channel handler(s) are installed by our code.

Initializer is a custom code we write to tell Netty how we would like to initialize each accepted connection (SocketChannel). This is where we specify the inbound and outbound handlers and whenever a connection is accepted, Netty will add the handlers to the pipeline of newly created channel.

Netty’s encapsulation over the bytes traveling on wire is ByteBuf (NIO buffer). We need to have appropriate encoder/decoder handlers in the pipeline to handle the conversion between ByteBuf and protobuf. Message encoders and decoders are also installed as handlers in the pipeline of a channel and Netty does a very good job at providing base classes and adapters which can easily be extended.

The flow of events in the channel through the handlers is in the same order as they are added in the pipeline. Netty takes care of differentiating inbound v/s outbound event handlers.

Bootstrap

This concept is exposed by Netty to configure and start the server by specifying the two event loop groups, socket channel initializer, handlers etc. We specify the port that server will bind to and be ready to accept connections.

Please read the code for a working implementation of RPC server using above concepts. Complete RPC protocol specification can be seen here.

Async worker pool

As you may notice in the code, the RPCRequestHandler uses Completable Future to run an async computation inside an executor pool and a callback code that sends the RPC response when future completes.

By now we should know that our channel handler code is invoked by event loop thread of Netty (Netty documentation refers to it as I/O or network thread). The handler code should be fairly quick (should not block) in order to let the I/O thread continue processing the events on the connections it is managing. This is a recommended best practice by Netty for scalability and performance.

This is where Completable Future comes into picture which allows us to use Java Future along with callback to be invoked when future completes rather than us doing future.get() and again blocking thus defeating the purpose of future. The worker pool (currently 2 threads) executes the task of talking to the in-memory store and running the get/put request. Once the task completes (implying future completes and we have the result of get/put), callback is invoked by worker pool to send the RPC response.

RPC Client

Event Loop and Channel Initializer

Unlike RPC server, we need a single event loop in RPC client to manage connections. The channel initializer is more or less same except for protobuf decoder which now converts the bytes in RPCResponse. Full working implementation can be seen here

Bootstrap

The concept of Bootstrap is also applicable in RPC client. We configure the client by providing the event loop and channel initializer. Finally, as and when needed, bootstrap is used to connect to RPC server endpoint at a particular address and port. All operations are asynchronous and thus remote connect returns a future

/**
 * Establish connection with an endpoint
 * @param endpoint {@link Endpoint} representing a peer with host * 
 * address/port
 * @return {@link ChannelFuture}
 */
ChannelFuture connectToPeer(Endpoint endpoint) {
  return bootstrap.connect(endpoint.getAddress(), endpoint.getPort());
}

RPC Response Handler, Response Listeners and Sequence Numbers

The final inbound handler that handles RPCResponse extends the basic inbound event handler implementation provided by Netty. As you may notice, the memory associated with incoming response message is not released by channelRead0() method. This is something that is taken care of by base class implementation which releases the resource after the call returns from channelRead0().

In the current implementation, as soon as the response is received, we just simply log a message (response data which also includes the original request and a sequence number). This is done within the event loop thread itself.

Every time, demo service thread makes a request via the local RPC client, it gets a new sequence number and RPC client code maintains a mapping between sequence number and corresponding listener (a sort of future but current implementation is dumb). Once the response arrives, we remove the corresponding listener from table and invoke the done() operation. I plan to improve the response listener semantics

class RPCResponseHandler extends SimpleChannelInboundHandler<KVStoreRPC.RPCResponse> {
    @Override
    public void channelRead0(ChannelHandlerContext context, KVStoreRPC.RPCResponse rpcResponse) {
      final long sequenceNum = rpcResponse.getSequenceNum();
      final RPCResponseListener listener = responseListeners.remove(sequenceNum);
      // right now just complete it in the network thread -- event loop thread
      listener.done(rpcResponse);
    }
  }

Connection Proxy

A simple wrapper over RPC Client for talking to a particular endpoint. It establishes connection and sends requests to a RPC server endpoint in the cluster. Demo service threads on a node use a single connection proxy (same physical connection to remote endpoint) to send requests.

Adding listeners to Netty’s ChannelFuture

As mentioned earlier, I/O operations (read, write, connect etc) on the channel are asynchronous. Pretty much all of them return ChannelFuture — Netty’s extension to Java Future. We can add custom listener code to this future as a callback to be invoked by the event loop thread when async operation completed. For example, the following listener is invoked in the client side once RPC request send operation completes. Similar listeners have been used in a couple of other places.

final ChannelFuture sendFuture = channel.writeAndFlush(rpcRequest);
sendFuture.addListener(new ChannelFutureListener() {
 @Override
 public void operationComplete(ChannelFuture future) throws Exception 
 {
    if (future.isSuccess()) {
      System.out.println("Request# " + sequenceNum + " sent successfully to " + future.channel().remoteAddress());
    } else {
      System.out.println("Unable to send request# " + sequenceNum);
    }
 }
});

Modules

• Daemon is the main KVStore process that runs on each node and starts/creates every other service — currently RPC , demo and in-memory store.

Deployment and Testing

KVStore is a maven project and distribution is done in the form of tarball. I have tested this both locally and on a 3 node EC2 cluster. The 3-node deployment architecture is same as shown earlier in V0 architecture diagram. Please see detailed instructions here

Future Work

There is a lot more I want to do here. Some immediate items are:

• Improvement to response listener
• Service discovery support using Zookeeper or Consul.
• Exception handling — currently there is basic resource cleanup.

More on Netty

This post is by no means a Netty cookbook. The purpose is to put something useful that can help people get started with Netty both theoretically and some initial tested distributed system implementing the concepts. I would encourage people to read Netty in Action for detailed information on Netty internals.

The tuple at a time processing methodology has some disadvantages when it comes to building high performance SQL execution engines for analytical queries.

• Excessive function calls – tuple-at-a-time model has a lot of overhead associated with a large number of function calls throughout query execution. As a result, it is difficult to write efficient query interpreters around this model. For some TPC-H queries, this can result in performance degradation by an order of magnitude.
• Defeats the purpose of using columnar format – Columnar format (in-memory or on-disk) has become the de facto format for building efficient analytical query engines to accelerate data warehouse or business intelligence style queries. Query processing algorithms written around tuple-at-a-time model are unable to take advantage of columnar format since it works on tuples as opposed to columns. Thus tuple-at-a-time model misses out on several key fundamental techniques that can be employed to run analytical queries efficiently on columnar format.

Vectorized query processing model is a significant departure from tuple-at-a-time model and has two key differences:

• Column oriented processing – Write query processing algorithms that operate on columns as long as possible in the execution plan. In other words, refrain from working on tuples till late in the plan until its absolutely necessary (during projecting result-set back to the user).
• Push column vectors through the query plan tree – Instead of passing around tuple from one operator to another, pass column(s) containing a fixed number of records. This is typically referred to a “block of vectors” or “batch of vectors”. This is the unit of work that is fed into the operator or the result produced by an operator.
  • Please don’t misunderstand this as “multiple tuples”. Forming tuples requires stitching values together from different columns to form rows and this disallows column oriented processing. A batch consists of multiple records but they are still passed as a set of columns — each column in its own data structure generally known as vector in columnar databases.

Some advantages associated with vectorized processing:

• Efficient interpretation – by working on a batch of vectors (as opposed to tuple-at-a-time) across the operators in execution plan, we see an overall improvement in query execution time.
• Better cache locality and efficient utilization of CPU cache – we can quickly loop through tightly packed values of a column and do the necessary processing — predicate evaluation, arithmetic computations etc. Cache lines are filled with related values (from the same column) as opposed to heterogeneous values from multiple columns in a tuple where some columns may not even be touched by the query.
• Better chance of native optimizations by the compiler – tight loop based vectorized algorithms are good candidates of automatic optimization by compilers.
• Leverage hardware acceleration – well aligned column data in densely packed arrays is amenable to acceleration using SIMD instructions. Common operations like FILTER, SUM, MIN, MAX can be accelerated by an order of magnitude by exploiting data-level parallelism of SIMD instructions.
• Directly operate on compressed columnar data – columnar format allows us to encode column values with lightweight compression algorithms (dictionary encoding, RLE etc) which trade compression ratio for better query performance. These compression schemes allow the query processing code to directly work (in some cases) on compressed column values without having to decompress everything upfront. For example:
  • FILTER on variable width columns can be efficiently executed if the column is dictionary encoded and we re-write the FILTER using dictionary codes (which are fixed width) which is amenable to SIMD processing.
  • RLE (Run Length Encoding) can be leveraged to do quick arithmetic on a set of column values without having to process/load each column value individually.

By executing query processing logic directly on compressed columnar data, we save a lot on CPU and this can further improve the query performance.

The following diagram shows a high level data flow during query execution in tuple-at-a-time and vectorized model respectively. 

I recently gave a talk at Strata Data Conference on Vectorized Processing using Apache Arrow and how Dremio has built a high performance execution engine on top of Apache Arrow. Please see the following links to view the presentation and learn more about Dremio and Apache Arrow.

• Strata Talk – slides can be downloaded from this link.
• Webinar
• Video

Please join us at the first Apache Arrow meetup in Bay Area on May 15th in San Francisco.

• Introduction
• Storage Organization
• Vectorized Query Execution
• CPU Cache Friendly
• Late Materialization
• Compression

Introduction

Analytic workloads comprise of operations like scans, joins, aggregations etc. These operations are concerned with data retrieval, some computations over the values stored in table cells, and predicate evaluation. Such operations typically touch a very large number of rows but only a few columns in the table. As far as I understand, this is the fundamental difference between analytic queries and OLTP-style operations. Here are some examples differentiating the two types of operations.

Now let’s talk about some of the factors that work in favor of column stores for efficient analytical queries.

1. Organization of Data

The fundamental (and most obvious) difference between column stores and row stores is the way they organize and store table data.

Row-oriented databases store data on a row-by-row basis. Each row has its data stored together (contiguously) in-memory or on-disk or both. Thus its easier to grab some/all columns of a “particular row” given its ROWID. So in a single seek, row’s data can be loaded from disk into memory.

However, if the operation requires to access only a handful of columns from a very large number of rows in the table (the case of analytic or data warehouse queries), row stores tend to become less efficient since we have to walk the row, skip over the columns not needed, extract the value from relevant column of interest and move on to the next row.

In columnar databases, values of a particular column are stored separately or individually — contiguous layout for values of a given column in-memory or on-disk. If the query touches only few columns, it is relatively faster to load all values of a “particular column” into memory from disk in fewer I/Os and further into CPU cache in fewer instructions.

In other words, column stores are capable of accessing values of a particular column independently without bothering about the other columns in the table which may not even be needed by the query plan. Thus in columnar storage format, there is no need to read (and then skip/discard) columns that are not needed and this provides better utilization of available I/O and CPU-memory bandwidth for analytical queries.

2. Vectorized Query Execution

The columnar storage format allows other efficient techniques to be built on top of it to deliver high performance for analytical or data warehouse type of queries.

Columnar databases are capable of doing faster predicate evaluation by exploiting data level parallelism through SIMD instructions. Multiple column values can be operated (+, =, <, > etc) on in a single CPU instruction using SIMD technology.

For example, with 128 bit SIMD register, predicate evaluation using “=” operator can be done on 8 2-byte column values in a single instruction — data level parallelism. Not only the evaluation is faster, but a single instruction can also load multiple column values into SIMD register from columnar in-memory storage format.

Example Query: SELECT COUNT(*) FROM EMPLOYEE WHERE AGE>40 AND AGE<55

AGE column can be efficiently bit-packed into fixed-width 1 byte numeric value instead of using a regular 4 or 8 byte numeric values. Thus 16 values from AGE column can be evaluated against the provided predicates in a single instruction.

SQL SUM, AVG, COUNT, MIN, MAX can all be accelerated using SIMD technology.

3. CPU Cache Friendly

The columnar storage format allows better utilization of CPU cache since cache lines are full of related values (same column) that are needed by query executor to run some operations (evaluation, computation etc). In other words, only data that really needs to be examined is brought into cache. Thus looping through values of a column packed into columnar format is faster.

This is completely different from tuple based iterator in conventional row stores where a tuple consists of values from multiple columns and operator might need value only from one column. Thus cache tends to contain mixed data.

4. Late Materialization

Any query eventually needs to send back the result-set (tuples) to the end user. This should happen regardless of what type (column store or row store) of data store is being used.

For example, the query should ultimately send back tuples <first_name, last_name, age> to the user even though the columns are stored separately in a column store.

SELECT FIRST_NAME, LAST_NAME, AGE FROM EMPLOYEE WHERE AGE>40 AND AGE<60

• AGE column can be scanned first to do efficient (SIMD scans) predicate evaluation and filter out the values that don’t pass the condition.
• The output of SIMD scan (typically a mask or bit-vector) can then be used to scan through the FIRST_NAME and LAST_NAME columns and pick the values corresponding to matched positions in AGE column and keep forming the resultant tuples that can be sent back to client.

What did we do above?

• The query plan for this query touches 3 columns  — FIRST_NAME, LAST_NAME and AGE with predicate evaluation needed on AGE column.
• There is typically one decision that needs to be made — when do we form a tuple? In other words, when do we materialize values from different columns into a tuple?
• Two directions can be taken here:

Early Materialization

• As soon as query plan touches a column (SELECT list or for WHERE clause or any other operator), the column value is added to the tuple. Note that word “touches” is very important here.
• The operator evaluation or any other processing hasn’t happened yet on the column value. The fact that it is needed by the query plan for something will dictate the addition of column value to the tuple.
• We end up forming tuples that eventually get discarded simply because operator evaluation failed on one of the columns.

<John, Smith, 35>, <Ray, Lane, 45>. The first tuple will get discarded later on when WHERE clause is executed on AGE column value.

Instead, why not delay the materialization of column values and ensure we only form relevant tuples  — tuples that are actually part of result-set.

Late Materialization

• Take advantage of the columnar format and process the individual column first.
• For example, in a tight for-loop AGE column was scanned first to do predicate evaluation on all the column values.
• The output of SIMD scan was then fed through the scan of FIRST_NAME and LAST_NAME columns to pick only the values that are needed for result-set and form relevant tuples.

How did Late Materialization help us?

• We formed only relevant tuples and that too much later during execution. No CPU was wasted on forming tuples upfront without knowing if a particular tuple might eventually be a part of result-set or not.
• This is more efficient utilization of CPU-memory bandwidth.
• Leveraged the storage format (columnar) and applied vectorized query processing during predicate evaluation on a particular column. If the column values are going to be materialized into tuples sooner in the query plan, then we lose the opportunity to use vectorized instructions on a column simply because after materialization the data is no longer columnar and we need to work with tuples.
• Because materialization is done later in the plan, we quickly loop through the values of a column and applied necessary filters.
• This is far more efficient than a tuple based iterator that we need to work with in the case of early materialization since for each tuple, we feed the right column value (AGE) into the operator (WHERE) and decide if the tuple is needed or not. The tuple based iteration is not so “CPU-cache friendly” because the tuple contains “N” fields and all we need is just one field to pass to the operator repeatedly (in a loop)
• There are more advantages of Late Materialization in the context of Compression and they will be discussed in the next section.

5. Compression

Data stored in database tables is usually compressed to optimize disk space storage and I/O during query processing — reduced number of bytes (compressed) will be read off disk into memory and down through the storage hierarchy.

Compression algorithms operate better if the input data is somewhat related (less entropy) and this gives better compression ratios. Columnar format can take advantage of this fact and each column can be compressed individually with a compression scheme most suitable for that column. This advantage is however not available in row stores since a row contains data from multiple different types of columns. This doesn’t imply that compression algorithms are guaranteed to give better compression ratios on columnar data. It’s just that related (all column values) data has the potential to be highly compressible.

Different factors can be taken into account when deciding upon a compression scheme to be used for a particular column. Factors like column cardinality, data type, sorted or not etc are important column-level characteristics that can be used to decide the compression method.

The focus is not really on getting best compression ratios. Instead, we should think about schemes that enable faster processing (predicate evaluation, decompression etc) on column values. Columnar format allows us to use simple and efficient compression methods like Run Length Encoding (RLE), Dictionary Encoding, bit-vector encoding, delta encoding etc that may not give the best compression ratios but allow faster decompression.

As an example, dictionary encoding is usually a good choice if the column has low cardinality — limited number of possible unique values. The encoded values (dictionary indexes) can be stored in fewer bits. COUNTRY column in a table can be efficiently represented using dictionary encoding.

There are some advantages of using such light-weight compression methods:

• Compliment vectorized query processing by potentially packing the data (uncompressed column values) into fixed-width values in an array (compressed column values).
• SIMD instructions can then work on this well packed data and process the compressed column values.
• The encoding schemes mostly allow us to work on compressed column values and process them before decompressing everything upfront. This way we can decompress only the needed values — values that passed predicate filters.
• Obviously raw column data might already be fixed width — for example AGE column might be internally represented using an array of 4 byte integers. So without doubt, SIMD techniques will be effective on raw column data as well in this case.
• However, if we compress the AGE column using bit-packing (1 byte is enough to represent any value in AGE column), the processing actually becomes much faster because the number of column values that can be loaded (and parallelly processed) into SIMD register (typically 128 bits) are far more if the column is compressed.

Late materialization becomes more effective since the idea of processing the column values before forming tuples actually allows us to work on compressed columnar data. Once a given column has been processed and we need to form tuples, we can decompress only the necessary values.

We want to find out all the states/regions in England that had more than half million in sales.

SELECT STATE FROM SALES WHERE COUNTRY=’ENGLAND’ AND SALEVALUE>500000

1. COUNTRY column can be stored as dictionary encoded when the table was loaded with data.
2. Let’s say SALEVALUE column is internally stored in 4-byte fixed width values (4 billion max sales). I feel there is no harm in storing this column as raw uncompressed.
3. So the query executor will first probe the dictionary to get fixed-width encoded value for predicate ‘ENGLAND’. Say this is 4.
4. Compressed COUNTRY column can be quickly scanned using vectorized instructions.
   • Multiple (16 if each encoded value is stored in 1 byte) encoded column values can be loaded into a 128 bit SIMD register and parallelly compared with value 4.
5. SALEVALUE column can then be scanned with vectorized instructions.
   • 4 column values can be loaded into a 128 bit SIMD register and parallelly compared with value 500000.
6. Bitwise AND can be done on the output of two SIMD scans (COUNTRY and SALEVALUE).
7. The output of SIMD scans is typically a bit vector or mask indicating where the predicate passed or failed.
8. All we are trying to do now is AND the vectors and find where both predicates passed or failed
   • bit_vector R = SIMD_SCAN_COUNTRY & SIMD_SCAN_SALEVALUE
9. Now we can use the bit-vector R and scan through the STATE column to pick a column value at position “N” if “Nth” bit is set in R.

With this I would like to conclude the post. I agree that the post doesn’t dig into specific details about using compression, vectorized processing, query plan etc. However, the goal of this post is to just give a brief overview. Subsequent posts will take a deep dive into individual topics.

Data structures (and their corresponding methods) implemented with coarse grained locks are hard to scale in highly parallel environments and workloads where the structure is required to be accessed by several concurrent threads simultaneously (parallelism).

In this post, I will summarize my understanding of a non-blocking (lock free) design of linked list data structure as presented in paper “A Pragmatic Implementation of Non-Blocking Linked-Lists” by Tim Harris.

The paper presents the design (and pseudo code) of a lock free linked list data structure that supports concurrent insert(), delete() and search() operations without taking explicit locks/mutexes etc. The proposed algorithms leverage CAS (Compare and Swap) low level atomic instruction to implement lock-free methods on linked list.

CAS is an atomic instruction available on all modern processors. The post will use the following semantics of CAS.

bool CAS (memory location L, expected value V at L, desired new value V1 at L).

If the expected value at memory location L matches with the current value at L, CAS succeeds by storing the user desired value V1 at L and returns TRUE.

For the following discussion, we will assume that linked list is sorted and take the following simplest definition of a linked list node.

typedef struct node
{
   int element;
   struct node *next;
} node;

Note: The paper obviously proposes its own generic linked list type and node definitions. However, I feel the above mentioned simple definition is good enough to understand the main ideas (lock free algorithms) proposed in paper.  Everything else is just low level implementation detail which the reader shouldn’t really care about.

Insert Operation – Insert new node with value 7 in the linked list structure shown above.

The generic algorithm to insert a new node into a sorted linked list is:

• Traverse the linked list from head to find the largest node strictly smaller than  value 7 (due for insertion). Let this node be N1.
• The node pointed to by “next” pointer of N1 is N2.
• We allocate a new node N and insert it between N1 and N2 as follows:

node* N = (node *)malloc(sizeof(node));
N->element = 7;
N2 = N1->next;
N->next = N2;
N1->next = N;

The above code snippet will effectively change the structure of linked list by creating and allocating a new node N between N1 (node with value 6) and N2 (node with value 10). This operation (specifically the act of changing the “next” pointer of N1 to point to new node N) needs to be atomic. The paper suggests the following simple approach to implement insert() in an atomic manner using CAS:

N->next = N2;
ret = CAS(&(N1->next), N2, N);
if(ret)
{
    // CAS succeeded. N was successfully inserted between N1 and N2
    return true;
}
else
{
    // CAS failed.
    // retry
}

CAS will fail if the “current” value at memory location is changed concurrently.

• In the context of this example, the “current” value at memory location “N1->next” is address of node N2. So if some other thread beats us by concurrently inserting a new node between N1 and N2 , CAS will fail because the reference stored in “N1->next” will no longer point to node N2.
• For example, if some other concurrent insert operation successfully inserted a node with value 9 between nodes with value 6 and 10, our CAS operation to insert node with value 7 will fail. This scenario is shown in the following diagram (Please ignore nodes labelled as H and T. The reason for having those nodes will be explained very soon).

Because we are no longer taking explicit locks, there is no guarantee that insert operation will succeed. Therefore, CAS based algorithms are usually implemented using a loop (aka CAS loop) to retry the operation when CAS fails. The pseudo code for insert operation will now look like:

Assumption: The node to be inserted is not already in the list. This corner case is not addressed right now.

boolean insert(int ele)
{
  // CREATE AND INITIALIZE THE NEW NODE N (DUE FOR INSERTION)
  node *N = (node *)malloc(sizeof(node));
  N->element = ele;
  N->next = NULL;
  node *N1 = NULL; 
  node *N2 = NULL;
  // CAS LOOP
  do
  {    
     // SEARCH FOR THE ELEMENT, get references to N1 and N2 nodes
     N2 = search(&N1, element);
     // CHECK ELEMENT SHOULD NOT ALREADY EXIST - IGNORE THE CORNER CASE FOR NOW
    
 
     N->next = N2;

     // SWING THE POINTER OF PREDECESSOR - ACTUAL STEP THAT INSERTS THE NEW NODE
     if(CAS(&(N1->next), N2, N))
     {
       // CAS SUCCEEDED - INSERT DONE
       return TRUE;
     }
  }
  while(TRUE);
}

The insert() operation internally uses the search algorithm to traverse the linked list and get references to nodes N1 and N2. There are some important details missing from the above description and code example(s). I will get back to them after discussing delete() and search() operations as then it will make much more sense to discuss more details, questions and concerns.

The insert() operation on a linked list can be divided into three cases: (1) insertion before head, (2) insertion after tail and (3) insertion in the middle.

The CAS based approach takes care of all these cases. Until now what we have discussed is case (2) which deals with inserting a new node with value 7 between existing nodes with value 6 and 10 respectively.

We can address cases (1) and (3) by converting them to case (2). This is possible by keeping dummy head and tail nodes that contain no meaningful value. The paper calls these as sentinel nodes but doesn’t discuss the reason for having them in the linked list. However, I feel these dummy/sentinel nodes are needed to conveniently handle cases (that change head and tail) in the form of case (2). So with dummy nodes H and T, the state of linked list will be:

Note that H and T are not the logical “head” and “tail” nodes of linked list.

How does this help in handling cases (1) and (3) ?

• Inserting before head — effectively inserting a new node between dummy node H and current head node (node with value 2).
• Inserting after tail — effectively inserting a new node between current tail node (node with value 10) and dummy node T.

Delete Operation – Delete existing node with value 7.

The generic deletion algorithm from a sorted linked list is:

• Traverse the linked list from head to find the largest node strictly smaller than 7. Let this node be N1.
• The node pointed to by “next” pointer of N1 is N2 (due for deletion).

// REMOVE N2 FROM LIST
N1->next = N2->next;
free(N2);

CAS based approach:

N3 = N2->next;
// SWING THE "NEXT" POINTER OF PREDECESSOR NODE
ret = CAS(&(N1->next), N2, N3);
if(ret)
{
    // CAS succeeded. N2 was successfully removed from the list. 
    // "next" field in N1 now points to node N3.
    return true;
}
else
{
    // CAS failed.
    // retry
}

The successful CAS operation that swings the “next” pointer of N1 to point to N3 only makes sure that N2 (node being deleted) is indeed the “next” node of N1 at the time CAS is executed. For example, if a concurrent insert() operation inserted a new node (with value 6.5) between nodes 6 (N1) and 7 (N2), our CAS of “next” pointer of N1 will fail while we are trying to delete node 7 (N2). This is completely fine and correctness is ensured.

The paper presents a very interesting problem associated with concurrent insert() and delete() operations. The single CAS operation doesn’t detect any changes happening after node N2 (due for deletion). In other words, as part of delete() when we swing the “next” pointer of N1 to point to N3, there is really no guarantee that N3 is in fact the successor to N2.

For example, if a concurrent insert() operation successfully inserts a new node with value 9 between nodes with value 7 (N2) and 10 (N3) respectively, our CAS operation will still succeed but will lead to an inconsistent (and corrupt) state of linked list. The following diagram shows this. CAS succeeds and swings the “next” pointer of N1 (6) to point to N3 (10). However, a concurrent insert happened after N2 (7) and N3 (10) was no longer the true successor to N2. Linked list is corrupted because even though node with value 9 is a valid node in the list, we have no reference pointing to it. It should _not_ be missing from the list.

To solve this problem, the paper introduces the concept of “marking” (or tagging) the node X that is due for deletion. Tagging a node X prevents the insert() operation to insert a new node Y as a successor to X.

When we use tagging, deletion happens in two steps (two CAS operations) — marking and removing. We mark a node by setting the LSB in its “next” pointer. Note that pointers are always aligned. So the LSB is available for such use. If the LSB of 8 byte reference stored in “next” field of a node N is set, it implies that N is marked (and someone intends to delete it).

Assumption: The node to be deleted is already in the list. This corner case is not addressed right now.

boolean delete(int ele)
{
  // CAS LOOP
  do
  {
     node *N1 = NULL;
     node *N2 = NULL;
     node *N3 = NULL;

     // SEARCH FOR THE ELEMENT, N1, AND N2 nodes.
     N2 = search(&N1, element);
     // CHECK ELEMENT SHOULD ALREADY EXIST - IGNORE THE CORNER CASE FOR NOW
         
 
     N3 = N2->next;

     // STEP1 -- TAGGING, MARK THE "NEXT" POINTER OF N2 (THE NODE TO BE DELETED)
     if( CAS(&(N2->next), N2->next, MARK_POINTER(N2->next)))
     {
         // IF WE ARE HERE, MARKING WAS SUCCESSFUL -- LSB set
         // STEP 2 -- SWING THE "NEXT" POINTER OF PREDECESSOR NODE TO REMOVE N2
         ret = CAS(&(N1->next), N2, N3);

         if(ret)
         {
             // CAS SUCCEEDED. N2 was successfully removed from the list. 
             // "next" field in N1 now points to node N3.
             return true;
         }
      }
   }
   while (TRUE);
}

Using the two-step approach for removing a node, the “linked list corruption” problem is solved because tagging node N2 (7) will prevent the concurrent insert of node with value 9 as a successor to N2.

The success of first CAS in the delete() algorithm ensures that node 7 has been marked for deletion (intent to delete) and any other concurrent insert() operation will not be able to insert a new node after node 7. Since the second CAS is executed only if the first CAS succeeds, it guarantees that the pointer swing (N1->next = N3) is not going to make the list inconsistent.

One of things that is confusing from the paper is the memory reclamation. The paper mentions that node removal comprises of two steps (1) logical deletion when the node is tagged by first CAS and (2) physical deletion when the “next” pointer of predecessor node is swung by second CAS.

I disagree here. Physical removal of the node is somewhat related to memory reclamation. The node is still there in a valid memory location because no free() or delete() call has been made to release the memory back to heap. Step (2) only ensures that node is not part of the list and in my opinion this is completely different from actually reclaiming the memory associated with the node.

It turns out that they use some reference counting based algorithm for memory reclamation to ensure that memory is reclaimed only after it is guaranteed that _no_ readers are holding any references to those nodes.

I haven’t discussed the full paper yet. Since I don’t like very long posts, I will cover the search() algorithm and various concurrency scenarios on all three (insert, search and delete) algorithms in my follow-up posts. Stay tuned

I would highly recommend reading the paper. It is a wonderful read.

• A read operation R on data object X is not guaranteed to return the value of most recent completed write operation W on data object X.
• Alternatively, in a distributed system based on eventual consistency, it is possible for the readers to see stale (not the most recent) copy of data.

Eventually all the nodes (replicas) storing copy of object X will converge to a single consistent copy and once this happens, readers will start seeing the most recent copy of data regardless of the node handling the read operation. Since such a state of the system is not guaranteed to be true at all instances of time, the semantics of consistency are “eventual” or “relaxed” implying that system will eventually reach such state.

Such consistency semantics are preferred to increase availability and lower the operation latency. Relaxed consistency semantics do not require the update operation to be visible/applied on each replicas of the data object. The system takes care of converging or syncing all the replicas in background (asynchronously) to a single consistent copy. Special algorithms have to be implemented to do this. The fact that this doesn’t happen on the critical path (client request) is the fundamental reason for improved availability and performance of the system since the request can complete even when some nodes are down (high availability) and the update operation can be potentially completed by executing on fewer replicas (low latency).

Discussing this in the context of CAP theorem, systems that exhibit eventual consistency semantics are typically classified as “AP” systems. However, this classification gives an impression that such systems prefer availability over consistency only in the presence of network partitions, faults etc. In my opinion, this is not always true. Distributed systems based on the principles of eventual consistency are “relaxed” about consistency requirements right from the beginning typically because they care more about high availability and low latency. They really don’t have to wait for the first fault to happen and then change the consistency semantics of the system.

R, W and N – These are configuration parameters typically used to set the consistency levels of distributed system. In most cases (Cassandra and Riak), they are tunable and thus developers writing applications on top of distributed data stores can set these to appropriate values depending on the use cases.

• N – Replication Factor
• W – Minimum number of replica nodes that should acknowledge a write request  for the system to declare the operation as successful.
• R – Minimum number of replica nodes that should acknowledge a read request for the system to declare the operation as successful.

Generally speaking, if the system has configuration R + W > N, we get strong consistency since the replica sets participating in read and write operations will always overlap (QUORUM). Thus a “successful” read request is guaranteed to return the result of most recent write since the request will always touch at least one replica that has the most recent copy of object.

Let’s take a simple system for discussion. The following diagram shows a 3-node distributed system.

N = 3, R = 2, W = 2 ; R+W > N

Write (item X, value V)

• A write operation is typically sent to all nodes regardless of the value of W. W controls the “minimum” number of successful responses that should be received by the coordinator node before it can return back to client with “SUCCESS”.
• Let’s say the item was successfully written on nodes N1 and N3 but the coordinator node couldn’t receive a response from N2.
• Since write quorum (W is 2) is anyways met, the update is successful.
• It is not absolutely necessary for the coordinator node to be a part of replica set. It can be an entirely different node in the cluster.
• In this example, N1 coordinates the update operation for item X.

Read (item X)

• Any subsequent successful read operation will touch at least two (read quorum) nodes from set {N1, N2, N3}.
• Hence it will see the latest value of item X regardless of the fact that latest update to item X is not yet propagated/replicated to N2.
• The distributed system’s implementation of read algorithm should have a way to ensure that copy present on N1/N3 is the latest one and N2 doesn’t have the change yet (or has the outdated change).
• How exactly this happens will not be discussed in this post since this topic deserves good details and will be covered in follow-up posts.

Now let’s consider a different state of the system to demonstrate eventual consistency. Let’s say all replicas N1, N2 and N3 store value V for object X to begin with. Consider this as the starting state for the sake of discussion.

Let’s take N = 3, W = 2, R = 1 ; R + W <= N

Write (item X, value V1) 

• Nodes N1 and N2 meet write quorum and update the value of item X to V1.
• Node N3 still has the old copy (value V).

Read (item X)

• Let’s say the operation was coordinated by N3.
• Since N3 is also one of the replicas for item X and R is 1, it returns “SUCCESS” with result as V — client sees a stale value.
• This is the point where we see the effects of eventual consistency — a read() doesn’t see the changes of most recent write() simply because the latest changes haven’t yet been propagated to node(s) handling the read request.
• On the other hand, if N1 coordinates the read, the client will see the latest changes.

This behavior is the fundamental characteristic of an eventually consistent system. It is strikingly different from what we saw earlier in a strongly consistent system (R+W > N) — the set of nodes required to participate in a successful read operation will have at least one node with the latest copy of data.

Consider the same system with R as 2 instead of 1. All readers see the latest changes and we again get a strongly consistent system with R + W > N.

For a strongly consistent system with W = N, R can be 1 — any replica can serve the read request and is guaranteed to have the most recent copy of data. But this configuration increases the latency of writes since the operation will touch all nodes. It also reduces availability since the write operation will fail even if one node is down.

Highly Available Configuration with Eventual Consistency

W=1 and R=1 (W + R <= N) has the potential to be a highly available system with low latency operations. A read and write will succeed as long as even one node is available in the system.

In my follow up posts, I will talk more about consistency semantics, update conflicts, asynchronous replication and time-stamps. Stay tuned

When implementing a collision resolution strategy for our hash table design, there are usually two directions that we take:

• Chaining – There is a per bucket overflow chain (a linked list of overflow buckets) containing items that earlier hashed to main bucket in the hash table but couldn’t be stored there because the bucket was already full (collision). Thus each time we hit such a condition on a given bucket B, we create a new overflow bucket (effectively creating a new node on linked list), and insert it at the head/tail of overflow chain of bucket B. The new item is then stored in this overflow bucket. Such hashing schemes are often referred to as closed addressed hashing or out-of-line hashing.
• Probing – Unlike the chaining approach, we can also design an inline strategy to handle collisions. In such design, there is no overflow chain on any bucket. The main hash table buckets are the only buckets we have at our disposal. So whenever there is a collision on bucket B, we simply probe in a linear order (starting from bucket B and peeking at B+1, B+2 …..) for the next vacant bucket B’ where the new item can be inserted. Such hashing schemes are often referred to open addressed hashing or inline hashing or linear probing.

Whatever the collision resolution mechanism is, the primary goal should be to keep the performance of get() and put() operations optimal. Some applications can tolerate a slow write (put operation on hash table) code path, but fast lookups is a requirement that should be supported by data stores that use some sort of hashing techniques (key value stores, hash indexes in databases etc).

Under what conditions can the performance of operations on hash table deteriorate ?

In almost all of the cases, load factor (aka occupancy percentage) has a significant impact on the performance of get() and put() operations. Load factor is the ratio of number of items (or records) in the hash table to the total number of buckets in the hash table array. It is basically an indicator of the degree of fullness of the hash table.

Consider the probing based hash tables. Study [see links below] shows that such hash table can show some performance degradation for both get() and put() operations even at a load factor of 50% (around half the buckets in hash table have items, and half are empty). This is because of “probing” which is required in the presence of collisions. If a key K hashes to bucket B and B is already occupied by some other key K1, the put() algorithm will linearly go over every subsequent bucket looking for an empty bucket which can hold key K. Any subsequent get() on key K will also have to do the same thing.

When linear probing based hashing schemes hit these situations often, the hash table is resized. This is so that performance of hash table operations  doesn’t degrade significantly. Although resizing will possibly provide optimal performance (because number of buckets will increase), this approach is not space efficient. The fact that hash table’s performance shows degradation even at low fill factors (hash table is 50% full) and forces a resize is a straight forward evidence that hash table is not memory efficient. Resizing (allocating more buckets or a bigger hash table) isn’t really a good idea when 40%-50% buckets are already empty in the existing hash table.

Similar problem can be understood for chaining based hash tables. On every collision, we create a  new overflow bucket thus effectively increasing the size of overflow chain and every subsequent get()/lookup operation has to traverse down the overflow chain on hash bucket to look for the presence of a Key-Value pair. The longer the size of chain, the slower our get() operation will be. Again, the hash table will be resized based on some internal statistics (average length of overflow chain etc etc).

It is possible to resize the hash table _purely_ on the basis of load factor thresholds. Let’s say we keep a threshold of 75%. Internally we need to track how many items have been inserted into the hash table. Once the hash table is 75% full, we trigger a resize operation that allocates a larger hash table and rehashes the existing items. This is fine as long as space efficiency is concerned. For example, if we start with hash table of size 16, with a load factor threshold of 75% we will resize when at least 12 buckets are full. This is reasonable.

However, in most cases this not really a good idea because it is like waiting for the hash table’e performance to degrade significantly and then doing something about it. If the collisions are happening (which implies longer overflow chains for chaining based tables and longer probe lengths for linear probing based tables), the slowdown in get() and put() operations will start becoming apparent. Most implementations of hash table will resize sooner than later and this is where the space inefficiency concern mentioned above is reinforced.

Nowadays, there are many hash table designs that use associativity in hash buckets to make them space efficient — implying performance of hash table operations is optimal (or at least acceptable/reasonable) even at high load factors. The concept of associativity is not really new. Since ages it has been used in the design of CPU caches (full-associative, 4-way set associative etc). In fact, the adoption of this design principle in hashing is also not new.

The fundamental idea is to use a fat bucket. Regular/traditional hash table designs are describes with the view that each bucket stores a single KV item. With associativity, each bucket is capable of storing “N” items — implying the hash table is N way set associative. Here are the advantages:

• For chaining based tables, we create an overflow bucket only when all the “N” slots in current bucket are full and there is “N+1″th collision. Thus we can possibly keep the length of chain shorter because each bucket now stores multiple items. The shorter the chain, the lesser the number of pointer de-references.
• Each read of a hash bucket gives us N items to look at whereas earlier we had to de-reference the pointer on overflow chain to look at every other item.
• Similarly, the probe lengths tend to be shorter for linear probing based hash tables because each bucket now stores “N” KV items.
• The performance of both get() and put() operations is not expected to degrade at low fill factors.
• We should try to align our fat hash bucket as per the size of CPU cache line (64 bytes on x86). This will also increase the performance by reducing cache misses as we move from one KV item to another.

Note that associativity is not an alternative to “resizing”. Hash table will eventually have to be resized. It is a typical Computer Science problem — What do we do when we exhaust the existing space taken by a particular data structure? The data structure  has to be resized. There is no doubt in that. The point I am trying to make is that  we should try to refrain from resizing the hash table at low load factors. Instead, the slowdown in get() and put() operations should be addressed by changing the design of hash table — which is what associativity does.

So it is perfectly fine to resize the hash table based on load factor thresholds (something like 75%, 80%). But we should try to use the “associative bucket” approach so that hash table’s performance remains reasonable as the load factors begins to increase from 0 to 20%, 30%, 50% and so on and we don’t keep on resizing at low occupancy rates.

The concept is being widely used in hash tables based on Cuckoo Hashing. Here are few useful links:

• A cool and practical alternative to traditional hash tables
• Efficient Hash Probes on Modern Processors>