OPEN - issue BATCH-1409: More efficient use of pool threads in repeat template (hence multi-threaded steps)

Use wait/notify instead of a Semaphore to allow calling thread to be re-scheduled if necessary.  Won't make any difference to repeats executed in the main thread, but if they are in a background thread it should be more fair to other tasks.

spring-batch-infrastructure/src/main/java/org/springframework/batch/repeat/support/ThrottleLimitResultQueue.java

@@ -19,7 +19,6 @@ package org.springframework.batch.repeat.support;
 import java.util.NoSuchElementException;
 import java.util.concurrent.BlockingQueue;
 import java.util.concurrent.LinkedBlockingQueue;
-import java.util.concurrent.Semaphore;
 
 /**
  * An implementation of the {@link ResultQueue} that throttles the number of
@@ -33,19 +32,21 @@ public class ThrottleLimitResultQueue<T> implements ResultQueue<T> {
 	private final BlockingQueue<T> results;
 
 	// Accumulation of dummy objects flagging expected results in the future.
-	private final Semaphore waits;
+	private volatile int waits = 0;
 
 	private final Object lock = new Object();
 
 	private volatile int count = 0;
 
+	private final int throttleLimit;
+
 	/**
 	 * @param throttleLimit the maximum number of results that can be expected
 	 * at any given time.
 	 */
 	public ThrottleLimitResultQueue(int throttleLimit) {
+		this.throttleLimit = throttleLimit;
 		results = new LinkedBlockingQueue<T>();
-		waits = new Semaphore(throttleLimit);
 	}
 
 	public boolean isEmpty() {
@@ -60,7 +61,7 @@ public class ThrottleLimitResultQueue<T> implements ResultQueue<T> {
 	public boolean isExpecting() {
 		// Base the decision about whether we expect more results on a
 		// counter of the number of expected results actually collected.
-		// Do not synchronize!  Otherwise put and expect can deadlock.
+		// Do not synchronize! Otherwise put and expect can deadlock.
 		return count > 0;
 	}
 
@@ -73,7 +74,10 @@ public class ThrottleLimitResultQueue<T> implements ResultQueue<T> {
 	 */
 	public void expect() throws InterruptedException {
 		synchronized (lock) {
-			waits.acquire();
+			while (waits >= throttleLimit) {
+				lock.wait();
+			}
+			waits++;
 			count++;
 		}
 	}
@@ -84,9 +88,12 @@ public class ThrottleLimitResultQueue<T> implements ResultQueue<T> {
 		}
 		// There should be no need to block here, or to use offer()
 		results.add(holder);
-		// Take from the waits queue now to allow another result to
+		// Take from the waits now to allow another result to
 		// accumulate. But don't decrement the counter.
-		waits.release();
+		synchronized (lock) {
+			waits--;
+			lock.notifyAll();
+		}
 	}
 
 	public T take() throws NoSuchElementException, InterruptedException {