Clock-Bound Wait

Read Restart

200ms is too high an interval to wait for every write request. This is why databases like CockroachDB or YugabyteDB implement a check in the read requests instead.

While serving a read request, cluster nodes check if there is a version available in the interval of readTimestamp and readTimestamp + maximum clock drift. If the version is available - assuming the reader's clock might be lagging - it is then asked to restart the read request with that version.

class KVStore…

  public void put(String key, String value) {
      HybridTimestamp writeTimestamp = clock.now();
      kv.put(new HybridClockKey(key, writeTimestamp), value);
  }

  public String get(String key, HybridTimestamp readTimestamp) {
      checksIfVersionInUncertaintyInterval(key, readTimestamp);
      return kv.floorEntry(new HybridClockKey(key, readTimestamp)).getValue();
  }

  private void checksIfVersionInUncertaintyInterval(String key, HybridTimestamp readTimestamp) {
      HybridTimestamp uncertaintyLimit = readTimestamp.add(maxOffset, 0);
      HybridClockKey versionedKey = kv.floorKey(new HybridClockKey(key, uncertaintyLimit));
      if (versionedKey == null) {
          return;
      }
      HybridTimestamp maxVersionBelowUncertainty = versionedKey.getVersion();
      if (maxVersionBelowUncertainty.after(readTimestamp)) {
          throw new ReadRestartException(readTimestamp, maxOffset, maxVersionBelowUncertainty);
      }
      ;
  }

class Client…

  String read(String key) {
      int attemptNo = 1;
      int maxAttempts = 5;
      while(attemptNo < maxAttempts) {
          try {
              HybridTimestamp now = clock.now();
              return kvStore.get(key, now);
          } catch (ReadRestartException e) {
              logger.info(" Got read restart error " + e + "Attempt No. " + attemptNo);
              Uninterruptibles.sleepUninterruptibly(e.getMaxOffset(), TimeUnit.MILLISECONDS);
              attemptNo++;
          }

      }
      throw new ReadTimeoutException("Unable to read after " + attemptNo + " attempts.");
  }

In the Alice and Bob example above, if there is a version for "title" available at timestamp 2, and Alice sends a read request with read timestamp 1, a ReadRestartException will be thrown asking Alice to restart the read request at readTimestamp 2.

Read restarts only happen if there is a version written in the uncertainty interval. Write request do not need to wait.

It’s important to remember that the configured value for maximum clock drift is an assumption, it is not guaranteed. In some cases, a bad server can have a clock drift more than the assumed value. In such cases, the problem will persist.

Using Clock Bound APIs

Cloud providers like Google and Amazon, implement clock machinery with atomic clocks and GPS to make sure that the clock drift across cluster nodes is kept below a few milliseconds. As we’ve just discussed, Google has True Time. AWS has AWS Time Sync Service and ClockBound.

There are two key requirements for cluster nodes to make sure these waits are implemented correctly.

• The clock drift across cluster nodes is kept to a minimum. Google's True-Time keeps it below 1ms in most cases (7ms in the worst cases)
• The possible clock drift is always available in the date-time API, this ensures programmers don't need to guess the value.

The clock machinery on cluster nodes computes error bounds for date-time values. Considering there is a possible error in timestamps returned by the local system clock, the API makes the error explicit. It will give the lower as well as the upper bound on clock values. The real time value is guaranteed to be within this interval.

public class ClockBound {
    public final long earliest;
    public final long latest;

    public ClockBound(long earliest, long latest) {
        this.earliest = earliest;
        this.latest = latest;
    }

    public boolean before(long timestamp) {
        return timestamp < earliest;
    }

    public boolean after(long timestamp)   {
        return timestamp > latest;
    }

As explained in this AWS blog the error is calculated at each cluster node as ClockErrorBound. The real time values will always be somewhere between local clock time and +- ClockErrorBound.

The error bounds are returned whenever date-time values are asked for.

public ClockBound now() {
    return now;
}

There are two properties guaranteed by the clock-bound API

• Clock bounds should overlap across cluster nodes
• For two time values t1 and t2, if t1 is less than t2, then clock_bound(t1).earliest is less than clock_bound(t2).latest across all cluster nodes

Imagine we have three cluster nodes: Green, Blue and Orange. Each node might have a different error bound. Let's say the error on Green is 1, Blue is 2 and Orange is 3. At time=4, the clock bound across cluster nodes will look like this:

In this scenario, two rules need to be followed to implement the commit-wait.

• For any write operation, the clock bound's latest value should be picked as the timestamp. This will ensure that it is always higher than any timestamp assigned to previous write operations (considering the second rule below).
• The system must wait until the write timestamp is less than the clock bound's earliest value, before storing the value.

  This is Because the earliest value is guaranteed to be lower than clock bound's latest values across all cluster nodes. This write operation will be accessible to anyone reading with the clock-bound's latest value in future. Also, this value is guaranteed to be ordered before any other write operation happen in future.

class KVStore…

  public void put(String key, String value) {
      ClockBound now = boundedClock.now();
      long writeTimestamp = now.latest;
      addPending(writeTimestamp);
      waitUntilTimeInPast(writeTimestamp);
      kv.put(new VersionedKey(key, writeTimestamp), value);
      removePending(writeTimestamp);
  }


  private void waitUntilTimeInPast(long writeTimestamp) {
      ClockBound now = boundedClock.now();
      while(now.earliest < writeTimestamp) {
          Uninterruptibles.sleepUninterruptibly(now.earliest - writeTimestamp, TimeUnit.MILLISECONDS);
          now = boundedClock.now();
      }
  }


  private void removePending(long writeTimestamp) {
      pendingWriteTimestamps.remove(writeTimestamp);
      try {
          lock.lock();
          cond.signalAll();
      } finally {
          lock.unlock();
      }
  }

  private void addPending(long writeTimestamp) {
      pendingWriteTimestamps.add(writeTimestamp);
  }

If we return to the Alice and Bob example above, when the value for "title"- "After Dawn" - is written by Philip on server Green, the put operation on Green waits until the chosen write timestamp is below the earliest value of the clock bound. This guarantees that every other cluster node is guaranteed to have a higher timestamp for the latest value of the clock bound. To illustrate, considering this scenario. Green has error bound of +-1. So, with a put operation which starts at time 4, when it stores the value, Green will pick up the latest value of clock bound which is 5. It then waits until the earliest value of the clock bound is more than 5. Essentially, Green waits for the uncertainty interval before actually storing the value in the key-value store.

When the value is made available in the key value store, that the clock bound's latest value is guaranteed to be higher than 5 on each and every cluster node. This means that Bob's request handled by Blue as well as Alice's request handled by Amber, are guaranteed to get the latest value of the title.

We will get the same result if Green has 'wider' time bounds. The greater the error bound, the longer the wait. If Green's error bound is maximum, it will continue to wait before making the values available in the key-value store. Neither Amber nor Blue will be able to get the value until their latest time value is past 7. When Alice gets the most up-to-date value of title at latest time 7, every other cluster node will be guaranteed to get it at it's latest time value.

Read-Wait

When reading the value, the client will always pick the maximum value from the clock bound from its cluster node.

The cluster node that is receiving the request needs to make sure that once a response is returned at the specific request timestamp, there are no values written at that timestamp or the lower timestamp.

If the timestamp in the request is higher than the timestamp at the server, the cluster node will wait until the clock catches up, before returning the response.

It will then check if there are any pending write requests at the lower timestamp, which are not yet stored. If there are, then the read requests will pause until the requests are complete.

The server will then read the values at the request timestamp and return the value. This ensures that once a response is returned at a particular timestamp, no values will ever be written at the lower timestamp. This guarantee is called Snapshot Isolation

class KVStore…

  final Lock lock = new ReentrantLock();
  Queue<Long> pendingWriteTimestamps = new ArrayDeque<>();
  final Condition cond  = lock.newCondition();

  public Optional<String> read(long readTimestamp) {
      waitUntilTimeInPast(readTimestamp);
      waitForPendingWrites(readTimestamp);
      Optional<VersionedKey> max = kv.keySet().stream().max(Comparator.naturalOrder());
      if(max.isPresent()) {
          return Optional.of(kv.get(max.get()));
      }
      return Optional.empty();
  }

  private void waitForPendingWrites(long readTimestamp) {
      try {
          lock.lock();
          while (pendingWriteTimestamps.stream().anyMatch(ts -> ts <= readTimestamp)) {
              cond.awaitUninterruptibly();
          }
      } finally {
          lock.unlock();
      }
  }

Consider this final scenario: Alice's read request is handled by server Amber with error bound of 3. It picks up the latest time as 7 to read the title. Meanwhile, Philip's write request is handled by Green (with an error bound of +-1), it picks up 5 to store the value. Alice's read request waits until the earliest time at Green is past 7 and the pending write request. It then returns the latest value with a timestamp below 7.