I just had the issue of debugging a large concurrent job that ran using Java’s Executor and ExecutorService classes for concurrency.
The job comprised of roughly 50,000 tasks submitted recursively over a multi-GB data set that took 5hrs to process. The bug left a single task in the queue hanging indefinitely, thus the job never completing. There’s no way to kill a Thread in Java, and if you have blocking I/O it can be difficult to implement interrupt() such that it works correctly.
This is the point I found it’s quite difficult to actually expose at run-time what the hell Java’s Executor is actually doing at any point in time.
The successful choice I found was to create a custom ThreadPoolExecutor which trapped the creation of a FutureTask and the start/stop of any given task. I expect these hooks are there for this purpose, but it sure isn’t obvious how to do this.
It seems odd that the API doesn’t include any way to gather info about what’s happening inside the Executor, and also, there’s not even a toString() implementation for wrapping classes like FutureTask which would bubble your Runnable or Callable classes’ toString() methods.
Oh well!
Here’s an example:
private static final int NUM_THREADS = 4;
private final Logger log = Logger.getLogger(getClass());
private LinkedBlockingQueue< Runnable> taskQueue = new LinkedBlockingQueue< Runnable>();
private List< Runnable> running = Collections.synchronizedList(new ArrayList());
public void doSomeStuffConcurrently() {
Executor executor =
new ThreadPoolExecutor(NUM_THREADS, NUM_THREADS,
0L, TimeUnit.MILLISECONDS,
taskQueue,
Executors.defaultThreadFactory())
{
@Override
protected < T> RunnableFuture< T> newTaskFor(final Runnable runnable, T value) {
return new FutureTask< T>(runnable, value) {
@Override
public String toString() {
return runnable.toString();
}
};
}
@Override
protected void beforeExecute(Thread t, Runnable r) {
super.beforeExecute(t, r);
running.add(r);
}
@Override
protected void afterExecute(Runnable r, Throwable t) {
super.afterExecute(r, t);
running.remove(r);
log.info("RUNNING: " + running);
}
};
executor.execute(new Runnable() {
@Override
public void run() {
// so some stuff
}
@Override
public String toString() {
return "describe the task here!";
}
});
}
