Skip to content

Store a data file's future before its check can read it #71

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
jwrosewell merged 3 commits into from
Nov 8, 2026
Merged

Store a data file's future before its check can read it #71

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
28d5891
Test a check that starts before registration stores its future
jwrosewell Nov 8, 2026
ce1a533
Store a data file's future before its check can read it
jwrosewell Nov 8, 2026
9424d53
Word the new comments to match what the lock and the tests cover
jwrosewell Nov 8, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
116 changes: 67 additions & 49 deletions ...ne.engines/src/main/java/fiftyone/pipeline/engines/services/DataUpdateServiceDefault.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -69,6 +69,14 @@ public class DataUpdateServiceDefault implements DataUpdateService {
private final boolean closeFutureFactory;
private final List<OnUpdateComplete> 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.
*/
Expand Down Expand Up @@ -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())));
}
}
}
}
Expand Down Expand Up @@ -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) {
Expand Down Expand Up @@ -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);
}
}

Expand Down Expand Up @@ -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);
}
}
}
}
Expand Down
Loading
Loading