Replace synchronized blocks and methods with locks

This commit replaces synchronized blocks and methods
that are used frequently or that guard blocking I/O
operations with locks. This is required to prevent
virtual threads pinning, as explained in JEP 444 [1].

Note that synchronized blocks and methods that are used
infrequently (like AutomaticJobRegistrar#start/stop) or
that guard in-memory operations were not replaced as this
is not required, see JEP 444 [1].

Resolves to #4399

---
[1]: https://openjdk.org/jeps/444

spring-batch-infrastructure/src/main/java/org/springframework/batch/item/data/AbstractPaginatedDataItemReader.java

@@ -22,6 +22,8 @@ import org.springframework.lang.Nullable;
 import org.springframework.util.Assert;
 
 import java.util.Iterator;
+import java.util.concurrent.locks.Lock;
+import java.util.concurrent.locks.ReentrantLock;
 
 /**
  * A base class that handles basic reading logic based on the paginated semantics of
@@ -44,7 +46,7 @@ public abstract class AbstractPaginatedDataItemReader<T> extends AbstractItemCou
 
 	protected Iterator<T> results;
 
-	private final Object lock = new Object();
+	private final Lock lock = new ReentrantLock();
 
 	/**
 	 * The number of items to be read with each page.
@@ -59,7 +61,8 @@ public abstract class AbstractPaginatedDataItemReader<T> extends AbstractItemCou
 	@Override
 	protected T doRead() throws Exception {
 
-		synchronized (lock) {
+		this.lock.lock();
+		try {
 			if (results == null || !results.hasNext()) {
 
 				results = doPageRead();
@@ -78,6 +81,9 @@ public abstract class AbstractPaginatedDataItemReader<T> extends AbstractItemCou
 				return null;
 			}
 		}
+		finally {
+			this.lock.unlock();
+		}
 	}
 
 	/**
@@ -101,7 +107,8 @@ public abstract class AbstractPaginatedDataItemReader<T> extends AbstractItemCou
 
 	@Override
 	protected void jumpToItem(int itemLastIndex) throws Exception {
-		synchronized (lock) {
+		this.lock.lock();
+		try {
 			page = itemLastIndex / pageSize;
 			int current = itemLastIndex % pageSize;
 
@@ -111,6 +118,9 @@ public abstract class AbstractPaginatedDataItemReader<T> extends AbstractItemCou
 				initialPage.next();
 			}
 		}
+		finally {
+			this.lock.unlock();
+		}
 	}
 
 }

spring-batch-infrastructure/src/main/java/org/springframework/batch/item/data/RepositoryItemReader.java

@@ -19,6 +19,8 @@ import java.lang.reflect.InvocationTargetException;
 import java.util.ArrayList;
 import java.util.List;
 import java.util.Map;
+import java.util.concurrent.locks.Lock;
+import java.util.concurrent.locks.ReentrantLock;
 
 import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
@@ -98,7 +100,7 @@ public class RepositoryItemReader<T> extends AbstractItemCountingItemStreamItemR
 
 	private volatile List<T> results;
 
-	private final Object lock = new Object();
+	private final Lock lock = new ReentrantLock();
 
 	private String methodName;
 
@@ -162,7 +164,8 @@ public class RepositoryItemReader<T> extends AbstractItemCountingItemStreamItemR
 	@Override
 	protected T doRead() throws Exception {
 
-		synchronized (lock) {
+		this.lock.lock();
+		try {
 			boolean nextPageNeeded = (results != null && current >= results.size());
 
 			if (results == null || nextPageNeeded) {
@@ -192,14 +195,21 @@ public class RepositoryItemReader<T> extends AbstractItemCountingItemStreamItemR
 				return null;
 			}
 		}
+		finally {
+			this.lock.unlock();
+		}
 	}
 
 	@Override
 	protected void jumpToItem(int itemLastIndex) throws Exception {
-		synchronized (lock) {
+		this.lock.lock();
+		try {
 			page = itemLastIndex / pageSize;
 			current = itemLastIndex % pageSize;
 		}
+		finally {
+			this.lock.unlock();
+		}
 	}
 
 	/**
@@ -236,11 +246,15 @@ public class RepositoryItemReader<T> extends AbstractItemCountingItemStreamItemR
 
 	@Override
 	protected void doClose() throws Exception {
-		synchronized (lock) {
+		this.lock.lock();
+		try {
 			current = 0;
 			page = 0;
 			results = null;
 		}
+		finally {
+			this.lock.unlock();
+		}
 	}
 
 	private Sort convertToSort(Map<String, Sort.Direction> sorts) {

spring-batch-infrastructure/src/main/java/org/springframework/batch/item/database/AbstractPagingItemReader.java

@@ -16,6 +16,8 @@
 package org.springframework.batch.item.database;
 
 import java.util.List;
+import java.util.concurrent.locks.Lock;
+import java.util.concurrent.locks.ReentrantLock;
 
 import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
@@ -58,7 +60,7 @@ public abstract class AbstractPagingItemReader<T> extends AbstractItemCountingIt
 
 	protected volatile List<T> results;
 
-	private final Object lock = new Object();
+	private final Lock lock = new ReentrantLock();
 
 	public AbstractPagingItemReader() {
 		setName(ClassUtils.getShortName(AbstractPagingItemReader.class));
@@ -101,7 +103,8 @@ public abstract class AbstractPagingItemReader<T> extends AbstractItemCountingIt
 	@Override
 	protected T doRead() throws Exception {
 
-		synchronized (lock) {
+		this.lock.lock();
+		try {
 
 			if (results == null || current >= pageSize) {
 
@@ -126,6 +129,9 @@ public abstract class AbstractPagingItemReader<T> extends AbstractItemCountingIt
 			}
 
 		}
+		finally {
+			this.lock.unlock();
+		}
 
 	}
 
@@ -142,22 +148,30 @@ public abstract class AbstractPagingItemReader<T> extends AbstractItemCountingIt
 	@Override
 	protected void doClose() throws Exception {
 
-		synchronized (lock) {
+		this.lock.lock();
+		try {
 			initialized = false;
 			current = 0;
 			page = 0;
 			results = null;
 		}
+		finally {
+			this.lock.unlock();
+		}
 
 	}
 
 	@Override
 	protected void jumpToItem(int itemIndex) throws Exception {
 
-		synchronized (lock) {
+		this.lock.lock();
+		try {
 			page = itemIndex / pageSize;
 			current = itemIndex % pageSize;
 		}
+		finally {
+			this.lock.unlock();
+		}
 
 		if (logger.isDebugEnabled()) {
 			logger.debug("Jumping to page " + getPage() + " and index " + current);

spring-batch-infrastructure/src/main/java/org/springframework/batch/item/database/ExtendedConnectionDataSourceProxy.java

@@ -24,6 +24,8 @@ import java.lang.reflect.Proxy;
 import java.sql.Connection;
 import java.sql.SQLException;
 import java.sql.SQLFeatureNotSupportedException;
+import java.util.concurrent.locks.Lock;
+import java.util.concurrent.locks.ReentrantLock;
 import java.util.logging.Logger;
 
 import javax.sql.DataSource;
@@ -92,7 +94,7 @@ public class ExtendedConnectionDataSourceProxy implements SmartDataSource, Initi
 	private boolean borrowedConnection = false;
 
 	/** Synchronization monitor for the shared Connection */
-	private final Object connectionMonitor = new Object();
+	private final Lock connectionMonitor = new ReentrantLock();
 
 	/**
 	 * No arg constructor for use when configured using JavaBean style.
@@ -143,12 +145,16 @@ public class ExtendedConnectionDataSourceProxy implements SmartDataSource, Initi
 	 * @param connection the {@link Connection} that close suppression is requested for
 	 */
 	public void startCloseSuppression(Connection connection) {
-		synchronized (this.connectionMonitor) {
+		this.connectionMonitor.lock();
+		try {
 			closeSuppressedConnection = connection;
 			if (TransactionSynchronizationManager.isActualTransactionActive()) {
 				borrowedConnection = true;
 			}
 		}
+		finally {
+			this.connectionMonitor.unlock();
+		}
 	}
 
 	/**
@@ -156,24 +162,36 @@ public class ExtendedConnectionDataSourceProxy implements SmartDataSource, Initi
 	 * off for
 	 */
 	public void stopCloseSuppression(Connection connection) {
-		synchronized (this.connectionMonitor) {
+		this.connectionMonitor.lock();
+		try {
 			closeSuppressedConnection = null;
 			borrowedConnection = false;
 		}
+		finally {
+			this.connectionMonitor.unlock();
+		}
 	}
 
 	@Override
 	public Connection getConnection() throws SQLException {
-		synchronized (this.connectionMonitor) {
+		this.connectionMonitor.lock();
+		try {
 			return initConnection(null, null);
 		}
+		finally {
+			this.connectionMonitor.unlock();
+		}
 	}
 
 	@Override
 	public Connection getConnection(String username, String password) throws SQLException {
-		synchronized (this.connectionMonitor) {
+		this.connectionMonitor.lock();
+		try {
 			return initConnection(username, password);
 		}
+		finally {
+			this.connectionMonitor.unlock();
+		}
 	}
 
 	private boolean completeCloseCall(Connection connection) {

spring-batch-infrastructure/src/main/java/org/springframework/batch/item/file/SimpleBinaryBufferedReaderFactory.java

@@ -20,6 +20,8 @@ import java.io.IOException;
 import java.io.InputStreamReader;
 import java.io.Reader;
 import java.io.UnsupportedEncodingException;
+import java.util.concurrent.locks.Lock;
+import java.util.concurrent.locks.ReentrantLock;
 
 import org.springframework.core.io.Resource;
 
@@ -60,12 +62,15 @@ public class SimpleBinaryBufferedReaderFactory implements BufferedReaderFactory
 	 * usual plain text conventions.
 	 *
 	 * @author Dave Syer
+	 * @author Mahmoud Ben Hassine
 	 *
 	 */
 	private static final class BinaryBufferedReader extends BufferedReader {
 
 		private final String ending;
 
+		private final Lock lock = new ReentrantLock();
+
 		private BinaryBufferedReader(Reader in, String ending) {
 			super(in);
 			this.ending = ending;
@@ -76,7 +81,8 @@ public class SimpleBinaryBufferedReaderFactory implements BufferedReaderFactory
 
 			StringBuilder buffer;
 
-			synchronized (lock) {
+			this.lock.lock();
+			try {
 
 				int next = read();
 				if (next == -1) {
@@ -92,6 +98,9 @@ public class SimpleBinaryBufferedReaderFactory implements BufferedReaderFactory
 				buffer.append(candidateEnding);
 
 			}
+			finally {
+				this.lock.unlock();
+			}
 
 			if (buffer != null && buffer.length() > 0) {
 				return buffer.toString();

spring-batch-infrastructure/src/main/java/org/springframework/batch/item/support/SynchronizedItemStreamReader.java

@@ -15,6 +15,9 @@
  */
 package org.springframework.batch.item.support;
 
+import java.util.concurrent.locks.Lock;
+import java.util.concurrent.locks.ReentrantLock;
+
 import org.springframework.batch.item.ExecutionContext;
 import org.springframework.batch.item.ItemStreamReader;
 import org.springframework.batch.item.NonTransientResourceException;
@@ -35,6 +38,7 @@ import org.springframework.util.Assert;
  * Here is the motivation behind this class: https://stackoverflow.com/a/20002493/2910265
  *
  * @author Matthew Ouyang
+ * @author Mahmoud Ben Hassine
  * @since 3.0.4
  * @param <T> type of object being read
  */
@@ -42,6 +46,8 @@ public class SynchronizedItemStreamReader<T> implements ItemStreamReader<T>, Ini
 
 	private ItemStreamReader<T> delegate;
 
+	private final Lock lock = new ReentrantLock();
+
 	public void setDelegate(ItemStreamReader<T> delegate) {
 		this.delegate = delegate;
 	}
@@ -50,8 +56,14 @@ public class SynchronizedItemStreamReader<T> implements ItemStreamReader<T>, Ini
 	 * This delegates to the read method of the <code>delegate</code>
 	 */
 	@Nullable
-	public synchronized T read() throws Exception {
-		return this.delegate.read();
+	public T read() throws Exception {
+		this.lock.lock();
+		try {
+			return this.delegate.read();
+		}
+		finally {
+			this.lock.unlock();
+		}
 	}
 
 	public void close() {

spring-batch-infrastructure/src/main/java/org/springframework/batch/item/support/SynchronizedItemStreamWriter.java

@@ -15,6 +15,9 @@
  */
 package org.springframework.batch.item.support;
 
+import java.util.concurrent.locks.Lock;
+import java.util.concurrent.locks.ReentrantLock;
+
 import org.springframework.batch.item.Chunk;
 import org.springframework.batch.item.ExecutionContext;
 import org.springframework.batch.item.ItemStreamException;
@@ -47,6 +50,8 @@ public class SynchronizedItemStreamWriter<T> implements ItemStreamWriter<T>, Ini
 
 	private ItemStreamWriter<T> delegate;
 
+	private final Lock lock = new ReentrantLock();
+
 	/**
 	 * Set the delegate {@link ItemStreamWriter}.
 	 * @param delegate the delegate to set
@@ -59,8 +64,14 @@ public class SynchronizedItemStreamWriter<T> implements ItemStreamWriter<T>, Ini
 	 * This method delegates to the {@code write} method of the {@code delegate}.
 	 */
 	@Override
-	public synchronized void write(Chunk<? extends T> items) throws Exception {
-		this.delegate.write(items);
+	public void write(Chunk<? extends T> items) throws Exception {
+		this.lock.lock();
+		try {
+			this.delegate.write(items);
+		}
+		finally {
+			this.lock.unlock();
+		}
 	}
 
 	@Override