From 6f7125f3340e9506f7494d99ff47c776cad26bc1 Mon Sep 17 00:00:00 2001 From: Tigran Mkrtchyan Date: Sun, 18 Oct 2026 18:36:43 +0200 Subject: [PATCH] pool: make nearline request lifecycle transitions atomic Motivation: The commit 4759b3fe2ee5ca4 ("pool: don't scan flush/restore/remove queue to get stats") introduced lock-free nearline subsistem queue statistics to improve the queue query throughout. However, the change missed the case where queue state and stats must be updated atomically. Thus, tape pools under heavy load might run into double accounting when many requests are canceled of removed files have been in the flush queue. Modification: Introduce per-queue lock that takes care that queue statistics and request state update atomically. The getXxxQueueSize has no locks as stats are used in AtomicReference Result: Queue statistics can no longer go negative and pools no longer gets stuck. Fixes: #7511 Acked-by: Lennart Sack Target: master, 12.0, 11.2 Require-book: no Require-notes: yes (cherry picked from commit 1a362ebe486aadf3aaa936c01903ee99668477d3) Signed-off-by: Tigran Mkrtchyan --- .../pool/nearline/NearlineStorageHandler.java | 235 ++++++++++-------- .../nearline/NearlineStorageHandlerTest.java | 72 ++++++ 2 files changed, 200 insertions(+), 107 deletions(-) diff --git a/modules/dcache/src/main/java/org/dcache/pool/nearline/NearlineStorageHandler.java b/modules/dcache/src/main/java/org/dcache/pool/nearline/NearlineStorageHandler.java index 1db15e13fbe..296b85407a8 100644 --- a/modules/dcache/src/main/java/org/dcache/pool/nearline/NearlineStorageHandler.java +++ b/modules/dcache/src/main/java/org/dcache/pool/nearline/NearlineStorageHandler.java @@ -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; @@ -507,9 +508,15 @@ protected enum State {QUEUED, ACTIVE, CANCELED, REMOVED} */ private final AtomicReference queueStat; - AbstractRequest(NearlineStorage storage, @Nonnull AtomicReference queueStat) { + /** + * Lock used to protect the queue statistics counters and state updates. + */ + private final ReentrantLock queueLock; + + AbstractRequest(NearlineStorage storage, @Nonnull AtomicReference queueStat, ReentrantLock queueLock) { this.storage = storage; this.queueStat = queueStat; + this.queueLock = queueLock; } // Implements NearlineRequest#setIncluded @@ -522,30 +529,46 @@ public long getCreatedAt() { } public void onQueued() { - incQueued(); + queueLock.lock(); + try { + incQueued(); + } finally { + queueLock.unlock(); + } } - protected synchronized ListenableFuture register(ListenableFuture 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 ListenableFuture register(ListenableFuture 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 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); } @@ -560,15 +583,27 @@ public Iterable> 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 @@ -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) { @@ -622,95 +655,78 @@ public int compareTo(AbstractRequest 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); } - } /** @@ -733,6 +749,11 @@ private abstract static class AbstractRequestContainer implements FlushR private final ReplicaDescriptor descriptor; private final StorageInfoMessage infoMsg; - public FlushRequestImpl(NearlineStorage nearlineStorage, AtomicReference stats, + public FlushRequestImpl(NearlineStorage nearlineStorage, AtomicReference 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()); @@ -1270,9 +1291,9 @@ private class StageRequestImpl extends AbstractRequest implements StageR private final ModifiableReplicaDescriptor descriptor; private ListenableFuture allocationFuture; - public StageRequestImpl(NearlineStorage storage, AtomicReference stats, + public StageRequestImpl(NearlineStorage storage, AtomicReference 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()); @@ -1448,8 +1469,8 @@ private class RemoveRequestImpl extends AbstractRequest implements RemoveRe private final URI uri; - RemoveRequestImpl(NearlineStorage storage, AtomicReference stats, URI uri) { - super(storage, stats); + RemoveRequestImpl(NearlineStorage storage, AtomicReference stats, ReentrantLock queueLock, URI uri) { + super(storage, stats, queueLock); this.uri = uri; LOGGER.debug("Remove request created for {}.", uri); } diff --git a/modules/dcache/src/test/java/org/dcache/pool/nearline/NearlineStorageHandlerTest.java b/modules/dcache/src/test/java/org/dcache/pool/nearline/NearlineStorageHandlerTest.java index afb064713f6..075b6308e40 100644 --- a/modules/dcache/src/test/java/org/dcache/pool/nearline/NearlineStorageHandlerTest.java +++ b/modules/dcache/src/test/java/org/dcache/pool/nearline/NearlineStorageHandlerTest.java @@ -6,6 +6,7 @@ import static org.mockito.ArgumentMatchers.eq; import static org.mockito.Mockito.doThrow; import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.never; import static org.mockito.Mockito.times; import static org.mockito.Mockito.verify; import static org.mockito.Mockito.when; @@ -21,9 +22,13 @@ import diskCacheV111.vehicles.GenericStorageInfo; import java.net.URI; import java.nio.channels.CompletionHandler; +import java.util.ArrayList; import java.util.Collection; import java.util.List; import java.util.Set; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.Future; import java.util.concurrent.TimeUnit; import java.util.stream.Collectors; import java.util.stream.StreamSupport; @@ -33,9 +38,12 @@ import org.dcache.pool.classic.ChecksumModule; import org.dcache.pool.nearline.spi.NearlineRequest; import org.dcache.pool.nearline.spi.NearlineStorage; +import org.dcache.pool.repository.CacheEntry; import org.dcache.pool.repository.FileStore; import org.dcache.pool.repository.ModifiableReplicaDescriptor; +import org.dcache.pool.repository.ReplicaState; import org.dcache.pool.repository.Repository; +import org.dcache.pool.repository.StateChangeEvent; import org.dcache.util.Checksum; import org.dcache.vehicles.FileAttributes; import org.junit.Before; @@ -385,6 +393,70 @@ public void testCompleteTwiceActiveRemove() throws CacheException { verify(hsmRemoveRequestCallack, times(1)).completed(any(), any()); } + + // Regression test for #7511: racing lifecycle transitions (activate, cancel, failure) + @Test + public void testConcurrentCancelActivateAndFailure() throws Exception { + int nFiles = 200; + List ids = new ArrayList<>(); + for (int i = 0; i < nFiles; i++) { + var fa = FileAttributes.of() + .fileType(FileType.REGULAR) + .accessLatency(AccessLatency.NEARLINE) + .retentionPolicy(RetentionPolicy.CUSTODIAL) + .pnfsId(InodeId.newID(0)) + .storageClass("a:b") + .hsm("foo") + .storageInfo(GenericStorageInfo.valueOf("a:b@foo", "*")) + .size(34567) + .build(); + var descriptor = mock(ModifiableReplicaDescriptor.class); + when(descriptor.getFileAttributes()).thenReturn(fa); + when(repository.openEntry(eq(fa.getPnfsId()), any())).thenReturn(descriptor); + ids.add(fa.getPnfsId()); + } + + nsh.flush("foo", ids, hsmMigrationRequestCallack); + + var ac = ArgumentCaptor.forClass(Iterable.class); + verify(nearlineStorage).flush(ac.capture()); + List requests = (List) StreamSupport.stream( + ac.getValue().spliterator(), false).collect( + Collectors.toList()); + + ExecutorService pool = Executors.newFixedThreadPool(8); + try { + List> futures = new ArrayList<>(); + for (int i = 0; i < requests.size(); i++) { + NearlineRequest request = requests.get(i); + PnfsId pnfsId = ids.get(i); + futures.add(pool.submit(() -> request.activate())); + futures.add(pool.submit(() -> request.activate())); + futures.add(pool.submit(() -> nsh.stateChanged(removedEvent(pnfsId)))); + futures.add(pool.submit(() -> nsh.stateChanged(removedEvent(pnfsId)))); + futures.add(pool.submit(() -> request.failed(1, "forcefully failed"))); + } + for (Future future : futures) { + /* propagates any uncaught invariant violation, e.g. a negative counter */ + future.get(30, TimeUnit.SECONDS); + } + } finally { + pool.shutdownNow(); + } + + assertThat(nsh.getStoreQueueSize(), is(0)); + assertThat(nsh.getActiveStoreJobs(), is(0)); + verify(hsmMigrationRequestCallack, times(nFiles)).failed(any(), any()); + verify(hsmMigrationRequestCallack, never()).completed(any(), any()); + } + + private StateChangeEvent removedEvent(PnfsId pnfsId) { + CacheEntry entry = mock(CacheEntry.class); + when(entry.getPnfsId()).thenReturn(pnfsId); + return new StateChangeEvent("file deleted", entry, entry, + ReplicaState.CACHED, ReplicaState.REMOVED); + } + private class FileBuilder { FileAttributes.Builder faBuilder;