4 minute read

Categories:
Tags:

For many data services, any easy way to reduce database load is to cache calls to semi-static data (ie: append-only, or refreshed only on a set schedule), and very recent calls due to backward user navigation. Not all methods and data are suitable for caching, so any implementation will require the ability to be selective.

Using Finagle’s Filters we can configure our caching in a central place, leaving individual method implementations clean of the concern.

The heavy lifting of our cache will be handled by Google Core Libraries: Guava, leaving us to sort out the small details.

Let’s start off writing a generic abstract filter class that will allow us to specify what method calls we want to cache. Here we are using an Option to allow us to cache all method calls if we pass in None.

import com.twitter.finagle.{ Service, SimpleFilter }
import com.twitter.util.Future
import com.google.common.cache.Cache
 
abstract class AbstractCacheFilter(val methodsToCache: Option[Seq[String]] = None) 
  extends SimpleFilter[Array[Byte], Array[Byte]] {
 
  import com.google.common.hash.Hashing
  import org.apache.commons.codec.binary.Base64
 
  /** Hash of request to use as key in cache. */
  def requestHashKey(request: Array[Byte]): String =
    Base64.encodeBase64String(Hashing.md5.hashBytes(request).asBytes)
 
  val cache: Cache[String, Array[Byte]]
 
  def apply(request: Array[Byte], 
    service: Service[Array[Byte], Array[Byte]]): Future[Array[Byte]]
}

The abstract cache has been keyed using a String, and requestHashKey will be used to uniquely identify all request queries. The question is: are identical queries represented by the same bytes? The answer is, in the case of Thrift, no – all requests are uniquely identified using a random SeqId. The SeqId is a unique Int32 assigned to each request, allowing multiple requests to be pipelined over the same network connection. But it isn’t a serious problem, we know how to find it in the protocol, and thankfully it’s easy to change.

Finagle uses the TBinaryProtocol, so all of this code also makes that assumption. The SeqId is a 4 byte integer, and is the third field in the binary protocol: right after the version and method name. We will need the ability to change this value in both the requests, and the responses. We’ll choose to store all requests/responses in our cache with SeqId=0, and then change them in any cached response before it’s returned to the client.

def bytesToInt(bytes: Array[Byte]): Int = java.nio.ByteBuffer.wrap(bytes).getInt
 
def binaryProtocolChangeSeqId(requestOrResponse: Array[Byte], 
  seqId: Array[Byte] = Array(0, 0, 0, 0)): Array[Byte] = {
 
  val methodNameLength = bytesToInt(requestOrResponse.slice(4, 8))
  val retVal = new Array[Byte](requestOrResponse.length)
  requestOrResponse.copyToArray(retVal)
  seqId.copyToArray(retVal, 8 + methodNameLength, 4)
  retVal
}

We can similarly discover the SeqId in any client request:

def binaryProtocolMethodNameSeqId(request: Array[Byte]): (String, Array[Byte]) = {
  val methodNameLength = bytesToInt(request.slice(4, 8))
  val methodName = new String(request.slice(8, 8 + methodNameLength), "UTF-8")
  val seqId = request.slice(8 + methodNameLength, 12 + methodNameLength)
  (methodName, seqId)
}

We now have code to peek into request method names, and to set/zero-out uniquely identifying SeqIds. But before we can write our caching code, we need to consider the case of Exceptions. It would be unwise to store exceptions into our cache, so let’s dig further into the Thrift protocol. Thrift allows exceptions to be sent back as part of a valid response, and any service methods allowing exception responses have a single byte exit status field.

/** Non-zero status is a thrown error */
def binaryResponseExitStatus(response: Array[Byte]): Byte = {
  val methodNameLength = bytesToInt(response.slice(4, 8))
  //skip over: version,methodLength,method,seqId,msgType,successStruct
  val positionMessageType = 4 + 4 + methodNameLength + 4 + 1 + 1
  response(positionMessageType)
}

There is a msgType field as part of the protocol which also indicates exceptions, however it is only used for unhandled exceptions. Any unhandled exception should occur before our caching code and terminate execution, as a result the rest of our code would not be executed. So we do not have to worry about handling these exception types within our code.

And finally, our apply method:

def apply(request: Array[Byte], 
  service: Service[Array[Byte], Array[Byte]]): Future[Array[Byte]] = {
 
  val (methodName, seqId) = binaryProtocolMethodNameSeqId(request)
  //cache all, or those specified
  if (!methodsToCache.isDefined || methodsToCache.get.contains(methodName)) {
    //key using a SeqId == 0
    val key = requestHashKey(binaryProtocolChangeSeqId(request))
    val cachedResult = cache.getIfPresent(key)
    if (cachedResult == null) {
      service(request).onSuccess(r => {
        //store with SeqId == 0, if clean exit
        if (binaryResponseExitStatus(r) == 0) 
          cache.put(key, binaryProtocolChangeSeqId(r))
      })
    } else {
      //change the SeqId to match request
      Future.value(binaryProtocolChangeSeqId(cachedResult, seqId))
    }
  } else service(request)
}

With our abstract caching filter complete, let’s create some example implementations to illustrate how to handle different scenarios and put to use some of Guava’s features. Guava is well documented, and there is a set of slides for the impatient.

Most eventually consistent datasets will benefit from having their recent read queries cached without having to worry about cache invalidation, as long as their total expiry time is less than the eventual consistency guarantee.

import com.google.common.cache.CacheBuilder
import com.twitter.conversions.time._
import com.twitter.util.Duration
 
class ExpiryCache(
  val slidingExpiry: Duration = 60 seconds,
  val maxExpiry: Duration = 10 minutes) 
  extends AbstractCacheFilter(None) {
 
  import java.util.concurrent.TimeUnit.SECONDS
  val cache = CacheBuilder.newBuilder()
    .expireAfterAccess(slidingExpiry.inUnit(SECONDS), SECONDS)
    .expireAfterWrite(maxExpiry.inUnit(SECONDS), SECONDS)
    .build[String, Array[Byte]]()
}

Next, we will take advantage of the weighting functionality in Guava to make a fixed memory size cache for our static data. Only a small subset of our read queries are for static, historical data, so eligible methods will be specified by name.

class FixedSizeCache(methodsToCache: Seq[String], val maxSizeMegabytes: Int = 100) 
  extends AbstractCacheFilter(Some(methodsToCache)) {
 
  import com.google.common.cache.Weigher
  val weight = maxSizeMegabytes * 1048576
  val weigher = new Weigher[String, Array[Byte]]() {
    def weigh(k: String, g: Array[Byte]): Int = g.length
  }
  val cache = CacheBuilder.newBuilder()
    .maximumWeight(weight).weigher(weigher)
    .build[String, Array[Byte]]()
}

As a final note, an assumption has been made that the Array[Byte] is encoded TBinaryProtocol. Alternatively, a TProtocolFactory could have been supplied in the AbstractCacheFilter constructor, and fields such as SeqId could be obtained by calling readMessageBegin. This alternative approach is revisited later in Thrift Client-Side Caching to Speed Up Unit Tests.

Full Sources: Finagle-Query-Cache-with-Guava.scala

Categories:
Tags:
Updated:

You May Also Enjoy

Job Queue Execution Management using ZIO Scopes

8 minute read

, ,

Job Queues are critical parts of Enterprise workloads. Complex queues use distributed nodes, state machines, and complex scheduling to trigger and track running jobs. But when simplicity allows the best approach is to create small idempotent jobs. The smaller the unit of work the easier progress can be tracked, jobs can be restarted or rerun with minimal waste, composability and reuse are increased, and logic is easier to reason about. These are the same arguments for Functional Programming and their Effect Systems, such as ZIO. Effect systems are congruent to the enterprise job queue, with ZIO fibers performing work and ZIO Resource Management forming the scheduling and supervision backbone. An efficient job queue can be written using ZIO constructs using surprisingly minimal amount of code. Table of Contents ZIO Resources and Scope Queue Features Class Interface ZIO Concurrency uses Fibers, Not Threads ...

Realtime Client Database using gRPC Bi-Directional Streams and ZIO Hub

1 minute read

, ,

Realtime pushed-based databases such as Google Firebase are a convenient way to ensure clients have the most recent data locally. Data updates are automatically streamed to clients immediately as they happen, or in the case of a client disconnect, immediately after reconnecting. gRPC server streaming and ZIO Hub allow this functionality to be easily replicated and customized beyond what expensive paid-for services such as Firebase can do.

Http4s Streams and Multipart Form-Data File Uploads

10 minute read

, ,

Streaming is the primary mechanism to reduce memory requirements for processing large datasets. The approach is to view only a small window of data at a time, allowing data to stream through in manageable amounts matching the data window size to the amount of RAM available. A practical example is a file-upload, where multi-GBs file streams can be handled by MBs of server RAM. However, enforcing streaming in software code is prone to errors, and misuse or incompatible method implementations will lead to breaking stream semantics, and ultimately to OOM exceptions. This article focuses on streams within the context of file uploads, using the Http4s library for examples.

Compiling Scala Native in a GitHub Action; Alternatives to GraalVM

11 minute read

, ,

Scala Native is a compiler and JDK written in Scala with the goal of removing Scala’s dependency on the JVM. This isn’t meant to achieve a higher performance such as with JDKs, and it is targeting a specialized use-case not considered to be today’s typical Scala development. Its competitors are Rust and Go, not GraalVM, Java or Kotlin. This article goes through common steps and challenges encountered when compiling Scala Native for linux with a GitHub Action.