4

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?

  • If there is a request for information that has to be retrieved from the payed API, can it wait until 100 requests are ready to send in a batch? Or do you need to combine it with requests for almost expired cached items; so that the response comes as soon as possible? – Kasper van den Berg Feb 5 '18 at 18:18
  • Batch is only for DynamoDb to make it more useful. Calls to API go as one item at a time -> request/response. Request goes to "GET someapi.com/product/123:333" and response goes as JSON so it was decided to cache it. We mostly process big input csv files with hundreds/thousands of lines, so we don't need to wait too long until batch fills up. For 95% cases batch will fill quickly. Probably batch will be half-filled for last part of file. But our calls are logic-based so its not straightforward. – Flamaker2018 Feb 5 '18 at 19:05
2

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.

  • Wow, tnx a lot for the detailed answer. I've heard about reactive java, but now it's knocking at my door. I'll use this, tnx! – Flamaker2018 Feb 9 '18 at 12:54
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.

  • I'm familiar with futures and thread pool executors. I understand that we have a thread pool with n threads. We submit the task and get list of futures DynamoDb batch call needs 100 resourceUrl to perform GET call. So we have thread pool for 100 threads, they enter the method class DynamoDbCache{ public String get(String resourceUrl){ executor.submit( (resourceUrl) -> { // but still here we need some collection to put 100 requests together for one batch call. Ok, future will not exit method until its done. } } } – Flamaker2018 Feb 6 '18 at 3:25
  • Thread pool would do some independent tasks, but we don't have independent lightweight tasks here, but we have one - batch call to dynamodb. We need to make 100 thread wait until information for them in retrieved in one batch call. We need to associate their resourceUrl and wait them up when we have result from batch call for them. Could you explain how you see this particular scenario arrive into the method "take his place in batch call" batch call is executed dynamodb response with 100 items is there each thread receives batch response – Flamaker2018 Feb 6 '18 at 3:28
  • Yeah, for scenarios like call api client fast I used thread pool to submit API calls, let 50 threads do the job and just used future.get() - that helped me to get things done. Here its more about waiting and blocking threads. If I would do 1 dynamodb call for every incoming thread, I would probably use thread pool to submit it to thread pool and just return the future. But its a bit different scenario here. Let me know how you see this. – Flamaker2018 Feb 6 '18 at 3:32
  • I think I understand what you want to do but let me try to explain it back to you and see if you agree. You are getting a single bulk response that you wish to handle across multiple threads. Is this correct or is there more to it? – JimmyJames Feb 6 '18 at 14:37
  • ------ begin --------- 1)There are many threads with "String resourceUrl" each 2)We want to place their resourceUrl into batch => String[] resourceUrls 3)Make them wait 4)Perform batch call to get 100 items in response 5)give each thread his String response for corresponding resourceUrl 6)wake each thread that he has his response ----end ------ Overall instead of 100 calls x 0.4 cents to make one 0.4 cent call – Flamaker2018 Feb 6 '18 at 15:20

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service, privacy policy and cookie policy

Not the answer you're looking for? Browse other questions tagged or ask your own question.