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

Backlog concurrency improvements (and tests)

Browse Source
This commit is contained in:
Iwein Fuld
2008-08-30 13:55:16 +00:00
parent 37b4ee5bd7
commit b009bb137c
2 changed files with 190 additions and 9 deletions
Download Patch File Download Diff File Expand all files Collapse all files

52
org.springframework.integration.adapter/src/main/java/org/springframework/integration/adapter/file/Backlog.java
View File

@@ -54,6 +54,7 @@ public class Backlog<T extends Comparable<T>> {
};
private Set<T> doneProcessing = Collections.synchronizedSet(new HashSet<T>());
private Set<T> currentlyProcessing = Collections.synchronizedSet(new HashSet<T>());
public synchronized void processSnapshot(List<T> currentSnapshot) {
@@ -83,8 +84,9 @@ public class Backlog<T extends Comparable<T>> {
* this method can be used to mark a subset of the processing buffer as
* processed. Calling this method in a threaded scenario without using
* <code>{@link #prepareForProcessing(int)}</code> will manipulate the
* backlog directly and allows for race conditions. This is only recommended
* when message duplication is not an issue.
* backlog directly and allows for race conditions. In threaded scenarios a
* call to {@link #selectForProcessing(int)} followed by a call to
* {@link #processed()} is recommended.
* @param items the items that have been processed
*/
public void fileProcessed(T... items) {
@@ -109,13 +111,25 @@ public class Backlog<T extends Comparable<T>> {
*/
public void prepareForProcessing(int maxBatchSize) {
List<T> processingBuffer = this.processingBuffer.get();
if (maxBatchSize == -1) {
this.backlog.drainTo(processingBuffer);
/*
* It is important to properly lock the access to backlog and
* currentlyProcessing in this case, because the removal from the
* backlog happens before addition to currently processed. If another
* thread accesses this type of state it might duplicate messages from
* the source into the backlog.
*/
synchronized (this) {
if (maxBatchSize == -1) {
this.backlog.drainTo(processingBuffer);
}
else {
this.backlog.drainTo(processingBuffer, maxBatchSize);
}
currentlyProcessing.addAll(processingBuffer);
}
else {
this.backlog.drainTo(processingBuffer, maxBatchSize);
if (logger.isDebugEnabled()) {
logger.debug("Preparing " + processingBuffer + " for processing");
}
currentlyProcessing.addAll(processingBuffer);
}
/**
@@ -134,6 +148,15 @@ public class Backlog<T extends Comparable<T>> {
* with <code>{@link #prepareForProcessing(int)}</code>
*/
public void processed() {
if (logger.isDebugEnabled()) {
logger.debug("Moving processing buffer " + processingBuffer.get() + " to doneProcessing");
}
/*
* this doesn't need synchronization because the processing buffer will
* first be moved to doneProcessing before it is removed from currently
* processing. The order and the thread safety of the used collections
* is essential.
*/
this.doneProcessing.addAll(this.processingBuffer.get());
currentlyProcessing.removeAll(processingBuffer.get());
this.processingBuffer.get().clear();
@@ -145,11 +168,18 @@ public class Backlog<T extends Comparable<T>> {
*/
public void processingFailed() {
if (logger.isDebugEnabled()) {
logger.debug("Moving all items from processing buffer to backlog. Processing has failed");
logger.debug("Moving processing buffer " + processingBuffer.get()
+ " back to backlog. Processing has failed");
}
/*
* this doesn't need synchronization because the processing buffer will
* first be moved to doneProcessing before it is removed from currently
* processing. The order and the thread safety of the used collections
* is essential.
*/
List<T> processing = this.processingBuffer.get();
currentlyProcessing.removeAll(processing);
this.backlog.addAll(processing);
currentlyProcessing.removeAll(processing);
processing.clear();
}
@@ -158,6 +188,10 @@ public class Backlog<T extends Comparable<T>> {
}
/**
* Asks the backlog if there are any more items to process. This means that
* this method is intended to return different results in different threads
* when at least one of the thread is processing. It is unlikely that it is
* useful to call this method during processing.
* @return <code>true</code> if both the thread local processing buffer and
* the backlog are empty.
*/

147
org.springframework.integration.adapter/src/test/java/org/springframework/integration/adapter/file/ConcurrentBacklogTests.java Normal file
View File

@@ -0,0 +1,147 @@
package org.springframework.integration.adapter.file;
import static org.junit.Assert.assertTrue;
import java.util.Arrays;
import java.util.Collection;
import java.util.List;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.atomic.AtomicBoolean;
import org.junit.Test;
import org.springframework.beans.DirectFieldAccessor;
@SuppressWarnings("unchecked")
public class ConcurrentBacklogTests {
@Test(timeout = 1000)
public void simultaniousPreparation() throws Exception {
final Backlog backlog = new Backlog();
backlog.processSnapshot(Arrays.asList(new String[] { "bert", "ernie", "pino", "whatsherface" }));
Runnable todo = new Runnable() {
public void run() {
backlog.prepareForProcessing(1);
}
};
CountDownLatch start = new CountDownLatch(1);
CountDownLatch done = doConcurrently(5, todo, start);
start.countDown();
done.await();
assertTrue(backlog.isEmpty());
}
@Test(timeout = 1000)
public void concurrentUnloading() throws Exception {
final Backlog backlog = new Backlog();
List<String> items = Arrays.asList(new String[] { "bert", "ernie", "pino", "whatsherface", "kaas", "pasf" });
backlog.processSnapshot(items);
final AtomicBoolean properlyUnloaded = new AtomicBoolean(false);
Runnable todo = new Runnable() {
public void run() {
backlog.prepareForProcessing(2);
backlog.processed();
properlyUnloaded.set(backlog.isEmpty() && backlog.getProcessingBuffer().isEmpty());
}
};
CountDownLatch start = new CountDownLatch(1);
CountDownLatch done = doConcurrently(3, todo, start);
start.countDown();
done.await();
assertTrue("currentlyProcessing not emptied correctly", ((Collection) new DirectFieldAccessor(backlog)
.getPropertyValue("currentlyProcessing")).isEmpty());
assertTrue("doneProcessing not populated correctly", ((Collection) new DirectFieldAccessor(backlog)
.getPropertyValue("doneProcessing")).containsAll(items));
}
@Test(timeout = 1000)
public void concurrentFailing() throws Exception {
final Backlog backlog = new Backlog();
List<String> items = Arrays.asList(new String[] { "bert", "ernie", "pino", "whatsherface", "kaas", "pasf" });
backlog.processSnapshot(items);
final AtomicBoolean properlyBackedUp = new AtomicBoolean(false);
Runnable todo = new Runnable() {
public void run() {
backlog.prepareForProcessing(2);
backlog.processingFailed();
properlyBackedUp.set(backlog.getProcessingBuffer().isEmpty());
}
};
CountDownLatch start = new CountDownLatch(1);
CountDownLatch done = doConcurrently(3, todo, start);
start.countDown();
done.await();
assertTrue("currentlyProcessing not emptied correctly", ((Collection) new DirectFieldAccessor(backlog)
.getPropertyValue("currentlyProcessing")).isEmpty());
assertTrue("backlog not repopulated correctly", ((Collection) new DirectFieldAccessor(backlog)
.getPropertyValue("backlog")).containsAll(items));
}
@Test(timeout = 1000)
public void concurrentSuccessFailure() throws Exception {
final Backlog backlog = new Backlog();
List<String> items = Arrays.asList(new String[] { "bert", "ernie", "pino", "whatsherface" });
backlog.processSnapshot(items);
final AtomicBoolean properlyBackedUp = new AtomicBoolean(true);
final AtomicBoolean properlyUnloaded = new AtomicBoolean(true);
Runnable doFailure = new Runnable() {
public void run() {
backlog.prepareForProcessing(1);
backlog.processingFailed();
properlyBackedUp.set(backlog.getProcessingBuffer().isEmpty() && properlyBackedUp.get());
}
};
Runnable doSuccess = new Runnable() {
public void run() {
backlog.prepareForProcessing(1);
backlog.processed();
properlyUnloaded.set(backlog.getProcessingBuffer().isEmpty() && properlyUnloaded.get());
}
};
CountDownLatch start = new CountDownLatch(1);
CountDownLatch doneSuccess = doConcurrently(2, doSuccess, start);
CountDownLatch doneFailure = doConcurrently(3, doFailure, start);
start.countDown();
doneSuccess.await();
doneFailure.await();
assertTrue(properlyBackedUp.get());
assertTrue(properlyUnloaded.get());
assertTrue("currentlyProcessing not emptied correctly", ((Collection) new DirectFieldAccessor(backlog)
.getPropertyValue("currentlyProcessing")).isEmpty());
Collection backlogQueue = (Collection) new DirectFieldAccessor(backlog).getPropertyValue("backlog");
assertTrue("backlog not repopulated correctly size is " + backlogQueue.size(), backlogQueue.size() == 2);
Collection doneProcessing = (Collection) new DirectFieldAccessor(backlog).getPropertyValue("doneProcessing");
assertTrue("doneProcessing not repopulated correctly size is " + doneProcessing.size(),
doneProcessing.size() == 2);
}
/**
* Convenience method to run part of a test concurrently in multiple threads
*
* @param numberOfThreads
* @param todo the runnable that should be run by all the threads
* @return a latch that will be counted down once all threads have run their
* runnable.
*/
private CountDownLatch doConcurrently(int numberOfThreads, final Runnable todo, final CountDownLatch start) {
final CountDownLatch started = new CountDownLatch(numberOfThreads);
final CountDownLatch done = new CountDownLatch(numberOfThreads);
for (int i = 0; i < numberOfThreads; i++) {
new Thread(new Runnable() {
public void run() {
started.countDown();
try {
started.await();
start.await();
}
catch (InterruptedException e) {
Thread.currentThread().interrupt();
}
todo.run();
done.countDown();
}
}).start();
}
return done;
}
}