Skip to content
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

Make all polling updaters wait for graph update finish #6262

Open
wants to merge 4 commits into
base: dev-2.x
Choose a base branch
from
+36 −34
Open

Make all polling updaters wait for graph update finish #6262

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
4 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
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.

Comment on lines +85 to +87
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. 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.


lastTimestamp = feedTimestamp;
} catch (Exception e) {
LOG.error("Failed to process GTFS-RT Alerts 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.service.vehicleparking.VehicleParkingRepository;
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