WIP: parallel raptor loop routes

.pkg

@@ -40,8 +40,8 @@
   commit=d8136b9c6a62db6ce09900ecdeb82bb793096cbd
 [boost]
   url=git@github.com:motis-project/boost.git
-  branch=master
-  commit=73549ebca677fe6214202a1ab580362b4f80e653
+  branch=boost-1.86.0
+  commit=c590212cf9f10badecd5411a8ce11bc1016c74c4
 [oh]
   url=git@github.com:felixguendling/oh.git
   branch=master

.pkg.lock

@@ -1,4 +1,4 @@
-11896755751842712608
+8595625129988798623
 cista 1fb6f1d9bb8ac9fdb4ca648df69695dd819dc07e
 PEGTL 0d37dcf8f02c12a84fdf521973801a5baab78e8f
 res b759b93316afeb529b6cb5b2548b24c41e382fb0
@@ -9,7 +9,7 @@ fmt dc10f83be70ac2873d5f8d1ce317596f1fd318a2
 utl 44ff7ba7e571b02c6c47d3981d722e187b86e576
 oh d9eb908452e808179afeb7954e8beaa5b3626c36
 zlib-ng 68ab3e2d80253ec5dc3c83691d9ff70477b32cd3
-boost 73549ebca677fe6214202a1ab580362b4f80e653
+boost c590212cf9f10badecd5411a8ce11bc1016c74c4
 doctest 832431f2d9dc77e51cc8b364337c186660bf267d
 geo 4f4877c2e7a2caee73bd9f00d955c56835a12f05
 miniz e857234a23de8e5a2d05d266e518220a495db0ce

CMakeLists.txt

@@ -1,5 +1,5 @@
 cmake_minimum_required(VERSION 3.22)
-project(nigiri)
+project(nigiri LANGUAGES C CXX ASM)
 
 if (NOT DEFINED CMAKE_MSVC_RUNTIME_LIBRARY)
   if (NIGIRI_MIMALLOC)
@@ -63,6 +63,7 @@ if ("${CMAKE_CXX_COMPILER_ID}" STREQUAL "Clang")
         -Wno-documentation-unknown-command
         -Wno-duplicate-enum
         -Wno-switch-default
+        -Wno-deprecated-declarations
         -Werror)
 elseif ("${CMAKE_CXX_COMPILER_ID}" STREQUAL "AppleClang")
   set(nigiri-compile-options -Wall -Wextra -Werror)
@@ -100,6 +101,7 @@ target_link_libraries(
   opentelemetry_api
   pugixml
   boost-json
+  Boost::fiber
 )
 target_compile_features(nigiri PUBLIC cxx_std_23)
 target_compile_options(nigiri PRIVATE ${nigiri-compile-options})

include/nigiri/routing/raptor/raptor.h

@@ -2,6 +2,10 @@
 
 #include <cassert>
 
+#include "boost/fiber/all.hpp"
+
+#include "cista/atomic.h"
+
 #include "nigiri/common/delta_t.h"
 #include "nigiri/common/linear_lower_bound.h"
 #include "nigiri/routing/journey.h"
@@ -46,17 +50,22 @@ struct raptor_stats {
   std::uint64_t route_update_prevented_by_lower_bound_{0ULL};
 };
 
+enum class exec { kSequential, kParallel };
 enum class search_mode { kOneToOne, kOneToAll };
 
+constexpr auto const kNThreads = 4U;
+
 template <direction SearchDir,
           bool Rt,
           via_offset_t Vias,
-          search_mode SearchMode>
+          search_mode SearchMode,
+          exec Exec = exec::kParallel>
 struct raptor {
   using algo_state_t = raptor_state;
   using algo_stats_t = raptor_stats;
 
-  static constexpr bool kUseLowerBounds = true;
+  static constexpr auto const Atomic = Exec == exec::kParallel;
+  static constexpr auto const kUseLowerBounds = true;
   static constexpr auto const kFwd = (SearchDir == direction::kForward);
   static constexpr auto const kBwd = (SearchDir == direction::kBackward);
   static constexpr auto const kInvalid = kInvalidDelta<SearchDir>;
@@ -70,6 +79,13 @@ struct raptor {
     return a;
   }();
 
+  static inline void update(delta_t& x, delta_t const new_value) {
+    if constexpr (Atomic) {
+      kFwd ? cista::fetch_min(x, new_value) : cista::fetch_max(x, new_value);
+    } else {
+      x = new_value;
+    }
+  }
   static bool is_better(auto a, auto b) { return kFwd ? a < b : a > b; }
   static bool is_better_or_eq(auto a, auto b) { return kFwd ? a <= b : a >= b; }
   static auto get_best(auto a, auto b) { return is_better(a, b) ? a : b; }
@@ -295,13 +311,12 @@ struct raptor {
 
   template <bool WithClaszFilter, bool WithBikeFilter, bool WithCarFilter>
   bool loop_routes(unsigned const k) {
-    auto any_marked = false;
-    state_.route_mark_.for_each_set_bit([&](auto const r_idx) {
-      auto const r = route_idx_t{r_idx};
+    auto const loop_route = [&](route_idx_t const r) {
+      auto const r_idx = to_idx(r);
 
       if constexpr (WithClaszFilter) {
         if (!is_allowed(allowed_claszes_, tt_.route_clasz_[r])) {
-          return;
+          return false;
         }
       }
 
@@ -313,7 +328,7 @@ struct raptor {
           auto const bikes_allowed_on_some_sections =
               tt_.route_bikes_allowed_.test(r_idx * 2 + 1);
           if (!bikes_allowed_on_some_sections) {
-            return;
+            return false;
           }
           section_bike_filter = true;
         }
@@ -327,22 +342,49 @@ struct raptor {
           auto const cars_allowed_on_some_sections =
               tt_.route_cars_allowed_.test(r_idx * 2 + 1);
           if (!cars_allowed_on_some_sections) {
-            return;
+            return false;
           }
           section_car_filter = true;
         }
       }
 
-      ++stats_.n_routes_visited_;
+      // ++stats_.n_routes_visited_;
       trace("┊ ├k={} updating route {}\n", k, r);
-      any_marked |=
-          section_bike_filter
-              ? (section_car_filter ? update_route<true, true>(k, r)
-                                    : update_route<true, false>(k, r))
-              : (section_car_filter ? update_route<false, true>(k, r)
-                                    : update_route<false, false>(k, r));
-    });
-    return any_marked;
+      return section_bike_filter
+                 ? (section_car_filter ? update_route<true, true>(k, r)
+                                       : update_route<true, false>(k, r))
+                 : (section_car_filter ? update_route<false, true>(k, r)
+                                       : update_route<false, false>(k, r));
+    };
+
+    if constexpr (Exec == exec::kParallel) {
+      auto barrier = std::make_shared<boost::fibers::barrier>(kNThreads + 1U);
+      auto any_marked = std::atomic_bool{false};
+      auto next = std::atomic_size_t{0U};
+      for (auto i = 0U; i != kNThreads; ++i) {
+        boost::fibers::fiber{[&, barrier]() {
+          auto next_route_idx = std::optional<std::size_t>{};
+          while ((next_route_idx = state_.route_mark_.get_next(next))
+                     .has_value()) {
+            assert(*next_route_idx < tt_.n_routes());
+            auto const marked = loop_route(route_idx_t{
+                static_cast<route_idx_t::value_t>(*next_route_idx)});
+            if (marked) {
+              any_marked.store(true);
+            }
+          }
+          barrier->wait();
+        }}.detach();
+      }
+      barrier->wait();
+      return any_marked.load();
+    } else {
+      auto any_marked = false;
+      state_.route_mark_.for_each_set_bit([&](std::size_t const r) {
+        any_marked |= loop_route(route_idx_t{r});
+      });
+      return any_marked;
+    }
   }
 
   template <bool WithClaszFilter, bool WithBikeFilter, bool WithCarFilter>
@@ -969,9 +1011,9 @@ struct raptor {
                 location{tt_, stp.location_idx()});
 
             ++stats_.n_earliest_arrival_updated_by_route_;
-            tmp_[l_idx][target_v] =
-                get_best(by_transport, tmp_[l_idx][target_v]);
-            state_.station_mark_.set(l_idx, true);
+            update(tmp_[l_idx][target_v],
+                   get_best(by_transport, tmp_[l_idx][target_v]));
+            state_.station_mark_.set<Atomic>(l_idx, true);
             current_best[v] = by_transport;
             any_marked = true;
           } else {
@@ -1029,8 +1071,9 @@ struct raptor {
                   location{tt_, stp.location_idx()});
 
               ++stats_.n_earliest_arrival_updated_by_route_;
-              tmp_[l_idx][dest_v] = get_best(by_transport, tmp_[l_idx][dest_v]);
-              state_.station_mark_.set(l_idx, true);
+              update(tmp_[l_idx][dest_v],
+                     get_best(by_transport, tmp_[l_idx][dest_v]));
+              state_.station_mark_.set<Atomic>(l_idx, true);
               any_marked = true;
             }
           }