diff --git a/spring-batch-infrastructure/src/main/java/org/springframework/batch/repeat/support/ResultHolderResultQueue.java b/spring-batch-infrastructure/src/main/java/org/springframework/batch/repeat/support/ResultHolderResultQueue.java new file mode 100644 index 000000000..e282a0dd7 --- /dev/null +++ b/spring-batch-infrastructure/src/main/java/org/springframework/batch/repeat/support/ResultHolderResultQueue.java @@ -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 { + + // Accumulation of result objects as they finish. + private final BlockingQueue 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(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.
+ *
+ * Release result immediately if: + *
    + *
  • There is a result that is continuable.
    • + *
+ * Otherwise block if either: + *
    + *
  • There is no result (as per contract of {@link ResultQueue}).
    • + *
  • The number of results is less than the number expected.
    • + *
+ * Error if either: + *
    + *
  • Not expecting.
    • + *
  • Interrupted.
    • + *
+ * + * @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 { + 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; + } + } + +} 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 411865fa5..71a8e0c71 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 @@ -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 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 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 queue, - ActivityBarrier lock) { + public ExecutingRunnable(RepeatCallback callback, RepeatContext context, ResultQueue 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 results; - private final ActivityBarrier lock; - /** * @param throttleLimit the throttle limit for the result queue */ public ResultQueueInternalState(int throttleLimit) { super(); - this.results = new ThrottleLimitResultQueue(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 { } - /** - *

- * 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. - *

- * - *

- * 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). - *

- * - *

- * 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. - *

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

- * Release the resources acquired in {@link #acquire()}. Call this - * method in a finally block after every repeat callback execution. - *

- * - * @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(); - } - } - - } - } diff --git a/spring-batch-infrastructure/src/test/java/org/springframework/batch/repeat/support/ResultHolderResultQueueTests.java b/spring-batch-infrastructure/src/test/java/org/springframework/batch/repeat/support/ResultHolderResultQueueTests.java new file mode 100644 index 000000000..bcca9d90a --- /dev/null +++ b/spring-batch-infrastructure/src/test/java/org/springframework/batch/repeat/support/ResultHolderResultQueueTests.java @@ -0,0 +1,66 @@ +package org.springframework.batch.repeat.support; + + +import static org.junit.Assert.assertFalse; +import static org.junit.Assert.assertTrue; + +import org.junit.Test; +import org.springframework.batch.repeat.RepeatContext; +import org.springframework.batch.repeat.RepeatStatus; + +public class ResultHolderResultQueueTests { + + private ResultHolderResultQueue queue = new ResultHolderResultQueue(10); + + @Test + public void testPutTake() throws Exception { + queue.expect(); + assertTrue(queue.isExpecting()); + assertTrue(queue.isEmpty()); + queue.put(new TestResultHolder(RepeatStatus.CONTINUABLE)); + assertFalse(queue.isEmpty()); + assertTrue(queue.take().getResult().isContinuable()); + assertFalse(queue.isExpecting()); + } + + @Test + public void testOrdering() throws Exception { + queue.expect(); + queue.expect(); + queue.put(new TestResultHolder(RepeatStatus.FINISHED)); + queue.put(new TestResultHolder(RepeatStatus.CONTINUABLE)); + assertFalse(queue.isEmpty()); + assertTrue(queue.take().getResult().isContinuable()); + assertFalse(queue.take().getResult().isContinuable()); + } + + private static class TestResultHolder implements ResultHolder { + + private RepeatStatus result; + + private Throwable error; + + public TestResultHolder(RepeatStatus result) { + super(); + this.result = result; + } + + public TestResultHolder(Throwable error) { + super(); + this.error = error; + } + + public RepeatContext getContext() { + return null; + } + + public Throwable getError() { + return error; + } + + public RepeatStatus getResult() { + return result; + } + } + +} diff --git a/spring-batch-infrastructure/src/test/java/org/springframework/batch/repeat/support/TaskExecutorRepeatTemplateBulkAsynchronousTests.java b/spring-batch-infrastructure/src/test/java/org/springframework/batch/repeat/support/TaskExecutorRepeatTemplateBulkAsynchronousTests.java index aa9f260dd..d3c93d000 100644 --- a/spring-batch-infrastructure/src/test/java/org/springframework/batch/repeat/support/TaskExecutorRepeatTemplateBulkAsynchronousTests.java +++ b/spring-batch-infrastructure/src/test/java/org/springframework/batch/repeat/support/TaskExecutorRepeatTemplateBulkAsynchronousTests.java @@ -47,7 +47,7 @@ public class TaskExecutorRepeatTemplateBulkAsynchronousTests { static Log logger = LogFactory.getLog(TaskExecutorRepeatTemplateBulkAsynchronousTests.class); - private static int TOTAL = 100; + private int total = 100; private int throttleLimit = 30; @@ -76,7 +76,7 @@ public class TaskExecutorRepeatTemplateBulkAsynchronousTests { public RepeatStatus doInIteration(RepeatContext context) throws Exception { int position = count.incrementAndGet(); - String item = position <= TOTAL ? "" + position : null; + String item = position <= total ? "" + position : null; items.add("" + item); if (item != null) { beBusy(); @@ -105,7 +105,7 @@ public class TaskExecutorRepeatTemplateBulkAsynchronousTests { int frequency = Collections.frequency(items, "null"); // System.err.println(items); // System.err.println("Frequency: " + frequency); - assertEquals(TOTAL, items.size() - frequency); + assertEquals(total, items.size() - frequency); assertTrue(frequency > 1); assertTrue(frequency <= throttleLimit + 1); @@ -120,7 +120,7 @@ public class TaskExecutorRepeatTemplateBulkAsynchronousTests { int frequency = Collections.frequency(items, "null"); // System.err.println("Frequency: " + frequency); // System.err.println("Items: " + items); - assertEquals(TOTAL, items.size() - frequency); + assertEquals(total, items.size() - frequency); assertTrue(frequency > 1); assertTrue(frequency <= throttleLimit + 1); @@ -131,7 +131,8 @@ public class TaskExecutorRepeatTemplateBulkAsynchronousTests { early = 2; SimpleAsyncTaskExecutor taskExecutor = new SimpleAsyncTaskExecutor(); - // Set the concurrency limit below the throttle limit for possible starvation condition + // Set the concurrency limit below the throttle limit for possible + // starvation condition taskExecutor.setConcurrencyLimit(20); template.setTaskExecutor(taskExecutor); @@ -140,7 +141,7 @@ public class TaskExecutorRepeatTemplateBulkAsynchronousTests { // System.err.println("Frequency: " + frequency); // System.err.println("Items: " + items); // Extra tasks will be submitted before the termination is detected - assertEquals(TOTAL, items.size() - frequency); + assertEquals(total, items.size() - frequency); assertTrue(frequency <= throttleLimit + 1); } @@ -148,18 +149,23 @@ public class TaskExecutorRepeatTemplateBulkAsynchronousTests { @Test public void testThrottleLimitEarlyFinishOneThread() throws Exception { - early = 2; + early = 4; SimpleAsyncTaskExecutor taskExecutor = new SimpleAsyncTaskExecutor(); taskExecutor.setConcurrencyLimit(1); + + // This is kind of slow with only one thread, so reduce size: + throttleLimit = 10; + total = 20; + + template.setThrottleLimit(throttleLimit); template.setTaskExecutor(taskExecutor); template.iterate(callback); int frequency = Collections.frequency(items, "null"); // System.err.println("Frequency: " + frequency); // System.err.println("Items: " + items); - // One extra task will be submitted before the termination is detected - assertEquals(early + 1, items.size() - frequency); - assertEquals(0, frequency); + assertEquals(total, items.size() - frequency); + assertTrue(frequency <= throttleLimit + 1); }