• Post Reply Bookmark Topic Watch Topic
  • New Topic
programming forums Java Mobile Certification Databases Caching Books Engineering Micro Controllers OS Languages Paradigms IDEs Build Tools Frameworks Application Servers Open Source This Site Careers Other Pie Elite all forums
this forum made possible by our volunteer staff, including ...
Marshals:
  • Campbell Ritchie
  • Jeanne Boyarsky
  • Ron McLeod
  • Paul Clapham
  • Liutauras Vilda
Sheriffs:
  • paul wheaton
  • Rob Spoor
  • Devaka Cooray
Saloon Keepers:
  • Stephan van Hulst
  • Tim Holloway
  • Carey Brown
  • Frits Walraven
  • Tim Moores
Bartenders:
  • Mikalai Zaikin

Group threads in ExecutorService

 
Ranch Hand
Posts: 352
  • Mark post as helpful
  • send pies
    Number of slices to send:
    Optional 'thank-you' note:
  • Quote
  • Report post to moderator
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,


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


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.
 
Saloon Keeper
Posts: 15491
363
  • Mark post as helpful
  • send pies
    Number of slices to send:
    Optional 'thank-you' note:
  • Quote
  • Report post to moderator
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
Ranch Hand
Posts: 352
  • Mark post as helpful
  • send pies
    Number of slices to send:
    Optional 'thank-you' note:
  • Quote
  • Report post to moderator
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
Ranch Hand
Posts: 352
  • Mark post as helpful
  • send pies
    Number of slices to send:
    Optional 'thank-you' note:
  • Quote
  • Report post to moderator
Where can i edit my question or reply?
 
author
Posts: 23951
142
jQuery Eclipse IDE Firefox Browser VI Editor C++ Chrome Java Linux Windows
  • Mark post as helpful
  • send pies
    Number of slices to send:
    Optional 'thank-you' note:
  • Quote
  • Report post to moderator

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
author
Posts: 23951
142
jQuery Eclipse IDE Firefox Browser VI Editor C++ Chrome Java Linux Windows
  • Mark post as helpful
  • send pies
    Number of slices to send:
    Optional 'thank-you' note:
  • Quote
  • Report post to moderator

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
Ranch Hand
Posts: 352
  • Mark post as helpful
  • send pies
    Number of slices to send:
    Optional 'thank-you' note:
  • Quote
  • Report post to moderator

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.  
 
Marshal
Posts: 28177
95
Eclipse IDE Firefox Browser MySQL Database
  • Mark post as helpful
  • send pies
    Number of slices to send:
    Optional 'thank-you' note:
  • Quote
  • Report post to moderator

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
Ranch Hand
Posts: 352
  • Mark post as helpful
  • send pies
    Number of slices to send:
    Optional 'thank-you' note:
  • Quote
  • Report post to moderator

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
Marshal
Posts: 28177
95
Eclipse IDE Firefox Browser MySQL Database
  • Mark post as helpful
  • send pies
    Number of slices to send:
    Optional 'thank-you' note:
  • Quote
  • Report post to moderator
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
Ranch Hand
Posts: 352
  • Mark post as helpful
  • send pies
    Number of slices to send:
    Optional 'thank-you' note:
  • Quote
  • Report post to moderator

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?



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?



 
Paul Clapham
Marshal
Posts: 28177
95
Eclipse IDE Firefox Browser MySQL Database
  • Mark post as helpful
  • send pies
    Number of slices to send:
    Optional 'thank-you' note:
  • Quote
  • Report post to moderator
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
Ranch Hand
Posts: 352
  • Mark post as helpful
  • send pies
    Number of slices to send:
    Optional 'thank-you' note:
  • Quote
  • Report post to moderator

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
Saloon Keeper
Posts: 15491
363
  • Mark post as helpful
  • send pies
    Number of slices to send:
    Optional 'thank-you' note:
  • Quote
  • Report post to moderator
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:
 
Paul Clapham
Marshal
Posts: 28177
95
Eclipse IDE Firefox Browser MySQL Database
  • Mark post as helpful
  • send pies
    Number of slices to send:
    Optional 'thank-you' note:
  • Quote
  • Report post to moderator
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
Ranch Hand
Posts: 352
  • Mark post as helpful
  • send pies
    Number of slices to send:
    Optional 'thank-you' note:
  • Quote
  • Report post to moderator

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:



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.

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
Ranch Hand
Posts: 352
  • Mark post as helpful
  • send pies
    Number of slices to send:
    Optional 'thank-you' note:
  • Quote
  • Report post to moderator
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
Saloon Keeper
Posts: 15491
363
  • Mark post as helpful
  • send pies
    Number of slices to send:
    Optional 'thank-you' note:
  • Quote
  • Report post to moderator

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
Saloon Keeper
Posts: 15491
363
  • Mark post as helpful
  • send pies
    Number of slices to send:
    Optional 'thank-you' note:
  • Quote
  • Report post to moderator
I also just noticed that the result field of the task could be volatile to make the getResult() method more reliable:
 
Harshana Dias
Ranch Hand
Posts: 352
  • Mark post as helpful
  • send pies
    Number of slices to send:
    Optional 'thank-you' note:
  • Quote
  • Report post to moderator

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.




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.



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?
 
Harshana Dias
Ranch Hand
Posts: 352
  • Mark post as helpful
  • send pies
    Number of slices to send:
    Optional 'thank-you' note:
  • Quote
  • Report post to moderator
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
Ranch Hand
Posts: 352
  • Mark post as helpful
  • send pies
    Number of slices to send:
    Optional 'thank-you' note:
  • Quote
  • Report post to moderator

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.

 
reply
    Bookmark Topic Watch Topic
  • New Topic