From c97b50d64be4119ce3118c14f7afec44fb91d583 Mon Sep 17 00:00:00 2001 From: Dave Syer Date: Fri, 11 Mar 2011 13:04:57 +0000 Subject: [PATCH] BATCH-1708: optimize TaskletStep --- .../batch/core/step/tasklet/TaskletStep.java | 23 ++++- .../StepExecutorInterruptionTests.java | 91 +++++++++++-------- 2 files changed, 70 insertions(+), 44 deletions(-) diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/step/tasklet/TaskletStep.java b/spring-batch-core/src/main/java/org/springframework/batch/core/step/tasklet/TaskletStep.java index b8de6175b..7e7677fec 100644 --- a/spring-batch-core/src/main/java/org/springframework/batch/core/step/tasklet/TaskletStep.java +++ b/spring-batch-core/src/main/java/org/springframework/batch/core/step/tasklet/TaskletStep.java @@ -97,8 +97,6 @@ public class TaskletStep extends AbstractStep { private Tasklet tasklet; - private final Semaphore semaphore = new Semaphore(1); - /** * Default constructor. */ @@ -245,6 +243,10 @@ public class TaskletStep extends AbstractStep { stream.update(stepExecution.getExecutionContext()); getJobRepository().updateExecutionContext(stepExecution); + // Shared semaphore per step execution, so other step executions can run + // in parallel without needing the lock + final Semaphore semaphore = createSemaphore(); + stepOperations.iterate(new StepContextRepeatCallback(stepExecution) { @Override @@ -260,7 +262,7 @@ public class TaskletStep extends AbstractStep { RepeatStatus result; try { result = (RepeatStatus) new TransactionTemplate(transactionManager, transactionAttribute) - .execute(new ChunkTransactionCallback(chunkContext)); + .execute(new ChunkTransactionCallback(chunkContext, semaphore)); } catch (UncheckedTransactionException e) { // Allow checked exceptions to be thrown inside callback @@ -281,6 +283,16 @@ public class TaskletStep extends AbstractStep { } + /** + * Extension point mainly for test purposes so that the behaviour of the + * lock can be manipulated to simulate various pathologies. + * + * @return a semaphore for locking access to the JobRepository + */ + protected Semaphore createSemaphore() { + return new Semaphore(1); + } + protected void close(ExecutionContext ctx) throws Exception { stream.close(); } @@ -312,9 +324,12 @@ public class TaskletStep extends AbstractStep { private boolean locked = false; - public ChunkTransactionCallback(ChunkContext chunkContext) { + private final Semaphore semaphore; + + public ChunkTransactionCallback(ChunkContext chunkContext, Semaphore semaphore) { this.chunkContext = chunkContext; this.stepExecution = chunkContext.getStepContext().getStepExecution(); + this.semaphore = semaphore; } @Override diff --git a/spring-batch-core/src/test/java/org/springframework/batch/core/step/tasklet/StepExecutorInterruptionTests.java b/spring-batch-core/src/test/java/org/springframework/batch/core/step/tasklet/StepExecutorInterruptionTests.java index 08686c53b..a2b6c04b6 100644 --- a/spring-batch-core/src/test/java/org/springframework/batch/core/step/tasklet/StepExecutorInterruptionTests.java +++ b/spring-batch-core/src/test/java/org/springframework/batch/core/step/tasklet/StepExecutorInterruptionTests.java @@ -20,7 +20,6 @@ import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertFalse; import static org.junit.Assert.assertTrue; -import java.lang.reflect.Field; import java.util.List; import java.util.concurrent.Semaphore; @@ -32,7 +31,10 @@ import org.springframework.batch.core.JobInterruptedException; 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.JobExecutionAlreadyRunningException; +import org.springframework.batch.core.repository.JobInstanceAlreadyCompleteException; import org.springframework.batch.core.repository.JobRepository; +import org.springframework.batch.core.repository.JobRestartException; import org.springframework.batch.core.repository.dao.MapExecutionContextDao; import org.springframework.batch.core.repository.dao.MapJobExecutionDao; import org.springframework.batch.core.repository.dao.MapJobInstanceDao; @@ -43,7 +45,6 @@ import org.springframework.batch.item.ItemWriter; import org.springframework.batch.repeat.policy.SimpleCompletionPolicy; import org.springframework.batch.repeat.support.RepeatTemplate; import org.springframework.batch.support.transaction.ResourcelessTransactionManager; -import org.springframework.util.ReflectionUtils; public class StepExecutorInterruptionTests { @@ -56,17 +57,18 @@ public class StepExecutorInterruptionTests { private StepExecution stepExecution; private JobRepository jobRepository; - - private Field semaphore; @Before public void setUp() throws Exception { - jobRepository = new SimpleJobRepository(new MapJobInstanceDao(), new MapJobExecutionDao(), new MapStepExecutionDao(), new MapExecutionContextDao()); + } + private void configureStep(TaskletStep step) throws JobExecutionAlreadyRunningException, JobRestartException, + JobInstanceAlreadyCompleteException { + + this.step = step; JobSupport job = new JobSupport(); - step = new TaskletStep("interruptedStep"); job.addStep(step); job.setBeanName("testJob"); jobExecution = jobRepository.createJobExecution(job.getName(), new JobParameters()); @@ -77,12 +79,13 @@ public class StepExecutorInterruptionTests { } }; stepExecution = new StepExecution(step.getName(), jobExecution); - semaphore = ReflectionUtils.findField(step.getClass(), "semaphore"); } @Test public void testInterruptStep() throws Exception { + configureStep(new TaskletStep("step")); + Thread processingThread = createThread(stepExecution); RepeatTemplate template = new RepeatTemplate(); @@ -124,6 +127,28 @@ public class StepExecutorInterruptionTests { @Test public void testInterruptOnInterruptedException() throws Exception { + // This simulates the unlikely sounding, but in practice all too common + // in Bamboo situation where the thread is interrupted before the lock + // is taken. + + configureStep(new TaskletStep("step") { + @Override + protected Semaphore createSemaphore() { + return new Semaphore(1) { + + @Override + public void acquire() throws InterruptedException { + Thread.currentThread().interrupt(); + throw new InterruptedException(); + } + + @Override + public void release() { + } + }; + } + }); + Thread processingThread = createThread(stepExecution); step.setTasklet(new TestingChunkOrientedTasklet(new ItemReader() { @@ -132,24 +157,6 @@ public class StepExecutorInterruptionTests { } }, itemWriter)); - // This simulates the unlikely sounding, but in practice all too common - // in Bamboo situation where the thread is interrupted before the lock - // is taken. - - ReflectionUtils.makeAccessible(semaphore); - ReflectionUtils.setField(semaphore, step, new Semaphore(1) { - - @Override - public void acquire() throws InterruptedException { - Thread.currentThread().interrupt(); - throw new InterruptedException(); - } - - @Override - public void release() { - } - }); - processingThread.start(); Thread.sleep(100); @@ -168,27 +175,31 @@ public class StepExecutorInterruptionTests { @Test public void testLockNotReleasedIfChunkFails() throws Exception { + configureStep(new TaskletStep("step") { + @Override + protected Semaphore createSemaphore() { + return new Semaphore(1) { + private boolean locked = false; + + @Override + public void acquire() throws InterruptedException { + locked = true; + } + + @Override + public void release() { + assertTrue("Lock released before it is acquired", locked); + } + }; + } + }); + step.setTasklet(new TestingChunkOrientedTasklet(new ItemReader() { public Object read() throws Exception { throw new RuntimeException("Planned!"); } }, itemWriter)); - ReflectionUtils.makeAccessible(semaphore); - ReflectionUtils.setField(semaphore, step, new Semaphore(1) { - private boolean locked = false; - - @Override - public void acquire() throws InterruptedException { - locked = true; - } - - @Override - public void release() { - assertTrue("Lock released before it is acquired", locked); - } - }); - jobRepository.add(stepExecution); step.execute(stepExecution);