Skip to content

Commit 0f5ddf5

Browse files
christophstroblmp911de
christophstrobl
authored and
mp911de
committed
Refine locking.
1 parent b3d9fb6 commit 0f5ddf5

File tree

4 files changed

+97
-46
lines changed

4 files changed

+97
-46
lines changed

‎spring-data-mongodb/src/main/java/org/springframework/data/mongodb/core/MongoOperations.java‎

Lines changed: 10 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -18,8 +18,12 @@
1818
import java.util.Collection;
1919
import java.util.List;
2020
import java.util.Set;
21+
import java.util.concurrent.atomic.AtomicReference;
22+
import java.util.concurrent.locks.Lock;
23+
import java.util.concurrent.locks.ReentrantLock;
2124
import java.util.function.Consumer;
2225
import java.util.function.Supplier;
26+
import java.util.function.UnaryOperator;
2327
import java.util.stream.Stream;
2428

2529
import org.bson.Document;
@@ -188,16 +192,19 @@ default SessionScoped withSession(Supplier<ClientSession> sessionProvider) {
188192

189193
return new SessionScoped() {
190194

191-
private final Object lock = new Object();
192-
private @Nullable ClientSession session = null;
195+
private final Lock lock = new ReentrantLock();
196+
private @Nullable ClientSession session;
193197

194198
@Override
195199
public <T> T execute(SessionCallback<T> action, Consumer<ClientSession> onComplete) {
196200

197-
synchronized (lock) {
201+
lock.lock();
202+
try {
198203
if (session == null) {
199204
session = sessionProvider.get();
200205
}
206+
} finally {
207+
lock.unlock();
201208
}
202209

203210
try {

‎spring-data-mongodb/src/main/java/org/springframework/data/mongodb/core/convert/LazyLoadingProxyFactory.java‎

Lines changed: 12 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -22,6 +22,10 @@
2222
import java.io.ObjectOutputStream;
2323
import java.io.Serializable;
2424
import java.lang.reflect.Method;
25+
import java.util.concurrent.locks.Lock;
26+
import java.util.concurrent.locks.ReadWriteLock;
27+
import java.util.concurrent.locks.ReentrantLock;
28+
import java.util.concurrent.locks.ReentrantReadWriteLock;
2529
import java.util.function.Supplier;
2630

2731
import org.aopalliance.intercept.MethodInterceptor;
@@ -171,6 +175,8 @@ public static class LazyLoadingInterceptor
171175
}
172176
}
173177

178+
private final ReadWriteLock lock = new ReentrantReadWriteLock();
179+
174180
private final MongoPersistentProperty property;
175181
private final DbRefResolverCallback callback;
176182
private final Object source;
@@ -339,8 +345,9 @@ private void readObject(ObjectInputStream in) throws IOException {
339345
}
340346

341347
@Nullable
342-
private synchronized Object resolve() {
348+
private Object resolve() {
343349

350+
lock.readLock().lock();
344351
if (resolved) {
345352

346353
if (LOGGER.isTraceEnabled()) {
@@ -349,13 +356,15 @@ private synchronized Object resolve() {
349356
}
350357
return result;
351358
}
359+
lock.readLock().unlock();
352360

353361
try {
354362
if (LOGGER.isTraceEnabled()) {
355363
LOGGER.trace(String.format("Resolving lazy loading property %s.%s",
356364
property.getOwner() != null ? property.getOwner().getName() : "unknown", property.getName()));
357365
}
358366

367+
lock.writeLock().lock();
359368
return callback.resolve(property);
360369

361370
} catch (RuntimeException ex) {
@@ -368,6 +377,8 @@ private synchronized Object resolve() {
368377

369378
throw new LazyLoadingException("Unable to lazily resolve DBRef",
370379
translatedException != null ? translatedException : ex);
380+
} finally {
381+
lock.writeLock().unlock();
371382
}
372383
}
373384
}

‎spring-data-mongodb/src/main/java/org/springframework/data/mongodb/core/messaging/CursorReadingTask.java‎

Lines changed: 33 additions & 18 deletions
Original file line numberDiff line numberDiff line change
@@ -18,6 +18,8 @@
1818
import java.time.Duration;
1919
import java.util.concurrent.CountDownLatch;
2020
import java.util.concurrent.TimeUnit;
21+
import java.util.concurrent.locks.Lock;
22+
import java.util.concurrent.locks.ReentrantLock;
2123
import java.util.function.Supplier;
2224

2325
import org.springframework.dao.DataAccessResourceFailureException;
@@ -39,7 +41,7 @@
3941
*/
4042
abstract class CursorReadingTask<T, R> implements Task {
4143

42-
private final Object lifecycleMonitor = new Object();
44+
private final Lock lock = new ReentrantLock();
4345

4446
private final MongoTemplate template;
4547
private final SubscriptionRequest<T, R, RequestOptions> request;
@@ -86,18 +88,18 @@ public void run() {
8688
}
8789
} catch (InterruptedException e) {
8890

89-
synchronized (lifecycleMonitor) {
90-
state = State.CANCELLED;
91-
}
91+
lock.lock();
92+
state = State.CANCELLED;
93+
lock.unlock();
9294
Thread.currentThread().interrupt();
9395
break;
9496
}
9597
}
9698
} catch (RuntimeException e) {
9799

98-
synchronized (lifecycleMonitor) {
99-
state = State.CANCELLED;
100-
}
100+
lock.lock();
101+
state = State.CANCELLED;
102+
lock.unlock();
101103

102104
errorHandler.handleError(e);
103105
}
@@ -114,18 +116,19 @@ public void run() {
114116
*/
115117
private void start() {
116118

117-
synchronized (lifecycleMonitor) {
118-
if (!State.RUNNING.equals(state)) {
119-
state = State.STARTING;
120-
}
119+
lock.lock();
120+
if (!State.RUNNING.equals(state)) {
121+
state = State.STARTING;
121122
}
123+
lock.unlock();
122124

123125
do {
124126

125127
boolean valid = false;
126128

127-
synchronized (lifecycleMonitor) {
129+
lock.lock();
128130

131+
try {
129132
if (State.STARTING.equals(state)) {
130133

131134
MongoCursor<T> cursor = execute(() -> initCursor(template, request.getRequestOptions(), targetType));
@@ -137,6 +140,8 @@ private void start() {
137140
cursor.close();
138141
}
139142
}
143+
} finally {
144+
lock.unlock();
140145
}
141146

142147
if (!valid) {
@@ -145,9 +150,9 @@ private void start() {
145150
Thread.sleep(100);
146151
} catch (InterruptedException e) {
147152

148-
synchronized (lifecycleMonitor) {
149-
state = State.CANCELLED;
150-
}
153+
lock.lock();
154+
state = State.CANCELLED;
155+
lock.unlock();
151156
Thread.currentThread().interrupt();
152157
}
153158
}
@@ -163,14 +168,18 @@ private void start() {
163168
@Override
164169
public void cancel() throws DataAccessResourceFailureException {
165170

166-
synchronized (lifecycleMonitor) {
171+
lock.lock();
172+
173+
try {
167174

168175
if (State.RUNNING.equals(state) || State.STARTING.equals(state)) {
169176
this.state = State.CANCELLED;
170177
if (cursor != null) {
171178
cursor.close();
172179
}
173180
}
181+
} finally {
182+
lock.unlock();
174183
}
175184
}
176185

@@ -182,8 +191,11 @@ public boolean isLongLived() {
182191
@Override
183192
public State getState() {
184193

185-
synchronized (lifecycleMonitor) {
194+
lock.lock();
195+
try {
186196
return state;
197+
} finally {
198+
lock.unlock();
187199
}
188200
}
189201

@@ -220,10 +232,13 @@ private void emitMessage(Message<T, R> message) {
220232
@Nullable
221233
private T getNext() {
222234

223-
synchronized (lifecycleMonitor) {
235+
lock.lock();
236+
try{
224237
if (State.RUNNING.equals(state)) {
225238
return cursor.tryNext();
226239
}
240+
} finally {
241+
lock.unlock();
227242
}
228243

229244
throw new IllegalStateException(String.format("Cursor %s is not longer open", cursor));

‎spring-data-mongodb/src/main/java/org/springframework/data/mongodb/core/messaging/DefaultMessageListenerContainer.java‎

Lines changed: 42 additions & 24 deletions
Original file line numberDiff line numberDiff line change
@@ -20,6 +20,8 @@
2020
import java.util.Map;
2121
import java.util.Optional;
2222
import java.util.concurrent.Executor;
23+
import java.util.concurrent.locks.ReadWriteLock;
24+
import java.util.concurrent.locks.ReentrantReadWriteLock;
2325

2426
import org.apache.commons.logging.Log;
2527
import org.apache.commons.logging.LogFactory;
@@ -49,9 +51,11 @@ public class DefaultMessageListenerContainer implements MessageListenerContainer
4951
private final TaskFactory taskFactory;
5052
private final Optional<ErrorHandler> errorHandler;
5153

52-
private final Object lifecycleMonitor = new Object();
5354
private final Map<SubscriptionRequest, Subscription> subscriptions = new LinkedHashMap<>();
5455

56+
ReadWriteLock lifecycleMonitor = new ReentrantReadWriteLock();
57+
ReadWriteLock subscriptionMonitor = new ReentrantReadWriteLock();
58+
5559
private boolean running = false;
5660

5761
/**
@@ -109,42 +113,47 @@ public void stop(Runnable callback) {
109113
@Override
110114
public void start() {
111115

112-
synchronized (lifecycleMonitor) {
113-
114-
if (this.running) {
115-
return;
116+
lifecycleMonitor.writeLock().lock();
117+
try {
118+
if (!this.running) {
119+
subscriptions.values().stream() //
120+
.filter(it -> !it.isActive()) //
121+
.filter(TaskSubscription.class::isInstance) //
122+
.map(TaskSubscription.class::cast) //
123+
.map(TaskSubscription::getTask) //
124+
.forEach(taskExecutor::execute);
125+
126+
running = true;
116127
}
117-
118-
subscriptions.values().stream() //
119-
.filter(it -> !it.isActive()) //
120-
.filter(TaskSubscription.class::isInstance) //
121-
.map(TaskSubscription.class::cast) //
122-
.map(TaskSubscription::getTask) //
123-
.forEach(taskExecutor::execute);
124-
125-
running = true;
128+
} finally {
129+
lifecycleMonitor.writeLock().unlock();
126130
}
127131
}
128132

129133
@Override
130134
public void stop() {
131135

132-
synchronized (lifecycleMonitor) {
136+
lifecycleMonitor.writeLock().lock();
133137

138+
try {
134139
if (this.running) {
135-
136140
subscriptions.values().forEach(Cancelable::cancel);
137-
138141
running = false;
139142
}
143+
} finally {
144+
lifecycleMonitor.writeLock().unlock();
140145
}
146+
141147
}
142148

143149
@Override
144150
public boolean isRunning() {
145151

146-
synchronized (this.lifecycleMonitor) {
152+
lifecycleMonitor.writeLock().lock();
153+
try {
147154
return running;
155+
} finally {
156+
lifecycleMonitor.writeLock().unlock();
148157
}
149158
}
150159

@@ -171,35 +180,42 @@ public <S, T> Subscription register(SubscriptionRequest<S, ? super T, ? extends
171180
@Override
172181
public Optional<Subscription> lookup(SubscriptionRequest<?, ?, ?> request) {
173182

174-
synchronized (lifecycleMonitor) {
183+
subscriptionMonitor.readLock();
184+
try {
175185
return Optional.ofNullable(subscriptions.get(request));
186+
} finally {
187+
subscriptionMonitor.readLock().unlock();
176188
}
189+
177190
}
178191

179192
public Subscription register(SubscriptionRequest request, Task task) {
180193

181194
Subscription subscription = new TaskSubscription(task);
182195

183-
synchronized (lifecycleMonitor) {
184-
196+
this.subscriptionMonitor.writeLock().lock();
197+
try {
185198
if (subscriptions.containsKey(request)) {
186199
return subscriptions.get(request);
187200
}
188201

189202
this.subscriptions.put(request, subscription);
190203

191-
if (this.running) {
204+
if (this.isRunning()) {
192205
taskExecutor.execute(task);
193206
}
207+
return subscription;
208+
} finally {
209+
this.subscriptionMonitor.writeLock().unlock();
194210
}
195211

196-
return subscription;
197212
}
198213

199214
@Override
200215
public void remove(Subscription subscription) {
201216

202-
synchronized (lifecycleMonitor) {
217+
this.subscriptionMonitor.writeLock().lock();
218+
try {
203219

204220
if (subscriptions.containsValue(subscription)) {
205221

@@ -209,6 +225,8 @@ public void remove(Subscription subscription) {
209225

210226
subscriptions.values().remove(subscription);
211227
}
228+
} finally {
229+
this.subscriptionMonitor.writeLock().unlock();
212230
}
213231
}
214232

0 commit comments

Comments
 (0)