diff --git a/spring-batch-infrastructure/src/main/java/org/springframework/batch/repeat/support/AbstractResultQueue.java b/spring-batch-infrastructure/src/main/java/org/springframework/batch/repeat/support/AbstractResultQueue.java
index 16f3b2635..daf89b57c 100644
--- a/spring-batch-infrastructure/src/main/java/org/springframework/batch/repeat/support/AbstractResultQueue.java
+++ b/spring-batch-infrastructure/src/main/java/org/springframework/batch/repeat/support/AbstractResultQueue.java
@@ -19,9 +19,11 @@ package org.springframework.batch.repeat.support;
import org.springframework.batch.repeat.RepeatException;
/**
- * Abstract implementation that can be extended for both Backport Concurrent and JDK 5 Concurrent
+ * Abstract implementation that can be extended for both Backport Concurrent and
+ * JDK 5 Concurrent.
*
* @author Ben Hale
+ * @author Dave Syer
*/
abstract class AbstractResultQueue extends RepeatInternalStateSupport implements ResultQueue {
@@ -31,14 +33,14 @@ abstract class AbstractResultQueue extends RepeatInternalStateSupport implements
// Arbitrary lock object.
Object hold = new Object();
- // Counter to monitor the difference between expected and actually collected results. When this
- // reaches zero there are really no more results.
+ // Counter to monitor the difference between expected and actually collected
+ // results. When this reaches zero there are really no more results.
volatile int count = 0;
public boolean isExpecting() {
synchronized (lock) {
- // Base the decision about whether we expect more results on a counter of the number of
- // expected results actually collected.
+ // Base the decision about whether we expect more results on a
+ // counter of the number of expected results actually collected.
return count > 0;
}
}
@@ -49,15 +51,17 @@ abstract class AbstractResultQueue extends RepeatInternalStateSupport implements
aquireWait();
count++;
}
- } catch (InterruptedException e) {
+ }
+ catch (InterruptedException e) {
Thread.currentThread().interrupt();
throw new RepeatException("InterruptedException waiting for to acquire lock on input.");
}
}
public void put(ResultHolder holder) {
- // There should be no need to block here, or to use offer(), but apparently the add()
- // sometimes takes so long on the CI build that the queue fills up...
+ // There should be no need to block here, or to use offer(), but
+ // apparently the add() sometimes takes so long on the CI build that the
+ // queue fills up, so we synchronize here...
synchronized (hold) {
addResult(holder);
// Take from the waits queue now to allow another result to
@@ -74,19 +78,41 @@ abstract class AbstractResultQueue extends RepeatInternalStateSupport implements
// Decrement the counter only when the result is collected.
count--;
}
- } catch (InterruptedException e) {
+ }
+ catch (InterruptedException e) {
Thread.currentThread().interrupt();
- throw new RepeatException("Interrupted while waiting for result.");
+ throw new RepeatException("InterruptedException while waiting for result.");
}
return value;
}
+ /**
+ * Acquire permission for one more task on the queue.
+ *
+ * @throws InterruptedException
+ */
protected abstract void aquireWait() throws InterruptedException;
+ /**
+ * Release the permit that we were holding while a task was processed.
+ */
protected abstract void releaseWait();
+ /**
+ * Add a {@link ResultHolder} with a finished result to the queue for
+ * collection. Should not block. May throw an exception to signal that the
+ * queue is full, but that would be an unexpected condition.
+ *
+ * @param resultHolder a {@link ResultHolder}
+ */
protected abstract void addResult(ResultHolder resultHolder);
+ /**
+ * Obtain a result from the queue, blocking until one becomes available.
+ *
+ * @return a {@link ResultHolder} with the completed result
+ * @throws InterruptedException if an interrupt is signalled
+ */
protected abstract ResultHolder takeResult() throws InterruptedException;
}
diff --git a/spring-batch-infrastructure/src/main/java/org/springframework/batch/repeat/support/TaskExecutorRepeatTemplate.java b/spring-batch-infrastructure/src/main/java/org/springframework/batch/repeat/support/TaskExecutorRepeatTemplate.java
index 66a866ef4..449bf1808 100644
--- a/spring-batch-infrastructure/src/main/java/org/springframework/batch/repeat/support/TaskExecutorRepeatTemplate.java
+++ b/spring-batch-infrastructure/src/main/java/org/springframework/batch/repeat/support/TaskExecutorRepeatTemplate.java
@@ -27,19 +27,22 @@ import org.springframework.core.task.TaskExecutor;
import org.springframework.util.Assert;
/**
- * Provides {@link RepeatOperations} support including interceptors that can be used to modify or monitor the behaviour
- * at run time.
+ * Provides {@link RepeatOperations} support including interceptors that can be
+ * used to modify or monitor the behaviour at run time.
*
- * This implementation is sufficient to be used to configure transactional behaviour for each item by making the
- * {@link RepeatCallback} transactional, or for the whole batch by making the execute method transactional (but only
+ * This implementation is sufficient to be used to configure transactional
+ * behaviour for each item by making the {@link RepeatCallback} transactional,
+ * or for the whole batch by making the execute method transactional (but only
* then if the task executor is synchronous).
*
- * This class is thread safe if its collaborators are thread safe (interceptors, terminationPolicy, callback). Normally
- * this will be the case, but clients need to be aware that if the task executor is asynchronous, then the other
- * collaborators should be also. In particular the {@link RepeatCallback} that is wrapped in the execute method must be
- * thread safe - often it is based on some form of data source, which itself should be both thread safe and
- * transactional (multiple threads could be accessing it at any given time, and each thread would have its own
- * transaction).
+ * This class is thread safe if its collaborators are thread safe (interceptors,
+ * terminationPolicy, callback). Normally this will be the case, but clients
+ * need to be aware that if the task executor is asynchronous, then the other
+ * collaborators should be also. In particular the {@link RepeatCallback} that
+ * is wrapped in the execute method must be thread safe - often it is based on
+ * some form of data source, which itself should be both thread safe and
+ * transactional (multiple threads could be accessing it at any given time, and
+ * each thread would have its own transaction).
*
* @author Dave Syer
*
@@ -47,7 +50,8 @@ import org.springframework.util.Assert;
public class TaskExecutorRepeatTemplate extends RepeatTemplate {
/**
- * Default limit for maximum number of concurrent unfinished results allowed by the template.
+ * Default limit for maximum number of concurrent unfinished results allowed
+ * by the template.
* {@link #getNextResult(RepeatContext, RepeatCallback, TerminationContext, List)}.
*/
public static final int DEFAULT_THROTTLE_LIMIT = 4;
@@ -68,16 +72,18 @@ public class TaskExecutorRepeatTemplate extends RepeatTemplate {
}
/**
- * Use the {@link #taskExecutor} to generate a result. The internal state in this case is a queue of unfinished
- * result holders of type {@link ResultHolder}. The holder with the return value should not be on the queue when
- * this method exits. The queue is scoped in the calling method so there is no need to synchronize access.
+ * Use the {@link #taskExecutor} to generate a result. The internal state in
+ * this case is a queue of unfinished result holders of type
+ * {@link ResultHolder}. The holder with the return value should not be on
+ * the queue when this method exits. The queue is scoped in the calling
+ * method so there is no need to synchronize access.
*
* @see org.springframework.batch.repeat.support.AbstracBatchemplate#getNextResult(org.springframework.batch.item.RepeatContext,
- * org.springframework.batch.repeat.RepeatCallback, org.springframework.batch.TerminationContext,
- * java.util.List)
+ * org.springframework.batch.repeat.RepeatCallback,
+ * org.springframework.batch.TerminationContext, java.util.List)
*/
protected ExitStatus getNextResult(RepeatContext context, RepeatCallback callback, RepeatInternalState state)
- throws Throwable {
+ throws Throwable {
ExecutingRunnable runnable = null;
@@ -86,23 +92,33 @@ public class TaskExecutorRepeatTemplate extends RepeatTemplate {
do {
/*
- * Wrap the callback in a runnable that will add its result to the queue when it is ready.
+ * Wrap the callback in a runnable that will add its result to the
+ * queue when it is ready.
*/
runnable = new ExecutingRunnable(callback, context, queue);
+ /**
+ * Tell the runnable that it can expect a result. This could have
+ * been in-lined with the constructor, but it might block, so it's
+ * better to do it here, since we have the option (it's a private
+ * class).
+ */
+ runnable.expect();
+
/*
* Start the task possibly concurrently / in the future.
*/
taskExecutor.execute(runnable);
/*
- * Allow termination policy to update its state. This must happen immediately before or after the call to
- * the task executor.
+ * Allow termination policy to update its state. This must happen
+ * immediately before or after the call to the task executor.
*/
update(context);
/*
- * Keep going until we get a result that is finished, or early termination...
+ * Keep going until we get a result that is finished, or early
+ * termination...
*/
} while (queue.isEmpty() && !isComplete(context));
@@ -114,27 +130,29 @@ public class TaskExecutorRepeatTemplate extends RepeatTemplate {
}
/**
- * Wait for all the results to appear on the queue and execute the after interceptors for each one.
+ * Wait for all the results to appear on the queue and execute the after
+ * interceptors for each one.
*
* @see org.springframework.batch.repeat.support.RepeatTemplate#waitForResults(org.springframework.batch.repeat.support.RepeatInternalState)
*/
protected boolean waitForResults(RepeatInternalState state) {
- ResultQueue futures = (ResultQueue) state;
+ ResultQueue queue = (ResultQueue) state;
boolean result = true;
- while (futures.isExpecting()) {
+ while (queue.isExpecting()) {
/*
- * Careful that no runnables that are not going to finish ever get onto the queue, else this may block
- * forever.
+ * Careful that no runnables that are not going to finish ever get
+ * onto the queue, else this may block forever.
*/
- ResultHolder future = (ResultHolder) futures.take();
+ ResultHolder future = (ResultHolder) queue.take();
if (future.getError() != null) {
state.getThrowables().add(future.getError());
- } else {
+ }
+ else {
ExitStatus status = future.getResult();
result = result && canContinue(status);
executeAfterInterceptors(future.getContext(), status);
@@ -142,7 +160,7 @@ public class TaskExecutorRepeatTemplate extends RepeatTemplate {
}
- Assert.state(futures.isEmpty(), "Future results should be empty at end of batch.");
+ Assert.state(queue.isEmpty(), "Future results queue should be empty at end of batch.");
return result;
}
@@ -177,38 +195,44 @@ public class TaskExecutorRepeatTemplate extends RepeatTemplate {
this.context = context;
this.queue = queue;
- /*
- * Tell the queue to expect a result.
- */
- queue.expect();
-
}
/**
- * Execute the batch callback, and store the result, or any exception that is thrown for retrieval later by
- * caller.
+ * Tell the queue to expect a result.
+ */
+ public void expect() {
+ queue.expect();
+ }
+
+ /**
+ * Execute the batch callback, and store the result, or any exception
+ * that is thrown for retrieval later by caller.
*
* @see java.lang.Runnable#run()
*/
public void run() {
try {
result = callback.doInIteration(context);
- } catch (Exception e) {
+ }
+ catch (Exception e) {
error = e;
- } finally {
+ }
+ finally {
queue.put(this);
}
}
/**
- * Get the result - never blocks because the queue manages waiting for the task to finish.
+ * Get the result - never blocks because the queue manages waiting for
+ * the task to finish.
*/
public ExitStatus getResult() {
return result;
}
/**
- * Get the error - never blocks because the queue manages waiting for the task to finish.
+ * Get the error - never blocks because the queue manages waiting for
+ * the task to finish.
*/
public Throwable getError() {
return error;
@@ -224,11 +248,13 @@ public class TaskExecutorRepeatTemplate extends RepeatTemplate {
}
/**
- * Public setter for the throttle limit. The throttle limit is the largest number of concurrent tasks that can be
- * executing at one time - if a new task arrives and the throttle limit is breached we wait for one of the executing
- * tasks to finish before submitting the new one to the {@link TaskExecutor}. Default value is
- * {@value #DEFAULT_THROTTLE_LIMIT}. N.B. when used with a thread pooled {@link TaskExecutor} it doesn't make sense
- * for the throttle limit to be less than the thread pool size.
+ * Public setter for the throttle limit. The throttle limit is the largest
+ * number of concurrent tasks that can be executing at one time - if a new
+ * task arrives and the throttle limit is breached we wait for one of the
+ * executing tasks to finish before submitting the new one to the
+ * {@link TaskExecutor}. Default value is {@value #DEFAULT_THROTTLE_LIMIT}.
+ * N.B. when used with a thread pooled {@link TaskExecutor} it doesn't make
+ * sense for the throttle limit to be less than the thread pool size.
*
* @param throttleLimit the throttleLimit to set.
*/