Merge pull request #6326 from Jnction/parallel-raptor-cache

Generate Raptor transfer cache in parallel

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

@@ -5,12 +5,19 @@
 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;
 
 public class RaptorTransferIndex {
 
+  private enum RequestSource {
+    SETUP,
+    REQUEST_SCOPE,
+  }
+
   private final List<RaptorTransfer>[] forwardTransfers;
 
   private final List<RaptorTransfer>[] reversedTransfers;
@@ -24,19 +31,47 @@ public RaptorTransferIndex(
     this.reversedTransfers = reversedTransfers.stream().map(List::copyOf).toArray(List[]::new);
   }
 
-  public static RaptorTransferIndex create(
+  /**
+   * Create an index for a route request configured in router-config.json
+   */
+  public static RaptorTransferIndex createInitialSetup(
+    List<List<Transfer>> transfersByStopIndex,
+    StreetSearchRequest request
+  ) {
+    return create(transfersByStopIndex, request, RequestSource.SETUP);
+  }
+
+  /**
+   * Create an index for a route request originated from the client
+   */
+  public static RaptorTransferIndex createRequestScope(
     List<List<Transfer>> transfersByStopIndex,
     StreetSearchRequest request
+  ) {
+    return create(transfersByStopIndex, request, RequestSource.REQUEST_SCOPE);
+  }
+
+  private static RaptorTransferIndex create(
+    List<List<Transfer>> transfersByStopIndex,
+    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.SETUP || 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 +84,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/request/RaptorRequestTransferCache.java

@@ -36,7 +36,7 @@ public LoadingCache<CacheKey, RaptorTransferIndex> getTransferCache() {
 
   public void put(List<List<Transfer>> transfersByStopIndex, RouteRequest request) {
     final CacheKey cacheKey = new CacheKey(transfersByStopIndex, request);
-    final RaptorTransferIndex raptorTransferIndex = RaptorTransferIndex.create(
+    final RaptorTransferIndex raptorTransferIndex = RaptorTransferIndex.createInitialSetup(
       transfersByStopIndex,
       cacheKey.request
     );
@@ -58,7 +58,10 @@ 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.createRequestScope(
+          cacheKey.transfersByStopIndex,
+          cacheKey.request
+        );
       }
     };
   }