From 25c01d039ca9bf6c1740f91527518e68c43173ad Mon Sep 17 00:00:00 2001 From: James Rosewell Date: Thu, 12 Nov 2026 20:14:15 +0000 Subject: [PATCH 1/6] Test that the file system watcher runs on its own setting The watcher is created with automatic updates off and the watcher on, and not with the watcher off. Tests that set automatic updates only to get a watcher no longer do. --- .../DataUpdateServiceLoggingTests.java | 3 +- .../services/DataUpdateServiceTests.java | 19 ++-- .../DataUpdateServiceWatchedFileTests.java | 86 ++++++++++++++++++- 3 files changed, 94 insertions(+), 14 deletions(-) diff --git a/pipeline/pipeline.engines/src/test/java/fiftyone/pipeline/engines/services/DataUpdateServiceLoggingTests.java b/pipeline/pipeline.engines/src/test/java/fiftyone/pipeline/engines/services/DataUpdateServiceLoggingTests.java index 15a2f9a1..dcd972ae 100644 --- a/pipeline/pipeline.engines/src/test/java/fiftyone/pipeline/engines/services/DataUpdateServiceLoggingTests.java +++ b/pipeline/pipeline.engines/src/test/java/fiftyone/pipeline/engines/services/DataUpdateServiceLoggingTests.java @@ -279,7 +279,8 @@ public void UpdateOnStartup_AutoUpdateOff_StartsNoTimer() throws Exception { @Test public void FileSystemWatcher_LogsCreatingFileSystemWatcher() { - AspectEngineDataFileDefault file = newDataFile(false, true, false); + // The watcher needs no automatic updates. + AspectEngineDataFileDefault file = newDataFile(false, false, false); file.getConfiguration().setFileSystemWatcherEnabled(true); dataUpdate.registerDataFile(file); diff --git a/pipeline/pipeline.engines/src/test/java/fiftyone/pipeline/engines/services/DataUpdateServiceTests.java b/pipeline/pipeline.engines/src/test/java/fiftyone/pipeline/engines/services/DataUpdateServiceTests.java index a1650c1d..b10b06a8 100644 --- a/pipeline/pipeline.engines/src/test/java/fiftyone/pipeline/engines/services/DataUpdateServiceTests.java +++ b/pipeline/pipeline.engines/src/test/java/fiftyone/pipeline/engines/services/DataUpdateServiceTests.java @@ -374,12 +374,9 @@ public void DataUpdateService_Register_FileSystemWatcher() throws IOException { dataUpdate.registerDataFile(file); // Assert - // The automatic update switch governs the file system watcher. - if (autoUpdateEnabled) { - assertNotNull(config.getWatchKey()); - } else { - assertNull(config.getWatchKey()); - } + // The watcher runs on its own setting, whether or not automatic + // updates are on. + assertNotNull(config.getWatchKey()); } finally { // Make sure we tidy up the temp file. File tempFile = new File(tempData); @@ -398,7 +395,8 @@ public void DataUpdateService_UpdateFromWatcher() throws IOException, Interrupte // Arrange OnPremiseAspectEngine engine = mock(OnPremiseAspectEngine.class); DataFileConfigurationDefault config = new DataFileConfigurationDefault(); - config.setAutomaticUpdatesEnabled(true); + // The watcher needs no automatic updates. + config.setAutomaticUpdatesEnabled(false); config.setFileSystemWatcherEnabled(true); config.setPollingIntervalSeconds(1); @@ -424,8 +422,7 @@ public Object answer(InvocationOnMock invocationOnMock) throws Throwable { }); // Configure a flag to be set when the watcher has applied the - // update. The polling timer's own check also completes, with no - // update found, and must not release the flag. + // update. final Semaphore completeFlag = new Semaphore(1); completeFlag.acquire(); dataUpdate.onUpdateComplete(new OnUpdateComplete() { @@ -1028,8 +1025,8 @@ public void DataUpdateService_CheckForUpdate_FileUpdated() { /** * A programmatic update compares the data file on disk whether or not - * Automatic Update from File is enabled, and with automatic update off - * no file system watcher exists whatever the watcher flag says. + * Automatic Update from File is enabled, here with automatic update off + * and the watcher flag at its default. */ @Test public void DataUpdateService_CheckForUpdate_FileUpdated_AutoUpdateOffWatcherFlagDefault() { diff --git a/pipeline/pipeline.engines/src/test/java/fiftyone/pipeline/engines/services/DataUpdateServiceWatchedFileTests.java b/pipeline/pipeline.engines/src/test/java/fiftyone/pipeline/engines/services/DataUpdateServiceWatchedFileTests.java index b57a16ca..74b97787 100644 --- a/pipeline/pipeline.engines/src/test/java/fiftyone/pipeline/engines/services/DataUpdateServiceWatchedFileTests.java +++ b/pipeline/pipeline.engines/src/test/java/fiftyone/pipeline/engines/services/DataUpdateServiceWatchedFileTests.java @@ -151,6 +151,79 @@ public void WatchedFileRefresh_PollingCarriesOn() throws Exception { } } + /** + * The file system watcher runs on its own setting. Watching a file + * downloads nothing, so the watcher does not wait for automatic updates + * to be on, and with its own setting off a replaced file is not loaded. + * Polling runs only with automatic updates on. + */ + @Test + public void WatcherRunsOnItsOwnSetting() throws Exception { + String[] names = { + "watcher on, automatic updates on", + "watcher on, automatic updates off", + "watcher off, automatic updates on"}; + boolean[] automatic = {true, false, true}; + boolean[] watching = {true, true, false}; + for (int i = 0; i < names.length; i++) { + checkWatcherSetting(names[i], automatic[i], watching[i]); + } + } + + private static void checkWatcherSetting( + String name, + boolean automatic, + boolean watching) throws Exception { + final List timers = new ArrayList<>(); + final AtomicReference watcher = new AtomicReference<>(); + FileWrapperFactory fileWrapperFactory = mock(FileWrapperFactory.class); + when(fileWrapperFactory.build(anyString())).then(invocation -> + new FileWrapperDefault((String) invocation.getArgument(0))); + when(fileWrapperFactory.getLastModified(anyString())).then( + invocation -> lastModified(invocation.getArgument(0))); + DataUpdateService dataUpdate = new DataUpdateServiceDefault( + null, + mock(HttpClient.class), + fileWrapperFactory, + futureFactory(timers, watcher)); + + OnPremiseAspectEngine engine = engine(); + Path directory = Files.createTempDirectory("watched"); + Path dataFile = directory.resolve("data.dat"); + Files.write(dataFile, "Old".getBytes()); + AspectEngineDataFileDefault file = + watchedFile(engine, dataFile, automatic, watching); + + try { + dataUpdate.registerDataFile(file); + assertEquals(name + ", a file system watcher was created", + watching, file.getConfiguration().getWatchKey() != null); + assertEquals(name + ", the watcher's task was scheduled", + watching, watcher.get() != null); + assertEquals(name + ", polling timers set", + automatic ? 1 : 0, timers.size()); + + // macOS watchers are too slow to wait for. The test holds the + // polling timers, so only a watcher can load the replaced file. + if (isMac() == false) { + Files.write(dataFile, "New".getBytes()); + Runnable task = watcher.get(); + long end = System.currentTimeMillis() + WAIT_FOR_EVENT_MILLIS; + while (task != null && refreshCount(engine) == 0 && + System.currentTimeMillis() < end) { + task.run(); + Thread.sleep(RETRY_MILLIS); + } + assertEquals(name + ", the replaced file was loaded", + watching ? 1 : 0, refreshCount(engine)); + } + } + finally { + cleanUp(file, directory, dataFile); + } + } + /** * A watched file that cannot be opened is tried, with a pause between * tries, until the wait ends. The update then reports @@ -361,9 +434,18 @@ private static AspectEngineDataFileDefault watchedFile( OnPremiseAspectEngine engine, Path dataFile) { + return watchedFile(engine, dataFile, true, true); + } + + private static AspectEngineDataFileDefault watchedFile( + OnPremiseAspectEngine engine, + Path dataFile, + boolean automatic, + boolean watching) { DataFileConfigurationDefault config = new DataFileConfigurationDefault(); - config.setAutomaticUpdatesEnabled(true); - config.setFileSystemWatcherEnabled(true); + config.setAutomaticUpdatesEnabled(automatic); + config.setFileSystemWatcherEnabled(watching); config.setPollingIntervalSeconds(1); config.setDataFilePath(dataFile.toString()); AspectEngineDataFileDefault file = new AspectEngineDataFileDefault(); From 7f96e53b7193bb2a403c6819b2ad1fdf97d9d3e5 Mon Sep 17 00:00:00 2001 From: James Rosewell Date: Thu, 12 Nov 2026 20:32:33 +0000 Subject: [PATCH 2/6] Run the file system watcher on its own setting The data update service creates the watcher when the watcher is on, whether or not automatic updates are on. Watching a file downloads nothing, so it needs neither automatic updates nor a license key. Polling, the update on start-up and the license key check are unchanged. --- .../engines/services/DataUpdateServiceDefault.java | 9 ++++----- 1 file changed, 4 insertions(+), 5 deletions(-) diff --git a/pipeline/pipeline.engines/src/main/java/fiftyone/pipeline/engines/services/DataUpdateServiceDefault.java b/pipeline/pipeline.engines/src/main/java/fiftyone/pipeline/engines/services/DataUpdateServiceDefault.java index b21f1a31..95bafae0 100644 --- a/pipeline/pipeline.engines/src/main/java/fiftyone/pipeline/engines/services/DataUpdateServiceDefault.java +++ b/pipeline/pipeline.engines/src/main/java/fiftyone/pipeline/engines/services/DataUpdateServiceDefault.java @@ -548,11 +548,10 @@ public void run() { } } - // If file system watcher is enabled then set it up. The - // automatic update switch governs the watcher as well as the - // polling timer. - if (aspectDataFile.getAutomaticUpdatesEnabled() && - aspectDataFile.getConfiguration().getFileSystemWatcherEnabled() && + // If file system watcher is enabled then set it up. The watcher + // runs on its own setting. Watching a file downloads nothing, so + // it needs neither automatic updates nor a license key. + if (aspectDataFile.getConfiguration().getFileSystemWatcherEnabled() && aspectDataFile.getConfiguration().getWatchKey() == null && aspectDataFile.getDataFilePath() != null && aspectDataFile.getDataFilePath().isEmpty() == false) { From 8fdd85d37c7139ba368d7fcdbe4de85e5b2ceaef Mon Sep 17 00:00:00 2001 From: James Rosewell Date: Fri, 13 Nov 2026 00:23:19 +0000 Subject: [PATCH 3/6] Test that the watcher alone registers a data file and closes its watch service The file system watcher runs on its own setting and needs no license key. An engine whose update source needs a key it does not have is built with automatic updates and the update on start-up off, and its data file should still be registered to be watched. The builder registers a data file only for the two options that download, so the new builder test fails until it also registers one for the watcher. Each watch service runs a thread of its own. Unregistering a watched data file should close its watch service, and a download should resume the watch on the same watch service rather than leave another one running. The watcher's task repeats, and a repeating task that fails is never run again, so it should do nothing while the watch is paused for a download. --- .../OnPremiseAspectEngineBuilderBaseTest.java | 96 +++++++++ .../DataUpdateServiceWatchedFileTests.java | 199 ++++++++++++++++++ 2 files changed, 295 insertions(+) diff --git a/pipeline/pipeline.engines/src/test/java/fiftyone/pipeline/engines/flowelements/OnPremiseAspectEngineBuilderBaseTest.java b/pipeline/pipeline.engines/src/test/java/fiftyone/pipeline/engines/flowelements/OnPremiseAspectEngineBuilderBaseTest.java index 196e6b9d..efbd8ade 100644 --- a/pipeline/pipeline.engines/src/test/java/fiftyone/pipeline/engines/flowelements/OnPremiseAspectEngineBuilderBaseTest.java +++ b/pipeline/pipeline.engines/src/test/java/fiftyone/pipeline/engines/flowelements/OnPremiseAspectEngineBuilderBaseTest.java @@ -26,7 +26,11 @@ import fiftyone.pipeline.engines.Constants; import fiftyone.pipeline.engines.configuration.DataFileConfiguration; import fiftyone.pipeline.engines.configuration.DataFileConfigurationDefault; +import fiftyone.pipeline.engines.data.AspectEngineDataFile; import fiftyone.pipeline.engines.services.DataUpdateService; +import fiftyone.pipeline.engines.services.DataUpdateServiceDefault; +import fiftyone.pipeline.engines.services.HttpClient; +import fiftyone.pipeline.engines.services.update.FutureFactory; import org.junit.After; import org.junit.Assume; import org.junit.Before; @@ -42,14 +46,18 @@ import java.util.Collections; import java.util.List; import java.util.Set; +import java.util.concurrent.ScheduledFuture; +import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertFalse; import static org.junit.Assert.assertThrows; import static org.junit.Assert.assertTrue; import static org.mockito.ArgumentMatchers.any; +import static org.mockito.ArgumentMatchers.anyLong; import static org.mockito.Mockito.mock; import static org.mockito.Mockito.never; import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.when; @SuppressWarnings("rawtypes") public class OnPremiseAspectEngineBuilderBaseTest { @@ -177,6 +185,94 @@ protected boolean isLicenseKeyRequired( verify(updateService, never()).registerDataFile(any()); } + /** + * The file system watcher runs on its own setting and needs no license + * key. An engine whose update source needs a key it does not have is + * built with automatic updates and the update on start-up off, and its + * data file is still registered to be watched. With the watcher off as + * well, nothing is registered. + */ + @Test + public void WatcherAlone_RegistersTheDataFile() throws Exception { + String[] names = { + "watcher on, automatic updates off, no license key", + "watcher off, automatic updates off, no license key"}; + boolean[] watching = {true, false}; + for (int i = 0; i < names.length; i++) { + checkWatcherAlone(names[i], watching[i]); + } + } + + private static void checkWatcherAlone(String name, boolean watching) + throws Exception { + FutureFactory futureFactory = mock(FutureFactory.class); + when(futureFactory.scheduleRepeating(any(Runnable.class), anyLong())) + .then(invocation -> mock(ScheduledFuture.class)); + DataUpdateService updateService = new DataUpdateServiceDefault( + null, mock(HttpClient.class), null, futureFactory); + OnPremiseAspectEngineBuilderBase builder = + keyRequiredBuilder(updateService); + + Path directory = Files.createTempDirectory("watched"); + Path dataFile = directory.resolve("data.dat"); + Files.write(dataFile, "Data".getBytes()); + DataFileConfigurationDefault config = new DataFileConfigurationDefault(); + config.setIdentifier("Watched"); + config.setDataFilePath(dataFile.toString()); + config.setAutomaticUpdatesEnabled(false); + config.setUpdateOnStartup(false); + config.setFileSystemWatcherEnabled(watching); + builder.addDataFile(config); + + try { + builder.buildEngine(); + AspectEngineDataFile file = + (AspectEngineDataFile) builder.dataFiles.get(0); + assertEquals(name + ", the data file is registered", + watching, file.getIsRegistered()); + assertEquals(name + ", a file system watcher was created", + watching, config.getWatchKey() != null); + } + finally { + for (Object file : builder.dataFiles) { + updateService.unregisterDataFile((AspectEngineDataFile) file); + } + if (config.getWatchKey() != null) { + config.getWatchKey().cancel(); + } + directory.toFile().deleteOnExit(); + if (dataFile.toFile().delete() == false) { + dataFile.toFile().deleteOnExit(); + } + directory.toFile().delete(); + } + } + + /** + * A builder whose update source serves nothing without a license key. + */ + private static OnPremiseAspectEngineBuilderBase keyRequiredBuilder( + DataUpdateService updateService) { + return new OnPremiseAspectEngineBuilderBase(null, updateService) { + @Override + public OnPremiseAspectEngineBuilderBase setPerformanceProfile( + Constants.PerformanceProfiles profile) { + return this; + } + + @Override + protected AspectEngine newEngine(List properties) { + return null; + } + + @Override + protected boolean isLicenseKeyRequired( + DataFileConfiguration dataFileConfig) { + return true; + } + }; + } + private static DataFileConfiguration autoUpdateConfiguration( String identifier, String licenseKey) { diff --git a/pipeline/pipeline.engines/src/test/java/fiftyone/pipeline/engines/services/DataUpdateServiceWatchedFileTests.java b/pipeline/pipeline.engines/src/test/java/fiftyone/pipeline/engines/services/DataUpdateServiceWatchedFileTests.java index 74b97787..09eb4245 100644 --- a/pipeline/pipeline.engines/src/test/java/fiftyone/pipeline/engines/services/DataUpdateServiceWatchedFileTests.java +++ b/pipeline/pipeline.engines/src/test/java/fiftyone/pipeline/engines/services/DataUpdateServiceWatchedFileTests.java @@ -37,11 +37,16 @@ import org.mockito.Answers; import org.slf4j.Logger; +import java.io.ByteArrayInputStream; import java.io.File; import java.io.IOException; +import java.net.HttpURLConnection; +import java.net.URL; import java.nio.file.Files; import java.nio.file.Path; +import java.nio.file.WatchKey; import java.util.ArrayList; +import java.util.Arrays; import java.util.List; import java.util.concurrent.CountDownLatch; import java.util.concurrent.ScheduledFuture; @@ -60,6 +65,7 @@ import static org.mockito.ArgumentMatchers.anyString; import static org.mockito.Mockito.mock; import static org.mockito.Mockito.mockingDetails; +import static org.mockito.Mockito.verify; import static org.mockito.Mockito.when; import static org.mockito.Mockito.withSettings; @@ -100,6 +106,10 @@ public class DataUpdateServiceWatchedFileTests { "File watchers are not well implemented in OS X, " + "so don't run this unit test as it is unlikely to pass."; + private static final String POLLER_MESSAGE = + "The watch service on OS X polls from a thread this test cannot " + + "pick out."; + @Test public void WatchedFileRefresh_PollingCarriesOn() throws Exception { assumeFalse(MAC_MESSAGE, isMac()); @@ -224,6 +234,145 @@ private static void checkWatcherSetting( } } + /** + * Each watch service runs a thread of its own. Unregistering a watched + * data file, as closing its engine does, closes the watch service so its + * thread ends. + */ + @Test + public void Unregistering_ClosesTheWatchService() throws Exception { + assumeFalse(POLLER_MESSAGE, isMac()); + DataUpdateService dataUpdate = new DataUpdateServiceDefault( + null, + mock(HttpClient.class), + fileWrapperFactory(), + futureFactory(new ArrayList<>(), new AtomicReference<>())); + Path directory = Files.createTempDirectory("watched"); + Path dataFile = directory.resolve("data.dat"); + Files.write(dataFile, "Old".getBytes()); + AspectEngineDataFileDefault file = + watchedFile(engine(), dataFile, false, true); + + try { + long before = watchServiceThreads(); + dataUpdate.registerDataFile(file); + assertTrue("Registering starts a watch service thread", + waitForWatchServiceThreads(before + 1)); + + dataUpdate.unregisterDataFile(file); + assertTrue("Unregistering ends the watch service thread", + waitForWatchServiceThreads(before)); + } + finally { + cleanUp(file, directory, dataFile); + } + } + + /** + * A downloaded data file is copied into place with the watch paused. + * The watch then resumes on the data file's own watch service, so no + * other watch service is left running, and unregistering ends the one + * there is. + */ + @Test + public void DownloadedFile_WatchResumesOnTheSameWatchService() + throws Exception { + assumeFalse(POLLER_MESSAGE, isMac()); + final List timers = new ArrayList<>(); + HttpURLConnection connection = mock(HttpURLConnection.class); + when(connection.getResponseCode()) + .thenReturn(HttpURLConnection.HTTP_OK); + when(connection.getInputStream()).thenReturn( + new ByteArrayInputStream("New".getBytes())); + HttpClient httpClient = mock(HttpClient.class); + when(httpClient.connect(any(URL.class))).thenReturn(connection); + DataUpdateService dataUpdate = new DataUpdateServiceDefault( + null, + httpClient, + fileWrapperFactory(), + timerFactory(timers)); + + OnPremiseAspectEngine engine = engine(); + Path directory = Files.createTempDirectory("watched"); + Path dataFile = directory.resolve("data.dat"); + Files.write(dataFile, "Old".getBytes()); + Path tempDirectory = Files.createTempDirectory("download"); + AspectEngineDataFileDefault file = + watchedFile(engine, dataFile, true, true); + file.setTempDataDirPath(tempDirectory.toString()); + file.getConfiguration().setDataUpdateUrl("http://www.test.com"); + file.getConfiguration().setDecompressContent(false); + file.getConfiguration().setVerifyMd5(false); + file.getConfiguration().setVerifyModifiedSince(false); + + try { + long before = watchServiceThreads(); + dataUpdate.registerDataFile(file); + assertTrue("Registering starts a watch service thread", + waitForWatchServiceThreads(before + 1)); + + // The polling timer downloads the new file. + timers.get(0).run(); + assertEquals("The downloaded file refreshes the engine", + 1, refreshCount(engine)); + assertNotNull("The watch resumes after the download", + file.getConfiguration().getWatchKey()); + assertTrue("The watch resumes on the same watch service", + waitForWatchServiceThreads(before + 1)); + + dataUpdate.unregisterDataFile(file); + assertTrue("Unregistering ends the watch service thread", + waitForWatchServiceThreads(before)); + } + finally { + cleanUp(file, directory, dataFile); + tempDirectory.toFile().delete(); + } + } + + /** + * The watcher's task repeats, and a repeating task that fails is never + * run again. So while the watch is paused, as it is when a downloaded + * file is copied into place, the task does nothing and carries on. + * Unregistering the data file while the watch is paused still stops the + * task. + */ + @Test + public void WatchPaused_WatcherTaskCarriesOn() throws Exception { + final AtomicReference watcher = new AtomicReference<>(); + DataUpdateService dataUpdate = new DataUpdateServiceDefault( + null, + mock(HttpClient.class), + fileWrapperFactory(), + futureFactory(new ArrayList<>(), watcher)); + Path directory = Files.createTempDirectory("watched"); + Path dataFile = directory.resolve("data.dat"); + Files.write(dataFile, "Old".getBytes()); + AspectEngineDataFileDefault file = + watchedFile(engine(), dataFile, false, true); + + try { + dataUpdate.registerDataFile(file); + WatchKey watchKey = file.getConfiguration().getWatchKey(); + assertNotNull("A file system watcher was created", watchKey); + ScheduledFuture task = file.getPollFuture(); + + file.getConfiguration().setWatchKey(null); + watcher.get().run(); + file.getConfiguration().setWatchKey(watchKey); + watcher.get().run(); + + file.getConfiguration().setWatchKey(null); + dataUpdate.unregisterDataFile(file); + verify(task).cancel(true); + watchKey.cancel(); + } + finally { + cleanUp(file, directory, dataFile); + } + } + /** * A watched file that cannot be opened is tried, with a pause between * tries, until the wait ends. The update then reports @@ -424,6 +573,56 @@ private static FutureFactory futureFactory( return futureFactory; } + /** + * Keeps each one-off timer's task for the test to run itself. + */ + private static FutureFactory timerFactory(final List timers) { + FutureFactory futureFactory = mock(FutureFactory.class); + when(futureFactory.schedule(any(Runnable.class), anyLong())) + .then(invocation -> { + timers.add((Runnable) invocation.getArgument(0)); + return mock(ScheduledFuture.class); + }); + when(futureFactory.scheduleRepeating(any(Runnable.class), anyLong())) + .then(invocation -> mock(ScheduledFuture.class)); + return futureFactory; + } + + private static FileWrapperFactory fileWrapperFactory() { + FileWrapperFactory fileWrapperFactory = mock(FileWrapperFactory.class); + when(fileWrapperFactory.build(anyString())).then(invocation -> + new FileWrapperDefault((String) invocation.getArgument(0))); + when(fileWrapperFactory.getLastModified(anyString())).then( + invocation -> lastModified(invocation.getArgument(0))); + return fileWrapperFactory; + } + + /** + * The number of watch service threads running, found by the watch + * service's poller in each thread's stack. + */ + private static long watchServiceThreads() { + return Thread.getAllStackTraces().values().stream() + .filter(stack -> Arrays.stream(stack).anyMatch(frame -> + frame.getClassName().endsWith("WatchService$Poller"))) + .count(); + } + + /** + * Wait for the number of watch service threads to become the count + * given, as a thread starts and ends a moment after its watch service + * is created or closed. + */ + private static boolean waitForWatchServiceThreads(long count) + throws InterruptedException { + long end = System.currentTimeMillis() + WAIT_FOR_EVENT_MILLIS; + while (watchServiceThreads() != count && + System.currentTimeMillis() < end) { + Thread.sleep(RETRY_MILLIS); + } + return watchServiceThreads() == count; + } + @SuppressWarnings("unchecked") private static OnPremiseAspectEngine engine() { From 5293783d718ac29a07cb3f84f67861492b6049c2 Mon Sep 17 00:00:00 2001 From: James Rosewell Date: Fri, 13 Nov 2026 00:42:45 +0000 Subject: [PATCH 4/6] Register a data file for the watcher alone and close its watch service The file system watcher runs on its own setting and needs no license key. The builder registered a data file with the data update service only for automatic updates or the update on start-up, so an engine with no license key, which must turn both off, was never watched. It now also registers a data file that is to be watched, when there is a data update service and a data file on disk. The data update service keeps each data file's watch service, as each runs a thread of its own. Unregistering the data file closes it, and the watch resumes on it after a download rather than on a new one, unless the data file was unregistered during the download. Unregistering stops the watcher's task even while the watch is paused, and the task does nothing while the watch is paused rather than fail, as a repeating task that fails is never run again. The notes on setDataFileSystemWatcher no longer say that automatic updates must be on for the watcher to work. --- .../DataFileConfigurationBuilderBase.java | 5 +- .../OnPremiseAspectEngineBuilderBase.java | 16 ++- .../SingleFileAspectEngineBuilderBase.java | 5 +- .../services/DataUpdateServiceDefault.java | 97 ++++++++++++++++--- 4 files changed, 103 insertions(+), 20 deletions(-) diff --git a/pipeline/pipeline.engines/src/main/java/fiftyone/pipeline/engines/configuration/DataFileConfigurationBuilderBase.java b/pipeline/pipeline.engines/src/main/java/fiftyone/pipeline/engines/configuration/DataFileConfigurationBuilderBase.java index 61589448..aeb9a1b2 100644 --- a/pipeline/pipeline.engines/src/main/java/fiftyone/pipeline/engines/configuration/DataFileConfigurationBuilderBase.java +++ b/pipeline/pipeline.engines/src/main/java/fiftyone/pipeline/engines/configuration/DataFileConfigurationBuilderBase.java @@ -148,8 +148,9 @@ public TBuilder setAutoUpdate(boolean enabled) { * disk and automatically refresh the engine as soon as the file is updated. * This setting enables/disables that feature. * - * The {@link #setAutoUpdate(boolean)} feature must also be enabled in order - * for the file system watcher to work. + * The file system watcher runs on its own setting, whether + * {@link #setAutoUpdate(boolean)} is on or off, and needs no license key, + * as watching a file downloads nothing. * If the engine is built from a byte[] then this setting does nothing. * @param enabled the cache configuration to use * @return this builder instance diff --git a/pipeline/pipeline.engines/src/main/java/fiftyone/pipeline/engines/flowelements/OnPremiseAspectEngineBuilderBase.java b/pipeline/pipeline.engines/src/main/java/fiftyone/pipeline/engines/flowelements/OnPremiseAspectEngineBuilderBase.java index a0995b9f..879f55f4 100644 --- a/pipeline/pipeline.engines/src/main/java/fiftyone/pipeline/engines/flowelements/OnPremiseAspectEngineBuilderBase.java +++ b/pipeline/pipeline.engines/src/main/java/fiftyone/pipeline/engines/flowelements/OnPremiseAspectEngineBuilderBase.java @@ -201,13 +201,27 @@ protected void preCreateEngine() { } if (dataFileConfig.getAutomaticUpdatesEnabled() || - dataFileConfig.getUpdateOnStartup()) { + dataFileConfig.getUpdateOnStartup() || + isWatched(dataFileConfig)) { dataUpdateService.registerDataFile(dataFile); } dataFiles.add(dataFile); } } + /** + * Whether the data file is to be watched for a new file on disk. The + * file system watcher runs on its own setting and needs neither + * automatic updates nor a license key, only a data update service to run + * it and a data file on disk. + */ + private boolean isWatched(DataFileConfiguration dataFileConfig) { + return dataUpdateService != null && + dataFileConfig.getFileSystemWatcherEnabled() && + dataFileConfig.getDataFilePath() != null && + dataFileConfig.getDataFilePath().isEmpty() == false; + } + /** * Refuse a data file configuration that needs a data update service or * a license key it does not have. Every data file is checked before any diff --git a/pipeline/pipeline.engines/src/main/java/fiftyone/pipeline/engines/flowelements/SingleFileAspectEngineBuilderBase.java b/pipeline/pipeline.engines/src/main/java/fiftyone/pipeline/engines/flowelements/SingleFileAspectEngineBuilderBase.java index 4a0cf866..84df5a4f 100644 --- a/pipeline/pipeline.engines/src/main/java/fiftyone/pipeline/engines/flowelements/SingleFileAspectEngineBuilderBase.java +++ b/pipeline/pipeline.engines/src/main/java/fiftyone/pipeline/engines/flowelements/SingleFileAspectEngineBuilderBase.java @@ -216,8 +216,9 @@ public TBuilder setAutoUpdate(boolean enabled) { * disk and automatically refresh the engine as soon as the file is updated. * This setting enables/disables that feature. *

- * The {@link #setAutoUpdate(boolean)} feature must also be enabled in order - * for the file system watcher to work. + * The file system watcher runs on its own setting, whether + * {@link #setAutoUpdate(boolean)} is on or off, and needs no license key, + * as watching a file downloads nothing. *

* If the engine is built from a byte[] then this setting does nothing. *

diff --git a/pipeline/pipeline.engines/src/main/java/fiftyone/pipeline/engines/services/DataUpdateServiceDefault.java b/pipeline/pipeline.engines/src/main/java/fiftyone/pipeline/engines/services/DataUpdateServiceDefault.java index 95bafae0..81675d1e 100644 --- a/pipeline/pipeline.engines/src/main/java/fiftyone/pipeline/engines/services/DataUpdateServiceDefault.java +++ b/pipeline/pipeline.engines/src/main/java/fiftyone/pipeline/engines/services/DataUpdateServiceDefault.java @@ -92,6 +92,13 @@ public class DataUpdateServiceDefault implements DataUpdateService { long watchedFileOpenTimeoutMillis = WATCHED_FILE_OPEN_TIMEOUT_MILLIS; final List configurations = new ArrayList<>(); + /** + * The watch service of each watched data file, held under the + * configuration lock. A watch service runs a thread of its own, so it is + * closed when its data file is unregistered. + */ + private final Map watchServices = + new IdentityHashMap<>(); private final Logger logger; private final boolean closeFutureFactory; private final List onUpdateCompleteList = new ArrayList<>(); @@ -567,9 +574,8 @@ public void run() { throw new IllegalStateException("Cannot find directory to watch"); } logger.debug("Getting a watcher using {}", aspectDataFilePath); - WatchService watcher = aspectDataFilePath - .getFileSystem() - .newWatchService(); + WatchService watcher = + watchService(aspectDataFile, aspectDataFilePath); logger.debug("Setting a watcher for {}", aspectDataFilePath.getParent()); aspectDataFile.getConfiguration().setWatchKey(aspectDataFilePath.getParent().register( watcher, @@ -587,8 +593,16 @@ public void run() { @Override public void run() { logger.debug("Watcher is running"); - for (WatchEvent event : - aspectDataFile.getConfiguration().getWatchKey().pollEvents()) { + // The watch is paused while a downloaded + // file is copied into place. A repeating + // task that fails never runs again, so + // this run does nothing. + WatchKey watchKey = + aspectDataFile.getConfiguration().getWatchKey(); + if (watchKey == null) { + return; + } + for (WatchEvent event : watchKey.pollEvents()) { if (event.kind() == StandardWatchEventKinds.ENTRY_MODIFY) { if (((Path) event.context()).endsWith(aspectDataFilePath.getFileName())) { dataFileUpdated(aspectDataFile); @@ -596,7 +610,7 @@ public void run() { } } // make sure to reset the watch key or we are not called again - aspectDataFile.getConfiguration().getWatchKey().reset(); + watchKey.reset(); } }, pollingInterval)); @@ -629,11 +643,13 @@ public void unregisterDataFile(AspectEngineDataFile dataFile) { } AspectEngineDataFileDefault aspectDataFile = (AspectEngineDataFileDefault) dataFile; + // The watcher's task is stopped even while the watch is + // paused for a download. + if (aspectDataFile.getPollFuture() != null) { + aspectDataFile.getPollFuture().cancel(true); + aspectDataFile.setPollFuture(null); + } if (aspectDataFile.getConfiguration().getWatchKey() != null) { - if (aspectDataFile.getPollFuture() != null) { - aspectDataFile.getPollFuture().cancel(true); - aspectDataFile.setPollFuture(null); - } aspectDataFile.getConfiguration().getWatchKey().cancel(); aspectDataFile.getConfiguration().setWatchKey(null); } @@ -644,12 +660,57 @@ public void unregisterDataFile(AspectEngineDataFile dataFile) { } } } + closeWatchService(dataFile); } if (dataFile != null) { dataFile.setDataUpdateService(null); } } + /** + * Get the data file's watch service, creating it the first time. The + * data file keeps the one watch service until it is unregistered, so the + * watch resumes on it after a download. + * + * @param dataFile the data file to watch + * @param path the absolute path of the data file + * @return the data file's watch service + * @throws IOException if a watch service cannot be created + */ + private WatchService watchService( + AspectEngineDataFile dataFile, + Path path) throws IOException { + synchronized (configLock) { + WatchService watcher = watchServices.get(dataFile); + if (watcher == null) { + watcher = path.getFileSystem().newWatchService(); + watchServices.put(dataFile, watcher); + } + return watcher; + } + } + + /** + * Close the data file's watch service, if it has one, which ends the + * watch service's thread. + * + * @param dataFile the data file no longer watched + */ + private void closeWatchService(AspectEngineDataFile dataFile) { + WatchService watcher; + synchronized (configLock) { + watcher = watchServices.remove(dataFile); + } + if (watcher != null) { + try { + watcher.close(); + } catch (IOException e) { + logger.warn("Could not close the file watcher for '{}'", + dataFile.getDataFilePath(), e); + } + } + } + /** * Get the polling interval for a data file plus a random number of * seconds, no greater than the configured maximum randomisation. @@ -1136,14 +1197,20 @@ private AutoUpdateStatus checkForUpdateFromUrl( dataFile.getDataFilePath(), ex); result = AutoUpdateStatus.AUTO_UPDATE_NEW_FILE_CANT_RENAME; } - // Make sure to enable the file watcher again if needed. + // Make sure to enable the file watcher again if needed. The + // watch resumes on the data file's own watch service, unless + // the data file was unregistered during the download. if (watchKey != null) { final Path filePath = Paths.get(dataFile.getDataFilePath()).toAbsolutePath(); - WatchService watcher = filePath.getFileSystem().newWatchService(); - watchKey = filePath.getParent().register(watcher, - StandardWatchEventKinds.ENTRY_MODIFY); - dataFile.getConfiguration().setWatchKey(watchKey); + synchronized (configLock) { + if (configurations.contains(dataFile)) { + watchKey = filePath.getParent().register( + watchService(dataFile, filePath), + StandardWatchEventKinds.ENTRY_MODIFY); + dataFile.getConfiguration().setWatchKey(watchKey); + } + } } } return result; From 324505b4bef6d050742d947139cbb2b1f862b393 Mon Sep 17 00:00:00 2001 From: James Rosewell Date: Fri, 13 Nov 2026 01:18:44 +0000 Subject: [PATCH 5/6] Test that the watcher's task needs a watch key and replaces an earlier one The watcher's task polls the watch key. With no watch key, as when the data file's folder does not exist, no task should be scheduled, as the task would otherwise run on doing nothing now that it no longer fails on a missing key. Registering a data file again that has no watch key should replace the watcher's task rather than leave the earlier one running beside it. The thread tests now follow the watch service threads started during each test, so a thread from an earlier test that ends late makes no difference. --- .../DataUpdateServiceWatchedFileTests.java | 132 ++++++++++++++---- 1 file changed, 106 insertions(+), 26 deletions(-) diff --git a/pipeline/pipeline.engines/src/test/java/fiftyone/pipeline/engines/services/DataUpdateServiceWatchedFileTests.java b/pipeline/pipeline.engines/src/test/java/fiftyone/pipeline/engines/services/DataUpdateServiceWatchedFileTests.java index 09eb4245..b3e6d13c 100644 --- a/pipeline/pipeline.engines/src/test/java/fiftyone/pipeline/engines/services/DataUpdateServiceWatchedFileTests.java +++ b/pipeline/pipeline.engines/src/test/java/fiftyone/pipeline/engines/services/DataUpdateServiceWatchedFileTests.java @@ -47,7 +47,10 @@ import java.nio.file.WatchKey; import java.util.ArrayList; import java.util.Arrays; +import java.util.HashSet; import java.util.List; +import java.util.Map; +import java.util.Set; import java.util.concurrent.CountDownLatch; import java.util.concurrent.ScheduledFuture; import java.util.concurrent.TimeUnit; @@ -58,6 +61,7 @@ import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertNotNull; +import static org.junit.Assert.assertNull; import static org.junit.Assert.assertTrue; import static org.junit.Assume.assumeFalse; import static org.mockito.ArgumentMatchers.any; @@ -254,14 +258,14 @@ public void Unregistering_ClosesTheWatchService() throws Exception { watchedFile(engine(), dataFile, false, true); try { - long before = watchServiceThreads(); + Set before = watchServiceThreads(); dataUpdate.registerDataFile(file); - assertTrue("Registering starts a watch service thread", - waitForWatchServiceThreads(before + 1)); + assertEquals("Registering starts a watch service thread", + 1, waitForThreadsStartedSince(before, 1).size()); dataUpdate.unregisterDataFile(file); - assertTrue("Unregistering ends the watch service thread", - waitForWatchServiceThreads(before)); + assertEquals("Unregistering ends the watch service thread", + 0, waitForThreadsStartedSince(before, 0).size()); } finally { cleanUp(file, directory, dataFile); @@ -307,10 +311,10 @@ public void DownloadedFile_WatchResumesOnTheSameWatchService() file.getConfiguration().setVerifyModifiedSince(false); try { - long before = watchServiceThreads(); + Set before = watchServiceThreads(); dataUpdate.registerDataFile(file); - assertTrue("Registering starts a watch service thread", - waitForWatchServiceThreads(before + 1)); + assertEquals("Registering starts a watch service thread", + 1, waitForThreadsStartedSince(before, 1).size()); // The polling timer downloads the new file. timers.get(0).run(); @@ -318,12 +322,12 @@ public void DownloadedFile_WatchResumesOnTheSameWatchService() 1, refreshCount(engine)); assertNotNull("The watch resumes after the download", file.getConfiguration().getWatchKey()); - assertTrue("The watch resumes on the same watch service", - waitForWatchServiceThreads(before + 1)); + assertEquals("The watch resumes on the same watch service", + 1, waitForThreadsStartedSince(before, 1).size()); dataUpdate.unregisterDataFile(file); - assertTrue("Unregistering ends the watch service thread", - waitForWatchServiceThreads(before)); + assertEquals("Unregistering ends the watch service thread", + 0, waitForThreadsStartedSince(before, 0).size()); } finally { cleanUp(file, directory, dataFile); @@ -373,6 +377,71 @@ public void WatchPaused_WatcherTaskCarriesOn() throws Exception { } } + /** + * The watcher's task polls the watch key, so with no watch key, here + * because the data file's folder does not exist, no task is scheduled. + */ + @Test + public void NoWatchKey_NoWatcherTask() throws Exception { + final AtomicReference watcher = new AtomicReference<>(); + DataUpdateService dataUpdate = new DataUpdateServiceDefault( + null, + mock(HttpClient.class), + fileWrapperFactory(), + futureFactory(new ArrayList<>(), watcher)); + Path directory = Files.createTempDirectory("watched"); + Path dataFile = directory.resolve("missing").resolve("data.dat"); + AspectEngineDataFileDefault file = + watchedFile(engine(), dataFile, false, true); + + try { + dataUpdate.registerDataFile(file); + assertNull("No watch key without the data file's folder", + file.getConfiguration().getWatchKey()); + assertNull("No watcher task without a watch key", + watcher.get()); + dataUpdate.unregisterDataFile(file); + } + finally { + directory.toFile().delete(); + } + } + + /** + * Registering a data file again that has no watch key, as it has while + * the watch is paused for a download, creates the watch key again and + * replaces the watcher's task rather than leave the old one running. + */ + @Test + public void RegisteredAgain_WatcherTaskReplaced() throws Exception { + DataUpdateService dataUpdate = new DataUpdateServiceDefault( + null, + mock(HttpClient.class), + fileWrapperFactory(), + futureFactory(new ArrayList<>(), new AtomicReference<>())); + Path directory = Files.createTempDirectory("watched"); + Path dataFile = directory.resolve("data.dat"); + Files.write(dataFile, "Old".getBytes()); + AspectEngineDataFileDefault file = + watchedFile(engine(), dataFile, false, true); + + try { + dataUpdate.registerDataFile(file); + ScheduledFuture first = file.getPollFuture(); + assertNotNull("The watcher's task was scheduled", first); + + file.getConfiguration().setWatchKey(null); + dataUpdate.registerDataFile(file); + assertNotNull("The watch key is created again", + file.getConfiguration().getWatchKey()); + verify(first).cancel(false); + dataUpdate.unregisterDataFile(file); + } + finally { + cleanUp(file, directory, dataFile); + } + } + /** * A watched file that cannot be opened is tried, with a pause between * tries, until the wait ends. The update then reports @@ -598,29 +667,40 @@ private static FileWrapperFactory fileWrapperFactory() { } /** - * The number of watch service threads running, found by the watch - * service's poller in each thread's stack. + * The watch service threads running, found by the watch service's + * poller in each thread's stack. */ - private static long watchServiceThreads() { - return Thread.getAllStackTraces().values().stream() - .filter(stack -> Arrays.stream(stack).anyMatch(frame -> - frame.getClassName().endsWith("WatchService$Poller"))) - .count(); + private static Set watchServiceThreads() { + Set threads = new HashSet<>(); + for (Map.Entry thread : + Thread.getAllStackTraces().entrySet()) { + if (Arrays.stream(thread.getValue()).anyMatch(frame -> + frame.getClassName().endsWith("WatchService$Poller"))) { + threads.add(thread.getKey()); + } + } + return threads; } /** - * Wait for the number of watch service threads to become the count - * given, as a thread starts and ends a moment after its watch service - * is created or closed. + * Wait for the watch service threads running that were not running + * before to number the count given, as a thread starts and ends a moment + * after its watch service is created or closed. Threads running before + * are left out, so one of those ending makes no difference. */ - private static boolean waitForWatchServiceThreads(long count) - throws InterruptedException { + private static Set waitForThreadsStartedSince( + Set before, + int count) throws InterruptedException { long end = System.currentTimeMillis() + WAIT_FOR_EVENT_MILLIS; - while (watchServiceThreads() != count && + Set started = watchServiceThreads(); + started.removeAll(before); + while (started.size() != count && System.currentTimeMillis() < end) { Thread.sleep(RETRY_MILLIS); + started = watchServiceThreads(); + started.removeAll(before); } - return watchServiceThreads() == count; + return started; } @SuppressWarnings("unchecked") From ba99c5da6f7f31a5bbcbc02d500037e341d6edcd Mon Sep 17 00:00:00 2001 From: James Rosewell Date: Fri, 13 Nov 2026 01:33:55 +0000 Subject: [PATCH 6/6] Schedule the watcher's task only with a watch key, replacing an earlier one The watcher's task polls the watch key, so it is scheduled only when the watch key was created. Before, a task was scheduled whatever happened, and failed on its first run when there was no watch key. Now that a run that finds no watch key does nothing, that task would run on for nothing. A data file registered again with no watch key gets a new task, and the earlier task is cancelled rather than left running beside it. --- .../services/DataUpdateServiceDefault.java | 72 ++++++++++--------- 1 file changed, 40 insertions(+), 32 deletions(-) diff --git a/pipeline/pipeline.engines/src/main/java/fiftyone/pipeline/engines/services/DataUpdateServiceDefault.java b/pipeline/pipeline.engines/src/main/java/fiftyone/pipeline/engines/services/DataUpdateServiceDefault.java index 81675d1e..398bfd9b 100644 --- a/pipeline/pipeline.engines/src/main/java/fiftyone/pipeline/engines/services/DataUpdateServiceDefault.java +++ b/pipeline/pipeline.engines/src/main/java/fiftyone/pipeline/engines/services/DataUpdateServiceDefault.java @@ -584,41 +584,49 @@ public void run() { logger.error("File watcher for '{}' could not be initialised. " + e.getMessage(), aspectDataFile.getDataFilePath(), e); } - try { - long pollingInterval = - aspectDataFile.getConfiguration().getPollingIntervalSeconds() * 1000L; - logger.debug("Setting poll future {} millis", pollingInterval); - aspectDataFile.setPollFuture(futureFactory.scheduleRepeating( - new Runnable() { - @Override - public void run() { - logger.debug("Watcher is running"); - // The watch is paused while a downloaded - // file is copied into place. A repeating - // task that fails never runs again, so - // this run does nothing. - WatchKey watchKey = - aspectDataFile.getConfiguration().getWatchKey(); - if (watchKey == null) { - return; - } - for (WatchEvent event : watchKey.pollEvents()) { - if (event.kind() == StandardWatchEventKinds.ENTRY_MODIFY) { - if (((Path) event.context()).endsWith(aspectDataFilePath.getFileName())) { - dataFileUpdated(aspectDataFile); + // The watcher's task polls the watch key, so it is scheduled + // only when there is one, and replaces any task scheduled + // before rather than run beside it. + if (aspectDataFile.getConfiguration().getWatchKey() != null) { + try { + if (aspectDataFile.getPollFuture() != null) { + aspectDataFile.getPollFuture().cancel(false); + } + long pollingInterval = + aspectDataFile.getConfiguration().getPollingIntervalSeconds() * 1000L; + logger.debug("Setting poll future {} millis", pollingInterval); + aspectDataFile.setPollFuture(futureFactory.scheduleRepeating( + new Runnable() { + @Override + public void run() { + logger.debug("Watcher is running"); + // The watch is paused while a downloaded + // file is copied into place. A repeating + // task that fails never runs again, so + // this run does nothing. + WatchKey watchKey = + aspectDataFile.getConfiguration().getWatchKey(); + if (watchKey == null) { + return; + } + for (WatchEvent event : watchKey.pollEvents()) { + if (event.kind() == StandardWatchEventKinds.ENTRY_MODIFY) { + if (((Path) event.context()).endsWith(aspectDataFilePath.getFileName())) { + dataFileUpdated(aspectDataFile); + } } } + // make sure to reset the watch key or we are not called again + watchKey.reset(); } - // make sure to reset the watch key or we are not called again - watchKey.reset(); - } - }, - pollingInterval)); - logger.debug("Poll check scheduled for {} mins", - aspectDataFile.getPollFuture().getDelay(TimeUnit.MINUTES)); - } catch (Exception e) { - logger.error("Could not schedule watcher for {}.", - aspectDataFile.getDataFilePath(), e); + }, + pollingInterval)); + logger.debug("Poll check scheduled for {} mins", + aspectDataFile.getPollFuture().getDelay(TimeUnit.MINUTES)); + } catch (Exception e) { + logger.error("Could not schedule watcher for {}.", + aspectDataFile.getDataFilePath(), e); + } } }