Merge remote-tracking branch 'jnction/parallel-raptor-cache' into dev…

…-2.x

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.street.search.request.StreetSearchRequest;
 
@@ -23,9 +25,16 @@ public RaptorTransferIndex(
     this.reversedTransfers = reversedTransfers.stream().map(List::copyOf).toArray(List[]::new);
   }
 
+  /**
+   * Create an index to be put into the transfer cache
+   *
+   * @param isRuntimeRequest true if the request originates from the client during the runtime,
+   *                         false if the request comes from transferCacheRequests in router-config.json
+   */
   public static RaptorTransferIndex create(
     List<List<Transfer>> transfersByStopIndex,
-    StreetSearchRequest request
+    StreetSearchRequest request,
+    boolean isRuntimeRequest
   ) {
     var forwardTransfers = new ArrayList<List<RaptorTransfer>>(transfersByStopIndex.size());
     var reversedTransfers = new ArrayList<List<RaptorTransfer>>(transfersByStopIndex.size());
@@ -35,7 +44,13 @@ public static RaptorTransferIndex create(
       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 (!isRuntimeRequest || 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
@@ -47,15 +62,18 @@ public static RaptorTransferIndex create(
         )
         .values();
 
+      // 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

@@ -38,7 +38,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,
+      false
     );
 
     LOG.info("Initializing cache with request: {}", cacheKey.options);
@@ -58,7 +59,7 @@ 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, true);
       }
     };
   }