Skip to content

pool: make nearline request lifecycle transitions atomic #8263

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
kofemann wants to merge 1 commit into
base: 12.0
Choose a base branch
from
Open

pool: make nearline request lifecycle transitions atomic #8263

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
235 changes: 128 additions & 107 deletions modules/dcache/src/main/java/org/dcache/pool/nearline/NearlineStorageHandler.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -81,6 +81,7 @@
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicReference;
import java.util.function.Consumer;
import java.util.concurrent.locks.ReentrantLock;
import java.util.function.Predicate;
import java.util.stream.Collectors;
import javax.annotation.Nonnull;
Expand Down Expand Up @@ -507,9 +508,15 @@ protected enum State {QUEUED, ACTIVE, CANCELED, REMOVED}
*/
private final AtomicReference<QueueStat> queueStat;

AbstractRequest(NearlineStorage storage, @Nonnull AtomicReference<QueueStat> queueStat) {
/**
* Lock used to protect the queue statistics counters and state updates.
*/
private final ReentrantLock queueLock;

AbstractRequest(NearlineStorage storage, @Nonnull AtomicReference<QueueStat> queueStat, ReentrantLock queueLock) {
this.storage = storage;
this.queueStat = queueStat;
this.queueLock = queueLock;
}

// Implements NearlineRequest#setIncluded
Expand All @@ -522,30 +529,46 @@ public long getCreatedAt() {
}

public void onQueued() {
incQueued();
queueLock.lock();
try {
incQueued();
} finally {
queueLock.unlock();
}
}

protected synchronized <T> ListenableFuture<T> register(ListenableFuture<T> future) {
if (state.get() == State.CANCELED) {
/*
* do not interrupt thread as we don't know how
* it will react on it (BerkeleyDB will require
* to re-open db).
*/
future.cancel(false);
} else {
asyncTasks.add(future);
protected <T> ListenableFuture<T> register(ListenableFuture<T> future) {
queueLock.lock();
try {

if (state.get() == State.CANCELED) {
/*
* do not interrupt thread as we don't know how
* it will react on it (BerkeleyDB will require
* to re-open db).
*/
future.cancel(false);
} else {
asyncTasks.add(future);
}
return future;
} finally {
queueLock.unlock();
}
return future;
}

public ListenableFuture<Void> activate() {
if (!state.compareAndSet(State.QUEUED, State.ACTIVE)) {
return Futures.immediateFailedFuture(
new IllegalStateException("Request is no longer queued."));
queueLock.lock();
try {
if (!state.compareAndSet(State.QUEUED, State.ACTIVE)) {
return Futures.immediateFailedFuture(
new IllegalStateException("Request is no longer queued."));
}
activatedAt = System.currentTimeMillis();
incActiveFromQueued();
} finally {
queueLock.unlock();
}
activatedAt = System.currentTimeMillis();
incActiveFromQueued();
return Futures.immediateFuture(null);
}

Expand All @@ -560,15 +583,27 @@ public Iterable<CompletionHandler<Void, K>> callbacks() {
}

public void cancel() {
State oldState = state.getAndSet(State.CANCELED);
if (oldState != State.CANCELED) {
if (oldState == State.ACTIVE) {
incCancelFromActive();
} else {
incCancelFromQueued();

queueLock.lock();
try {
State oldState = state.get();

if (oldState == State.REMOVED) {
// removed from queue. the cancel path is already done, so nothing to do here.
return;
}
storage.cancel(uuid);
synchronized (this) {

/*
Request already canceled, but if attempted again, probably ignored by HSM, so remind it.
Assume that canceling is idempotent.
*/
if (oldState != State.CANCELED) {
state.set(State.CANCELED);
if (oldState == State.ACTIVE) {
incCancelFromActive();
} else {
incCancelFromQueued();
}
for (Future<?> task : asyncTasks) {
/*
* do not interrupt thread as we don't know how
Expand All @@ -578,13 +613,11 @@ public void cancel() {
task.cancel(false);
}
}
} else {
/*
Request already canceled, but if attempted again, probably ignored by HSM, so remind it.
Assume that canceling is idempotent.
*/
storage.cancel(uuid);
} finally {
queueLock.unlock();
}

storage.cancel(uuid);
}

public void failed(int rc, String msg) {
Expand Down Expand Up @@ -622,95 +655,78 @@ public int compareTo(AbstractRequest<K> o) {
}

public void removeFromQueue() {
State currentState = state.getAndSet(State.REMOVED);
switch (currentState) {
case QUEUED:
decQueued();
break;
case ACTIVE:
decActive();
break;
case CANCELED:
decCanceled();
break;
case REMOVED:
LOGGER.warn("Request {} was already removed from the queue.", this);
break;
default:
throw new RuntimeException();
queueLock.lock();
try {
State currentState = state.getAndSet(State.REMOVED);
switch (currentState) {
case QUEUED:
decQueued();
break;
case ACTIVE:
decActive();
break;
case CANCELED:
decCanceled();
break;
case REMOVED:
LOGGER.warn("Request {} was already removed from the queue.", this);
break;
default:
throw new RuntimeException();
}
} finally {
queueLock.unlock();
}
}

private void incQueued() {
boolean isSet;
do {
QueueStat current = queueStat.get();
QueueStat withThis = new QueueStat(current.queued() + 1, current.active(),
current.canceled());
isSet = queueStat.compareAndSet(current, withThis);
} while (!isSet);
QueueStat current = queueStat.get();
QueueStat withThis = new QueueStat(current.queued() + 1, current.active(),
current.canceled());
queueStat.set(withThis);
}

private void incActiveFromQueued() {
boolean isSet;
do {
QueueStat current = queueStat.get();
QueueStat withThis = new QueueStat(current.queued() - 1, current.active() + 1,
current.canceled());
isSet = queueStat.compareAndSet(current, withThis);
} while (!isSet);
QueueStat current = queueStat.get();
QueueStat withThis = new QueueStat(current.queued() - 1, current.active() + 1,
current.canceled());
queueStat.set(withThis);
}

private void incCancelFromActive() {
boolean isSet;
do {
QueueStat current = queueStat.get();
QueueStat withThis = new QueueStat(current.queued(), current.active() - 1,
current.canceled() + 1);
isSet = queueStat.compareAndSet(current, withThis);
} while (!isSet);
QueueStat current = queueStat.get();
QueueStat withThis = new QueueStat(current.queued(), current.active() - 1,
current.canceled() + 1);
queueStat.set(withThis);
}

private void incCancelFromQueued() {
boolean isSet;
do {
QueueStat current = queueStat.get();
QueueStat withThis = new QueueStat(current.queued() - 1, current.active(),
current.canceled() + 1);
isSet = queueStat.compareAndSet(current, withThis);
} while (!isSet);
QueueStat current = queueStat.get();
QueueStat withThis = new QueueStat(current.queued() - 1, current.active(),
current.canceled() + 1);
queueStat.set(withThis);
}

private void decQueued() {
boolean isSet;
do {
QueueStat current = queueStat.get();
QueueStat withThis = new QueueStat(current.queued() - 1, current.active(),
current.canceled());
isSet = queueStat.compareAndSet(current, withThis);
} while (!isSet);
QueueStat current = queueStat.get();
QueueStat withThis = new QueueStat(current.queued() - 1, current.active(),
current.canceled());
queueStat.set(withThis);
}

private void decActive() {
boolean isSet;
do {
QueueStat current = queueStat.get();
QueueStat withThis = new QueueStat(current.queued(), current.active() - 1,
current.canceled());
isSet = queueStat.compareAndSet(current, withThis);
} while (!isSet);
QueueStat current = queueStat.get();
QueueStat withThis = new QueueStat(current.queued(), current.active() - 1,
current.canceled());
queueStat.set(withThis);
}

private void decCanceled() {
boolean isSet;
do {
QueueStat current = queueStat.get();
QueueStat withThis = new QueueStat(current.queued(), current.active(),
current.canceled() - 1);
isSet = queueStat.compareAndSet(current, withThis);
} while (!isSet);
QueueStat current = queueStat.get();
QueueStat withThis = new QueueStat(current.queued(), current.active(),
current.canceled() - 1);
queueStat.set(withThis);
}

}

/**
Expand All @@ -733,6 +749,11 @@ private abstract static class AbstractRequestContainer<K, F, R extends AbstractR

private final ContainerState state = new ContainerState();

/**
* Lock used to protect the queue statistics counters.
*/
protected final ReentrantLock queueLock = new ReentrantLock();

/**
* Queue usage statistics.
*/
Expand Down Expand Up @@ -912,7 +933,7 @@ protected PnfsId extractKey(PnfsId id) {
@Override
protected FlushRequestImpl createRequest(NearlineStorage storage, PnfsId id)
throws CacheException, InterruptedException {
return new FlushRequestImpl(storage, queueStats, id);
return new FlushRequestImpl(storage, queueStats, queueLock, id);
}

@Override
Expand All @@ -933,7 +954,7 @@ protected PnfsId extractKey(FileAttributes file) {
protected StageRequestImpl createRequest(NearlineStorage storage,
FileAttributes file)
throws CacheException {
return new StageRequestImpl(storage, queueStats, file);
return new StageRequestImpl(storage, queueStats, queueLock, file);
}

@Override
Expand All @@ -952,7 +973,7 @@ protected URI extractKey(URI uri) {

@Override
protected RemoveRequestImpl createRequest(NearlineStorage storage, URI uri) {
return new RemoveRequestImpl(storage, queueStats, uri);
return new RemoveRequestImpl(storage, queueStats, queueLock, uri);
}

@Override
Expand All @@ -966,9 +987,9 @@ private class FlushRequestImpl extends AbstractRequest<PnfsId> implements FlushR
private final ReplicaDescriptor descriptor;
private final StorageInfoMessage infoMsg;

public FlushRequestImpl(NearlineStorage nearlineStorage, AtomicReference<QueueStat> stats,
public FlushRequestImpl(NearlineStorage nearlineStorage, AtomicReference<QueueStat> stats, ReentrantLock queueLock,
PnfsId pnfsId) throws CacheException, InterruptedException {
super(nearlineStorage, stats);
super(nearlineStorage, stats, queueLock);
descriptor = repository.openEntry(pnfsId, EnumSet.of(OpenFlags.NOATIME));
infoMsg = new StorageInfoMessage(cellAddress, pnfsId, false);
infoMsg.setStorageInfo(descriptor.getFileAttributes().getStorageInfo());
Expand Down Expand Up @@ -1270,9 +1291,9 @@ private class StageRequestImpl extends AbstractRequest<PnfsId> implements StageR
private final ModifiableReplicaDescriptor descriptor;
private ListenableFuture<Void> allocationFuture;

public StageRequestImpl(NearlineStorage storage, AtomicReference<QueueStat> stats,
public StageRequestImpl(NearlineStorage storage, AtomicReference<QueueStat> stats, ReentrantLock queueLock,
FileAttributes fileAttributes) throws CacheException {
super(storage, stats);
super(storage, stats, queueLock);
PnfsId pnfsId = fileAttributes.getPnfsId();
infoMsg = new StorageInfoMessage(cellAddress, pnfsId, true);
infoMsg.setStorageInfo(fileAttributes.getStorageInfo());
Expand Down Expand Up @@ -1448,8 +1469,8 @@ private class RemoveRequestImpl extends AbstractRequest<URI> implements RemoveRe

private final URI uri;

RemoveRequestImpl(NearlineStorage storage, AtomicReference<QueueStat> stats, URI uri) {
super(storage, stats);
RemoveRequestImpl(NearlineStorage storage, AtomicReference<QueueStat> stats, ReentrantLock queueLock, URI uri) {
super(storage, stats, queueLock);
this.uri = uri;
LOGGER.debug("Remove request created for {}.", uri);
}
Expand Down
Loading
Loading