diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/configuration/support/MapJobRegistry.java b/spring-batch-core/src/main/java/org/springframework/batch/core/configuration/support/MapJobRegistry.java index daf442000..3663e6f4d 100644 --- a/spring-batch-core/src/main/java/org/springframework/batch/core/configuration/support/MapJobRegistry.java +++ b/spring-batch-core/src/main/java/org/springframework/batch/core/configuration/support/MapJobRegistry.java @@ -20,6 +20,10 @@ import java.util.Collections; import java.util.HashMap; import java.util.HashSet; import java.util.Map; +import java.util.Set; + +import java.util.concurrent.ConcurrentMap; +import java.util.concurrent.ConcurrentHashMap; import org.springframework.batch.core.Job; import org.springframework.batch.core.configuration.DuplicateJobException; @@ -29,50 +33,50 @@ import org.springframework.batch.core.launch.NoSuchJobException; import org.springframework.util.Assert; /** - * Simple map-based implementation of {@link JobRegistry}. Access to the map is - * synchronized, guarded by an internal lock. + * Simple, thread-safe, map-based implementation of {@link JobRegistry}. * * @author Dave Syer + * @author Robert Fischer * */ public class MapJobRegistry implements JobRegistry { - private Map map = new HashMap(); + /** + * The map holding the registered job factories. + */ + // The "final" ensures that it is visible and initialized when the constructor resolves. + private final ConcurrentMap map = new ConcurrentHashMap(); public void register(JobFactory jobFactory) throws DuplicateJobException { Assert.notNull(jobFactory); String name = jobFactory.getJobName(); Assert.notNull(name, "Job configuration must have a name."); - synchronized (map) { - if (map.containsKey(name)) { - throw new DuplicateJobException("A job configuration with this name [" + name - + "] was already registered"); - } - map.put(name, jobFactory); + JobFactory previousValue = map.putIfAbsent(name, jobFactory); + if(previousValue != null) { + throw new DuplicateJobException("A job configuration with this name [" + name + + "] was already registered"); } } public void unregister(String name) { Assert.notNull(name, "Job configuration must have a name."); - synchronized (map) { - map.remove(name); - } - + map.remove(name); } public Job getJob(String name) throws NoSuchJobException { - synchronized (map) { - if (!map.containsKey(name)) { - throw new NoSuchJobException("No job configuration with the name [" + name + "] was registered"); - } - return map.get(name).createJob(); + JobFactory factory = map.get(name); + if(factory == null) { + throw new NoSuchJobException("No job configuration with the name [" + name + "] was registered"); + } else { + return factory.createJob(); } } - public Collection getJobNames() { - synchronized (map) { - return Collections.unmodifiableCollection(new HashSet(map.keySet())); - } + /** + * Provides an unmodifiable view of the job names. + */ + public Set getJobNames() { + return Collections.unmodifiableSet(map.keySet()); } } diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/repository/dao/MapExecutionContextDao.java b/spring-batch-core/src/main/java/org/springframework/batch/core/repository/dao/MapExecutionContextDao.java index 7904a1958..8713fd944 100644 --- a/spring-batch-core/src/main/java/org/springframework/batch/core/repository/dao/MapExecutionContextDao.java +++ b/spring-batch-core/src/main/java/org/springframework/batch/core/repository/dao/MapExecutionContextDao.java @@ -16,8 +16,11 @@ package org.springframework.batch.core.repository.dao; +import java.util.concurrent.ConcurrentMap; import java.util.Map; +import java.io.Serializable; + import org.springframework.batch.core.JobExecution; import org.springframework.batch.core.StepExecution; import org.springframework.batch.item.ExecutionContext; @@ -32,15 +35,61 @@ import org.springframework.batch.support.transaction.TransactionAwareProxyFactor */ public class MapExecutionContextDao implements ExecutionContextDao { - private Map contextsByStepExecutionId = TransactionAwareProxyFactory + private final ConcurrentMap contexts = TransactionAwareProxyFactory .createAppendOnlyTransactionalMap(); - private Map contextsByJobExecutionId = TransactionAwareProxyFactory - .createAppendOnlyTransactionalMap(); + private static final class ContextKey implements Comparable, Serializable { - public void clear() { - contextsByJobExecutionId.clear(); - contextsByStepExecutionId.clear(); + private static enum Type { STEP, JOB; } + + private final Type type; + private final long id; + + private ContextKey(Type type, long id) { + if(type == null) throw new IllegalStateException("Need a non-null type for a context"); + this.type = type; + this.id = id; + } + + @Override + public int compareTo(ContextKey them) { + if(them == null) return 1; + final int idCompare = new Long(this.id).compareTo(new Long(them.id)); // JDK6 Make this Long.compare(x,y) + if(idCompare != 0) return idCompare; + final int typeCompare = this.type.compareTo(them.type); + if(typeCompare != 0) return typeCompare; + return 0; + } + + @Override + public boolean equals(Object them) { + if(them == null) return false; + if(them instanceof ContextKey) return this.equals((ContextKey)them); + return false; + } + + public boolean equals(ContextKey them) { + if(them == null) return false; + return this.id == them.id && this.type.equals(them.type); + } + + @Override + public int hashCode() { + int value = (int)(id^(id>>>32)); + switch(type) { + case STEP: return value; + case JOB: return ~value; + default: throw new IllegalStateException("Unknown type encountered in switch: " + type); + } + } + + public static ContextKey step(long id) { return new ContextKey(Type.STEP, id); } + + public static ContextKey job(long id) { return new ContextKey(Type.JOB, id); } + } + + public void clear() { + contexts.clear(); } private static ExecutionContext copy(ExecutionContext original) { @@ -48,24 +97,24 @@ public class MapExecutionContextDao implements ExecutionContextDao { } public ExecutionContext getExecutionContext(StepExecution stepExecution) { - return copy(contextsByStepExecutionId.get(stepExecution.getId())); + return copy(contexts.get(ContextKey.step(stepExecution.getId()))); } public void updateExecutionContext(StepExecution stepExecution) { ExecutionContext executionContext = stepExecution.getExecutionContext(); if (executionContext != null) { - contextsByStepExecutionId.put(stepExecution.getId(), copy(executionContext)); + contexts.put(ContextKey.step(stepExecution.getId()), copy(executionContext)); } } public ExecutionContext getExecutionContext(JobExecution jobExecution) { - return copy(contextsByJobExecutionId.get(jobExecution.getId())); + return copy(contexts.get(ContextKey.job(jobExecution.getId()))); } public void updateExecutionContext(JobExecution jobExecution) { ExecutionContext executionContext = jobExecution.getExecutionContext(); if (executionContext != null) { - contextsByJobExecutionId.put(jobExecution.getId(), copy(executionContext)); + contexts.put(ContextKey.job(jobExecution.getId()), copy(executionContext)); } } diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/repository/dao/MapJobExecutionDao.java b/spring-batch-core/src/main/java/org/springframework/batch/core/repository/dao/MapJobExecutionDao.java index 2b4effdb5..fbb2f5181 100644 --- a/spring-batch-core/src/main/java/org/springframework/batch/core/repository/dao/MapJobExecutionDao.java +++ b/spring-batch-core/src/main/java/org/springframework/batch/core/repository/dao/MapJobExecutionDao.java @@ -23,7 +23,11 @@ import java.util.HashSet; import java.util.List; import java.util.Map; import java.util.Set; + import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.ConcurrentMap; + +import java.util.concurrent.atomic.AtomicLong; import org.springframework.batch.core.JobExecution; import org.springframework.batch.core.JobInstance; @@ -36,9 +40,10 @@ import org.springframework.util.Assert; */ public class MapJobExecutionDao implements JobExecutionDao { - private Map executionsById = new ConcurrentHashMap(); + // JDK6 Make this into a ConcurrentSkipListMap: adds and removes tend to be very near the front or back + private final ConcurrentMap executionsById = new ConcurrentHashMap(); - private long currentId = 0; + private final AtomicLong currentId = new AtomicLong(0L); public void clear() { executionsById.clear(); @@ -51,7 +56,7 @@ public class MapJobExecutionDao implements JobExecutionDao { public void saveJobExecution(JobExecution jobExecution) { Assert.isTrue(jobExecution.getId() == null); - Long newId = currentId++; + Long newId = currentId.getAndIncrement(); jobExecution.setId(newId); jobExecution.incrementVersion(); executionsById.put(newId, copy(jobExecution)); diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/repository/dao/MapJobInstanceDao.java b/spring-batch-core/src/main/java/org/springframework/batch/core/repository/dao/MapJobInstanceDao.java index 552e71d78..ec62ddae6 100644 --- a/spring-batch-core/src/main/java/org/springframework/batch/core/repository/dao/MapJobInstanceDao.java +++ b/spring-batch-core/src/main/java/org/springframework/batch/core/repository/dao/MapJobInstanceDao.java @@ -21,8 +21,12 @@ import java.util.Collection; import java.util.Collections; import java.util.Comparator; import java.util.List; +import java.util.Set; + import java.util.concurrent.CopyOnWriteArraySet; +import java.util.concurrent.atomic.AtomicLong; + import org.springframework.batch.core.JobExecution; import org.springframework.batch.core.JobInstance; import org.springframework.batch.core.JobParameters; @@ -33,9 +37,10 @@ import org.springframework.util.Assert; */ public class MapJobInstanceDao implements JobInstanceDao { - private Collection jobInstances = new CopyOnWriteArraySet(); + // JDK6 Make a ConcurrentSkipListSet: tends to add on end + private final Set jobInstances = new CopyOnWriteArraySet(); - private long currentId = 0; + private final AtomicLong currentId = new AtomicLong(0L); public void clear() { jobInstances.clear(); @@ -45,7 +50,7 @@ public class MapJobInstanceDao implements JobInstanceDao { Assert.state(getJobInstance(jobName, jobParameters) == null, "JobInstance must not already exist"); - JobInstance jobInstance = new JobInstance(currentId++, jobParameters, jobName); + JobInstance jobInstance = new JobInstance(currentId.getAndIncrement(), jobParameters, jobName); jobInstance.incrementVersion(); jobInstances.add(jobInstance); diff --git a/spring-batch-core/src/test/java/org/springframework/batch/core/launch/support/SimpleJobOperatorTests.java b/spring-batch-core/src/test/java/org/springframework/batch/core/launch/support/SimpleJobOperatorTests.java index 85ecf00ac..9de0d88da 100644 --- a/spring-batch-core/src/test/java/org/springframework/batch/core/launch/support/SimpleJobOperatorTests.java +++ b/spring-batch-core/src/test/java/org/springframework/batch/core/launch/support/SimpleJobOperatorTests.java @@ -26,6 +26,7 @@ import static org.junit.Assert.fail; import java.util.Arrays; import java.util.Collection; import java.util.Collections; +import java.util.HashSet; import java.util.List; import java.util.Map; import java.util.Properties; @@ -100,8 +101,8 @@ public class SimpleJobOperatorTests { } @Override - public Collection getJobNames() { - return Arrays.asList(new String[] { "foo", "bar" }); + public Set getJobNames() { + return new HashSet(Arrays.asList(new String[] { "foo", "bar" })); } }); diff --git a/spring-batch-core/src/test/java/org/springframework/batch/core/repository/dao/MapExecutionContextDaoTests.java b/spring-batch-core/src/test/java/org/springframework/batch/core/repository/dao/MapExecutionContextDaoTests.java index 7fd900d4c..f941fa8a5 100644 --- a/spring-batch-core/src/test/java/org/springframework/batch/core/repository/dao/MapExecutionContextDaoTests.java +++ b/spring-batch-core/src/test/java/org/springframework/batch/core/repository/dao/MapExecutionContextDaoTests.java @@ -3,6 +3,7 @@ package org.springframework.batch.core.repository.dao; import static org.junit.Assert.*; import org.junit.Test; +import org.junit.Ignore; import org.junit.internal.runners.JUnit4ClassRunner; import org.junit.runner.RunWith; import org.springframework.batch.core.JobExecution; @@ -34,7 +35,31 @@ public class MapExecutionContextDaoTests extends AbstractExecutionContextDaoTest protected ExecutionContextDao getExecutionContextDao() { return new MapExecutionContextDao(); } + + @Test + public void testSaveBothJobAndStepContextWithSameId() throws Exception { + MapExecutionContextDao tested = new MapExecutionContextDao(); + JobExecution jobExecution = new JobExecution(1L); + StepExecution stepExecution = new StepExecution("stepName", jobExecution, 1L); + + assertTrue(stepExecution.getId() == jobExecution.getId()); + + jobExecution.getExecutionContext().put("type", "job"); + stepExecution.getExecutionContext().put("type", "step"); + assertTrue(!jobExecution.getExecutionContext().get("type").equals(stepExecution.getExecutionContext().get("type"))); + assertEquals("job", jobExecution.getExecutionContext().get("type")); + assertEquals("step", stepExecution.getExecutionContext().get("type")); + + tested.saveExecutionContext(jobExecution); + tested.saveExecutionContext(stepExecution); + ExecutionContext jobCtx = tested.getExecutionContext(jobExecution); + ExecutionContext stepCtx = tested.getExecutionContext(stepExecution); + + assertEquals("job", jobCtx.get("type")); + assertEquals("step", stepCtx.get("type")); + } + @Test public void testPersistentCopy() throws Exception { MapExecutionContextDao tested = new MapExecutionContextDao(); diff --git a/spring-batch-core/src/test/java/org/springframework/batch/core/repository/dao/MapJobExecutionDaoTests.java b/spring-batch-core/src/test/java/org/springframework/batch/core/repository/dao/MapJobExecutionDaoTests.java index d0a83d5f7..fc451aadf 100644 --- a/spring-batch-core/src/test/java/org/springframework/batch/core/repository/dao/MapJobExecutionDaoTests.java +++ b/spring-batch-core/src/test/java/org/springframework/batch/core/repository/dao/MapJobExecutionDaoTests.java @@ -1,6 +1,18 @@ package org.springframework.batch.core.repository.dao; +import java.util.ArrayList; +import java.util.Collections; import java.util.Date; +import java.util.HashSet; +import java.util.List; +import java.util.Set; +import java.util.SortedSet; +import java.util.TreeSet; + +import java.util.concurrent.CopyOnWriteArraySet; +import java.util.concurrent.CountDownLatch; + +import java.util.concurrent.atomic.AtomicReference; import org.junit.Test; import org.junit.runner.RunWith; @@ -45,4 +57,63 @@ public class MapJobExecutionDaoTests extends AbstractJobExecutionDaoTests { } + /** + * Verify that the ids are properly generated even under heavy concurrent load + */ + @Test + public void testConcurrentSaveJobExecution() throws Exception { + final int iterations = 100; + + // Object under test + final JobExecutionDao tested = new MapJobExecutionDao(); + + // Support objects for this testing + final CountDownLatch latch = new CountDownLatch(1); + final SortedSet ids = Collections.synchronizedSortedSet(new TreeSet()); // TODO Change to SkipList w/JDK6 + final AtomicReference exception = new AtomicReference(null); + + // Implementation of the high-concurrency code + final Runnable codeUnderTest = new Runnable() { + public void run() { + try { + JobExecution jobExecution = new JobExecution(new JobInstance((long) -1, new JobParameters(), "mapJob")); + latch.await(); + tested.saveJobExecution(jobExecution); + ids.add(jobExecution.getId()); + } catch(Exception e) { + exception.set(e); + } + } + }; + + // Create the threads + final Thread[] threads = new Thread[iterations]; + for(int i = 0; i < iterations; i++) { + Thread t = new Thread(codeUnderTest, "Map Job Thread #" + (i+1)); + t.setPriority(Thread.MAX_PRIORITY); + t.setDaemon(true); + t.start(); + Thread.yield(); + threads[i] = t; + } + + // Let the high concurrency abuse begin! + do { latch.countDown(); } while(latch.getCount() > 0); + for(Thread t : threads) { t.join(); } + + // Ensure no general exceptions arose + if(exception.get() != null) throw new RuntimeException("Excepion occurred under high concurrency usage", exception.get()); + + // Validate the ids: we'd expect one of these three things to fail + if(ids.size() < iterations) { + fail("Duplicate id generated during high concurrency usage"); + } + if(ids.first() < 0) { + fail("Generated an id less than zero during high concurrency usage: " + ids.first()); + } + if(ids.last() > iterations) { + fail("Generated an id larger than expected during high concurrency usage: " + ids.last()); + } + } + } diff --git a/spring-batch-core/src/test/java/org/springframework/batch/core/step/item/FaultTolerantStepFactoryBeanRollbackTests.java b/spring-batch-core/src/test/java/org/springframework/batch/core/step/item/FaultTolerantStepFactoryBeanRollbackTests.java index a94e23fe6..f44cf2f21 100644 --- a/spring-batch-core/src/test/java/org/springframework/batch/core/step/item/FaultTolerantStepFactoryBeanRollbackTests.java +++ b/spring-batch-core/src/test/java/org/springframework/batch/core/step/item/FaultTolerantStepFactoryBeanRollbackTests.java @@ -17,6 +17,7 @@ import java.util.Map; import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; +import org.junit.After; import org.junit.Before; import org.junit.Test; import org.springframework.batch.core.BatchStatus; @@ -58,15 +59,13 @@ public class FaultTolerantStepFactoryBeanRollbackTests { private JobRepository repository; - public FaultTolerantStepFactoryBeanRollbackTests() throws Exception { - reader = new SkipReaderStub(); - processor = new SkipProcessorStub(); - writer = new SkipWriterStub(); - } - @SuppressWarnings("unchecked") @Before public void setUp() throws Exception { + reader = new SkipReaderStub(); + processor = new SkipProcessorStub(); + writer = new SkipWriterStub(); + factory = new FaultTolerantStepFactoryBean(); factory.setBeanName("stepName"); @@ -96,6 +95,14 @@ public class FaultTolerantStepFactoryBeanRollbackTests { stepExecution = jobExecution.createStepExecution(factory.getName()); repository.add(stepExecution); } + + @After + public void tearDown() throws Exception { + reader = null; + processor = null; + writer = null; + factory = null; + } @Test public void testBeforeChunkListenerException() throws Exception{ 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 a2b6c04b6..9254b449b 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 @@ -222,6 +222,8 @@ public class StepExecutorInterruptionTests { } } }; + processingThread.setDaemon(true); + processingThread.setPriority(Thread.MIN_PRIORITY); return processingThread; } diff --git a/spring-batch-infrastructure/src/main/java/org/springframework/batch/support/transaction/TransactionAwareProxyFactory.java b/spring-batch-infrastructure/src/main/java/org/springframework/batch/support/transaction/TransactionAwareProxyFactory.java index efac6507a..c54ac474e 100644 --- a/spring-batch-infrastructure/src/main/java/org/springframework/batch/support/transaction/TransactionAwareProxyFactory.java +++ b/spring-batch-infrastructure/src/main/java/org/springframework/batch/support/transaction/TransactionAwareProxyFactory.java @@ -24,6 +24,7 @@ import java.util.List; import java.util.Map; import java.util.Set; import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.ConcurrentMap; import java.util.concurrent.CopyOnWriteArrayList; import java.util.concurrent.CopyOnWriteArraySet; @@ -165,8 +166,8 @@ public class TransactionAwareProxyFactory { return (Map) new TransactionAwareProxyFactory>(new ConcurrentHashMap(map)).createInstance(); } - public static Map createAppendOnlyTransactionalMap() { - return (Map) new TransactionAwareProxyFactory>(new ConcurrentHashMap(), true).createInstance(); + public static ConcurrentMap createAppendOnlyTransactionalMap() { + return new TransactionAwareProxyFactory>(new ConcurrentHashMap(), true).createInstance(); } public static Set createAppendOnlyTransactionalSet() {