Async mmap population

src/common/util/CMakeLists.txt

@@ -70,6 +70,7 @@ target_sources(${TARGET_NAME}
         ${CMAKE_CURRENT_SOURCE_DIR}/src/file_util.cpp
         ${CMAKE_CURRENT_SOURCE_DIR}/src/log.cpp
         ${CMAKE_CURRENT_SOURCE_DIR}/src/memory.cpp
+        ${CMAKE_CURRENT_SOURCE_DIR}/src/memory_prefetch.hpp
         ${CMAKE_CURRENT_SOURCE_DIR}/src/native_stream.cpp
         ${CMAKE_CURRENT_SOURCE_DIR}/src/native_streambuf.cpp
         ${CMAKE_CURRENT_SOURCE_DIR}/src/parallel_read_streambuf.cpp
@@ -94,6 +95,7 @@ if (WIN32)
 endif()
 
 target_include_directories(${TARGET_NAME} PUBLIC $<BUILD_INTERFACE:${UTIL_INCLUDE_DIR}>)
+target_include_directories(${TARGET_NAME} PRIVATE ${CMAKE_CURRENT_SOURCE_DIR}/src)
 
 ov_add_clang_format_target(${TARGET_NAME}_clang FOR_TARGETS ${TARGET_NAME})
 ov_ncc_naming_style(FOR_TARGET ${TARGET_NAME}

src/common/util/include/openvino/util/memory.hpp

@@ -116,18 +116,4 @@ void vm_decommit(void* ptr, size_t size) noexcept;
  * @pre  ptr != nullptr && size > 0; violated preconditions are a programming error (assert fires in debug).
  */
 void vm_release(void* ptr, size_t size) noexcept;
-
-/**
- * @brief Pre-fetch a committed VM range into physical memory.
- *
- * Works with both anonymous (@ref vm_commit) and file-backed (mmap) regions.
- *
- * @param ptr         Base address of the range. Must be page-aligned.
- * @param size        Number of bytes to pre-fetch. Must be a multiple of the system page size.
- * @param num_threads Strategy selector:
- *                    - @c 0 (default) → OS advisory hint (async, low overhead).
- *                    - @c N >= 1      → parallel touch with N threads (synchronous).
- */
-void vm_prefetch(void* ptr, size_t size, size_t num_threads = 0) noexcept;
-
 }  // namespace ov::util

src/common/util/include/openvino/util/mmap_object.hpp

@@ -16,6 +16,7 @@
 #include <string>
 
 #include "openvino/util/file_util.hpp"
+#include "openvino/util/memory.hpp"
 
 namespace ov {
 
@@ -48,6 +49,17 @@ class MappedMemory {
      *               mapping when set to auto_size.
      */
     virtual void hint_prefetch(size_t offset = 0, size_t size = auto_size) = 0;
+
+    /**
+     * @brief Asynchronous variant of @ref hint_prefetch: starts populating the given region in the
+     * background and returns immediately. Any background work is joined before this object is
+     * destroyed. The default implementation is a no-op.
+     *
+     * @param offset Offset within the mapping where prefetching starts.
+     * @param size   Number of bytes to prefetch. Defaults to the rest of the
+     *               mapping when set to auto_size.
+     */
+    virtual void hint_prefetch_async(size_t offset = 0, size_t size = auto_size) {}
 };
 
 /**

src/common/util/include/openvino/util/parallel_io.hpp

@@ -9,6 +9,7 @@
 
 #pragma once
 
+#include <algorithm>
 #include <cstddef>
 #include <filesystem>
 
@@ -25,6 +26,15 @@ inline constexpr size_t default_parallel_io_min_chunk = 2UL * 1024 * 1024;  ///<
  */
 inline constexpr size_t default_parallel_io_prefetch_cap = 32UL * 1024 * 1024;
 
+/**
+ * @brief Number of parallel chunks a @p size byte job should be split into: at least one, at most
+ * @p max_chunks, targeting roughly one chunk per @p min_chunk bytes.
+ */
+constexpr size_t split_chunk_count(size_t size, size_t min_chunk, size_t max_chunks) noexcept {
+    const size_t by_size = (min_chunk == 0) ? max_chunks : size / min_chunk;
+    return std::max<size_t>(1, std::min<size_t>(max_chunks, by_size));
+}
+
 /**
  * @brief Open a file for reading and retrieve its size.
  *

src/common/util/src/memory.cpp

@@ -5,41 +5,18 @@
 #include "openvino/util/memory.hpp"
 
 #include <algorithm>
-#include <cassert>
 #include <cstdint>
 #include <thread>
 #include <vector>
 
+#include "memory_prefetch.hpp"
 #include "openvino/util/math_util.hpp"
 #include "openvino/util/mmap_object.hpp"
 
 namespace ov::util {
 void populate_pages(void* ptr, size_t size, size_t num_threads) noexcept;
 
-namespace {
-
-/**
- * @brief Functor that touches one byte per page over [m_begin, m_end) to force the pages resident.
- *
- * The volatile accumulator prevents the compiler from optimizing the read loop away.
- */
-struct PageToucher {
-    const uint8_t* m_begin;
-    const uint8_t* m_end;
-    const size_t m_page_size;
-
-    void operator()() const noexcept {
-        volatile uint8_t local = 0;  // prevents the compiler from optimizing the loop away
-        for (auto begin = m_begin; begin < m_end; begin += m_page_size) {
-            local += *begin;
-        }
-    }
-};
-
-}  // namespace
-
 void populate_pages(void* ptr, size_t size, size_t num_threads) noexcept {
-    // ptr and size are guaranteed page-aligned by vm_prefetch's precondition.
     const auto page_size = static_cast<size_t>(get_system_page_size());
     const auto chunk_size = std::max<size_t>(align_size_up(size / num_threads, page_size), one_mib);
 

src/common/util/src/memory_prefetch.hpp

@@ -0,0 +1,157 @@
+// Copyright (C) 2018-2026 Intel Corporation
+// SPDX-License-Identifier: Apache-2.0
+//
+
+#pragma once
+
+#include <algorithm>
+#include <cstddef>
+#include <cstdint>
+#include <future>
+#include <thread>
+#include <vector>
+
+#include "openvino/util/memory.hpp"
+#include "openvino/util/mmap_object.hpp"
+#include "openvino/util/parallel_io.hpp"
+
+namespace ov::util {
+
+// Touches one byte per page over [m_begin, m_end) to force the pages resident. The volatile
+// accumulator keeps the compiler from eliminating the read loop.
+struct PageToucher {
+    const uint8_t* m_begin;
+    const uint8_t* m_end;
+    const size_t m_page_size;
+
+    void operator()() const noexcept {
+        volatile uint8_t local = 0;
+        for (auto begin = m_begin; begin < m_end; begin += m_page_size) {
+            local += *begin;
+        }
+    }
+};
+
+class PrefetchToken;
+
+/**
+ * @brief Pre-fetches a page-aligned, committed VM range into physical memory, blocking until every
+ * page is resident.
+ *
+ * @param ptr          Page-aligned base address of the range.
+ * @param size         Multiple of the system page size.
+ * @param num_threads  Number of population jobs to split the range into; @c 0 requests only a
+ *                     lightweight advisory OS hint instead of touching pages.
+ */
+void vm_prefetch(void* ptr, size_t size, size_t num_threads) noexcept;
+
+/**
+ * @brief Asynchronous variant of @ref vm_prefetch: submits page-population to the shared pool and
+ * returns immediately with a @ref PrefetchToken to wait on. Returns an empty token if the work
+ * could not be scheduled.
+ */
+PrefetchToken vm_prefetch_async(void* ptr, size_t size) noexcept;
+
+/**
+ * @brief Move-only RAII handle for background page-population started by @ref vm_prefetch_async.
+ *
+ * Destruction (or an explicit @ref wait) joins the outstanding work, so nothing is ever left
+ * running uncontrolled. The token does not keep the populated memory alive: the caller must keep
+ * that memory valid until the token completes, is destroyed, or its futures are @ref detach "detached".
+ */
+class PrefetchToken {
+public:
+    PrefetchToken() noexcept = default;
+    explicit PrefetchToken(std::vector<std::future<void>>&& tasks) noexcept : m_tasks(std::move(tasks)) {}
+
+    PrefetchToken(const PrefetchToken&) = delete;
+    PrefetchToken& operator=(const PrefetchToken&) = delete;
+    PrefetchToken(PrefetchToken&&) noexcept = default;
+
+    PrefetchToken& operator=(PrefetchToken&& other) noexcept {
+        if (this != &other) {
+            wait();
+            m_tasks = std::move(other.m_tasks);
+        }
+        return *this;
+    }
+
+    ~PrefetchToken() {
+        wait();
+    }
+
+    void wait() noexcept {
+        for (auto& task : m_tasks) {
+            if (task.valid()) {
+                task.wait();
+            }
+        }
+        m_tasks.clear();
+    }
+
+    std::vector<std::future<void>> detach() noexcept {
+        auto tasks = std::move(m_tasks);
+        m_tasks.clear();
+        return tasks;
+    }
+
+    bool valid() const noexcept {
+        return !m_tasks.empty();
+    }
+
+    explicit operator bool() const noexcept {
+        return valid();
+    }
+
+private:
+    std::vector<std::future<void>> m_tasks;
+};
+
+/**
+ * @brief Clamps [offset, offset + size) to [0, mapping_size) and page-aligns the result. Returns an
+ * empty region (m_length == 0) for a null/empty mapping, an offset at or past the end, or a
+ * sub-page request.
+ */
+inline AlignedRegion clamp_align_region(const void* data, size_t mapping_size, size_t offset, size_t size) noexcept {
+    const auto page_size = static_cast<size_t>(get_system_page_size());
+    if (data == nullptr || mapping_size == 0 || offset >= mapping_size || size < page_size) {
+        return {};
+    }
+    const auto available = mapping_size - offset;
+    const auto raw_len = (size == auto_size) ? available : std::min(size, available);
+    return align_region(reinterpret_cast<uintptr_t>(data) + offset, raw_len, page_size);
+}
+
+/** @brief Aligned region and page-aligned size shared between the sync and async hint_prefetch(). */
+struct PrefetchPlan {
+    uintptr_t m_address = 0;
+    size_t m_aligned_size = 0;
+};
+
+/**
+ * @brief Computes the region and page-aligned size for a hint_prefetch()/hint_prefetch_async()
+ * call, or an empty plan (m_aligned_size == 0) when the region is below the parallel-I/O threshold
+ * (a real population pass would not be worth it).
+ */
+inline PrefetchPlan make_prefetch_plan(const void* data, size_t mapping_size, size_t offset, size_t size) noexcept {
+    if (const auto region = clamp_align_region(data, mapping_size, offset, size);
+        region.m_length > default_parallel_io_threshold) {
+        return {region.m_address, align_size_up(region.m_length, static_cast<size_t>(get_system_page_size()))};
+    }
+    return {};
+}
+
+/** @brief Upper bound on the shared page-population pool's worker threads. */
+inline constexpr size_t max_prefetch_threads = 8;
+
+/**
+ * @brief Number of page-population jobs a @p size byte region is split into, honoring the shared
+ * parallel-I/O minimum chunk size and the pool worker cap.
+ */
+inline size_t prefetch_thread_count(size_t size) noexcept {
+    const auto pool_cap =
+        std::max<size_t>(1, std::min<size_t>(max_prefetch_threads, std::thread::hardware_concurrency()));
+    return split_chunk_count(size, default_parallel_io_min_chunk, pool_cap);
+}
+
+}  // namespace ov::util