Use CompletableFuture to avoid using a "blocking" await

Question

This is a blocking system (synchronous). The idea behind it is:

If our webapp receives N identical requests (N different Tomcat http requests that generate the same 'cache id'), just the first is sent to the api server, the rest are blocked on this 'cache id'.
When this first request ends up and the response is received from the api server, this response is stored on our cache.
Right after this, the 'unlock' method is called to wake up all threads blocked on this 'cache id' (the rest N-1 requests).
This N-1 requests are awaken and they consult the cache, finding the response for this 'cache id', so they are not sent to the api server and so this api saves to process them.

This is the method (simplified) that tries to get a document from cache.

public dtoCache getDocumentFromCache(String cacheId)
{
 dtoCache objCache = CacheFacade.getInstance().getCacheEntry(cacheId);

 // If document not found in cache ...
 if (objCache == null)
 {
  // If I'm the first, just return null (and the process to send the request 
  // to the server will start outside this method). Otherwise block on this method.
  if (!this.objLockAndAwait.lock(cacheId)) return null;

  // At this point the first request with this 'cacheId' saved the document on cache, 
  // so after awaking go to cache again because we will find the document there.
  return (this.getDocumentFromCache(cacheId));  
 }

 return (objCache);
}

The 'lock' method is very simple, just consult if the 'mapCacheId' contains the 'cacheId' key.

public boolean lock(String cacheId)
{
  // Get a 'lockObject' for this 'cacheId' and acquire it.
  LockObject lockObject = this.stripedLock.getLockObject(cacheId);
  lockObject.lock();

  try
  {
   // Check if this 'cacheId' is going to the server right now. Two cases:
   // 1. If the map does not contain this 'cacheId' --> don't block, I'm the 
   // first --> return false and continue
   // 1. If the map contains this 'cacheId' --> block on it.
   if (this.mapCacheId.putIfAbsent(cacheId, (byte)0) != null)
   {
    lockObject.await(cacheId);
    return true;
   }

   return false;
  }
  finally { lockObject.unlock(); }
}

And this is the 'unlock' method, which is called every time a request comes from the server. It signals on 'cacheId' to wake up all threads blocked on it.

public void unlock(String cacheId)
{
  LockObject lockObject = this.stripedLock.getLockObject(cacheId);
  lockObject.lock();

  try
  {
   lockObject.signalAll(cacheId);
   this.mapCacheId.remove(cacheId);
  }
  finally { lockObject.unlock(); }
}

Now I need to rewrite this code to make it asynchronous. I don't want the threads to be blocked on the await condition, if one thread must be blocked because the 'cache id' is found in the 'mapCacheId', I need this thread to be released to take another task, and retake the code in a callback only when it's notified that the request sent to the server finished.

I'm checking the 'CompletableFuture', because it seems the solution to this, but after playing with it for some days I'm not able to see how to modify the code to make it asynchronous with CF. It seems I need this N-1 CF to wait for the completion of the first one, but the CF api is complicated and I couldn't find any similar examples.

On the first thread, easiest is to just block it. On subsequent requests, if a cache population is in progress, call ServletRequest.startAsync(), store the AsyncContext somewhere (perhaps a threadsafe queue) and return from the request service method - this releases the http worker thread. When the first request completes, simple finish all the other queued requests by calling writing the response and calling AsyncContext.complete(). — Boris the Spider
This logic can be generalized by starting, or taking from a pool, a cache worker thread and placing all requests - including the first - in a queue by the above method. CompletableFuture can be used to implement this background cache worker thread logic. But the key to releasing http worker threads is the startAsync method - I suggest you read up on it. — Boris the Spider
That's great Boris. I was already using ReadListener and WriteListener not to block on read/write io operations, but I hadn't though about using req.startAsync() for this. This will clearly solve my problem. Thanks a lot. — user8398650

Alexei Kaigorodov Alexei Kaigorodov · Accepted Answer · 2018-08-19T03:17:01

let class DtoCache extends CompletableFuture<Document>. Then

// webapp receives a request
public void handleRequest(Request request) {
   String cacheId = request.getCacheId();
   boolean first = false;
   DtoCache objCache;
   synchronized(cache) {
       objCache = cache.get(cacheId);
       if (objCache == null) {
           first = true;
           objCache = new DtoCache();
           cache.put(cacheId, objCache);
       }
   }
   if (first) {
       // only the first request asks the Api server
       Document doc = askApiServer(cacheId);
       // reply to our request
       request.reply(doc);
       // reply to the other requests
       objCache.complete(doc);
   } else {
      // arrange an asynchronous reply to our request
      objCache.thenApply(request::reply);
   }

Use CompletableFuture to avoid using a "blocking" await

1 Answers