From 0fb758ee8d2c2cd4158a66597bf15877211a0d7b Mon Sep 17 00:00:00 2001 From: dsyer Date: Mon, 10 Aug 2009 10:46:42 +0000 Subject: [PATCH] RESOLVED - issue BATCH-1362: Threads spinning doing nothing at end of multi-threaded Step http://jira.springframework.org/browse/BATCH-1362 --- .../support/TaskExecutorRepeatTemplate.java | 236 ++++++++++++------ ...RepeatTemplateSimpleAsynchronousTests.java | 179 ++++++------- 2 files changed, 237 insertions(+), 178 deletions(-) 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 de2814a7c..2739dfa99 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,9 +16,8 @@ package org.springframework.batch.repeat.support; -import java.util.concurrent.atomic.AtomicBoolean; -import java.util.concurrent.atomic.AtomicInteger; - +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; @@ -59,15 +58,25 @@ public class TaskExecutorRepeatTemplate extends RepeatTemplate { */ public static final int DEFAULT_THROTTLE_LIMIT = 4; - private static final String ACTIVE_COUNT = TaskExecutorRepeatTemplate.class.getName() + ".ACTIVE_COUNT"; - - private static final String PAUSED = TaskExecutorRepeatTemplate.class.getName() + ".PAUSED"; - private int throttleLimit = DEFAULT_THROTTLE_LIMIT; private TaskExecutor taskExecutor = new SyncTaskExecutor(); - private static final Object lock = new Object(); + /** + * 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 {@link #DEFAULT_THROTTLE_LIMIT}. + * N.B. when used with a thread pooled {@link TaskExecutor} the thread pool + * might prevent the throttle limit actually being reached (so make the core + * pool size larger than the throttle limit if possible). + * + * @param throttleLimit the throttleLimit to set. + */ + public void setThrottleLimit(int throttleLimit) { + this.throttleLimit = throttleLimit; + } /** * Setter for task executor to be used to run the individual item callbacks. @@ -94,11 +103,7 @@ public class TaskExecutorRepeatTemplate extends RepeatTemplate { ExecutingRunnable runnable = null; ResultQueue queue = ((ResultQueueInternalState) state).getResultQueue(); - - if (!context.hasAttribute(ACTIVE_COUNT)) { - context.setAttribute(ACTIVE_COUNT, new AtomicInteger()); - context.setAttribute(PAUSED, new AtomicBoolean(false)); - } + ActivityBarrier lock = ((ResultQueueInternalState) state).getLock(); do { @@ -106,7 +111,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); + runnable = new ExecutingRunnable(callback, context, queue, lock); /** * Tell the runnable that it can expect a result. This could have @@ -154,15 +159,13 @@ 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()) { - synchronized (lock) { - logger.debug("Notifying other waiting callbacks while waiting for results."); - lock.notifyAll(); - } + lock.release(); /* * Careful that no runnables that are not going to finish ever get @@ -216,19 +219,17 @@ public class TaskExecutorRepeatTemplate extends RepeatTemplate { private volatile Throwable error; - private final AtomicInteger active; + private final ActivityBarrier lock; - private final AtomicBoolean paused; - - public ExecutingRunnable(RepeatCallback callback, RepeatContext context, ResultQueue queue) { + public ExecutingRunnable(RepeatCallback callback, RepeatContext context, ResultQueue queue, + ActivityBarrier lock) { super(); this.callback = callback; this.context = context; this.queue = queue; - this.active = (AtomicInteger) context.getAttribute(ACTIVE_COUNT); - this.paused = (AtomicBoolean) context.getAttribute(PAUSED); + this.lock = lock; } @@ -259,9 +260,7 @@ public class TaskExecutorRepeatTemplate extends RepeatTemplate { RepeatSynchronizationManager.register(context); } - active.incrementAndGet(); - paused.set(false); - + lock.acquire(); result = callback.doInIteration(context); } @@ -270,49 +269,12 @@ public class TaskExecutorRepeatTemplate extends RepeatTemplate { } finally { - boolean stillActive = active.decrementAndGet()>0 || paused.get(); + lock.release(isComplete(context), result); if (clearContext) { RepeatSynchronizationManager.clear(); } - - if (logger.isDebugEnabled()) { - logger.debug("Completed callback with result = " + result + ", and " + stillActive - + " active callbacks."); - } - - if (result == RepeatStatus.FINISHED) { - if (stillActive) { - synchronized (lock) { - logger.debug("Waiting for other active callbacks to finish."); - try { - lock.wait(); - } - catch (InterruptedException e) { - logger.info("Interrupted waiting for active callbacks"); - Thread.currentThread().interrupt(); - } - } - } - else { - synchronized (lock) { - logger.debug("Notifying other waiting callbacks on finish."); - lock.notifyAll(); - } - } - } - else { - if (isComplete(context)) { - synchronized (lock) { - logger.debug("Notifying other waiting callbacks on policy based completion."); - lock.notifyAll(); - } - } else { - paused.set(true); - } - } - queue.put(this); } @@ -351,19 +313,26 @@ 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(); } - /* - * (non-Javadoc) - * - * @seeorg.springframework.batch.repeat.support.RepeatInternalState# - * getResultQueue() + /** + * @return the lock instance + */ + public ActivityBarrier getLock() { + return lock; + } + + /** + * @return the result queue */ public ResultQueue getResultQueue() { return results; @@ -372,18 +341,125 @@ 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 {@link #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. + *

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

* - * @param throttleLimit the throttleLimit to set. + *

+ * 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 */ - public void setThrottleLimit(int throttleLimit) { - this.throttleLimit = throttleLimit; + 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) { + try { + lock.wait(); + } + 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/TaskExecutorRepeatTemplateSimpleAsynchronousTests.java b/spring-batch-infrastructure/src/test/java/org/springframework/batch/repeat/support/TaskExecutorRepeatTemplateSimpleAsynchronousTests.java index b581a56e8..dadedf7b9 100644 --- a/spring-batch-infrastructure/src/test/java/org/springframework/batch/repeat/support/TaskExecutorRepeatTemplateSimpleAsynchronousTests.java +++ b/spring-batch-infrastructure/src/test/java/org/springframework/batch/repeat/support/TaskExecutorRepeatTemplateSimpleAsynchronousTests.java @@ -26,84 +26,55 @@ import java.util.concurrent.atomic.AtomicInteger; import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; +import org.junit.Before; import org.junit.Test; -import org.springframework.batch.item.ExecutionContext; import org.springframework.batch.repeat.RepeatCallback; import org.springframework.batch.repeat.RepeatContext; import org.springframework.batch.repeat.RepeatStatus; import org.springframework.batch.repeat.policy.SimpleCompletionPolicy; -import org.springframework.batch.repeat.support.AbstractTradeBatchTests.TradeItemReader; -import org.springframework.core.io.ClassPathResource; import org.springframework.core.task.SimpleAsyncTaskExecutor; +/** + * Simple tests for concurrent behaviour in repeat template, in particular the + * barrier at the end of the iteration. N.B. these tests may fail if + * insufficient threads are available (e.g. on a single-core machine, or under + * load). They shouldn't deadlock though. + * + * @author Dave Syer + * + */ public class TaskExecutorRepeatTemplateSimpleAsynchronousTests { - static Log logger = LogFactory - .getLog(TaskExecutorRepeatTemplateSimpleAsynchronousTests.class); + static Log logger = LogFactory.getLog(TaskExecutorRepeatTemplateSimpleAsynchronousTests.class); private static int TOTAL = 100; - @Test - public void testThrottleLimit() throws Exception { + private int throttleLimit = 30; - int throttleLimit = 20; + private volatile int early = Integer.MAX_VALUE; - TaskExecutorRepeatTemplate template = new TaskExecutorRepeatTemplate(); + private TaskExecutorRepeatTemplate template; + + private RepeatCallback callback; + + private List items; + + @Before + public void setUp() { + + template = new TaskExecutorRepeatTemplate(); SimpleAsyncTaskExecutor taskExecutor = new SimpleAsyncTaskExecutor(); taskExecutor.setConcurrencyLimit(300); template.setTaskExecutor(taskExecutor); template.setThrottleLimit(throttleLimit); - final List items = Collections - .synchronizedList(new ArrayList()); + items = Collections.synchronizedList(new ArrayList()); - final RepeatCallback callback = new RepeatCallback() { - - private volatile int count = 0; - - public RepeatStatus doInIteration(RepeatContext context) - throws Exception { - String item = count < TOTAL ? "" + count : null; - count++; - items.add("" + item); - if (item != null) { - beBusy(); - } - return RepeatStatus.continueIf(item != null); - } - }; - - template.iterate(callback); - int frequency = Collections.frequency(items, "null"); -// System.err.println(items); -// System.err.println("Frequency: " + frequency); - assertEquals(TOTAL, items.size() - frequency); - assertTrue(frequency > 1); - assertTrue(frequency <= throttleLimit + 1); - } - - @Test - public void testThrottleLimitWithRetry() throws Exception { - - int throttleLimit = 30; - - TaskExecutorRepeatTemplate template = new TaskExecutorRepeatTemplate(); - SimpleAsyncTaskExecutor taskExecutor = new SimpleAsyncTaskExecutor(); - taskExecutor.setConcurrencyLimit(300); - template.setTaskExecutor(taskExecutor); - template.setThrottleLimit(throttleLimit); - - final List items = Collections - .synchronizedList(new ArrayList()); - - final RepeatCallback callback = new RepeatCallback() { + callback = new RepeatCallback() { private volatile AtomicInteger count = new AtomicInteger(0); - private volatile int early = 2; - - public RepeatStatus doInIteration(RepeatContext context) - throws Exception { + public RepeatStatus doInIteration(RepeatContext context) throws Exception { int position = count.incrementAndGet(); String item = position <= TOTAL ? "" + count : null; items.add("" + item); @@ -117,75 +88,87 @@ public class TaskExecutorRepeatTemplateSimpleAsynchronousTests { * happens for instance if there is a failure and you want to * retry the work.) */ - RepeatStatus result = RepeatStatus.continueIf(position != early - && item != null); + RepeatStatus result = RepeatStatus.continueIf(position != early && item != null); logger.debug("Returning " + result + " for count=" + position); return result; } }; - template.iterate(callback); - int frequency = Collections.frequency(items, "null"); - assertEquals(TOTAL, items.size() - frequency); - // System.err.println("Frequency: " + frequency); - assertTrue(frequency > 1); - assertTrue(frequency <= throttleLimit + 1); } @Test - public void testThrottleLimitWithRetryAndEarlyCompletion() throws Exception { + public void testThrottleLimit() throws Exception { - int throttleLimit = 30; + template.iterate(callback); + int frequency = Collections.frequency(items, "null"); + // System.err.println(items); + // System.err.println("Frequency: " + frequency); + assertEquals(TOTAL, items.size() - frequency); + assertTrue(frequency > 1); + assertTrue(frequency <= throttleLimit + 1); - TaskExecutorRepeatTemplate template = new TaskExecutorRepeatTemplate(); + } + + @Test + public void testThrottleLimitEarlyFinish() throws Exception { + + early = 2; + + template.iterate(callback); + int frequency = Collections.frequency(items, "null"); + // System.err.println("Frequency: " + frequency); + // System.err.println("Items: " + items); + assertEquals(TOTAL, items.size() - frequency); + assertTrue(frequency > 1); + assertTrue(frequency <= throttleLimit + 1); + + } + + @Test + public void testThrottleLimitEarlyFinishOneThread() throws Exception { + + early = 2; SimpleAsyncTaskExecutor taskExecutor = new SimpleAsyncTaskExecutor(); - taskExecutor.setConcurrencyLimit(300); - template.setCompletionPolicy(new SimpleCompletionPolicy(10)); + taskExecutor.setConcurrencyLimit(1); template.setTaskExecutor(taskExecutor); - template.setThrottleLimit(throttleLimit); - final List items = Collections - .synchronizedList(new ArrayList()); + 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); - final RepeatCallback callback = new RepeatCallback() { + } - private volatile AtomicInteger count = new AtomicInteger(0); - private volatile int early = 2; + @Test + public void testThrottleLimitWithEarlyCompletion() throws Exception { - public RepeatStatus doInIteration(RepeatContext context) - throws Exception { - - int position = count.incrementAndGet(); - String item = position <= TOTAL ? "" + count : null; - items.add("" + item); - if (item != null) { - beBusy(); - } - RepeatStatus result = RepeatStatus.continueIf(position != early - && item != null); - logger.debug("Returning " + result + " for count=" + position); - return result; - } - }; + early = 2; + template.setCompletionPolicy(new SimpleCompletionPolicy(10)); template.iterate(callback); int frequency = Collections.frequency(items, "null"); assertEquals(10, items.size() - frequency); // System.err.println("Frequency: " + frequency); assertEquals(0, frequency); + } + /** + * Slightly flakey convenience method. If this doesn't do something that + * lasts sufficiently long for another worker to be launched while it is + * busy, the early completion tests will fail. "Sufficiently long" is the + * problem so we try and block until we know someone else is busy? + * + * @throws Exception + */ private void beBusy() throws Exception { - // Do some more I/O - for (int i = 0; i < 10; i++) { - TradeItemReader provider = new TradeItemReader( - new ClassPathResource("trades.csv", getClass())); - provider.open(new ExecutionContext()); - while (provider.read() != null) - continue; - provider.close(); + synchronized (this) { + wait(100L); + notifyAll(); } - } }