Group threads in ExecutorService

[Harshana Dias]

In ExecutorService, how can I group all the threads processing a specific course id and then upon one thread finish process that course id, close all the other threads which are in the process of that course id?

The number of threads which can process the course id untill one thread finish process, is not a fixed number too.

So for example, if those threads per course id is map in a collection, it would be as below,

List l1 = new ArrayList();
l1.add(thread1);

Map t = new Hashmap();
t.add("courseid1":l1)

Any time a new thread enter to process the same course id, I would add it to the list.

l1.add(thread2);

If any of the thread finish process the courseid1, then i am stop processing all threads in the list and return status 200 for each threads.

[Stephan van Hulst]

Don't work with threads. Instead, when you create your tasks, assign them an object through which they can signal that they're finished. All other tasks that are part of the same group periodically check the same object if any of them has finished yet. Then they can stop themselves.

This sounds like very questionable design though. Why would you have multiple tasks like that running at the same time, only to cancel all of them when one is done?

[Harshana Dias]

Actually, Its a long poll request. Firstly a thread would listen to a course status in a redis cache. If its not complete, go back to sleep for 10 seconds and wake and check again untill for 30 mits. If status is not complete with in that time, it will send error. Now with in 39 mints, if user refresh the page, another thread initiate and do the same process. Mean time, the first thead read the status from redis cache as complete and remove it from cache. Now when second thread wake and check for course status in redis cache, it will never find it. So i want to finish that thread execution

[Harshana Dias]

Where can i edit my question or reply?

[Henry Wong]

Harshana Dias wrote:Now with in 39 mints, if user refresh the page, another thread initiate and do the same process. Mean time, the first thead read the status from redis cache as complete and remove it from cache. Now when second thread wake and check for course status in redis cache, it will never find it. So i want to finish that thread execution

Sounds like a complicated design to me. What is wrong with the second (or third, or fourth, or fifth, etc) thread simply having the capability to detect that the state has changed, that it is no longer needed, clean up, and exit?  Surely, this is easier that having the first thread hang around for awhile, looking for an undermined number of threads that may be started by accident, and communicating with them (or the executor) in order to abort those threads.

Henry

[Henry Wong]

Harshana Dias wrote:Where can i edit my question or reply?

The details of eligibility to be edit, and how it is to be edit is somewhat complex on these forums... so, I won't go into it...

Regardless, if a post can be edited, you should see an "edit" button on top right of the post.

Henry

[Harshana Dias]

Henry Wong wrote:  Surely, this is easier that having the first thread hang around for awhile, looking for an undermined number of threads that may be started by accident, and communicating with them (or the executor) in order to abort those threads.

Henry

'
How would you suggest to first thread to detect what other threads accidentally doing the same process and communicate to them, "ok i am done with this process and here is the result. so stop keep processing for 30 mints and quit"?

What i am currently do is kind of  similar to what you suggest, instead keeping the first thread hang around for a while even it see the result to return, I keep the redis cache entry for 30 mints which is the entry, a thread want to see to end the process. So that all the accidentally trigger threads will not hang there for 30 mints due to the unavailability of cache entry instead when they wake in next 10 second, they see the cache entry and return the response. But this approach keeps a redis entry for a unwanted 30 mints time period.

Instead I want to know how can I utilize the java 8 or other java strategy for proper solution.  

[Paul Clapham]

Stephan van Hulst wrote:Don't work with threads. Instead, when you create your tasks, assign them an object through which they can signal that they're finished. All other tasks that are part of the same group periodically check the same object if any of them has finished yet. Then they can stop themselves.

You didn't like this answer? Why not?

[Harshana Dias]

Paul Clapham wrote:You didn't like this answer? Why not?

I have to work with threads in order to check a redis entry in time to time (by sleep and wake the thread)

[Paul Clapham]

I don't see what that has to do with Stephan's suggestion. After all we already know that your tasks are going to run in separate threads, so what's the objection to having those tasks/threads communicate via a shared object?

[Harshana Dias]

Paul Clapham wrote:I don't see what that has to do with Stephan's suggestion. After all we already know that your tasks are going to run in separate threads, so what's the objection to having those tasks/threads communicate via a shared object?

I assume you referering to use ExecutorService of concurrent package.

Suppose below is my method which each thread/task enters. I belive executor shoud be a class variable. Under Callable call method, I have to write the logic which checks the common object (in my case redis cache entry). Am I correct please?

public class A{

ExecutorService executor = Executors.newFixedThreadPool(2);
 
public void mymethod(){
MyCallable callable = new MyCallable(5000);
FutureTask<String> futureTask = new FutureTask<String>(callable);
executor.execute(futureTask);

while (true) {
  if(futureTask.isDone()){
    executor.shutdown();
  }else{
    Thread.sleep(2000);
  }
}

}
}

But then again I still need to group those tasks to course id because I cant just shut down all tasks in the executor. I can shut down only tasks which process same course.  Based on my above code despite of which course a thread process, all add to the same ExecutorService. Can you provide suggestions to improve it?

I can think of having a map with key as the course id and value as ExecutorService object. So basically, I have multiple ExecutorService per course id's. Does below approach is correct?

public class A{

Map<String, ExecutorService> map = new HashMap<String, ExecutorService>();

public void mymethod(){

ExecutorService executor = null;
MyCallable callable = new MyCallable(5000);
FutureTask<String> futureTask = new FutureTask<String>(callable);

if(map.get(courseId) == null){
  executor = Executors.newFixedThreadPool(2);
  map.add(courseId,executor);
  executor.execute(futureTask);
}else{
  executor = (ExecutorService)map.get(courseId);
  executor.execute(futureTask);
}

executor.execute(futureTask);

while (true) {
  if(futureTask.isDone()){
    executor.shutdown();
  }else{
    Thread.sleep(2000);
  }
}

}
}

[Paul Clapham]

Yes, a Map<String, Set<FutureTask<String>>> sounds like a reasonable idea. But I would only use one ExecutorService -- I don't see the point of having more than one.

So when one of those FutureTask objects completes, presumably it knows the courseId so it looks in that Map and finds all of the FutureTask<String> objects which are processing that courseId, and then stops them all.

[Harshana Dias]

Paul Clapham wrote:
So when one of those FutureTask objects completes, presumably it knows the courseId so it looks in that Map and finds all of the FutureTask<String> objects which are processing that courseId, and then stops them all.

But can only stop those tasks via the ExecutorService shutdown method know?

[Stephan van Hulst]

No. Like I suggested earlier, tasks should stop themselves. They need shared state to determine when to shut down. Here's how it roughly looks:

import java.util.*;
import java.util.concurrent.locks.*;

import static java.util.concurrent.TimeUnit.*;

final class Task {

  private final Lock taskLock = new ReentrantLock();

  private final Condition taskDoneCondition = taskLock.newCondition();

  private final String id;

  private Result result;

  Task(String id) {
    if (id == null)
      throw new IllegalArgumentException();

    this.id = id;
  }

  String id() {
    return id;
  }
  
  Optional<Result> getResult() {
    return Optional.ofNullable(result);
  }

  Result perform() throws InterruptedException {
    taskLock.lock();

    try {
      while (result == null) {
        // Do redis stuff
        
        if (/* redis stuff succeeded */) {
          result = new Result(/* data from redis */);
          taskDoneCondition.signalAll();
          return result;
        }
        
        taskDoneCondition.await(10, SECONDS);
      }
      
      throw new InterruptedException("Another thread already completed this task.");
    }

    finally {
      taskLock.unlock();
    }
  }
}

You need to keep a Map<String, Task> somewhere, and whenever you load the page, you just submit a new callable to your executor service:

Task task = tasks.computeIfAbsent(id, Task::new);

if (!task.getResult().isPresent()) {
  Future<Result> futureResult = executor.submit(task::perform);
}

[Paul Clapham]

That almost looks like "Only start one task for each course ID. Then worrying about stopping the other tasks which are also processing the course when one of them finishes becomes unnecessary."

I always wondered about the design where you set off several tasks which are (apparently) doing the same thing and then when one of them wins, tell the others to stop. If the tasks are updating some resource then it just seems like a recipe for disaster to have multiple tasks updating that resource in the same way, especially if some of them can be interrupted unexpectedly.

[Harshana Dias]

Stephan van Hulst wrote:No. Like I suggested earlier, tasks should stop themselves. They need shared state to determine when to shut down. Here's how it roughly looks:
You need to keep a Map<String, Task> somewhere, and whenever you load the page, you just submit a new callable to your executor service:

Task task = tasks.computeIfAbsent(id, Task::new);

if (!task.getResult().isPresent()) {
  Future<Result> futureResult = executor.submit(task::perform);
}

Thank you for your answer with code. Surely it helps me to learn about concurrent.lock package, which something I learn for the first time.

I just have few questions for instance about this id you mentioned in computeIfAbsent. I assume its "courseId" in my context of the question.

Eg: Task task = tasks.computeIfAbsent("course1", Task::new);

If so, what ever tasks (threads) hiting for same course id, would not be add to the tasks map right? Only the initially added task for given course id would be there right?

If above assumption is correct, I am wondering, since the Task class also not static, for a particular course, only one Task object will be initiate right?
Then does this locks and  taskDoneCondition.signalAll() is required?

[Harshana Dias]

I mean since Task class is not static, it will not be share between seperate tasks right? Because each course id has its own Task object?

[Stephan van Hulst]

Harshana Dias wrote:what ever tasks (threads)

Tasks are NOT threads. A task is a job that needs to be done. Threads perform jobs. Your question was about getting ONE task done by multiple threads, and interrupt threads when the task has already been performed by another thread.

Yes, each course ID will be associated with only one Task object. When you reload the page, you resubmit the same Task instance to the executor service. That means that the same Task instance will be shared between multiple threads, and that's why we need to use a Lock object to synchronize between them, and use signalAll() to wake them all up when the task is finished.

[Stephan van Hulst]

I also just noticed that the result field of the task could be volatile to make the getResult() method more reliable:

private volatile Result result;

[Harshana Dias]

Stephan van Hulst wrote: That means that the same Task instance will be shared between multiple threads, and that's why we need to use a Lock object to synchronize between them, and use signalAll() to wake them all up when the task is finished.

Thank you. I can now understand your implementation.

Just few clarifications

1. I just wrote the main method using Future so that I can generate the behaviour of multiple threads invoke task perform.

public static void main(String[] args) {

        Map<String, MyTask> taskMap = new HashMap<String, MyTask>();

        MyTask t1 = new MyTask("456");
        ProcessTask fc = new ProcessTask(taskMap, t1, "123");
        System.out.println("Gonna submit1");
        Future<?> future = threadpool.submit(fc);
        System.out.println("Submitted1");

        MyTask t2 = new MyTask("457");
        ProcessTask fc2 = new ProcessTask(taskMap, t2, "123");
        System.out.println("Gonna submit2");
        Future<?> future2 = threadpool.submit(fc2);
        System.out.println("Submitted2");        

       //threadpool.shutdown();

 }

And I pass courseId to ProcessTask which implement Callable.  So from the call() of ProcessTask, I call below,

Just before the executor.submit(task::perform), I add below print statement.

   public String processTasks(Map<String, MyTask> taskMap, MyTask taskPar, String courseID) {
        MyTask task = taskMap.computeIfAbsent(courseID, MyTask -> taskPar);
        ExecutorService executor = Executors.newFixedThreadPool(2);

        if (!task.getResult().isPresent()) {
            System.out.println("processTasks: " + courseID + " " +task.id()+" "+ Thread.currentThread().getId());
            Future<Result> futureResult = executor.submit(task::perform);
            try {
                while (!futureResult.isDone()) {
                    System.out.println("Task is not completed yet...." + Thread.currentThread().getId());
                    try {
                        Thread.sleep(5000);
                    } catch (InterruptedException e) {
                    }
                }
                System.out.println("Task completed...." + Thread.currentThread().getId());
                Result result = futureResult.get();
                System.out.println("Returning result: " + result.getName() + " " + Thread.currentThread().getId());
                return result.getName();
            } catch (InterruptedException | ExecutionException e) {
                e.printStackTrace();
            }
        }
        return null;
    }

But I see below out put,

Gonna submit1
Submitted1
Gonna submit2
Submitted2
processTasks: 123 456 10
processTasks: 123 457 11

How come it print 457 with thread is 11 as there is a already task for course id 123? I expect 123 456 10.

2. How can I decide a number for threadpool?

ExecutorService executor = Executors.newFixedThreadPool(2);

[Harshana Dias]

Correction:
1.How come it print 457 with thread id 11 as there is a already task for course id 123? I expect 123 456 11.

[Harshana Dias]

Harshana Dias wrote:Correction:
1.How come it print 457 with thread id 11 as there is a already task for course id 123? I expect 123 456 11.

Sorry. Its because I add a new Task object to ProcessTask instance. Instead I add below and it prints expected 123 456 11.

ProcessTask fc2 = new ProcessTask(taskMap, t1, "123");