Skip to content

perf(gtfs): resolve lock contention in spatial queries #730

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

Open
AhmedAlian7 wants to merge 1 commit into
base: main
Choose a base branch
from
Open

perf(gtfs): resolve lock contention in spatial queries #730

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
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
5 changes: 4 additions & 1 deletion internal/gtfs/gtfs_manager.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -452,7 +452,7 @@ type stopWithDistance struct {

// GetStopsForLocation retrieves stops near a given location using the spatial index.
// It supports filtering by route types and querying for specific stop codes.
// IMPORTANT: Caller must hold manager.RLock() before calling this method.
// This method manages its own locking internally; callers must NOT hold any Manager locks.
func (manager *Manager) GetStopsForLocation(
ctx context.Context,
lat, lon, radius, latSpan, lonSpan float64,
Expand Down Expand Up @@ -486,7 +486,10 @@ func (manager *Manager) GetStopsForLocation(
return []gtfsdb.Stop{}
}

// Acquire the read lock only for the in-memory spatial search.
manager.staticMutex.RLock()
dbStops := queryStopsInBounds(manager.stopSpatialIndex, bounds)
manager.staticMutex.RUnlock()

for _, dbStop := range dbStops {
if ctx.Err() != nil {
Expand Down
13 changes: 4 additions & 9 deletions internal/restapi/arrivals_and_departures_for_stop_handler_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1044,19 +1044,16 @@ func TestGetNearbyStopIDs_UsesResolvedAgency(t *testing.T) {
// The RABA agency ID is "25".
rabaAgencyID := "25"

// GetStopsForLocation requires the caller to hold RLock.
api.GtfsManager.RLock()
// GetStopsForLocation manages its own locking internally.
stops := api.GtfsManager.GetStopsForLocation(ctx, 40.589123, -122.390830, 2000, 0, 0, "", 10, false, []int{}, mockClock.Now())
api.GtfsManager.RUnlock()
require.NotEmpty(t, stops, "precondition: RABA should have stops near Redding, CA")

currentStop := stops[0]

// Call getNearbyStopIDs with a wrong fallback agency.
// If batch resolution works, nearby stops should use "25", not the fallback.
api.GtfsManager.RLock()
// getNearbyStopIDs calls GetStopsForLocation which manages its own locking.
result := getNearbyStopIDs(api, ctx, currentStop.Lat, currentStop.Lon, currentStop.ID, "WrongFallbackAgency")
api.GtfsManager.RUnlock()
require.NotEmpty(t, result, "should find nearby stops")

for _, combinedID := range result {
Expand All @@ -1075,16 +1072,14 @@ func TestGetNearbyStopIDs_ExcludesCurrentStop(t *testing.T) {

ctx := context.Background()

api.GtfsManager.RLock()
// GetStopsForLocation manages its own locking internally.
stops := api.GtfsManager.GetStopsForLocation(ctx, 40.589123, -122.390830, 2000, 0, 0, "", 10, false, []int{}, mockClock.Now())
api.GtfsManager.RUnlock()
require.NotEmpty(t, stops)

currentStop := stops[0]

api.GtfsManager.RLock()
// getNearbyStopIDs calls GetStopsForLocation which manages its own locking.
result := getNearbyStopIDs(api, ctx, currentStop.Lat, currentStop.Lon, currentStop.ID, "25")
api.GtfsManager.RUnlock()

for _, combinedID := range result {
_, codeID, _ := utils.ExtractAgencyIDAndCodeID(combinedID)
Expand Down
17 changes: 12 additions & 5 deletions internal/restapi/routes_for_location_handler.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -59,9 +59,7 @@ func (api *RestAPI) routesForLocationHandler(w http.ResponseWriter, r *http.Requ
return
}

api.GtfsManager.RLock()
defer api.GtfsManager.RUnlock()

// GetStopsForLocation manages its own locking internally.
stops := api.GtfsManager.GetStopsForLocation(ctx, lat, lon, radius, latSpan, lonSpan, query, maxCount, true, nil, time.Time{})

var results = []models.Route{}
Expand All @@ -76,10 +74,15 @@ func (api *RestAPI) routesForLocationHandler(w http.ResponseWriter, r *http.Requ

if len(stopIDs) == 0 {
// Return empty response if no stops found
// Scope the read lock narrowly for in-memory data access only.
api.GtfsManager.RLock()
agencies := utils.FilterAgencies(api.GtfsManager.GetAgencies(), agencyIDs)
outOfBounds := checkIfOutOfBounds(api, lat, lon, latSpan, lonSpan, radius)
api.GtfsManager.RUnlock()

references := models.NewEmptyReferences()
references.Agencies = agencies
response := models.NewListResponseWithRange(results, *references, checkIfOutOfBounds(api, lat, lon, latSpan, lonSpan, radius), api.Clock, false)
response := models.NewListResponseWithRange(results, *references, outOfBounds, api.Clock, false)
api.sendResponse(w, r, response)
return
}
Expand Down Expand Up @@ -133,7 +136,11 @@ func (api *RestAPI) routesForLocationHandler(w http.ResponseWriter, r *http.Requ
return
}

// Scope the read lock narrowly for in-memory data access only.
api.GtfsManager.RLock()
agencies := utils.FilterAgencies(api.GtfsManager.GetAgencies(), agencyIDs)
outOfBounds := checkIfOutOfBounds(api, lat, lon, latSpan, lonSpan, radius)
api.GtfsManager.RUnlock()

// Populate situation references for alerts affecting the returned routes
alerts := api.collectAlertsForRoutes(resultRawRouteIDs)
Expand All @@ -143,7 +150,7 @@ func (api *RestAPI) routesForLocationHandler(w http.ResponseWriter, r *http.Requ
references.Agencies = agencies
references.Situations = situations

response := models.NewListResponseWithRange(results, *references, checkIfOutOfBounds(api, lat, lon, latSpan, lonSpan, radius), api.Clock, isLimitExceeded)
response := models.NewListResponseWithRange(results, *references, outOfBounds, api.Clock, isLimitExceeded)
api.sendResponse(w, r, response)
}

Expand Down
18 changes: 13 additions & 5 deletions internal/restapi/stops_for_location_handler.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -101,9 +101,7 @@ func (api *RestAPI) stopsForLocationHandler(w http.ResponseWriter, r *http.Reque
return
}

api.GtfsManager.RLock()
defer api.GtfsManager.RUnlock()

// GetStopsForLocation manages its own locking internally.
stops := api.GtfsManager.GetStopsForLocation(ctx, lat, lon, radius, latSpan, lonSpan, query, maxCount, false, routeTypes, queryTime)

// Referenced Java code: "here we sort by distance for possible truncation, but later it will be re-sorted by stopId"
Expand All @@ -124,7 +122,12 @@ func (api *RestAPI) stopsForLocationHandler(w http.ResponseWriter, r *http.Reque

if len(stopIDs) == 0 {
// Return empty response if no stops found
// Scope the read lock narrowly for in-memory data access only.
api.GtfsManager.RLock()
agencies := utils.FilterAgencies(api.GtfsManager.GetAgencies(), agencyIDs)
outOfBounds := checkIfOutOfBounds(api, lat, lon, latSpan, lonSpan, radius)
api.GtfsManager.RUnlock()

if agencies == nil {
agencies = []models.AgencyReference{}
}
Expand All @@ -138,7 +141,7 @@ func (api *RestAPI) stopsForLocationHandler(w http.ResponseWriter, r *http.Reque
references.Agencies = agencies
references.Routes = routes

response := models.NewListResponseWithRange(results, *references, checkIfOutOfBounds(api, lat, lon, latSpan, lonSpan, radius), api.Clock, false)
response := models.NewListResponseWithRange(results, *references, outOfBounds, api.Clock, false)
api.sendResponse(w, r, response)
return
}
Expand Down Expand Up @@ -249,7 +252,12 @@ func (api *RestAPI) stopsForLocationHandler(w http.ResponseWriter, r *http.Reque
return
}

// Scope the read lock narrowly for in-memory data access only.
api.GtfsManager.RLock()
agencies := utils.FilterAgencies(api.GtfsManager.GetAgencies(), agencyIDs)
outOfBounds := checkIfOutOfBounds(api, lat, lon, latSpan, lonSpan, radius)
api.GtfsManager.RUnlock()

routes := utils.FilterRoutes(api.GtfsManager.GtfsDB.Queries, ctx, routeIDs)

if agencies == nil {
Expand All @@ -268,6 +276,6 @@ func (api *RestAPI) stopsForLocationHandler(w http.ResponseWriter, r *http.Reque
references.Routes = routes
references.Situations = situations

response := models.NewListResponseWithRange(results, *references, checkIfOutOfBounds(api, lat, lon, latSpan, lonSpan, radius), api.Clock, isLimitExceeded)
response := models.NewListResponseWithRange(results, *references, outOfBounds, api.Clock, isLimitExceeded)
api.sendResponse(w, r, response)
}
15 changes: 11 additions & 4 deletions internal/restapi/trips_for_location_handler.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -20,9 +20,6 @@ import (
func (api *RestAPI) tripsForLocationHandler(w http.ResponseWriter, r *http.Request) {
ctx := r.Context()

api.GtfsManager.RLock()
defer api.GtfsManager.RUnlock()

lat, lon, latSpan, lonSpan, includeTrip, includeSchedule, currentLocation, todayMidnight, serviceDate, fieldErrors, err := api.parseAndValidateRequest(r)
if fieldErrors != nil {
api.validationErrorResponse(w, r, fieldErrors)
Expand All @@ -39,6 +36,7 @@ func (api *RestAPI) tripsForLocationHandler(w http.ResponseWriter, r *http.Reque
// Note: re-deriving currentTime here rather than returning it from parseAndValidateRequest(line: 150)
currentTime := api.Clock.Now().In(currentLocation)

// GetStopsForLocation manages its own locking internally.
stops := api.GtfsManager.GetStopsForLocation(ctx, lat, lon, -1, latSpan, lonSpan, "", 100, false, []int{}, api.Clock.Now())
stopIDs := extractStopIDs(stops)
stopTimes, err := api.GtfsManager.GtfsDB.Queries.GetStopTimesByStopIDs(ctx, stopIDs)
Expand Down Expand Up @@ -119,7 +117,12 @@ func (api *RestAPI) tripsForLocationHandler(w http.ResponseWriter, r *http.Reque
Stops: stops,
Trips: result,
})
response := models.NewListResponseWithRange(result, references, checkIfOutOfBounds(api, lat, lon, latSpan, lonSpan, 0), api.Clock, false)
// Scope the read lock narrowly for in-memory data access only (GetRegionBounds).
api.GtfsManager.RLock()
outOfBounds := checkIfOutOfBounds(api, lat, lon, latSpan, lonSpan, 0)
api.GtfsManager.RUnlock()

response := models.NewListResponseWithRange(result, references, outOfBounds, api.Clock, false)
api.sendResponse(w, r, response)
}

Expand All @@ -141,7 +144,11 @@ func (api *RestAPI) parseAndValidateRequest(r *http.Request) (
includeTrip = queryParams.Get("includeTrip") == "true"
includeSchedule = queryParams.Get("includeSchedule") == "true"

// Scope the read lock narrowly for in-memory data access only (GetAgencies).
api.GtfsManager.RLock()
agencies := api.GtfsManager.GetAgencies()
api.GtfsManager.RUnlock()

if len(agencies) == 0 {
return 0, 0, 0, 0, false, false, nil, time.Time{}, time.Time{}, nil, errors.New("no agencies configured in GTFS manager")
}
Expand Down
Loading