Optimize dGPU path

Author: riverlijunjie

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

@@ -21,8 +21,8 @@
 #    ifndef WIN32_LEAN_AND_MEAN
 #        define WIN32_LEAN_AND_MEAN
 #    endif
-#    include <windows.h>
 #    include <psapi.h>
+#    include <windows.h>
 #else
 #    include <sys/mman.h>
 #endif
@@ -78,21 +78,18 @@ class ParallelMemStreamBuf : public std::streambuf {
             MEMORY_BASIC_INFORMATION mbi{};
             if (VirtualQuery(data, &mbi, sizeof(mbi)) && mbi.Type == MEM_MAPPED) {
                 wchar_t dev_path[MAX_PATH] = {};
-                if (GetMappedFileNameW(GetCurrentProcess(),
-                                       const_cast<void*>(data),
-                                       dev_path,
-                                       MAX_PATH) > 0) {
+                if (GetMappedFileNameW(GetCurrentProcess(), const_cast<void*>(data), dev_path, MAX_PATH) > 0) {
                     // Convert device path (\Device\HarddiskVolume3\...) to Win32 path.
                     wchar_t win32_path[MAX_PATH] = {};
                     if (resolve_device_path(dev_path, win32_path, MAX_PATH)) {
                         // Compute the file offset: the region base may be at a
                         // different page boundary than the pointer we received.
                         // AllocationBase is the start of the mapped view.
                         const std::streamoff file_offset =
-                            reinterpret_cast<const char*>(data) -
-                            reinterpret_cast<const char*>(mbi.AllocationBase);
-                        m_file_buf = std::make_unique<ParallelReadStreamBuf>(
-                            std::filesystem::path(win32_path), file_offset, threshold);
+                            reinterpret_cast<const char*>(data) - reinterpret_cast<const char*>(mbi.AllocationBase);
+                        m_file_buf = std::make_unique<ParallelReadStreamBuf>(std::filesystem::path(win32_path),
+                                                                             file_offset,
+                                                                             threshold);
                     }
                 }
             }
@@ -298,9 +295,7 @@ class ParallelMemStreamBuf : public std::streambuf {
     // Parse /proc/self/maps to find the file backing an mmap address.
     // Returns true and fills out_path/out_offset if the address is inside
     // a named file mapping (i.e. not anonymous / [stack] / [heap]).
-    static bool get_mmap_file_info(const void* addr,
-                                   std::filesystem::path& out_path,
-                                   std::streamoff& out_offset) {
+    static bool get_mmap_file_info(const void* addr, std::filesystem::path& out_path, std::streamoff& out_offset) {
         std::ifstream maps_file("/proc/self/maps");
         if (!maps_file.is_open())
             return false;
@@ -316,21 +311,17 @@ class ParallelMemStreamBuf : public std::streambuf {
             const auto dash = addr_range.find('-');
             if (dash == std::string::npos)
                 continue;
-            const auto range_start =
-                static_cast<uintptr_t>(std::stoull(addr_range.substr(0, dash), nullptr, 16));
-            const auto range_end =
-                static_cast<uintptr_t>(std::stoull(addr_range.substr(dash + 1), nullptr, 16));
+            const auto range_start = static_cast<uintptr_t>(std::stoull(addr_range.substr(0, dash), nullptr, 16));
+            const auto range_end = static_cast<uintptr_t>(std::stoull(addr_range.substr(dash + 1), nullptr, 16));
             if (addr_val < range_start || addr_val >= range_end)
                 continue;
             // Read optional pathname
             std::string path;
             if (!(iss >> path) || path.empty() || path[0] != '/')
                 return false;  // anonymous or special region, no benefit
             out_path = path;
-            const auto map_offset =
-                static_cast<std::streamoff>(std::stoull(offset_str, nullptr, 16));
-            out_offset = map_offset +
-                         static_cast<std::streamoff>(addr_val - range_start);
+            const auto map_offset = static_cast<std::streamoff>(std::stoull(offset_str, nullptr, 16));
+            out_offset = map_offset + static_cast<std::streamoff>(addr_val - range_start);
             return true;
         }
         return false;

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

@@ -9,6 +9,7 @@
 #include <atomic>
 #include <cstring>
 #include <filesystem>
+#include <fstream>
 #include <future>
 #include <iostream>
 #include <stdexcept>
@@ -166,9 +167,8 @@ class ParallelReadStreamBuf : public std::streambuf {
         }
         // Read a batch of up to UNDERFLOW_BUF bytes so that character-by-character
         // consumers (std::getline, operator>>) don't issue one pread per char.
-        const size_t to_read = static_cast<size_t>(
-            std::min(static_cast<std::streamoff>(UNDERFLOW_BUF),
-                     m_file_size - m_file_offset));
+        const size_t to_read =
+            static_cast<size_t>(std::min(static_cast<std::streamoff>(UNDERFLOW_BUF), m_file_size - m_file_offset));
         if (!single_read(m_underflow_buf.data(), to_read, static_cast<size_t>(m_file_offset))) {
             return traits_type::eof();
         }
@@ -335,20 +335,22 @@ class ParallelReadStreamBuf : public std::streambuf {
                 CloseHandle(t_handle);
             }));
 #else
-            const int fd = m_fd;
-            futures.emplace_back(std::async(std::launch::async, [fd, thread_file_offset, ptr, read_size, &success] {
-                char* cur = ptr;
-                size_t remaining = read_size;
-                off_t cur_file_offset = static_cast<off_t>(thread_file_offset);
-                while (remaining > 0 && success) {
-                    const ssize_t n = ::pread(fd, cur, remaining, cur_file_offset);
-                    if (n <= 0) {
-                        success = false;
-                        break;
-                    }
-                    cur += n;
-                    cur_file_offset += n;
-                    remaining -= static_cast<size_t>(n);
+            // Each worker opens its own std::ifstream so that Linux's per-file-
+            // description readahead state (file_ra_state / f_ra) is independent
+            // per thread. Sharing a single fd causes concurrent pread() calls to
+            // corrupt each other's sequential readahead prediction, collapsing
+            // throughput from ~3.5 GB/s sequential to ~0.5 GB/s.
+            const std::filesystem::path t_path = m_path;
+            futures.emplace_back(std::async(std::launch::async, [t_path, thread_file_offset, ptr, read_size, &success] {
+                std::ifstream t_ifs(t_path, std::ios::binary);
+                if (!t_ifs.is_open()) {
+                    success = false;
+                    return;
+                }
+                t_ifs.seekg(static_cast<std::streamoff>(thread_file_offset), std::ios::beg);
+                t_ifs.read(ptr, static_cast<std::streamsize>(read_size));
+                if (!t_ifs.good()) {
+                    success = false;
                 }
             }));
 #endif

src/inference/src/cache_manager.hpp

@@ -18,10 +18,10 @@
 
 #include "openvino/runtime/icache_manager.hpp"
 #include "openvino/runtime/shared_buffer.hpp"
-#include "openvino/util/parallel_read_streambuf.hpp"
 #include "openvino/runtime/tensor.hpp"
 #include "openvino/util/file_util.hpp"
 #include "openvino/util/mmap_object.hpp"
+#include "openvino/util/parallel_read_streambuf.hpp"
 
 namespace ov {
 

src/plugins/intel_gpu/include/intel_gpu/primitives/data.hpp

@@ -422,13 +422,65 @@ struct data : public primitive_base<data> {
             if (is_alloc_host_accessible(_allocation_type)) {
                 ib >> make_data(mem->buffer_ptr(), data_size);
             } else {
-                const size_t DATA_BLOCK_SIZE = 2 * 1024 * 1024;
-                auto& strm = ib.get_engine().get_service_stream();
+                const size_t DATA_BLOCK_SIZE = 4 * 1024 * 1024;
+                auto& eng = ib.get_engine();
+                auto& strm = eng.get_service_stream();
                 if (data_size < DATA_BLOCK_SIZE || output_layout.format.is_image_2d()) {
                     std::vector<uint8_t> _buf(data_size);
                     ib >> make_data(_buf.data(), data_size);
                     mem->copy_from(strm, _buf.data());
+                } else if (eng.supports_allocation(allocation_type::usm_host)) {
+                    // Use USM host staging buffers so the driver can DMA directly
+                    // from pinned host memory to device, avoiding the hidden
+                    // pageable-to-pinned bounce copy that std::vector staging incurs.
+                    // The pipeline is the same ping-pong as the fallback below:
+                    //   CPU reads into host_bufA  while  GPU copies from host_bufB
+                    // Safety: each staging buffer is reused every other iteration;
+                    // the wait() before re-submitting a copy ensures the previous
+                    // GPU read from that buffer has finished before the CPU writes
+                    // new data into it (the OTHER buffer's event is waited, but that
+                    // event was submitted after waiting for THIS buffer's previous
+                    // event, so THIS buffer is always free by the time we reuse it).
+                    const layout staging_layout{{static_cast<int64_t>(DATA_BLOCK_SIZE)},
+                                                data_types::u8, format::bfyx};
+                    auto host_buf1 = eng.allocate_memory(staging_layout, allocation_type::usm_host, false);
+                    auto host_buf2 = eng.allocate_memory(staging_layout, allocation_type::usm_host, false);
+                    bool buf_flag = true;
+                    event::ptr ev1, ev2;
+                    ev1 = ev2 = nullptr;
+                    size_t dst_offset = 0;
+                    while (dst_offset < data_size) {
+                        const bool is_blocking = false;
+                        const size_t src_offset = 0;
+                        size_t copy_size =
+                            (data_size > (dst_offset + DATA_BLOCK_SIZE)) ? DATA_BLOCK_SIZE : (data_size - dst_offset);
+                        if (buf_flag) {
+                            ib >> make_data(static_cast<uint8_t*>(host_buf1->buffer_ptr()), copy_size);
+                            if (ev2 != nullptr) {
+                                ev2->wait();
+                                ev2 = nullptr;
+                            }
+                            ev1 = mem->copy_from(strm, *host_buf1, src_offset, dst_offset, copy_size, is_blocking);
+                        } else {
+                            ib >> make_data(static_cast<uint8_t*>(host_buf2->buffer_ptr()), copy_size);
+                            if (ev1 != nullptr) {
+                                ev1->wait();
+                                ev1 = nullptr;
+                            }
+                            ev2 = mem->copy_from(strm, *host_buf2, src_offset, dst_offset, copy_size, is_blocking);
+                        }
+                        dst_offset += DATA_BLOCK_SIZE;
+                        buf_flag = !buf_flag;
+                    }
+                    if (ev2 != nullptr) {
+                        ev2->wait();
+                    }
+                    if (ev1 != nullptr) {
+                        ev1->wait();
+                    }
                 } else {
+                    // Fallback for devices that do not support USM host allocation:
+                    // use pageable std::vector staging buffers (original behaviour).
                     std::vector<uint8_t> _buf1(DATA_BLOCK_SIZE);
                     std::vector<uint8_t> _buf2(DATA_BLOCK_SIZE);
                     bool buf_flag = true;