From 1e09b36035ccb21950465e3d30edeb3505473bff Mon Sep 17 00:00:00 2001 From: shifengmin Date: Tue, 7 Jul 2026 20:20:53 +0800 Subject: [PATCH] feat: add request trace dump with runtime toggle. - dump input/output and raw HTTP body to JSONL or per-file. - flush streaming deltas as one record; backfill finish_reason. Co-authored-by: Cursor --- xllm/api_service/api_service.cpp | 15 + xllm/api_service/call.h | 11 + xllm/core/common/global_flags.h | 6 + xllm/core/framework/config/service_config.cpp | 47 +++ xllm/core/framework/config/service_config.h | 11 +- xllm/core/framework/request/CMakeLists.txt | 3 + .../core/framework/request/request_tracer.cpp | 364 ++++++++++++++++++ xllm/core/framework/request/request_tracer.h | 142 +++++++ .../scheduler/async_response_processor.cpp | 6 + 9 files changed, 604 insertions(+), 1 deletion(-) create mode 100644 xllm/core/framework/request/request_tracer.cpp create mode 100644 xllm/core/framework/request/request_tracer.h diff --git a/xllm/api_service/api_service.cpp b/xllm/api_service/api_service.cpp index 3b6f0ad2cf..b6572bbfb5 100644 --- a/xllm/api_service/api_service.cpp +++ b/xllm/api_service/api_service.cpp @@ -29,6 +29,7 @@ limitations under the License. #include "call.h" #include "chat.pb.h" #include "common.pb.h" +#include "common/global_flags.h" #include "completion.pb.h" #include "core/common/constants.h" #include "core/common/metrics.h" @@ -197,6 +198,11 @@ void APIService::CompletionsHttp(::google::protobuf::RpcController* controller, auto ctrl = reinterpret_cast(controller); + std::string raw_body; + if (FLAGS_enable_request_trace) { + raw_body = ctrl->request_attachment().to_string(); + } + auto [preprocess_status, processed_json] = preprocess_completion_prompt(ctrl->request_attachment().to_string()); if (!preprocess_status.ok()) { @@ -218,6 +224,8 @@ void APIService::CompletionsHttp(::google::protobuf::RpcController* controller, std::shared_ptr call = std::make_shared( ctrl, done_guard.release(), req_pb, resp_pb, arena != nullptr); + call->set_raw_request_body(std::move(raw_body)); + call->set_request_endpoint("/v1/completions"); if (completion_service_impl_) { completion_service_impl_->process_async(call); } else if (rec_completion_service_impl_) { @@ -340,6 +348,11 @@ void chat_completions_http_impl(std::unique_ptr& service, std::string attachment; ctrl->request_attachment().copy_to(&attachment, content_len, 0); + std::string raw_body; + if (FLAGS_enable_request_trace) { + raw_body = attachment; + } + auto [preprocess_status, processed_json] = chat_json_parser.preprocess(std::move(attachment)); if (!preprocess_status.ok()) { @@ -361,6 +374,8 @@ void chat_completions_http_impl(std::unique_ptr& service, auto call = std::make_shared( ctrl, guard.release(), req_pb, resp_pb, arena != nullptr /*use_arena*/); + call->set_raw_request_body(std::move(raw_body)); + call->set_request_endpoint("/v1/chat/completions"); service->process_async(call); } diff --git a/xllm/api_service/call.h b/xllm/api_service/call.h index dc934729f6..4aaec70c5a 100644 --- a/xllm/api_service/call.h +++ b/xllm/api_service/call.h @@ -32,6 +32,15 @@ class Call { std::string take_request_payload() { return std::move(request_payload_); } void init_request_payload(); + void set_raw_request_body(std::string body) { + raw_request_body_ = std::move(body); + } + void set_request_endpoint(std::string path) { + request_endpoint_ = std::move(path); + } + const std::string& raw_request_body() const { return raw_request_body_; } + const std::string& request_endpoint() const { return request_endpoint_; } + virtual bool is_disconnected() const = 0; protected: @@ -44,6 +53,8 @@ class Call { std::string x_request_time_; std::string request_payload_; + std::string raw_request_body_; + std::string request_endpoint_; }; } // namespace xllm diff --git a/xllm/core/common/global_flags.h b/xllm/core/common/global_flags.h index a560f85c1c..9729c09e45 100755 --- a/xllm/core/common/global_flags.h +++ b/xllm/core/common/global_flags.h @@ -387,3 +387,9 @@ DECLARE_bool(enable_aclnn_swiglu); DECLARE_bool(use_cpp_chat_template); DECLARE_int32(health_check_interval_ms); + +DECLARE_bool(enable_request_trace); + +DECLARE_string(request_trace_path); + +DECLARE_bool(request_trace_per_file); diff --git a/xllm/core/framework/config/service_config.cpp b/xllm/core/framework/config/service_config.cpp index 8f80d6ff44..4ea33f7018 100644 --- a/xllm/core/framework/config/service_config.cpp +++ b/xllm/core/framework/config/service_config.cpp @@ -55,6 +55,41 @@ DEFINE_int32(health_check_interval_ms, 3000, "Worker health check interval in milliseconds."); +// --- request input/output trace(dump) config --- +DEFINE_bool(enable_request_trace, + false, + "Whether to dump the input and output of each request to a file. " + "For streaming requests, the complete output is saved after all " + "the streaming outputs are finished."); + +DEFINE_string(request_trace_path, + "./request_trace.jsonl", + "The path to dump the request input/output trace. When " + "request_trace_per_file is false, this is a file that JSON " + "records are appended to (one record per line). When " + "request_trace_per_file is true, this is a directory and each " + "request is written to /.json. Only used when " + "enable_request_trace is true."); + +DEFINE_bool(request_trace_per_file, + false, + "Whether to dump each request to a separate file named " + "/.json instead of appending all " + "requests to a single JSON-lines file."); + +namespace { +// brpc's /flags endpoint refuses to modify a gflag at runtime unless it has a +// registered validator (it returns "A reloadable gflag must have validator"). +// Register a no-op validator so enable_request_trace can be toggled without +// restarting the service, as documented for the request trace feature. +bool validate_enable_request_trace(const char* /*unused*/, bool /*unused*/) { + return true; +} +const bool kEnableRequestTraceValidatorDummy = + google::RegisterFlagValidator(&FLAGS_enable_request_trace, + &validate_enable_request_trace); +} // namespace + namespace xllm { void ServiceConfig::from_flags() { @@ -68,6 +103,9 @@ void ServiceConfig::from_flags() { XLLM_CONFIG_ASSIGN_FROM_FLAG(num_request_handling_threads); XLLM_CONFIG_ASSIGN_FROM_FLAG(num_response_handling_threads); XLLM_CONFIG_ASSIGN_FROM_FLAG(health_check_interval_ms); + XLLM_CONFIG_ASSIGN_FROM_FLAG(enable_request_trace); + XLLM_CONFIG_ASSIGN_FROM_FLAG(request_trace_path); + XLLM_CONFIG_ASSIGN_FROM_FLAG(request_trace_per_file); } void ServiceConfig::from_json(const JsonReader& json) { @@ -82,6 +120,9 @@ void ServiceConfig::from_json(const JsonReader& json) { XLLM_CONFIG_ASSIGN_FROM_JSON(num_request_handling_threads); XLLM_CONFIG_ASSIGN_FROM_JSON(num_response_handling_threads); XLLM_CONFIG_ASSIGN_FROM_JSON(health_check_interval_ms); + XLLM_CONFIG_ASSIGN_FROM_JSON(enable_request_trace); + XLLM_CONFIG_ASSIGN_FROM_JSON(request_trace_path); + XLLM_CONFIG_ASSIGN_FROM_JSON(request_trace_per_file); } void ServiceConfig::append_config_json( @@ -106,6 +147,12 @@ void ServiceConfig::append_config_json( config_json, default_config, num_response_handling_threads); APPEND_CONFIG_JSON_VALUE_IF_NOT_DEFAULT( config_json, default_config, health_check_interval_ms); + APPEND_CONFIG_JSON_VALUE_IF_NOT_DEFAULT( + config_json, default_config, enable_request_trace); + APPEND_CONFIG_JSON_VALUE_IF_NOT_DEFAULT( + config_json, default_config, request_trace_path); + APPEND_CONFIG_JSON_VALUE_IF_NOT_DEFAULT( + config_json, default_config, request_trace_per_file); } ServiceConfig& ServiceConfig::get_instance() { diff --git a/xllm/core/framework/config/service_config.h b/xllm/core/framework/config/service_config.h index 0c2882a060..054bbf2508 100644 --- a/xllm/core/framework/config/service_config.h +++ b/xllm/core/framework/config/service_config.h @@ -50,7 +50,10 @@ class ServiceConfig final { "max_concurrent_requests", "num_request_handling_threads", "num_response_handling_threads", - "health_check_interval_ms"}}; + "health_check_interval_ms", + "enable_request_trace", + "request_trace_path", + "request_trace_per_file"}}; return kOptionCategory; } @@ -73,6 +76,12 @@ class ServiceConfig final { PROPERTY(int32_t, num_response_handling_threads) = 4; PROPERTY(int32_t, health_check_interval_ms) = 3000; + + PROPERTY(bool, enable_request_trace) = false; + + PROPERTY(std::string, request_trace_path) = "./request_trace.jsonl"; + + PROPERTY(bool, request_trace_per_file) = false; }; } // namespace xllm diff --git a/xllm/core/framework/request/CMakeLists.txt b/xllm/core/framework/request/CMakeLists.txt index 33d1b09828..7311e4b5bb 100644 --- a/xllm/core/framework/request/CMakeLists.txt +++ b/xllm/core/framework/request/CMakeLists.txt @@ -16,6 +16,7 @@ cc_library( dit_request_output.h dit_request_params.h request_params.h + request_tracer.h sample_slot.h sequence.h sequence_logprob_state.h @@ -34,6 +35,7 @@ cc_library( request_output.cpp dit_request_output.cpp request_params.cpp + request_tracer.cpp sample_slot.cpp dit_request_params.cpp sequence.cpp @@ -55,5 +57,6 @@ cc_library( absl::strings absl::time proto::xllm_proto + nlohmann_json::nlohmann_json torch ) \ No newline at end of file diff --git a/xllm/core/framework/request/request_tracer.cpp b/xllm/core/framework/request/request_tracer.cpp new file mode 100644 index 0000000000..a0e323fce4 --- /dev/null +++ b/xllm/core/framework/request/request_tracer.cpp @@ -0,0 +1,364 @@ +/* Copyright 2025 The xLLM Authors. All Rights Reserved. + +Licensed under the Apache License, Version 2.0 (the "License"); +you may not use this file except in compliance with the License. +You may obtain a copy of the License at + + https://github.com/jd-opensource/xllm/blob/main/LICENSE + +Unless required by applicable law or agreed to in writing, software +distributed under the License is distributed on an "AS IS" BASIS, +WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +See the License for the specific language governing permissions and +limitations under the License. +==============================================================================*/ + +#include "request_tracer.h" + +#include +#include +#include + +#include +#include +#include + +#include "api_service/call.h" +#include "common/global_flags.h" +#include "finish_reason.h" +#include "request.h" + +namespace xllm { + +namespace { + +std::string sanitize_filename(const std::string& name) { + std::string out; + out.reserve(name.size()); + for (char c : name) { + if (std::isalnum(static_cast(c)) || c == '-' || c == '_' || + c == '.') { + out.push_back(c); + } else { + out.push_back('_'); + } + } + if (out.empty()) { + out = "unknown"; + } + return out; +} + +} // namespace + +RequestTracer& RequestTracer::get_instance() { + static RequestTracer instance; + return instance; +} + +bool RequestTracer::enabled() const { return FLAGS_enable_request_trace; } + +RequestTracer::RequestTracer() = default; + +RequestTracer::~RequestTracer() { + { + std::lock_guard lock(queue_mutex_); + stop_ = true; + } + queue_cv_.notify_one(); + if (writer_thread_.joinable()) { + writer_thread_.join(); + } + if (ofs_.is_open()) { + ofs_.flush(); + ofs_.close(); + } +} + +void RequestTracer::ensure_initialized() { + std::call_once(init_flag_, [this] { + per_file_ = FLAGS_request_trace_per_file; + trace_path_ = FLAGS_request_trace_path; + + try { + if (per_file_) { + std::filesystem::create_directories(trace_path_); + } else { + std::filesystem::path file_path(trace_path_); + if (file_path.has_parent_path()) { + std::filesystem::create_directories(file_path.parent_path()); + } + } + } catch (const std::exception& e) { + LOG(ERROR) << "Failed to create directory for request trace: " + << trace_path_ << ", error: " << e.what(); + init_failed_ = true; + return; + } + + if (!per_file_) { + ofs_.open(trace_path_, std::ios::out | std::ios::app); + if (!ofs_.is_open()) { + LOG(ERROR) << "Failed to open request trace file: " << trace_path_; + init_failed_ = true; + return; + } + } + + writer_thread_ = std::thread(&RequestTracer::writer_loop, this); + + LOG(INFO) << "Request input/output trace enabled, dumping to: " + << trace_path_ + << (per_file_ ? " (one file per request)" : " (jsonl)"); + }); +} + +void RequestTracer::enqueue(WriteTask task) { + { + std::lock_guard lock(queue_mutex_); + write_queue_.push_back(std::move(task)); + } + queue_cv_.notify_one(); +} + +void RequestTracer::writer_loop() { + while (true) { + std::deque batch; + { + std::unique_lock lock(queue_mutex_); + queue_cv_.wait(lock, [this] { return stop_ || !write_queue_.empty(); }); + batch.swap(write_queue_); + } + + for (auto& task : batch) { + if (task.file_path.empty()) { + if (ofs_.is_open()) { + ofs_ << task.serialized << "\n"; + } + } else { + std::ofstream file_ofs(task.file_path, std::ios::out | std::ios::trunc); + if (file_ofs.is_open()) { + file_ofs << task.serialized; + } else { + LOG(ERROR) << "Failed to open request trace file: " << task.file_path; + } + } + } + + if (!batch.empty() && ofs_.is_open()) { + ofs_.flush(); + } + + if (stop_) { + std::lock_guard lock(queue_mutex_); + for (auto& task : write_queue_) { + if (task.file_path.empty()) { + if (ofs_.is_open()) { + ofs_ << task.serialized << "\n"; + } + } else { + std::ofstream file_ofs(task.file_path, + std::ios::out | std::ios::trunc); + if (file_ofs.is_open()) { + file_ofs << task.serialized; + } + } + } + write_queue_.clear(); + break; + } + } +} + +std::map RequestTracer::build_seq_records( + const std::vector& outputs) { + std::map records; + for (const auto& seq_output : outputs) { + auto& record = records[seq_output.index]; + record.text = seq_output.text; + if (seq_output.finish_reason.has_value()) { + record.finish_reason = seq_output.finish_reason; + } + record.token_ids = seq_output.token_ids; + } + return records; +} + +void RequestTracer::write_record(Request& request, + const std::map& seq_records, + const std::optional& usage, + bool finished, + bool cancelled) { + ensure_initialized(); + if (init_failed_) { + return; + } + + const auto& state = request.state(); + + nlohmann::json record; + record["request_id"] = request.request_id(); + record["x_request_id"] = request.x_request_id(); + record["x_request_time"] = request.x_request_time(); + record["service_request_id"] = request.service_request_id(); + record["timestamp"] = absl::FormatTime( + "%Y-%m-%dT%H:%M:%E3S%z", absl::Now(), absl::LocalTimeZone()); + record["stream"] = state.stream; + + // input + nlohmann::json input; + input["prompt"] = state.prompt; + input["prompt_token_ids"] = state.prompt_tokens; + input["prompt_tokens_num"] = state.prompt_tokens.size(); + input["max_tokens"] = state.stopping_checker.get_max_generated_tokens(); + input["n"] = state.n; + input["best_of"] = state.best_of; + input["temperature"] = state.sampling_param.temperature; + input["top_p"] = state.sampling_param.top_p; + input["top_k"] = state.sampling_param.top_k; + input["frequency_penalty"] = state.sampling_param.frequency_penalty; + input["presence_penalty"] = state.sampling_param.presence_penalty; + input["repetition_penalty"] = state.sampling_param.repetition_penalty; + record["input"] = std::move(input); + + // output + nlohmann::json output; + nlohmann::json sequences = nlohmann::json::array(); + for (const auto& [index, seq] : seq_records) { + nlohmann::json seq_json; + seq_json["index"] = index; + seq_json["text"] = seq.text; + seq_json["token_ids"] = seq.token_ids; + if (seq.finish_reason.has_value()) { + seq_json["finish_reason"] = seq.finish_reason.value(); + } + sequences.push_back(std::move(seq_json)); + } + output["sequences"] = std::move(sequences); + if (usage.has_value()) { + nlohmann::json usage_json; + usage_json["prompt_tokens"] = usage->num_prompt_tokens; + usage_json["generated_tokens"] = usage->num_generated_tokens; + usage_json["total_tokens"] = usage->num_total_tokens; + output["usage"] = std::move(usage_json); + } + output["finished"] = finished; + output["cancelled"] = cancelled; + record["output"] = std::move(output); + + // raw HTTP request for replay + if (state.call_.has_value() && state.call_.value() != nullptr) { + const auto& raw_body = state.call_.value()->raw_request_body(); + if (!raw_body.empty()) { + nlohmann::json raw_req; + raw_req["body"] = + nlohmann::json::parse(raw_body, nullptr, /*allow_exceptions=*/false); + raw_req["path"] = state.call_.value()->request_endpoint(); + record["raw_request"] = std::move(raw_req); + } + } + + WriteTask task; + if (per_file_) { + task.file_path = (std::filesystem::path(trace_path_) / + (sanitize_filename(request.request_id()) + ".json")) + .string(); + task.serialized = record.dump( + /*indent=*/2, + /*indent_char=*/' ', + /*ensure_ascii=*/false, + nlohmann::json::error_handler_t::replace); + } else { + task.serialized = record.dump( + /*indent=*/-1, + /*indent_char=*/' ', + /*ensure_ascii=*/false, + nlohmann::json::error_handler_t::replace); + } + + enqueue(std::move(task)); +} + +void RequestTracer::trace_completed_request(Request& request, + const RequestOutput& output) { + if (request.state().stream) { + if (enabled()) { + flush_stream_state(request); + } else { + // Tracing may have been disabled at runtime while this streaming request + // was in flight. Drop any accumulated state so it does not leak. + std::lock_guard lock(stream_mutex_); + stream_states_.erase(request.request_id()); + } + return; + } + if (!enabled()) { + return; + } + write_record(request, + build_seq_records(output.outputs), + output.usage, + output.finished, + output.cancelled); +} + +void RequestTracer::trace_stream_output(Request& request, + const RequestOutput& output) { + if (!enabled()) { + return; + } + + // Only accumulate the delta here. The accumulated state is flushed exactly + // once by trace_completed_request (which routes streaming requests through + // flush_stream_state) when the request is finished or cancelled. Flushing + // here based on request.finished()/cancelled() is unsafe under + // enable_schedule_overlap: an earlier stream callback may observe the + // request as already finished because a later scheduling step has advanced + // the sequences, and flush partial state, producing one record per chunk plus + // trailing empty records. + std::lock_guard lock(stream_mutex_); + auto& stream_state = stream_states_[request.request_id()]; + for (const auto& seq_output : output.outputs) { + auto& record = stream_state[seq_output.index]; + record.text += seq_output.text; + record.token_ids.insert(record.token_ids.end(), + seq_output.token_ids.begin(), + seq_output.token_ids.end()); + if (seq_output.finish_reason.has_value()) { + record.finish_reason = seq_output.finish_reason; + } + } +} + +void RequestTracer::flush_stream_state(Request& request) { + std::map seq_records; + { + std::lock_guard lock(stream_mutex_); + auto it = stream_states_.find(request.request_id()); + if (it == stream_states_.end()) { + return; + } + seq_records = std::move(it->second); + stream_states_.erase(it); + } + + // Streaming deltas never carry finish_reason (generate_streaming_output does + // not populate it), so backfill it from each sequence's final finish reason. + const auto& sequences = request.sequences(); + for (auto& [index, record] : seq_records) { + if (!record.finish_reason.has_value() && index < sequences.size()) { + const auto& seq = sequences[index]; + if (seq->finish_reason() != FinishReason::NONE) { + record.finish_reason = seq->finish_reason().to_string(); + } + } + } + + write_record(request, + seq_records, + /*usage=*/std::nullopt, + request.finished(), + request.cancelled()); +} + +} // namespace xllm diff --git a/xllm/core/framework/request/request_tracer.h b/xllm/core/framework/request/request_tracer.h new file mode 100644 index 0000000000..95fff1bb5f --- /dev/null +++ b/xllm/core/framework/request/request_tracer.h @@ -0,0 +1,142 @@ +/* Copyright 2025 The xLLM Authors. All Rights Reserved. + +Licensed under the Apache License, Version 2.0 (the "License"); +you may not use this file except in compliance with the License. +You may obtain a copy of the License at + + https://github.com/jd-opensource/xllm/blob/main/LICENSE + +Unless required by applicable law or agreed to in writing, software +distributed under the License is distributed on an "AS IS" BASIS, +WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +See the License for the specific language governing permissions and +limitations under the License. +==============================================================================*/ + +#pragma once + +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include + +#include "request_output.h" + +namespace xllm { + +class Request; + +// RequestTracer dumps the input and output of each request to a file when +// FLAGS_enable_request_trace is set. The flag is checked on every request so +// tracing can be enabled/disabled at runtime without restarting the service. +// +// Two output layouts are supported (FLAGS_request_trace_per_file): +// - false: append one JSON record per line to FLAGS_request_trace_path. +// - true : write one pretty-printed JSON file per request, named +// "/.json". +// +// For non-streaming requests the complete output is written directly once the +// request is finished. For streaming requests the delta outputs (text and +// token ids) are accumulated per request and the complete output is written +// only after all the streaming outputs are finished (or the request is +// cancelled). +// +// All file I/O is performed asynchronously on a dedicated background thread +// to avoid blocking the response threads. +class RequestTracer { + public: + static RequestTracer& get_instance(); + + // Check the live flag value; callers use this for the fast-path skip. + bool enabled() const; + + // Dump a finished non-streaming request. The given output already contains + // the complete generated text. + void trace_completed_request(Request& request, const RequestOutput& output); + + // Accumulate the delta output of a streaming request. The complete record is + // written out once the request is finished or cancelled. + void trace_stream_output(Request& request, const RequestOutput& output); + + private: + RequestTracer(); + ~RequestTracer(); + RequestTracer(const RequestTracer&) = delete; + RequestTracer& operator=(const RequestTracer&) = delete; + + // Fully assembled output for a single sequence of a request. + struct SeqRecord { + std::string text; + std::optional finish_reason; + // raw generated token id sequence (the original/unparsed sequence). + std::vector token_ids; + }; + + // A unit of work for the background writer thread. + struct WriteTask { + std::string serialized; + std::string file_path; // non-empty only in per-file mode + }; + + // Lazily initialize the writer thread and output file/directory on first + // use (called when the flag transitions from off to on at runtime). + void ensure_initialized(); + + // Write out and erase any accumulated streaming state for the request. Used + // both for the normal streaming finish and as a cleanup path when a streaming + // request is finalized without a final streaming callback (e.g. cancelled). + void flush_stream_state(Request& request); + + // Build the seq records (text/finish_reason/token_ids) from a RequestOutput. + static std::map build_seq_records( + const std::vector& outputs); + + // Serialize and enqueue a record for async writing. + void write_record(Request& request, + const std::map& seq_records, + const std::optional& usage, + bool finished, + bool cancelled); + + // Enqueue a serialized record for async writing. + void enqueue(WriteTask task); + + // Background writer thread entry point. + void writer_loop(); + + std::once_flag init_flag_; + bool init_failed_ = false; + + // when true, write one file per request; otherwise append JSONL. + bool per_file_ = false; + + // the configured output path (file when per_file_ is false, directory when + // per_file_ is true). + std::string trace_path_; + + // Guards stream_states_ only (no longer guards file I/O). + std::mutex stream_mutex_; + + // request_id -> accumulated streaming state (per sequence index). + std::unordered_map> stream_states_; + + // --- Async writer --- + std::thread writer_thread_; + std::mutex queue_mutex_; + std::condition_variable queue_cv_; + std::deque write_queue_; + bool stop_ = false; + + // only used when per_file_ is false; owned by writer thread. + std::ofstream ofs_; +}; + +} // namespace xllm diff --git a/xllm/core/scheduler/async_response_processor.cpp b/xllm/core/scheduler/async_response_processor.cpp index f50fb7d1c0..6f7a74e67f 100644 --- a/xllm/core/scheduler/async_response_processor.cpp +++ b/xllm/core/scheduler/async_response_processor.cpp @@ -26,6 +26,7 @@ limitations under the License. #include "core/framework/config/service_config.h" #include "framework/request/finish_reason.h" #include "framework/request/request.h" +#include "framework/request/request_tracer.h" #include "framework/request/sequence.h" #include "util/blocking_counter.h" #include "util/env_var.h" @@ -110,6 +111,7 @@ void AsyncResponseProcessor::process_completed_request( if (!disable_log_stats_) { request->log_statistic(end_2_end_latency_seconds); } + RequestTracer::get_instance().trace_completed_request(*request, req_output); request->state().output_func(req_output); }; if (request->state().response_thread_id < 0) { @@ -150,6 +152,8 @@ void AsyncResponseProcessor::batch_process_completed_requests( // currently only support one sequence when enable_service_routing request_output->finished_on_prefill_instance = true; } + RequestTracer::get_instance().trace_completed_request(*request, + *request_output); counter->decrement_count(); }; if (request->state().response_thread_id < 0) { @@ -231,6 +235,7 @@ void AsyncResponseProcessor::process_stream_request( req_output.outputs.push_back(std::move(seq_output.value())); } } + RequestTracer::get_instance().trace_stream_output(*request, req_output); if (!request->state().output_func(req_output)) { // cancel the request if on_stream returns false request->set_cancel(); @@ -306,6 +311,7 @@ void AsyncResponseProcessor::batch_process_stream_requests( } } } + RequestTracer::get_instance().trace_stream_output(*request, *req_output); counter->decrement_count(); }; if (request->state().response_thread_id < 0) {