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

OPEN - issue BATCH-1362: Threads spinning doing nothing at end of multi-threaded Step

Browse Source 
Fixed, but can be tidied up.
This commit is contained in:
dsyer
2009-08-08 12:10:36 +00:00
parent d2f4849598
commit 410029ebbe
6 changed files with 296 additions and 72 deletions
Download Patch File Download Diff File Expand all files Collapse all files

11
spring-batch-core/src/main/java/org/springframework/batch/core/scope/context/StepContextRepeatCallback.java
View File

@@ -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<ChunkContext> attributeQueue = new LinkedBlockingQueue<ChunkContext>();
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()) {

41
spring-batch-core/src/test/java/org/springframework/batch/core/scope/context/StepContextRepeatCallbackTests.java
View File

@@ -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();
}
}

47
spring-batch-core/src/test/java/org/springframework/batch/core/step/tasklet/AsyncTaskletStepTests.java
View File

@@ -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<String> itemWriter = new ItemWriter<String>() {
public void write(List<? extends String> 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<String> 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<String> 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<String>(new ListItemReader<String>(items), itemWriter, chunkTemplate));
step.setTasklet(new TestingChunkOrientedTasklet<String>(
new ListItemReader<String>(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);
}
}

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

@@ -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<ResultHolder> 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<ResultHolder> 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);
}
}

191
spring-batch-infrastructure/src/test/java/org/springframework/batch/repeat/support/TaskExecutorRepeatTemplateSimpleAsynchronousTests.java Normal file
View File

@@ -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<String> items = Collections
.synchronizedList(new ArrayList<String>());
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<String> items = Collections
.synchronizedList(new ArrayList<String>());
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<String> items = Collections
.synchronizedList(new ArrayList<String>());
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();
}
}
}

2
spring-batch-infrastructure/src/test/resources/log4j.properties
View File

@@ -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