smp: add a function that barriers memory prefault work

include/seastar/core/reactor.hh

@@ -374,14 +374,14 @@ private:
         std::atomic<uint64_t> _pending_signals;
         std::unordered_map<int, signal_handler> _signal_handlers;
     };
-
     signals _signals;
     std::unique_ptr<thread_pool> _thread_pool;
+
     friend class thread_pool;
     friend class thread_context;
     friend class internal::cpu_stall_detector;
-
     friend void handle_signal(int signo, noncopyable_function<void ()>&& handler, bool once);
+    friend void barrier_memory_prefault();
 
     uint64_t pending_task_count() const;
     void run_tasks(task_queue& tq);

include/seastar/core/smp.hh

@@ -335,6 +335,7 @@ public:
     void cleanup_cpu();
     void arrive_at_event_loop_end();
     void join_all();
+    void join_memory_prefault();
     static bool main_thread() { return std::this_thread::get_id() == _tmain; }
 
     /// Runs a function on a remote core.
@@ -484,6 +485,13 @@ public:
     static unsigned count;
 };
 
+/// Barriers memory prefault background work. 
+/// This function returns after all memory prefault work is finished. 
+/// It is safe to call from multiple threads.
+/// 
+/// Note: This is a blocking operation, so it should be used with caution.
+void barrier_memory_prefault();
+
 SEASTAR_MODULE_EXPORT_END
 
 }

src/core/prefault.hh

@@ -36,11 +36,16 @@ namespace seastar::internal {
 class memory_prefaulter {
     std::atomic<bool> _stop_request = false;
     std::vector<posix_thread> _worker_threads;
+
+    std::mutex join_mtx;
+    bool joined = false;
+
     // Keep this in object scope to avoid allocating in worker thread
     std::unordered_map<unsigned, std::vector<memory::internal::memory_range>> _layout_by_node_id;
 public:
     explicit memory_prefaulter(const resource::resources& res, memory::internal::numa_layout layout);
     ~memory_prefaulter();
+    void join_all();
 private:
     void work(std::vector<memory::internal::memory_range>& ranges, size_t page_size, std::optional<size_t> huge_page_size_opt);
 };

src/core/reactor.cc

@@ -4637,6 +4637,10 @@ void smp::configure(const smp_options& smp_opts, const reactor_options& reactor_
     }
 }
 
+void barrier_memory_prefault() {
+    engine()._smp->join_memory_prefault();
+}
+
 bool smp::poll_queues() {
     size_t got = 0;
     for (unsigned i = 0; i < count; i++) {

src/core/smp.cc

@@ -179,6 +179,13 @@ smp::setup_prefaulter(const seastar::resource::resources& res, seastar::memory::
 #endif
 }
 
+void
+smp::join_memory_prefault() {
+    if (_prefaulter) {
+        _prefaulter->join_all();
+    }
+}
+
 static
 std::optional<size_t>
 get_huge_page_size() {
@@ -224,11 +231,20 @@ internal::memory_prefaulter::memory_prefaulter(const resource::resources& res, m
     }
 }
 
+void
+internal::memory_prefaulter::join_all() {
+    const std::lock_guard<std::mutex> lock(join_mtx);
+    if (!joined) {
+        for (auto& t : _worker_threads) {
+            t.join();
+        }
+        joined = true;
+    }
+}
+
 internal::memory_prefaulter::~memory_prefaulter() {
     _stop_request.store(true, std::memory_order_relaxed);
-    for (auto& t : _worker_threads) {
-        t.join();
-    }
+    join_all();
 }
 
 void