BATCH-1409: By using a special queue we can force the repeat template to block only on calling thread, not on workers.
This commit is contained in:
dsyer
@@ -0,0 +1,175 @@
|
||||
/*
|
||||
* Copyright 2002-2007 the original author or authors.
|
||||
*
|
||||
* Licensed under the Apache License, Version 2.0 (the "License");
|
||||
* you may not use this file except in compliance with the License.
|
||||
* You may obtain a copy of the License at
|
||||
*
|
||||
* http://www.apache.org/licenses/LICENSE-2.0
|
||||
*
|
||||
* Unless required by applicable law or agreed to in writing, software
|
||||
* distributed under the License is distributed on an "AS IS" BASIS,
|
||||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
* See the License for the specific language governing permissions and
|
||||
* limitations under the License.
|
||||
*/
|
||||
|
||||
package org.springframework.batch.repeat.support;
|
||||
|
||||
import java.util.Comparator;
|
||||
import java.util.NoSuchElementException;
|
||||
import java.util.concurrent.BlockingQueue;
|
||||
import java.util.concurrent.PriorityBlockingQueue;
|
||||
import java.util.concurrent.Semaphore;
|
||||
|
||||
import org.springframework.batch.repeat.RepeatStatus;
|
||||
|
||||
/**
|
||||
* An implementation of the {@link ResultQueue} that throttles the number of
|
||||
* expected results, limiting it to a maximum at any given time.
|
||||
*
|
||||
* @author Dave Syer
|
||||
*/
|
||||
public class ResultHolderResultQueue implements ResultQueue<ResultHolder> {
|
||||
|
||||
// Accumulation of result objects as they finish.
|
||||
private final BlockingQueue<ResultHolder> results;
|
||||
|
||||
// Accumulation of dummy objects flagging expected results in the future.
|
||||
private final Semaphore waits;
|
||||
|
||||
private final Object lock = new Object();
|
||||
|
||||
private volatile int count = 0;
|
||||
|
||||
/**
|
||||
* @param throttleLimit the maximum number of results that can be expected
|
||||
* at any given time.
|
||||
*/
|
||||
public ResultHolderResultQueue(int throttleLimit) {
|
||||
results = new PriorityBlockingQueue<ResultHolder>(throttleLimit, new ResultHolderComparator());
|
||||
waits = new Semaphore(throttleLimit);
|
||||
}
|
||||
|
||||
public boolean isEmpty() {
|
||||
return results.isEmpty();
|
||||
}
|
||||
|
||||
/*
|
||||
* (non-Javadoc)
|
||||
*
|
||||
* @see org.springframework.batch.repeat.support.ResultQueue#isExpecting()
|
||||
*/
|
||||
public boolean isExpecting() {
|
||||
// Base the decision about whether we expect more results on a
|
||||
// counter of the number of expected results actually collected.
|
||||
// Do not synchronize! Otherwise put and expect can deadlock.
|
||||
return count > 0;
|
||||
}
|
||||
|
||||
/**
|
||||
* Tell the queue to expect one more result. Blocks until a new result is
|
||||
* available if already expecting too many (as determined by the throttle
|
||||
* limit).
|
||||
*
|
||||
* @see ResultQueue#expect()
|
||||
*/
|
||||
public void expect() throws InterruptedException {
|
||||
synchronized (lock) {
|
||||
waits.acquire();
|
||||
count++;
|
||||
}
|
||||
}
|
||||
|
||||
public void put(ResultHolder holder) throws IllegalArgumentException {
|
||||
if (!isExpecting()) {
|
||||
throw new IllegalArgumentException("Not expecting a result. Call expect() before put().");
|
||||
}
|
||||
results.add(holder);
|
||||
// Take from the waits queue now to allow another result to
|
||||
// accumulate. But don't decrement the counter.
|
||||
waits.release();
|
||||
synchronized (lock) {
|
||||
lock.notifyAll();
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Get the next result as soon as it becomes available. <br/>
|
||||
* <br/>
|
||||
* Release result immediately if:
|
||||
* <ul>
|
||||
* <li>There is a result that is continuable.</li>
|
||||
* </ul>
|
||||
* Otherwise block if either:
|
||||
* <ul>
|
||||
* <li>There is no result (as per contract of {@link ResultQueue}).</li>
|
||||
* <li>The number of results is less than the number expected.</li>
|
||||
* </ul>
|
||||
* Error if either:
|
||||
* <ul>
|
||||
* <li>Not expecting.</li>
|
||||
* <li>Interrupted.</li>
|
||||
* </ul>
|
||||
*
|
||||
* @see ResultQueue#take()
|
||||
*/
|
||||
public ResultHolder take() throws NoSuchElementException, InterruptedException {
|
||||
if (!isExpecting()) {
|
||||
throw new NoSuchElementException("Not expecting a result. Call expect() before take().");
|
||||
}
|
||||
ResultHolder value;
|
||||
synchronized (lock) {
|
||||
value = results.take();
|
||||
if (isContinuable(value)) {
|
||||
// Decrement the counter only when the result is collected.
|
||||
count--;
|
||||
return value;
|
||||
}
|
||||
}
|
||||
results.put(value);
|
||||
synchronized (lock) {
|
||||
while (count > results.size()) {
|
||||
lock.wait();
|
||||
}
|
||||
value = results.take();
|
||||
count--;
|
||||
}
|
||||
return value;
|
||||
}
|
||||
|
||||
private boolean isContinuable(ResultHolder value) {
|
||||
return value.getResult() != null && value.getResult().isContinuable();
|
||||
}
|
||||
|
||||
/**
|
||||
* Compares ResultHolders so that one that is continuable ranks lowest.
|
||||
*
|
||||
* @author Dave Syer
|
||||
*
|
||||
*/
|
||||
private static class ResultHolderComparator implements Comparator<ResultHolder> {
|
||||
public int compare(ResultHolder h1, ResultHolder h2) {
|
||||
RepeatStatus result1 = h1.getResult();
|
||||
RepeatStatus result2 = h2.getResult();
|
||||
if (result1 == null && result2 == null) {
|
||||
return 0;
|
||||
}
|
||||
if (result1 == null) {
|
||||
return -1;
|
||||
}
|
||||
else if (result2 == null) {
|
||||
return 1;
|
||||
}
|
||||
if ((result1.isContinuable() && result2.isContinuable())
|
||||
|| (!result1.isContinuable() && !result2.isContinuable())) {
|
||||
return 0;
|
||||
}
|
||||
if (result1.isContinuable()) {
|
||||
return -1;
|
||||
}
|
||||
return 1;
|
||||
}
|
||||
}
|
||||
|
||||
}
|
||||
@@ -16,8 +16,6 @@
|
||||
|
||||
package org.springframework.batch.repeat.support;
|
||||
|
||||
import org.apache.commons.logging.Log;
|
||||
import org.apache.commons.logging.LogFactory;
|
||||
import org.springframework.batch.repeat.RepeatCallback;
|
||||
import org.springframework.batch.repeat.RepeatContext;
|
||||
import org.springframework.batch.repeat.RepeatException;
|
||||
@@ -103,7 +101,6 @@ public class TaskExecutorRepeatTemplate extends RepeatTemplate {
|
||||
ExecutingRunnable runnable = null;
|
||||
|
||||
ResultQueue<ResultHolder> queue = ((ResultQueueInternalState) state).getResultQueue();
|
||||
ActivityBarrier lock = ((ResultQueueInternalState) state).getLock();
|
||||
|
||||
do {
|
||||
|
||||
@@ -111,7 +108,7 @@ public class TaskExecutorRepeatTemplate extends RepeatTemplate {
|
||||
* Wrap the callback in a runnable that will add its result to the
|
||||
* queue when it is ready.
|
||||
*/
|
||||
runnable = new ExecutingRunnable(callback, context, queue, lock);
|
||||
runnable = new ExecutingRunnable(callback, context, queue);
|
||||
|
||||
/**
|
||||
* Tell the runnable that it can expect a result. This could have
|
||||
@@ -159,14 +156,11 @@ public class TaskExecutorRepeatTemplate extends RepeatTemplate {
|
||||
protected boolean waitForResults(RepeatInternalState state) {
|
||||
|
||||
ResultQueue<ResultHolder> queue = ((ResultQueueInternalState) state).getResultQueue();
|
||||
ActivityBarrier lock = ((ResultQueueInternalState) state).getLock();
|
||||
|
||||
boolean result = true;
|
||||
|
||||
while (queue.isExpecting()) {
|
||||
|
||||
lock.release();
|
||||
|
||||
/*
|
||||
* Careful that no runnables that are not going to finish ever get
|
||||
* onto the queue, else this may block forever.
|
||||
@@ -219,17 +213,13 @@ public class TaskExecutorRepeatTemplate extends RepeatTemplate {
|
||||
|
||||
private volatile Throwable error;
|
||||
|
||||
private final ActivityBarrier lock;
|
||||
|
||||
public ExecutingRunnable(RepeatCallback callback, RepeatContext context, ResultQueue<ResultHolder> queue,
|
||||
ActivityBarrier lock) {
|
||||
public ExecutingRunnable(RepeatCallback callback, RepeatContext context, ResultQueue<ResultHolder> queue) {
|
||||
|
||||
super();
|
||||
|
||||
this.callback = callback;
|
||||
this.context = context;
|
||||
this.queue = queue;
|
||||
this.lock = lock;
|
||||
|
||||
}
|
||||
|
||||
@@ -260,7 +250,10 @@ public class TaskExecutorRepeatTemplate extends RepeatTemplate {
|
||||
RepeatSynchronizationManager.register(context);
|
||||
}
|
||||
|
||||
lock.acquire();
|
||||
if (logger.isDebugEnabled()) {
|
||||
logger.debug("Repeat operation about to start at count=" + context.getStartedCount());
|
||||
}
|
||||
|
||||
result = callback.doInIteration(context);
|
||||
|
||||
}
|
||||
@@ -269,8 +262,6 @@ public class TaskExecutorRepeatTemplate extends RepeatTemplate {
|
||||
}
|
||||
finally {
|
||||
|
||||
lock.release(isComplete(context), result);
|
||||
|
||||
if (clearContext) {
|
||||
RepeatSynchronizationManager.clear();
|
||||
}
|
||||
@@ -313,22 +304,12 @@ public class TaskExecutorRepeatTemplate extends RepeatTemplate {
|
||||
|
||||
private final ResultQueue<ResultHolder> results;
|
||||
|
||||
private final ActivityBarrier lock;
|
||||
|
||||
/**
|
||||
* @param throttleLimit the throttle limit for the result queue
|
||||
*/
|
||||
public ResultQueueInternalState(int throttleLimit) {
|
||||
super();
|
||||
this.results = new ThrottleLimitResultQueue<ResultHolder>(throttleLimit);
|
||||
this.lock = new ActivityBarrier();
|
||||
}
|
||||
|
||||
/**
|
||||
* @return the lock instance
|
||||
*/
|
||||
public ActivityBarrier getLock() {
|
||||
return lock;
|
||||
this.results = new ResultHolderResultQueue(throttleLimit);
|
||||
}
|
||||
|
||||
/**
|
||||
@@ -340,137 +321,4 @@ public class TaskExecutorRepeatTemplate extends RepeatTemplate {
|
||||
|
||||
}
|
||||
|
||||
/**
|
||||
* <p>
|
||||
* Encapsulates the locking concerns needed when waiting for concurrent
|
||||
* repeat tasks to complete. Some tasks may return FINISHED before the rest
|
||||
* have completed, and furthermore one or more of the rest might be
|
||||
* CONTINUABLE or might have ended in an error and still have more work to
|
||||
* do.
|
||||
* </p>
|
||||
*
|
||||
* <p>
|
||||
* Uses wait and notify to co-ordinate the end of the repeat iterations, so
|
||||
* that the process can end only when either the completion policy signals
|
||||
* completion, or all workers have signalled that they are FINISHED (and had
|
||||
* a chance to accept more work if they are initially CONTINUABLE).
|
||||
* </p>
|
||||
*
|
||||
* <p>
|
||||
* The net effect of using this lock protocol is that all workers (up to the
|
||||
* throttle limit) may perform no work for the last iteration and return
|
||||
* FINISHED. Contrast this with the single threaded case where the only one
|
||||
* call resulting in FINISHED is made to the repeat callback.
|
||||
* </p>
|
||||
*
|
||||
* @author Dave Syer
|
||||
*/
|
||||
private static class ActivityBarrier {
|
||||
|
||||
private static Log logger = LogFactory.getLog(ActivityBarrier.class);
|
||||
|
||||
private volatile int active = 0;
|
||||
|
||||
private volatile boolean paused = false;
|
||||
|
||||
private final Object lock = new Object();
|
||||
|
||||
/**
|
||||
* Atomic acquisition of resources needed to track concurrent execution.
|
||||
* Call this method before every repeat callback execution.
|
||||
*/
|
||||
public void acquire() {
|
||||
synchronized (lock) {
|
||||
active++;
|
||||
paused = false;
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* <p>
|
||||
* Release the resources acquired in {@link #acquire()}. Call this
|
||||
* method in a finally block after every repeat callback execution.
|
||||
* </p>
|
||||
*
|
||||
* @param complete true if we know from the completion policy that the
|
||||
* iteration should end
|
||||
* @param result the latest result from a callback
|
||||
*/
|
||||
public void release(boolean complete, RepeatStatus result) {
|
||||
|
||||
boolean stillActive = false;
|
||||
|
||||
synchronized (lock) {
|
||||
|
||||
active--;
|
||||
stillActive = active > 0 || paused;
|
||||
|
||||
if (logger.isDebugEnabled()) {
|
||||
logger.debug("Completed callback with result = " + result + ", " + active
|
||||
+ " active callbacks, and paused=" + paused);
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
if (result == RepeatStatus.FINISHED) {
|
||||
if (stillActive) {
|
||||
logger.debug("Waiting for other active callbacks to finish.");
|
||||
synchronized (lock) {
|
||||
while (stillActive) {
|
||||
try {
|
||||
/*
|
||||
* Need a timed wait here to avoid deadlock in
|
||||
* the case that stillActive changed its value
|
||||
* between where it was computed and this wait.
|
||||
* In that case another thread might have
|
||||
* already sent the notify() message and then we
|
||||
* would miss it here.
|
||||
*/
|
||||
lock.wait(2000L);
|
||||
stillActive = active > 0 || paused;
|
||||
}
|
||||
catch (InterruptedException e) {
|
||||
logger.info("Interrupted waiting for active callbacks");
|
||||
Thread.currentThread().interrupt();
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
else {
|
||||
synchronized (lock) {
|
||||
logger.debug("Notifying other waiting callbacks on finish.");
|
||||
paused = false;
|
||||
lock.notifyAll();
|
||||
}
|
||||
}
|
||||
}
|
||||
else {
|
||||
if (complete) {
|
||||
synchronized (lock) {
|
||||
logger.debug("Notifying other waiting callbacks on policy based completion.");
|
||||
paused = false;
|
||||
lock.notifyAll();
|
||||
}
|
||||
}
|
||||
else {
|
||||
synchronized (lock) {
|
||||
paused = true;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
/**
|
||||
* Release all waiting workers unconditionally. Call this method when
|
||||
* iteration has ended in case any workers are still waiting.
|
||||
*/
|
||||
public void release() {
|
||||
synchronized (lock) {
|
||||
lock.notifyAll();
|
||||
}
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user