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 entries) { + beginInitialization(); + + _lock.lock(); + try { + _context.getRepository().addListener(this); + + for (CacheEntry entry : entries) { + if (accept(entry)) { + add(entry); + } + } + + if (getState() == State.INITIALIZING) { + setState(State.RUNNING); + } + } catch (Throwable e) { + LOGGER.error("Migration job initialization failed", e); + throw e; + } finally { + _lock.unlock(); + finishInitialization(); + } + } + + /** + * Marks the job as initializing and schedules the periodic source and target pool refresh. + * + *

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);