Get rid of all synchronized blocks and methods in production code

JAVA-5105

Author: stIncMale

bson/src/main/org/bson/codecs/pojo/LazyPropertyModelCodec.java

@@ -27,6 +27,8 @@
 
 import java.util.ArrayList;
 import java.util.List;
+import java.util.concurrent.locks.Lock;
+import java.util.concurrent.locks.ReentrantLock;
 
 import static java.lang.String.format;
 import static org.bson.codecs.pojo.PojoSpecializationHelper.specializeTypeData;
@@ -35,7 +37,7 @@ class LazyPropertyModelCodec<T> implements Codec<T> {
     private final PropertyModel<T> propertyModel;
     private final CodecRegistry registry;
     private final PropertyCodecRegistry propertyCodecRegistry;
-
+    private final Lock codecLock = new ReentrantLock();
     private volatile Codec<T> codec;
 
     LazyPropertyModelCodec(final PropertyModel<T> propertyModel, final CodecRegistry registry,
@@ -61,11 +63,17 @@ public Class<T> getEncoderClass() {
     }
 
     private Codec<T> getPropertyModelCodec() {
+        Codec<T> codec = this.codec;
         if (codec == null) {
-            synchronized (this) {
+            codecLock.lock();
+            try {
+                codec = this.codec;
                 if (codec == null) {
                     codec = createCodec();
+                    this.codec = codec;
                 }
+            } finally {
+                codecLock.unlock();
             }
         }
         return codec;

driver-core/src/main/com/mongodb/connection/netty/NettyStream.java

@@ -64,6 +64,8 @@
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.Future;
 import java.util.concurrent.ScheduledFuture;
+import java.util.concurrent.locks.Lock;
+import java.util.concurrent.locks.ReentrantLock;
 
 import static com.mongodb.assertions.Assertions.assertNotNull;
 import static com.mongodb.assertions.Assertions.isTrueArgument;
@@ -120,8 +122,8 @@ final class NettyStream implements Stream {
     private volatile Channel channel;
 
     private final LinkedList<io.netty.buffer.ByteBuf> pendingInboundBuffers = new LinkedList<>();
-    /* The fields pendingReader, pendingException are always written/read inside synchronized blocks
-     * that use the same NettyStream object, so they can be plain.*/
+    private final Lock lock = new ReentrantLock();
+    // access to the fields `pendingReader`, `pendingException` is guarded by `lock`
     private PendingReader pendingReader;
     private Throwable pendingException;
     /* The fields readTimeoutTask, readTimeoutMillis are each written only in the ChannelInitializer.initChannel method
@@ -282,7 +284,8 @@ public void readAsync(final int numBytes, final AsyncCompletionHandler<ByteBuf>
     private void readAsync(final int numBytes, final AsyncCompletionHandler<ByteBuf> handler, final long readTimeoutMillis) {
         ByteBuf buffer = null;
         Throwable exceptionResult = null;
-        synchronized (this) {
+        lock.lock();
+        try {
             exceptionResult = pendingException;
             if (exceptionResult == null) {
                 if (!hasBytesAvailable(numBytes)) {
@@ -316,6 +319,8 @@ private void readAsync(final int numBytes, final AsyncCompletionHandler<ByteBuf>
                 cancel(pendingReader.timeout);
                 this.pendingReader = null;
             }
+        } finally {
+            lock.unlock();
         }
         if (exceptionResult != null) {
             handler.failed(exceptionResult);
@@ -338,13 +343,16 @@ private boolean hasBytesAvailable(final int numBytes) {
 
     private void handleReadResponse(@Nullable final io.netty.buffer.ByteBuf buffer, @Nullable final Throwable t) {
         PendingReader localPendingReader = null;
-        synchronized (this) {
+        lock.lock();
+        try {
             if (buffer != null) {
                 pendingInboundBuffers.add(buffer.retain());
             } else {
                 pendingException = t;
             }
             localPendingReader = pendingReader;
+        } finally {
+            lock.unlock();
         }
 
         if (localPendingReader != null) {
@@ -359,16 +367,21 @@ public ServerAddress getAddress() {
     }
 
     @Override
-    public synchronized void close() {
-        isClosed = true;
-        if (channel != null) {
-            channel.close();
-            channel = null;
-        }
-        for (Iterator<io.netty.buffer.ByteBuf> iterator = pendingInboundBuffers.iterator(); iterator.hasNext();) {
-            io.netty.buffer.ByteBuf nextByteBuf = iterator.next();
-            iterator.remove();
-            nextByteBuf.release();
+    public void close() {
+        lock.lock();
+        try {
+            isClosed = true;
+            if (channel != null) {
+                channel.close();
+                channel = null;
+            }
+            for (Iterator<io.netty.buffer.ByteBuf> iterator = pendingInboundBuffers.iterator(); iterator.hasNext();) {
+                io.netty.buffer.ByteBuf nextByteBuf = iterator.next();
+                iterator.remove();
+                nextByteBuf.release();
+            }
+        } finally {
+            lock.unlock();
         }
     }
 
@@ -504,7 +517,8 @@ private class OpenChannelFutureListener implements ChannelFutureListener {
 
         @Override
         public void operationComplete(final ChannelFuture future) {
-            synchronized (NettyStream.this) {
+            lock.lock();
+            try {
                 if (future.isSuccess()) {
                     if (isClosed) {
                         channelFuture.channel().close();
@@ -522,6 +536,8 @@ public void operationComplete(final ChannelFuture future) {
                         initializeChannel(handler, socketAddressQueue);
                     }
                 }
+            } finally {
+                lock.unlock();
             }
         }
     }

driver-core/src/main/com/mongodb/internal/authentication/AzureCredentialHelper.java

@@ -26,6 +26,8 @@
 import java.util.HashMap;
 import java.util.Map;
 import java.util.Optional;
+import java.util.concurrent.locks.Lock;
+import java.util.concurrent.locks.ReentrantLock;
 
 import static com.mongodb.internal.authentication.HttpHelper.getHttpContents;
 
@@ -37,42 +39,52 @@
 public final class AzureCredentialHelper {
     private static final String ACCESS_TOKEN_FIELD = "access_token";
     private static final String EXPIRES_IN_FIELD = "expires_in";
+    private static final Lock CACHED_ACCESS_TOKEN_LOCK = new ReentrantLock();
+    private static volatile ExpirableValue<String> cachedAccessToken = ExpirableValue.expired();
 
-    private static ExpirableValue<String> cachedAccessToken = ExpirableValue.expired();
-
-    public static synchronized BsonDocument obtainFromEnvironment() {
+    public static BsonDocument obtainFromEnvironment() {
         String accessToken;
         Optional<String> cachedValue = cachedAccessToken.getValue();
         if (cachedValue.isPresent()) {
             accessToken = cachedValue.get();
         } else {
-            String endpoint = "http://" + "169.254.169.254:80"
-                    + "/metadata/identity/oauth2/token?api-version=2018-02-01&resource=https://vault.azure.net";
+            CACHED_ACCESS_TOKEN_LOCK.lock();
+            try {
+                cachedValue = cachedAccessToken.getValue();
+                if (cachedValue.isPresent()) {
+                    accessToken = cachedValue.get();
+                } else {
+                    String endpoint = "http://" + "169.254.169.254:80"
+                            + "/metadata/identity/oauth2/token?api-version=2018-02-01&resource=https://vault.azure.net";
 
-            Map<String, String> headers = new HashMap<>();
-            headers.put("Metadata", "true");
-            headers.put("Accept", "application/json");
+                    Map<String, String> headers = new HashMap<>();
+                    headers.put("Metadata", "true");
+                    headers.put("Accept", "application/json");
 
-            long startNanoTime = System.nanoTime();
-            BsonDocument responseDocument;
-            try {
-                responseDocument = BsonDocument.parse(getHttpContents("GET", endpoint, headers));
-            } catch (JsonParseException e) {
-                throw new MongoClientException("Exception parsing JSON from Azure IMDS metadata response.", e);
-            }
+                    long startNanoTime = System.nanoTime();
+                    BsonDocument responseDocument;
+                    try {
+                        responseDocument = BsonDocument.parse(getHttpContents("GET", endpoint, headers));
+                    } catch (JsonParseException e) {
+                        throw new MongoClientException("Exception parsing JSON from Azure IMDS metadata response.", e);
+                    }
 
-            if (!responseDocument.isString(ACCESS_TOKEN_FIELD)) {
-                throw new MongoClientException(String.format(
-                        "The %s field from Azure IMDS metadata response is missing or is not a string", ACCESS_TOKEN_FIELD));
-            }
-            if (!responseDocument.isString(EXPIRES_IN_FIELD)) {
-                throw new MongoClientException(String.format(
-                        "The %s field from Azure IMDS metadata response is missing or is not a string", EXPIRES_IN_FIELD));
+                    if (!responseDocument.isString(ACCESS_TOKEN_FIELD)) {
+                        throw new MongoClientException(String.format(
+                                "The %s field from Azure IMDS metadata response is missing or is not a string", ACCESS_TOKEN_FIELD));
+                    }
+                    if (!responseDocument.isString(EXPIRES_IN_FIELD)) {
+                        throw new MongoClientException(String.format(
+                                "The %s field from Azure IMDS metadata response is missing or is not a string", EXPIRES_IN_FIELD));
+                    }
+                    accessToken = responseDocument.getString(ACCESS_TOKEN_FIELD).getValue();
+                    int expiresInSeconds = Integer.parseInt(responseDocument.getString(EXPIRES_IN_FIELD).getValue());
+                    cachedAccessToken = ExpirableValue.expirable(accessToken, Duration.ofSeconds(expiresInSeconds).minus(Duration.ofMinutes(1)),
+                            startNanoTime);
+                }
+            } finally {
+                CACHED_ACCESS_TOKEN_LOCK.unlock();
             }
-            accessToken = responseDocument.getString(ACCESS_TOKEN_FIELD).getValue();
-            int expiresInSeconds = Integer.parseInt(responseDocument.getString(EXPIRES_IN_FIELD).getValue());
-            cachedAccessToken = ExpirableValue.expirable(accessToken, Duration.ofSeconds(expiresInSeconds).minus(Duration.ofMinutes(1)),
-                    startNanoTime);
        }
        return new BsonDocument("accessToken", new BsonString(accessToken));
     }

driver-core/src/main/com/mongodb/internal/connection/AsynchronousChannelStream.java

@@ -69,16 +69,20 @@ public PowerOfTwoBufferPool getBufferProvider() {
         return bufferProvider;
     }
 
-    public synchronized ExtendedAsynchronousByteChannel getChannel() {
+    public ExtendedAsynchronousByteChannel getChannel() {
         return channel;
     }
 
-    protected synchronized void setChannel(final ExtendedAsynchronousByteChannel channel) {
+    protected void setChannel(final ExtendedAsynchronousByteChannel channel) {
         isTrue("current channel is null", this.channel == null);
         if (isClosed) {
             closeChannel(channel);
         } else {
             this.channel = channel;
+            if (isClosed) {
+                this.channel = null;
+                closeChannel(channel);
+            }
         }
     }
 
@@ -158,16 +162,14 @@ public ServerAddress getAddress() {
     }
 
     @Override
-    public synchronized void close() {
+    public void close() {
         isClosed = true;
-        try {
-            closeChannel(channel);
-        } finally {
-            channel = null;
-        }
+        ExtendedAsynchronousByteChannel channel = this.channel;
+        this.channel = null;
+        closeChannel(channel);
     }
 
-    private void closeChannel(final ExtendedAsynchronousByteChannel channel) {
+    private void closeChannel(@Nullable final ExtendedAsynchronousByteChannel channel) {
         try {
             if (channel != null) {
                 channel.close();

driver-core/src/main/com/mongodb/internal/connection/DefaultServerMonitor.java

@@ -244,11 +244,8 @@ private ServerDescription lookupServerDescription(final ServerDescription curren
                 }
             } catch (Throwable t) {
                 averageRoundTripTime.reset();
-                InternalConnection localConnection;
-                synchronized (this) {
-                    localConnection = connection;
-                    connection = null;
-                }
+                InternalConnection localConnection = connection;
+                connection = null;
                 if (localConnection != null) {
                     localConnection.close();
                 }
@@ -310,14 +307,9 @@ private long waitForSignalOrTimeout() throws InterruptedException {
         }
 
         public void cancelCurrentCheck() {
-            InternalConnection localConnection = null;
-            synchronized (this) {
-                if (connection != null && !currentCheckCancelled) {
-                    localConnection = connection;
-                    currentCheckCancelled = true;
-                }
-            }
-            if (localConnection != null) {
+            InternalConnection localConnection = connection;
+            if (localConnection != null && !currentCheckCancelled) {
+                currentCheckCancelled = true;
                 localConnection.close();
             }
         }

driver-core/src/main/com/mongodb/internal/connection/ExponentiallyWeightedMovingAverage.java

@@ -16,32 +16,37 @@
 
 package com.mongodb.internal.connection;
 
+import java.util.concurrent.atomic.AtomicLong;
+
 import static com.mongodb.assertions.Assertions.isTrueArgument;
 
 class ExponentiallyWeightedMovingAverage {
+    private static final long EMPTY = -1;
+
     private final double alpha;
-    private long average = -1;
+    private final AtomicLong average;
 
     ExponentiallyWeightedMovingAverage(final double alpha) {
         isTrueArgument("alpha >= 0.0 and <= 1.0", alpha >= 0.0 && alpha <= 1.0);
         this.alpha = alpha;
+        average = new AtomicLong(EMPTY);
     }
 
-    synchronized void reset() {
-        average = -1;
+    void reset() {
+        average.set(EMPTY);
     }
 
-    synchronized long addSample(final long sample) {
-        if (average == -1) {
-            average = sample;
-        } else {
-            average = (long) (alpha * sample + (1 - alpha) * average);
-        }
-
-        return average;
+    long addSample(final long sample) {
+        return average.accumulateAndGet(sample, (average, givenSample) -> {
+            if (average == EMPTY) {
+                return givenSample;
+            }
+            return (long) (alpha * givenSample + (1 - alpha) * average);
+        });
     }
 
-    synchronized long getAverage() {
-        return average == -1 ? 0 : average;
+    long getAverage() {
+        long average = this.average.get();
+        return average == EMPTY ? 0 : average;
     }
 }