this is my first time using thread pools and I dont quite understand how the executorservice works. I put watermarks over an image and merge them onto an empty picture. But even if i only use one thread it will still only draw half.
This is my WorkerThread class:
public class WorkerThread implements Runnable {
BufferedImage source;
BufferedImage toDraw;
int x;
int y;
BufferedImage target;
ParallelWatermarkFilter pf;
public WorkerThread(BufferedImage source, BufferedImage toDraw, int x, int y, BufferedImage target){
this.source = source;
this.toDraw = toDraw;
this.x = x;
this.y = y;
this.target = target;
pf = new ParallelWatermarkFilter(source, 5);
}
@Override
public void run() {
pf.mergeImages(source, toDraw, x, y, target);
}
}
And this is is how i use the ExecutorService in my FilterClass:
public BufferedImage apply(BufferedImage input) {
ExecutorService threadpool = Executors.newFixedThreadPool(numThreads);
for (int w = 0; w < imgWidth; w += watermarkWidth) {
for (int h = 0; h < imgHeight; h += watermarkHeight) {
Runnable worker = new WorkerThread(input, watermark, w, h, result);
System.out.println("WIDTH: " + w + " HEIGHT: " + h);
threadpool.execute(worker);
}
}
threadpool.shutdown();
Do the threads not wait until one thread is done ?
The thing ThreadPoolExecutor
shutdown and task execution/draining work queue/taking from work queue is racy. So you cannot rely on thread interruptiong mechanism or something else. All you are guaranteed with is:
Initiates an orderly shutdown in which previously submitted tasks are executed, but no new tasks will be accepted. Invocation has no additional effect if already shut down.
This method does not wait for previously submitted tasks to complete execution.
To dig a bit deeper into the ThreadPoolExecutor
implementation lets take a look at the main execution method:
final void runWorker(Worker w) {
Thread wt = Thread.currentThread();
Runnable task = w.firstTask;
w.firstTask = null;
w.unlock(); // allow interrupts
boolean completedAbruptly = true;
try {
while (task != null || (task = getTask()) != null) {
w.lock();
// If pool is stopping, ensure thread is interrupted;
// if not, ensure thread is not interrupted. This
// requires a recheck in second case to deal with
// shutdownNow race while clearing interrupt
if ((runStateAtLeast(ctl.get(), STOP) ||
(Thread.interrupted() &&
runStateAtLeast(ctl.get(), STOP))) &&
!wt.isInterrupted())
wt.interrupt();
try {
beforeExecute(wt, task);
Throwable thrown = null;
try {
task.run();
} catch (RuntimeException x) {
thrown = x; throw x;
} catch (Error x) {
thrown = x; throw x;
} catch (Throwable x) {
thrown = x; throw new Error(x);
} finally {
afterExecute(task, thrown);
}
} finally {
task = null;
w.completedTasks++;
w.unlock();
}
}
completedAbruptly = false;
} finally {
processWorkerExit(w, completedAbruptly);
}
}
The crucial part here is calling to getTask()
. Its fragment is:
if (rs >= SHUTDOWN && (rs >= STOP || workQueue.isEmpty())) {
decrementWorkerCount();
return null;
}
The method is not synchronized and relies only on ordering supplied by CAS'ed ctl
value. ctl
here is the global pool state stored inside AtomicInteger
(for non-blocking atomic ThreadPoolExecutor
state acquiring).
So the following case is possible.
getTask
RUNNING
.ctl
accordingly.