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

Add TODO for possible race condition

Browse Source
This commit is contained in:
dsyer
2008-02-06 07:08:18 +00:00
parent a0c4af8351
commit c86454fbe0
1 changed files with 30 additions and 31 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -69,10 +69,8 @@ public class TaskExecutorRepeatTemplate extends RepeatTemplate {
/**
* Setter for task executor to be used to run the individual item callbacks.
*
* @param taskExecutor
* a TaskExecutor
* @throws IllegalArgumentException
* if the argument is null
* @param taskExecutor a TaskExecutor
* @throws IllegalArgumentException if the argument is null
*/
public void setTaskExecutor(TaskExecutor taskExecutor) {
Assert.notNull(taskExecutor);
@@ -87,11 +85,10 @@ public class TaskExecutorRepeatTemplate extends RepeatTemplate {
* 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)
protected ExitStatus getNextResult(RepeatContext context, RepeatCallback callback, RepeatInternalState state)
throws Throwable {
ExecutingRunnable runnable = null;
@@ -126,7 +123,8 @@ public class TaskExecutorRepeatTemplate extends RepeatTemplate {
Object result;
try {
result = queue.take().getResult();
} catch (InterruptedException e) {
}
catch (InterruptedException e) {
Thread.currentThread().interrupt();
throw e;
}
@@ -159,13 +157,15 @@ public class TaskExecutorRepeatTemplate extends RepeatTemplate {
Object value;
try {
value = future.getResult();
} catch (InterruptedException e) {
}
catch (InterruptedException e) {
Thread.currentThread().interrupt();
value = e;
}
if (value instanceof Throwable) {
state.getThrowables().add(value);
} else {
}
else {
ExitStatus status = (ExitStatus) value;
result = result && canContinue(status);
executeAfterInterceptors(future.getContext(), status);
@@ -173,8 +173,7 @@ public class TaskExecutorRepeatTemplate extends RepeatTemplate {
}
Assert.state(futures.isEmpty(),
"Future results should be empty at end of batch.");
Assert.state(futures.isEmpty(), "Future results should be empty at end of batch.");
return result;
}
@@ -199,8 +198,7 @@ public class TaskExecutorRepeatTemplate extends RepeatTemplate {
Object result;
public ExecutingRunnable(RepeatCallback callback,
RepeatContext context, ResultQueue queue) {
public ExecutingRunnable(RepeatCallback callback, RepeatContext context, ResultQueue queue) {
super();
@@ -224,9 +222,11 @@ public class TaskExecutorRepeatTemplate extends RepeatTemplate {
public void run() {
try {
result = callback.doInIteration(context);
} catch (Exception e) {
}
catch (Exception e) {
result = e;
} finally {
}
finally {
queue.put(this);
}
}
@@ -235,8 +235,7 @@ public class TaskExecutorRepeatTemplate extends RepeatTemplate {
* Get the result - never blocks because the queue manages waiting for
* the task to finish.
*
* @throws InterruptedException
* if the thread is interrupted.
* @throws InterruptedException if the thread is interrupted.
*/
public Object getResult() throws InterruptedException {
return result;
@@ -296,15 +295,17 @@ public class TaskExecutorRepeatTemplate extends RepeatTemplate {
synchronized (lock) {
count++;
}
} catch (InterruptedException e) {
}
catch (InterruptedException e) {
Thread.currentThread().interrupt();
throw new RepeatException(
"InterruptedException waiting for to acquire lock on input.");
throw new RepeatException("InterruptedException waiting for to acquire lock on input.");
}
}
public void put(ResultHolder holder) {
// There should be no need to block here:
// There should be no need to block here, or to use offer():
// TODO: fix race condition where many calls come into put() at once
// and the queue fills up...
results.add(holder);
// Take from the waits queue now to allow another result to
// accumulate. But don't decrement the counter.
@@ -319,10 +320,10 @@ public class TaskExecutorRepeatTemplate extends RepeatTemplate {
// 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("Interrupted while waiting for result.");
}
return value;
}
@@ -343,9 +344,8 @@ public class TaskExecutorRepeatTemplate extends RepeatTemplate {
* until it is ready.
*
* @return the result.
* @throws InterruptedException
* if the thread is interrupted while waiting for the
* result.
* @throws InterruptedException if the thread is interrupted while
* waiting for the result.
* @throws IllegalStateException
*/
Object getResult() throws InterruptedException;
@@ -367,8 +367,7 @@ public class TaskExecutorRepeatTemplate extends RepeatTemplate {
* 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.
* @param throttleLimit the throttleLimit to set.
*/
public void setThrottleLimit(int throttleLimit) {
this.throttleLimit = throttleLimit;