FixedThreadPool with a Blocking submit

[Hadi Zeftin]

Hi, been searching a whole days now, and still haven't found any answers I can understand of.

        ExecutorService exec = Executors.newFixedThreadPool(this.maxThread);
        Transaction transaction;
        while ((transaction = this.transactionFactory.getNextTransaction()) != null) {
            exec.execute(transaction);  // i want it to block if the pool already working on a number of maxThread threads.
        }
        exec.shutdown();

I thought this would be a simple question with hopefully a simple answer too. How can I make the main/current thread block if the executor already working on a number of maxThread threads, for example if maxThreads=10 then exec.execute() will block if already submitting 10 threads and all of those threads are still running, until there some space available (i.e. some threads finish and idle). why ? because this.transaction.getNextTransaction() may always returning a result for a long time, putting it to executor will probably cost memory usage ?? not sure, just doesn't feel right.

maybe I'm using a wrong set of Classes there, if I am please point me to the right direction.

[Chris Hurst]

You could use the CallerRunsPolicy if that's helpful (not blocking though :-( ) ...

Anyway failing that (from Java Concurrency in Practice (very recommended book) ) there is

no predefined saturation policy to make execute block when the work queue is full. However the effect can be accomplished by using a Semaphore to bound the task injection rate

. It has a listing to achieve that sure you can google one.

[Hadi Zeftin]

yes it does, thanks so much for the hint. now it became like this

        ExecutorService exec = Executors.newFixedThreadPool(this.maxThread);
        Transaction transaction;
        BoundedExecutor bex = new BoundedExecutor(exec, this.maxThread);

        while ((transaction = this.transactionFactory.getNextTransaction()) != null) {
            try {
                bex.submitTask(transaction);
            } catch (InterruptedException e) {
                logger.error("error on a transaction submition", e);
            }
            // exec.execute(transaction);
        }
        exec.shutdown();

and for anyone that had the same question, the bounded executor

package net.jcip.examples;
import java.util.concurrent.*;
import net.jcip.annotations.*;

/**
 * BoundedExecutor
 * <p/>
 * Using a Semaphore to throttle task submission
 *
 * @author Brian Goetz and Tim Peierls
 */
@ThreadSafe
public class BoundedExecutor {
    private final Executor exec;
    private final Semaphore semaphore;
    public BoundedExecutor(Executor exec, int bound) {
        this.exec = exec;
        this.semaphore = new Semaphore(bound);
    }
    public void submitTask(final Runnable command)
            throws InterruptedException {
        semaphore.acquire();
        try {
            exec.execute(new Runnable() {
                public void run() {
                    try {
                        command.run();
                    } finally {
                        semaphore.release();
                    }
                }
            });
        } catch (RejectedExecutionException e) {
            semaphore.release();
        }
    }
}

[rudra tripathy]

Doesn't thread pool take care of that?if my thread pool size is 10, and i have 100 threads to execute, then this should automatically take care of,right, or i misunderstood the question?

[Steve Luke]

rudra tripathy wrote:Doesn't thread pool take care of that?if my thread pool size is 10, and i have 100 threads to execute, then this should automatically take care of,right, or i misunderstood the question?

An ExecutorService will usually have two parts: 1 Pool of Threads and 1 Queue for holding tasks. When there are more tasks than threads, then the tasks get pushed into the Queue for execution when a thread becomes available. But the the thread which submits the task does not wait for a worker thread to become available - the task just goes into a Queue and the submitter continues on.

So if there are 10 threads already working, then the 11th task gets submitted, this extra task gets put into the Queue and the thread which submits the task continues on its way. What the OP wants is the submitter thread to block - stop executing - until the task gets consumed by a worker thread.

[rudra tripathy]

Hi Steve,

Thanks for the confirmation.