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

Avoid synchronizing on this and use an internal monitor instead

Browse Source 
Where possible, code that previously synchronized on this (or on the
class in the case of static methods) has been updated to use an
internal monitor object instead. This allows the locking model that's
employed to be an implementation detail rather than part of the
class's API.

Classes that override a synchronized method continue to declare
the overriding method as synchronized. This ensures that locking
is consistent across the superclass and its subclass.

Closes gh-6262
This commit is contained in:
Andy Wilkinson
2016-07-01 10:00:29 +01:00
parent f9c7db1137
commit 92bb24e365
25 changed files with 685 additions and 511 deletions
Download Patch File Download Diff File Expand all files Collapse all files

4
spring-boot-devtools/src/main/java/org/springframework/boot/devtools/livereload/Connection.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2012-2015 the original author or authors.
* Copyright 2012-2016 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.
@@ -139,7 +139,7 @@ class Connection {
}
}
private synchronized void writeWebSocketFrame(Frame frame) throws IOException {
private void writeWebSocketFrame(Frame frame) throws IOException {
frame.write(this.outputStream);
}

92
spring-boot-devtools/src/main/java/org/springframework/boot/devtools/livereload/LiveReloadServer.java
View File

@@ -53,6 +53,13 @@ public class LiveReloadServer {
private static final int READ_TIMEOUT = (int) TimeUnit.SECONDS.toMillis(4);
private final ExecutorService executor = Executors
.newCachedThreadPool(new WorkerThreadFactory());
private final List<Connection> connections = new ArrayList<Connection>();
private final Object monitor = new Object();
private final int port;
private final ThreadFactory threadFactory;
@@ -61,11 +68,6 @@ public class LiveReloadServer {
private Thread listenThread;
private ExecutorService executor = Executors
.newCachedThreadPool(new WorkerThreadFactory());
private List<Connection> connections = new ArrayList<Connection>();
/**
* Create a new {@link LiveReloadServer} listening on the default port.
*/
@@ -112,29 +114,33 @@ public class LiveReloadServer {
* Start the livereload server and accept incoming connections.
* @throws IOException in case of I/O errors
*/
public synchronized void start() throws IOException {
Assert.state(!isStarted(), "Server already started");
logger.debug("Starting live reload server on port " + this.port);
this.serverSocket = new ServerSocket(this.port);
this.listenThread = this.threadFactory.newThread(new Runnable() {
public void start() throws IOException {
synchronized (this.monitor) {
Assert.state(!isStarted(), "Server already started");
logger.debug("Starting live reload server on port " + this.port);
this.serverSocket = new ServerSocket(this.port);
this.listenThread = this.threadFactory.newThread(new Runnable() {
@Override
public void run() {
acceptConnections();
}
@Override
public void run() {
acceptConnections();
}
});
this.listenThread.setDaemon(true);
this.listenThread.setName("Live Reload Server");
this.listenThread.start();
});
this.listenThread.setDaemon(true);
this.listenThread.setName("Live Reload Server");
this.listenThread.start();
}
}
/**
* Return if the server has been started.
* @return {@code true} if the server is running
*/
public synchronized boolean isStarted() {
return this.listenThread != null;
public boolean isStarted() {
synchronized (this.monitor) {
return this.listenThread != null;
}
}
/**
@@ -168,30 +174,32 @@ public class LiveReloadServer {
* Gracefully stop the livereload server.
* @throws IOException in case of I/O errors
*/
public synchronized void stop() throws IOException {
if (this.listenThread != null) {
closeAllConnections();
try {
this.executor.shutdown();
this.executor.awaitTermination(1, TimeUnit.MINUTES);
public void stop() throws IOException {
synchronized (this.monitor) {
if (this.listenThread != null) {
closeAllConnections();
try {
this.executor.shutdown();
this.executor.awaitTermination(1, TimeUnit.MINUTES);
}
catch (InterruptedException ex) {
Thread.currentThread().interrupt();
}
this.serverSocket.close();
try {
this.listenThread.join();
}
catch (InterruptedException ex) {
Thread.currentThread().interrupt();
}
this.listenThread = null;
this.serverSocket = null;
}
catch (InterruptedException ex) {
Thread.currentThread().interrupt();
}
this.serverSocket.close();
try {
this.listenThread.join();
}
catch (InterruptedException ex) {
Thread.currentThread().interrupt();
}
this.listenThread = null;
this.serverSocket = null;
}
}
private void closeAllConnections() throws IOException {
synchronized (this.connections) {
synchronized (this.monitor) {
for (Connection connection : this.connections) {
connection.close();
}
@@ -202,7 +210,7 @@ public class LiveReloadServer {
* Trigger livereload of all connected clients.
*/
public void triggerReload() {
synchronized (this.connections) {
synchronized (this.monitor) {
for (Connection connection : this.connections) {
try {
connection.triggerReload();
@@ -215,13 +223,13 @@ public class LiveReloadServer {
}
private void addConnection(Connection connection) {
synchronized (this.connections) {
synchronized (this.monitor) {
this.connections.add(connection);
}
}
private void removeConnection(Connection connection) {
synchronized (this.connections) {
synchronized (this.monitor) {
this.connections.remove(connection);
}
}

59
spring-boot-devtools/src/main/java/org/springframework/boot/devtools/restart/Restarter.java
View File

@@ -80,10 +80,24 @@ import org.springframework.util.ReflectionUtils;
*/
public class Restarter {
private static final Object INSTANCE_MONITOR = new Object();
private static final String[] NO_ARGS = {};
private static Restarter instance;
private final Set<URL> urls = new LinkedHashSet<URL>();
private final ClassLoaderFiles classLoaderFiles = new ClassLoaderFiles();
private final Map<String, Object> attributes = new HashMap<String, Object>();
private final BlockingDeque<LeakSafeThread> leakSafeThreads = new LinkedBlockingDeque<LeakSafeThread>();
private final Lock stopLock = new ReentrantLock();
private final Object monitor = new Object();
private Log logger = new DeferredLog();
private final boolean forceReferenceCleanup;
@@ -100,18 +114,8 @@ public class Restarter {
private final UncaughtExceptionHandler exceptionHandler;
private final Set<URL> urls = new LinkedHashSet<URL>();
private final ClassLoaderFiles classLoaderFiles = new ClassLoaderFiles();
private final Map<String, Object> attributes = new HashMap<String, Object>();
private final BlockingDeque<LeakSafeThread> leakSafeThreads = new LinkedBlockingDeque<LeakSafeThread>();
private boolean finished = false;
private final Lock stopLock = new ReentrantLock();
private volatile ConfigurableApplicationContext rootContext;
/**
@@ -394,15 +398,20 @@ public class Restarter {
* Called to finish {@link Restarter} initialization when application logging is
* available.
*/
synchronized void finish() {
if (!isFinished()) {
this.logger = DeferredLog.replay(this.logger, LogFactory.getLog(getClass()));
this.finished = true;
void finish() {
synchronized (this.monitor) {
if (!isFinished()) {
this.logger = DeferredLog.replay(this.logger,
LogFactory.getLog(getClass()));
this.finished = true;
}
}
}
synchronized boolean isFinished() {
return this.finished;
boolean isFinished() {
synchronized (this.monitor) {
return this.finished;
}
}
void prepare(ConfigurableApplicationContext applicationContext) {
@@ -514,7 +523,7 @@ public class Restarter {
public static void initialize(String[] args, boolean forceReferenceCleanup,
RestartInitializer initializer, boolean restartOnInitialize) {
Restarter localInstance = null;
synchronized (Restarter.class) {
synchronized (INSTANCE_MONITOR) {
if (instance == null) {
localInstance = new Restarter(Thread.currentThread(), args,
forceReferenceCleanup, initializer);
@@ -531,9 +540,11 @@ public class Restarter {
* {@link #initialize(String[]) initialization}.
* @return the restarter
*/
public synchronized static Restarter getInstance() {
Assert.state(instance != null, "Restarter has not been initialized");
return instance;
public static Restarter getInstance() {
synchronized (INSTANCE_MONITOR) {
Assert.state(instance != null, "Restarter has not been initialized");
return instance;
}
}
/**
@@ -541,7 +552,9 @@ public class Restarter {
* @param instance the instance to set
*/
final static void setInstance(Restarter instance) {
Restarter.instance = instance;
synchronized (INSTANCE_MONITOR) {
Restarter.instance = instance;
}
}
/**
@@ -549,7 +562,9 @@ public class Restarter {
* application code.
*/
public static void clearInstance() {
instance = null;
synchronized (INSTANCE_MONITOR) {
instance = null;
}
}
/**

2
spring-boot-devtools/src/main/java/org/springframework/boot/devtools/tunnel/client/HttpTunnelConnection.java
View File

@@ -150,7 +150,7 @@ public class HttpTunnelConnection implements TunnelConnection {
return size;
}
private synchronized void openNewConnection(final HttpTunnelPayload payload) {
private void openNewConnection(final HttpTunnelPayload payload) {
HttpTunnelConnection.this.executor.execute(new Runnable() {
@Override

63
spring-boot-devtools/src/main/java/org/springframework/boot/devtools/tunnel/client/TunnelClient.java
View File

@@ -46,12 +46,14 @@ public class TunnelClient implements SmartInitializingSingleton {
private static final Log logger = LogFactory.getLog(TunnelClient.class);
private final TunnelClientListeners listeners = new TunnelClientListeners();
private final Object monitor = new Object();
private final int listenPort;
private final TunnelConnection tunnelConnection;
private TunnelClientListeners listeners = new TunnelClientListeners();
private ServerThread serverThread;
public TunnelClient(int listenPort, TunnelConnection tunnelConnection) {
@@ -63,12 +65,14 @@ public class TunnelClient implements SmartInitializingSingleton {
@Override
public void afterSingletonsInstantiated() {
if (this.serverThread == null) {
try {
start();
}
catch (IOException ex) {
throw new IllegalStateException(ex);
synchronized (this.monitor) {
if (this.serverThread == null) {
try {
start();
}
catch (IOException ex) {
throw new IllegalStateException(ex);
}
}
}
}
@@ -77,35 +81,42 @@ public class TunnelClient implements SmartInitializingSingleton {
* Start the client and accept incoming connections on the port.
* @throws IOException in case of I/O errors
*/
public synchronized void start() throws IOException {
Assert.state(this.serverThread == null, "Server already started");
ServerSocketChannel serverSocketChannel = ServerSocketChannel.open();
serverSocketChannel.socket().bind(new InetSocketAddress(this.listenPort));
logger.trace("Listening for TCP traffic to tunnel on port " + this.listenPort);
this.serverThread = new ServerThread(serverSocketChannel);
this.serverThread.start();
public void start() throws IOException {
synchronized (this.monitor) {
Assert.state(this.serverThread == null, "Server already started");
ServerSocketChannel serverSocketChannel = ServerSocketChannel.open();
serverSocketChannel.socket().bind(new InetSocketAddress(this.listenPort));
logger.trace(
"Listening for TCP traffic to tunnel on port " + this.listenPort);
this.serverThread = new ServerThread(serverSocketChannel);
this.serverThread.start();
}
}
/**
* Stop the client, disconnecting any servers.
* @throws IOException in case of I/O errors
*/
public synchronized void stop() throws IOException {
if (this.serverThread != null) {
logger.trace("Closing tunnel client on port " + this.listenPort);
this.serverThread.close();
try {
this.serverThread.join(2000);
public void stop() throws IOException {
synchronized (this.monitor) {
if (this.serverThread != null) {
logger.trace("Closing tunnel client on port " + this.listenPort);
this.serverThread.close();
try {
this.serverThread.join(2000);
}
catch (InterruptedException ex) {
// Ignore
}
this.serverThread = null;
}
catch (InterruptedException ex) {
// Ignore
}
this.serverThread = null;
}
}
protected final ServerThread getServerThread() {
return this.serverThread;
synchronized (this.monitor) {
return this.serverThread;
}
}
public void addListener(TunnelClientListener listener) {

6
spring-boot-devtools/src/main/java/org/springframework/boot/devtools/tunnel/client/TunnelClientListeners.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2012-2015 the original author or authors.
* Copyright 2012-2016 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.
@@ -17,8 +17,8 @@
package org.springframework.boot.devtools.tunnel.client;
import java.nio.channels.SocketChannel;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.CopyOnWriteArrayList;
import org.springframework.util.Assert;
@@ -29,7 +29,7 @@ import org.springframework.util.Assert;
*/
class TunnelClientListeners {
private final List<TunnelClientListener> listeners = new ArrayList<TunnelClientListener>();
private final List<TunnelClientListener> listeners = new CopyOnWriteArrayList<TunnelClientListener>();
public void addListener(TunnelClientListener listener) {
Assert.notNull(listener, "Listener must not be null");

36
spring-boot-devtools/src/main/java/org/springframework/boot/devtools/tunnel/payload/HttpTunnelPayloadForwarder.java
View File

@@ -34,12 +34,14 @@ public class HttpTunnelPayloadForwarder {
private static final int MAXIMUM_QUEUE_SIZE = 100;
private final Map<Long, HttpTunnelPayload> queue = new HashMap<Long, HttpTunnelPayload>();
private final Object monitor = new Object();
private final WritableByteChannel targetChannel;
private long lastRequestSeq = 0;
private final Map<Long, HttpTunnelPayload> queue = new HashMap<Long, HttpTunnelPayload>();
/**
* Create a new {@link HttpTunnelPayloadForwarder} instance.
* @param targetChannel the target channel
@@ -49,20 +51,22 @@ public class HttpTunnelPayloadForwarder {
this.targetChannel = targetChannel;
}
public synchronized void forward(HttpTunnelPayload payload) throws IOException {
long seq = payload.getSequence();
if (this.lastRequestSeq != seq - 1) {
Assert.state(this.queue.size() < MAXIMUM_QUEUE_SIZE,
"Too many messages queued");
this.queue.put(seq, payload);
return;
}
payload.logOutgoing();
payload.writeTo(this.targetChannel);
this.lastRequestSeq = seq;
HttpTunnelPayload queuedItem = this.queue.get(seq + 1);
if (queuedItem != null) {
forward(queuedItem);
public void forward(HttpTunnelPayload payload) throws IOException {
synchronized (this.monitor) {
long seq = payload.getSequence();
if (this.lastRequestSeq != seq - 1) {
Assert.state(this.queue.size() < MAXIMUM_QUEUE_SIZE,
"Too many messages queued");
this.queue.put(seq, payload);
return;
}
payload.logOutgoing();
payload.writeTo(this.targetChannel);
this.lastRequestSeq = seq;
HttpTunnelPayload queuedItem = this.queue.get(seq + 1);
if (queuedItem != null) {
forward(queuedItem);
}
}
}