CODEWITHSUNDEEP
javamultithreadingconcurrency
hakos
hakos

Reputation: 270

Make asynchronous queries synchronous

I have an underlying asynchronous out-of-order query/response system that I want to make synchronous. Queries and responses can be mapped by marking the queries with unique IDs that in turn will accompany the corresponding responses.

My attempt at making it synchronous uses two ConcurrentHashMaps: one that maps from IDs to results, and one that maps from the same IDs to CountDownLatches. The code goes like this when executing a query:

public Result execute(Query query) throws InterruptedException {
    int id = atomicInteger.incrementAndGet();
    CountDownLatch latch = new CountDownLatch(1);
    latchMap.put(id, latch);
    query.executeAsyncWithId(id); // Probably returns before result is ready
    try {
        latch.await(); // Blocks until result is ready
        return resultMap.remove(id);
    } catch (InterruptedException e) {
        latchMap.remove(id); // We are not waiting anymore
        throw e;
    }
}

And the code for handling an incoming result:

public void handleResult(Result result) {
    int id = result.getId();
    CountDownLatch latch = latchMap.remove(id);
    if (latch == null) {
        return; // Nobody wants result
    }
    resultMap.put(id, result);
    latch.countDown();
}

This method is called from a thread that reads all the incoming results from the underlying system (there is only one such reader thread).

First of all I am uncertain about the thread-safety, but it also seems unnecessary to use two HashMaps for this (particularly because IDs are never reused). Any ideas for improvements?

Upvotes: 3

Views: 3396

Answers (3)

sjlee
sjlee

Reputation: 7866

I think the version with the CountDownLatch is probably the best solution. "Java Concurrency in Practice" (Brian Goetz) actually talks about this (I think he calls it the value latch). It's essentially a one-time synchronization mechanism on a value being set.

Although it is possible to implement such a value latch with the wait()-notifyAll() mechanism, using the CountDownLatch yields a simpler implementation.

The CountDownLatch await()-countDown() methods provide a proper happens-before relationship.

Upvotes: 0

hakos
hakos

Reputation: 270

A new attempt inspired by this answer to a similar question: 

public class ResultFuture {

    private volatile Result result = null;
    private final CountDownLatch latch = new CountDownLatch(1);

    public Result get() throws InterruptedException {
        latch.await();
        return result;
    }

    public void set(Result result) {
        this.result = result;
        latch.countDown();
    }
}

Now I need only one HashMap of these ResultFutures:

public Result execute(Query query) throws InterruptedException {
    int id = atomicInteger.incrementAndGet();
    ResultFuture resultFuture = new ResultFuture();
    resultFutureMap.put(id, resultFuture);
    query.executeAsyncWithId(id); // Probably returns before result is ready
    try {
        return resultFuture.get(); // Blocks until result is ready
    } finally {
        resultFutureMap.remove(id);
    }
}

public void handleResult(Result result) {
    int id = result.getId();
    ResultFuture resultFuture = resultFutureMap.get(id);
    if (resultFuture == null) {
        return; // Nobody wants result
    }
    resultFuture.set(result);
}

Upvotes: 4

Jack
Jack

Reputation: 133567

Usage of ConcurrentHashMap implies that you should trust just methods that are defined as atomic in documentation like:

  • putIfAbsent(K key, V val)
  • replace(K key, V val)
  • remove(K key, V val)

so if you plan to keep them you should change your usage, this will at least guarantee thread safety of your hashmaps.

Apart fromt hat just create a new Executor for each query requested that returns the result itself so that the thread will wait for its completion before keeping up its work: in this way you will have the whole thing in a synchronized fashion..

Upvotes: 0

Related Questions