This question is inspired by:
where the problem is to run tasks on a scheduled basis, in parallel, and have individual timeouts for each job.
For example, consider this use-case:
BatchedJobHandler.TaskSupplier<Long, SleepTask> supplier =
() -> getTasksForExecution();
BatchedJobHandler.TaskCompletion<Long, SleepTask> completion =
(t, b, f) -> taskAvailable(t, b, f);
BatchedJobHandler<Long, SleepTask> batcher = new BatchedJobHandler<>(
supplier, completion, 10, 5, TimeUnit.SECONDS);
batcher.start();
Which sets up a supplier, and a completion consumer. The supplier is called every 10 seconds, and the individual jobs are expected to be complete within 5 seconds. The completion consumer will be called for each task when the task completes, or it has been running for 5 seconds, which ever comes first. If the timeout happens first, the timedout value will be true.
Note that even though the above code uses Java8 Functional interfaces, the core class is fully Java7 compatible.
I'm looking for reviews of the concurrency model, any issues you may see at all.
One issue I am aware of is that the system requires at least one listener thread for each task. Since the listener threads do nothing other than wait, they are low-impact, but, there may be a lot of them. There may be the need for a 'throttle' on the number of active tasks.
The code is intended to be compatible with Java7.
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ScheduledFuture;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicLong;
import java.util.concurrent.atomic.AtomicReference;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.ReentrantLock;
/**
* Class able to pull batches of tasks from a supplier on a schedule, and
* process them in parallel, sending results to a listener.
*
* @author rolfl
*
* @param <V>
* The value returned by the tasks
* @param <T>
* The task to run (a Callable).
*/
public final class BatchedJobHandler<V, T extends Callable<V>> {
/**
* Defines the source of the tasks to process.
*
* @param <Q>
* the task's return value.
* @param <C>
* the task's type (a Callable).
*/
public interface TaskSupplier<Q, C extends Callable<Q>> {
/**
* Produce a batch of tasks to process.
*
* @return the batch in the form of an Iterable.
*/
Iterable<C> tasks();
}
/**
* Defines the listener for completed tasks.
*
* @param <Q>
* the task's return value.
* @param <C>
* the task's type (a Callable).
*/
public interface TaskCompletion<Q, C extends Callable<Q>> {
/**
* Called when a task is complete.
* This call may happen from any thread context, and concurrently with other
* calls too. This method needs to be thread safe.
*
* @param task
* the completed task
* @param timedout
* true if the completion was the result of a timeout. Note
* that the future may complete successfully despite this
* value being true.
* @param future
* a future allowing the task result to be retrieved. Expect
* the isDone() to be false, and get() to block if timedout
* is true.
*/
void taskComplete(C task, boolean timedout, Future<Q> future);
}
private enum RunState {
PENDING, RUNNING, COMPLETE
};
private static final AtomicLong THREAD_ID = new AtomicLong();
private static final String DEFAULT_NAME = "Batched Job Handler";
private static final class CustomThreadMaker implements ThreadFactory {
private final String prefix;
private final boolean daemon;
public CustomThreadMaker(String prefix, boolean daemon) {
super();
this.prefix = prefix;
this.daemon = daemon;
}
@Override
public Thread newThread(Runnable torun) {
Thread thread = new Thread(torun, String.format("%s %d",
prefix, THREAD_ID.incrementAndGet()));
thread.setDaemon(daemon);
return thread;
}
}
private final class TimeTrackingCallable implements Callable<V> {
private final T delegate;
private final ReentrantLock lock = new ReentrantLock();
private final Condition started = lock.newCondition();
private final Condition complete = lock.newCondition();
private RunState state = RunState.PENDING;
private long delayto = 0L;
public TimeTrackingCallable(T delegate) {
this.delegate = delegate;
}
@Override
public V call() throws Exception {
long to = System.currentTimeMillis() + timeUnit.toMillis(timeout);
V result = null;
log("On Thread %s", delegate);
lock.lock();
try {
state = RunState.RUNNING;
delayto = to;
started.signalAll();
} finally {
lock.unlock();
}
try {
result = delegate.call();
log("Off Thread %s", delegate);
return result;
} finally {
lock.lock();
try {
state = RunState.COMPLETE;
complete.signalAll();
} finally {
lock.unlock();
}
}
}
public void waitOrTimeout(final Future<V> fut) {
boolean timeout = true;
log("Awaiting %s", delegate);
lock.lock();
try {
while (state == RunState.PENDING) {
log("Awaiting PENDING %s", delegate);
started.await();
}
long now = System.currentTimeMillis();
while (state == RunState.RUNNING && now < delayto) {
log("Awaiting RUNNING %s %d", delegate, delayto - now);
complete.await(delayto - now, TimeUnit.MILLISECONDS);
now = System.currentTimeMillis();
}
timeout = state != RunState.COMPLETE;
log("Awaited %s %s %s", state, delegate, timeout);
} catch (InterruptedException ie) {
// we don't handle it, we let others deal with the fallout.
Thread.currentThread().interrupt();
} finally {
lock.unlock();
}
completion.taskComplete(delegate, timeout, fut);
}
}
private final class QueueDrainer implements Runnable {
@Override
public void run() {
final List<T> drain = new ArrayList<>();
log("Starting Queue Drainer");
while (queueManager.get() == this) {
try {
drain.clear();
final T act = taskQueue.poll(delay, TimeUnit.SECONDS);
if (act != null) {
drain.add(act);
taskQueue.drainTo(drain);
}
for (final T t : drain) {
log("Adding Job %s", t);
final TimeTrackingCallable tracker = new TimeTrackingCallable(
t);
final Future<V> fut = workerPool.submit(tracker);
listenerPool.execute(new Runnable() {
@Override
public void run() {
log("Listening for %s", t);
tracker.waitOrTimeout(fut);
}
});
}
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
queueManager.compareAndSet(this, null);
return;
}
}
log("Completing Queue Drainer");
}
}
private final AtomicReference<ScheduledFuture<?>> timedTask = new AtomicReference<>();
private final AtomicReference<QueueDrainer> queueManager = new AtomicReference<>();
private final TaskSupplier<V, T> supplier;
private final TaskCompletion<V, T> completion;
private final ScheduledExecutorService scheduler;
private final ExecutorService workerPool;
private final ExecutorService listenerPool;
private final String name;
private final long delay;
private final long timeout;
private final TimeUnit timeUnit;
private final LinkedBlockingQueue<T> taskQueue = new LinkedBlockingQueue<>();
// This runnable can be scheduled repeatedly, and will add tasks to the
// queue.
private final Runnable tickRunner = new Runnable() {
@Override
public void run() {
log("Ticker Ticking");
if (queueManager.get() == null) {
log("QueueManager is empty");
QueueDrainer drainer = new QueueDrainer();
if (queueManager.compareAndSet(null, drainer)) {
listenerPool.execute(drainer);
}
log("QueueManager is now %s", queueManager.get());
}
for (T t : supplier.tasks()) {
taskQueue.add(t);
}
log("Ticker Complete");
}
};
/**
* Create a new batch processor.
*
* @param name
* The name to put on threads related to this batch processor.
* @param supplier
* The source for tasks.
* @param completion
* The destination for completion notifications.
* @param scheduler
* The ScheduledExecutorService that will be used to schedule
* task retrievals.
* @param workerPool
* The ExecutorService used to process tasks on.
* @param delay
* how long to delay from one task retrieval to the next.
* @param timeout
* how long to wait for each task to complete (once it is
* actually running).
* @param timeUnit
* the time unit used for both the delay, and the timeoout.
*/
public BatchedJobHandler(String name, TaskSupplier<V, T> supplier,
TaskCompletion<V, T> completion,
ScheduledExecutorService scheduler, ExecutorService workerPool,
int delay, int timeout, TimeUnit timeUnit) {
this.name = name;
this.supplier = supplier;
this.completion = completion;
this.scheduler = scheduler;
this.workerPool = workerPool;
this.listenerPool = Executors
.newCachedThreadPool(new CustomThreadMaker(name + " listener",
true));
this.delay = delay;
this.timeout = timeout;
this.timeUnit = timeUnit;
if (logqueue != null) {
listenerPool.execute(logRunner);
}
}
/**
* Create a new batch processor using default (independent) scheduler and
* work threads.
*
* @param supplier
* The source for tasks.
* @param completion
* The destination for completion notifications.
* @param delay
* how long to delay from one task retrieval to the next.
* @param timeout
* how long to wait for each task to complete (once it is
* actually running).
* @param timeUnit
* the time unit used for both the delay, and the timeoout.
*/
public BatchedJobHandler(TaskSupplier<V, T> supplier,
TaskCompletion<V, T> completion, int delay, int timeout,
TimeUnit timeUnit) {
this(DEFAULT_NAME, supplier, completion, Executors
.newScheduledThreadPool(1, new CustomThreadMaker(DEFAULT_NAME
+ " Schedule", true)), Executors.newFixedThreadPool(
Runtime.getRuntime().availableProcessors(),
new CustomThreadMaker(DEFAULT_NAME + " Worker", true)), delay,
timeout, timeUnit);
}
/**
* Schedule the task retriever to collect tasks to process. Ensure the
* system is ready to process tasks.
*/
public final void start() {
log("Starting");
if (timedTask.get() == null) {
// try not to start multiple scheduled tasks, but, if we do, it's
// OK.
ScheduledFuture<?> ntick = scheduler.scheduleAtFixedRate(
tickRunner, delay, delay, timeUnit);
if (!timedTask.compareAndSet(null, ntick)) {
// some other thread started and raced us, and won.
ntick.cancel(true);
}
}
}
/**
* Stop the scheduled task retriever. No further tasks will be collected.
* Currently running tasks will complete.
*/
public final void stop() {
log("Stopping");
ScheduledFuture<?> tokill = timedTask.getAndSet(null);
if (tokill != null) {
tokill.cancel(false);
}
}
@Override
public String toString() {
return String.format("'%s' state: %s", name,
timedTask.get() == null ? "idle" : "running");
}
/*
* Cheat way to do logging.... Replace with real logging later.
*/
private static final boolean LOG_ENABLED = Boolean
.getBoolean("BATCHJOBDEBUG");
private final LinkedBlockingQueue<String> logqueue = LOG_ENABLED ? new LinkedBlockingQueue<>()
: null;
private final Runnable logRunner = new Runnable() {
public void run() {
log("Starting log in %s", Thread.currentThread().getName());
while (true) {
try {
System.out.println(logqueue.take());
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}
};
private final void log(String format, Object... parms) {
if (logqueue != null) {
logqueue.add(String.format("%tc [%s] - %s", System
.currentTimeMillis(), Thread.currentThread().getName(),
String.format(format, parms)));
}
}
}
I have tested this code with the following system that's not for review, but provides an idea of usage. Note, it uses some Java8 concepts...
import java.util.ArrayList;
import java.util.List;
import java.util.Random;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicLong;
public class SleepJobs {
private static final class SleepTask implements Callable<Long> {
private final int id;
private final long time;
private final AtomicLong start = new AtomicLong(-1);
public SleepTask(int id, long time) {
super();
this.id = id;
this.time = time;
}
@Override
public Long call() throws Exception {
long begin = System.currentTimeMillis();
start.set(begin);
System.out.println("Sleep " + time);
Thread.sleep(time);
return System.currentTimeMillis() - begin;
}
public int getId() {
return id;
}
public long getTime() {
return time;
}
public long getStart() {
return start.get();
}
@Override
public String toString() {
return String.format("Job %d with sleep time %d", id, time);
}
}
private static final Random rand = new Random();
private static final AtomicInteger idgen = new AtomicInteger();
private static Iterable<SleepTask> getTasksForExecution() {
int cnt = 5 + rand.nextInt(10);
System.out.println("Getting " + cnt + " tasks");
List<SleepTask> ret = new ArrayList<>(cnt);
for (int i = 0; i < cnt; i++) {
ret.add(new SleepTask(idgen.incrementAndGet(), 2000 + rand
.nextInt(5000)));
}
return ret;
}
private static void taskAvailable(SleepTask task, boolean timedout,
Future<Long> control) {
try {
long duration = control.isDone() ? control.get() : -1;
long act = System.currentTimeMillis() - task.getStart();
System.out.printf(
"Task %d at %d for %d %s after %d with result %d%n", task
.getId(), task.getStart(), task.getTime(),
timedout ? "bailed " : "completed", act, duration);
} catch (ExecutionException | InterruptedException ee) {
long act = System.currentTimeMillis() - task.getStart();
System.out.printf(
"Task %d at %d for %d %s after %d with EXCEPTION %s%n",
task.getId(), task.getStart(), task.getTime(),
timedout ? "bailed " : "completed", act, ee.getMessage());
ee.printStackTrace();
}
}
public static void main(String[] args) throws InterruptedException {
BatchedJobHandler.TaskSupplier<Long, SleepTask> supplier = () -> getTasksForExecution();
BatchedJobHandler.TaskCompletion<Long, SleepTask> completion = (t, b, f) -> taskAvailable(
t, b, f);
BatchedJobHandler<Long, SleepTask> batcher = new BatchedJobHandler<>(
supplier, completion, 10, 5, TimeUnit.SECONDS);
Thread.sleep(1000);
batcher.start();
Thread.sleep(30000);
batcher.stop();
System.out.println(batcher.toString());
}
}
If you run the above code, and set -DBATCHJOBDEBUG=true, then the code will output additional debug messages.
teverywhere... \$\endgroup\$super();in the constructor ofCustomThreadMaker? \$\endgroup\$