Skip to content

Make all polling updaters wait for graph update finish #6262

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
merged 12 commits into from
Jan 9, 2025
Merged

Make all polling updaters wait for graph update finish #6262

Show file tree
Hide file tree
Changes from 1 commit
Commits
Show all changes
12 commits
Select commit Hold shift + click to select a range
12b4995
make all polling updater wait for graph update finish
miklcct Nov 19, 2024
5edfd73
Merge branch 'dev-2.x' into fix-update-lag
miklcct Nov 26, 2024
ed778da
Merge branch 'dev-2.x' into fix-update-lag
miklcct Nov 27, 2024
c74c9e5
Merge branch 'dev-2.x' into fix-update-lag
miklcct Dec 4, 2024
0944921
extract future handling logic to a new method
miklcct Dec 12, 2024
4df65a1
add feature flag
miklcct Dec 12, 2024
d8a34dd
refactor update graph logic to PollingGraphUpdater
miklcct Dec 12, 2024
69ac485
add test for desired feature flag
miklcct Dec 12, 2024
0037b5a
implement feature flag
miklcct Dec 12, 2024
2ce1a6d
Merge branch 'dev-2.x' into fix-update-lag
miklcct Dec 17, 2024
c0f7a3f
Merge branch 'dev-2.x' into fix-update-lag
miklcct Jan 7, 2025
74241bd
Merge branch 'dev-2.x' into fix-update-lag
miklcct Jan 9, 2025
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
45 changes: 21 additions & 24 deletions application/src/main/java/org/opentripplanner/updater/alert/GtfsRealtimeAlertsUpdater.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -2,6 +2,7 @@

import com.google.transit.realtime.GtfsRealtime.FeedMessage;
import java.net.URI;
import java.util.concurrent.ExecutionException;
import org.opentripplanner.framework.io.OtpHttpClient;
import org.opentripplanner.framework.io.OtpHttpClientFactory;
import org.opentripplanner.routing.impl.TransitAlertServiceImpl;
Expand Down Expand Up @@ -63,32 +64,28 @@ public String toString() {
}

@Override
protected void runPolling() {
try {
final FeedMessage feed = otpHttpClient.getAndMap(
URI.create(url),
this.headers.asMap(),
FeedMessage.PARSER::parseFrom
);
protected void runPolling() throws InterruptedException, ExecutionException {
final FeedMessage feed = otpHttpClient.getAndMap(
URI.create(url),
this.headers.asMap(),
FeedMessage.PARSER::parseFrom
);

long feedTimestamp = feed.getHeader().getTimestamp();
if (feedTimestamp == lastTimestamp) {
LOG.debug("Ignoring feed with a timestamp that has not been updated from {}", url);
return;
}
if (feedTimestamp < lastTimestamp) {
LOG.info("Ignoring feed with older than previous timestamp from {}", url);
return;
}
long feedTimestamp = feed.getHeader().getTimestamp();
if (feedTimestamp == lastTimestamp) {
LOG.debug("Ignoring feed with a timestamp that has not been updated from {}", url);
return;
}
if (feedTimestamp < lastTimestamp) {
LOG.info("Ignoring feed with older than previous timestamp from {}", url);
return;
}

// Handle update in graph writer runnable
saveResultOnGraph.execute(context ->
updateHandler.update(feed, context.gtfsRealtimeFuzzyTripMatcher())
);
// Handle update in graph writer runnable
saveResultOnGraph
.execute(context -> updateHandler.update(feed, context.gtfsRealtimeFuzzyTripMatcher()))
.get();
Copy link
Member

@t2gran t2gran Nov 22, 2024 •
edited
Loading

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I think we should have a timeout here - what happens if a task take too much time or does not return ?

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

The task is to write the update into the graph. I am not sure what will happen if the task to update the graph is forcibly terminated.

Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

If we don't want to use the asynchronicity of the Future, why do we even need a Future at all? I think in such a case the method could just return void.

In any case, we need to discuss this in the dev meeting.

Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

After further discussion I finally realized what's going on here. In fact @leonardehrenfried, this is using the Future to delay completion of the polling graph updater until the subtask it enqueues on a different ExecutorService completes. I will comment further on the main discussion thread.

Copy link
Member

@t2gran t2gran Dec 10, 2024 •
edited by abyrd
Loading

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This is a good change, but we need to be able to turn this off in case it have some side-effects we have missed. This is critical code, and it is possible that we have not understood all implications. So, it would be nice to introduce a OTPFeature to be able to turn this the synchronisation off. This need to be done for all calls to Feature#get() introduced in this PR.

Suggested change
saveResultOnGraph
.execute(context -> updateHandler.update(feed, context.gtfsRealtimeFuzzyTripMatcher()))
.get();
var f = saveResultOnGraph.execute(context -> updateHandler.update(feed, context.gtfsRealtimeFuzzyTripMatcher()));
if(OTPFeature.WaitOnPullingUpdatersBeforeResheduling.isOn()) {
f.get();
}

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I will add a method to the updater and a unit test to do that, such that the behaviour is testable.

Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This is a good change, but we need to be able to turn this off in case it have some side-effects we have missed.

In the alternative I suggested, this could be accomplished by an additional condition

if (OTPFeature.AwaitPollingUpdaterCompletion.isOff() || status == null || status.isDone()) {
        status = runPolling();
}

Though I think safety measures (and maybe even the OTPOption) are less necessary in an implementation where there are no blocking calls.

Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

(@t2gran sorry, I accidentally edited your comment instead of hitting "quote reply". I think I restored your comment to its original form.)


lastTimestamp = feedTimestamp;
} catch (Exception e) {
LOG.error("Error reading gtfs-realtime feed from " + url, e);
}
lastTimestamp = feedTimestamp;
}
}
5 changes: 3 additions & 2 deletions application/src/main/java/org/opentripplanner/updater/trip/PollingTripUpdater.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -2,6 +2,7 @@

import com.google.transit.realtime.GtfsRealtime.TripUpdate;
import java.util.List;
import java.util.concurrent.ExecutionException;
import java.util.function.Consumer;
import org.opentripplanner.updater.spi.PollingGraphUpdater;
import org.opentripplanner.updater.spi.UpdateResult;
Expand Down Expand Up @@ -73,7 +74,7 @@ public void setup(WriteToGraphCallback writeToGraphCallback) {
* applies those updates to the graph.
*/
@Override
public void runPolling() {
public void runPolling() throws InterruptedException, ExecutionException {
// Get update lists from update source
List<TripUpdate> updates = updateSource.getUpdates();
var incrementality = updateSource.incrementalityOfLastUpdates();
Expand All @@ -89,7 +90,7 @@ public void runPolling() {
feedId,
recordMetrics
);
saveResultOnGraph.execute(runnable);
saveResultOnGraph.execute(runnable).get();
}
}

Expand Down
5 changes: 3 additions & 2 deletions ...n/java/org/opentripplanner/updater/vehicle_parking/VehicleParkingAvailabilityUpdater.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -2,6 +2,7 @@

import java.util.List;
import java.util.Map;
import java.util.concurrent.ExecutionException;
import java.util.function.Function;
import java.util.stream.Collectors;
import org.opentripplanner.routing.vehicle_parking.VehicleParking;
Expand Down Expand Up @@ -49,12 +50,12 @@ public void setup(WriteToGraphCallback writeToGraphCallback) {
}

@Override
protected void runPolling() {
protected void runPolling() throws InterruptedException, ExecutionException {
if (source.update()) {
var updates = source.getUpdates();

var graphWriterRunnable = new AvailabilityUpdater(updates);
saveResultOnGraph.execute(graphWriterRunnable);
saveResultOnGraph.execute(graphWriterRunnable).get();
}
}

Expand Down
5 changes: 3 additions & 2 deletions ...tion/src/main/java/org/opentripplanner/updater/vehicle_parking/VehicleParkingUpdater.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -6,6 +6,7 @@
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.ExecutionException;
import java.util.function.Function;
import java.util.stream.Collectors;
import org.opentripplanner.routing.graph.Graph;
Expand Down Expand Up @@ -69,7 +70,7 @@ public void setup(WriteToGraphCallback writeToGraphCallback) {
}

@Override
protected void runPolling() {
protected void runPolling() throws InterruptedException, ExecutionException {
LOG.debug("Updating vehicle parkings from {}", source);
if (!source.update()) {
LOG.debug("No updates");
Expand All @@ -81,7 +82,7 @@ protected void runPolling() {
VehicleParkingGraphWriterRunnable graphWriterRunnable = new VehicleParkingGraphWriterRunnable(
vehicleParkings
);
saveResultOnGraph.execute(graphWriterRunnable);
saveResultOnGraph.execute(graphWriterRunnable).get();
}

private class VehicleParkingGraphWriterRunnable implements GraphWriterRunnable {
Expand Down
5 changes: 3 additions & 2 deletions ...main/java/org/opentripplanner/updater/vehicle_position/PollingVehiclePositionUpdater.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -3,6 +3,7 @@
import com.google.transit.realtime.GtfsRealtime.VehiclePosition;
import java.util.List;
import java.util.Set;
import java.util.concurrent.ExecutionException;
import org.opentripplanner.service.realtimevehicles.RealtimeVehicleRepository;
import org.opentripplanner.service.realtimevehicles.model.RealtimeVehicle;
import org.opentripplanner.standalone.config.routerconfig.updaters.VehiclePositionsUpdaterConfig;
Expand Down Expand Up @@ -64,7 +65,7 @@ public void setup(WriteToGraphCallback writeToGraphCallback) {
* applies those updates to the graph.
*/
@Override
public void runPolling() {
public void runPolling() throws InterruptedException, ExecutionException {
// Get update lists from update source
List<VehiclePosition> updates = vehiclePositionSource.getPositions();

Expand All @@ -77,7 +78,7 @@ public void runPolling() {
fuzzyTripMatching,
updates
);
saveResultOnGraph.execute(runnable);
saveResultOnGraph.execute(runnable).get();
}
}

Expand Down
5 changes: 3 additions & 2 deletions ...cation/src/main/java/org/opentripplanner/updater/vehicle_rental/VehicleRentalUpdater.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -8,6 +8,7 @@
import java.util.Map;
import java.util.Map.Entry;
import java.util.Set;
import java.util.concurrent.ExecutionException;
import java.util.stream.Collectors;
import java.util.stream.Stream;
import org.opentripplanner.routing.linking.DisposableEdgeCollection;
Expand Down Expand Up @@ -124,7 +125,7 @@ public String getConfigRef() {
}

@Override
protected void runPolling() {
protected void runPolling() throws InterruptedException, ExecutionException {
LOG.debug("Updating vehicle rental stations from {}", nameForLogging);
if (!source.update()) {
LOG.debug("No updates from {}", nameForLogging);
Expand All @@ -138,7 +139,7 @@ protected void runPolling() {
stations,
geofencingZones
);
saveResultOnGraph.execute(graphWriterRunnable);
saveResultOnGraph.execute(graphWriterRunnable).get();
}

private class VehicleRentalGraphWriterRunnable implements GraphWriterRunnable {
Expand Down
Loading