diff --git a/modules/dcache/src/main/java/org/dcache/pool/migration/Job.java b/modules/dcache/src/main/java/org/dcache/pool/migration/Job.java index b36c19348d4..57009fb168f 100644 --- a/modules/dcache/src/main/java/org/dcache/pool/migration/Job.java +++ b/modules/dcache/src/main/java/org/dcache/pool/migration/Job.java @@ -135,11 +135,79 @@ public Job(MigrationContext context, JobDefinition definition) { .getCellName(); } + /** + * Starts a job by asynchronously scanning the repository on the executor and queuing any + * matching entries that are discovered during initialization. + */ public void start() { + beginInitialization(); + _context.getExecutor().submit(new FireAndForgetTask(() -> { + try { + _context.getRepository().addListener(Job.this); + populate(); + + _lock.lock(); + try { + if (getState() == State.INITIALIZING) { + setState(State.RUNNING); + } + } finally { + _lock.unlock(); + } + } catch (InterruptedException e) { + LOGGER.error("Migration job was interrupted"); + } finally { + finishInitialization(); + } + })); + } + + /** + * Starts a job with a pre-selected set of entries, bypassing the generic repository scan. + * + *
Typical callers are hot-file replication, which already knows the exact {@link CacheEntry}
+ * that triggered replication. The method registers the job as a repository listener, enqueues
+ * the supplied entries, and then transitions the job to {@link State#RUNNING}. Because the
+ * work is performed synchronously on the calling thread, the caller blocks until all entries
+ * have been examined and queued.
+ *
+ * @param entries an {@link Iterable} of {@link CacheEntry} objects that should be migrated.
+ * @see #start() the asynchronous variant that performs a full-repository scan.
+ */
+ public void start(Iterable This method must be called exactly once when transitioning a job out of {@link State#NEW}.
+ */
+ private void beginInitialization() {
_lock.lock();
try {
checkState(_state == State.NEW);
- _state = State.INITIALIZING;
long refreshPeriod = _definition.refreshPeriod;
ScheduledExecutorService executor = _context.getExecutor();
@@ -150,37 +218,34 @@ public void start() {
_definition.poolList.refresh();
}), 0, refreshPeriod, TimeUnit.MILLISECONDS);
- executor.submit(new FireAndForgetTask(() -> {
- try {
- _context.getRepository().addListener(Job.this);
- populate();
+ _state = State.INITIALIZING;
+ } finally {
+ _lock.unlock();
+ }
+ }
- _lock.lock();
- try {
- if (getState() == State.INITIALIZING) {
- setState(State.RUNNING);
- }
- } finally {
- _lock.unlock();
- }
- } catch (InterruptedException e) {
- LOGGER.error("Migration job was interrupted");
- } finally {
- _lock.lock();
- try {
- switch (getState()) {
- case INITIALIZING:
- setState(State.FAILED);
- break;
- case CANCELLING:
- schedule();
- break;
- }
- } finally {
- _lock.unlock();
- }
- }
- }));
+ /**
+ * Completes job initialization bookkeeping.
+ *
+ * If initialization never progressed beyond {@link State#INITIALIZING}, the job is marked
+ * failed. If cancellation was requested while initializing, scheduling is resumed so the cancel
+ * path can complete.
+ */
+ private void finishInitialization() {
+ _lock.lock();
+ try {
+ switch (getState()) {
+ case INITIALIZING:
+ setState(State.FAILED);
+ break;
+ case CANCELLING:
+ schedule();
+ break;
+ default:
+ // No additional cleanup is needed once initialization has reached RUNNING,
+ // SLEEPING, PAUSED, SUSPENDED, STOPPING, CANCELLED, FINISHED, or FAILED.
+ break;
+ }
} finally {
_lock.unlock();
}
diff --git a/modules/dcache/src/main/java/org/dcache/pool/migration/MigrationModule.java b/modules/dcache/src/main/java/org/dcache/pool/migration/MigrationModule.java
index ef99476459b..74aadfa1102 100644
--- a/modules/dcache/src/main/java/org/dcache/pool/migration/MigrationModule.java
+++ b/modules/dcache/src/main/java/org/dcache/pool/migration/MigrationModule.java
@@ -1377,7 +1377,7 @@ public synchronized void reportFileRequest(PnfsId pnfsId, long numberOfRequests,
"About to start migration job with id {} for pnfsId {}. Job definition: {}",
jobId, pnfsId, job.getDefinition());
try {
- job.start();
+ job.start(Collections.singletonList(cacheEntry));
LOGGER.debug("Started migration job with id {} for pnfsId {}", jobId,
pnfsId);
} catch (Exception e) {
diff --git a/modules/dcache/src/test/java/org/dcache/pool/migration/MigrationModuleTest.java b/modules/dcache/src/test/java/org/dcache/pool/migration/MigrationModuleTest.java
index 5e9b7abb408..6f21bb67d80 100644
--- a/modules/dcache/src/test/java/org/dcache/pool/migration/MigrationModuleTest.java
+++ b/modules/dcache/src/test/java/org/dcache/pool/migration/MigrationModuleTest.java
@@ -29,6 +29,8 @@
import static org.mockito.ArgumentMatchers.isA;
import static org.mockito.Mockito.doNothing;
import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.never;
+import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when;
import com.google.common.util.concurrent.ListenableFuture;
@@ -137,6 +139,24 @@ public void testReportFileRequestAboveThreshold() throws Exception {
assertTrue(module.hasJob("hotfile-" + pnfsId));
}
+ @Test
+ public void testReportFileRequestDoesNotScanRepositoryToStartHotfileJob() throws Exception {
+ PnfsId pnfsId = new PnfsId("0000A1B2C3D4E5F7");
+ ProtocolInfo protocolInfo = mock(ProtocolInfo.class);
+ when(protocolInfo.getProtocol()).thenReturn("DCap");
+ when(protocolInfo.getMajorVersion()).thenReturn(3);
+ when(entry.getFileAttributes()).thenReturn(fileAttributes);
+ when(entry.getLastAccessTime()).thenReturn(0L);
+ when(entry.getPnfsId()).thenReturn(pnfsId);
+ when(repository.getEntry(pnfsId)).thenReturn(entry);
+
+ module.setThreshold(0L);
+ module.reportFileRequest(pnfsId, 1L, protocolInfo);
+
+ verify(repository, never()).iterator();
+ assertTrue(module.hasJob("hotfile-" + pnfsId));
+ }
+
@Test
public void testHotfileJobHousekeeping() throws Exception {
SerializablePoolMonitor poolMonitor = mock(SerializablePoolMonitor.class);