Wait for completion of threadpool tasks using Phaser

Question

I want to return from my executeTasks() method only after all the tasks submitted to the threadpool are finished. Please note that my thread pool has configurable threadpoolsize and uses SynchronousQueue as the backing queue, so my for loop proceeds safely by submitting a task only if a thread is available. So, I just want to wait for the final tasks. I am using Phaser for this.

I've created a Phaser with 1 registered party i.e., the current thread and I register a new party to the Phaser before submitting a task to the threadpool, when the task finishes I deregister the task party. When for loop finishes by submitting the final tasks, I am hoping that my arriveAndAwaitAdvance() will wait for registered parties to arrive but it will only discover that all those parties are deregistered after some time and then move forward and return from my method.

I think that this will solve my problem. Please let me know if I am wrong or if there is any other better way to do this. Countdownlatch is not going to help as my threadpoolsize is configurable. I know that having a counter and monitor will solve this problem but I want out-of-the-box solution like Phaser.

private void executeTasks(TheadPoolExecutor threadPool, Iterator<String> it) {
    final Phaser phaser = new Phaser(1);

    for (final String id : IteratorUtils.iterable(it)) {
        phaser.register();
        threadPool.execute(() -> {
        // phaser.arrive();
            try {
                thread.sleep(10000 * id.length());
            } finally {
                phaser.arriveAndDeregister();
            }
        });
    }
    phaser.arriveAndAwaitAdvance();
    phaser.arriveAndDeregister();
}

I implemented this and this works. If anyone finds any loophole or a better way, please let me know. — uday, Mar 11 '17 at 02:22

Dimitri · Answer 1 · 2017-03-10T16:39:21.230

0

I never used a Phaser before but I think a CountDownLatch is the better way to handle this task.

A CountDownLatch is a synchronization barier that allows one or more threads to wait until a set of operations being performed in other threads completes.

2 methods are useful when using a CountDownLatch :

countDown that decrements the counter when a task is finish.
await is for the current thread (main for instance) to wait the others to complete.

*

private void executeTasks(TheadPoolExecutor threadPool, Iterator<String> it) {
    final CountDownLatch countDownLatch = new CountDownLatch(threadPool.getPoolSize());

    for (final String id : IteratorUtils.iterable(it)) {
        threadPool.execute(() -> {
            try {
                thread.sleep(10000 * id.length());
                countDownLatch.countDown();
            } catch (InterruptedException ex) {}
        });
    }
    countDownLatch.await();
  }

Here the CountDownLatch is initialized with the number of threads in the threadpool.

edited Mar 10 '17 at 16:39

answered Mar 10 '17 at 16:31

Dimitri

8,122
19
71
128

Oh I am sorry. I should have mentioned that my thread pool size is configurable and countdownlatch won't help me – uday Mar 10 '17 at 16:32
It does not matter, because you pass a threadPool as a parameter. It only needs to get the number of threads in the poolsize – Dimitri Mar 10 '17 at 16:33
What if I want to increase the pool size? I cannot increment countdown latch right – uday Mar 10 '17 at 16:41
The `CountDownLatch ` does not need to know about increasing the pool size. If you see in the method `executeTasks`, I pass the number of threads in the thread pool as a parameter for the latch. – Dimitri Mar 10 '17 at 16:52
Yes I understand that but I am saying that during the execution of my tasks I want to increase thread pool size. But since countdownlatch is created with initial thread pool size, it cannot be adjusted. Also, your implementation assumes that each thread just one task which is not the case. – uday Mar 10 '17 at 17:01
Well you need to give more info of what you want to achieve – Dimitri Mar 10 '17 at 17:02
Well look at this answer http://stackoverflow.com/questions/2791851/can-you-dynamically-resize-a-java-util-concurrent-threadpoolexecutor-while-it-st – Dimitri Mar 10 '17 at 17:17
Sorry for not providing more details. I am able to adjust my threadpoolsize using setCorePoolSize() and setMaxPoolSize() – uday Mar 10 '17 at 17:45
I provided more details in the question now – uday Mar 10 '17 at 17:48

Wait for completion of threadpool tasks using Phaser

1 Answers1