Skip to content

Commit 47c99b5

Browse files
stIncMalestIncMale
authored
Introduce and use Cluster.withLock to achieve cluster-wide locking (#869)
JAVA-4471
1 parent 8854c2c commit 47c99b5

15 files changed

+126
-88
lines changed

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

Lines changed: 12 additions & 13 deletions
Original file line numberDiff line numberDiff line change
@@ -89,17 +89,16 @@ MongoException getSrvResolutionException() {
8989

9090
protected void initialize(final Collection<ServerAddress> serverAddresses) {
9191
ClusterDescription currentDescription = getCurrentDescription();
92-
ClusterDescription newDescription;
9392

9493
// synchronizing this code because addServer registers a callback which is re-entrant to this instance.
9594
// In other words, we are leaking a reference to "this" from the constructor.
96-
synchronized (this) {
95+
withLock(() -> {
9796
for (final ServerAddress serverAddress : serverAddresses) {
9897
addServer(serverAddress);
9998
}
100-
newDescription = updateDescription();
99+
ClusterDescription newDescription = updateDescription();
101100
fireChangeEvent(newDescription, currentDescription);
102-
}
101+
});
103102
}
104103

105104
@Override
@@ -111,14 +110,14 @@ protected void connect() {
111110

112111
@Override
113112
public void close() {
114-
synchronized (this) {
113+
withLock(() -> {
115114
if (!isClosed()) {
116115
for (final ServerTuple serverTuple : addressToServerTupleMap.values()) {
117116
serverTuple.server.close();
118117
}
119118
}
120119
super.close();
121-
}
120+
});
122121
}
123122

124123
@Override
@@ -141,7 +140,7 @@ public void serverDescriptionChanged(final ServerDescriptionChangedEvent event)
141140
}
142141

143142
void onChange(final Collection<ServerAddress> newHosts) {
144-
synchronized (this) {
143+
withLock(() -> {
145144
if (isClosed()) {
146145
return;
147146
}
@@ -165,14 +164,11 @@ void onChange(final Collection<ServerAddress> newHosts) {
165164
ClusterDescription newClusterDescription = updateDescription();
166165

167166
fireChangeEvent(newClusterDescription, oldClusterDescription);
168-
}
167+
});
169168
}
170169

171170
private void onChange(final ServerDescriptionChangedEvent event) {
172-
ClusterDescription oldClusterDescription = null;
173-
ClusterDescription newClusterDescription = null;
174-
boolean shouldUpdateDescription = true;
175-
synchronized (this) {
171+
withLock(() -> {
176172
if (isClosed()) {
177173
return;
178174
}
@@ -193,6 +189,7 @@ private void onChange(final ServerDescriptionChangedEvent event) {
193189
return;
194190
}
195191

192+
boolean shouldUpdateDescription = true;
196193
if (newDescription.isOk()) {
197194
if (clusterType == UNKNOWN && newDescription.getType() != REPLICA_SET_GHOST) {
198195
clusterType = newDescription.getClusterType();
@@ -216,6 +213,8 @@ private void onChange(final ServerDescriptionChangedEvent event) {
216213
}
217214
}
218215

216+
ClusterDescription oldClusterDescription = null;
217+
ClusterDescription newClusterDescription = null;
219218
if (shouldUpdateDescription) {
220219
serverTuple.description = newDescription;
221220
oldClusterDescription = getCurrentDescription();
@@ -224,7 +223,7 @@ private void onChange(final ServerDescriptionChangedEvent event) {
224223
if (shouldUpdateDescription) {
225224
fireChangeEvent(newClusterDescription, oldClusterDescription);
226225
}
227-
}
226+
});
228227
}
229228

230229
private boolean handleReplicaSetMemberChanged(final ServerDescription newDescription) {

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

Lines changed: 36 additions & 25 deletions
Original file line numberDiff line numberDiff line change
@@ -240,13 +240,15 @@ public boolean isClosed() {
240240
return isClosed;
241241
}
242242

243-
protected synchronized void updateDescription(final ClusterDescription newDescription) {
244-
if (LOGGER.isDebugEnabled()) {
245-
LOGGER.debug(format("Updating cluster description to %s", newDescription.getShortDescription()));
246-
}
243+
protected void updateDescription(final ClusterDescription newDescription) {
244+
withLock(() -> {
245+
if (LOGGER.isDebugEnabled()) {
246+
LOGGER.debug(format("Updating cluster description to %s", newDescription.getShortDescription()));
247+
}
247248

248-
description = newDescription;
249-
updatePhase();
249+
description = newDescription;
250+
updatePhase();
251+
});
250252
}
251253

252254
/**
@@ -265,8 +267,13 @@ public ClusterDescription getCurrentDescription() {
265267
return description;
266268
}
267269

268-
private synchronized void updatePhase() {
269-
phase.getAndSet(new CountDownLatch(1)).countDown();
270+
@Override
271+
public synchronized void withLock(final Runnable action) {
272+
action.run();
273+
}
274+
275+
private void updatePhase() {
276+
withLock(() -> phase.getAndSet(new CountDownLatch(1)).countDown());
270277
}
271278

272279
private long getMaxWaitTimeNanos() {
@@ -387,7 +394,7 @@ private ServerSelector getCompositeServerSelector(final ServerSelector serverSel
387394

388395
protected ClusterableServer createServer(final ServerAddress serverAddress,
389396
final ServerDescriptionChangedListener serverDescriptionChangedListener) {
390-
return serverFactory.create(serverAddress, serverDescriptionChangedListener, clusterClock);
397+
return serverFactory.create(this, serverAddress, serverDescriptionChangedListener, clusterClock);
391398
}
392399

393400
private void throwIfIncompatible(final ClusterDescription curDescription) {
@@ -456,26 +463,30 @@ long getRemainingTime() {
456463
}
457464
}
458465

459-
private synchronized void notifyWaitQueueHandler(final ServerSelectionRequest request) {
460-
if (isClosed) {
461-
return;
462-
}
466+
private void notifyWaitQueueHandler(final ServerSelectionRequest request) {
467+
withLock(() -> {
468+
if (isClosed) {
469+
return;
470+
}
463471

464-
waitQueue.add(request);
472+
waitQueue.add(request);
465473

466-
if (waitQueueHandler == null) {
467-
waitQueueHandler = new Thread(new WaitQueueHandler(), "cluster-" + clusterId.getValue());
468-
waitQueueHandler.setDaemon(true);
469-
waitQueueHandler.start();
470-
} else {
471-
updatePhase();
472-
}
474+
if (waitQueueHandler == null) {
475+
waitQueueHandler = new Thread(new WaitQueueHandler(), "cluster-" + clusterId.getValue());
476+
waitQueueHandler.setDaemon(true);
477+
waitQueueHandler.start();
478+
} else {
479+
updatePhase();
480+
}
481+
});
473482
}
474483

475-
private synchronized void stopWaitQueueHandler() {
476-
if (waitQueueHandler != null) {
477-
waitQueueHandler.interrupt();
478-
}
484+
private void stopWaitQueueHandler() {
485+
withLock(() -> {
486+
if (waitQueueHandler != null) {
487+
waitQueueHandler.interrupt();
488+
}
489+
});
479490
}
480491

481492
private final class WaitQueueHandler implements Runnable {

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

Lines changed: 7 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -92,4 +92,11 @@ public interface Cluster extends Closeable {
9292
* @return true if all the servers in this cluster have been closed
9393
*/
9494
boolean isClosed();
95+
96+
/**
97+
* Does the supplied {@code action} while holding a reentrant cluster-wide lock.
98+
*
99+
* @param action The action to {@linkplain Runnable#run() do}.
100+
*/
101+
void withLock(Runnable action);
95102
}

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

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -20,8 +20,8 @@
2020
import com.mongodb.connection.ServerSettings;
2121

2222
public interface ClusterableServerFactory {
23-
ClusterableServer create(ServerAddress serverAddress, ServerDescriptionChangedListener serverDescriptionChangedListener,
24-
ClusterClock clusterClock);
23+
ClusterableServer create(Cluster cluster, ServerAddress serverAddress,
24+
ServerDescriptionChangedListener serverDescriptionChangedListener, ClusterClock clusterClock);
2525

2626
ServerSettings getSettings();
2727
}

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

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -76,7 +76,7 @@ public DefaultClusterableServerFactory(final ClusterId clusterId, final ClusterS
7676
}
7777

7878
@Override
79-
public ClusterableServer create(final ServerAddress serverAddress,
79+
public ClusterableServer create(final Cluster cluster, final ServerAddress serverAddress,
8080
final ServerDescriptionChangedListener serverDescriptionChangedListener,
8181
final ClusterClock clusterClock) {
8282
ServerId serverId = new ServerId(clusterId, serverAddress);
@@ -91,7 +91,7 @@ mongoDriverInformation, emptyList(), null, serverApi),
9191
mongoDriverInformation, compressorList, commandListener, serverApi),
9292
connectionPoolSettings, internalConnectionPoolSettings, sdamProvider);
9393
ServerListener serverListener = singleServerListener(serverSettings);
94-
SdamServerDescriptionManager sdam = new DefaultSdamServerDescriptionManager(serverId, serverDescriptionChangedListener,
94+
SdamServerDescriptionManager sdam = new DefaultSdamServerDescriptionManager(cluster, serverId, serverDescriptionChangedListener,
9595
serverListener, serverMonitor, connectionPool, clusterSettings.getMode());
9696
sdamProvider.initialize(sdam);
9797
serverMonitor.start();

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

Lines changed: 8 additions & 23 deletions
Original file line numberDiff line numberDiff line change
@@ -24,9 +24,6 @@
2424
import com.mongodb.event.ServerDescriptionChangedEvent;
2525
import com.mongodb.event.ServerListener;
2626

27-
import java.util.concurrent.locks.Lock;
28-
import java.util.concurrent.locks.ReentrantLock;
29-
3027
import static com.mongodb.assertions.Assertions.assertFalse;
3128
import static com.mongodb.assertions.Assertions.assertNotNull;
3229
import static com.mongodb.assertions.Assertions.assertTrue;
@@ -36,34 +33,34 @@
3633

3734
@ThreadSafe
3835
final class DefaultSdamServerDescriptionManager implements SdamServerDescriptionManager {
36+
private final Cluster cluster;
3937
private final ServerId serverId;
4038
private final ServerDescriptionChangedListener serverDescriptionChangedListener;
4139
private final ServerListener serverListener;
4240
private final ServerMonitor serverMonitor;
4341
private final ConnectionPool connectionPool;
4442
private final ClusterConnectionMode connectionMode;
45-
private final Lock lock;
4643
private volatile ServerDescription description;
4744

48-
DefaultSdamServerDescriptionManager(final ServerId serverId,
45+
DefaultSdamServerDescriptionManager(final Cluster cluster,
46+
final ServerId serverId,
4947
final ServerDescriptionChangedListener serverDescriptionChangedListener,
5048
final ServerListener serverListener, final ServerMonitor serverMonitor,
5149
final ConnectionPool connectionPool,
5250
final ClusterConnectionMode connectionMode) {
51+
this.cluster = cluster;
5352
this.serverId = assertNotNull(serverId);
5453
this.serverDescriptionChangedListener = assertNotNull(serverDescriptionChangedListener);
5554
this.serverListener = assertNotNull(serverListener);
5655
this.serverMonitor = assertNotNull(serverMonitor);
5756
this.connectionPool = assertNotNull(connectionPool);
5857
this.connectionMode = assertNotNull(connectionMode);
5958
description = unknownConnectingServerDescription(serverId, null);
60-
lock = new ReentrantLock();
6159
}
6260

6361
@Override
6462
public void update(final ServerDescription candidateDescription) {
65-
lock.lock();
66-
try {
63+
cluster.withLock(() -> {
6764
if (TopologyVersionHelper.newer(description.getTopologyVersion(), candidateDescription.getTopologyVersion())) {
6865
return;
6966
}
@@ -85,29 +82,17 @@ public void update(final ServerDescription candidateDescription) {
8582
assertFalse(markedPoolReady);
8683
connectionPool.invalidate(candidateDescriptionException);
8784
}
88-
} finally {
89-
lock.unlock();
90-
}
85+
});
9186
}
9287

9388
@Override
9489
public void handleExceptionBeforeHandshake(final SdamIssue sdamIssue) {
95-
lock.lock();
96-
try {
97-
handleException(sdamIssue, true);
98-
} finally {
99-
lock.unlock();
100-
}
90+
cluster.withLock(() -> handleException(sdamIssue, true));
10191
}
10292

10393
@Override
10494
public void handleExceptionAfterHandshake(final SdamIssue sdamIssue) {
105-
lock.lock();
106-
try {
107-
handleException(sdamIssue, false);
108-
} finally {
109-
lock.unlock();
110-
}
95+
cluster.withLock(() -> handleException(sdamIssue, false));
11196
}
11297

11398
@Override

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

Lines changed: 6 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -159,7 +159,7 @@ private void init(final ClusterId clusterId, final ClusterableServerFactory serv
159159
.address(host)
160160
.build()),
161161
settings, serverFactory.getSettings());
162-
server = serverFactory.create(host, event -> { }, clusterClock);
162+
server = serverFactory.create(this, host, event -> { }, clusterClock);
163163

164164
clusterListener.clusterDescriptionChanged(new ClusterDescriptionChangedEvent(clusterId, description, initialDescription));
165165
}
@@ -282,6 +282,11 @@ public boolean isClosed() {
282282
return closed.get();
283283
}
284284

285+
@Override
286+
public void withLock(final Runnable action) {
287+
fail();
288+
}
289+
285290
private void handleServerSelectionRequest(final ServerSelectionRequest serverSelectionRequest) {
286291
assertTrue(initializationCompleted);
287292
if (srvRecordResolvedToMultipleHosts) {

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

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -70,7 +70,7 @@ public LoadBalancedClusterableServerFactory(final ClusterId clusterId, final Ser
7070
}
7171

7272
@Override
73-
public ClusterableServer create(final ServerAddress serverAddress,
73+
public ClusterableServer create(final Cluster cluster, final ServerAddress serverAddress,
7474
final ServerDescriptionChangedListener serverDescriptionChangedListener,
7575
final ClusterClock clusterClock) {
7676
ConnectionPool connectionPool = new DefaultConnectionPool(new ServerId(clusterId, serverAddress),

0 commit comments

Comments
 (0)