Ggtcf

In the system, I have an object - let's call it TaskProcessor. It holds queue of tasks, which are executed by some pool of threads (ExecutorService + PriorityBlockingQueue)
The result of each task is saved in the database under some unique identifier.

The user, who knows this unique identifier, may check the result of this task. The result could be in the database, but also the task could still wait in the queue for execution. In that case, UserThread should wait until the task will be finished.

Additionally, the following assumptions are valid:

• Someone else could enqueue the task to TaskProcessor and some random UserThread can access the result if he knows the unique identifier.




• UserThread and TaskProcess are in the same app. TaskProcessor contains a pool of threads, and UserThread is simply servlet Thread.




• UserThread should be blocked when asking for the result, and the result is not completed yet. UserThread should be unblocked immediately after TaskProcessor complete task (or tasks) grouped by a unique identifier

My first attempt (the naive one), was to check the result in the loop and sleep for some time:

// UserThread
while(!checkResultIsInDatabase(uniqueIdentifier))
 sleep(someTime)

But I don't like it. First of all, I am wasting database connections. Moreover, if the task would be finished right after sleep, then the user will wait even if the result just appeared.

Next attempt was based on wait/notify:

//UserThread 
while (!checkResultIsInDatabase())
 taskProcessor.wait()

//TaskProcessor
... some complicated calculations
this.notifyAll()

But I don't like it either. If more UserThreads will use TaskProcessor, then they will be wakened up unnecessarily every time some task would be completed and moreover - they will make unnecessary database calls.

The last attempt was based on something which I called waitingRoom:

//UserThread
Object mutex = new Object();
taskProcessor.addToWaitingRoom(uniqueIdentifier, mutex)
while (!checkResultIsInDatabase())
 mutex.wait()

//TaskProcessor
... Some complicated calculations
if (uniqueIdentifierExistInWaitingRoom(taskUniqueIdentifier))
 getMutexFromWaitingRoom(taskUniqueIdentifier).notify()

But it seems to be not secure. Between database check and wait(), the task could be completed (notify() wouldn't be effective because UserThread didn't invoke wait() yet), which may end up with deadlock.

It seems, that I should synchronize it somewhere. But I am afraid that it will be not effective.
Is there a way to correct any of my attempts, to make them secure and effective? Or maybe there is some other, better way to do this?

I believe replacing of mutex with CountDownLatch in waitingRoom approach prevents deadlock.

CountDownLatch latch = new CountDownLatch(1)
taskProcessor.addToWaitingRoom(uniqueIdentifier, latch)
while (!checkResultIsInDatabase())
 // consider timed version
 latch.await()

//TaskProcessor
... Some complicated calculations
if (uniqueIdentifierExistInWaitingRoom(taskUniqueIdentifier))
 getLatchFromWaitingRoom(taskUniqueIdentifier).countDown()

You seem to be looking for some sort of future / promise abstraction. Take a look at CompletableFuture, available since Java 8.

CompletableFuture<Void> future = CompletableFuture.runAsync(db::yourExpensiveOperation, executor);

// best approach: attach some callback to run when the future is complete, and handle any errors
future.thenRun(this::onSuccess)
 .exceptionally(ex -> logger.error("err", ex));

// if you really need the current thread to block, waiting for the async result:
future.join(); // blocking! returns the result when complete or throws a CompletionException on error

You can also return a (meaningful) value from your async operation and pass the result to the callback. To make use of this, take a look at supplyAsync(), thenAccept(), thenApply(), whenComplete() and the like.

You can also combine multiple futures into one and a lot more.

With CompletableFuture and a ConcurrentHashMap you can achieve it:

/* Server class, i.e. your TaskProcessor */
// Map of queued tasks (either pending or ongoing)
private static final ConcurrentHashMap<String, CompletableFuture<YourTaskResult>> tasks = new ConcurrentHashMap<>();

// Launch method. By default, CompletableFuture uses ForkJoinPool which implicitly enqueues tasks.
private CompletableFuture<YourTaskResult> launchTask(final String taskId) {
 return tasks.computeIfAbsent(taskId, v -> CompletableFuture // return ongoing task if any, or launch a new one
 .supplyAsync(() -> 
 doYourThing(taskId)) // get from DB or calculate or whatever
 .whenCompleteAsync((integer, throwable) -> 
 if (throwable != null) 
 log.error("Failed task: ", taskId, throwable);
 
 tasks.remove(taskId);
 )
 );


/* Client class, i.e. your UserThread */
// Usage
YourTaskResult taskResult = taskProcessor.launchTask(taskId).get(); // block until we get a result

Any time a user asks for the result of a taskId, they will either:

• enqueue a new task if they are the first to ask for this taskId; or


• get the result of the ongoing task with id taskId, if someone else enqueued it first.

This is production code currently used by hundreds of users concurrently.

In our app, users ask for any given file, via a REST endpoint (every user on its own thread). Our taskIds are filenames, and our doYourThing(taskId) retrieves the file from the local filesystem or downloads it from an S3 bucket if it doesn't exist.

Obviously we don't want to download the same file more than once. With this solution I implemented, any number of users can ask for the same file at the same or different times, and the file will be downloaded exactly once. All users that asked for it while it was downloading will get it at the same time the moment it finishes downloading; all users that ask for it later, will get it instantly from the local filesystem.

Works like a charm.

What I understood from the question details is-

When UserThread requests for result, there are 3 possibilities:

1. Task has been already completed so no blocking of user thread and directly get result from DB.


2. Task is in queue or executing but not yet completed, so block the user thread(till now there should not be any db queries) and just after completion of task(the task result must be saved in DB at this point), unblock user thread(now user thread can query the DB for result)


3. There is no task submitted ever for the given uniqueIdentifier which user has requested, in this case there will be empty result from db.

For point 1 and 3, Its straight forward, there will not be any blocking of UserThread, just query the result from DB.

For point 2 - I have written a simple implementation of TaskProcessor. Here I have used ConcurrentHashMap to keep the current tasks which are not yet completed. This map contains the mapping between UniqueIdentifier and corresponding task. I have used computeIfPresent() (introduced in JAVA - 1.8) method of ConcurrentHashMap which guarantees that the invocation of this method is thread safe for the same key. Below is what java doc says:
Link

If the value for the specified key is present, attempts to compute a
new mapping given the key and its current mapped value. The entire
method invocation is performed atomically. Some attempted update
operations on this map by other threads may be blocked while
computation is in progress, so the computation should be short and
simple, and must not attempt to update any other mappings of this map.

So with use of this method, whenever there is a user thread request for a task T1 and if the task T1 is in queue or executing but not completed yet, then user thread will wait on that task.
When the task T1 will be completed, all the user requests thread which were waiting on task T1 will be notified and then we will remove task T1 from the above map.

Other classes reference used in below code are present on this link.

TaskProcessor.java:

import java.util.Map;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.function.BiFunction;

public class TaskProcessor implements ITaskProcessor 

 //This map will contain all the tasks which are in queue and not yet completed
 //If there is scenario where there may be multiple tasks corresponding to same uniqueIdentifier, in that case below map can be modified accordingly to have the list of corresponding tasks which are not completed yet
 private final Map<String, Task> taskInProgresssByUniqueIdentifierMap = new ConcurrentHashMap<>();
 private final int QUEUE_SIZE = 100;
 private final BlockingQueue<Task> taskQueue = new ArrayBlockingQueue<Task>(QUEUE_SIZE);
 private final TaskRunner taskRunner = new TaskRunner();

 private Executor executor;
 private AtomicBoolean isStarted;
 private final DBManager dbManager = new DBManager();

 @Override
 public void start() 
 executor = Executors.newCachedThreadPool();
 while(isStarted.get()) 
 try 
 Task task = taskQueue.take();
 executeTaskInSeperateThread(task);

 catch (InterruptedException e) 
 e.printStackTrace();
 
 

 

 private void executeTaskInSeperateThread(Task task) 
 executor.execute(() -> 
 taskRunner.execute(task, new ITaskProgressListener() 

 @Override
 public void onTaskCompletion(TaskResult taskResult) 
 task.setCompleted(true);
 //TODO: we can also propagate the taskResult to waiting users, Implement it if it is required.
 notifyAllWaitingUsers(task);
 

 @Override
 public void onTaskFailure(Exception e) 
 notifyAllWaitingUsers(task);
 
 );
 );
 

 private void notifyAllWaitingUsers(Task task) 
 taskInProgresssByUniqueIdentifierMap.computeIfPresent(task.getUniqueIdentifier(), new BiFunction<String, Task, Task>() 
 @Override
 public Task apply(String s, Task task) 
 synchronized (task) 
 task.notifyAll();
 
 return null;
 
 );
 
 //User thread
 @Override
 public ITaskResult getTaskResult(String uniqueIdentifier) 
 TaskResult result = null;
 Task task = taskInProgresssByUniqueIdentifierMap.computeIfPresent(uniqueIdentifier, new BiFunction<String, Task, Task>() 
 @Override
 public Task apply(String s, Task task) 
 synchronized (task) 
 try 
 //
 task.wait();
 catch (InterruptedException e) 
 e.printStackTrace();
 
 
 return task;
 
 );
 //If task is null, it means the task was not there in queue, so we direcltly query the db for the task result
 if(task != null && !task.isCompleted()) 
 return null; // Handle this condition gracefully, If task is not completed, it means there was some exception
 
 ITaskResult taskResult = getResultFromDB(uniqueIdentifier); // At this point the result must be already saved in DB if the corresponding task has been processed ever.
 return taskResult;
 

 private ITaskResult getResultFromDB(String uniqueIdentifier) 
 return dbManager.getTaskResult(uniqueIdentifier);
 

 //Other thread 
 @Override
 public void enqueueTask(Task task) 
 if(isStarted.get()) 
 taskInProgresssByUniqueIdentifierMap.putIfAbsent(task.getUniqueIdentifier(), task);
 taskQueue.offer(task);
 
 

 @Override
 public void stop() 
 isStarted.compareAndSet(true, false);
 

Let me know in comments if you have any queries.
Thanks.