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..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 @@ -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 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. + */ + 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); + } } } } 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..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 @@ -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 finds no update and starts before registration has + * stored its future must still leave the next check scheduled and + * stored. A check whose download fails takes the same 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,154 @@ private void configureTimerImmediateCallbackOnce() { configureTimerImmediateCallback(1, null); } + private static final long RACE_WAIT_SECONDS = 10; + + /** + * 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(); + 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";