Secon pass - handle multi-lock cases

Author: tzolov

spring-integration-core/src/main/java/org/springframework/integration/expression/ReloadableResourceBundleExpressionSource.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2022 the original author or authors.
+ * Copyright 2002-2023 the original author or authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -25,6 +25,8 @@
 import java.util.Locale;
 import java.util.Map;
 import java.util.Properties;
+import java.util.concurrent.locks.Lock;
+import java.util.concurrent.locks.ReentrantLock;
 
 import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
@@ -80,16 +82,22 @@ public class ReloadableResourceBundleExpressionSource implements ExpressionSourc
 	 */
 	private final Map<String, Map<Locale, List<String>>> cachedFilenames = new HashMap<>();
 
+	private final Lock cachedFilenamesMonitor = new ReentrantLock();
+
 	/**
 	 * Cache to hold already loaded properties per filename.
 	 */
 	private final Map<String, PropertiesHolder> cachedProperties = new HashMap<>();
 
+	private final Lock cachedPropertiesMonitor = new ReentrantLock();
+
 	/**
 	 * Cache to hold merged loaded properties per locale.
 	 */
 	private final Map<Locale, PropertiesHolder> cachedMergedProperties = new HashMap<>();
 
+	private final Lock cachedMergedPropertiesMonitor = new ReentrantLock();
+
 	private final ExpressionParser parser = new SpelExpressionParser(new SpelParserConfiguration(true, true));
 
 	private String[] basenames = {};
@@ -282,7 +290,8 @@ private String getExpressionString(String key, Locale locale) {
 	 * cached forever.
 	 */
 	private PropertiesHolder getMergedProperties(Locale locale) {
-		synchronized (this.cachedMergedProperties) {
+		this.cachedMergedPropertiesMonitor.lock();
+		try {
 			PropertiesHolder mergedHolder = this.cachedMergedProperties.get(locale);
 			if (mergedHolder != null) {
 				return mergedHolder;
@@ -303,6 +312,9 @@ private PropertiesHolder getMergedProperties(Locale locale) {
 			this.cachedMergedProperties.put(locale, mergedHolder);
 			return mergedHolder;
 		}
+		finally {
+			this.cachedMergedPropertiesMonitor.unlock();
+		}
 	}
 
 	/**
@@ -316,7 +328,8 @@ private PropertiesHolder getMergedProperties(Locale locale) {
 	 * @see #calculateFilenamesForLocale
 	 */
 	private List<String> calculateAllFilenames(String basename, Locale locale) {
-		synchronized (this.cachedFilenames) {
+		this.cachedFilenamesMonitor.lock();
+		try {
 			Map<Locale, List<String>> localeMap = this.cachedFilenames.get(basename);
 			if (localeMap != null) {
 				List<String> filenames = localeMap.get(locale);
@@ -345,6 +358,9 @@ private List<String> calculateAllFilenames(String basename, Locale locale) {
 			}
 			return filenames;
 		}
+		finally {
+			this.cachedFilenamesMonitor.unlock();
+		}
 	}
 
 	/**
@@ -392,7 +408,8 @@ private List<String> calculateFilenamesForLocale(String basename, Locale locale)
 	 * @return the current PropertiesHolder for the bundle
 	 */
 	private PropertiesHolder getProperties(String filename) {
-		synchronized (this.cachedProperties) {
+		this.cachedPropertiesMonitor.lock();
+		try {
 			PropertiesHolder propHolder = this.cachedProperties.get(filename);
 			if (propHolder != null &&
 					(propHolder.getRefreshTimestamp() < 0 ||
@@ -401,6 +418,9 @@ private PropertiesHolder getProperties(String filename) {
 			}
 			return refreshProperties(filename, propHolder);
 		}
+		finally {
+			this.cachedPropertiesMonitor.unlock();
+		}
 	}
 
 	/**
@@ -539,12 +559,21 @@ private void loadFromProperties(Resource resource, String filename, InputStream
 	 */
 	public void clearCache() {
 		LOGGER.debug("Clearing entire resource bundle cache");
-		synchronized (this.cachedProperties) {
+		this.cachedPropertiesMonitor.lock();
+		try {
 			this.cachedProperties.clear();
 		}
-		synchronized (this.cachedMergedProperties) {
+		finally {
+			this.cachedPropertiesMonitor.unlock();
+		}
+
+		this.cachedMergedPropertiesMonitor.lock();
+		try {
 			this.cachedMergedProperties.clear();
 		}
+		finally {
+			this.cachedMergedPropertiesMonitor.unlock();
+		}
 	}
 
 	@Override

spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/AbstractConnectionFactory.java

@@ -80,6 +80,8 @@ public abstract class AbstractConnectionFactory extends IntegrationObjectSupport
 
 	private final Map<String, TcpConnectionSupport> connections = new ConcurrentHashMap<>();
 
+	private final Lock connectionsMonitor = new ReentrantLock();
+
 	private final BlockingQueue<PendingIO> delayedReads = new LinkedBlockingQueue<>();
 
 	private final List<TcpSender> senders = Collections.synchronizedList(new ArrayList<>());
@@ -568,7 +570,8 @@ protected Executor getTaskExecutor() {
 	@Override
 	public void stop() {
 		this.active = false;
-		synchronized (this.connections) {
+		this.connectionsMonitor.lock();
+		try {
 			Iterator<Entry<String, TcpConnectionSupport>> iterator = this.connections.entrySet().iterator();
 			while (iterator.hasNext()) {
 				TcpConnectionSupport connection = iterator.next().getValue();
@@ -582,6 +585,10 @@ public void stop() {
 				}
 			}
 		}
+		finally {
+			this.connectionsMonitor.unlock();
+		}
+
 		this.lifecycleMonitor.lock();
 		try {
 			if (this.privateExecutor) {
@@ -860,22 +867,27 @@ protected void doAccept(final Selector selector, ServerSocketChannel server, lon
 	}
 
 	protected void addConnection(TcpConnectionSupport connection) {
-		synchronized (this.connections) {
+		this.connectionsMonitor.lock();
+		try {
 			if (!this.active) {
 				connection.close();
 				return;
 			}
 			this.connections.put(connection.getConnectionId(), connection);
 			logger.debug(() -> getComponentName() + ": Added new connection: " + connection.getConnectionId());
 		}
+		finally {
+			this.connectionsMonitor.unlock();
+		}
 	}
 
 	/**
 	 * Cleans up this.connections by removing any closed connections.
 	 * @return a list of open connection ids.
 	 */
 	private List<String> removeClosedConnectionsAndReturnOpenConnectionIds() {
-		synchronized (this.connections) {
+		this.connectionsMonitor.lock();
+		try {
 			List<String> openConnectionIds = new ArrayList<>();
 			Iterator<Entry<String, TcpConnectionSupport>> iterator = this.connections.entrySet().iterator();
 			while (iterator.hasNext()) {
@@ -899,6 +911,9 @@ private List<String> removeClosedConnectionsAndReturnOpenConnectionIds() {
 			}
 			return openConnectionIds;
 		}
+		finally {
+			this.connectionsMonitor.unlock();
+		}
 	}
 
 	/**
@@ -959,7 +974,8 @@ public List<String> getOpenConnectionIds() {
 	public boolean closeConnection(String connectionId) {
 		Assert.notNull(connectionId, "'connectionId' to close must not be null");
 		// closed connections are removed from #connections in #harvestClosedConnections()
-		synchronized (this.connections) {
+		this.connectionsMonitor.lock();
+		try {
 			boolean closed = false;
 			TcpConnectionSupport connection = this.connections.remove(connectionId);
 			if (connection != null) {
@@ -975,6 +991,9 @@ public boolean closeConnection(String connectionId) {
 			}
 			return closed;
 		}
+		finally {
+			this.connectionsMonitor.unlock();
+		}
 	}
 
 	@Override