feat(dat): async ReadFile with a callback, delivered by the dat worker

Add GwDatArchive::ReadFileAsync(file_id, callback, stream_id): it queues the
request on an internal list and returns immediately. The dat worker, which
now wakes on a short poll interval as well as on tasks, scans the list each
pass - when a file resolves it fires the callback with the decompressed
bytes and drops the entry; if the file hasn't appeared within ~1 minute it
fires the callback once with an empty vector and drops it.

This lets callers wait for files the client streams in on demand (e.g. the
Steam .snapshot archive, which the game persists to disk) without blocking:
the existing throttled MaybeRefresh picks the file up on a later poll. Undelivered
requests are dropped if the worker is stopped.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

Author: 3vcloud

GWToolboxdll/Utils/GwDat/GwDatArchive.cpp

@@ -1,7 +1,9 @@
 #include <Windows.h>
 
+#include <chrono>
 #include <cstring>
 #include <cwchar>
+#include <utility>
 
 #include "GwDatArchive.h"
 #include "xentax.h"
@@ -10,6 +12,8 @@ namespace {
     const uint8_t kDatMagic[4] = {0x33, 0x41, 0x4e, 0x1a}; // "3ANa"
     constexpr size_t kMaxView = 0x800000;                  // cap each mapped window at 8 MB
     constexpr uint32_t kRefreshThrottleMs = 2000;          // min gap between re-parses on a read miss
+    constexpr uint32_t kAsyncReadTimeoutMs = 60000;        // give up on an async read after ~1 minute
+    constexpr uint32_t kAsyncPollMs = 250;                 // worker re-scan cadence for pending async reads
 
     DWORD AllocationGranularity()
     {
@@ -136,13 +140,19 @@ void GwDatArchive::WorkerLoop()
         std::function<void()> task;
         {
             std::unique_lock<std::mutex> lock(m_task_mutex);
-            m_task_cv.wait(lock, [this] { return m_worker_stop || !m_tasks.empty(); });
+            // Wake for queued tasks, or periodically to re-scan pending async reads.
+            m_task_cv.wait_for(lock, std::chrono::milliseconds(kAsyncPollMs),
+                               [this] { return m_worker_stop || !m_tasks.empty(); });
             if (m_worker_stop)
                 return; // drop any queued tasks on shutdown
-            task = std::move(m_tasks.front());
-            m_tasks.pop();
+            if (!m_tasks.empty()) {
+                task = std::move(m_tasks.front());
+                m_tasks.pop();
+            }
         }
-        task();
+        if (task)
+            task();
+        ProcessPendingReads();
     }
 }
 
@@ -167,8 +177,13 @@ void GwDatArchive::StopWorker()
     m_task_cv.notify_all();
     if (m_worker.joinable())
         m_worker.join();
-    std::scoped_lock<std::mutex> lock(m_task_mutex);
-    m_worker_running = false;
+    {
+        std::scoped_lock<std::mutex> lock(m_task_mutex);
+        m_worker_running = false;
+    }
+    // Drop any undelivered async reads; callers tolerate non-delivery across a worker stop.
+    std::scoped_lock<std::mutex> lock(m_pending_mutex);
+    m_pending_reads.clear();
 }
 
 void GwDatArchive::EnqueueTask(std::function<void()> task)
@@ -396,3 +411,47 @@ bool GwDatArchive::ReadFile(uint32_t file_id, std::vector<uint8_t>& out, uint32_
     }
     return !out.empty();
 }
+
+void GwDatArchive::ReadFileAsync(uint32_t file_id, ReadCallback callback, uint32_t stream_id)
+{
+    if (!callback)
+        return;
+    if (!file_id) {
+        std::vector<uint8_t> empty;
+        callback(empty); // nothing to look up
+        return;
+    }
+    StartWorker(); // the poll loop must be running to deliver
+    const uint32_t deadline = GetTickCount() + kAsyncReadTimeoutMs;
+    std::scoped_lock<std::mutex> lock(m_pending_mutex);
+    m_pending_reads.push_back({file_id, stream_id, deadline, std::move(callback)});
+}
+
+// Worker-thread pass over the pending async reads: deliver any that now resolve, expire any past
+// their deadline. Callbacks fire outside the lock so they can't stall other ReadFileAsync callers.
+void GwDatArchive::ProcessPendingReads()
+{
+    std::vector<std::pair<ReadCallback, std::vector<uint8_t>>> ready;
+    {
+        std::scoped_lock<std::mutex> lock(m_pending_mutex);
+        if (m_pending_reads.empty())
+            return;
+        const uint32_t now = GetTickCount();
+        for (auto it = m_pending_reads.begin(); it != m_pending_reads.end();) {
+            std::vector<uint8_t> data;
+            if (ReadFile(it->file_id, data, it->stream_id)) {
+                ready.emplace_back(std::move(it->callback), std::move(data));
+                it = m_pending_reads.erase(it);
+            }
+            else if (static_cast<int32_t>(now - it->deadline_ms) >= 0) { // deadline passed (wrap-safe)
+                ready.emplace_back(std::move(it->callback), std::vector<uint8_t>{});
+                it = m_pending_reads.erase(it);
+            }
+            else {
+                ++it;
+            }
+        }
+    }
+    for (auto& [callback, data] : ready)
+        callback(data);
+}

GWToolboxdll/Utils/GwDat/GwDatArchive.h

@@ -37,6 +37,14 @@ class GwDatArchive {
     // Decompressed bytes for a GW file id; stream_id picks a stream (0 = the file's own data). False if absent.
     bool ReadFile(uint32_t file_id, std::vector<uint8_t>& out, uint32_t stream_id = 0);
 
+    // Async variant of ReadFile: the worker retries the read until the file appears in the dat, then
+    // invokes `callback` (on the worker thread) with the decompressed bytes; if it hasn't shown up
+    // within ~1 minute the callback fires once with an empty vector. Lets callers wait for files the
+    // client streams in on demand (e.g. Steam .snapshot installs) without blocking. Starts the worker
+    // if needed. Undelivered requests are dropped (callback not called) if the worker is stopped.
+    using ReadCallback = std::function<void(std::vector<uint8_t>&)>;
+    void ReadFileAsync(uint32_t file_id, ReadCallback callback, uint32_t stream_id = 0);
+
     // Dedicated worker so reads/decodes run off the caller's thread; EnqueueTask runs inline if not started.
     void StartWorker();
     void StopWorker();
@@ -46,6 +54,7 @@ class GwDatArchive {
     GwDatArchive() = default;
 
     void WorkerLoop();
+    void ProcessPendingReads(); // deliver/expire queued ReadFileAsync requests (worker thread)
 
 #pragma pack(push, 1)
     struct MainHeader {
@@ -95,6 +104,16 @@ class GwDatArchive {
     std::queue<std::function<void()>> m_tasks;
     bool m_worker_running = false;
     bool m_worker_stop = false;
+
+    // Async ReadFile requests, retried by the worker until the file appears or the deadline passes.
+    struct PendingRead {
+        uint32_t file_id;
+        uint32_t stream_id;
+        uint32_t deadline_ms; // GetTickCount() value at which this request gives up
+        ReadCallback callback;
+    };
+    std::mutex m_pending_mutex; // guards m_pending_reads
+    std::vector<PendingRead> m_pending_reads;
     std::shared_mutex m_index_mutex;               // guards m_mapping/m_slots/m_fileid_to_slot against MaybeRefresh
     uint32_t m_last_refresh_ms = 0;                // GetTickCount of the last re-parse, to throttle refreshes
     void* m_mapping = nullptr;      // file-mapping HANDLE for the dat, replaced by MaybeRefresh when the dat grows