diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/scope/context/StepContextRepeatCallback.java b/spring-batch-core/src/main/java/org/springframework/batch/core/scope/context/StepContextRepeatCallback.java index a8d2adbbe..558cc945e 100644 --- a/spring-batch-core/src/main/java/org/springframework/batch/core/scope/context/StepContextRepeatCallback.java +++ b/spring-batch-core/src/main/java/org/springframework/batch/core/scope/context/StepContextRepeatCallback.java @@ -17,7 +17,6 @@ package org.springframework.batch.core.scope.context; import java.util.Queue; import java.util.concurrent.LinkedBlockingQueue; -import java.util.concurrent.atomic.AtomicInteger; import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; @@ -39,8 +38,6 @@ public abstract class StepContextRepeatCallback implements RepeatCallback { private final Queue attributeQueue = new LinkedBlockingQueue(); - private final AtomicInteger workerCount = new AtomicInteger(0); - private final StepExecution stepExecution; private final Log logger = LogFactory.getLog(StepContextRepeatCallback.class); @@ -64,8 +61,6 @@ public abstract class StepContextRepeatCallback implements RepeatCallback { */ public RepeatStatus doInIteration(RepeatContext context) throws Exception { - workerCount.incrementAndGet(); - // The StepContext has to be the same for all chunks, // otherwise step-scoped beans will be re-initialised for each chunk. StepContext stepContext = StepSynchronizationManager.register(stepExecution); @@ -77,12 +72,10 @@ public abstract class StepContextRepeatCallback implements RepeatCallback { } try { - logger.debug("Chunk execution starting: worker count="+workerCount.get()+", queue size="+attributeQueue.size()); - return RepeatStatus.continueIf(doInChunkContext(context, chunkContext).isContinuable() - || !attributeQueue.isEmpty() || workerCount.get()>1); + logger.debug("Chunk execution starting: queue size="+attributeQueue.size()); + return doInChunkContext(context, chunkContext); } finally { - workerCount.decrementAndGet(); // Still some stuff to do with the data in this chunk, // pass it back. if (!chunkContext.isComplete()) { diff --git a/spring-batch-core/src/test/java/org/springframework/batch/core/scope/context/StepContextRepeatCallbackTests.java b/spring-batch-core/src/test/java/org/springframework/batch/core/scope/context/StepContextRepeatCallbackTests.java index 48cec9dfc..fb30c2681 100644 --- a/spring-batch-core/src/test/java/org/springframework/batch/core/scope/context/StepContextRepeatCallbackTests.java +++ b/spring-batch-core/src/test/java/org/springframework/batch/core/scope/context/StepContextRepeatCallbackTests.java @@ -15,25 +15,17 @@ */ package org.springframework.batch.core.scope.context; -import static org.junit.Assert.fail; import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertFalse; import static org.junit.Assert.assertTrue; -import java.util.concurrent.CountDownLatch; - -import org.apache.commons.logging.Log; -import org.apache.commons.logging.LogFactory; import org.junit.After; import org.junit.Test; import org.springframework.batch.core.ExitStatus; import org.springframework.batch.core.JobExecution; import org.springframework.batch.core.StepExecution; -import org.springframework.batch.core.scope.context.StepContextRepeatCallback; -import org.springframework.batch.core.scope.context.StepSynchronizationManager; import org.springframework.batch.repeat.RepeatContext; import org.springframework.batch.repeat.RepeatStatus; -import org.springframework.batch.repeat.context.RepeatContextSupport; /** * @author Dave Syer @@ -87,37 +79,4 @@ public class StepContextRepeatCallbackTests { assertFalse(removedAttribute); } - @Test - public void testUnfinishedWork() throws Exception { - StepSynchronizationManager.register(stepExecution); - final CountDownLatch background = new CountDownLatch(1); - final CountDownLatch foreground = new CountDownLatch(1); - final StepContextRepeatCallback callback = new StepContextRepeatCallback(stepExecution) { - private Log logger = LogFactory.getLog(getClass()); - @Override - public RepeatStatus doInChunkContext(RepeatContext context, ChunkContext chunkContext) throws Exception { - foreground.countDown(); - if (context==null) { - logger.debug("Waiting for latch"); - background.await(); - logger.debug("Released"); - } - return RepeatStatus.FINISHED; - } - }; - new Thread(new Runnable() { - public void run() { - try { - callback.doInIteration(null); - } - catch (Exception e) { - fail(e.getMessage()); - } - } - }).start(); - foreground.await(); - assertEquals(RepeatStatus.CONTINUABLE, callback.doInIteration(new RepeatContextSupport(null))); - background.countDown(); - } - } diff --git a/spring-batch-core/src/test/java/org/springframework/batch/core/step/tasklet/AsyncTaskletStepTests.java b/spring-batch-core/src/test/java/org/springframework/batch/core/step/tasklet/AsyncTaskletStepTests.java index ddee07080..f21985dac 100644 --- a/spring-batch-core/src/test/java/org/springframework/batch/core/step/tasklet/AsyncTaskletStepTests.java +++ b/spring-batch-core/src/test/java/org/springframework/batch/core/step/tasklet/AsyncTaskletStepTests.java @@ -17,6 +17,7 @@ package org.springframework.batch.core.step.tasklet; import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertTrue; import java.util.Arrays; import java.util.List; @@ -26,13 +27,9 @@ import org.junit.Before; import org.junit.Test; import org.springframework.batch.core.BatchStatus; import org.springframework.batch.core.JobExecution; -import org.springframework.batch.core.JobInstance; import org.springframework.batch.core.JobParameters; import org.springframework.batch.core.StepExecution; -import org.springframework.batch.core.job.JobSupport; -import org.springframework.batch.core.repository.dao.MapJobExecutionDao; -import org.springframework.batch.core.repository.dao.MapJobInstanceDao; -import org.springframework.batch.core.repository.dao.MapStepExecutionDao; +import org.springframework.batch.core.repository.JobRepository; import org.springframework.batch.core.step.JobRepositorySupport; import org.springframework.batch.item.ExecutionContext; import org.springframework.batch.item.ItemStreamSupport; @@ -51,35 +48,41 @@ public class AsyncTaskletStepTests { private TaskletStep step; - private JobInstance jobInstance; + private int throttleLimit = 20; ItemWriter itemWriter = new ItemWriter() { public void write(List data) throws Exception { + // Thread.sleep(100L); processed.addAll(data); } }; + private JobRepository jobRepository; + @Before public void setUp() throws Exception { - MapJobInstanceDao.clear(); - MapStepExecutionDao.clear(); - MapJobExecutionDao.clear(); step = new TaskletStep("stepName"); ResourcelessTransactionManager transactionManager = new ResourcelessTransactionManager(); step.setTransactionManager(transactionManager); - List items = Arrays.asList(StringUtils - .commaDelimitedListToStringArray("1,2,3,4,5,6,7,8,9,10,11,12,13,14,15,16,17,18,19,20,21,22,23,24,25")); + List items = Arrays + .asList(StringUtils + .commaDelimitedListToStringArray("1,2,3,4,5,6,7,8,9,10,11,12,13,14,15,16,17,18,19,20,21,22,23,24,25")); RepeatTemplate chunkTemplate = new RepeatTemplate(); chunkTemplate.setCompletionPolicy(new SimpleCompletionPolicy(2)); - step.setTasklet(new TestingChunkOrientedTasklet(new ListItemReader(items), itemWriter, chunkTemplate)); + step.setTasklet(new TestingChunkOrientedTasklet( + new ListItemReader(items), itemWriter, chunkTemplate)); - step.setJobRepository(new JobRepositorySupport()); + jobRepository = new JobRepositorySupport(); + step.setJobRepository(jobRepository); TaskExecutorRepeatTemplate template = new TaskExecutorRepeatTemplate(); - template.setTaskExecutor(new SimpleAsyncTaskExecutor()); + template.setThrottleLimit(throttleLimit); + SimpleAsyncTaskExecutor taskExecutor = new SimpleAsyncTaskExecutor(); + taskExecutor.setConcurrencyLimit(300); + template.setTaskExecutor(taskExecutor); step.setStepOperations(template); step.registerStream(new ItemStreamSupport() { @@ -91,9 +94,6 @@ public class AsyncTaskletStepTests { } }); - JobSupport job = new JobSupport("FOO"); - jobInstance = new JobInstance(0L, new JobParameters(), job.getName()); - } /** @@ -102,8 +102,10 @@ public class AsyncTaskletStepTests { @Test public void testStepExecutionUpdates() throws Exception { - JobExecution jobExecution = new JobExecution(jobInstance); - StepExecution stepExecution = jobExecution.createStepExecution(step.getName()); + JobExecution jobExecution = jobRepository.createJobExecution("JOB", + new JobParameters()); + StepExecution stepExecution = jobExecution.createStepExecution(step + .getName()); step.execute(stepExecution); @@ -111,6 +113,13 @@ public class AsyncTaskletStepTests { assertEquals(25, stepExecution.getReadCount()); assertEquals(25, processed.size()); + // System.err.println(stepExecution.getCommitCount()); + // Check commit count didn't spin out of control waiting for other + // threads to finish... + assertTrue(stepExecution.getCommitCount() > processed.size()); + assertTrue(stepExecution.getCommitCount() <= processed.size() + + throttleLimit + 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 abb5a65d4..de2814a7c 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,11 +16,14 @@ package org.springframework.batch.repeat.support; -import org.springframework.batch.repeat.RepeatStatus; +import java.util.concurrent.atomic.AtomicBoolean; +import java.util.concurrent.atomic.AtomicInteger; + import org.springframework.batch.repeat.RepeatCallback; import org.springframework.batch.repeat.RepeatContext; import org.springframework.batch.repeat.RepeatException; import org.springframework.batch.repeat.RepeatOperations; +import org.springframework.batch.repeat.RepeatStatus; import org.springframework.core.task.SyncTaskExecutor; import org.springframework.core.task.TaskExecutor; import org.springframework.util.Assert; @@ -56,10 +59,16 @@ 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(); + /** * Setter for task executor to be used to run the individual item callbacks. * @@ -86,6 +95,11 @@ public class TaskExecutorRepeatTemplate extends RepeatTemplate { ResultQueue queue = ((ResultQueueInternalState) state).getResultQueue(); + if (!context.hasAttribute(ACTIVE_COUNT)) { + context.setAttribute(ACTIVE_COUNT, new AtomicInteger()); + context.setAttribute(PAUSED, new AtomicBoolean(false)); + } + do { /* @@ -145,6 +159,11 @@ public class TaskExecutorRepeatTemplate extends RepeatTemplate { while (queue.isExpecting()) { + synchronized (lock) { + logger.debug("Notifying other waiting callbacks while waiting for results."); + lock.notifyAll(); + } + /* * Careful that no runnables that are not going to finish ever get * onto the queue, else this may block forever. @@ -185,7 +204,7 @@ public class TaskExecutorRepeatTemplate extends RepeatTemplate { * @author Dave Syer * */ - private static class ExecutingRunnable implements Runnable, ResultHolder { + private class ExecutingRunnable implements Runnable, ResultHolder { private final RepeatCallback callback; @@ -197,6 +216,10 @@ public class TaskExecutorRepeatTemplate extends RepeatTemplate { private volatile Throwable error; + private final AtomicInteger active; + + private final AtomicBoolean paused; + public ExecutingRunnable(RepeatCallback callback, RepeatContext context, ResultQueue queue) { super(); @@ -204,6 +227,8 @@ public class TaskExecutorRepeatTemplate extends RepeatTemplate { this.callback = callback; this.context = context; this.queue = queue; + this.active = (AtomicInteger) context.getAttribute(ACTIVE_COUNT); + this.paused = (AtomicBoolean) context.getAttribute(PAUSED); } @@ -233,16 +258,63 @@ public class TaskExecutorRepeatTemplate extends RepeatTemplate { clearContext = true; RepeatSynchronizationManager.register(context); } + + active.incrementAndGet(); + paused.set(false); + result = callback.doInIteration(context); + } catch (Exception e) { error = e; } finally { + + boolean stillActive = active.decrementAndGet()>0 || paused.get(); + 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); + } } 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 new file mode 100644 index 000000000..b581a56e8 --- /dev/null +++ b/spring-batch-infrastructure/src/test/java/org/springframework/batch/repeat/support/TaskExecutorRepeatTemplateSimpleAsynchronousTests.java @@ -0,0 +1,191 @@ +/* + * Copyright 2006-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 static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertTrue; + +import java.util.ArrayList; +import java.util.Collections; +import java.util.List; +import java.util.concurrent.atomic.AtomicInteger; + +import org.apache.commons.logging.Log; +import org.apache.commons.logging.LogFactory; +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; + +public class TaskExecutorRepeatTemplateSimpleAsynchronousTests { + + static Log logger = LogFactory + .getLog(TaskExecutorRepeatTemplateSimpleAsynchronousTests.class); + + private static int TOTAL = 100; + + @Test + public void testThrottleLimit() throws Exception { + + int throttleLimit = 20; + + 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() { + + 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() { + + private volatile AtomicInteger count = new AtomicInteger(0); + private volatile int early = 2; + + public RepeatStatus doInIteration(RepeatContext context) + throws Exception { + + int position = count.incrementAndGet(); + String item = position <= TOTAL ? "" + count : null; + items.add("" + item); + if (item != null) { + beBusy(); + } + /* + * In a multi-threaded task, one of the callbacks can call + * FINISHED early, while other threads are still working, and + * would do more work if the callback was called again. (This + * happens for instance if there is a failure and you want to + * retry the work.) + */ + 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 { + + int throttleLimit = 30; + + TaskExecutorRepeatTemplate template = new TaskExecutorRepeatTemplate(); + SimpleAsyncTaskExecutor taskExecutor = new SimpleAsyncTaskExecutor(); + taskExecutor.setConcurrencyLimit(300); + template.setCompletionPolicy(new SimpleCompletionPolicy(10)); + template.setTaskExecutor(taskExecutor); + template.setThrottleLimit(throttleLimit); + + final List items = Collections + .synchronizedList(new ArrayList()); + + final RepeatCallback callback = new RepeatCallback() { + + private volatile AtomicInteger count = new AtomicInteger(0); + private volatile int early = 2; + + 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; + } + }; + + template.iterate(callback); + int frequency = Collections.frequency(items, "null"); + assertEquals(10, items.size() - frequency); + // System.err.println("Frequency: " + frequency); + assertEquals(0, frequency); + } + + 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(); + } + + } + +} diff --git a/spring-batch-infrastructure/src/test/resources/log4j.properties b/spring-batch-infrastructure/src/test/resources/log4j.properties index b34176f9a..3f796307f 100644 --- a/spring-batch-infrastructure/src/test/resources/log4j.properties +++ b/spring-batch-infrastructure/src/test/resources/log4j.properties @@ -4,5 +4,5 @@ log4j.appender.stdout=org.apache.log4j.ConsoleAppender log4j.appender.stdout.layout=org.apache.log4j.PatternLayout log4j.appender.stdout.layout.ConversionPattern=%d %p %t [%c] - <%m>%n -#log4j.category.org.springframework.batch=DEBUG +log4j.category.org.springframework.batch=DEBUG #log4j.category.org.springframework.transaction=DEBUG