Supercache

.github/workflows/code-review.yml

@@ -7,6 +7,8 @@ on:
 jobs:
   qoder-review:
     runs-on: ubuntu-latest
+    env:
+      QODER_PERSONAL_ACCESS_TOKEN: ${{ secrets.QODER_PERSONAL_ACCESS_TOKEN }}
     permissions:
       contents: read
       pull-requests: write
@@ -19,9 +21,14 @@ jobs:
           fetch-depth: 0
 
       - name: Run Qoder Code Review
+        if: env.QODER_PERSONAL_ACCESS_TOKEN != ''
         uses: QoderAI/qoder-action@v0
         with:
-          qoder_personal_access_token: ${{ secrets.QODER_PERSONAL_ACCESS_TOKEN }}
+          qoder_personal_access_token: ${{ env.QODER_PERSONAL_ACCESS_TOKEN }}
           prompt: |
             /review-pr
             REPO:${{ github.repository }} PR_NUMBER:${{ github.event.pull_request.number }}
+
+      - name: Skip Qoder Code Review
+        if: env.QODER_PERSONAL_ACCESS_TOKEN == ''
+        run: echo "QODER_PERSONAL_ACCESS_TOKEN is unavailable; skipping Qoder review."

docs/source/deployment/mooncake-store-deployment-guide.md

@@ -598,10 +598,15 @@ Local hot cache provides a DRAM read cache on top of SSD-resident objects for fa
 | `MC_MMAP_ARENA_POOL_SIZE` | unset | Pre-allocated arena pool size (e.g., `8gb`). Explicitly set to enable the arena |
 | `MC_DISABLE_MMAP_ARENA` | unset | Set `1` to disable arena, fall back to per-call `mmap()` |
 
-### yalantinglibs Log Level
+### yalantinglibs Logging
 
 ```bash
 export MC_YLT_LOG_LEVEL=info
+export MC_YLT_LOG_PATH=logs/rpc.log
+export MC_YLT_LOG_MAX_FILE_SIZE=1048576000
+export MC_YLT_LOG_MAX_FILES=3
 ```
 
 Available: `trace`, `debug`, `info`, `warn` (or `warning`), `error`, `critical`.
+`MC_YLT_LOG_MAX_FILE_SIZE` is specified in bytes. Invalid or unset values fall
+back to the defaults shown above.

docs/yh/log-reference.md

@@ -681,6 +681,34 @@ export MC_HIFREQ_LOG_SAMPLE_RATE=1.0
 export MC_HIFREQ_LOG_SAMPLE_RATE=0
 ```
 
+#### Correlated latency diagnostics
+
+High-frequency diagnostics now use deterministic `trace_id` sampling. A
+sampled request emits correlated `*_breakdown`, `transfer_diag`,
+`storage_read_breakdown`, `BatchGetReplicaListItem`, and
+`urma_queue_depth` records across worker threads and processes. SSD offload
+RPCs propagate the trace id to the storage owner. Calls without a trace id
+retain random local sampling.
+
+- `transfer_diag.submit_us`: validation, segment lookup, request construction,
+  and submission until a future is returned.
+- `transfer_diag.wait_us`: time blocked in `future.get()`.
+- `local_endpoints_us`: time spent locking and copying the locally mounted
+  Transfer Engine endpoint set.
+- `select_replica_us`: time spent scanning the returned replicas and choosing
+  the preferred complete replica.
+- Transfer-layer records intentionally do not carry object keys; correlate
+  them with Store records through the existing trace id.
+- `storage_read_breakdown.alloc_us`: owner ClientBuffer batch allocation.
+- `storage_read_breakdown.plan_us`: metadata lookup and read-plan construction.
+- `storage_read_breakdown.file_open_us`: path resolution and file open time.
+- `storage_read_breakdown.disk_read_us`: cumulative time inside actual
+  `read_aligned`/`vector_read` calls.
+- `storage_read_breakdown.total_us`: wall time of the owner `BatchGet`.
+
+Batch endpoint records are emitted as `batch_transfer_item`, one record per
+key. Transfer failure ERROR records are never sampled and include the key.
+
 ---
 
 ## 9. 异步日志与 TraceId

mooncake-common/include/default_config.h

@@ -7,8 +7,16 @@
 #endif
 #include <yaml-cpp/yaml.h>
 
+#include <algorithm>
+#include <cctype>
+#include <charconv>
 #include <cstdint>
+#include <cstdlib>
+#include <cstring>
+#include <limits>
+#include <mutex>
 #include <string>
+#include <system_error>
 #include <unordered_map>
 #include <ylt/easylog.hpp>
 
@@ -148,36 +156,63 @@ class DefaultConfig {
     std::unordered_map<std::string, Node> data_;
 };
 
-// usage: export MC_YLT_LOG_LEVEL=info or export MC_YLT_LOG_LEVEL=debug etc.
+// Configure yalantinglibs logging with MC_YLT_LOG_LEVEL, MC_YLT_LOG_PATH,
+// MC_YLT_LOG_MAX_FILE_SIZE (bytes), and MC_YLT_LOG_MAX_FILES.
 inline void init_ylt_log_level() {
-    const char* env_level = std::getenv("MC_YLT_LOG_LEVEL");
-    if (!env_level || !*env_level) {
-        // default is WARN
-        easylog::set_min_severity(easylog::Severity::WARN);
-        return;
-    }
-    std::string level_str(env_level);
-    std::transform(level_str.begin(), level_str.end(), level_str.begin(),
-                   [](unsigned char c) { return std::tolower(c); });
-    easylog::Severity severity;
-    if (level_str == "trace") {
-        severity = easylog::Severity::TRACE;
-    } else if (level_str == "debug") {
-        severity = easylog::Severity::DEBUG;
-    } else if (level_str == "info") {
-        severity = easylog::Severity::INFO;
-    } else if (level_str == "warn" || level_str == "warning") {
-        severity = easylog::Severity::WARN;
-    } else if (level_str == "error") {
-        severity = easylog::Severity::ERROR;
-    } else if (level_str == "critical") {
-        severity = easylog::Severity::CRITICAL;
-    } else {
-        // rollback to WARN
-        severity = easylog::Severity::WARN;
-    }
+    static std::once_flag once;
+    std::call_once(once, [] {
+        easylog::Severity severity = easylog::Severity::WARN;
+        const char* env_level = std::getenv("MC_YLT_LOG_LEVEL");
+        if (env_level && *env_level) {
+            std::string level_str(env_level);
+            std::transform(level_str.begin(), level_str.end(),
+                           level_str.begin(),
+                           [](unsigned char c) { return std::tolower(c); });
+            if (level_str == "trace") {
+                severity = easylog::Severity::TRACE;
+            } else if (level_str == "debug") {
+                severity = easylog::Severity::DEBUG;
+            } else if (level_str == "info") {
+                severity = easylog::Severity::INFO;
+            } else if (level_str == "warn" || level_str == "warning") {
+                severity = easylog::Severity::WARN;
+            } else if (level_str == "error") {
+                severity = easylog::Severity::ERROR;
+            } else if (level_str == "critical") {
+                severity = easylog::Severity::CRITICAL;
+            }
+        }
+
+        std::string log_path = "logs/rpc.log";
+        const char* env_log_path = std::getenv("MC_YLT_LOG_PATH");
+        if (env_log_path && *env_log_path) {
+            log_path = env_log_path;
+        }
 
-    easylog::set_min_severity(severity);
+        auto get_size_env = [](const char* name, size_t default_value) {
+            const char* value = std::getenv(name);
+            if (!value || !*value) return default_value;
+
+            uint64_t parsed = 0;
+            const char* end = value + std::strlen(value);
+            const auto result = std::from_chars(value, end, parsed);
+            if (result.ec != std::errc() || result.ptr != end ||
+                parsed > std::numeric_limits<size_t>::max()) {
+                return default_value;
+            }
+            return static_cast<size_t>(parsed);
+        };
+
+        constexpr size_t kDefaultMaxFileSize = 1000ULL * 1024 * 1024;
+        constexpr size_t kDefaultMaxFiles = 3;
+        const size_t max_file_size = get_size_env(
+            "MC_YLT_LOG_MAX_FILE_SIZE", kDefaultMaxFileSize);
+        const size_t max_files =
+            get_size_env("MC_YLT_LOG_MAX_FILES", kDefaultMaxFiles);
+
+        easylog::init_log(severity, log_path, true, false, max_file_size,
+                          max_files, false);
+    });
 }
 
 }  // namespace mooncake

mooncake-common/include/environ.h

@@ -51,6 +51,13 @@ class Environ {
     bool GetPathRoundrobin() const { return path_roundrobin_; }
     bool GetWithNvidiaPeermem() const { return with_nvidia_peermem_; }
     int GetEfaCqThreads() const { return efa_cq_threads_; }
+    size_t GetOffloadRpcThreadNum(size_t default_value = 8) const;
+    uint32_t GetYltRpcPoolMaxConnection(uint32_t default_value = 100) const;
+    size_t GetYltRpcPoolIdleTimeoutMs(size_t default_value) const;
+    size_t GetYltRpcPoolShortIdleTimeoutMs(size_t default_value) const;
+    bool GetYltRpcPoolWarmupEnabled(bool default_value = true) const;
+    size_t GetYltRpcPoolWarmupConnections(size_t default_value) const;
+    bool GetStoreWarmupEnabled(bool default_value = false) const;
 
     // Helper method to get int from env
     static int GetInt(const char* name, int default_value);

mooncake-common/include/mooncake_logging.h

@@ -16,10 +16,13 @@ void ApplyMooncakeLogEnableToGlog();
 
 // High-frequency log sampling.  Parses MC_HIFREQ_LOG_SAMPLE_RATE once (default
 // 0.1, clamped to [0,1]) and caches it process-wide.  ShouldSampleHiFreqLog()
-// rolls a thread-local RNG: each call returns true with probability == rate.
-// Used to gate per-request high-frequency logs (e.g. real_client breakdown).
+// without an id rolls a thread-local RNG. The trace-id overload hashes the id
+// so all layers handling the same request make the same sampling decision.
 double HiFreqLogSampleRate();
 bool ShouldSampleHiFreqLog();
+// Deterministic variant used to correlate high-frequency logs across
+// processes and worker threads. A zero trace id falls back to random sampling.
+bool ShouldSampleHiFreqLog(uint64_t trace_id);
 
 class ScopedTraceId {
    public:

mooncake-common/src/environ.cpp

@@ -31,6 +31,44 @@ int Environ::GetInt(const char* name, int default_value) {
     return default_value;
 }
 
+size_t Environ::GetOffloadRpcThreadNum(size_t default_value) const {
+    const size_t value = GetSizeT("MC_OFFLOAD_RPC_THREAD_NUM", default_value);
+    return value == 0 ? default_value : value;
+}
+
+uint32_t Environ::GetYltRpcPoolMaxConnection(uint32_t default_value) const {
+    const size_t value =
+        GetSizeT("MC_YLT_RPC_POOL_MAX_CONNECTION", default_value);
+    if (value == 0 || value > UINT32_MAX) {
+        std::cerr << "[Mooncake] Warning: invalid value for env "
+                  << "MC_YLT_RPC_POOL_MAX_CONNECTION, using default "
+                  << default_value << std::endl;
+        return default_value;
+    }
+    return static_cast<uint32_t>(value);
+}
+
+size_t Environ::GetYltRpcPoolIdleTimeoutMs(size_t default_value) const {
+    return GetSizeT("MC_YLT_RPC_POOL_IDLE_TIMEOUT_MS", default_value);
+}
+
+size_t Environ::GetYltRpcPoolShortIdleTimeoutMs(
+    size_t default_value) const {
+    return GetSizeT("MC_YLT_RPC_POOL_SHORT_IDLE_TIMEOUT_MS", default_value);
+}
+
+bool Environ::GetYltRpcPoolWarmupEnabled(bool default_value) const {
+    return GetBool("MC_YLT_RPC_POOL_WARMUP", default_value);
+}
+
+size_t Environ::GetYltRpcPoolWarmupConnections(size_t default_value) const {
+    return GetSizeT("MC_YLT_RPC_POOL_WARMUP_CONNECTIONS", default_value);
+}
+
+bool Environ::GetStoreWarmupEnabled(bool default_value) const {
+    return GetBool("MC_STORE_WARMUP", default_value);
+}
+
 size_t Environ::GetSizeT(const char* name, size_t default_value) {
     const char* val = std::getenv(name);
     if (val) {

mooncake-common/src/mooncake_logging.cpp

@@ -323,6 +323,22 @@ bool ShouldSampleHiFreqLog() {
     return dist(rng) < rate;
 }
 
+bool ShouldSampleHiFreqLog(uint64_t trace_id) {
+    if (trace_id == 0) return ShouldSampleHiFreqLog();
+    const double rate = HiFreqLogSampleRate();
+    if (rate >= 1.0) return true;
+    if (rate <= 0.0) return false;
+
+    // SplitMix64 finalizer: stable, cheap, and sufficiently uniform for
+    // deterministic sampling. Use the top 53 bits to match double precision.
+    uint64_t value = trace_id + 0x9e3779b97f4a7c15ULL;
+    value = (value ^ (value >> 30)) * 0xbf58476d1ce4e5b9ULL;
+    value = (value ^ (value >> 27)) * 0x94d049bb133111ebULL;
+    value ^= value >> 31;
+    constexpr double kScale = 1.0 / static_cast<double>(1ULL << 53);
+    return static_cast<double>(value >> 11) * kScale < rate;
+}
+
 bool ShouldLog(google::LogSeverity severity) {
     // MC_LOG_ENABLE was removed: MC_LOG now behaves like plain glog LOG and is
     // gated only by glog's own severity threshold (still async + trace_id).

mooncake-integration/store/mooncake_perf_points.def

@@ -105,6 +105,9 @@ PERF_KEY_DEF(PUT_BATCH_COLLECT_RESULTS,      "client_service.cpp::BatchPut",
 
 // === batch_get_into_offload_object_internal 子步骤 (SSD read) ===
 PERF_KEY_DEF(GET_SSD_OFFLOAD_RPC,            "real_client.cpp::batch_get_into_offload_object_internal",  "OffloadRpc")
+PERF_KEY_DEF(GET_SSD_OFFLOAD_RPC_POOL,       "real_client.cpp::ClientRequester::invoke_rpc",             "RpcPoolLookup")
+PERF_KEY_DEF(GET_SSD_OFFLOAD_RPC_CALL,       "real_client.cpp::ClientRequester::invoke_rpc",             "RpcCall")
+PERF_KEY_DEF(GET_SSD_OFFLOAD_RPC_RESULT_GET, "real_client.cpp::ClientRequester::invoke_rpc",             "RpcResultGet")
 PERF_KEY_DEF(GET_SSD_TRANSFER_DATA,          "real_client.cpp::batch_get_into_offload_object_internal",  "TransferData")
 PERF_KEY_DEF(GET_SSD_RELEASE_BUFFER,         "real_client.cpp::batch_get_into_offload_object_internal",  "ReleaseBuffer")
 
@@ -116,6 +119,9 @@ PERF_KEY_DEF(GET_SSD_OWNER_LOAD_PLAN,        "storage_backend.cpp::BatchLoad",
 PERF_KEY_DEF(GET_SSD_OWNER_LOAD_URING,       "storage_backend.cpp::BatchLoad",                           "OwnerLoadUring")
 PERF_KEY_DEF(GET_SSD_OWNER_LOAD_POSIX,       "storage_backend.cpp::BatchLoad",                           "OwnerLoadPosix")
 PERF_KEY_DEF(GET_SSD_OWNER_PUSH_WRITE,       "real_client.cpp::batch_get_offload_object_push",           "OwnerPushWrite")
+PERF_KEY_DEF(GET_SSD_OWNER_RPC_QUEUE,        "real_client.cpp::batch_get_offload_object",                "OwnerRpcQueue")
+PERF_KEY_DEF(GET_SSD_OWNER_RPC_BATCH_GET,    "real_client.cpp::batch_get_offload_object",                "OwnerRpcBatchGet")
+PERF_KEY_DEF(GET_SSD_OWNER_RPC_RESUME,       "real_client.cpp::batch_get_offload_object",                "OwnerRpcResume")
 PERF_KEY_DEF(GET_SSD_OWNER_RELEASE,          "file_storage.cpp::ReleaseBuffer",                          "OwnerReleaseBuffer")
 
 // === UB / URMA endpoint first connection path ===

mooncake-store/benchmarks/stress_cluster_bench.cpp

@@ -11,6 +11,7 @@
 #include <latch>
 #include <memory>
 #include <numeric>
+#include <random>
 #include <sstream>
 #include <string>
 #include <thread>
@@ -199,6 +200,10 @@ DEFINE_uint64(num_threads, 1, "Number of concurrent reader threads");
 DEFINE_uint64(warmup_keys, 5, "Number of warmup keys (not counted in stats)");
 DEFINE_uint64(wait_seconds, 5,
               "Seconds to wait before reading (for remote scenarios)");
+DEFINE_uint64(client_init_wait_seconds, 0,
+              "Seconds to wait after RealClient setup succeeds and before "
+              "benchmark run starts. This can give asynchronous RPC/transfer "
+              "warmup and background initialization time to settle.");
 DEFINE_bool(verify, true, "Verify data integrity after read");
 DEFINE_uint64(replica_num, 1, "Number of replicas for each object");
 DEFINE_bool(hard_pin, false,
@@ -218,6 +223,9 @@ DEFINE_uint64(duration, 0,
               "scenario (0 = read num_keys once)");
 DEFINE_uint64(statis_interval, 5,
               "Statistics print interval in seconds for segment_read scenario");
+DEFINE_uint64(shuffle_seed, 0,
+              "Seed for deterministic shuffle of segment_read keys "
+              "(0 = disabled)");
 
 using Clock = std::chrono::steady_clock;
 using Nanos = std::chrono::nanoseconds;
@@ -819,6 +827,12 @@ class StressBenchmark {
                 all_keys.push_back(MakeSegmentKey(read_segments[s], i));
             }
         }
+        if (FLAGS_shuffle_seed != 0) {
+            std::mt19937_64 rng(FLAGS_shuffle_seed);
+            std::shuffle(all_keys.begin(), all_keys.end(), rng);
+            LOG(INFO) << "Shuffled segment_read keys with seed="
+                      << FLAGS_shuffle_seed;
+        }
         LOG(INFO) << "Total keys to read: " << all_keys.size();
 
         size_t warmup_end =
@@ -1572,6 +1586,8 @@ int main(int argc, char* argv[]) {
     LOG(INFO) << "  Read seg nums:  " << FLAGS_read_segment_nums;
     LOG(INFO) << "  Duration:       " << FLAGS_duration << "s";
     LOG(INFO) << "  Stats interval: " << FLAGS_statis_interval << "s";
+    LOG(INFO) << "  Client init wait: " << FLAGS_client_init_wait_seconds
+              << "s";
 
     size_t total_data = FLAGS_num_keys * FLAGS_value_size;
     if (total_data > FLAGS_global_segment_size * 9.5 / 10) {
@@ -1589,6 +1605,14 @@ int main(int argc, char* argv[]) {
         return ret;
     }
 
+    if (FLAGS_client_init_wait_seconds > 0) {
+        LOG(INFO) << "Waiting " << FLAGS_client_init_wait_seconds
+                  << " seconds after client setup before benchmark run...";
+        std::this_thread::sleep_for(
+            std::chrono::seconds(FLAGS_client_init_wait_seconds));
+        LOG(INFO) << "Client init wait complete";
+    }
+
     ret = bench.Run();
     return ret;
 }

mooncake-store/include/client_service.h

@@ -139,6 +139,11 @@ class Client {
         ErrorCode>
     QueryByRegex(const std::string& str);
 
+    /**
+     * @brief Returns unique offload RPC endpoints currently known by master.
+     */
+    tl::expected<std::vector<std::string>, ErrorCode> GetOffloadEndpoints();
+
     /**
      * @brief Batch query object metadata without transferring data
      * @param object_keys Keys to query