From b009bb137ca25a3b8085a9fd22a48e8b91a8692d Mon Sep 17 00:00:00 2001 From: Iwein Fuld Date: Sat, 30 Aug 2008 13:55:16 +0000 Subject: [PATCH] Backlog concurrency improvements (and tests) --- .../integration/adapter/file/Backlog.java | 52 +++++-- .../adapter/file/ConcurrentBacklogTests.java | 147 ++++++++++++++++++ 2 files changed, 190 insertions(+), 9 deletions(-) create mode 100644 org.springframework.integration.adapter/src/test/java/org/springframework/integration/adapter/file/ConcurrentBacklogTests.java diff --git a/org.springframework.integration.adapter/src/main/java/org/springframework/integration/adapter/file/Backlog.java b/org.springframework.integration.adapter/src/main/java/org/springframework/integration/adapter/file/Backlog.java index 9e1d31daff..9dcde18b80 100644 --- a/org.springframework.integration.adapter/src/main/java/org/springframework/integration/adapter/file/Backlog.java +++ b/org.springframework.integration.adapter/src/main/java/org/springframework/integration/adapter/file/Backlog.java @@ -54,6 +54,7 @@ public class Backlog> { }; private Set doneProcessing = Collections.synchronizedSet(new HashSet()); + private Set currentlyProcessing = Collections.synchronizedSet(new HashSet()); public synchronized void processSnapshot(List currentSnapshot) { @@ -83,8 +84,9 @@ public class Backlog> { * this method can be used to mark a subset of the processing buffer as * processed. Calling this method in a threaded scenario without using * {@link #prepareForProcessing(int)} 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> { */ public void prepareForProcessing(int maxBatchSize) { List 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> { * with {@link #prepareForProcessing(int)} */ 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> { */ 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 processing = this.processingBuffer.get(); - currentlyProcessing.removeAll(processing); this.backlog.addAll(processing); + currentlyProcessing.removeAll(processing); processing.clear(); } @@ -158,6 +188,10 @@ public class Backlog> { } /** + * 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 true if both the thread local processing buffer and * the backlog are empty. */ diff --git a/org.springframework.integration.adapter/src/test/java/org/springframework/integration/adapter/file/ConcurrentBacklogTests.java b/org.springframework.integration.adapter/src/test/java/org/springframework/integration/adapter/file/ConcurrentBacklogTests.java new file mode 100644 index 0000000000..9e78b8d2bd --- /dev/null +++ b/org.springframework.integration.adapter/src/test/java/org/springframework/integration/adapter/file/ConcurrentBacklogTests.java @@ -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 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 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 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; + } +}