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;