Threaded batching algorithm

Question

Ok, I'm working on DynamoDb-based caching. We have an API where each call costs us real money and the information retrieved is "fresh" for 2-3 weeks, so it makes sense to cache it. There's no problem with calling dynamoDbClient.get(apiUrl) and retrieving the stored JSON response. They have batches - 25 batch items for PUT and 100 batch items for GET. So it means if we need get JSON response for 100 items we do one call instead of 100.

The question I have is how to organize this in the best way for threading.

Here's general idea and I'm open to suggestions.

Assuming we have a batch call with 100 items, we can use a BlockingQueue<String> to store the keys which we can use later. We would use blockingQueue.put() method, to make other threads wait if 150 threads arrived into method having a batch of only 100 available.

So thread that enters method, gets his place in batch queue, needs some "locking" mechanism to wait until response arrives and "wake him up" that his response is available.

pseudoCodeMethod(String resourceUrl) {
  blockingQueue.put(resourceUrl); // sleep if no place
  LockableResponse lockableResponse = getLockableResponse();
  lockableResponse.setKey(resourceUrl);
  lockableResponse.getSemaphore().acquire();

  String jsonApiResponse = lockableResponse.getJsonApiResponse();
  lockableResponse.clean();

  return jsonApiResponse;
}

LockableResponse{
  String resourceUrl;
  String jsonApiResponse;
  Semaphore semaphore = new Semaphore(0);

  public void clean(){
     resourceUrl = null;
     jsonApiResponse = null;         
  } 
}

My though is that we would associate a resourceUrl for each dynamodb response with a data structure that has semaphore to await for api response which arrives for ~100 items in one batch call.

A separate thread performs the call and then iterates through response, assigns jsonApiResponse to proper resourceUrl [concurrentMap for that?? ] and then calls lockableResponse.release() to wake up thread, so thread takes resourceUrl and exists the method.

The clean() method could be added to allow reuse of the same structure. I'm thinking about array of lockable resource according to batch capacity. So they could be cleared and reused for other batch GET call.

Thoughts? Suggestions?

Answer 1

Here is a reactive approach to your problem. RxJava allows you to manage resources, especially threads, in a style similar to that used in Java streams.

The request queue for GETs is a SerializedSubject which is where requests are sent. SerializedSubject is thread safe, and allows requests to be made from any thread.

SerializedSubject<Pair<Observable<JsonResponse>, String> requestGetQueue =
  PublishSubject.<Pair<Observable<JsonResponse>, String>create().toSerial();

Once the subject is declared, set up processing of the queued requests. The observeOn() operator tells the observer chain to perform processing on a particular scheduler, which then selects a thread from its pool to perform all the operations on. RxJava takes of the thread-hopping to go from the calling thread to the thread that handles the requests.

requestGetQueue
  .observeOn( Schedulers.io() )
  .buffer( 100 )  // batch size
  .subscribe( requestList -> processRequestList( requestList ),
    error -> log.error( error ) );

The buffer() operator batches up a set of requests into a list. The batch size is 100, per the original posting. The buffer() operator can take an additional parameter to set a timeout, so that eventually a lone group of requests will be handled and nothing gets stuck. That's a business decision whether you want to handle less than 100 requests at any time.

Observable<JsonResponse> getApiValue( String url ) {
  BehaviorSubject<JsonResponse> responseFromApi = BehaviorSubject.create();
  requestQueue.onNext( new Pair<>(responseFromApi, url) );
  return responseFromApi;
}

The API request for clients is very simple. What the caller gets back is the equivalent of a future. The client makes the request and establishes a call-back using the subscribe() step to handle the JsonResponse that will eventually come through. An error handler is also required, since ... well, errors happen. The observeOn() operator is used to move the response handling back on to another thread. More on that in a bit.

getApiValue( apiUrlString )
  .observeOn( clientScheduler )
  .subscribe( jsonResponse -> { ... },
    error -> { ... } );

Handling the request is simply batching up the list of URLs, waiting for the list of responses to come back and pairing up the responses to the requests. The code below assumes all responses come back in order. The response is emitted to the client by using onNext() followed by onCompleted().

void processRequestList( List<Pair<Observable<JsonResponse>,String>> requestList ) {
  List<String> uriList = new ArrayList<>();
  for ( int s = 0; s < requestList.size(); s++ ) {
    uriList.add( requestList.getSecond() );
  }
  List<JsonResponse> results = sendBatchRequest( uriList );
  for ( int i = 0; i < requestList.size(); i++ ) {
    requestList.get(i).getFirst().onNext( results.get(i) );
    requestList.get(i).getFirst().onCompleted();
  }
}

I mentioned earlier that the client needs to use a scheduler to move processing back to its own thread. How you do this depends on how client threads are set up. You can create the client scheduler from an executor service:

clientScheduler = Schedulers.from( executorService );

and using observeOn( clientScheduler ) will cause subsequent operations to move on to the client thread.

Summary

RxJava, and similar reactive platforms, manage almost all of the details of threads, locks, semaphores, mutexes, blocking queues, timers, responses, etc, for you. You still have to understand what things are needed for thread safety, and when processing is performed on particular threads, but almost of the finicky, tricky stuff is kept behind the scenes.

Answer 2

To recap: the problem you want to solve is akin to a 'shuttle bus' situation. You have requests that are coming in and you wish to bundle them together in batches of a certain size and send them together.

I thought about this and built a little prototype and you can get this behavior with a couple of simple concurrency constructs in Java: a CountDownLatch and a ConcurrentQueue. The basic idea is to take all the requests into the queue and create an object to hold the request and it's corresponding response with a CountDownLatch that will block on the retrieval of the result until it is retrieved. When the queue depth reaches the batch size, you pull off that many requests and send the batch. It then sets the results and frees the latch allowing the calling thread to retrieve the result. A code example follows. I've done only minimal testing of this and it's not necessarily the best solution but it should give you an overview of the approach.

First some 'primitives':

Request

import java.util.concurrent.CountDownLatch;

class Request<Q,R>
{
  private final Q value;
  private final CountDownLatch latch = new CountDownLatch(1);
  private volatile R result;

  Request(Q value)
  {
    this.value = value;
  }

  public Q value()
  {
    return value;
  }

  void setResult(R result)
  {
    this.result = result;
    latch.countDown();
  }

  public R getResult() throws InterruptedException
  {
    latch.await();
    return result;
  }
}

Processor

public interface Processor<Q, R>
{
  void execute(Collection<Request<Q,R>> requests);
}

The above is responsible for issuing the batch request and then setting the responses for each corresponding request. You can build stuff around this to fit your needs.

And then the heavy lifting:

RequestBatcher

import java.util.ArrayList;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Future;
import java.util.concurrent.FutureTask;

public class RequestBatcher<Q, R>
{
  private final int limit;
  private final BlockingQueue<Request<Q,R>> queue;
  private final Processor<Q,R> processor;

  RequestBatcher(Processor<Q,R> processor, int limit)
  {
    this.limit = limit;
    this.processor = processor;
    queue = new ArrayBlockingQueue<>(limit * 2);
  }

  public Future<R> process(Q request) throws InterruptedException
  {
    Request<Q,R> holder = new Request<>(request); 
    BatchedCallable callable = new BatchedCallable(holder);

    queue.put(holder);

    if (queue.size() >= limit) batchRequests();

    return callable.future();
  }

  public R await(Q request) throws InterruptedException, ExecutionException
  {
    Future<R> response = process(request);

    return response.get();
  }

  private class BatchedCallable implements Callable<R>
  {
    final Request<Q,R> request;

    BatchedCallable(Request<Q,R> request)
    {
      this.request = request;
    }

    @Override
    public R call() throws Exception
    {
      return request.getResult();
    }

    public Future<R> future()
    {
      FutureTask<R> future = new FutureTask<R>(this);
      future.run();

      return future;
    }
  }

  private void batchRequests() throws InterruptedException
  {
    while(queue.size() >= limit) {
      ArrayList<Request<Q,R>> requests = new ArrayList<>(limit);

      for (int i =0; i < limit; i++) {
        requests.add(queue.take());
      }

      processor.execute(requests);
    }
  }
}

This implements what I think you have said you want to do in your question. I've added a blocking await() as well as the non-blocking process() method. The former is good if you want this to work just a like a synchronous call per request. The latter will allow you to use executors or similar to make things more parallel.

There are a couple of issues with the above that I think are worth mentioning. One problem is that nothing will execute until you reach your batching limit. If you don't have a steady constant stream of requests coming in, you will basically have a bunch of hung threads. If you use the blocking version, that means you need that mean threads to each make a single request. I would guess that you don't have 100 threads that are all constantly making requests at the same time so this could be problematic. One answer would be to have a single thread do the batching and use another CountDownLatch that it waits on. When you hit the batch number, you count down and it goes. This allows you to set a timeout in the await() which you can use to have non-full batches run after waiting past a time limit.

On a side note, I would also avoid attempts to reuse data structures. This might seem like it will improve performance but what it will often do is cause objects to live long enough to survive from the eden or young spaces and into the where old objects reside. The cost of collecting these is much higher than short-lived objects. You are likely better off just creating new objects as needed and letting them be collected.

The first thing I would say to this is that you should familiarize yourself with the Executor framework. I think the basic outline of what you are proposing is easily implemented using a fixed thread pool executor or one of the other built-in ExecutorService implementations.