From 28d58916db5a3adf2d8bbf0eb3fb7303ed88c46f Mon Sep 17 00:00:00 2001 From: James Rosewell Date: Sun, 8 Nov 2026 09:47:42 +0000 Subject: [PATCH 1/3] Test a check that starts before registration stores its future registerDataFile schedules the first check and only then stores the future it gets back, so a check with a short delay can start in between. If the check reads the future in that gap it finds none, clears nothing, and its own re-registration then finds the stored future and schedules no next check. DataUpdateService_UpdateFromFile_FileUpdated failed this way once in a full build on macOS x64 with Java 11. The two new tests force that order on every run, one for a check that applies an update and one for a check that applies none, which is the path the retry after a 429 response takes. Registration is held after scheduling until the check has read the future or is blocked on a lock that registration holds, and the check's first read waits until registration has stored the future. Every wait is bounded, so a design that deadlocks fails rather than hangs. Each failure message lists the order in which the two threads read and wrote the future. For #70 --- .../services/DataUpdateServiceTests.java | 219 ++++++++++++++++++ 1 file changed, 219 insertions(+) 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 1e24bb485..e315eb06e 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 @@ -51,6 +51,9 @@ import java.io.File; import java.io.FileWriter; import java.io.IOException; +import java.lang.management.ManagementFactory; +import java.lang.management.ThreadInfo; +import java.lang.management.ThreadMXBean; import java.net.HttpURLConnection; import java.net.URL; import java.nio.file.Files; @@ -61,8 +64,10 @@ import java.util.Calendar; import java.util.Collection; import java.util.Date; +import java.util.List; import java.util.concurrent.*; import java.util.concurrent.atomic.AtomicInteger; +import java.util.concurrent.atomic.AtomicReference; import static fiftyone.pipeline.engines.services.DataUpdateService.AutoUpdateStatus.AUTO_UPDATE_HTTPS_ERR; import static fiftyone.pipeline.engines.services.DataUpdateService.AutoUpdateStatus.AUTO_UPDATE_NOT_NEEDED; @@ -492,6 +497,73 @@ public void call(Object sender, DataUpdateService.DataUpdateCompleteArgs args) { anyLong()); } + /** + * A check that finds an update and starts before registration has + * stored its future must still leave the next check scheduled and + * stored. + */ + @Test + public void DataUpdateService_UpdateFromFile_FileUpdated_CheckBeforeFutureStored() + throws InterruptedException { + // Arrange + RaceDataFile file = new RaceDataFile(); + OnPremiseAspectEngine engine = + mock(OnPremiseAspectEngine.class); + DataFileConfigurationDefault config = new DataFileConfigurationDefault(); + config.setAutomaticUpdatesEnabled(true); + config.setFileSystemWatcherEnabled(false); + file.setEngine(engine); + file.setConfiguration(config); + when(engine.getDataFileMetaData()).thenReturn(file); + configureFileUpdate(engine); + configureScheduleRace(file); + Semaphore completeFlag = onUpdateCompleteFlag(); + + // Act + dataUpdate.registerDataFile(file); + boolean completed = completeFlag.tryAcquire( + RACE_WAIT_SECONDS * 3, TimeUnit.SECONDS); + + // Assert + assertTrue("The OnUpdateComplete event was never fired", completed); + verify(engine, times(1)).refreshData((String) any()); + assertNextCheckScheduled(file); + } + + /** + * A check that applies no update and starts before registration has + * stored its future must still leave the next check scheduled and + * stored. Every check that applies no update takes this path, the retry + * after a 429 response among them. + */ + @Test + public void DataUpdateService_UpdateFromFile_FileNotUpdated_CheckBeforeFutureStored() + throws InterruptedException { + // Arrange + RaceDataFile file = new RaceDataFile(); + OnPremiseAspectEngine engine = + mock(OnPremiseAspectEngine.class); + DataFileConfigurationDefault config = new DataFileConfigurationDefault(); + config.setAutomaticUpdatesEnabled(true); + config.setFileSystemWatcherEnabled(false); + file.setEngine(engine); + file.setConfiguration(config); + when(engine.getDataFileMetaData()).thenReturn(file); + configureFileNoUpdate(engine); + configureScheduleRace(file); + Semaphore completeFlag = onUpdateCompleteFlag(); + + // Act + dataUpdate.registerDataFile(file); + boolean completed = completeFlag.tryAcquire( + RACE_WAIT_SECONDS * 3, TimeUnit.SECONDS); + + // Assert + assertTrue("The OnUpdateComplete event was never fired", completed); + verify(engine, never()).refreshData(anyString()); + assertNextCheckScheduled(file); + } + @Test public void DataUpdateService_UpdateFromUrl_NoUpdate() throws InterruptedException, IOException { // Arrange @@ -1409,6 +1481,153 @@ private void configureTimerImmediateCallbackOnce() { configureTimerImmediateCallback(1, null); } + private static final long RACE_WAIT_SECONDS = 10; + + /** + * Data file that records each read and write of its future. The check's + * first read waits until registration has stored the future, so a check + * that reads it too early goes on with a stale value. + */ + private static class RaceDataFile extends AspectEngineDataFileDefault { + private final Thread registering = Thread.currentThread(); + private final List events = new CopyOnWriteArrayList<>(); + private final AtomicInteger schedules = new AtomicInteger(); + private final AtomicReference> first = + new AtomicReference<>(); + private final AtomicReference> next = + new AtomicReference<>(); + private final AtomicReference check = new AtomicReference<>(); + private final CountDownLatch checkRead = new CountDownLatch(1); + private final CountDownLatch stored = new CountDownLatch(1); + private volatile boolean registrationReleased; + private volatile boolean storedInTime = true; + + private String who() { + return Thread.currentThread() == registering ? + "registration" : "check"; + } + + private String name(ScheduledFuture future) { + if (future == null) { + return "no future"; + } + if (future == first.get()) { + return "the first future"; + } + return future == next.get() ? "the next future" : "another future"; + } + + @Override + public ScheduledFuture getFuture() { + ScheduledFuture future = super.getFuture(); + if (Thread.currentThread() != registering) { + events.add("check read " + name(future)); + if (checkRead.getCount() > 0) { + checkRead.countDown(); + try { + storedInTime = stored.await( + RACE_WAIT_SECONDS, TimeUnit.SECONDS); + } catch (InterruptedException e) { + storedInTime = false; + Thread.currentThread().interrupt(); + } + } + } + return future; + } + + @Override + public void setFuture(ScheduledFuture future) { + super.setFuture(future); + events.add(future == null ? + who() + " cleared the future" : + who() + " stored " + name(future)); + if (Thread.currentThread() == registering) { + stored.countDown(); + } + } + } + + /** + * Run the first check at once, and hold registration after scheduling + * it until the check has read the future or waits for a lock that + * registration holds. Later checks are never run. + */ + private void configureScheduleRace(final RaceDataFile file) { + final ThreadMXBean threads = ManagementFactory.getThreadMXBean(); + doAnswer(invocation -> { + final Runnable check = invocation.getArgument(0); + if (file.schedules.incrementAndGet() > 1) { + ScheduledFuture next = mock(ScheduledFuture.class); + file.next.set(next); + file.events.add(file.who() + " scheduled the next check"); + return next; + } + file.events.add(file.who() + " scheduled the first check"); + ScheduledFuture first = executorService.schedule(() -> { + file.check.set(Thread.currentThread()); + check.run(); + }, 0, TimeUnit.MILLISECONDS); + file.first.set(first); + file.registrationReleased = waitForCheck(file, threads); + return first; + }).when(futureFactory).schedule(any(Runnable.class), anyLong()); + } + + /** + * Wait until the check has read the future, or is blocked on a lock + * that the registering thread holds and so cannot read it until + * registration has finished. Thread.getId() is deprecated from Java 19 + * but is the id that ThreadMXBean reports on every version. + */ + @SuppressWarnings("deprecation") + private static boolean waitForCheck( + RaceDataFile file, + ThreadMXBean threads) throws InterruptedException { + long deadline = System.nanoTime() + + TimeUnit.SECONDS.toNanos(RACE_WAIT_SECONDS); + while (System.nanoTime() < deadline) { + if (file.checkRead.await(10, TimeUnit.MILLISECONDS)) { + return true; + } + Thread check = file.check.get(); + if (check != null) { + ThreadInfo info = threads.getThreadInfo(check.getId()); + if (info != null && + info.getLockOwnerId() == file.registering.getId()) { + return true; + } + } + } + return false; + } + + /** + * Assert that one check was scheduled at registration and one after it, + * and that the future stored is the next check's. Each message carries + * the order in which the two threads read and wrote the future. + */ + private static void assertNextCheckScheduled(RaceDataFile file) { + String events = " Events: " + file.events; + assertTrue("The check neither read the future nor waited for " + + "registration." + events, + file.registrationReleased); + assertTrue("Registration did not store the future while the " + + "check waited." + events, + file.storedInTime); + assertEquals("Schedules." + events, 2, file.schedules.get()); + assertSame("The future stored is not the next check's." + events, + file.next.get(), + file.getFuture()); + } + + private Semaphore onUpdateCompleteFlag() throws InterruptedException { + final Semaphore completeFlag = new Semaphore(1); + completeFlag.acquire(); + dataUpdate.onUpdateComplete((sender, args) -> completeFlag.release()); + return completeFlag; + } + private void configureFileNoUpdate(OnPremiseAspectEngine engine) { String dataFile = "C:/test/tempFile.dat"; String tempFile = "C:/test/dataFile.dat"; From ce1a533bdcfb4fa067cbaf62175861f06464916b Mon Sep 17 00:00:00 2001 From: James Rosewell Date: Sun, 8 Nov 2026 09:47:43 +0000 Subject: [PATCH 2/3] Store a data file's future before its check can read it Every read and write of a data file's future now happens while holding one lock in the data update service. registerDataFile holds it from scheduling the check to storing the future, and the check takes it before it cancels, clears or replaces the future, so a check that starts at once waits until registration has stored its future. Unregistering takes it for the same reason. The lock is the service's own rather than the data file's update lock, because the file system watcher holds the update lock while it refreshes the engine and while it waits for the new file to open. Taking that lock for the future would make unregistering wait for a refresh. Only scheduling, cancelling and storing happen while holding the new lock, so it is never held while waiting for the configuration lock or an update lock. For #70 --- .../services/DataUpdateServiceDefault.java | 116 ++++++++++-------- 1 file changed, 67 insertions(+), 49 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 7430ea3cc..df145a374 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 @@ -69,6 +69,14 @@ public class DataUpdateServiceDefault implements DataUpdateService { private final boolean closeFutureFactory; private final List onUpdateCompleteList = new ArrayList<>(); private final Object configLock = new Object(); + /** + * Held for every read and write of a data file's scheduled check, so a + * check cannot read or clear its future before + * {@link #registerDataFile} has stored it. Only the future is scheduled, + * cancelled and stored while holding it, so it is never held while + * waiting for the configuration lock or a data file's update lock. + */ + private final Object futureLock = new Object(); /** * HTTP client used to download new data files. */ @@ -226,16 +234,18 @@ private AutoUpdateStatus checkForUpdate(Object state, boolean manualUpdate) { // This will repeat until the update is acquired. if (manualUpdate == false) { logger.debug("Cancelling and resetting update timer"); - if (dataFile.getFuture() != null) { - dataFile.getFuture().cancel(true); - dataFile.setFuture(futureFactory.schedule( - new Runnable() { - @Override - public void run() { - checkForUpdate(dataFile, false); - } - }, - getInterval(dataFile.getConfiguration()))); + synchronized (futureLock) { + if (dataFile.getFuture() != null) { + dataFile.getFuture().cancel(true); + dataFile.setFuture(futureFactory.schedule( + new Runnable() { + @Override + public void run() { + checkForUpdate(dataFile, false); + } + }, + getInterval(dataFile.getConfiguration()))); + } } } } @@ -268,10 +278,12 @@ public void run() { private AutoUpdateStatus updatedFileAvailable(AspectEngineDataFile dataFile) { AspectEngineDataFileDefault aspectDataFile = (AspectEngineDataFileDefault) dataFile; - if (aspectDataFile.getFuture() != null) { - // Dispose of the old timer object - aspectDataFile.getFuture().cancel(true); - aspectDataFile.setFuture(null); + synchronized (futureLock) { + if (aspectDataFile.getFuture() != null) { + // Dispose of the old timer object + aspectDataFile.getFuture().cancel(true); + aspectDataFile.setFuture(null); + } } if (dataFile.getEngine() == null) { @@ -417,39 +429,43 @@ public void registerDataFile(final AspectEngineDataFile dataFile) { // Only create an automatic update timer if auto updates are // enabled for this engine and there is not already an associated - // timer. - if (aspectDataFile.getAutomaticUpdatesEnabled() && - aspectDataFile.getFuture() == null) { - long delay = getInterval(aspectDataFile.getConfiguration()); - if (aspectDataFile.getUpdateAvailableTime() != null) { - if (aspectDataFile.getUpdateAvailableTime().getTime() - > System.currentTimeMillis()) { - delay = applyIntervalRandomisation( - aspectDataFile.getUpdateAvailableTime().getTime() - - System.currentTimeMillis(), - aspectDataFile.getConfiguration()); + // timer. The lock is held from scheduling to storing, because + // the check can start as soon as it is scheduled and must find + // its own future. + synchronized (futureLock) { + if (aspectDataFile.getAutomaticUpdatesEnabled() && + aspectDataFile.getFuture() == null) { + long delay = getInterval(aspectDataFile.getConfiguration()); + if (aspectDataFile.getUpdateAvailableTime() != null) { + if (aspectDataFile.getUpdateAvailableTime().getTime() + > System.currentTimeMillis()) { + delay = applyIntervalRandomisation( + aspectDataFile.getUpdateAvailableTime().getTime() + - System.currentTimeMillis(), + aspectDataFile.getConfiguration()); + } + } + // Create a timer that will go off when the engine expects + // updated data to be available. + logger.debug("Scheduling update check with delay {}", delay); + try { + ScheduledFuture future = futureFactory.schedule( + new Runnable() { + @Override + public void run() { + logger.debug("running checkForUpdate {}", + aspectDataFile.getDataFilePath()); + checkForUpdate(aspectDataFile, false); + logger.debug("Returning from checkForUpdate"); + } + }, + delay); + logger.debug("Update check scheduled for {} millis", + future.getDelay(TimeUnit.MILLISECONDS)); + aspectDataFile.setFuture(future); + } catch (Throwable t) { + logger.error("Exception when scheduling update check", t); } - } - // Create a timer that will go off when the engine expects - // updated data to be available. - logger.debug("Scheduling update check with delay {}", delay); - try { - ScheduledFuture future = futureFactory.schedule( - new Runnable() { - @Override - public void run() { - logger.debug("running checkForUpdate {}", - aspectDataFile.getDataFilePath()); - checkForUpdate(aspectDataFile, false); - logger.debug("Returning from checkForUpdate"); - } - }, - delay); - logger.debug("Update check scheduled for {} millis", - future.getDelay(TimeUnit.MILLISECONDS)); - aspectDataFile.setFuture(future); - } catch (Throwable t) { - logger.error("Exception when scheduling update check", t); } } @@ -539,9 +555,11 @@ public void unregisterDataFile(AspectEngineDataFile dataFile) { aspectDataFile.getConfiguration().getWatchKey().cancel(); aspectDataFile.getConfiguration().setWatchKey(null); } - if (aspectDataFile.getFuture() != null) { - aspectDataFile.getFuture().cancel(true); - aspectDataFile.setFuture(null); + synchronized (futureLock) { + if (aspectDataFile.getFuture() != null) { + aspectDataFile.getFuture().cancel(true); + aspectDataFile.setFuture(null); + } } } } From 9424d538dc55d348e0d78c1a9e6631350cd935a4 Mon Sep 17 00:00:00 2001 From: James Rosewell Date: Sun, 8 Nov 2026 09:53:20 +0000 Subject: [PATCH 3/3] Word the new comments to match what the lock and the tests cover The lock's comment now says it covers the service's own reads and writes of the future, the test data file's comment says it records the check's reads, and the second test's comment says which checks share its path. No code changes. For #70 --- .../engines/services/DataUpdateServiceDefault.java | 4 ++-- .../engines/services/DataUpdateServiceTests.java | 11 ++++++----- 2 files changed, 8 insertions(+), 7 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 df145a374..dc2aabe8b 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 @@ -70,8 +70,8 @@ public class DataUpdateServiceDefault implements DataUpdateService { private final List onUpdateCompleteList = new ArrayList<>(); private final Object configLock = new Object(); /** - * Held for every read and write of a data file's scheduled check, so a - * check cannot read or clear its future before + * Held for every read and write this service makes of a data file's + * future, so a check cannot read or clear its future before * {@link #registerDataFile} has stored it. Only the future is scheduled, * cancelled and stored while holding it, so it is never held while * waiting for the configuration lock or a data file's update lock. 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 e315eb06e..d8768cf6c 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 @@ -531,9 +531,9 @@ public void DataUpdateService_UpdateFromFile_FileUpdated_CheckBeforeFutureStored } /** - * A check that applies no update and starts before registration has + * A check that finds no update and starts before registration has * stored its future must still leave the next check scheduled and - * stored. Every check that applies no update takes this path, the retry + * stored. A check whose download fails takes the same path, the retry * after a 429 response among them. */ @Test @@ -1484,9 +1484,10 @@ private void configureTimerImmediateCallbackOnce() { private static final long RACE_WAIT_SECONDS = 10; /** - * Data file that records each read and write of its future. The check's - * first read waits until registration has stored the future, so a check - * that reads it too early goes on with a stale value. + * Data file that records every write of its future and every read the + * check makes. The check's first read waits until registration has + * stored the future, so a check that reads it too early goes on with a + * stale value. */ private static class RaceDataFile extends AspectEngineDataFileDefault { private final Thread registering = Thread.currentThread();