From e96e0db49de57913d344a041a624a9529c34f8ca Mon Sep 17 00:00:00 2001 From: Oleg Pipikin Date: Wed, 1 Jul 2026 15:41:05 +0200 Subject: [PATCH 1/5] Async mmap population on Linux --- .../util/include/openvino/util/memory.hpp | 114 ++++++++++- .../include/openvino/util/mmap_object.hpp | 22 +++ src/common/util/src/os/lin/lin_memory.cpp | 181 ++++++++++++++++- .../util/src/os/lin/lin_mmap_object.cpp | 87 ++++++++- src/common/util/src/os/win/win_memory.cpp | 14 +- .../util/src/os/win/win_mmap_object.cpp | 6 +- src/core/tests/file_load_benchmark.cpp | 16 +- src/core/tests/mmap_object.cpp | 184 ++++++++++++++++++ .../common_test_utils/tests/memory_test.cpp | 8 +- 9 files changed, 605 insertions(+), 27 deletions(-) diff --git a/src/common/util/include/openvino/util/memory.hpp b/src/common/util/include/openvino/util/memory.hpp index 351ca2df028469..0c8c37b2f66e5d 100644 --- a/src/common/util/include/openvino/util/memory.hpp +++ b/src/common/util/include/openvino/util/memory.hpp @@ -7,8 +7,10 @@ #include #include #include +#include #include #include +#include namespace ov::util { @@ -122,12 +124,116 @@ void vm_release(void* ptr, size_t size) noexcept; * * 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 fast Strategy selector: + * - @c true (default) → OS advisory hint (async, low overhead). + * - @c false → parallel synchronous touch that blocks until every page is + * resident. The degree of parallelism is chosen internally (see the shared pool + * used by @ref vm_prefetch_async), not by the caller. + */ +void vm_prefetch(void* ptr, size_t size, bool fast = true) noexcept; + +/** + * @brief Move-only RAII token representing background page-population work started by + * @ref vm_prefetch_async. + * + * The work itself runs on a shared, bounded background thread pool (see @ref vm_prefetch_async): + * the token owns the `std::future`s of the submitted tasks, not dedicated OS threads. Call + * wait() to block until the population completes, or simply let the token go out of scope — + * its destructor waits for all outstanding tasks, so no background work is ever left running + * uncontrolled. + * + * A default-constructed (or moved-from, or detached) token is "empty": wait() is a no-op and + * valid()/operator bool() return false. + * + * @note The token does not extend the lifetime of the underlying memory mapping/buffer that + * is being populated. The caller is responsible for keeping that memory alive until the + * token has completed (via wait(), destruction, or by taking over the futures via detach()). + */ +class PrefetchToken { +public: + PrefetchToken() noexcept = default; + explicit PrefetchToken(std::vector>&& 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(); // avoid abandoning still-running tasks owned by *this via member assignment + m_tasks = std::move(other.m_tasks); + } + return *this; + } + + ~PrefetchToken() { + wait(); + } + + /** + * @brief Blocks until all background population tasks complete, then releases them. + * Safe to call multiple times and on an empty token (no-op). + */ + void wait() { + for (auto& task : m_tasks) { + if (task.valid()) { + task.wait(); + } + } + m_tasks.clear(); + } + + /** + * @brief Detaches the background tasks from this token, transferring ownership of their + * futures to the caller, and returns them; the token's destructor will no longer wait for + * them (mirrors `std::thread::detach()`). + * + * The caller becomes responsible for the memory being populated remaining valid until the + * returned futures complete — either by storing and waiting on them itself (e.g. an object + * whose memory is being populated can join them before it tears that memory down), or, if + * discarded outright, by guaranteeing the memory outlives the background work some other + * way. + * + * After detach(), the token is empty (valid() == false). + */ + std::vector> detach() noexcept { + auto tasks = std::move(m_tasks); + m_tasks.clear(); + return tasks; + } + + /** @brief Returns true if the token owns outstanding background work. */ + bool valid() const noexcept { + return !m_tasks.empty(); + } + + explicit operator bool() const noexcept { + return valid(); + } + +private: + std::vector> m_tasks; +}; + +/** + * @brief Asynchronous variant of @ref vm_prefetch. + * + * Starts pre-fetching a committed VM range into physical memory by submitting page-touching + * tasks to a small, shared background thread pool, and returns immediately with a + * @ref PrefetchToken that must be used to wait for completion (explicitly via wait(), or + * implicitly by letting the token go out of scope). + * + * Unlike spawning dedicated threads per call, repeated calls reuse the same bounded set of + * pool worker threads: tasks queue up and are picked up by whichever worker becomes free. + * * @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). + * @return A @ref PrefetchToken owning the submitted tasks' futures (empty if the pages could not + * be scheduled for population, e.g. on allocation failure). */ -void vm_prefetch(void* ptr, size_t size, size_t num_threads = 0) noexcept; +PrefetchToken vm_prefetch_async(void* ptr, size_t size) noexcept; } // namespace ov::util diff --git a/src/common/util/include/openvino/util/mmap_object.hpp b/src/common/util/include/openvino/util/mmap_object.hpp index 56e0567ac7fe61..2ae2390ad295aa 100644 --- a/src/common/util/include/openvino/util/mmap_object.hpp +++ b/src/common/util/include/openvino/util/mmap_object.hpp @@ -16,6 +16,7 @@ #include #include "openvino/util/file_util.hpp" +#include "openvino/util/memory.hpp" namespace ov { @@ -48,8 +49,29 @@ 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 of the mapping in background threads and returns + * immediately. Unlike the lower-level @ref util::vm_prefetch_async, no token is handed back + * to the caller: the implementation is fully responsible for tracking the background work + * internally and joining it before this object is destroyed (so the mapping is never torn + * down while a background task might still be touching it), regardless of what the caller + * does afterwards. This avoids requiring the caller to correctly manage a token's lifetime + * relative to this object's lifetime, which is an easy way to introduce a use-after-free. + * + * The default implementation is a no-op, so existing implementers of this interface keep + * compiling/behaving unchanged unless they override it. + * + * @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) {} }; + /** * @brief Returns mapped memory for a file from provided path. * Instead of reading files, we can map the memory via mmap for Linux diff --git a/src/common/util/src/os/lin/lin_memory.cpp b/src/common/util/src/os/lin/lin_memory.cpp index 6ad74c2c6d76d6..7e3dc244a19f52 100644 --- a/src/common/util/src/os/lin/lin_memory.cpp +++ b/src/common/util/src/os/lin/lin_memory.cpp @@ -4,14 +4,25 @@ #include +#include #include #include +#include #include +#include #include #include +#include +#include +#include +#include +#include #include +#include +#include "openvino/util/common_util.hpp" #include "openvino/util/memory.hpp" +#include "openvino/util/mmap_object.hpp" namespace ov::util { @@ -22,9 +33,140 @@ void madvise_hint(void* ptr, size_t size) noexcept { madvise(ptr, size, MADV_WILLNEED); } -} // namespace +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; + } + } +}; + +/** + * @brief A small, bounded, process-wide thread pool used to run page-population tasks. + * + * Rather than spawning dedicated OS threads for every vm_prefetch(_async) call, tasks are + * queued and picked up by whichever of the pool's fixed worker threads becomes free. This + * keeps the number of live threads bounded regardless of how many prefetch calls are made + * concurrently. + */ +class PageTouchThreadPool { +public: + static PageTouchThreadPool& instance() { + static PageTouchThreadPool pool; + return pool; + } + + PageTouchThreadPool(const PageTouchThreadPool&) = delete; + PageTouchThreadPool& operator=(const PageTouchThreadPool&) = delete; + + /** + * @brief Queues @p jobs for execution on the pool and returns a future per job that + * becomes ready once that job has run. + * + * Enqueueing is all-or-nothing: every throwing operation (task/future creation) happens on + * local state first, and the finished batch is spliced into the shared queue under the lock + * with a non-throwing list splice. If preparation throws (e.g. std::bad_alloc), nothing is + * enqueued and no job runs, so callers never observe a partially-submitted batch whose jobs + * might touch memory the caller believes was never scheduled. + */ + std::vector> submit(std::vector>&& jobs) { + std::vector> futures; + futures.reserve(jobs.size()); + std::list> pending; + for (auto& job : jobs) { + auto task = std::make_shared>(std::move(job)); + futures.push_back(task->get_future()); + pending.emplace_back([task]() { + (*task)(); + }); + } + { + std::lock_guard lock(m_mutex); + m_queue.splice(m_queue.end(), pending); // non-throwing: all-or-nothing enqueue + } + m_cv.notify_all(); + return futures; + } -void populate_pages(void* ptr, size_t size, size_t num_threads) noexcept; + size_t worker_count() const noexcept { + return m_workers.size(); + } + +private: + PageTouchThreadPool() + : m_workers(std::max(1, std::min(10, std::thread::hardware_concurrency()))) { + for (auto& worker : m_workers) { + worker = std::thread([this]() { + worker_loop(); + }); + } + } + + ~PageTouchThreadPool() { + { + std::lock_guard lock(m_mutex); + m_stop = true; + } + m_cv.notify_all(); + for (auto& worker : m_workers) { + if (worker.joinable()) { + worker.join(); + } + } + } + + void worker_loop() { + for (;;) { + std::function job; + { + std::unique_lock lock(m_mutex); + m_cv.wait(lock, [this]() { + return m_stop || !m_queue.empty(); + }); + if (m_queue.empty()) { + if (m_stop) { + return; // fully drained: safe to let this worker exit. + } + continue; + } + job = std::move(m_queue.front()); + m_queue.pop_front(); + } + job(); + } + } + + std::vector m_workers; + std::list> m_queue; + std::mutex m_mutex; + std::condition_variable m_cv; + bool m_stop = false; +}; + +/** + * @brief Splits [ptr, ptr + size) into page-toucher jobs (one per chunk) and submits them to the + * shared @ref PageTouchThreadPool, returning a future per job. + */ +std::vector> submit_page_toucher_tasks(void* ptr, size_t size) { + // ptr and size are guaranteed page-aligned by vm_prefetch's precondition. + const auto page_size = static_cast(get_system_page_size()); + const auto num_threads = PageTouchThreadPool::instance().worker_count(); + const auto chunk_size = std::max(align_size_up(size / num_threads, page_size), 1024 * 1024); + + std::vector> jobs; + jobs.reserve(ceil_div(size, chunk_size)); + + for (auto first = reinterpret_cast(ptr), last = first + size; first < last; first += chunk_size) { + jobs.emplace_back(PageToucher{first, std::min(first + chunk_size, last), page_size}); + } + return PageTouchThreadPool::instance().submit(std::move(jobs)); +} +} // namespace void* aligned_alloc(size_t size, size_t alignment) noexcept { if (alignment == 0) { @@ -74,13 +216,40 @@ void vm_release(void* ptr, size_t size) noexcept { std::ignore = munmap(ptr, size); } -void vm_prefetch(void* ptr, size_t size, size_t num_threads) noexcept { +void vm_prefetch(void* ptr, size_t size, bool fast) noexcept { assert(ptr != nullptr && size > 0); - if (num_threads == 0) { + // assert if region is not mmap-backed. + + if (fast) { + // Option 1: OS advisory hints — async, low overhead. madvise_hint(ptr, size); } else { - // blocks until every page has been faulted in. - populate_pages(ptr, size, num_threads); + // Option 2: parallel synchronous touch — blocks until every page is resident. + // Must not be called from a PageTouchThreadPool worker thread: it blocks on that shared + // pool, so if every worker were parked here no worker would remain to run the jobs. + // Prefetching is best-effort, so on allocation failure fall back to an advisory hint + // rather than letting an exception escape this noexcept function (which would + // std::terminate). + try { + PrefetchToken(submit_page_toucher_tasks(ptr, size)).wait(); + } catch (...) { + madvise_hint(ptr, size); + } + } +} + +PrefetchToken vm_prefetch_async(void* ptr, size_t size) noexcept { + assert(ptr != nullptr && size > 0); + + // Submit the touchers to the shared pool but do not wait on them here — ownership of the + // futures is transferred to the token, whose destructor (or explicit wait()) waits on them. + // Prefetching is best-effort: if submission fails (e.g. std::bad_alloc) fall back to an + // advisory hint and return an empty token instead of throwing out of this noexcept function. + try { + return PrefetchToken(submit_page_toucher_tasks(ptr, size)); + } catch (...) { + madvise_hint(ptr, size); + return PrefetchToken{}; } } diff --git a/src/common/util/src/os/lin/lin_mmap_object.cpp b/src/common/util/src/os/lin/lin_mmap_object.cpp index 763c01175fbab1..6a6f5948b0dedc 100644 --- a/src/common/util/src/os/lin/lin_mmap_object.cpp +++ b/src/common/util/src/os/lin/lin_mmap_object.cpp @@ -9,7 +9,10 @@ #include #include +#include #include +#include +#include #include #include @@ -56,6 +59,35 @@ inline util::AlignedRegion make_madvise_region(const void* data, size_t mapping_ return util::align_region(reinterpret_cast(data) + offset, raw_len, page_size); } } + +/** + * @brief Plan computed by @ref make_prefetch_plan, shared between the sync and async + * hint_prefetch() implementations. + */ +struct PrefetchPlan { + uintptr_t m_address = 0; + size_t m_aligned_size = 0; +}; + +/** + * @brief Computes the aligned region and page-aligned size to use for a + * hint_prefetch()/hint_prefetch_async() call. + * + * @param data The base address of the mapped memory region. + * @param mapping_size The size of the mapped memory region. + * @param offset Offset within the mapping where prefetching starts. + * @param size Number of bytes to prefetch. + * @return PrefetchPlan with m_aligned_size == 0 when the requested region is below the + * 4 MiB 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) { + constexpr size_t one_mb = 1024 * 1024; + // Below 4 MiB the overhead of populating pages exceeds the benefit; skip. + if (const auto region = make_madvise_region(data, mapping_size, offset, size); region.m_length > 4 * one_mb) { + return {region.m_address, align_size_up(region.m_length, static_cast(get_system_page_size()))}; + } + return {}; +} } // namespace util class HandleHolder { @@ -103,6 +135,38 @@ class MapHolder final : public MappedMemory { size_t m_size = 0; uint64_t m_id = std::numeric_limits::max(); HandleHolder m_handle; + // Tasks handed off via PrefetchToken::detach(): must be waited on before unmapping, see + // wait_for_pending_prefetch() / ~MapHolder(). + std::mutex m_pending_prefetch_mutex; + std::vector> m_pending_prefetch; + + void adopt_pending_prefetch(std::vector>&& tasks) { + std::lock_guard lock(m_pending_prefetch_mutex); + // Opportunistically reap already-finished futures (non-blocking check via wait_for(0)) + // so the vector doesn't grow without bound if hint_prefetch_async().detach() is called + // repeatedly over the lifetime of this mapping. + m_pending_prefetch.erase(std::remove_if(m_pending_prefetch.begin(), + m_pending_prefetch.end(), + [](std::future& task) { + return !task.valid() || + task.wait_for(std::chrono::seconds(0)) == + std::future_status::ready; + }), + m_pending_prefetch.end()); + m_pending_prefetch.insert(m_pending_prefetch.end(), + std::make_move_iterator(tasks.begin()), + std::make_move_iterator(tasks.end())); + } + + void wait_for_pending_prefetch() noexcept { + std::lock_guard lock(m_pending_prefetch_mutex); + for (auto& task : m_pending_prefetch) { + if (task.valid()) { + task.wait(); + } + } + m_pending_prefetch.clear(); + } public: MapHolder() = default; @@ -150,6 +214,9 @@ class MapHolder final : public MappedMemory { } ~MapHolder() { + // Detached prefetch tasks (see hint_prefetch_async()) may still be touching this + // mapping's pages; they must complete before the mapping is torn down. + wait_for_pending_prefetch(); if (m_mapped_view != MAP_FAILED) { munmap(m_mapped_view, m_mapped_view_size); } @@ -172,12 +239,20 @@ class MapHolder final : public MappedMemory { } void hint_prefetch(size_t offset, size_t size) override { - constexpr size_t one_mb = 1024 * 1024; - // Below 4 MiB the overhead of spawning threads exceeds the benefit; skip. - if (const auto region = util::make_madvise_region(m_data, m_size, offset, size); region.m_length > 4 * one_mb) { - const auto num_threads = std::min(10, std::thread::hardware_concurrency()); - const auto aligned_size = util::align_size_up(region.m_length, util::get_system_page_size()); - util::vm_prefetch(reinterpret_cast(region.m_address), aligned_size, num_threads); + if (const auto plan = util::make_prefetch_plan(m_data, m_size, offset, size); plan.m_aligned_size) { + util::vm_prefetch(reinterpret_cast(plan.m_address), plan.m_aligned_size, /*fast=*/false); + } + } + + void hint_prefetch_async(size_t offset, size_t size) override { + if (const auto plan = util::make_prefetch_plan(m_data, m_size, offset, size); plan.m_aligned_size) { + // Never hand the token back to the caller: adopt its tasks into m_pending_prefetch + // right here, in this same call, before returning. This guarantees ~MapHolder() + // will always join them before munmap, regardless of what the caller does + // afterwards (e.g. dropping its last reference to this object right away) — there + // is no window where the tasks' lifetime is decoupled from this object's lifetime. + auto token = util::vm_prefetch_async(reinterpret_cast(plan.m_address), plan.m_aligned_size); + adopt_pending_prefetch(token.detach()); } } }; diff --git a/src/common/util/src/os/win/win_memory.cpp b/src/common/util/src/os/win/win_memory.cpp index f0b5d7cdd68f51..c33890e863fd0e 100644 --- a/src/common/util/src/os/win/win_memory.cpp +++ b/src/common/util/src/os/win/win_memory.cpp @@ -9,9 +9,11 @@ #include #include +#include #include #include #include +#include #include #include "openvino/util/memory.hpp" @@ -59,15 +61,23 @@ void vm_release(void* ptr, size_t) noexcept { std::ignore = VirtualFree(ptr, 0, MEM_RELEASE); } -void vm_prefetch(void* ptr, size_t size, size_t num_threads) noexcept { +void vm_prefetch(void* ptr, size_t size, bool fast) noexcept { assert(ptr != nullptr && size > 0); - if (num_threads == 0) { + if (fast) { WIN32_MEMORY_RANGE_ENTRY entry{ptr, size}; ::PrefetchVirtualMemory(::GetCurrentProcess(), 1, &entry, 0); } else { // blocks until every page has been faulted in. + const auto num_threads = std::max(1, std::min(10, std::thread::hardware_concurrency())); populate_pages(ptr, size, num_threads); } } +PrefetchToken vm_prefetch_async(void* ptr, size_t size) noexcept { + assert(ptr != nullptr && size > 0); + // CVS-186579 + // No background work is started on Windows; mirrors the vm_prefetch() no-op stub. + return {}; +} + } // namespace ov::util diff --git a/src/common/util/src/os/win/win_mmap_object.cpp b/src/common/util/src/os/win/win_mmap_object.cpp index c1f637968aa145..c4d27adbc0be50 100644 --- a/src/common/util/src/os/win/win_mmap_object.cpp +++ b/src/common/util/src/os/win/win_mmap_object.cpp @@ -8,7 +8,6 @@ #include #include #include -#include #include #include "openvino/util/file_util.hpp" @@ -305,6 +304,8 @@ class MapHolder : public ov::MappedMemory { void hint_prefetch(size_t offset, size_t size) override; + void hint_prefetch_async(size_t /*offset*/, size_t /*size*/) override {} + private: /** * @brief Remaps a placeholder region by replacing it with a file-backed view. @@ -764,10 +765,9 @@ util::AlignedRegion clamp_align_region(const void* data, size_t mapping_size, si void MapHolder::hint_prefetch(size_t offset, size_t size) { // Below 4 MiB the overhead of spawning threads exceeds the benefit; skip. if (const auto region = clamp_align_region(m_data, m_size, offset, size); region.m_length > 4 * util::one_mib) { - const auto num_threads = std::min(10, std::thread::hardware_concurrency()); const auto aligned_size = util::align_size_up(region.m_length, static_cast(util::get_system_page_size())); - util::vm_prefetch(reinterpret_cast(region.m_address), aligned_size, num_threads); + util::vm_prefetch(reinterpret_cast(region.m_address), aligned_size, /*fast=*/false); } } diff --git a/src/core/tests/file_load_benchmark.cpp b/src/core/tests/file_load_benchmark.cpp index d14bd1bf9bf6fc..cc78e057410f46 100644 --- a/src/core/tests/file_load_benchmark.cpp +++ b/src/core/tests/file_load_benchmark.cpp @@ -17,7 +17,6 @@ #include #include #include -#include #include #include @@ -205,7 +204,7 @@ namespace strategy { // Note: the mmap destructor (munmap + close) runs inside the timed window; void sync_vm_prefetch_mem_lock(const std::filesystem::path& path, size_t /*file_size*/) { auto mapped = load_mmap_object(path); - util::vm_prefetch(mapped->data(), mapped->size(), std::thread::hardware_concurrency()); + util::vm_prefetch(mapped->data(), mapped->size(), /*fast=*/false); ensure_memory_resident(mapped); // should be near no-op and just lock/unlock resident pages } @@ -240,6 +239,19 @@ void compute_over_mapped(const std::shared_ptr& mapped) { (void)sink; } +void parallel_loop_sync_then_memcpy(const std::filesystem::path& path, size_t file_size) { + auto mapped = load_mmap_object(path); + util::vm_prefetch(mapped->data(), mapped->size(), /*fast=*/false); + constexpr size_t chunk_size = 128 * util::one_mib; + std::vector buffer(std::min(chunk_size, file_size)); + volatile char sink = 0; + for (size_t offset = 0; offset < file_size; offset += chunk_size) { + const size_t copy_size = std::min(chunk_size, file_size - offset); + std::memcpy(buffer.data(), mapped->data() + offset, copy_size); + sink += buffer[0] + buffer[copy_size / 2] + buffer[copy_size - 1]; // prevents optimization + } +} + void mmap_then_compute(const std::filesystem::path& path, size_t /*file_size*/) { auto mapped = load_mmap_object(path); compute_over_mapped(mapped); diff --git a/src/core/tests/mmap_object.cpp b/src/core/tests/mmap_object.cpp index d5974e39e80051..cd5f1190f4d47d 100644 --- a/src/core/tests/mmap_object.cpp +++ b/src/core/tests/mmap_object.cpp @@ -8,10 +8,13 @@ #include #include +#include #include #include #include #include +#include +#include #include "common_test_utils/common_utils.hpp" #include "common_test_utils/file_utils.hpp" @@ -473,4 +476,185 @@ TEST_F(HintPrefetchTest, hint_prefetch_sequential_eviction_check) { EXPECT_EQ(pages_after, pages_before) << "hint_prefetch evicted pages."; } +class HintPrefetchAsyncTest : public ::testing::Test { +protected: + std::filesystem::path m_file_path; + + void TearDown() override { + std::filesystem::remove(m_file_path); + } + + static std::vector read_mapped(MappedMemory& mm) { + return {reinterpret_cast(mm.data()), reinterpret_cast(mm.data()) + mm.size()}; + } + + static std::vector make_pattern(size_t size) { + std::vector data(size); + for (size_t i = 0; i < size; ++i) + data[i] = static_cast(i % 251); + return data; + } + + void write_file(const std::vector& data) { + std::ofstream f(m_file_path, std::ios::binary); + f.write(reinterpret_cast(data.data()), data.size()); + } + + // hint_prefetch_async() no longer returns a token to wait on: completion is tracked and + // joined internally by the MappedMemory implementation. To observe residency in tests we + // simply poll until the background population catches up (or a generous timeout elapses). + static size_t wait_for_resident_pages(const char* addr, + size_t size, + size_t expected_pages, + std::chrono::milliseconds timeout = std::chrono::seconds(10)) { + const auto deadline = std::chrono::steady_clock::now() + timeout; + size_t pages_resident = 0; + do { + pages_resident = utils::count_resident_pages(addr, size); + if (pages_resident >= expected_pages) + break; + std::this_thread::sleep_for(std::chrono::milliseconds(10)); + } while (std::chrono::steady_clock::now() < deadline); + return pages_resident; + } +}; + +TEST_F(HintPrefetchAsyncTest, pages_resident_eventually) { +#ifndef __linux__ + GTEST_SKIP() << "utils::count_resident_pages is not implemented on this platform yet CVS-186579"; +#endif + m_file_path = std::filesystem::path(utils::generateTestFilePrefix() + "_prefetch_async_wait.bin"); + constexpr size_t file_size = 8 * 1024 * 1024; // 8 MiB (above 4 MiB threshold) + const auto data = make_pattern(file_size); + write_file(data); + + auto mapped = load_mmap_object(m_file_path); + ASSERT_NE(mapped, nullptr); + + const size_t page = static_cast(util::get_system_page_size()); + const size_t total_pages = (file_size + page - 1) / page; + + mapped->hint_prefetch_async(); + + const size_t pages_resident = wait_for_resident_pages(mapped->data(), file_size, total_pages); + EXPECT_EQ(pages_resident, total_pages) << "Expected all pages resident after hint_prefetch_async()."; +} + +TEST_F(HintPrefetchAsyncTest, prefetch_then_immediate_destruction_is_safe) { + // Regression test: hint_prefetch_async() must never leave background page-touching tasks + // racing with this object's destruction. Dropping the last reference to the mapping right + // after kicking off the prefetch (with zero waiting) must not crash / use-after-free, since + // the implementation is required to join any in-flight tasks before unmapping. + m_file_path = std::filesystem::path(utils::generateTestFilePrefix() + "_prefetch_async_destroy.bin"); + constexpr size_t file_size = 8 * 1024 * 1024; // 8 MiB (above 4 MiB threshold) + const auto data = make_pattern(file_size); + write_file(data); + + auto mapped = load_mmap_object(m_file_path); + ASSERT_NE(mapped, nullptr); + + mapped->hint_prefetch_async(); + mapped.reset(); // must not crash, regardless of whether the background tasks finished yet +} + +TEST_F(HintPrefetchAsyncTest, partial_region_populated_and_correct) { +#ifndef __linux__ + GTEST_SKIP() << "utils::count_resident_pages is not implemented on this platform yet CVS-186579"; +#endif + m_file_path = std::filesystem::path(utils::generateTestFilePrefix() + "_prefetch_async_partial.bin"); + constexpr size_t file_size = 8 * 1024 * 1024; // 8 MiB + constexpr size_t prefetch_offset = 1 * 1024 * 1024; + constexpr size_t prefetch_size = 5 * 1024 * 1024; + const auto data = make_pattern(file_size); + write_file(data); + + auto mapped = load_mmap_object(m_file_path); + ASSERT_NE(mapped, nullptr); + + const size_t page = static_cast(util::get_system_page_size()); + const size_t region_pages = (prefetch_size + page - 1) / page; + + mapped->hint_prefetch_async(prefetch_offset, prefetch_size); + + const size_t pages_resident = wait_for_resident_pages(mapped->data() + prefetch_offset, prefetch_size, region_pages); + EXPECT_EQ(pages_resident, region_pages) << "Expected the requested region to be fully resident."; + + EXPECT_EQ(read_mapped(*mapped), data); +} + +TEST_F(HintPrefetchAsyncTest, below_threshold_is_safe_noop) { + m_file_path = std::filesystem::path(utils::generateTestFilePrefix() + "_prefetch_async_small.bin"); + constexpr size_t file_size = 1024; // 1 KiB - below the 4 MiB threshold + const auto data = make_pattern(file_size); + write_file(data); + + auto mapped = load_mmap_object(m_file_path); + ASSERT_NE(mapped, nullptr); + + EXPECT_NO_THROW(mapped->hint_prefetch_async()); + EXPECT_EQ(read_mapped(*mapped), data); +} + +TEST_F(HintPrefetchAsyncTest, data_correct_immediately_after_call) { + m_file_path = std::filesystem::path(utils::generateTestFilePrefix() + "_prefetch_async_alive.bin"); + constexpr size_t file_size = 8 * 1024 * 1024; // 8 MiB (above 4 MiB threshold) + const auto data = make_pattern(file_size); + write_file(data); + + auto mapped = load_mmap_object(m_file_path); + ASSERT_NE(mapped, nullptr); + + mapped->hint_prefetch_async(); + // The mapping is always readable/correct regardless of prefetch completion; background + // threads only accelerate residency. + EXPECT_EQ(read_mapped(*mapped), data); +} + +TEST(PrefetchTokenTest, move_transfers_ownership) { + // MappedMemory::hint_prefetch_async() no longer exposes a util::PrefetchToken at all, so + // exercise PrefetchToken's move semantics directly against the lower-level free function + // instead, using a page-aligned buffer that is intentionally never freed for the lifetime + // of the process (mirrors detach_releases_futures_to_caller below). + const auto page = static_cast(util::get_system_page_size()); + const size_t buf_size = util::align_size_down(8 * 1024 * 1024, page); + static void* leaked_buffer = util::aligned_alloc(buf_size, page); + ASSERT_NE(leaked_buffer, nullptr); + + auto token = util::vm_prefetch_async(leaked_buffer, buf_size); + ASSERT_TRUE(static_cast(token)); + + util::PrefetchToken moved(std::move(token)); + EXPECT_FALSE(static_cast(token)); // moved-from token is empty + EXPECT_TRUE(static_cast(moved)); // ownership transferred to the new token + + util::PrefetchToken move_assigned; + move_assigned = std::move(moved); + EXPECT_FALSE(static_cast(moved)); // moved-from token is empty again + EXPECT_TRUE(static_cast(move_assigned)); // ownership transferred via move-assignment + + move_assigned.wait(); + EXPECT_FALSE(static_cast(move_assigned)); +} + +TEST_F(HintPrefetchAsyncTest, detach_releases_futures_to_caller) { + // Use a page-aligned buffer that is intentionally never freed for the lifetime of the + // process, so that discarding the futures returned by detach() (fire-and-forget, like + // std::thread::detach()) cannot cause a real use-after-free, regardless of when the tasks + // actually finish running. + const auto page = static_cast(util::get_system_page_size()); + const size_t buf_size = util::align_size_down(8 * 1024 * 1024, page); + static void* leaked_buffer = util::aligned_alloc(buf_size, page); + ASSERT_NE(leaked_buffer, nullptr); + + auto token = util::vm_prefetch_async(leaked_buffer, buf_size); + ASSERT_TRUE(static_cast(token)); + + // detach() transfers ownership of the futures to the caller and empties the token; the + // caller here simply discards them (fire-and-forget). + auto tasks = token.detach(); + EXPECT_FALSE(tasks.empty()); + EXPECT_FALSE(static_cast(token)); +} + } // namespace ov::test + diff --git a/src/tests/test_utils/common_test_utils/tests/memory_test.cpp b/src/tests/test_utils/common_test_utils/tests/memory_test.cpp index ce21086ff61c4e..5a1d4c23b5b371 100644 --- a/src/tests/test_utils/common_test_utils/tests/memory_test.cpp +++ b/src/tests/test_utils/common_test_utils/tests/memory_test.cpp @@ -157,7 +157,7 @@ TEST_F(AlignedAllocTest, free_nullptr_is_noop) { EXPECT_NO_FATAL_FAILURE(util::aligned_free(nullptr)); } -class VmPrefetchMappedFileTest : public testing::TestWithParam { +class VmPrefetchMappedFileTest : public testing::TestWithParam { protected: std::filesystem::path m_file_path; @@ -167,7 +167,7 @@ class VmPrefetchMappedFileTest : public testing::TestWithParam { }; TEST_P(VmPrefetchMappedFileTest, prefetch_faults_in_mapped_file_and_preserves_data) { - const size_t num_threads = GetParam(); + const bool fast = GetParam(); const size_t size = 64 * util::min_page_alignment; const auto expected = utils::make_modulo_sequence_pattern(size); @@ -179,11 +179,11 @@ TEST_P(VmPrefetchMappedFileTest, prefetch_faults_in_mapped_file_and_preserves_da ASSERT_NE(mapped, nullptr); ASSERT_EQ(mapped->size(), size); - EXPECT_NO_FATAL_FAILURE(util::vm_prefetch(mapped->data(), mapped->size(), num_threads)); + EXPECT_NO_FATAL_FAILURE(util::vm_prefetch(mapped->data(), mapped->size(), fast)); EXPECT_THAT(expected, ElementsAreArray(reinterpret_cast(mapped->data()), mapped->size())); } -INSTANTIATE_TEST_SUITE_P(NumThreads, VmPrefetchMappedFileTest, testing::Values(0u, 5u, 10u)); +INSTANTIATE_TEST_SUITE_P(Fast, VmPrefetchMappedFileTest, testing::Bool()); } // namespace ov::test From f733690435a718587047bb3f9b9fbaf4fe822c76 Mon Sep 17 00:00:00 2001 From: Oleg Pipikin Date: Wed, 15 Jul 2026 14:31:51 +0200 Subject: [PATCH 2/5] Refactor and comments --- src/common/util/CMakeLists.txt | 2 + .../util/include/openvino/util/memory.hpp | 120 ----------- .../include/openvino/util/mmap_object.hpp | 15 +- .../include/openvino/util/parallel_io.hpp | 10 + src/common/util/src/memory.cpp | 11 +- src/common/util/src/memory_prefetch.hpp | 144 +++++++++++++ src/common/util/src/os/lin/lin_memory.cpp | 189 ++++++++---------- .../util/src/os/lin/lin_mmap_object.cpp | 78 ++------ src/common/util/src/os/win/win_memory.cpp | 17 +- .../util/src/os/win/win_mmap_object.cpp | 29 +-- .../util/src/parallel_read_streambuf.cpp | 3 +- src/core/tests/file_load_benchmark.cpp | 4 +- src/core/tests/mmap_object.cpp | 87 ++++---- .../common_test_utils/tests/memory_test.cpp | 31 --- 14 files changed, 326 insertions(+), 414 deletions(-) create mode 100644 src/common/util/src/memory_prefetch.hpp diff --git a/src/common/util/CMakeLists.txt b/src/common/util/CMakeLists.txt index da7305dd89baa7..d3fffcb2b7961b 100644 --- a/src/common/util/CMakeLists.txt +++ b/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 $) +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} diff --git a/src/common/util/include/openvino/util/memory.hpp b/src/common/util/include/openvino/util/memory.hpp index 0c8c37b2f66e5d..da1f1b1f54dc4e 100644 --- a/src/common/util/include/openvino/util/memory.hpp +++ b/src/common/util/include/openvino/util/memory.hpp @@ -7,10 +7,8 @@ #include #include #include -#include #include #include -#include namespace ov::util { @@ -118,122 +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 fast Strategy selector: - * - @c true (default) → OS advisory hint (async, low overhead). - * - @c false → parallel synchronous touch that blocks until every page is - * resident. The degree of parallelism is chosen internally (see the shared pool - * used by @ref vm_prefetch_async), not by the caller. - */ -void vm_prefetch(void* ptr, size_t size, bool fast = true) noexcept; - -/** - * @brief Move-only RAII token representing background page-population work started by - * @ref vm_prefetch_async. - * - * The work itself runs on a shared, bounded background thread pool (see @ref vm_prefetch_async): - * the token owns the `std::future`s of the submitted tasks, not dedicated OS threads. Call - * wait() to block until the population completes, or simply let the token go out of scope — - * its destructor waits for all outstanding tasks, so no background work is ever left running - * uncontrolled. - * - * A default-constructed (or moved-from, or detached) token is "empty": wait() is a no-op and - * valid()/operator bool() return false. - * - * @note The token does not extend the lifetime of the underlying memory mapping/buffer that - * is being populated. The caller is responsible for keeping that memory alive until the - * token has completed (via wait(), destruction, or by taking over the futures via detach()). - */ -class PrefetchToken { -public: - PrefetchToken() noexcept = default; - explicit PrefetchToken(std::vector>&& 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(); // avoid abandoning still-running tasks owned by *this via member assignment - m_tasks = std::move(other.m_tasks); - } - return *this; - } - - ~PrefetchToken() { - wait(); - } - - /** - * @brief Blocks until all background population tasks complete, then releases them. - * Safe to call multiple times and on an empty token (no-op). - */ - void wait() { - for (auto& task : m_tasks) { - if (task.valid()) { - task.wait(); - } - } - m_tasks.clear(); - } - - /** - * @brief Detaches the background tasks from this token, transferring ownership of their - * futures to the caller, and returns them; the token's destructor will no longer wait for - * them (mirrors `std::thread::detach()`). - * - * The caller becomes responsible for the memory being populated remaining valid until the - * returned futures complete — either by storing and waiting on them itself (e.g. an object - * whose memory is being populated can join them before it tears that memory down), or, if - * discarded outright, by guaranteeing the memory outlives the background work some other - * way. - * - * After detach(), the token is empty (valid() == false). - */ - std::vector> detach() noexcept { - auto tasks = std::move(m_tasks); - m_tasks.clear(); - return tasks; - } - - /** @brief Returns true if the token owns outstanding background work. */ - bool valid() const noexcept { - return !m_tasks.empty(); - } - - explicit operator bool() const noexcept { - return valid(); - } - -private: - std::vector> m_tasks; -}; - -/** - * @brief Asynchronous variant of @ref vm_prefetch. - * - * Starts pre-fetching a committed VM range into physical memory by submitting page-touching - * tasks to a small, shared background thread pool, and returns immediately with a - * @ref PrefetchToken that must be used to wait for completion (explicitly via wait(), or - * implicitly by letting the token go out of scope). - * - * Unlike spawning dedicated threads per call, repeated calls reuse the same bounded set of - * pool worker threads: tasks queue up and are picked up by whichever worker becomes free. - * - * @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. - * @return A @ref PrefetchToken owning the submitted tasks' futures (empty if the pages could not - * be scheduled for population, e.g. on allocation failure). - */ -PrefetchToken vm_prefetch_async(void* ptr, size_t size) noexcept; - } // namespace ov::util diff --git a/src/common/util/include/openvino/util/mmap_object.hpp b/src/common/util/include/openvino/util/mmap_object.hpp index 2ae2390ad295aa..944ac145c4af63 100644 --- a/src/common/util/include/openvino/util/mmap_object.hpp +++ b/src/common/util/include/openvino/util/mmap_object.hpp @@ -51,18 +51,9 @@ class MappedMemory { 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 of the mapping in background threads and returns - * immediately. Unlike the lower-level @ref util::vm_prefetch_async, no token is handed back - * to the caller: the implementation is fully responsible for tracking the background work - * internally and joining it before this object is destroyed (so the mapping is never torn - * down while a background task might still be touching it), regardless of what the caller - * does afterwards. This avoids requiring the caller to correctly manage a token's lifetime - * relative to this object's lifetime, which is an easy way to introduce a use-after-free. - * - * The default implementation is a no-op, so existing implementers of this interface keep - * compiling/behaving unchanged unless they override it. + * @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 diff --git a/src/common/util/include/openvino/util/parallel_io.hpp b/src/common/util/include/openvino/util/parallel_io.hpp index 1911cf3b8d2d65..ce31df31894ae6 100644 --- a/src/common/util/include/openvino/util/parallel_io.hpp +++ b/src/common/util/include/openvino/util/parallel_io.hpp @@ -9,6 +9,7 @@ #pragma once +#include #include #include @@ -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(1, std::min(max_chunks, by_size)); +} + /** * @brief Open a file for reading and retrieve its size. * diff --git a/src/common/util/src/memory.cpp b/src/common/util/src/memory.cpp index 612945e5a86b66..203610e4ac47ab 100644 --- a/src/common/util/src/memory.cpp +++ b/src/common/util/src/memory.cpp @@ -5,7 +5,6 @@ #include "openvino/util/memory.hpp" #include -#include #include #include #include @@ -18,18 +17,15 @@ 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. - */ +// 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; // prevents the compiler from optimizing the loop away + volatile uint8_t local = 0; for (auto begin = m_begin; begin < m_end; begin += m_page_size) { local += *begin; } @@ -39,7 +35,6 @@ struct PageToucher { } // 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(get_system_page_size()); const auto chunk_size = std::max(align_size_up(size / num_threads, page_size), one_mib); diff --git a/src/common/util/src/memory_prefetch.hpp b/src/common/util/src/memory_prefetch.hpp new file mode 100644 index 00000000000000..b5e4aea3f166a1 --- /dev/null +++ b/src/common/util/src/memory_prefetch.hpp @@ -0,0 +1,144 @@ +// Copyright (C) 2018-2026 Intel Corporation +// SPDX-License-Identifier: Apache-2.0 +// + +#pragma once + +#include +#include +#include +#include +#include +#include + +#include "openvino/util/memory.hpp" +#include "openvino/util/mmap_object.hpp" +#include "openvino/util/parallel_io.hpp" + +namespace ov::util { + +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>&& 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> 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> 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(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(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(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(1, std::min(max_prefetch_threads, std::thread::hardware_concurrency())); + return split_chunk_count(size, default_parallel_io_min_chunk, pool_cap); +} + +} // namespace ov::util diff --git a/src/common/util/src/os/lin/lin_memory.cpp b/src/common/util/src/os/lin/lin_memory.cpp index 7e3dc244a19f52..7f32eeb07d1e0f 100644 --- a/src/common/util/src/os/lin/lin_memory.cpp +++ b/src/common/util/src/os/lin/lin_memory.cpp @@ -5,7 +5,6 @@ #include #include -#include #include #include #include @@ -20,9 +19,11 @@ #include #include +#include "memory_prefetch.hpp" #include "openvino/util/common_util.hpp" #include "openvino/util/memory.hpp" #include "openvino/util/mmap_object.hpp" +#include "openvino/util/parallel_io.hpp" namespace ov::util { @@ -33,47 +34,76 @@ void madvise_hint(void* ptr, size_t size) noexcept { madvise(ptr, size, MADV_WILLNEED); } +// 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; // prevents the compiler from optimizing the loop away + volatile uint8_t local = 0; for (auto begin = m_begin; begin < m_end; begin += m_page_size) { local += *begin; } } }; -/** - * @brief A small, bounded, process-wide thread pool used to run page-population tasks. - * - * Rather than spawning dedicated OS threads for every vm_prefetch(_async) call, tasks are - * queued and picked up by whichever of the pool's fixed worker threads becomes free. This - * keeps the number of live threads bounded regardless of how many prefetch calls are made - * concurrently. - */ -class PageTouchThreadPool { +// Thread-safe queue of population jobs. Owns only the queueing/notification concern. +class TaskQueue { public: - static PageTouchThreadPool& instance() { - static PageTouchThreadPool pool; + void push(std::list>&& batch) noexcept { + { + std::lock_guard lock(m_mutex); + m_queue.splice(m_queue.end(), batch); + } + m_cv.notify_all(); + } + + // Blocks until a job is available, or returns false once the queue is stopped and drained. + bool wait_and_pop(std::function& job) noexcept { + std::unique_lock lock(m_mutex); + m_cv.wait(lock, [this] { + return m_stop || !m_queue.empty(); + }); + if (m_queue.empty()) { + return false; + } + job = std::move(m_queue.front()); + m_queue.pop_front(); + return true; + } + + void stop() noexcept { + { + std::lock_guard lock(m_mutex); + m_stop = true; + } + m_cv.notify_all(); + } + +private: + std::mutex m_mutex; + std::condition_variable m_cv; + std::list> m_queue; + bool m_stop = false; +}; + +// Process-wide, bounded pool of worker threads that drain a shared TaskQueue. Keeping the worker +// count bounded means repeated prefetch calls reuse the same threads instead of spawning new ones. +class ThreadPool { +public: + static ThreadPool& instance() { + static ThreadPool pool; return pool; } - PageTouchThreadPool(const PageTouchThreadPool&) = delete; - PageTouchThreadPool& operator=(const PageTouchThreadPool&) = delete; + ThreadPool(const ThreadPool&) = delete; + ThreadPool& operator=(const ThreadPool&) = delete; - /** - * @brief Queues @p jobs for execution on the pool and returns a future per job that - * becomes ready once that job has run. - * - * Enqueueing is all-or-nothing: every throwing operation (task/future creation) happens on - * local state first, and the finished batch is spliced into the shared queue under the lock - * with a non-throwing list splice. If preparation throws (e.g. std::bad_alloc), nothing is - * enqueued and no job runs, so callers never observe a partially-submitted batch whose jobs - * might touch memory the caller believes was never scheduled. - */ + // Enqueueing is all-or-nothing: every throwing operation (task/future creation) happens on + // local state first, and the finished batch is spliced into the queue with a non-throwing + // splice. If preparation throws, nothing is enqueued and no job runs. std::vector> submit(std::vector>&& jobs) { std::vector> futures; futures.reserve(jobs.size()); @@ -85,34 +115,24 @@ class PageTouchThreadPool { (*task)(); }); } - { - std::lock_guard lock(m_mutex); - m_queue.splice(m_queue.end(), pending); // non-throwing: all-or-nothing enqueue - } - m_cv.notify_all(); + m_queue.push(std::move(pending)); return futures; } - size_t worker_count() const noexcept { - return m_workers.size(); - } - private: - PageTouchThreadPool() - : m_workers(std::max(1, std::min(10, std::thread::hardware_concurrency()))) { - for (auto& worker : m_workers) { - worker = std::thread([this]() { + ThreadPool() { + const auto workers = + std::max(1, std::min(max_prefetch_threads, std::thread::hardware_concurrency())); + m_workers.reserve(workers); + for (size_t i = 0; i < workers; ++i) { + m_workers.emplace_back([this]() { worker_loop(); }); } } - ~PageTouchThreadPool() { - { - std::lock_guard lock(m_mutex); - m_stop = true; - } - m_cv.notify_all(); + ~ThreadPool() { + m_queue.stop(); for (auto& worker : m_workers) { if (worker.joinable()) { worker.join(); @@ -120,43 +140,22 @@ class PageTouchThreadPool { } } - void worker_loop() { - for (;;) { - std::function job; - { - std::unique_lock lock(m_mutex); - m_cv.wait(lock, [this]() { - return m_stop || !m_queue.empty(); - }); - if (m_queue.empty()) { - if (m_stop) { - return; // fully drained: safe to let this worker exit. - } - continue; - } - job = std::move(m_queue.front()); - m_queue.pop_front(); - } + void worker_loop() noexcept { + std::function job; + while (m_queue.wait_and_pop(job)) { job(); } } + TaskQueue m_queue; std::vector m_workers; - std::list> m_queue; - std::mutex m_mutex; - std::condition_variable m_cv; - bool m_stop = false; }; -/** - * @brief Splits [ptr, ptr + size) into page-toucher jobs (one per chunk) and submits them to the - * shared @ref PageTouchThreadPool, returning a future per job. - */ -std::vector> submit_page_toucher_tasks(void* ptr, size_t size) { - // ptr and size are guaranteed page-aligned by vm_prefetch's precondition. +// Splits [ptr, ptr + size) into num_threads page-toucher jobs and submits them to the shared pool. +std::vector> submit_page_toucher_tasks(void* ptr, size_t size, size_t num_threads) { const auto page_size = static_cast(get_system_page_size()); - const auto num_threads = PageTouchThreadPool::instance().worker_count(); - const auto chunk_size = std::max(align_size_up(size / num_threads, page_size), 1024 * 1024); + const auto chunk_size = + std::max(align_size_up(size / num_threads, page_size), default_parallel_io_min_chunk); std::vector> jobs; jobs.reserve(ceil_div(size, chunk_size)); @@ -164,7 +163,7 @@ std::vector> submit_page_toucher_tasks(void* ptr, size_t size) for (auto first = reinterpret_cast(ptr), last = first + size; first < last; first += chunk_size) { jobs.emplace_back(PageToucher{first, std::min(first + chunk_size, last), page_size}); } - return PageTouchThreadPool::instance().submit(std::move(jobs)); + return ThreadPool::instance().submit(std::move(jobs)); } } // namespace @@ -199,7 +198,6 @@ void vm_commit(void* ptr, size_t size, std::error_code& ec) noexcept { } void vm_decommit(void* ptr, size_t size) noexcept { - assert(ptr != nullptr && size > 0); #if defined(__linux__) std::ignore = mprotect(ptr, size, PROT_NONE); std::ignore = madvise(ptr, size, MADV_DONTNEED); @@ -212,41 +210,30 @@ void vm_decommit(void* ptr, size_t size) noexcept { } void vm_release(void* ptr, size_t size) noexcept { - assert(ptr != nullptr && size > 0); std::ignore = munmap(ptr, size); } -void vm_prefetch(void* ptr, size_t size, bool fast) noexcept { - assert(ptr != nullptr && size > 0); - // assert if region is not mmap-backed. - - if (fast) { - // Option 1: OS advisory hints — async, low overhead. +void vm_prefetch(void* ptr, size_t size, size_t num_threads) noexcept { + if (num_threads == 0) { + // Advisory-only: let the OS decide, no page touching. + madvise_hint(ptr, size); + return; + } + // Parallel synchronous touch, blocking until every page is resident. Must not run on a pool + // worker thread (it would deadlock waiting on the pool). Prefetching is best-effort, so on + // allocation failure fall back to an advisory hint rather than escaping this noexcept function. + try { + PrefetchToken(submit_page_toucher_tasks(ptr, size, num_threads)).wait(); + } catch (...) { madvise_hint(ptr, size); - } else { - // Option 2: parallel synchronous touch — blocks until every page is resident. - // Must not be called from a PageTouchThreadPool worker thread: it blocks on that shared - // pool, so if every worker were parked here no worker would remain to run the jobs. - // Prefetching is best-effort, so on allocation failure fall back to an advisory hint - // rather than letting an exception escape this noexcept function (which would - // std::terminate). - try { - PrefetchToken(submit_page_toucher_tasks(ptr, size)).wait(); - } catch (...) { - madvise_hint(ptr, size); - } } } PrefetchToken vm_prefetch_async(void* ptr, size_t size) noexcept { - assert(ptr != nullptr && size > 0); - - // Submit the touchers to the shared pool but do not wait on them here — ownership of the - // futures is transferred to the token, whose destructor (or explicit wait()) waits on them. - // Prefetching is best-effort: if submission fails (e.g. std::bad_alloc) fall back to an - // advisory hint and return an empty token instead of throwing out of this noexcept function. + // Ownership of the futures is transferred to the token; it waits on them on destruction or via + // an explicit wait(). If submission fails, fall back to an advisory hint and return an empty token. try { - return PrefetchToken(submit_page_toucher_tasks(ptr, size)); + return PrefetchToken(submit_page_toucher_tasks(ptr, size, prefetch_thread_count(size))); } catch (...) { madvise_hint(ptr, size); return PrefetchToken{}; diff --git a/src/common/util/src/os/lin/lin_mmap_object.cpp b/src/common/util/src/os/lin/lin_mmap_object.cpp index 6a6f5948b0dedc..2e8a5159bcb6ba 100644 --- a/src/common/util/src/os/lin/lin_mmap_object.cpp +++ b/src/common/util/src/os/lin/lin_mmap_object.cpp @@ -16,10 +16,13 @@ #include #include +#include "memory_prefetch.hpp" +#include "openvino/util/common_util.hpp" #include "openvino/util/file_util.hpp" #include "openvino/util/hash_util.hpp" #include "openvino/util/memory.hpp" #include "openvino/util/mmap_object.hpp" +#include "openvino/util/parallel_io.hpp" namespace ov { namespace util { @@ -39,55 +42,6 @@ inline util::AlignedRegion make_mmap_region(size_t offset, size_t size) { const auto page_size = static_cast(util::get_system_page_size()); return util::align_region(static_cast(offset), size, page_size); } - -/** - * @brief Creates a memory region for madvise operations. - * - * @param data The base address of the mapped memory region. - * @param mapping_size The size of the mapped memory region. - * @param offset The offset within the mapped memory region. - * @param size The size of the region. - * @return AlignedRegion The aligned memory region. - */ -inline util::AlignedRegion make_madvise_region(const void* data, size_t mapping_size, size_t offset, size_t size) { - const auto page_size = static_cast(util::get_system_page_size()); - if (data == nullptr || mapping_size == 0 || offset >= mapping_size || size < page_size) { - return {}; - } else { - const auto available = mapping_size - offset; - const auto raw_len = (size == auto_size) ? available : std::min(size, available); - return util::align_region(reinterpret_cast(data) + offset, raw_len, page_size); - } -} - -/** - * @brief Plan computed by @ref make_prefetch_plan, shared between the sync and async - * hint_prefetch() implementations. - */ -struct PrefetchPlan { - uintptr_t m_address = 0; - size_t m_aligned_size = 0; -}; - -/** - * @brief Computes the aligned region and page-aligned size to use for a - * hint_prefetch()/hint_prefetch_async() call. - * - * @param data The base address of the mapped memory region. - * @param mapping_size The size of the mapped memory region. - * @param offset Offset within the mapping where prefetching starts. - * @param size Number of bytes to prefetch. - * @return PrefetchPlan with m_aligned_size == 0 when the requested region is below the - * 4 MiB 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) { - constexpr size_t one_mb = 1024 * 1024; - // Below 4 MiB the overhead of populating pages exceeds the benefit; skip. - if (const auto region = make_madvise_region(data, mapping_size, offset, size); region.m_length > 4 * one_mb) { - return {region.m_address, align_size_up(region.m_length, static_cast(get_system_page_size()))}; - } - return {}; -} } // namespace util class HandleHolder { @@ -135,16 +89,14 @@ class MapHolder final : public MappedMemory { size_t m_size = 0; uint64_t m_id = std::numeric_limits::max(); HandleHolder m_handle; - // Tasks handed off via PrefetchToken::detach(): must be waited on before unmapping, see - // wait_for_pending_prefetch() / ~MapHolder(). + // Tasks adopted from hint_prefetch_async()'s token; joined before unmapping (see ~MapHolder). std::mutex m_pending_prefetch_mutex; std::vector> m_pending_prefetch; void adopt_pending_prefetch(std::vector>&& tasks) { std::lock_guard lock(m_pending_prefetch_mutex); - // Opportunistically reap already-finished futures (non-blocking check via wait_for(0)) - // so the vector doesn't grow without bound if hint_prefetch_async().detach() is called - // repeatedly over the lifetime of this mapping. + // Reap already-finished futures so the vector doesn't grow without bound across repeated + // hint_prefetch_async() calls over this mapping's lifetime. m_pending_prefetch.erase(std::remove_if(m_pending_prefetch.begin(), m_pending_prefetch.end(), [](std::future& task) { @@ -214,8 +166,7 @@ class MapHolder final : public MappedMemory { } ~MapHolder() { - // Detached prefetch tasks (see hint_prefetch_async()) may still be touching this - // mapping's pages; they must complete before the mapping is torn down. + // Detached prefetch tasks may still be touching this mapping's pages; join them first. wait_for_pending_prefetch(); if (m_mapped_view != MAP_FAILED) { munmap(m_mapped_view, m_mapped_view_size); @@ -232,7 +183,7 @@ class MapHolder final : public MappedMemory { void hint_evict(size_t offset, size_t size) noexcept override { if (m_mapped_view != MAP_FAILED) { - if (const auto region = util::make_madvise_region(m_data, m_size, offset, size); region.m_length > 0) { + if (const auto region = util::clamp_align_region(m_data, m_size, offset, size); region.m_length > 0) { std::ignore = madvise(reinterpret_cast(region.m_address), region.m_length, MADV_DONTNEED); } } @@ -240,17 +191,18 @@ class MapHolder final : public MappedMemory { void hint_prefetch(size_t offset, size_t size) override { if (const auto plan = util::make_prefetch_plan(m_data, m_size, offset, size); plan.m_aligned_size) { - util::vm_prefetch(reinterpret_cast(plan.m_address), plan.m_aligned_size, /*fast=*/false); + util::vm_prefetch(reinterpret_cast(plan.m_address), + plan.m_aligned_size, + util::prefetch_thread_count(plan.m_aligned_size)); } } void hint_prefetch_async(size_t offset, size_t size) override { if (const auto plan = util::make_prefetch_plan(m_data, m_size, offset, size); plan.m_aligned_size) { - // Never hand the token back to the caller: adopt its tasks into m_pending_prefetch - // right here, in this same call, before returning. This guarantees ~MapHolder() - // will always join them before munmap, regardless of what the caller does - // afterwards (e.g. dropping its last reference to this object right away) — there - // is no window where the tasks' lifetime is decoupled from this object's lifetime. + // Adopt the token's tasks into m_pending_prefetch within this same call, before + // returning, so ~MapHolder() always joins them before munmap regardless of what the + // caller does next. The token is never handed back, so there is no window where the + // tasks' lifetime is decoupled from this object's. auto token = util::vm_prefetch_async(reinterpret_cast(plan.m_address), plan.m_aligned_size); adopt_pending_prefetch(token.detach()); } diff --git a/src/common/util/src/os/win/win_memory.cpp b/src/common/util/src/os/win/win_memory.cpp index c33890e863fd0e..0cf4ed51ad7ae5 100644 --- a/src/common/util/src/os/win/win_memory.cpp +++ b/src/common/util/src/os/win/win_memory.cpp @@ -10,12 +10,12 @@ #include #include -#include #include #include #include #include +#include "memory_prefetch.hpp" #include "openvino/util/memory.hpp" namespace ov::util { @@ -52,31 +52,26 @@ void vm_commit(void* ptr, size_t size, std::error_code& ec) noexcept { } void vm_decommit(void* ptr, size_t size) noexcept { - assert(ptr != nullptr && size > 0); std::ignore = VirtualFree(ptr, size, MEM_DECOMMIT); } void vm_release(void* ptr, size_t) noexcept { - assert(ptr != nullptr); std::ignore = VirtualFree(ptr, 0, MEM_RELEASE); } -void vm_prefetch(void* ptr, size_t size, bool fast) noexcept { - assert(ptr != nullptr && size > 0); - if (fast) { +void vm_prefetch(void* ptr, size_t size, size_t num_threads) noexcept { + if (num_threads == 0) { + // Advisory-only: let the OS decide, no page touching. WIN32_MEMORY_RANGE_ENTRY entry{ptr, size}; ::PrefetchVirtualMemory(::GetCurrentProcess(), 1, &entry, 0); } else { - // blocks until every page has been faulted in. - const auto num_threads = std::max(1, std::min(10, std::thread::hardware_concurrency())); + // Blocks until every page has been faulted in. populate_pages(ptr, size, num_threads); } } PrefetchToken vm_prefetch_async(void* ptr, size_t size) noexcept { - assert(ptr != nullptr && size > 0); - // CVS-186579 - // No background work is started on Windows; mirrors the vm_prefetch() no-op stub. + // CVS-186579: no background work is started on Windows yet; mirrors the vm_prefetch() stub. return {}; } diff --git a/src/common/util/src/os/win/win_mmap_object.cpp b/src/common/util/src/os/win/win_mmap_object.cpp index c4d27adbc0be50..dff759a191af99 100644 --- a/src/common/util/src/os/win/win_mmap_object.cpp +++ b/src/common/util/src/os/win/win_mmap_object.cpp @@ -10,10 +10,13 @@ #include #include +#include "memory_prefetch.hpp" +#include "openvino/util/common_util.hpp" #include "openvino/util/file_util.hpp" #include "openvino/util/hash_util.hpp" #include "openvino/util/memory.hpp" #include "openvino/util/mmap_object.hpp" +#include "openvino/util/parallel_io.hpp" // clang-format off #ifndef NOMINMAX @@ -745,29 +748,11 @@ bool MapHolder::try_remap_slot(uintptr_t fault_addr) { return remap_placeholder(proc, placeholder_base, placeholder_size); } -namespace { - -// 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. -util::AlignedRegion clamp_align_region(const void* data, size_t mapping_size, size_t offset, size_t size) noexcept { - const auto page_size = static_cast(util::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 util::align_region(reinterpret_cast(data) + offset, raw_len, page_size); -} - -} // namespace - void MapHolder::hint_prefetch(size_t offset, size_t size) { - // Below 4 MiB the overhead of spawning threads exceeds the benefit; skip. - if (const auto region = clamp_align_region(m_data, m_size, offset, size); region.m_length > 4 * util::one_mib) { - const auto aligned_size = - util::align_size_up(region.m_length, static_cast(util::get_system_page_size())); - util::vm_prefetch(reinterpret_cast(region.m_address), aligned_size, /*fast=*/false); + if (const auto plan = util::make_prefetch_plan(m_data, m_size, offset, size); plan.m_aligned_size) { + util::vm_prefetch(reinterpret_cast(plan.m_address), + plan.m_aligned_size, + util::prefetch_thread_count(plan.m_aligned_size)); } } diff --git a/src/common/util/src/parallel_read_streambuf.cpp b/src/common/util/src/parallel_read_streambuf.cpp index 5102193310508f..3e6d2af700cadc 100644 --- a/src/common/util/src/parallel_read_streambuf.cpp +++ b/src/common/util/src/parallel_read_streambuf.cpp @@ -268,8 +268,7 @@ bool ParallelReadStreamBuf::single_read(char* dst, size_t size, size_t file_offs // Parallel positional read bool ParallelReadStreamBuf::parallel_read(char* dst, size_t size, size_t file_offset) { const size_t hw_threads = (std::max)(size_t{1}, static_cast(std::thread::hardware_concurrency())); - const size_t max_by_size = size / (1024 * 1024); // 1 thread per MB - const size_t num_threads = (std::max)(size_t{1}, (std::min)(hw_threads, max_by_size)); + const size_t num_threads = split_chunk_count(size, one_mib, hw_threads); // 1 thread per MB, capped by HW if (num_threads == 1) { return single_read(dst, size, file_offset); diff --git a/src/core/tests/file_load_benchmark.cpp b/src/core/tests/file_load_benchmark.cpp index cc78e057410f46..4f25f170ca82f7 100644 --- a/src/core/tests/file_load_benchmark.cpp +++ b/src/core/tests/file_load_benchmark.cpp @@ -204,7 +204,7 @@ namespace strategy { // Note: the mmap destructor (munmap + close) runs inside the timed window; void sync_vm_prefetch_mem_lock(const std::filesystem::path& path, size_t /*file_size*/) { auto mapped = load_mmap_object(path); - util::vm_prefetch(mapped->data(), mapped->size(), /*fast=*/false); + mapped->hint_prefetch(); ensure_memory_resident(mapped); // should be near no-op and just lock/unlock resident pages } @@ -241,7 +241,7 @@ void compute_over_mapped(const std::shared_ptr& mapped) { void parallel_loop_sync_then_memcpy(const std::filesystem::path& path, size_t file_size) { auto mapped = load_mmap_object(path); - util::vm_prefetch(mapped->data(), mapped->size(), /*fast=*/false); + mapped->hint_prefetch(); constexpr size_t chunk_size = 128 * util::one_mib; std::vector buffer(std::min(chunk_size, file_size)); volatile char sink = 0; diff --git a/src/core/tests/mmap_object.cpp b/src/core/tests/mmap_object.cpp index cd5f1190f4d47d..a3adb211d2f12b 100644 --- a/src/core/tests/mmap_object.cpp +++ b/src/core/tests/mmap_object.cpp @@ -12,6 +12,7 @@ #include #include #include +#include #include #include #include @@ -610,50 +611,52 @@ TEST_F(HintPrefetchAsyncTest, data_correct_immediately_after_call) { EXPECT_EQ(read_mapped(*mapped), data); } -TEST(PrefetchTokenTest, move_transfers_ownership) { - // MappedMemory::hint_prefetch_async() no longer exposes a util::PrefetchToken at all, so - // exercise PrefetchToken's move semantics directly against the lower-level free function - // instead, using a page-aligned buffer that is intentionally never freed for the lifetime - // of the process (mirrors detach_releases_futures_to_caller below). - const auto page = static_cast(util::get_system_page_size()); - const size_t buf_size = util::align_size_down(8 * 1024 * 1024, page); - static void* leaked_buffer = util::aligned_alloc(buf_size, page); - ASSERT_NE(leaked_buffer, nullptr); - - auto token = util::vm_prefetch_async(leaked_buffer, buf_size); - ASSERT_TRUE(static_cast(token)); - - util::PrefetchToken moved(std::move(token)); - EXPECT_FALSE(static_cast(token)); // moved-from token is empty - EXPECT_TRUE(static_cast(moved)); // ownership transferred to the new token - - util::PrefetchToken move_assigned; - move_assigned = std::move(moved); - EXPECT_FALSE(static_cast(moved)); // moved-from token is empty again - EXPECT_TRUE(static_cast(move_assigned)); // ownership transferred via move-assignment - - move_assigned.wait(); - EXPECT_FALSE(static_cast(move_assigned)); +TEST_F(HintPrefetchAsyncTest, random_access_after_async_is_correct) { + // Reading pages in a shuffled order concurrently with in-flight background population must + // always return the correct bytes, whether or not a given page has been faulted in yet. + m_file_path = std::filesystem::path(utils::generateTestFilePrefix() + "_prefetch_async_random.bin"); + constexpr size_t file_size = 8 * 1024 * 1024; // 8 MiB (above the parallel-I/O threshold) + const auto data = make_pattern(file_size); + write_file(data); + + auto mapped = load_mmap_object(m_file_path); + ASSERT_NE(mapped, nullptr); + + mapped->hint_prefetch_async(); + + const size_t page = static_cast(util::get_system_page_size()); + const size_t total_pages = (file_size + page - 1) / page; + std::vector order(total_pages); + std::iota(order.begin(), order.end(), size_t{0}); + std::shuffle(order.begin(), order.end(), std::mt19937{12345}); + + const auto* base = reinterpret_cast(mapped->data()); + for (const size_t idx : order) { + const size_t begin = idx * page; + const size_t end = std::min(begin + page, file_size); + for (size_t i = begin; i < end; ++i) { + ASSERT_EQ(base[i], data[i]) << "mismatch at offset " << i; + } + } } -TEST_F(HintPrefetchAsyncTest, detach_releases_futures_to_caller) { - // Use a page-aligned buffer that is intentionally never freed for the lifetime of the - // process, so that discarding the futures returned by detach() (fire-and-forget, like - // std::thread::detach()) cannot cause a real use-after-free, regardless of when the tasks - // actually finish running. - const auto page = static_cast(util::get_system_page_size()); - const size_t buf_size = util::align_size_down(8 * 1024 * 1024, page); - static void* leaked_buffer = util::aligned_alloc(buf_size, page); - ASSERT_NE(leaked_buffer, nullptr); - - auto token = util::vm_prefetch_async(leaked_buffer, buf_size); - ASSERT_TRUE(static_cast(token)); - - // detach() transfers ownership of the futures to the caller and empties the token; the - // caller here simply discards them (fire-and-forget). - auto tasks = token.detach(); - EXPECT_FALSE(tasks.empty()); - EXPECT_FALSE(static_cast(token)); +TEST_F(HintPrefetchAsyncTest, backward_access_after_async_is_correct) { + // Reading the mapping back-to-front (opposite the population's forward stride) concurrently + // with in-flight background population must still return the correct bytes. + m_file_path = std::filesystem::path(utils::generateTestFilePrefix() + "_prefetch_async_backward.bin"); + constexpr size_t file_size = 8 * 1024 * 1024; // 8 MiB (above the parallel-I/O threshold) + const auto data = make_pattern(file_size); + write_file(data); + + auto mapped = load_mmap_object(m_file_path); + ASSERT_NE(mapped, nullptr); + + mapped->hint_prefetch_async(); + + const auto* base = reinterpret_cast(mapped->data()); + for (size_t i = file_size; i-- > 0;) { + ASSERT_EQ(base[i], data[i]) << "mismatch at offset " << i; + } } } // namespace ov::test diff --git a/src/tests/test_utils/common_test_utils/tests/memory_test.cpp b/src/tests/test_utils/common_test_utils/tests/memory_test.cpp index 5a1d4c23b5b371..00dbb4762f2cff 100644 --- a/src/tests/test_utils/common_test_utils/tests/memory_test.cpp +++ b/src/tests/test_utils/common_test_utils/tests/memory_test.cpp @@ -18,8 +18,6 @@ namespace ov::test { -using testing::ElementsAreArray; - // Verify the constexpr contract at compile time for the most common alignments. static_assert(ov::util::align_size_up(0, 64) == 0); static_assert(ov::util::align_size_up(1, 64) == 64); @@ -157,33 +155,4 @@ TEST_F(AlignedAllocTest, free_nullptr_is_noop) { EXPECT_NO_FATAL_FAILURE(util::aligned_free(nullptr)); } -class VmPrefetchMappedFileTest : public testing::TestWithParam { -protected: - std::filesystem::path m_file_path; - - void TearDown() override { - std::filesystem::remove(m_file_path); - } -}; - -TEST_P(VmPrefetchMappedFileTest, prefetch_faults_in_mapped_file_and_preserves_data) { - const bool fast = GetParam(); - const size_t size = 64 * util::min_page_alignment; - - const auto expected = utils::make_modulo_sequence_pattern(size); - - m_file_path = utils::generateTestFilePrefix() + "_vm_prefetch.bin"; - ov::util::save_binary(m_file_path, expected.data(), expected.size()); - - auto mapped = load_mmap_object(m_file_path); - ASSERT_NE(mapped, nullptr); - ASSERT_EQ(mapped->size(), size); - - EXPECT_NO_FATAL_FAILURE(util::vm_prefetch(mapped->data(), mapped->size(), fast)); - - EXPECT_THAT(expected, ElementsAreArray(reinterpret_cast(mapped->data()), mapped->size())); -} - -INSTANTIATE_TEST_SUITE_P(Fast, VmPrefetchMappedFileTest, testing::Bool()); - } // namespace ov::test From 596d001cb1244503d65bed8bbb34075d0fcd9e0c Mon Sep 17 00:00:00 2001 From: Oleg Pipikin Date: Wed, 15 Jul 2026 15:38:13 +0200 Subject: [PATCH 3/5] fixes --- src/common/util/src/memory.cpp | 20 +----- src/common/util/src/memory_prefetch.hpp | 15 +++++ src/common/util/src/os/lin/lin_memory.cpp | 67 +++++-------------- .../util/src/os/lin/lin_mmap_object.cpp | 4 -- src/common/util/src/os/win/win_memory.cpp | 1 - 5 files changed, 33 insertions(+), 74 deletions(-) diff --git a/src/common/util/src/memory.cpp b/src/common/util/src/memory.cpp index 203610e4ac47ab..11d509709c3b65 100644 --- a/src/common/util/src/memory.cpp +++ b/src/common/util/src/memory.cpp @@ -9,31 +9,13 @@ #include #include +#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 { - -// 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; - } - } -}; - -} // namespace - void populate_pages(void* ptr, size_t size, size_t num_threads) noexcept { const auto page_size = static_cast(get_system_page_size()); const auto chunk_size = std::max(align_size_up(size / num_threads, page_size), one_mib); diff --git a/src/common/util/src/memory_prefetch.hpp b/src/common/util/src/memory_prefetch.hpp index b5e4aea3f166a1..4f093ff91fafcc 100644 --- a/src/common/util/src/memory_prefetch.hpp +++ b/src/common/util/src/memory_prefetch.hpp @@ -17,6 +17,21 @@ 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; /** diff --git a/src/common/util/src/os/lin/lin_memory.cpp b/src/common/util/src/os/lin/lin_memory.cpp index 7f32eeb07d1e0f..8c56e3799404e2 100644 --- a/src/common/util/src/os/lin/lin_memory.cpp +++ b/src/common/util/src/os/lin/lin_memory.cpp @@ -10,7 +10,6 @@ #include #include #include -#include #include #include #include @@ -20,7 +19,7 @@ #include #include "memory_prefetch.hpp" -#include "openvino/util/common_util.hpp" +#include "openvino/util/math_util.hpp" #include "openvino/util/memory.hpp" #include "openvino/util/mmap_object.hpp" #include "openvino/util/parallel_io.hpp" @@ -34,22 +33,6 @@ void madvise_hint(void* ptr, size_t size) noexcept { madvise(ptr, size, MADV_WILLNEED); } -// 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; - } - } -}; - -// Thread-safe queue of population jobs. Owns only the queueing/notification concern. class TaskQueue { public: void push(std::list>&& batch) noexcept { @@ -89,8 +72,6 @@ class TaskQueue { bool m_stop = false; }; -// Process-wide, bounded pool of worker threads that drain a shared TaskQueue. Keeping the worker -// count bounded means repeated prefetch calls reuse the same threads instead of spawning new ones. class ThreadPool { public: static ThreadPool& instance() { @@ -101,9 +82,6 @@ class ThreadPool { ThreadPool(const ThreadPool&) = delete; ThreadPool& operator=(const ThreadPool&) = delete; - // Enqueueing is all-or-nothing: every throwing operation (task/future creation) happens on - // local state first, and the finished batch is spliced into the queue with a non-throwing - // splice. If preparation throws, nothing is enqueued and no job runs. std::vector> submit(std::vector>&& jobs) { std::vector> futures; futures.reserve(jobs.size()); @@ -151,19 +129,23 @@ class ThreadPool { std::vector m_workers; }; -// Splits [ptr, ptr + size) into num_threads page-toucher jobs and submits them to the shared pool. -std::vector> submit_page_toucher_tasks(void* ptr, size_t size, size_t num_threads) { - const auto page_size = static_cast(get_system_page_size()); - const auto chunk_size = - std::max(align_size_up(size / num_threads, page_size), default_parallel_io_min_chunk); +std::vector> submit_page_toucher_tasks(void* ptr, size_t size, size_t num_threads) noexcept { + try { + const auto page_size = static_cast(get_system_page_size()); + const auto chunk_size = + std::max(align_size_up(size / num_threads, page_size), default_parallel_io_min_chunk); - std::vector> jobs; - jobs.reserve(ceil_div(size, chunk_size)); + std::vector> jobs; + jobs.reserve(ceil_div(size, chunk_size)); - for (auto first = reinterpret_cast(ptr), last = first + size; first < last; first += chunk_size) { - jobs.emplace_back(PageToucher{first, std::min(first + chunk_size, last), page_size}); + for (auto first = reinterpret_cast(ptr), last = first + size; first < last; + first += chunk_size) { + jobs.emplace_back(PageToucher{first, std::min(first + chunk_size, last), page_size}); + } + return ThreadPool::instance().submit(std::move(jobs)); + } catch (...) { + return {}; } - return ThreadPool::instance().submit(std::move(jobs)); } } // namespace @@ -215,29 +197,14 @@ void vm_release(void* ptr, size_t size) noexcept { void vm_prefetch(void* ptr, size_t size, size_t num_threads) noexcept { if (num_threads == 0) { - // Advisory-only: let the OS decide, no page touching. madvise_hint(ptr, size); return; } - // Parallel synchronous touch, blocking until every page is resident. Must not run on a pool - // worker thread (it would deadlock waiting on the pool). Prefetching is best-effort, so on - // allocation failure fall back to an advisory hint rather than escaping this noexcept function. - try { - PrefetchToken(submit_page_toucher_tasks(ptr, size, num_threads)).wait(); - } catch (...) { - madvise_hint(ptr, size); - } + PrefetchToken(submit_page_toucher_tasks(ptr, size, num_threads)).wait(); } PrefetchToken vm_prefetch_async(void* ptr, size_t size) noexcept { - // Ownership of the futures is transferred to the token; it waits on them on destruction or via - // an explicit wait(). If submission fails, fall back to an advisory hint and return an empty token. - try { - return PrefetchToken(submit_page_toucher_tasks(ptr, size, prefetch_thread_count(size))); - } catch (...) { - madvise_hint(ptr, size); - return PrefetchToken{}; - } + return PrefetchToken(submit_page_toucher_tasks(ptr, size, prefetch_thread_count(size))); } } // namespace ov::util diff --git a/src/common/util/src/os/lin/lin_mmap_object.cpp b/src/common/util/src/os/lin/lin_mmap_object.cpp index 2e8a5159bcb6ba..f38f56016b3560 100644 --- a/src/common/util/src/os/lin/lin_mmap_object.cpp +++ b/src/common/util/src/os/lin/lin_mmap_object.cpp @@ -199,10 +199,6 @@ class MapHolder final : public MappedMemory { void hint_prefetch_async(size_t offset, size_t size) override { if (const auto plan = util::make_prefetch_plan(m_data, m_size, offset, size); plan.m_aligned_size) { - // Adopt the token's tasks into m_pending_prefetch within this same call, before - // returning, so ~MapHolder() always joins them before munmap regardless of what the - // caller does next. The token is never handed back, so there is no window where the - // tasks' lifetime is decoupled from this object's. auto token = util::vm_prefetch_async(reinterpret_cast(plan.m_address), plan.m_aligned_size); adopt_pending_prefetch(token.detach()); } diff --git a/src/common/util/src/os/win/win_memory.cpp b/src/common/util/src/os/win/win_memory.cpp index 0cf4ed51ad7ae5..e91ca42afd4489 100644 --- a/src/common/util/src/os/win/win_memory.cpp +++ b/src/common/util/src/os/win/win_memory.cpp @@ -71,7 +71,6 @@ void vm_prefetch(void* ptr, size_t size, size_t num_threads) noexcept { } PrefetchToken vm_prefetch_async(void* ptr, size_t size) noexcept { - // CVS-186579: no background work is started on Windows yet; mirrors the vm_prefetch() stub. return {}; } From ab5e33004626d5f5b20dbb5e0fd36718fc24c471 Mon Sep 17 00:00:00 2001 From: Oleg Pipikin Date: Wed, 15 Jul 2026 17:07:04 +0200 Subject: [PATCH 4/5] fix --- src/common/util/src/os/lin/lin_mmap_object.cpp | 1 - src/common/util/src/os/win/win_mmap_object.cpp | 1 - src/core/tests/file_load_benchmark.cpp | 13 ------------- 3 files changed, 15 deletions(-) diff --git a/src/common/util/src/os/lin/lin_mmap_object.cpp b/src/common/util/src/os/lin/lin_mmap_object.cpp index f38f56016b3560..ef4e7ebb5abcd6 100644 --- a/src/common/util/src/os/lin/lin_mmap_object.cpp +++ b/src/common/util/src/os/lin/lin_mmap_object.cpp @@ -17,7 +17,6 @@ #include #include "memory_prefetch.hpp" -#include "openvino/util/common_util.hpp" #include "openvino/util/file_util.hpp" #include "openvino/util/hash_util.hpp" #include "openvino/util/memory.hpp" diff --git a/src/common/util/src/os/win/win_mmap_object.cpp b/src/common/util/src/os/win/win_mmap_object.cpp index dff759a191af99..d7081da2f38b5a 100644 --- a/src/common/util/src/os/win/win_mmap_object.cpp +++ b/src/common/util/src/os/win/win_mmap_object.cpp @@ -11,7 +11,6 @@ #include #include "memory_prefetch.hpp" -#include "openvino/util/common_util.hpp" #include "openvino/util/file_util.hpp" #include "openvino/util/hash_util.hpp" #include "openvino/util/memory.hpp" diff --git a/src/core/tests/file_load_benchmark.cpp b/src/core/tests/file_load_benchmark.cpp index 4f25f170ca82f7..63af1443fc1eef 100644 --- a/src/core/tests/file_load_benchmark.cpp +++ b/src/core/tests/file_load_benchmark.cpp @@ -239,19 +239,6 @@ void compute_over_mapped(const std::shared_ptr& mapped) { (void)sink; } -void parallel_loop_sync_then_memcpy(const std::filesystem::path& path, size_t file_size) { - auto mapped = load_mmap_object(path); - mapped->hint_prefetch(); - constexpr size_t chunk_size = 128 * util::one_mib; - std::vector buffer(std::min(chunk_size, file_size)); - volatile char sink = 0; - for (size_t offset = 0; offset < file_size; offset += chunk_size) { - const size_t copy_size = std::min(chunk_size, file_size - offset); - std::memcpy(buffer.data(), mapped->data() + offset, copy_size); - sink += buffer[0] + buffer[copy_size / 2] + buffer[copy_size - 1]; // prevents optimization - } -} - void mmap_then_compute(const std::filesystem::path& path, size_t /*file_size*/) { auto mapped = load_mmap_object(path); compute_over_mapped(mapped); From 9a894ff0ba5896d510eb56540d22935799804e37 Mon Sep 17 00:00:00 2001 From: Oleg Pipikin Date: Thu, 16 Jul 2026 09:55:26 +0200 Subject: [PATCH 5/5] Fix clang-format --- src/common/util/include/openvino/util/mmap_object.hpp | 1 - src/common/util/src/memory_prefetch.hpp | 2 -- src/common/util/src/os/lin/lin_mmap_object.cpp | 5 ++--- src/core/tests/mmap_object.cpp | 4 ++-- 4 files changed, 4 insertions(+), 8 deletions(-) diff --git a/src/common/util/include/openvino/util/mmap_object.hpp b/src/common/util/include/openvino/util/mmap_object.hpp index 944ac145c4af63..61c8fff10c6cd8 100644 --- a/src/common/util/include/openvino/util/mmap_object.hpp +++ b/src/common/util/include/openvino/util/mmap_object.hpp @@ -62,7 +62,6 @@ class MappedMemory { virtual void hint_prefetch_async(size_t offset = 0, size_t size = auto_size) {} }; - /** * @brief Returns mapped memory for a file from provided path. * Instead of reading files, we can map the memory via mmap for Linux diff --git a/src/common/util/src/memory_prefetch.hpp b/src/common/util/src/memory_prefetch.hpp index 4f093ff91fafcc..563cd58cd16ef0 100644 --- a/src/common/util/src/memory_prefetch.hpp +++ b/src/common/util/src/memory_prefetch.hpp @@ -52,7 +52,6 @@ void vm_prefetch(void* ptr, size_t size, size_t num_threads) noexcept; */ 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. * @@ -142,7 +141,6 @@ inline PrefetchPlan make_prefetch_plan(const void* data, size_t mapping_size, si return {}; } - /** @brief Upper bound on the shared page-population pool's worker threads. */ inline constexpr size_t max_prefetch_threads = 8; diff --git a/src/common/util/src/os/lin/lin_mmap_object.cpp b/src/common/util/src/os/lin/lin_mmap_object.cpp index ef4e7ebb5abcd6..b05f57578fd3a5 100644 --- a/src/common/util/src/os/lin/lin_mmap_object.cpp +++ b/src/common/util/src/os/lin/lin_mmap_object.cpp @@ -99,9 +99,8 @@ class MapHolder final : public MappedMemory { m_pending_prefetch.erase(std::remove_if(m_pending_prefetch.begin(), m_pending_prefetch.end(), [](std::future& task) { - return !task.valid() || - task.wait_for(std::chrono::seconds(0)) == - std::future_status::ready; + return !task.valid() || task.wait_for(std::chrono::seconds(0)) == + std::future_status::ready; }), m_pending_prefetch.end()); m_pending_prefetch.insert(m_pending_prefetch.end(), diff --git a/src/core/tests/mmap_object.cpp b/src/core/tests/mmap_object.cpp index a3adb211d2f12b..4ea3885f985299 100644 --- a/src/core/tests/mmap_object.cpp +++ b/src/core/tests/mmap_object.cpp @@ -577,7 +577,8 @@ TEST_F(HintPrefetchAsyncTest, partial_region_populated_and_correct) { mapped->hint_prefetch_async(prefetch_offset, prefetch_size); - const size_t pages_resident = wait_for_resident_pages(mapped->data() + prefetch_offset, prefetch_size, region_pages); + const size_t pages_resident = + wait_for_resident_pages(mapped->data() + prefetch_offset, prefetch_size, region_pages); EXPECT_EQ(pages_resident, region_pages) << "Expected the requested region to be fully resident."; EXPECT_EQ(read_mapped(*mapped), data); @@ -660,4 +661,3 @@ TEST_F(HintPrefetchAsyncTest, backward_access_after_async_is_correct) { } } // namespace ov::test -