diff --git a/.github/workflows/code-review.yml b/.github/workflows/code-review.yml index 5d94a0a712..5feb21d926 100644 --- a/.github/workflows/code-review.yml +++ b/.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." diff --git a/docs/source/deployment/mooncake-store-deployment-guide.md b/docs/source/deployment/mooncake-store-deployment-guide.md index aa4799448b..77ea125f9c 100644 --- a/docs/source/deployment/mooncake-store-deployment-guide.md +++ b/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. diff --git a/docs/yh/log-reference.md b/docs/yh/log-reference.md index fbc6d4c8ce..8f9f811aad 100644 --- a/docs/yh/log-reference.md +++ b/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 diff --git a/extern/ubdiag b/extern/ubdiag new file mode 160000 index 0000000000..7deea5f897 --- /dev/null +++ b/extern/ubdiag @@ -0,0 +1 @@ +Subproject commit 7deea5f89786784a2e2aa074d315a0a8e2561963 diff --git a/extern/yalantinglibs b/extern/yalantinglibs index 6a0e067d9a..b12fbfd86d 160000 --- a/extern/yalantinglibs +++ b/extern/yalantinglibs @@ -1 +1 @@ -Subproject commit 6a0e067d9a43492cf8e4e280b531924fbd724dbd +Subproject commit b12fbfd86db241b9b79a696c373ffd3306e65369 diff --git a/mooncake-common/include/default_config.h b/mooncake-common/include/default_config.h index f072baf212..1c2149e239 100644 --- a/mooncake-common/include/default_config.h +++ b/mooncake-common/include/default_config.h @@ -7,8 +7,16 @@ #endif #include +#include +#include +#include #include +#include +#include +#include +#include #include +#include #include #include @@ -148,36 +156,63 @@ class DefaultConfig { std::unordered_map 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::max()) { + return default_value; + } + return static_cast(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 diff --git a/mooncake-common/include/environ.h b/mooncake-common/include/environ.h index 78f8f45028..bd3b7ed946 100644 --- a/mooncake-common/include/environ.h +++ b/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); diff --git a/mooncake-common/include/mooncake_logging.h b/mooncake-common/include/mooncake_logging.h index ce3b24b1ee..df7ef39564 100644 --- a/mooncake-common/include/mooncake_logging.h +++ b/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: diff --git a/mooncake-common/src/environ.cpp b/mooncake-common/src/environ.cpp index 46aa1e9a9c..291c7c848e 100644 --- a/mooncake-common/src/environ.cpp +++ b/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(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) { diff --git a/mooncake-common/src/mooncake_logging.cpp b/mooncake-common/src/mooncake_logging.cpp index 4175c67567..e79b6cb27c 100644 --- a/mooncake-common/src/mooncake_logging.cpp +++ b/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(1ULL << 53); + return static_cast(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). diff --git a/mooncake-integration/store/mooncake_perf_points.def b/mooncake-integration/store/mooncake_perf_points.def index 251a02bfcb..e1b45a7e78 100644 --- a/mooncake-integration/store/mooncake_perf_points.def +++ b/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 === diff --git a/mooncake-store/benchmarks/stress_cluster_bench.cpp b/mooncake-store/benchmarks/stress_cluster_bench.cpp index f5b4e76142..6c70652171 100644 --- a/mooncake-store/benchmarks/stress_cluster_bench.cpp +++ b/mooncake-store/benchmarks/stress_cluster_bench.cpp @@ -11,6 +11,7 @@ #include #include #include +#include #include #include #include @@ -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; } diff --git a/mooncake-store/include/client_service.h b/mooncake-store/include/client_service.h index 383dedfddb..a4c910fb45 100644 --- a/mooncake-store/include/client_service.h +++ b/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, ErrorCode> GetOffloadEndpoints(); + /** * @brief Batch query object metadata without transferring data * @param object_keys Keys to query diff --git a/mooncake-store/include/master_client.h b/mooncake-store/include/master_client.h index c3dc9a7e6f..20c4a1f5bd 100644 --- a/mooncake-store/include/master_client.h +++ b/mooncake-store/include/master_client.h @@ -12,6 +12,7 @@ #include #include +#include "environ.h" #include "client_metric.h" #include "replica.h" #include "segment.h" @@ -59,6 +60,14 @@ class MasterClient { metrics_(metrics) { coro_io::client_pool::pool_config pool_conf{}; + pool_conf.max_connection = Environ::Get().GetYltRpcPoolMaxConnection( + pool_conf.max_connection); + pool_conf.idle_timeout = std::chrono::milliseconds( + Environ::Get().GetYltRpcPoolIdleTimeoutMs( + pool_conf.idle_timeout.count())); + pool_conf.short_connect_idle_timeout = std::chrono::milliseconds( + Environ::Get().GetYltRpcPoolShortIdleTimeoutMs( + pool_conf.short_connect_idle_timeout.count())); // Disable alive_detect to prevent stale reconnection logs after HA // failover. Old client_pool objects remain in client_pools_ map and @@ -442,6 +451,9 @@ class MasterClient { const UUID& client_id, const std::vector& tasks, const std::vector& metadatas); + [[nodiscard]] tl::expected, ErrorCode> + GetOffloadEndpoints(); + /** * @brief Heartbeat-driven pull of pending L2->L1 promotion work for a * client. Returns tenant-scoped tasks the caller should read from local @@ -663,6 +675,8 @@ class MasterClient { [[nodiscard]] std::vector> invoke_batch_rpc(size_t input_size, Args&&... args); + void WarmupRpcPool(); + /** * @brief Accessor for the coro_rpc_client pool. Since coro_rpc_client pool * cannot reconnect to a different address, a new coro_rpc_client pool is diff --git a/mooncake-store/include/master_service.h b/mooncake-store/include/master_service.h index a6ff2482e5..499fb91f89 100644 --- a/mooncake-store/include/master_service.h +++ b/mooncake-store/include/master_service.h @@ -592,6 +592,13 @@ class MasterService { const std::vector& metadatas) -> tl::expected; + /** + * @brief Returns unique RPC endpoints that own completed LOCAL_DISK + * replicas. Used by clients to warm up offload RPC pools before traffic. + */ + auto GetOffloadEndpoints() + -> tl::expected, ErrorCode>; + /** * @brief Heartbeat-driven pull of pending promotion work for a client. * Returns tenant-scoped promotion tasks for the holder client and clears diff --git a/mooncake-store/include/pyclient.h b/mooncake-store/include/pyclient.h index 8e39d45de2..b9e2d190ed 100644 --- a/mooncake-store/include/pyclient.h +++ b/mooncake-store/include/pyclient.h @@ -5,9 +5,11 @@ #include #include #include +#include #include #include #include +#include #include #include "client_service.h" @@ -140,6 +142,14 @@ struct ShmFdResponse { class ClientRequester { public: + struct RpcTiming { + uint64_t pool_lookup_us{0}; + uint64_t rpc_call_us{0}; + uint64_t result_get_us{0}; + uint64_t result_parse_us{0}; + uint64_t total_us{0}; + }; + ClientRequester(); /** @@ -151,7 +161,8 @@ class ClientRequester { tl::expected batch_get_offload_object(const std::string &client_addr, const std::vector &keys, - const std::vector &sizes); + const std::vector &sizes, + RpcTiming *timing = nullptr); /** * @brief Push-mode offload: invoke the owner's push handler, which WRITEs @@ -163,17 +174,20 @@ class ClientRequester { tl::expected batch_get_offload_object_push( const std::string &client_addr, - const BatchGetOffloadObjectPushRequest &req); + const BatchGetOffloadObjectPushRequest &req, + RpcTiming *timing = nullptr); /** - * @brief Notifies remote FileStorage to release buffer after transfer - * completion. This is a fire-and-forget call - errors are logged but not - * propagated. + * @brief Synchronously notifies remote FileStorage to release its buffer + * after transfer completion. Errors are logged but not propagated because + * the owner's lease-based GC provides a fallback. * @param client_addr Network address of the remote FileStorage service. * @param batch_id The batch_id returned from batch_get_offload_object. */ void release_offload_buffer(const std::string &client_addr, - uint64_t batch_id); + uint64_t batch_id, RpcTiming *timing = nullptr); + + void WarmupRpcPool(const std::string &client_addr); private: /** @@ -203,6 +217,8 @@ class ClientRequester { }; mutable std::shared_mutex client_pool_mutex_; + mutable std::shared_mutex warmed_rpc_pool_mutex_; + std::unordered_set warmed_rpc_pools_; std::shared_ptr> client_pools_; @@ -216,7 +232,7 @@ class ClientRequester { */ template [[nodiscard]] tl::expected invoke_rpc( - const std::string &client_addr, Args &&...args); + const std::string &client_addr, RpcTiming *timing, Args &&...args); }; // Python-specific wrapper class for client interface diff --git a/mooncake-store/include/real_client.h b/mooncake-store/include/real_client.h index 373e672c6f..ec04e9192a 100644 --- a/mooncake-store/include/real_client.h +++ b/mooncake-store/include/real_client.h @@ -506,7 +506,8 @@ class RealClient : public PyClient { const std::string &ipc_socket_path = "", int local_rpc_port = 50052, bool enable_ssd_offload = false, bool start_offload_rpc_server = false, const std::string &ssd_offload_path = "", - const std::string &tenant_id = "default"); + const std::string &tenant_id = "default", + size_t offload_rpc_thread_num = 8); // Overload that accepts a configuration dictionary tl::expected setup_internal(const ConfigDict &config); @@ -532,7 +533,12 @@ class RealClient : public PyClient { uint64_t total_size; int64_t query_us; int64_t select_us; + int64_t local_endpoints_us; + int64_t select_replica_us; std::string replica_type; + size_t replica_count{0}; + std::string local_endpoint{"-"}; + std::string remote_endpoint{"-"}; }; tl::expected @@ -680,7 +686,8 @@ class RealClient : public PyClient { async_simple::coro::Lazy< tl::expected> batch_get_offload_object(const std::vector &keys, - const std::vector &sizes); + const std::vector &sizes, + uint64_t trace_id = 0); /** * @brief Push-mode offload handler, run on the data owner. Reads the @@ -702,7 +709,7 @@ class RealClient : public PyClient { * @param batch_id The unique identifier of the batch to release * @return true if batch was found and released, false otherwise */ - bool release_offload_buffer(uint64_t batch_id); + bool release_offload_buffer(uint64_t batch_id, uint64_t trace_id = 0); /** * @brief Retrieves multiple stored objects from a remote service. diff --git a/mooncake-store/include/rpc_service.h b/mooncake-store/include/rpc_service.h index b7c118411a..079750cdd1 100644 --- a/mooncake-store/include/rpc_service.h +++ b/mooncake-store/include/rpc_service.h @@ -202,6 +202,8 @@ class WrappedMasterService { const UUID& client_id, const std::vector& tasks, const std::vector& metadatas); + tl::expected, ErrorCode> GetOffloadEndpoints(); + // Promotion-on-hit RPCs. tl::expected, ErrorCode> PromotionObjectHeartbeat(const UUID& client_id); diff --git a/mooncake-store/include/rpc_types.h b/mooncake-store/include/rpc_types.h index e1f3091240..3f60320fd4 100644 --- a/mooncake-store/include/rpc_types.h +++ b/mooncake-store/include/rpc_types.h @@ -296,11 +296,12 @@ struct BatchGetOffloadObjectPushRequest { std::vector sizes; // total bytes per key (for FileStorage) std::string requester_te_addr; // requester's transfer engine endpoint std::vector> dst_slices; // per-key dst slices + uint64_t trace_id{0}; BatchGetOffloadObjectPushRequest() = default; }; YLT_REFL(BatchGetOffloadObjectPushRequest, keys, sizes, requester_te_addr, - dst_slices); + dst_slices, trace_id); struct BatchGetOffloadObjectPushResponse { ErrorCode error_code; // overall result; data is already in requester memory diff --git a/mooncake-store/include/storage_backend.h b/mooncake-store/include/storage_backend.h index dfeabfb54a..ebc0203665 100644 --- a/mooncake-store/include/storage_backend.h +++ b/mooncake-store/include/storage_backend.h @@ -251,6 +251,29 @@ struct FileStorageConfig { static FileStorageConfig FromEnvironment(); }; +struct StorageReadStats { + uint64_t plan_us{0}; + uint64_t file_open_us{0}; + uint64_t disk_read_us{0}; + uint64_t slowest_disk_read_us{0}; + std::string slowest_key{"-"}; + std::string io_mode{"unknown"}; + std::string status{"ok"}; + std::string error_key{"-"}; + ErrorCode error_code{ErrorCode::OK}; +}; + +StorageReadStats* CurrentStorageReadStats(); + +class ScopedStorageReadStats { + public: + explicit ScopedStorageReadStats(StorageReadStats* stats); + ~ScopedStorageReadStats(); + + private: + StorageReadStats* previous_; +}; + class StorageBackendInterface { public: StorageBackendInterface(const FileStorageConfig& file_storage_config); @@ -1214,4 +1237,4 @@ class OffsetAllocatorStorageBackend : public StorageBackendInterface { tl::expected, ErrorCode> CreateStorageBackend(const FileStorageConfig& config); -} // namespace mooncake \ No newline at end of file +} // namespace mooncake diff --git a/mooncake-store/src/client_service.cpp b/mooncake-store/src/client_service.cpp index 34d2d51cf3..5dfdb28e4c 100644 --- a/mooncake-store/src/client_service.cpp +++ b/mooncake-store/src/client_service.cpp @@ -1072,6 +1072,11 @@ Client::QueryByRegex(const std::string& str) { return result; } +tl::expected, ErrorCode> +Client::GetOffloadEndpoints() { + return master_client_.GetOffloadEndpoints(); +} + tl::expected Client::Query( const std::string& object_key) { std::chrono::steady_clock::time_point start_time = @@ -1380,6 +1385,8 @@ std::vector> Client::BatchGet( const std::vector& query_results, std::unordered_map>& slices, bool prefer_alloc_in_same_node) { + const bool batch_item_log = mooncake::logging::ShouldSampleHiFreqLog( + mooncake::logging::CurrentTraceId()); if (!transfer_submitter_) { MC_LOG(ERROR) << "TransferSubmitter not initialized"; std::vector> results; @@ -1535,6 +1542,36 @@ std::vector> Client::BatchGet( } } } + if (batch_item_log) { + std::string selected_type = "disk"; + std::string remote_endpoint = "-"; + if (stored_replica.is_memory_replica()) { + selected_type = "memory"; + remote_endpoint = + stored_replica.get_memory_descriptor() + .buffer_descriptor.transport_endpoint_; + } else if (stored_replica.is_nof_replica()) { + selected_type = "nof"; + remote_endpoint = + stored_replica.get_nof_descriptor() + .buffer_descriptor.transport_endpoint_; + } + const auto slices_it = slices.find(key); + const size_t bytes = + slices_it == slices.end() + ? 0 + : CalculateSliceSize(slices_it->second); + MC_LOG(INFO) + << "batch_transfer_item op=batch_get key=" << key + << " batch_index=" << index + << " selected_type=" << selected_type + << " strategy=" << future.strategy() + << " local_endpoint=" << GetSegmentEndpoint() + << " remote_endpoint=" << remote_endpoint + << " bytes=" << bytes << " status=" + << (result == ErrorCode::OK ? "ok" : "transfer_fail") + << " error_code=" << static_cast(result); + } } // As lease expired is a rare case, we check all the results with the same @@ -1679,6 +1716,9 @@ tl::expected Client::Put(const ObjectKey& key, ErrorCode transfer_err = TransferWrite(replica, slices); pt_tw.End(transfer_err == ErrorCode::OK ? 0 : -1); if (transfer_err != ErrorCode::OK) { + MC_LOG(ERROR) << "transfer_write_failed key=" << key + << " error_code=" + << static_cast(transfer_err); // Revoke put operation UbDiag::PerfPoint pt_revoke(PerfKey::PUT_SINGLE_PUT_REVOKE, UbDiag::PerfLevel::MODULE); @@ -1787,6 +1827,9 @@ tl::expected Client::Upsert(const ObjectKey& key, if (replica.is_memory_replica()) { ErrorCode transfer_err = TransferWrite(replica, slices); if (transfer_err != ErrorCode::OK) { + MC_LOG(ERROR) << "transfer_write_failed key=" << key + << " error_code=" + << static_cast(transfer_err); auto revoke_result = master_client_.UpsertRevoke(key, ReplicaType::MEMORY); if (!revoke_result) { @@ -2108,6 +2151,8 @@ void Client::StartBatchUpsert(std::vector& ops, } void Client::SubmitTransfers(std::vector& ops) { + const bool batch_item_log = mooncake::logging::ShouldSampleHiFreqLog( + mooncake::logging::CurrentTraceId()); if (!transfer_submitter_) { MC_LOG(ERROR) << "TransferSubmitter not initialized"; for (auto& op : ops) { @@ -2181,7 +2226,33 @@ void Client::SubmitTransfers(std::vector& ops) { replica, op.slices, TransferRequest::WRITE); } pt_submit.End(submit_result ? 0 : -1); + if (batch_item_log) { + const auto& handle = + replica.is_memory_replica() + ? replica.get_memory_descriptor().buffer_descriptor + : replica.get_nof_descriptor().buffer_descriptor; + MC_LOG(INFO) + << "batch_transfer_item op=batch_put key=" << op.key + << " selected_type=" + << (replica.is_memory_replica() ? "memory" : "nof") + << " strategy=" + << (submit_result ? submit_result->strategy() + : TransferStrategy::EMPTY) + << " local_endpoint=" << GetSegmentEndpoint() + << " remote_endpoint=" << handle.transport_endpoint_ + << " bytes=" << CalculateSliceSize(op.slices) + << " status=" + << (submit_result ? "submitted" : "submit_fail") + << " error_code=" + << static_cast( + submit_result ? ErrorCode::OK + : ErrorCode::TRANSFER_FAIL); + } if (!submit_result) { + MC_LOG(ERROR) + << "transfer_failed op=batch_put key=" << op.key + << " direction=write stage=submit error_code=" + << static_cast(ErrorCode::TRANSFER_FAIL); std::string failure_context = "Failed to submit transfer for replica " + std::to_string(replica_idx); @@ -2212,6 +2283,10 @@ void Client::WaitForTransfers(std::vector& ops) { auto& pending_transfer = op.pending_transfers[i]; ErrorCode transfer_result = pending_transfer.future.get(); if (transfer_result != ErrorCode::OK) { + MC_LOG(ERROR) + << "transfer_failed op=batch_put key=" << op.key + << " direction=write stage=wait error_code=" + << static_cast(transfer_result); op.transfer_summary.RecordFailure(pending_transfer.replica_type, transfer_result); std::string error_context = @@ -3322,13 +3397,26 @@ tl::expected Client::BatchGetOffloadObject( auto future = transfer_submitter_->submit_batch_get_offload_object( transfer_engine_addr, keys, pointers, batch_slices); if (!future) { - MC_LOG(ERROR) << "Failed to submit transfer operation"; + for (size_t i = 0; i < keys.size(); ++i) { + MC_LOG(ERROR) + << "transfer_failed op=batch_get_offload key=" << keys[i] + << " batch_index=" << i + << " direction=read remote_endpoint=" << transfer_engine_addr + << " stage=submit error_code=" + << static_cast(ErrorCode::TRANSFER_FAIL); + } return tl::make_unexpected(ErrorCode::TRANSFER_FAIL); } MC_VLOG(1) << "Using transfer strategy: " << future->strategy(); auto result = future->get(); if (result != ErrorCode::OK) { - MC_LOG(ERROR) << "Transfer failed, error code is " << result; + for (size_t i = 0; i < keys.size(); ++i) { + MC_LOG(ERROR) + << "transfer_failed op=batch_get_offload key=" << keys[i] + << " batch_index=" << i + << " direction=read remote_endpoint=" << transfer_engine_addr + << " stage=wait error_code=" << static_cast(result); + } return tl::make_unexpected(result); } return {}; @@ -3342,13 +3430,26 @@ tl::expected Client::BatchPushOffloadObject( auto future = transfer_submitter_->submit_batch_push_offload_object( requester_te_addr, keys, src_pointers, dst_slices); if (!future) { - MC_LOG(ERROR) << "Failed to submit push transfer operation"; + for (size_t i = 0; i < keys.size(); ++i) { + MC_LOG(ERROR) + << "transfer_failed op=batch_push_offload key=" << keys[i] + << " batch_index=" << i + << " direction=write remote_endpoint=" << requester_te_addr + << " stage=submit error_code=" + << static_cast(ErrorCode::TRANSFER_FAIL); + } return tl::make_unexpected(ErrorCode::TRANSFER_FAIL); } MC_VLOG(1) << "Using transfer strategy: " << future->strategy(); auto result = future->get(); if (result != ErrorCode::OK) { - MC_LOG(ERROR) << "Push transfer failed, error code is " << result; + for (size_t i = 0; i < keys.size(); ++i) { + MC_LOG(ERROR) + << "transfer_failed op=batch_push_offload key=" << keys[i] + << " batch_index=" << i + << " direction=write remote_endpoint=" << requester_te_addr + << " stage=wait error_code=" << static_cast(result); + } return tl::make_unexpected(result); } return {}; @@ -3733,6 +3834,24 @@ ErrorCode Client::TransferData(const Replica::Descriptor& replica_descriptor, std::vector& slices, TransferRequest::OpCode op_code) { bool is_write = (op_code == TransferRequest::WRITE); + std::string remote_endpoint = "-"; + if (replica_descriptor.is_memory_replica()) { + remote_endpoint = + replica_descriptor.get_memory_descriptor() + .buffer_descriptor.transport_endpoint_; + } else if (replica_descriptor.is_nof_replica()) { + remote_endpoint = + replica_descriptor.get_nof_descriptor() + .buffer_descriptor.transport_endpoint_; + } else if (replica_descriptor.is_local_disk_replica()) { + remote_endpoint = + replica_descriptor.get_local_disk_descriptor().transport_endpoint; + } + const bool diag_log = mooncake::logging::ShouldSampleHiFreqLog( + mooncake::logging::CurrentTraceId()); + const auto submit_start = + diag_log ? std::chrono::steady_clock::now() + : std::chrono::steady_clock::time_point{}; UbDiag::PerfPoint pt_full(is_write ? PerfKey::PUT_SINGLE_TRANSFER_FULL : PerfKey::GET_SINGLE_TRANSFER_FULL, UbDiag::PerfLevel::MODULE); @@ -3762,10 +3881,30 @@ ErrorCode Client::TransferData(const Replica::Descriptor& replica_descriptor, } pt_submit.End(future ? 0 : -1); if (!future) { - MC_LOG(ERROR) << "Failed to submit transfer operation"; + if (diag_log) { + const auto submit_us = + std::chrono::duration_cast( + std::chrono::steady_clock::now() - submit_start) + .count(); + MC_LOG(INFO) << "transfer_diag op=" + << (is_write ? "put" : "get") + << " direction=" << (is_write ? "write" : "read") + << " strategy=unknown request_count=" << slices.size() + << " total_bytes=" << CalculateSliceSize(slices) + << " submit_us=" << submit_us + << (is_write ? " local_endpoint=" : "") + << (is_write ? GetSegmentEndpoint() : "") + << (is_write ? " remote_endpoint=" : "") + << (is_write ? remote_endpoint : "") + << " wait_us=0 status=submit_fail error_code=" + << static_cast(ErrorCode::TRANSFER_FAIL); + } pt_full.End(-1); return ErrorCode::TRANSFER_FAIL; } + const auto submit_end = + diag_log ? std::chrono::steady_clock::now() + : std::chrono::steady_clock::time_point{}; MC_VLOG(1) << "Using transfer strategy: " << future->strategy(); @@ -3774,8 +3913,34 @@ ErrorCode Client::TransferData(const Replica::Descriptor& replica_descriptor, UbDiag::PerfLevel::DEBUG); pt_wait.Start(); auto result = future->get(); + const auto wait_end = + diag_log ? std::chrono::steady_clock::now() + : std::chrono::steady_clock::time_point{}; pt_wait.End(result == ErrorCode::OK ? 0 : -1); pt_full.End(result == ErrorCode::OK ? 0 : -1); + if (diag_log) { + const auto submit_us = + std::chrono::duration_cast( + submit_end - submit_start) + .count(); + const auto wait_us = + std::chrono::duration_cast(wait_end - + submit_end) + .count(); + MC_LOG(INFO) << "transfer_diag op=" << (is_write ? "put" : "get") + << " direction=" << (is_write ? "write" : "read") + << " strategy=" << future->strategy() + << " request_count=" << slices.size() + << " total_bytes=" << CalculateSliceSize(slices) + << (is_write ? " local_endpoint=" : "") + << (is_write ? GetSegmentEndpoint() : "") + << (is_write ? " remote_endpoint=" : "") + << (is_write ? remote_endpoint : "") + << " submit_us=" << submit_us << " wait_us=" << wait_us + << " status=" + << (result == ErrorCode::OK ? "ok" : "transfer_fail") + << " error_code=" << static_cast(result); + } return result; } @@ -3787,16 +3952,42 @@ ErrorCode Client::TransferReadInternal( return ErrorCode::INVALID_PARAMS; } + const bool diag_log = mooncake::logging::ShouldSampleHiFreqLog( + mooncake::logging::CurrentTraceId()); + const auto submit_start = + diag_log ? std::chrono::steady_clock::now() + : std::chrono::steady_clock::time_point{}; auto future = transfer_submitter_->submitRangeRead(replica_descriptor, slices, src_offset); if (!future) { - MC_LOG(ERROR) << "Failed to submit range read operation"; return ErrorCode::TRANSFER_FAIL; } + const auto submit_end = + diag_log ? std::chrono::steady_clock::now() + : std::chrono::steady_clock::time_point{}; MC_VLOG(1) << "Using transfer strategy: " << future->strategy(); - - return future->get(); + const auto result = future->get(); + if (diag_log) { + const auto wait_end = std::chrono::steady_clock::now(); + MC_LOG(INFO) + << "transfer_diag op=get_into_range" + << " direction=read strategy=" << future->strategy() + << " request_count=" << slices.size() + << " total_bytes=" << CalculateSliceSize(slices) + << " submit_us=" + << std::chrono::duration_cast( + submit_end - submit_start) + .count() + << " wait_us=" + << std::chrono::duration_cast(wait_end - + submit_end) + .count() + << " status=" + << (result == ErrorCode::OK ? "ok" : "transfer_fail") + << " error_code=" << static_cast(result); + } + return result; } ErrorCode Client::TransferWrite(const Replica::Descriptor& replica_descriptor, diff --git a/mooncake-store/src/file_storage.cpp b/mooncake-store/src/file_storage.cpp index 754fbfda17..263cda3530 100644 --- a/mooncake-store/src/file_storage.cpp +++ b/mooncake-store/src/file_storage.cpp @@ -1,9 +1,11 @@ #include "file_storage.h" #include +#include #include #include "aligned_client_buffer.hpp" +#include "mooncake_logging.h" #include "storage_backend.h" #include "client_metric.h" #include "utils.h" @@ -331,6 +333,41 @@ tl::expected FileStorage::Init() { tl::expected FileStorage::BatchGet( const std::vector& keys, const std::vector& sizes) { auto start_time = std::chrono::steady_clock::now(); + const uint64_t trace_id = mooncake::logging::CurrentTraceId(); + const bool breakdown_log = + mooncake::logging::ShouldSampleHiFreqLog(trace_id); + const uint64_t total_bytes = + std::accumulate(sizes.begin(), sizes.end(), uint64_t{0}); + uint64_t alloc_us = 0; + StorageReadStats read_stats; + auto log_breakdown = [&](const char* fallback_status, + ErrorCode fallback_error) { + if (!breakdown_log) return; + const auto total_us = + std::chrono::duration_cast( + std::chrono::steady_clock::now() - start_time) + .count(); + const char* status = + read_stats.status == "ok" ? fallback_status + : read_stats.status.c_str(); + const ErrorCode error = + read_stats.error_code == ErrorCode::OK ? fallback_error + : read_stats.error_code; + MC_LOG(INFO) << "storage_read_breakdown num_keys=" << keys.size() + << " total_bytes=" << total_bytes + << " alloc_us=" << alloc_us + << " plan_us=" << read_stats.plan_us + << " file_open_us=" << read_stats.file_open_us + << " disk_read_us=" << read_stats.disk_read_us + << " total_us=" << total_us + << " slowest_key=" << read_stats.slowest_key + << " slowest_disk_read_us=" + << read_stats.slowest_disk_read_us + << " io_mode=" << read_stats.io_mode + << " status=" << status + << " error_key=" << read_stats.error_key + << " error_code=" << static_cast(error); + }; // Owner-side SSD read total (OwnerSsdRead); broken down into buffer // allocation (OwnerAllocBuffer) and disk load (OwnerDiskLoad). End() is on // the success path only — the error early-returns let the dtor Abandon the @@ -341,20 +378,27 @@ tl::expected FileStorage::BatchGet( UbDiag::PerfPoint pt_alloc(PerfKey::GET_SSD_OWNER_ALLOC, UbDiag::PerfLevel::MODULE); pt_alloc.Start(); + const auto alloc_start = std::chrono::steady_clock::now(); auto allocate_res = AllocateBatch(keys, sizes); + alloc_us = std::chrono::duration_cast( + std::chrono::steady_clock::now() - alloc_start) + .count(); pt_alloc.End(allocate_res ? 0 : -1); if (!allocate_res) { LOG(ERROR) << "Failed to allocate batch objects"; + log_breakdown("alloc_fail", allocate_res.error()); return tl::make_unexpected(allocate_res.error()); } auto allocated_batch = allocate_res.value(); UbDiag::PerfPoint pt_load(PerfKey::GET_SSD_OWNER_LOAD, UbDiag::PerfLevel::MODULE); pt_load.Start(); + ScopedStorageReadStats stats_scope(breakdown_log ? &read_stats : nullptr); auto result = BatchLoad(allocated_batch->slices); pt_load.End(result ? 0 : -1); if (!result) { LOG(ERROR) << "Batch load object failed,err_code = " << result.error(); + log_breakdown("read_fail", result.error()); return tl::make_unexpected(result.error()); } @@ -381,6 +425,7 @@ tl::expected FileStorage::BatchGet( .count(); VLOG(1) << "Time taken for FileStorage::BatchGet: " << elapsed_time << "us, key size: " << keys.size() << ", batch_id: " << batch_id; + log_breakdown("ok", ErrorCode::OK); pt_read.End(0); return batch_result; } @@ -1039,19 +1084,31 @@ bool FileStorage::ReleaseBuffer(uint64_t batch_id) { UbDiag::PerfPoint pt_release(PerfKey::GET_SSD_OWNER_RELEASE, UbDiag::PerfLevel::MODULE); pt_release.Start(); + const auto start = std::chrono::steady_clock::now(); + const bool breakdown_log = mooncake::logging::ShouldSampleHiFreqLog( + mooncake::logging::CurrentTraceId()); MutexLocker locker(&client_buffer_mutex_); auto it = client_buffer_allocated_batches_.find(batch_id); - if (it != client_buffer_allocated_batches_.end()) { + const bool found = it != client_buffer_allocated_batches_.end(); + if (found) { VLOG(1) << "Releasing buffer for batch_id: " << batch_id << " (transfer completed)"; client_buffer_allocated_batches_.erase(it); - pt_release.End(0); - return true; + } else { + VLOG(1) << "batch_id " << batch_id + << " not found (may have been GC'd already)"; + } + const auto total_us = std::chrono::duration_cast( + std::chrono::steady_clock::now() - start) + .count(); + if (breakdown_log) { + MC_LOG(INFO) << "storage_release_breakdown batch_id=" << batch_id + << " total_us=" << total_us + << " found=" << (found ? 1 : 0) + << " status=" << (found ? "ok" : "not_found"); } - VLOG(1) << "batch_id " << batch_id - << " not found (may have been GC'd already)"; - pt_release.End(-1); - return false; + pt_release.End(found ? 0 : -1); + return found; } tl::expected FileStorage::ReRegisterOffloadedObjects() { diff --git a/mooncake-store/src/master_client.cpp b/mooncake-store/src/master_client.cpp index c1ba116831..9a372b35e7 100644 --- a/mooncake-store/src/master_client.cpp +++ b/mooncake-store/src/master_client.cpp @@ -5,7 +5,9 @@ #include #include +#include #include +#include #include #include #include @@ -238,6 +240,11 @@ struct RpcNameTraits<&WrappedMasterService::NotifyOffloadSuccess> { static constexpr const char* value = "NotifyOffloadSuccess"; }; +template <> +struct RpcNameTraits<&WrappedMasterService::GetOffloadEndpoints> { + static constexpr const char* value = "GetOffloadEndpoints"; +}; + template <> struct RpcNameTraits<&WrappedMasterService::PromotionObjectHeartbeat> { static constexpr const char* value = "PromotionObjectHeartbeat"; @@ -325,7 +332,19 @@ struct RpcNameTraits<&WrappedMasterService::BatchEvictDiskReplica> { template tl::expected MasterClient::invoke_rpc(Args&&... args) { + const uint64_t trace_id = mooncake::logging::CurrentTraceId(); + const bool breakdown_log = + mooncake::logging::ShouldSampleHiFreqLog(trace_id); + const auto total_start = + breakdown_log ? std::chrono::steady_clock::now() + : std::chrono::steady_clock::time_point{}; auto pool = client_accessor_.GetClientPool(); + const auto pool_end = + breakdown_log ? std::chrono::steady_clock::now() + : std::chrono::steady_clock::time_point{}; + uint64_t rpc_call_us = 0; + uint64_t result_get_us = 0; + uint64_t result_parse_us = 0; // Increment RPC counter if (metrics_) { @@ -333,7 +352,7 @@ tl::expected MasterClient::invoke_rpc(Args&&... args) { } auto start_time = std::chrono::steady_clock::now(); - return async_simple::coro::syncAwait( + auto rpc_result = async_simple::coro::syncAwait( [&]() -> async_simple::coro::Lazy> { auto ret = co_await pool->send_request( [&](coro_io::client_reuse_hint, @@ -341,11 +360,26 @@ tl::expected MasterClient::invoke_rpc(Args&&... args) { return client.send_request( std::forward(args)...); }); + if (breakdown_log) { + rpc_call_us = + std::chrono::duration_cast( + std::chrono::steady_clock::now() - pool_end) + .count(); + } if (!ret.has_value()) { LOG(ERROR) << "Client not available"; co_return tl::make_unexpected(ErrorCode::RPC_FAIL); } + const auto result_get_start = + breakdown_log ? std::chrono::steady_clock::now() + : std::chrono::steady_clock::time_point{}; auto result = co_await std::move(ret.value()); + if (breakdown_log) { + result_get_us = + std::chrono::duration_cast( + std::chrono::steady_clock::now() - result_get_start) + .count(); + } if (!result) { LOG(ERROR) << "RPC call failed: " << result.error().msg; co_return tl::make_unexpected(ErrorCode::RPC_FAIL); @@ -358,8 +392,48 @@ tl::expected MasterClient::invoke_rpc(Args&&... args) { metrics_->rpc_latency.observe( {RpcNameTraits::value}, latency.count()); } - co_return result->result(); + const auto result_parse_start = + breakdown_log ? std::chrono::steady_clock::now() + : std::chrono::steady_clock::time_point{}; + if constexpr (std::is_void_v) { + result->result(); + if (breakdown_log) { + result_parse_us = + std::chrono::duration_cast( + std::chrono::steady_clock::now() - + result_parse_start) + .count(); + } + co_return tl::expected{}; + } else { + auto response = result->result(); + if (breakdown_log) { + result_parse_us = + std::chrono::duration_cast( + std::chrono::steady_clock::now() - + result_parse_start) + .count(); + } + co_return std::move(response); + } }()); + if (breakdown_log) { + MC_LOG(INFO) << "master_rpc_client_breakdown method=" + << RpcNameTraits::value + << " pool_lookup_us=" + << std::chrono::duration_cast( + pool_end - total_start) + .count() + << " rpc_call_us=" << rpc_call_us + << " result_get_us=" << result_get_us + << " result_parse_us=" << result_parse_us + << " total_us=" + << std::chrono::duration_cast( + std::chrono::steady_clock::now() - total_start) + .count() + << " status=" << (rpc_result ? "ok" : "rpc_fail"); + } + return rpc_result; } template @@ -412,6 +486,46 @@ std::vector> MasterClient::invoke_batch_rpc( MasterClient::~MasterClient() = default; +void MasterClient::WarmupRpcPool() { + auto pool = client_accessor_.GetClientPool(); + if (!pool) { + return; + } + if (!Environ::Get().GetYltRpcPoolWarmupEnabled(true)) { + LOG(INFO) << "Master RPC pool warmup disabled by " + << "MC_YLT_RPC_POOL_WARMUP"; + return; + } + const size_t target_connections = Environ::Get().GetYltRpcPoolWarmupConnections( + pool->get_pool_config().max_connection); + if (target_connections == 0) { + LOG(INFO) << "Master RPC pool warmup disabled: target_connections=0"; + return; + } + + LOG(INFO) << "Warming up master RPC pool to " << target_connections + << " connection(s), max_connection=" + << pool->get_pool_config().max_connection; + std::vector> futures; + futures.reserve(target_connections); + for (size_t i = 0; i < target_connections; ++i) { + futures.emplace_back(std::async(std::launch::async, [this]() { + auto result = + invoke_rpc<&WrappedMasterService::ServiceReady, std::string>(); + return result.has_value(); + })); + } + + size_t ok_count = 0; + for (auto& future : futures) { + if (future.get()) { + ++ok_count; + } + } + LOG(INFO) << "Master RPC pool warmup completed: " << ok_count << "/" + << target_connections << " succeeded"; +} + ErrorCode MasterClient::Connect(const std::string& master_addr) { ScopedVLogTimer timer(1, "MasterClient::Connect"); timer.LogRequest("master_addr=", master_addr); @@ -442,6 +556,7 @@ ErrorCode MasterClient::Connect(const std::string& master_addr) { timer.LogResponse("error_code=", ErrorCode::INVALID_VERSION); return ErrorCode::INVALID_VERSION; } + WarmupRpcPool(); timer.LogResponse("error_code=", ErrorCode::OK); return ErrorCode::OK; } @@ -1040,6 +1155,17 @@ tl::expected MasterClient::NotifyOffloadSuccess( return result; } +tl::expected, ErrorCode> +MasterClient::GetOffloadEndpoints() { + ScopedVLogTimer timer(1, "MasterClient::GetOffloadEndpoints"); + timer.LogRequest("action=get_offload_endpoints"); + + auto result = invoke_rpc<&WrappedMasterService::GetOffloadEndpoints, + std::vector>(); + timer.LogResponseExpected(result); + return result; +} + tl::expected, ErrorCode> MasterClient::PromotionObjectHeartbeat(const UUID& client_id) { ScopedVLogTimer timer(1, "MasterClient::PromotionObjectHeartbeat"); diff --git a/mooncake-store/src/master_service.cpp b/mooncake-store/src/master_service.cpp index 864fa72597..9614b7b473 100644 --- a/mooncake-store/src/master_service.cpp +++ b/mooncake-store/src/master_service.cpp @@ -1311,6 +1311,42 @@ auto MasterService::GetReplicaListByRegex(const std::string& regex_pattern, return results; } +auto MasterService::GetOffloadEndpoints() + -> tl::expected, ErrorCode> { + std::unordered_set unique_endpoints; + + std::shared_lock shared_lock(snapshot_mutex_); + for (size_t i = 0; i < kNumShards; ++i) { + MetadataShardAccessorRO shard(this, i); + for (const auto& tenant_it : shard->tenants) { + for (const auto& metadata_it : tenant_it.second.metadata) { + const auto& metadata = metadata_it.second; + metadata.VisitReplicas( + [](const Replica& replica) { + return replica.is_completed() && + replica.is_local_disk_replica(); + }, + [&unique_endpoints](const Replica& replica) { + const auto& endpoint = + replica.get_descriptor() + .get_local_disk_descriptor() + .transport_endpoint; + if (!endpoint.empty()) { + unique_endpoints.emplace(endpoint); + } + }); + } + } + } + + std::vector endpoints; + endpoints.reserve(unique_endpoints.size()); + for (const auto& endpoint : unique_endpoints) { + endpoints.emplace_back(endpoint); + } + return endpoints; +} + auto MasterService::GetReplicaList(const std::string& key, const std::string& tenant_id) -> tl::expected { diff --git a/mooncake-store/src/real_client.cpp b/mooncake-store/src/real_client.cpp index 48fa2b8257..9d0987b2f1 100644 --- a/mooncake-store/src/real_client.cpp +++ b/mooncake-store/src/real_client.cpp @@ -18,9 +18,12 @@ #include #include #include +#include #include +#include #include #include +#include #include #include "mooncake_logging.h" @@ -36,6 +39,7 @@ #include "file_storage.h" #include "gpu_staging_utils.h" #include "default_config.h" +#include "environ.h" #include "shm_helper.h" #include "memory_location.h" #define UBDIAG_PERF_DEF_FILE "mooncake_perf_points.def" @@ -57,6 +61,26 @@ DEFINE_int32(http_port, 9300, namespace mooncake { namespace { +struct OffloadReadTiming { + uint64_t offload_rpc_us{0}; + uint64_t transfer_data_us{0}; + uint64_t release_buffer_us{0}; +}; + +thread_local OffloadReadTiming *current_offload_read_timing = nullptr; + +class ScopedOffloadReadTiming { + public: + explicit ScopedOffloadReadTiming(OffloadReadTiming *timing) + : previous_(current_offload_read_timing) { + current_offload_read_timing = timing; + } + ~ScopedOffloadReadTiming() { current_offload_read_timing = previous_; } + + private: + OffloadReadTiming *previous_; +}; + // Format a wall-clock time_point as "YYYY-MM-DD HH:MM:SS.ffffff" for the // request-start (t0) field in the *_breakdown logs. Only called at log // emission time (guarded by breakdown_log); the cheap system_clock::now() that @@ -661,7 +685,8 @@ tl::expected RealClient::setup_internal( const std::shared_ptr &transfer_engine, const std::string &ipc_socket_path, int local_rpc_port, bool enable_ssd_offload, bool start_offload_rpc_server, - const std::string &ssd_offload_path, const std::string &tenant_id) { + const std::string &ssd_offload_path, const std::string &tenant_id, + size_t offload_rpc_thread_num) { this->protocol = protocol; this->ipc_socket_path_ = ipc_socket_path; const bool should_use_hugepage = @@ -923,14 +948,21 @@ tl::expected RealClient::setup_internal( if (enable_ssd_offload && start_offload_rpc_server) { // Start RPC server for offload operations (batch_get / release_buffer). // Use port 0 to let the OS auto-allocate an available port. + if (offload_rpc_thread_num == 0) { + LOG(WARNING) << "Invalid offload_rpc_thread_num=0, using 1"; + offload_rpc_thread_num = 1; + } offload_rpc_server_ = - std::make_unique(1, 0, "0.0.0.0"); + std::make_unique( + offload_rpc_thread_num, 0, "0.0.0.0"); offload_rpc_server_ ->register_handler<&RealClient::batch_get_offload_object>(this); offload_rpc_server_ ->register_handler<&RealClient::batch_get_offload_object_push>(this); offload_rpc_server_ ->register_handler<&RealClient::release_offload_buffer>(this); + offload_rpc_server_ + ->register_handler<&RealClient::service_ready_internal>(this); offload_rpc_server_->async_start(); auto err = offload_rpc_server_->get_errc(); if (err) { @@ -940,7 +972,8 @@ tl::expected RealClient::setup_internal( return tl::unexpected(ErrorCode::INTERNAL_ERROR); } offload_rpc_port_ = offload_rpc_server_->port(); - LOG(INFO) << "Offload RPC server started on port " << offload_rpc_port_; + LOG(INFO) << "Offload RPC server started on port " << offload_rpc_port_ + << " with " << offload_rpc_thread_num << " thread(s)"; // Build local_rpc_addr from hostname + auto-allocated port this->local_rpc_addr = buildHostNameWithPort( @@ -962,6 +995,22 @@ tl::expected RealClient::setup_internal( } } client_requester_ = std::make_shared(); + if (Environ::Get().GetYltRpcPoolWarmupEnabled(true)) { + auto endpoints_result = client_->GetOffloadEndpoints(); + if (!endpoints_result) { + LOG(WARNING) << "Failed to fetch offload RPC endpoints for pool " + << "warmup: " + << toString(endpoints_result.error()); + } else { + const auto& endpoints = endpoints_result.value(); + LOG(INFO) << "Warming up offload RPC pools for " + << endpoints.size() << " endpoint(s) during client " + << "initialization"; + for (const auto& endpoint : endpoints) { + client_requester_->WarmupRpcPool(endpoint); + } + } + } if (FLAGS_enable_http_server) { if (start_http_server() != 0) { LOG(ERROR) << "Failed to start HTTP server on port " @@ -976,15 +1025,7 @@ tl::expected RealClient::setup_internal( // Controlled by MC_STORE_WARMUP (default: disabled to // preserve existing behaviour). Set to 1/true/yes to enable. if (client_) { - const char* env = std::getenv("MC_STORE_WARMUP"); - bool enable_warmup = false; - if (env) { - std::string val = env; - std::transform(val.begin(), val.end(), val.begin(), - [](unsigned char c) { return std::tolower(c); }); - enable_warmup = - (val == "1" || val == "true" || val == "yes" || val == "on"); - } + const bool enable_warmup = Environ::Get().GetStoreWarmupEnabled(false); if (enable_warmup) { // warmup issues RDMA transfers using a buffer from // client_buffer_allocator_, which must be registered with the @@ -1023,7 +1064,7 @@ int RealClient::setup_real( local_hostname, metadata_server, global_segment_size, local_buffer_size, protocol, rdma_devices, master_server_addr, transfer_engine, ipc_socket_path, 50052, enable_ssd_offload, true, ssd_offload_path, - tenant_id)); + tenant_id, Environ::Get().GetOffloadRpcThreadNum(8))); } namespace { @@ -1125,6 +1166,13 @@ tl::expected RealClient::setup_internal( std::string ssd_offload_path = get_config(config, "ssd_offload_path"); std::string tenant_id = get_config(config, CONFIG_KEY_TENANT_ID, "default"); + auto offload_rpc_thread_num_opt = + get_config_size(config, "offload_rpc_thread_num", + Environ::Get().GetOffloadRpcThreadNum(8)); + if (!offload_rpc_thread_num_opt.has_value()) { + return tl::unexpected(ErrorCode::INVALID_PARAMS); + } + size_t offload_rpc_thread_num = offload_rpc_thread_num_opt.value(); std::string enable_ssd_offload_str = get_config(config, "enable_ssd_offload", "false"); @@ -1137,7 +1185,8 @@ tl::expected RealClient::setup_internal( return setup_internal( local_hostname, metadata_server, global_segment_size, local_buffer_size, protocol, rdma_devices, master_server_addr, nullptr, ipc_socket_path, - 50052, enable_ssd_offload, true, ssd_offload_path, tenant_id); + 50052, enable_ssd_offload, true, ssd_offload_path, tenant_id, + offload_rpc_thread_num); } tl::expected RealClient::initAll_internal( @@ -1766,7 +1815,8 @@ tl::expected RealClient::put_internal( } // One dice roll per request (MC_HIFREQ_LOG_SAMPLE_RATE, default 1.0) gates // both the breakdown timing capture and its emission below. - const bool breakdown_log = mooncake::logging::ShouldSampleHiFreqLog(); + const bool breakdown_log = mooncake::logging::ShouldSampleHiFreqLog( + mooncake::logging::CurrentTraceId()); auto t0 = breakdown_log ? std::chrono::steady_clock::now() : std::chrono::steady_clock::time_point{}; auto t0_wall = breakdown_log ? std::chrono::system_clock::now() @@ -1879,7 +1929,8 @@ tl::expected RealClient::put_batch_internal( } // One dice roll per request (MC_HIFREQ_LOG_SAMPLE_RATE, default 1.0) gates // both the breakdown timing capture and its emission below. - const bool breakdown_log = mooncake::logging::ShouldSampleHiFreqLog(); + const bool breakdown_log = mooncake::logging::ShouldSampleHiFreqLog( + mooncake::logging::CurrentTraceId()); auto t0 = breakdown_log ? std::chrono::steady_clock::now() : std::chrono::steady_clock::time_point{}; auto t0_wall = breakdown_log ? std::chrono::system_clock::now() @@ -2717,13 +2768,15 @@ std::shared_ptr RealClient::get_buffer_internal( // One dice roll per request (MC_HIFREQ_LOG_SAMPLE_RATE, default 1.0) gates // both the breakdown timing capture and its emission below. - const bool breakdown_log = mooncake::logging::ShouldSampleHiFreqLog(); + const bool breakdown_log = mooncake::logging::ShouldSampleHiFreqLog( + mooncake::logging::CurrentTraceId()); auto t0 = breakdown_log ? std::chrono::steady_clock::now() : std::chrono::steady_clock::time_point{}; auto t0_wall = breakdown_log ? std::chrono::system_clock::now() : std::chrono::system_clock::time_point{}; - auto t_query = t0, t_select = t0, t_alloc = t0; + auto t_query = t0, t_local_endpoints = t0, t_select = t0, t_alloc = t0; std::string replica_type; + std::string remote_endpoint = "-"; // Query the object info UbDiag::PerfPoint pt_query(PerfKey::GET_INTERNAL_QUERY, @@ -2757,6 +2810,9 @@ std::shared_ptr RealClient::get_buffer_internal( UbDiag::PerfLevel::MODULE); pt_select.Start(); auto local_endpoints = client_->GetLocalEndpoints(); + if (breakdown_log) { + t_local_endpoints = std::chrono::steady_clock::now(); + } const auto *best_replica = SelectBestReplica(replica_list, local_endpoints); if (breakdown_log) t_select = std::chrono::steady_clock::now(); pt_select.End(best_replica ? 0 : -1); @@ -2772,10 +2828,12 @@ std::shared_ptr RealClient::get_buffer_internal( if (replica.is_memory_replica()) { const auto &endpoint = replica.get_memory_descriptor() .buffer_descriptor.transport_endpoint_; + remote_endpoint = endpoint; bool is_local = local_endpoints.count(endpoint) > 0; replica_type = is_local ? "memory_local" : "memory_remote"; } else if (replica.is_local_disk_replica()) { const auto &ld_desc = replica.get_local_disk_descriptor(); + remote_endpoint = ld_desc.transport_endpoint; bool is_local_holder = (ld_desc.client_id == client_->getClientId()); replica_type = is_local_holder ? "local_disk_local" : "local_disk_remote"; @@ -2811,6 +2869,14 @@ std::shared_ptr RealClient::get_buffer_internal( auto select_us = std::chrono::duration_cast( t_select - t_query) .count(); + auto local_endpoints_us = + std::chrono::duration_cast( + t_local_endpoints - t_query) + .count(); + auto select_replica_us = + std::chrono::duration_cast( + t_select - t_local_endpoints) + .count(); auto alloc_us = std::chrono::duration_cast( t_alloc - t_select) .count(); @@ -2822,10 +2888,16 @@ std::shared_ptr RealClient::get_buffer_internal( .count(); MC_LOG(INFO) << "get_breakdown key[" << key << "] start_time[" << FormatWallClock(t0_wall) << "] query_us[" << query_us - << "] select_us[" << select_us << "] alloc_us[" << alloc_us + << "] select_us[" << select_us + << "] local_endpoints_us[" << local_endpoints_us + << "] select_replica_us[" << select_replica_us + << "] alloc_us[" << alloc_us << "] read_us[" << read_us << "] total_us[" << total_us - << "] type[" << replica_type << "] status[" << status - << "]"; + << "] type[" << replica_type << "] replica_count[" + << replica_list.size() << "] selected_type[" + << replica_type << "] local_endpoint[" + << client_->GetSegmentEndpoint() << "] remote_endpoint[" + << remote_endpoint << "] status[" << status << "]"; }; if (best_replica->is_local_disk_replica()) { @@ -3095,7 +3167,8 @@ RealClient::batch_get_buffer_internal( // One dice roll per request (MC_HIFREQ_LOG_SAMPLE_RATE, default 1.0) gates // both the breakdown timing capture and its emission below. - const bool breakdown_log = mooncake::logging::ShouldSampleHiFreqLog(); + const bool breakdown_log = mooncake::logging::ShouldSampleHiFreqLog( + mooncake::logging::CurrentTraceId()); auto t0 = breakdown_log ? std::chrono::steady_clock::now() : std::chrono::steady_clock::time_point{}; auto t0_wall = breakdown_log ? std::chrono::system_clock::now() @@ -3503,6 +3576,7 @@ RealClient::resolve_ranged_read_metadata(const std::string &key) { UbDiag::PerfLevel::MODULE); pt_select.Start(); auto local_endpoints = client_->GetLocalEndpoints(); + auto t_local_endpoints = std::chrono::steady_clock::now(); const auto *best_replica = SelectBestReplica(replica_list, local_endpoints); auto t_select = std::chrono::steady_clock::now(); pt_select.End(best_replica ? 0 : -1); @@ -3511,17 +3585,21 @@ RealClient::resolve_ranged_read_metadata(const std::string &key) { return tl::unexpected(ErrorCode::INVALID_REPLICA); } + const size_t replica_count = replica_list.size(); auto query_value = std::move(query_result.value()); auto replica = *best_replica; auto total_size = calculate_total_size(replica); std::string replica_type; + std::string remote_endpoint = "-"; if (replica.is_memory_replica()) { const auto &endpoint = replica.get_memory_descriptor() .buffer_descriptor.transport_endpoint_; + remote_endpoint = endpoint; bool is_local = local_endpoints.count(endpoint) > 0; replica_type = is_local ? "memory_local" : "memory_remote"; } else if (replica.is_local_disk_replica()) { const auto &ld_desc = replica.get_local_disk_descriptor(); + remote_endpoint = ld_desc.transport_endpoint; bool is_local_holder = (ld_desc.client_id == client_->getClientId()); replica_type = is_local_holder ? "local_disk_local" : "local_disk_remote"; @@ -3539,7 +3617,18 @@ RealClient::resolve_ranged_read_metadata(const std::string &key) { .select_us = std::chrono::duration_cast( t_select - t_query) .count(), - .replica_type = std::move(replica_type)}; + .local_endpoints_us = + std::chrono::duration_cast( + t_local_endpoints - t_query) + .count(), + .select_replica_us = + std::chrono::duration_cast( + t_select - t_local_endpoints) + .count(), + .replica_type = std::move(replica_type), + .replica_count = replica_count, + .local_endpoint = client_->GetSegmentEndpoint(), + .remote_endpoint = std::move(remote_endpoint)}; } tl::expected RealClient::execute_ranged_read( @@ -3743,7 +3832,8 @@ tl::expected RealClient::get_into_range_internal( size_t size, bool size_is_buffer_capacity) { // One dice roll per request (MC_HIFREQ_LOG_SAMPLE_RATE, default 1.0) gates // both the breakdown timing capture and its emission below. - const bool breakdown_log = mooncake::logging::ShouldSampleHiFreqLog(); + const bool breakdown_log = mooncake::logging::ShouldSampleHiFreqLog( + mooncake::logging::CurrentTraceId()); auto t0 = breakdown_log ? std::chrono::steady_clock::now() : std::chrono::steady_clock::time_point{}; auto t0_wall = breakdown_log ? std::chrono::system_clock::now() @@ -3762,7 +3852,10 @@ tl::expected RealClient::get_into_range_internal( .count(); MC_LOG(INFO) << "get_into_breakdown key[" << key << "] start_time[" << FormatWallClock(t0_wall) - << "] query_us[0] select_us[0] read_us[0] total_us[" + << "] query_us[0] select_us[0] local_endpoints_us[0]" + " select_replica_us[0] read_us[0]" + " offload_rpc_us[0] transfer_data_us[0]" + " release_buffer_us[0] read_overhead_us[0] total_us[" << total_us << "] type[unknown] mode[unknown] status[" << toString(metadata_result.error()) << "]"; } @@ -3770,6 +3863,9 @@ tl::expected RealClient::get_into_range_internal( } const auto &metadata = metadata_result.value(); + OffloadReadTiming offload_timing; + ScopedOffloadReadTiming offload_timing_scope( + breakdown_log ? &offload_timing : nullptr); auto read_start = breakdown_log ? std::chrono::steady_clock::now() : std::chrono::steady_clock::time_point{}; auto read_result = @@ -3802,12 +3898,31 @@ tl::expected RealClient::get_into_range_internal( auto total_us = std::chrono::duration_cast(read_end - t0) .count(); + const uint64_t known_read_us = + offload_timing.offload_rpc_us + offload_timing.transfer_data_us + + offload_timing.release_buffer_us; + const uint64_t read_overhead_us = + read_us > static_cast(known_read_us) + ? static_cast(read_us) - known_read_us + : 0; MC_LOG(INFO) << "get_into_breakdown key[" << key << "] start_time[" << FormatWallClock(t0_wall) << "] query_us[" << metadata.query_us << "] select_us[" - << metadata.select_us << "] read_us[" << read_us + << metadata.select_us << "] local_endpoints_us[" + << metadata.local_endpoints_us << "] select_replica_us[" + << metadata.select_replica_us << "] read_us[" << read_us + << "] offload_rpc_us[" << offload_timing.offload_rpc_us + << "] transfer_data_us[" + << offload_timing.transfer_data_us + << "] release_buffer_us[" + << offload_timing.release_buffer_us + << "] read_overhead_us[" << read_overhead_us << "] total_us[" << total_us << "] type[" << metadata.replica_type << "] mode[" << mode + << "] replica_count[" << metadata.replica_count + << "] selected_type[" << metadata.replica_type + << "] local_endpoint[" << metadata.local_endpoint + << "] remote_endpoint[" << metadata.remote_endpoint << "] status[" << status << "]"; } @@ -4137,7 +4252,8 @@ std::vector> RealClient::batch_put_from_internal( // One dice roll per request (MC_HIFREQ_LOG_SAMPLE_RATE, default 1.0) gates // both the breakdown timing capture and its emission below. - const bool breakdown_log = mooncake::logging::ShouldSampleHiFreqLog(); + const bool breakdown_log = mooncake::logging::ShouldSampleHiFreqLog( + mooncake::logging::CurrentTraceId()); auto t0 = breakdown_log ? std::chrono::steady_clock::now() : std::chrono::steady_clock::time_point{}; auto t0_wall = breakdown_log ? std::chrono::system_clock::now() @@ -4234,7 +4350,8 @@ tl::expected RealClient::put_from_internal( // One dice roll per request (MC_HIFREQ_LOG_SAMPLE_RATE, default 1.0) gates // both the breakdown timing capture and its emission below. - const bool breakdown_log = mooncake::logging::ShouldSampleHiFreqLog(); + const bool breakdown_log = mooncake::logging::ShouldSampleHiFreqLog( + mooncake::logging::CurrentTraceId()); auto t0 = breakdown_log ? std::chrono::steady_clock::now() : std::chrono::steady_clock::time_point{}; auto t0_wall = breakdown_log ? std::chrono::system_clock::now() @@ -4974,7 +5091,8 @@ RealClient::batch_get_into_internal(const std::vector &keys, auto start_time = std::chrono::steady_clock::now(); // One dice roll per request (MC_HIFREQ_LOG_SAMPLE_RATE, default 1.0) gates // both the breakdown timing capture and its emission below. - const bool breakdown_log = mooncake::logging::ShouldSampleHiFreqLog(); + const bool breakdown_log = mooncake::logging::ShouldSampleHiFreqLog( + mooncake::logging::CurrentTraceId()); auto t0_wall = breakdown_log ? std::chrono::system_clock::now() : std::chrono::system_clock::time_point{}; auto t_query = start_time, t_prep = start_time; @@ -6248,38 +6366,128 @@ tl::expected RealClient::query_task( } async_simple::coro::Lazy> RealClient::batch_get_offload_object(const std::vector &keys, - const std::vector &sizes) { + const std::vector &sizes, + uint64_t trace_id) { + const auto handler_start = std::chrono::steady_clock::now(); + const bool breakdown_log = + mooncake::logging::ShouldSampleHiFreqLog(trace_id); // Run SSD I/O on a dedicated thread pool so the coro_rpc IO thread is // free to handle ping and other RPCs. Same heap-owned state pattern as // batch_get_into_dummy_helper: the lambda captures only a raw pointer. struct CallState { + CallState() + : pt_queue(PerfKey::GET_SSD_OWNER_RPC_QUEUE, + UbDiag::PerfLevel::DEBUG), + pt_batch_get(PerfKey::GET_SSD_OWNER_RPC_BATCH_GET, + UbDiag::PerfLevel::MODULE), + pt_resume(PerfKey::GET_SSD_OWNER_RPC_RESUME, + UbDiag::PerfLevel::DEBUG) {} std::vector keys; std::vector sizes; std::shared_ptr file_storage; + uint64_t trace_id; + std::chrono::steady_clock::time_point worker_start; + std::chrono::steady_clock::time_point worker_end; + UbDiag::PerfPoint pt_queue; + UbDiag::PerfPoint pt_batch_get; + UbDiag::PerfPoint pt_resume; }; auto state = std::make_unique(); state->keys = keys; state->sizes = sizes; state->file_storage = file_storage_; + state->trace_id = trace_id; auto *s = state.get(); + s->pt_queue.Start(); auto try_result = co_await coro_io::post( - [s]() { return s->file_storage->BatchGet(s->keys, s->sizes); }); + [s]() { + s->worker_start = std::chrono::steady_clock::now(); + s->pt_queue.End(0); + mooncake::logging::ScopedTraceId trace(s->trace_id); + s->pt_batch_get.Start(); + auto result = s->file_storage->BatchGet(s->keys, s->sizes); + s->pt_batch_get.End(result ? 0 : -1); + s->worker_end = std::chrono::steady_clock::now(); + s->pt_resume.Start(); + return result; + }); + const auto resumed = std::chrono::steady_clock::now(); + s->pt_resume.End(0); auto result = try_result.value(); if (!result) { LOG(ERROR) << "Batch get offload object failed,err_code = " << result.error(); + if (breakdown_log) { + mooncake::logging::ScopedTraceId trace(trace_id); + MC_LOG(INFO) + << "offload_rpc_server_breakdown mode=pull" + << " num_keys=" << keys.size() + << " queue_us=" + << std::chrono::duration_cast( + s->worker_start - handler_start) + .count() + << " batch_get_us=" + << std::chrono::duration_cast( + s->worker_end - s->worker_start) + .count() + << " resume_us=" + << std::chrono::duration_cast( + resumed - s->worker_end) + .count() + << " response_us=0 total_us=" + << std::chrono::duration_cast( + resumed - handler_start) + .count() + << " status=read_fail error_code=" + << static_cast(result.error()); + } co_return tl::make_unexpected(result.error()); } - co_return BatchGetOffloadObjectResponse( + const auto response_start = std::chrono::steady_clock::now(); + BatchGetOffloadObjectResponse response( result.value().batch_id, std::move(result.value().pointers), client_->GetSegmentEndpoint(), file_storage_->config_.client_buffer_gc_ttl_ms); + const auto response_end = std::chrono::steady_clock::now(); + const auto queue_us = std::chrono::duration_cast( + s->worker_start - handler_start) + .count(); + const auto batch_get_us = + std::chrono::duration_cast(s->worker_end - + s->worker_start) + .count(); + const auto resume_us = + std::chrono::duration_cast(resumed - + s->worker_end) + .count(); + const auto response_us = + std::chrono::duration_cast(response_end - + response_start) + .count(); + const auto total_us = + std::chrono::duration_cast(response_end - + handler_start) + .count(); + if (breakdown_log) { + mooncake::logging::ScopedTraceId trace(trace_id); + MC_LOG(INFO) << "offload_rpc_server_breakdown mode=pull" + << " num_keys=" << keys.size() + << " queue_us=" << queue_us + << " batch_get_us=" << batch_get_us + << " resume_us=" << resume_us + << " response_us=" << response_us + << " total_us=" << total_us << " status=ok"; + } + co_return response; } async_simple::coro::Lazy< tl::expected> RealClient::batch_get_offload_object_push( const BatchGetOffloadObjectPushRequest &req) { + const auto handler_start = std::chrono::steady_clock::now(); + const bool breakdown_log = + mooncake::logging::ShouldSampleHiFreqLog(req.trace_id); if (!file_storage_) { LOG(ERROR) << "batch_get_offload_object_push called but file_storage_ is null"; @@ -6297,23 +6505,53 @@ RealClient::batch_get_offload_object_push( // heap-owned state pattern as batch_get_offload_object: the lambda captures // only a raw pointer. struct CallState { + CallState() + : pt_queue(PerfKey::GET_SSD_OWNER_RPC_QUEUE, + UbDiag::PerfLevel::DEBUG), + pt_batch_get(PerfKey::GET_SSD_OWNER_RPC_BATCH_GET, + UbDiag::PerfLevel::MODULE), + pt_resume(PerfKey::GET_SSD_OWNER_RPC_RESUME, + UbDiag::PerfLevel::DEBUG) {} BatchGetOffloadObjectPushRequest req; std::shared_ptr file_storage; Client *client; + uint64_t trace_id; + std::chrono::steady_clock::time_point worker_start; + std::chrono::steady_clock::time_point worker_end; + uint64_t batch_get_us{0}; + uint64_t transfer_us{0}; + uint64_t release_us{0}; + UbDiag::PerfPoint pt_queue; + UbDiag::PerfPoint pt_batch_get; + UbDiag::PerfPoint pt_resume; }; auto state = std::make_unique(); state->req = req; state->file_storage = file_storage_; state->client = client_.get(); + state->trace_id = req.trace_id; auto *s = state.get(); + s->pt_queue.Start(); auto try_result = co_await coro_io::post([s]() -> tl::expected { + s->worker_start = std::chrono::steady_clock::now(); + s->pt_queue.End(0); + mooncake::logging::ScopedTraceId trace(s->trace_id); // Stage the on-disk blobs into the local ClientBuffer // (FileStorage::BatchGet carries the OwnerSsdRead breakdown). + const auto batch_get_start = std::chrono::steady_clock::now(); + s->pt_batch_get.Start(); auto result = s->file_storage->BatchGet(s->req.keys, s->req.sizes); + s->pt_batch_get.End(result ? 0 : -1); + s->batch_get_us = + std::chrono::duration_cast( + std::chrono::steady_clock::now() - batch_get_start) + .count(); if (!result) { LOG(ERROR) << "Push offload BatchGet failed, err_code = " << result.error(); + s->worker_end = std::chrono::steady_clock::now(); + s->pt_resume.Start(); return tl::make_unexpected(result.error()); } const uint64_t batch_id = result.value().batch_id; @@ -6321,32 +6559,99 @@ RealClient::batch_get_offload_object_push( UbDiag::PerfPoint pt_write(PerfKey::GET_SSD_OWNER_PUSH_WRITE, UbDiag::PerfLevel::MODULE); pt_write.Start(); + const auto transfer_start = std::chrono::steady_clock::now(); auto write_result = s->client->BatchPushOffloadObject( s->req.requester_te_addr, s->req.keys, result.value().pointers, s->req.dst_slices); + s->transfer_us = + std::chrono::duration_cast( + std::chrono::steady_clock::now() - transfer_start) + .count(); pt_write.End(write_result ? 0 : -1); // The WRITE has completed (BatchPushOffloadObject blocks on the // transfer future), so the ClientBuffer can be reclaimed // immediately instead of waiting for the GC lease // (FileStorage::ReleaseBuffer carries OwnerReleaseBuffer). + const auto release_start = std::chrono::steady_clock::now(); s->file_storage->ReleaseBuffer(batch_id); + s->release_us = + std::chrono::duration_cast( + std::chrono::steady_clock::now() - release_start) + .count(); + s->worker_end = std::chrono::steady_clock::now(); + s->pt_resume.Start(); return write_result; }); + const auto resumed = std::chrono::steady_clock::now(); + s->pt_resume.End(0); auto pushed = try_result.value(); if (!pushed) { LOG(ERROR) << "Push offload transfer failed, err_code = " << pushed.error(); + if (breakdown_log) { + mooncake::logging::ScopedTraceId trace(req.trace_id); + MC_LOG(INFO) + << "offload_rpc_server_breakdown mode=push" + << " num_keys=" << req.keys.size() + << " queue_us=" + << std::chrono::duration_cast( + s->worker_start - handler_start) + .count() + << " batch_get_us=" << s->batch_get_us + << " transfer_data_us=" << s->transfer_us + << " release_buffer_us=" << s->release_us + << " resume_us=" + << std::chrono::duration_cast( + resumed - s->worker_end) + .count() + << " response_us=0 total_us=" + << std::chrono::duration_cast( + resumed - handler_start) + .count() + << " status=failed error_code=" + << static_cast(pushed.error()); + } co_return tl::make_unexpected(pushed.error()); } - co_return BatchGetOffloadObjectPushResponse(ErrorCode::OK); -} - -bool RealClient::release_offload_buffer(uint64_t batch_id) { + const auto response_start = std::chrono::steady_clock::now(); + BatchGetOffloadObjectPushResponse response(ErrorCode::OK); + const auto response_end = std::chrono::steady_clock::now(); + if (breakdown_log) { + mooncake::logging::ScopedTraceId trace(req.trace_id); + MC_LOG(INFO) + << "offload_rpc_server_breakdown mode=push" + << " num_keys=" << req.keys.size() + << " queue_us=" + << std::chrono::duration_cast( + s->worker_start - handler_start) + .count() + << " batch_get_us=" << s->batch_get_us + << " transfer_data_us=" << s->transfer_us + << " release_buffer_us=" << s->release_us + << " resume_us=" + << std::chrono::duration_cast( + resumed - s->worker_end) + .count() + << " response_us=" + << std::chrono::duration_cast( + response_end - response_start) + .count() + << " total_us=" + << std::chrono::duration_cast( + response_end - handler_start) + .count() + << " status=ok"; + } + co_return response; +} + +bool RealClient::release_offload_buffer(uint64_t batch_id, uint64_t trace_id) { if (!file_storage_) { LOG(WARNING) << "release_offload_buffer called but file_storage_ is null"; return false; } + mooncake::logging::ScopedTraceId trace(trace_id); // Owner-side buffer release triggered by the requester's RPC (pull path). // FileStorage::ReleaseBuffer carries the OwnerReleaseBuffer perf point. return file_storage_->ReleaseBuffer(batch_id); @@ -6358,6 +6663,8 @@ RealClient::batch_get_into_offload_object_internal( std::unordered_map> &objects) { offload_rpc_read_count_.fetch_add(1, std::memory_order_relaxed); auto start_time = std::chrono::steady_clock::now(); + const bool breakdown_log = mooncake::logging::ShouldSampleHiFreqLog( + mooncake::logging::CurrentTraceId()); std::vector keys; std::vector storage_keys; std::vector sizes; @@ -6394,12 +6701,22 @@ RealClient::batch_get_into_offload_object_internal( push_req.sizes = sizes; push_req.requester_te_addr = client_->GetSegmentEndpoint(); push_req.dst_slices = std::move(dst_slices); + push_req.trace_id = mooncake::logging::CurrentTraceId(); UbDiag::PerfPoint pt_push(PerfKey::GET_SSD_OFFLOAD_RPC, UbDiag::PerfLevel::MODULE); pt_push.Start(); + const auto rpc_start = std::chrono::steady_clock::now(); + ClientRequester::RpcTiming rpc_timing; auto pushResp = client_requester_->batch_get_offload_object_push( - target_rpc_service_addr, push_req); + target_rpc_service_addr, push_req, &rpc_timing); + const auto rpc_us = + std::chrono::duration_cast( + std::chrono::steady_clock::now() - rpc_start) + .count(); + if (current_offload_read_timing) { + current_offload_read_timing->offload_rpc_us = rpc_us; + } pt_push.End(pushResp ? 0 : -1); if (!pushResp) { LOG(ERROR) << "Push offload failed with error: " @@ -6412,23 +6729,32 @@ RealClient::batch_get_into_offload_object_internal( return tl::make_unexpected(pushResp->error_code); } auto end_time = std::chrono::steady_clock::now(); - auto elapsed_time = static_cast( - std::chrono::duration_cast(end_time - + const auto total_us = + std::chrono::duration_cast(end_time - start_time) - .count()); - LOG(INFO) << "Time taken for batch_get_into_offload_object_internal " - "(push): " - << elapsed_time << "ms, with target_rpc_service_addr: " - << target_rpc_service_addr - << ", key size: " << objects.size(); + .count(); + if (breakdown_log) { + MC_LOG(INFO) << "offload_read_breakdown mode=push" + << " endpoint=" << target_rpc_service_addr + << " num_keys=" << objects.size() + << " total_us=" << total_us << " status=ok"; + } return {}; } UbDiag::PerfPoint pt_rpc(PerfKey::GET_SSD_OFFLOAD_RPC, UbDiag::PerfLevel::MODULE); pt_rpc.Start(); + const auto rpc_start = std::chrono::steady_clock::now(); + ClientRequester::RpcTiming rpc_timing; auto batchGetResp = client_requester_->batch_get_offload_object( - target_rpc_service_addr, storage_keys, sizes); + target_rpc_service_addr, storage_keys, sizes, &rpc_timing); + const auto rpc_us = std::chrono::duration_cast( + std::chrono::steady_clock::now() - rpc_start) + .count(); + if (current_offload_read_timing) { + current_offload_read_timing->offload_rpc_us = rpc_us; + } pt_rpc.End(batchGetResp ? 0 : -1); if (!batchGetResp) { LOG(ERROR) << "Batch get offload object failed with error: " @@ -6443,29 +6769,53 @@ RealClient::batch_get_into_offload_object_internal( UbDiag::PerfPoint pt_transfer(PerfKey::GET_SSD_TRANSFER_DATA, UbDiag::PerfLevel::MODULE); pt_transfer.Start(); + const auto transfer_start = std::chrono::steady_clock::now(); auto result = client_->BatchGetOffloadObject(batchGetResp->transfer_engine_addr, keys, batchGetResp->pointers, objects); + const auto transfer_us = + std::chrono::duration_cast( + std::chrono::steady_clock::now() - transfer_start) + .count(); + if (current_offload_read_timing) { + current_offload_read_timing->transfer_data_us = transfer_us; + } pt_transfer.End(result ? 0 : -1); auto end_time = std::chrono::steady_clock::now(); auto elapsed_time = static_cast( std::chrono::duration_cast(end_time - start_time) .count()); - LOG(INFO) << "Time taken for batch_get_into_offload_object_internal: " - << elapsed_time - << "ms, with target_rpc_service_addr: " << target_rpc_service_addr - << ", key size: " << objects.size() - << ", batch_id: " << batchGetResp->batch_id - << ", gc ttl: " << batchGetResp->gc_ttl_ms << "ms."; - - // Release buffer immediately after transfer completion (fire-and-forget) - // This allows early buffer reclamation instead of waiting for GC lease + if (breakdown_log) { + MC_LOG(INFO) << "offload_read_breakdown mode=pull" + << " endpoint=" << target_rpc_service_addr + << " num_keys=" << objects.size() + << " batch_id=" << batchGetResp->batch_id + << " gc_ttl_ms=" << batchGetResp->gc_ttl_ms + << " pre_release_total_us=" + << std::chrono::duration_cast( + end_time - start_time) + .count() + << " status=" << (result ? "ok" : "transfer_fail"); + } + + // Release the owner buffer immediately after transfer completion. This RPC + // is synchronous; measuring it separately exposes owner lock/reclaim stalls. UbDiag::PerfPoint pt_release(PerfKey::GET_SSD_RELEASE_BUFFER, UbDiag::PerfLevel::MODULE); pt_release.Start(); + const auto release_start = std::chrono::steady_clock::now(); + ClientRequester::RpcTiming release_timing; client_requester_->release_offload_buffer(target_rpc_service_addr, - batchGetResp->batch_id); + batchGetResp->batch_id, + &release_timing); + const auto release_us = + std::chrono::duration_cast( + std::chrono::steady_clock::now() - release_start) + .count(); + if (current_offload_read_timing) { + current_offload_read_timing->release_buffer_us = release_us; + } pt_release.End(0); if (!result) { @@ -6481,6 +6831,14 @@ RealClient::batch_get_into_offload_object_internal( ClientRequester::ClientRequester() { coro_io::client_pool::pool_config pool_conf{}; + pool_conf.max_connection = + Environ::Get().GetYltRpcPoolMaxConnection(pool_conf.max_connection); + pool_conf.idle_timeout = std::chrono::milliseconds( + Environ::Get().GetYltRpcPoolIdleTimeoutMs( + pool_conf.idle_timeout.count())); + pool_conf.short_connect_idle_timeout = std::chrono::milliseconds( + Environ::Get().GetYltRpcPoolShortIdleTimeoutMs( + pool_conf.short_connect_idle_timeout.count())); const char *value = std::getenv("MC_RPC_PROTOCOL"); if (value && std::string_view(value) == "rdma") { pool_conf.client_config.socket_config = @@ -6503,13 +6861,80 @@ ClientRequester::ClientRequester() { pool_conf); } +void ClientRequester::WarmupRpcPool(const std::string &client_addr) { + if (!Environ::Get().GetYltRpcPoolWarmupEnabled(true)) { + return; + } + + std::unique_lock lock(warmed_rpc_pool_mutex_); + const auto [_, inserted] = warmed_rpc_pools_.insert(client_addr); + if (!inserted) { + return; + } + + auto client_pool = client_pools_->at(client_addr); + const size_t target_connections = + Environ::Get().GetYltRpcPoolWarmupConnections( + client_pool->get_pool_config().max_connection); + if (target_connections == 0) { + LOG(INFO) << "Offload RPC pool warmup disabled for " << client_addr + << " by MC_YLT_RPC_POOL_WARMUP_CONNECTIONS=0"; + return; + } + + LOG(INFO) << "Warming up offload RPC pool for " << client_addr << " to " + << target_connections << " connection(s), max_connection=" + << client_pool->get_pool_config().max_connection; + + std::vector> futures; + futures.reserve(target_connections); + for (size_t i = 0; i < target_connections; ++i) { + futures.emplace_back(std::async(std::launch::async, [this, client_addr]() { + auto result = + invoke_rpc<&RealClient::service_ready_internal, void>( + client_addr, nullptr); + return result.has_value(); + })); + } + + size_t ok_count = 0; + for (auto &future : futures) { + try { + if (future.get()) { + ++ok_count; + } + } catch (const std::exception &e) { + LOG(WARNING) << "Offload RPC pool warmup task failed for " + << client_addr << ": " << e.what(); + } + } + + LOG(INFO) << "Offload RPC pool warmup completed for " << client_addr + << ": " << ok_count << "/" << target_connections + << " succeeded"; +} + tl::expected ClientRequester::batch_get_offload_object(const std::string &client_addr, const std::vector &keys, - const std::vector &sizes) { + const std::vector &sizes, + RpcTiming *timing) { + const uint64_t trace_id = mooncake::logging::CurrentTraceId(); auto result = invoke_rpc<&RealClient::batch_get_offload_object, - BatchGetOffloadObjectResponse>(client_addr, keys, sizes); + BatchGetOffloadObjectResponse>( + client_addr, timing, keys, sizes, trace_id); + if (timing && mooncake::logging::ShouldSampleHiFreqLog(trace_id)) { + MC_LOG(INFO) << "offload_rpc_client_breakdown mode=pull" + << " endpoint=" << client_addr + << " num_keys=" << keys.size() + << " pool_lookup_us=" << timing->pool_lookup_us + << " rpc_call_us=" << timing->rpc_call_us + << " result_get_us=" << timing->result_get_us + << " result_parse_us=" << timing->result_parse_us + << " total_us=" << timing->total_us + << " status=" << (result ? "ok" : "rpc_fail"); + } if (!result) { LOG(ERROR) << "Failed to invoke batch_get_offload_object, client_addr = " @@ -6521,10 +6946,22 @@ ClientRequester::batch_get_offload_object(const std::string &client_addr, tl::expected ClientRequester::batch_get_offload_object_push( const std::string &client_addr, - const BatchGetOffloadObjectPushRequest &req) { + const BatchGetOffloadObjectPushRequest &req, RpcTiming *timing) { auto result = invoke_rpc<&RealClient::batch_get_offload_object_push, - BatchGetOffloadObjectPushResponse>(client_addr, - req); + BatchGetOffloadObjectPushResponse>( + client_addr, timing, req); + if (timing && mooncake::logging::ShouldSampleHiFreqLog(req.trace_id)) { + mooncake::logging::ScopedTraceId trace(req.trace_id); + MC_LOG(INFO) << "offload_rpc_client_breakdown mode=push" + << " endpoint=" << client_addr + << " num_keys=" << req.keys.size() + << " pool_lookup_us=" << timing->pool_lookup_us + << " rpc_call_us=" << timing->rpc_call_us + << " result_get_us=" << timing->result_get_us + << " result_parse_us=" << timing->result_parse_us + << " total_us=" << timing->total_us + << " status=" << (result ? "ok" : "rpc_fail"); + } if (!result) { LOG(ERROR) << "Failed to invoke batch_get_offload_object_push, client_addr = " @@ -6534,10 +6971,22 @@ ClientRequester::batch_get_offload_object_push( } void ClientRequester::release_offload_buffer(const std::string &client_addr, - uint64_t batch_id) { - // Fire-and-forget: attempt to release buffer, log errors but don't block + uint64_t batch_id, + RpcTiming *timing) { + // Synchronous RPC. Failure is non-fatal because GC eventually reclaims it. + const uint64_t trace_id = mooncake::logging::CurrentTraceId(); auto result = invoke_rpc<&RealClient::release_offload_buffer, bool>( - client_addr, batch_id); + client_addr, timing, batch_id, trace_id); + if (timing && mooncake::logging::ShouldSampleHiFreqLog(trace_id)) { + MC_LOG(INFO) << "offload_release_client_breakdown endpoint=" + << client_addr << " batch_id=" << batch_id + << " pool_lookup_us=" << timing->pool_lookup_us + << " rpc_call_us=" << timing->rpc_call_us + << " result_get_us=" << timing->result_get_us + << " result_parse_us=" << timing->result_parse_us + << " total_us=" << timing->total_us + << " status=" << (result ? "ok" : "rpc_fail"); + } if (!result) { // This is expected in some cases (e.g., network issues, buffer already // GC'd) Log at INFO level since GC will eventually clean up anyway @@ -6552,26 +7001,86 @@ void ClientRequester::release_offload_buffer(const std::string &client_addr, template tl::expected ClientRequester::invoke_rpc( - const std::string &client_addr, Args &&...args) { + const std::string &client_addr, RpcTiming *timing, Args &&...args) { + const auto total_start = std::chrono::steady_clock::now(); + UbDiag::PerfPoint pt_pool(PerfKey::GET_SSD_OFFLOAD_RPC_POOL, + UbDiag::PerfLevel::DEBUG); + pt_pool.Start(); auto client_pool = client_pools_->at(client_addr); - return async_simple::coro::syncAwait( + pt_pool.End(0); + if (timing) { + timing->pool_lookup_us = + std::chrono::duration_cast( + std::chrono::steady_clock::now() - total_start) + .count(); + } + auto rpc_result = async_simple::coro::syncAwait( [&]() -> async_simple::coro::Lazy> { + const auto rpc_call_start = std::chrono::steady_clock::now(); + UbDiag::PerfPoint pt_rpc_call( + PerfKey::GET_SSD_OFFLOAD_RPC_CALL, UbDiag::PerfLevel::DEBUG); + pt_rpc_call.Start(); auto ret = co_await client_pool->send_request( [&](coro_io::client_reuse_hint, coro_rpc::coro_rpc_client &client) { return client.send_request( std::forward(args)...); }); + pt_rpc_call.End(ret.has_value() ? 0 : -1); + if (timing) { + timing->rpc_call_us = + std::chrono::duration_cast( + std::chrono::steady_clock::now() - rpc_call_start) + .count(); + } if (!ret.has_value()) { LOG(ERROR) << "Dummy Client not available"; co_return tl::make_unexpected(ErrorCode::RPC_FAIL); } + const auto result_get_start = std::chrono::steady_clock::now(); + UbDiag::PerfPoint pt_result_get( + PerfKey::GET_SSD_OFFLOAD_RPC_RESULT_GET, + UbDiag::PerfLevel::MODULE); + pt_result_get.Start(); auto result = co_await std::move(ret.value()); + pt_result_get.End(result ? 0 : -1); + if (timing) { + timing->result_get_us = + std::chrono::duration_cast( + std::chrono::steady_clock::now() - result_get_start) + .count(); + } if (!result) { LOG(ERROR) << "RPC call failed: " << result.error().msg; co_return tl::make_unexpected(ErrorCode::RPC_FAIL); } - co_return result->result(); + const auto parse_start = std::chrono::steady_clock::now(); + if constexpr (std::is_void_v) { + result->result(); + if (timing) { + timing->result_parse_us = + std::chrono::duration_cast( + std::chrono::steady_clock::now() - parse_start) + .count(); + } + co_return tl::expected{}; + } else { + auto response = std::move(result->result()); + if (timing) { + timing->result_parse_us = + std::chrono::duration_cast( + std::chrono::steady_clock::now() - parse_start) + .count(); + } + co_return std::move(response); + } }()); + if (timing) { + timing->total_us = + std::chrono::duration_cast( + std::chrono::steady_clock::now() - total_start) + .count(); + } + return rpc_result; } } // namespace mooncake diff --git a/mooncake-store/src/real_client_main.cpp b/mooncake-store/src/real_client_main.cpp index 10cd3a71c7..677a0dafd0 100644 --- a/mooncake-store/src/real_client_main.cpp +++ b/mooncake-store/src/real_client_main.cpp @@ -1,4 +1,5 @@ #include +#include #include #include @@ -25,6 +26,10 @@ DEFINE_bool(start_offload_rpc_server, true, "(batch_get_offload_object / release_offload_buffer). " "Effective only when --enable_offload is true. " "Disable for a write-only owner."); +DEFINE_int32(offload_rpc_thread_num, 8, + "Number of threads for the offload RPC server. " + "Effective only when --enable_offload and " + "--start_offload_rpc_server are true."); DECLARE_bool(enable_http_server); DECLARE_int32(http_port); @@ -115,11 +120,18 @@ int main(int argc, char *argv[]) { #endif auto client_inst = RealClient::create(); + const size_t offload_rpc_thread_num = + static_cast(std::max(1, FLAGS_offload_rpc_thread_num)); + if (FLAGS_offload_rpc_thread_num <= 0) { + LOG(WARNING) << "Invalid --offload_rpc_thread_num=" + << FLAGS_offload_rpc_thread_num << ", using 1"; + } auto res = client_inst->setup_internal( FLAGS_host, FLAGS_metadata_server, global_segment_size, 0, FLAGS_protocol, FLAGS_device_names, FLAGS_master_server_address, nullptr, "@mooncake_client_" + std::to_string(FLAGS_port) + ".sock", - FLAGS_port, FLAGS_enable_offload, FLAGS_start_offload_rpc_server); + FLAGS_port, FLAGS_enable_offload, FLAGS_start_offload_rpc_server, "", + "default", offload_rpc_thread_num); if (!res) { LOG(FATAL) << "Failed to setup client: " << toString(res.error()); return -1; diff --git a/mooncake-store/src/rpc_service.cpp b/mooncake-store/src/rpc_service.cpp index 21fe227f48..9d700f35a0 100644 --- a/mooncake-store/src/rpc_service.cpp +++ b/mooncake-store/src/rpc_service.cpp @@ -1012,7 +1012,8 @@ WrappedMasterService::GetReplicaList(const std::string& key, trace_scope_ = std::make_unique(client_trace_id); } - const bool sample = mooncake::logging::ShouldSampleHiFreqLog(); + const bool sample = + mooncake::logging::ShouldSampleHiFreqLog(client_trace_id); const auto t0 = sample ? std::chrono::steady_clock::now() : std::chrono::steady_clock::time_point{}; auto result = execute_rpc( @@ -1031,6 +1032,8 @@ WrappedMasterService::GetReplicaList(const std::string& key, MC_LOG(INFO) << "GetReplicaList host=" << ResolveClientHost(master_service_, client_id) << " key=" << key << " tenant=" << tenant_id + << " replica_count=" + << (result ? result->replicas.size() : 0) << " latency_us=" << latency_us << " status=" << (result.has_value() ? "ok" : toString(result.error())); } @@ -1047,7 +1050,8 @@ WrappedMasterService::BatchGetReplicaList(const std::vector& keys, trace_scope_ = std::make_unique(client_trace_id); } - const bool sample = mooncake::logging::ShouldSampleHiFreqLog(); + const bool sample = + mooncake::logging::ShouldSampleHiFreqLog(client_trace_id); const auto t0 = sample ? std::chrono::steady_clock::now() : std::chrono::steady_clock::time_point{}; UbDiag::PerfPoint pt(PerfKey::MASTER_RPC_BATCH_GET_REPLICA, @@ -1062,8 +1066,24 @@ WrappedMasterService::BatchGetReplicaList(const std::vector& keys, std::vector> results; results.reserve(keys.size()); - for (const auto& key : keys) { - results.emplace_back(master_service_.GetReplicaList(key, tenant_id)); + for (size_t i = 0; i < keys.size(); ++i) { + const auto item_start = sample ? std::chrono::steady_clock::now() + : std::chrono::steady_clock::time_point{}; + results.emplace_back( + master_service_.GetReplicaList(keys[i], tenant_id)); + if (sample) { + const auto item_us = + std::chrono::duration_cast( + std::chrono::steady_clock::now() - item_start) + .count(); + const auto& item = results.back(); + MC_LOG(INFO) + << "BatchGetReplicaListItem key=" << keys[i] + << " batch_index=" << i << " replica_count=" + << (item ? item->replicas.size() : 0) + << " latency_us=" << item_us << " status=" + << (item ? "ok" : toString(item.error())); + } } size_t failure_count = 0; @@ -1120,7 +1140,8 @@ WrappedMasterService::PutStart(const UUID& client_id, const std::string& key, trace_scope_ = std::make_unique(client_trace_id); } - const bool sample = mooncake::logging::ShouldSampleHiFreqLog(); + const bool sample = + mooncake::logging::ShouldSampleHiFreqLog(client_trace_id); const auto t0 = sample ? std::chrono::steady_clock::now() : std::chrono::steady_clock::time_point{}; auto result = execute_rpc( @@ -1158,7 +1179,8 @@ tl::expected WrappedMasterService::PutEnd( trace_scope_ = std::make_unique(client_trace_id); } - const bool sample = mooncake::logging::ShouldSampleHiFreqLog(); + const bool sample = + mooncake::logging::ShouldSampleHiFreqLog(client_trace_id); const auto t0 = sample ? std::chrono::steady_clock::now() : std::chrono::steady_clock::time_point{}; auto result = execute_rpc( @@ -2079,6 +2101,16 @@ tl::expected WrappedMasterService::NotifyOffloadSuccess( return result; } +tl::expected, ErrorCode> +WrappedMasterService::GetOffloadEndpoints() { + ScopedVLogTimer timer(1, "GetOffloadEndpoints"); + timer.LogRequest("action=get_offload_endpoints"); + + auto result = master_service_.GetOffloadEndpoints(); + timer.LogResponseExpected(result); + return result; +} + tl::expected, ErrorCode> WrappedMasterService::PromotionObjectHeartbeat(const UUID& client_id) { ScopedVLogTimer timer(1, "PromotionObjectHeartbeat"); @@ -2244,6 +2276,9 @@ void RegisterRpcService( server.register_handler< &mooncake::WrappedMasterService::NotifyOffloadSuccess>( &wrapped_master_service); + server.register_handler< + &mooncake::WrappedMasterService::GetOffloadEndpoints>( + &wrapped_master_service); server.register_handler< &mooncake::WrappedMasterService::PromotionObjectHeartbeat>( &wrapped_master_service); diff --git a/mooncake-store/src/storage/distributed/distributed_storage_backend.cpp b/mooncake-store/src/storage/distributed/distributed_storage_backend.cpp index 7c8e6ffbb1..cb8c3609f4 100644 --- a/mooncake-store/src/storage/distributed/distributed_storage_backend.cpp +++ b/mooncake-store/src/storage/distributed/distributed_storage_backend.cpp @@ -3,6 +3,7 @@ #include #include +#include #include #include #include @@ -189,6 +190,8 @@ tl::expected DistributedStorageBackend::BatchOffload( tl::expected DistributedStorageBackend::BatchLoad( std::unordered_map& batched_slices) { + auto* stats = CurrentStorageReadStats(); + if (stats) stats->io_mode = "distributed"; if (!initialized_) { LOG(ERROR) << "DistributedStorageBackend is not initialized"; return tl::make_unexpected(ErrorCode::INTERNAL_ERROR); @@ -197,11 +200,33 @@ tl::expected DistributedStorageBackend::BatchLoad( for (auto& [key, slice] : batched_slices) { auto path = GetObjectPath(key); + const auto read_start = std::chrono::steady_clock::now(); auto result = fs_adapter_->ReadFile(path, slice.ptr, slice.size); + if (stats) { + const auto read_us = + std::chrono::duration_cast( + std::chrono::steady_clock::now() - read_start) + .count(); + stats->disk_read_us += read_us; + if (read_us > stats->slowest_disk_read_us) { + stats->slowest_disk_read_us = read_us; + stats->slowest_key = key; + } + } if (!result) { + if (stats) { + stats->status = "read_fail"; + stats->error_key = key; + stats->error_code = result.error(); + } return tl::make_unexpected(result.error()); } if (*result != slice.size) { + if (stats) { + stats->status = "short_read"; + stats->error_key = key; + stats->error_code = ErrorCode::FILE_READ_FAIL; + } return tl::make_unexpected(ErrorCode::FILE_READ_FAIL); } } diff --git a/mooncake-store/src/storage_backend.cpp b/mooncake-store/src/storage_backend.cpp index 93c5150cf8..7675c6638b 100644 --- a/mooncake-store/src/storage_backend.cpp +++ b/mooncake-store/src/storage_backend.cpp @@ -30,6 +30,23 @@ namespace mooncake { +namespace { +thread_local StorageReadStats* current_storage_read_stats = nullptr; +} + +StorageReadStats* CurrentStorageReadStats() { + return current_storage_read_stats; +} + +ScopedStorageReadStats::ScopedStorageReadStats(StorageReadStats* stats) + : previous_(current_storage_read_stats) { + current_storage_read_stats = stats; +} + +ScopedStorageReadStats::~ScopedStorageReadStats() { + current_storage_read_stats = previous_; +} + bool FilePerKeyConfig::Validate() const { if (fsdir.empty()) { LOG(ERROR) << "FilePerKeyConfig: fsdir is invalid"; @@ -1083,7 +1100,10 @@ tl::expected StorageBackendAdaptor::IsExist( tl::expected StorageBackendAdaptor::BatchLoad( std::unordered_map& batched_slices) { + auto* stats = CurrentStorageReadStats(); + if (stats) stats->io_mode = "preadv"; for (const auto& [key, slice] : batched_slices) { + const auto io_start = std::chrono::steady_clock::now(); KVEntry kv; kv.key = key; auto path = @@ -1097,6 +1117,11 @@ tl::expected StorageBackendAdaptor::BatchLoad( auto r = storage_backend_->LoadObject(path, kv_buf, kv_buf.size()); if (!r) { + if (stats) { + stats->status = "read_fail"; + stats->error_key = key; + stats->error_code = r.error(); + } LOG(ERROR) << "Failed to load from file"; return tl::make_unexpected(r.error()); } @@ -1106,6 +1131,17 @@ tl::expected StorageBackendAdaptor::BatchLoad( if (!kv.value.empty()) { std::memcpy(slice.ptr, kv.value.data(), kv.value.size()); } + if (stats) { + const auto io_us = + std::chrono::duration_cast( + std::chrono::steady_clock::now() - io_start) + .count(); + stats->disk_read_us += io_us; + if (io_us > stats->slowest_disk_read_us) { + stats->slowest_disk_read_us = io_us; + stats->slowest_key = key; + } + } } return {}; } @@ -1381,6 +1417,8 @@ tl::expected BucketStorageBackend::BatchQuery( tl::expected BucketStorageBackend::BatchLoad( std::unordered_map& batch_object) { + auto* stats = CurrentStorageReadStats(); + const auto plan_start = std::chrono::steady_clock::now(); // Step 1: Build read plan by copying metadata under lock // BucketReadGuard increments inflight_reads_ to prevent deletion during IO. // When the guard goes out of scope, it decrements the counter. @@ -1410,6 +1448,15 @@ tl::expected BucketStorageBackend::BatchLoad( // Lookup key -> metadata auto object_it = object_bucket_map_.find(key); if (object_it == object_bucket_map_.end()) { + if (stats) { + stats->plan_us = + std::chrono::duration_cast( + std::chrono::steady_clock::now() - plan_start) + .count(); + stats->status = "plan_fail"; + stats->error_key = key; + stats->error_code = ErrorCode::INVALID_KEY; + } LOG(ERROR) << "Key not found: " << key; return tl::make_unexpected(ErrorCode::INVALID_KEY); } @@ -1418,6 +1465,15 @@ tl::expected BucketStorageBackend::BatchLoad( // Lookup bucket -> BucketMetadata auto bucket_it = buckets_.find(metadata.bucket_id); if (bucket_it == buckets_.end()) { + if (stats) { + stats->plan_us = + std::chrono::duration_cast( + std::chrono::steady_clock::now() - plan_start) + .count(); + stats->status = "plan_fail"; + stats->error_key = key; + stats->error_code = ErrorCode::BUCKET_NOT_FOUND; + } LOG(ERROR) << "Bucket not found for key: " << key << ", bucket_id=" << metadata.bucket_id; return tl::make_unexpected(ErrorCode::BUCKET_NOT_FOUND); @@ -1425,6 +1481,15 @@ tl::expected BucketStorageBackend::BatchLoad( // Validate size if (metadata.data_size != static_cast(dest_slice.size)) { + if (stats) { + stats->plan_us = + std::chrono::duration_cast( + std::chrono::steady_clock::now() - plan_start) + .count(); + stats->status = "plan_fail"; + stats->error_key = key; + stats->error_code = ErrorCode::INVALID_PARAMS; + } LOG(ERROR) << "Size mismatch for key: " << key << ", expected: " << metadata.data_size << ", got: " << dest_slice.size; @@ -1455,6 +1520,12 @@ tl::expected BucketStorageBackend::BatchLoad( } } pt_plan.End(0); + if (stats) { + stats->plan_us = + std::chrono::duration_cast( + std::chrono::steady_clock::now() - plan_start) + .count(); + } // Lock released here - bucket files protected by BucketReadGuards // which remain alive until this function returns (~line bucket_guards // destructor), keeping inflight_reads_ > 0 throughout the I/O phase. @@ -1462,15 +1533,34 @@ tl::expected BucketStorageBackend::BatchLoad( // Step 2: Perform IO without holding any locks for (auto& [bucket_id, read_plans] : bucket_read_plans) { // Open file for this bucket (cheap syscall, no lock needed) + const auto open_start = std::chrono::steady_clock::now(); auto filepath_res = GetBucketDataPath(bucket_id); if (!filepath_res) { + if (stats) { + stats->file_open_us += + std::chrono::duration_cast( + std::chrono::steady_clock::now() - open_start) + .count(); + stats->status = "open_fail"; + stats->error_code = ErrorCode::INTERNAL_ERROR; + } LOG(ERROR) << "Failed to get bucket data path, bucket_id=" << bucket_id; return tl::make_unexpected(ErrorCode::INTERNAL_ERROR); } auto file_res = OpenFile(filepath_res.value(), FileMode::Read); + if (stats) { + stats->file_open_us += + std::chrono::duration_cast( + std::chrono::steady_clock::now() - open_start) + .count(); + } if (!file_res) { + if (stats) { + stats->status = "open_fail"; + stats->error_code = file_res.error(); + } LOG(ERROR) << "Failed to open bucket file: " << filepath_res.value(); return tl::make_unexpected(file_res.error()); @@ -1481,11 +1571,14 @@ tl::expected BucketStorageBackend::BatchLoad( for (const auto& plan : read_plans) { int64_t actual_offset = plan.offset + plan.key_size; tl::expected read_res; + const auto disk_start = std::chrono::steady_clock::now(); + const char* io_mode = "preadv"; #ifdef USE_URING // Try to use read_aligned for O_DIRECT I/O if file is UringFile UringFile* uring_file = dynamic_cast(file.get()); if (uring_file != nullptr) { + io_mode = "io_uring_direct"; // Calculate aligned read range int64_t aligned_offset = align_down(actual_offset, kDirectIOAlignment); @@ -1525,8 +1618,29 @@ tl::expected BucketStorageBackend::BatchLoad( read_res = file->vector_read(&iov, 1, actual_offset); pt_posix.End(read_res ? 0 : -1); } + if (stats) { + const auto read_us = + std::chrono::duration_cast( + std::chrono::steady_clock::now() - disk_start) + .count(); + stats->disk_read_us += read_us; + if (read_us > stats->slowest_disk_read_us) { + stats->slowest_disk_read_us = read_us; + stats->slowest_key = plan.key; + } + if (stats->io_mode == "unknown") { + stats->io_mode = io_mode; + } else if (stats->io_mode != io_mode) { + stats->io_mode = "mixed"; + } + } if (!read_res) { + if (stats) { + stats->status = "read_fail"; + stats->error_key = plan.key; + stats->error_code = read_res.error(); + } LOG(ERROR) << "vector_read failed for key: " << plan.key << ", bucket_id=" << plan.bucket_id << ", error: " << read_res.error(); @@ -1534,6 +1648,11 @@ tl::expected BucketStorageBackend::BatchLoad( } if (read_res.value() != plan.dest_slice.size) { + if (stats) { + stats->status = "short_read"; + stats->error_key = plan.key; + stats->error_code = ErrorCode::FILE_READ_FAIL; + } LOG(ERROR) << "Read size mismatch for key: " << plan.key << ", expected: " << plan.dest_slice.size << ", got: " << read_res.value(); @@ -2947,6 +3066,8 @@ tl::expected OffsetAllocatorStorageBackend::BatchOffload( tl::expected OffsetAllocatorStorageBackend::BatchLoad( std::unordered_map& batched_slices) { + auto* stats = CurrentStorageReadStats(); + const auto plan_start = std::chrono::steady_clock::now(); if (!initialized_.load(std::memory_order_acquire)) { LOG(ERROR) << "Storage backend is not initialized. Call Init() before use."; @@ -3000,7 +3121,15 @@ tl::expected OffsetAllocatorStorageBackend::BatchLoad( // Step 2: Perform disk I/O without holding any locks // Allocations are kept alive by shared_ptr references in read_plans + if (stats) { + stats->plan_us = + std::chrono::duration_cast( + std::chrono::steady_clock::now() - plan_start) + .count(); + stats->io_mode = "preadv"; + } for (const auto& plan : read_plans) { + const auto disk_start = std::chrono::steady_clock::now(); // Read header first RecordHeader header; iovec header_iovs[2] = {{&header.key_len, sizeof(header.key_len)}, @@ -3064,6 +3193,17 @@ tl::expected OffsetAllocatorStorageBackend::BatchLoad( << ", got: " << read_value_result.value(); return tl::make_unexpected(ErrorCode::FILE_READ_FAIL); } + if (stats) { + const auto read_us = + std::chrono::duration_cast( + std::chrono::steady_clock::now() - disk_start) + .count(); + stats->disk_read_us += read_us; + if (read_us > stats->slowest_disk_read_us) { + stats->slowest_disk_read_us = read_us; + stats->slowest_key = plan.key; + } + } } // read_plans destructor releases all AllocationPtr references diff --git a/mooncake-transfer-engine/include/transport/kunpeng_transport/ub_context.h b/mooncake-transfer-engine/include/transport/kunpeng_transport/ub_context.h index 9b4acb4bc0..10d5cba337 100644 --- a/mooncake-transfer-engine/include/transport/kunpeng_transport/ub_context.h +++ b/mooncake-transfer-engine/include/transport/kunpeng_transport/ub_context.h @@ -42,6 +42,14 @@ class UbWorkerPool { // Add slices to queue, called by Transport int submitPostSend(const std::vector& slice_list); + uint64_t pendingSliceCount() const { + const uint64_t submitted = + submitted_slice_count_.load(std::memory_order_relaxed); + const uint64_t processed = + processed_slice_count_.load(std::memory_order_relaxed); + return submitted >= processed ? submitted - processed : 0; + } + private: void performPostSend(int thread_id); diff --git a/mooncake-transfer-engine/include/transport/kunpeng_transport/urma/urma_endpoint.h b/mooncake-transfer-engine/include/transport/kunpeng_transport/urma/urma_endpoint.h index 86c8727b8e..846ac640c9 100644 --- a/mooncake-transfer-engine/include/transport/kunpeng_transport/urma/urma_endpoint.h +++ b/mooncake-transfer-engine/include/transport/kunpeng_transport/urma/urma_endpoint.h @@ -82,6 +82,9 @@ class UrmaContext : public UbContext { urma_jfr_t* jfr(); urma_jfce_t* JFCE(); urma_transport_mode_t transMode() const { return trans_mode_; } + uint64_t pendingSliceCount() const { + return worker_pool_ ? worker_pool_->pendingSliceCount() : 0; + } static bool uninit(); static bool init(); diff --git a/mooncake-transfer-engine/src/transport/kunpeng_transport/urma/urma_endpoint.cpp b/mooncake-transfer-engine/src/transport/kunpeng_transport/urma/urma_endpoint.cpp index 66fe750f6a..7c33b276ff 100644 --- a/mooncake-transfer-engine/src/transport/kunpeng_transport/urma/urma_endpoint.cpp +++ b/mooncake-transfer-engine/src/transport/kunpeng_transport/urma/urma_endpoint.cpp @@ -1090,12 +1090,48 @@ int UrmaEndpoint::submitPostSend( std::vector& failed_slice_list) { RWSpinlock::WriteGuard guard(lock_); if (!active_) return 0; + const bool queue_log = + !slice_list.empty() && + mooncake::logging::ShouldSampleHiFreqLog(slice_list.front()->trace_id); int jetty_index = SimpleRandom::Get().next(jetty_list_.size()); + size_t incoming = 0; + int jetty_before = 0; + int jfc_before = 0; + if (queue_log) { + incoming = slice_list.size(); + jetty_before = wr_depth_list_[jetty_index]; + jfc_before = *jfc_outstanding_; + } int wr_count = std::min(max_wr_depth_ - wr_depth_list_[jetty_index], (int)slice_list.size()); wr_count = std::min(int(globalConfig().max_jfc_e) - *jfc_outstanding_, wr_count); - if (wr_count <= 0) return 0; + if (wr_count <= 0) { + if (queue_log) { + const char* status = + jetty_before >= max_wr_depth_ ? "jetty_full" : "jfc_full"; + MC_LOG(INFO) + << "urma_queue_depth direction=" + << (slice_list.front()->opcode == + Transport::TransferRequest::READ + ? "read" + : "write") + << " local_nic=" << context_->nicPath() + << " remote_nic=" << peer_nic_path_ + << " incoming=" << incoming + << " posted=0 software_remaining=" << incoming + << " software_pending=" << context_->pendingSliceCount() + << " jetty_index=" << jetty_index + << " jetty_depth_before=" << jetty_before + << " jetty_depth_after=" << jetty_before + << " jetty_depth_max=" << max_wr_depth_ + << " jfc_outstanding_before=" << jfc_before + << " jfc_outstanding_after=" << jfc_before + << " jfc_depth_max=" << globalConfig().max_jfc_e + << " total_bytes=0 retry_count_max=0 status=" << status; + } + return 0; + } urma_jfs_wr_t wr_list[wr_count], *bad_wr = nullptr; urma_sge_t l_sge_list[wr_count]; @@ -1150,10 +1186,12 @@ int UrmaEndpoint::submitPostSend( } int rc = urma_post_jetty_send_wr(jetty_list_[jetty_index], wr_list, &bad_wr); + int failed_post_count = 0; if (rc) { PLOG(ERROR) << "Failed to urma_post_jetty_send_wr"; while (bad_wr) { int i = bad_wr - wr_list; + ++failed_post_count; LOG(ERROR) << "slice (" << i << ") post send failed."; failed_slice_list.push_back(slice_list[i]); __sync_fetch_and_sub(&wr_depth_list_[jetty_index], 1); @@ -1161,6 +1199,36 @@ int UrmaEndpoint::submitPostSend( bad_wr = bad_wr->next; } } + if (queue_log) { + uint64_t total_bytes = 0; + uint32_t retry_count_max = 0; + for (int i = 0; i < wr_count; ++i) { + total_bytes += slice_list[i]->length; + retry_count_max = + std::max(retry_count_max, slice_list[i]->ub.retry_cnt); + } + MC_LOG(INFO) + << "urma_queue_depth direction=" + << (slice_list.front()->opcode == Transport::TransferRequest::READ + ? "read" + : "write") + << " local_nic=" << context_->nicPath() + << " remote_nic=" << peer_nic_path_ << " incoming=" << incoming + << " posted=" << (wr_count - failed_post_count) + << " software_remaining=" + << (incoming - wr_count + failed_post_count) + << " software_pending=" << context_->pendingSliceCount() + << " jetty_index=" << jetty_index + << " jetty_depth_before=" << jetty_before + << " jetty_depth_after=" << wr_depth_list_[jetty_index] + << " jetty_depth_max=" << max_wr_depth_ + << " jfc_outstanding_before=" << jfc_before + << " jfc_outstanding_after=" << *jfc_outstanding_ + << " jfc_depth_max=" << globalConfig().max_jfc_e + << " total_bytes=" << total_bytes + << " retry_count_max=" << retry_count_max + << " status=" << (rc ? "post_fail" : "posted"); + } slice_list.erase(slice_list.begin(), slice_list.begin() + wr_count); return 0; }