Generate Raptor transfer cache in parallel

application/src/main/java/org/opentripplanner/routing/algorithm/raptoradapter/transit/RaptorTransferIndex.java

@@ -5,6 +5,8 @@
 import java.util.ArrayList;
 import java.util.List;
 import java.util.function.Function;
+import java.util.stream.IntStream;
+import org.opentripplanner.framework.application.OTPFeature;
 import org.opentripplanner.raptor.api.model.RaptorTransfer;
 import org.opentripplanner.routing.api.request.StreetMode;
 import org.opentripplanner.street.search.request.StreetSearchRequest;
@@ -24,19 +26,30 @@ public RaptorTransferIndex(
     this.reversedTransfers = reversedTransfers.stream().map(List::copyOf).toArray(List[]::new);
   }
 
+  /**
+   * Create an index to be put into the transfer cache
+   */
   public static RaptorTransferIndex create(
     List<List<Transfer>> transfersByStopIndex,
-    StreetSearchRequest request
+    StreetSearchRequest request,
+    RequestSource requestSource
   ) {
     var forwardTransfers = new ArrayList<List<RaptorTransfer>>(transfersByStopIndex.size());
     var reversedTransfers = new ArrayList<List<RaptorTransfer>>(transfersByStopIndex.size());
     StreetMode mode = request.mode();
 
     for (int i = 0; i < transfersByStopIndex.size(); i++) {
+      forwardTransfers.add(new ArrayList<>());
       reversedTransfers.add(new ArrayList<>());
     }
 
-    for (int fromStop = 0; fromStop < transfersByStopIndex.size(); fromStop++) {
+    var stopIndices = IntStream.range(0, transfersByStopIndex.size());
+    // we want to always parallelize the cache building during the startup
+    // and only parallelize during runtime requests if the feature flag is on
+    if (requestSource == RequestSource.CONFIG || OTPFeature.ParallelRouting.isOn()) {
+      stopIndices = stopIndices.parallel();
+    }
+    stopIndices.forEach(fromStop -> {
       // The transfers are filtered so that there is only one possible directional transfer
       // for a stop pair.
       var transfers = transfersByStopIndex
@@ -49,15 +62,18 @@ public static RaptorTransferIndex create(
         )
         .values();
 
-      forwardTransfers.add(new ArrayList<>(transfers));
+      // forwardTransfers is not modified here, and no two threads will access the same element
+      // in it, so this is still thread safe.
+      forwardTransfers.get(fromStop).addAll(transfers);
+    });
 
-      for (RaptorTransfer forwardTransfer : transfers) {
+    for (int fromStop = 0; fromStop < transfersByStopIndex.size(); fromStop++) {
+      for (var forwardTransfer : forwardTransfers.get(fromStop)) {
         reversedTransfers
           .get(forwardTransfer.stop())
           .add(DefaultRaptorTransfer.reverseOf(fromStop, forwardTransfer));
       }
     }
-
     return new RaptorTransferIndex(forwardTransfers, reversedTransfers);
   }
 

application/src/main/java/org/opentripplanner/routing/algorithm/raptoradapter/transit/RequestSource.java

@@ -0,0 +1,12 @@
+package org.opentripplanner.routing.algorithm.raptoradapter.transit;
+
+public enum RequestSource {
+  /**
+   * The request comes from transferCacheRequests in router-config.json
+   */
+  CONFIG,
+  /**
+   * The request comes from a client routing request
+   */
+  RUNTIME,
+}

application/src/main/java/org/opentripplanner/routing/algorithm/raptoradapter/transit/request/RaptorRequestTransferCache.java

@@ -7,6 +7,7 @@
 import java.util.Objects;
 import java.util.concurrent.ExecutionException;
 import org.opentripplanner.routing.algorithm.raptoradapter.transit.RaptorTransferIndex;
+import org.opentripplanner.routing.algorithm.raptoradapter.transit.RequestSource;
 import org.opentripplanner.routing.algorithm.raptoradapter.transit.Transfer;
 import org.opentripplanner.routing.api.request.RouteRequest;
 import org.opentripplanner.routing.api.request.StreetMode;
@@ -38,7 +39,8 @@ public void put(List<List<Transfer>> transfersByStopIndex, RouteRequest request)
     final CacheKey cacheKey = new CacheKey(transfersByStopIndex, request);
     final RaptorTransferIndex raptorTransferIndex = RaptorTransferIndex.create(
       transfersByStopIndex,
-      cacheKey.request
+      cacheKey.request,
+      RequestSource.CONFIG
     );
 
     LOG.info("Initializing cache with request: {}", cacheKey.options);
@@ -58,7 +60,11 @@ private CacheLoader<CacheKey, RaptorTransferIndex> cacheLoader() {
       @Override
       public RaptorTransferIndex load(CacheKey cacheKey) {
         LOG.info("Adding runtime request to cache: {}", cacheKey.options);
-        return RaptorTransferIndex.create(cacheKey.transfersByStopIndex, cacheKey.request);
+        return RaptorTransferIndex.create(
+          cacheKey.transfersByStopIndex,
+          cacheKey.request,
+          RequestSource.RUNTIME
+        );
       }
     };
   }