Skip to content

Commit

Permalink
Refine locking.
Browse files Browse the repository at this point in the history 
Closes: #4429
Original Pull Request: #4431
  • Loading branch information
@christophstrobl
christophstrobl committed Oct 5, 2023
1 parent fd96974 commit cc56a93
Show file tree
Hide file tree
Showing 4 changed files with 139 additions and 89 deletions.
13 changes: 10 additions & 3 deletions spring-data-mongodb/src/main/java/org/springframework/data/mongodb/core/MongoOperations.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -18,8 +18,12 @@
import java.util.Collection;
import java.util.List;
import java.util.Set;
import java.util.concurrent.atomic.AtomicReference;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;
import java.util.function.Consumer;
import java.util.function.Supplier;
import java.util.function.UnaryOperator;
import java.util.stream.Stream;

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

return new SessionScoped() {

private final Object lock = new Object();
private @Nullable ClientSession session = null;
private final Lock lock = new ReentrantLock();
private @Nullable ClientSession session;

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

synchronized (lock) {
lock.lock();
try {
if (session == null) {
session = sessionProvider.get();
}
} finally {
lock.unlock();
}

try {
Expand Down
48 changes: 34 additions & 14 deletions .../src/main/java/org/springframework/data/mongodb/core/convert/LazyLoadingProxyFactory.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -22,6 +22,9 @@
import java.io.ObjectOutputStream;
import java.io.Serializable;
import java.lang.reflect.Method;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReadWriteLock;
import java.util.concurrent.locks.ReentrantReadWriteLock;
import java.util.function.Supplier;

import org.aopalliance.intercept.MethodInterceptor;
Expand Down Expand Up @@ -134,7 +137,8 @@ public Object createLazyLoadingProxy(MongoPersistentProperty property, DbRefReso
}

return prepareProxyFactory(propertyType,
() -> new LazyLoadingInterceptor(property, callback, source, exceptionTranslator)).getProxy(LazyLoadingProxy.class.getClassLoader());
() -> new LazyLoadingInterceptor(property, callback, source, exceptionTranslator))
.getProxy(LazyLoadingProxy.class.getClassLoader());
}

/**
Expand Down Expand Up @@ -171,6 +175,8 @@ public static class LazyLoadingInterceptor
}
}

private final ReadWriteLock lock = new ReentrantReadWriteLock();

private final MongoPersistentProperty property;
private final DbRefResolverCallback callback;
private final Object source;
Expand Down Expand Up @@ -339,25 +345,29 @@ private void readObject(ObjectInputStream in) throws IOException {
}

@Nullable
private synchronized Object resolve() {
private Object resolve() {

if (resolved) {
lock.readLock().lock();
try {
if (resolved) {

if (LOGGER.isTraceEnabled()) {
LOGGER.trace(String.format("Accessing already resolved lazy loading property %s.%s",
property.getOwner() != null ? property.getOwner().getName() : "unknown", property.getName()));
if (LOGGER.isTraceEnabled()) {
LOGGER.trace(String.format("Accessing already resolved lazy loading property %s.%s",
property.getOwner() != null ? property.getOwner().getName() : "unknown", property.getName()));
}
return result;
}
return result;
} finally {
lock.readLock().unlock();
}

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

return callback.resolve(property);
if (LOGGER.isTraceEnabled()) {
LOGGER.trace(String.format("Resolving lazy loading property %s.%s",
property.getOwner() != null ? property.getOwner().getName() : "unknown", property.getName()));
}

try {
return executeWhileLocked(lock.writeLock(), () -> callback.resolve(property));
} catch (RuntimeException ex) {

DataAccessException translatedException = exceptionTranslator.translateExceptionIfPossible(ex);
Expand All @@ -370,6 +380,16 @@ private synchronized Object resolve() {
translatedException != null ? translatedException : ex);
}
}

private static <T> T executeWhileLocked(Lock lock, Supplier<T> stuff) {

lock.lock();
try {
return stuff.get();
} finally {
lock.unlock();
}
}
}

}
82 changes: 47 additions & 35 deletions ...godb/src/main/java/org/springframework/data/mongodb/core/messaging/CursorReadingTask.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -18,6 +18,8 @@
import java.time.Duration;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;
import java.util.function.Supplier;

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

private final Object lifecycleMonitor = new Object();
private final Lock lock = new ReentrantLock();

private final MongoTemplate template;
private final SubscriptionRequest<T, R, RequestOptions> request;
Expand Down Expand Up @@ -86,19 +88,14 @@ public void run() {
}
} catch (InterruptedException e) {

synchronized (lifecycleMonitor) {
state = State.CANCELLED;
}
doWhileLocked(lock, () -> state = State.CANCELLED);
Thread.currentThread().interrupt();
break;
}
}
} catch (RuntimeException e) {

synchronized (lifecycleMonitor) {
state = State.CANCELLED;
}

doWhileLocked(lock, () -> state = State.CANCELLED);
errorHandler.handleError(e);
}
}
Expand All @@ -114,40 +111,40 @@ public void run() {
*/
private void start() {

synchronized (lifecycleMonitor) {
doWhileLocked(lock, () -> {
if (!State.RUNNING.equals(state)) {
state = State.STARTING;
}
}
});

do {

boolean valid = false;
// boolean valid = false;

synchronized (lifecycleMonitor) {
boolean valid = executeWhileLocked(lock, () -> {

if (State.STARTING.equals(state)) {
if (!State.STARTING.equals(state)) {
return false;
}

MongoCursor<T> cursor = execute(() -> initCursor(template, request.getRequestOptions(), targetType));
valid = isValidCursor(cursor);
if (valid) {
this.cursor = cursor;
state = State.RUNNING;
} else if (cursor != null) {
cursor.close();
}
MongoCursor<T> cursor = execute(() -> initCursor(template, request.getRequestOptions(), targetType));
boolean isValid = isValidCursor(cursor);
if (isValid) {
this.cursor = cursor;
state = State.RUNNING;
} else if (cursor != null) {
cursor.close();
}
}
return isValid;
});

if (!valid) {

try {
Thread.sleep(100);
} catch (InterruptedException e) {

synchronized (lifecycleMonitor) {
state = State.CANCELLED;
}
doWhileLocked(lock, () -> state = State.CANCELLED);
Thread.currentThread().interrupt();
}
}
Expand All @@ -163,15 +160,15 @@ private void start() {
@Override
public void cancel() throws DataAccessResourceFailureException {

synchronized (lifecycleMonitor) {
doWhileLocked(lock, () -> {

if (State.RUNNING.equals(state) || State.STARTING.equals(state)) {
this.state = State.CANCELLED;
if (cursor != null) {
cursor.close();
}
}
}
});
}

@Override
Expand All @@ -181,10 +178,7 @@ public boolean isLongLived() {

@Override
public State getState() {

synchronized (lifecycleMonitor) {
return state;
}
return executeWhileLocked(lock, () -> state);
}

@Override
Expand Down Expand Up @@ -220,13 +214,12 @@ private void emitMessage(Message<T, R> message) {
@Nullable
private T getNext() {

synchronized (lifecycleMonitor) {
return executeWhileLocked(lock, () -> {
if (State.RUNNING.equals(state)) {
return cursor.tryNext();
}
}

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

private static boolean isValidCursor(@Nullable MongoCursor<?> cursor) {
Expand Down Expand Up @@ -263,4 +256,23 @@ private <V> V execute(Supplier<V> callback) {
throw translated != null ? translated : e;
}
}

private static void doWhileLocked(Lock lock, Runnable action) {

executeWhileLocked(lock, () -> {
action.run();
return null;
});
}

@Nullable
private static <T> T executeWhileLocked(Lock lock, Supplier<T> stuff) {

lock.lock();
try {
return stuff.get();
} finally {
lock.unlock();
}
}
}
Loading

0 comments on commit cc56a93

Please sign in to comment.