Spring/spring-batch
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Reviewed concurrency in TaskExecutorRepeatTemplate. Removed a blocking call from a constructor, and added some docos.

Browse Source
This commit is contained in:
dsyer
2008-03-13 14:14:43 +00:00
parent cc39916c19
commit 570c04e6f3
2 changed files with 107 additions and 55 deletions
Download Patch File Download Diff File Expand all files Collapse all files

46
spring-batch-infrastructure/src/main/java/org/springframework/batch/repeat/support/AbstractResultQueue.java
View File

@@ -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;
}

116
spring-batch-infrastructure/src/main/java/org/springframework/batch/repeat/support/TaskExecutorRepeatTemplate.java
View File

@@ -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.<br/>
* Provides {@link RepeatOperations} support including interceptors that can be
* used to modify or monitor the behaviour at run time.<br/>
*
* 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).<br/>
*
* 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).<br/>
* 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).<br/>
*
* @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.
*/