diff --git a/ggml/CMakeLists.txt b/ggml/CMakeLists.txt index c4a8450d1ca..d68f4018709 100644 --- a/ggml/CMakeLists.txt +++ b/ggml/CMakeLists.txt @@ -329,6 +329,7 @@ set(GGML_PUBLIC_HEADERS include/ggml-blas.h include/ggml-cann.h include/ggml-cpp.h + include/ggml-trace.h include/ggml-cuda.h include/ggml-opt.h include/ggml-metal.h diff --git a/ggml/include/ggml-trace.h b/ggml/include/ggml-trace.h new file mode 100644 index 00000000000..834eb282d86 --- /dev/null +++ b/ggml/include/ggml-trace.h @@ -0,0 +1,83 @@ +// Event tracer for the RPC backend and llama.cpp. +// +// The tracer is off unless the environment variable GGML_RPC_TRACE is set to a file path (the +// rpc-server also accepts --trace ). When it is off the only cost at a call site is one +// load and one branch on ggml_trace_flag; no clock is read and nothing is formatted. +// +// Every process writes JSON lines to its own file: one header object, then one object per event. +// scripts/rpc_trace/merge.py aligns the files of the two nodes with the clock offset measured by +// the client at connect time and emits a Chrome trace plus a per decode step summary. + +#pragma once + +#include "ggml.h" +#include "ggml-backend.h" + +#include +#include + +#ifdef __cplusplus +extern "C" { +#endif + + // 1 while a trace file is open. Read it directly at the call sites, do not call a function: + // + // const int64_t t0 = ggml_trace_flag ? ggml_time_us() : 0; + // + GGML_API int ggml_trace_flag; + + // Opens the trace file. `path` may be NULL, then GGML_RPC_TRACE is used. `role` names the + // process in the merged trace ("client", "rpc-server"). Safe to call more than once, the + // first call with a usable path wins. Returns 1 if tracing is on afterwards. + GGML_API int ggml_trace_open(const char * path, const char * role); + GGML_API void ggml_trace_close(void); + + // Monotonic microseconds in the same clock as every timestamp written to the trace. + GGML_API int64_t ggml_trace_time_us(void); + + // Stable small integer for the calling thread, used as the Chrome trace tid. + GGML_API int ggml_trace_tid(void); + + // llama-server tags the threads that drive one pipeline group so that every event raised + // underneath, down to the individual RPC command, carries the group id. -1 means no group. + GGML_API void ggml_trace_set_group(int group); + GGML_API int ggml_trace_get_group(void); + + // Names the tensor or graph the next RPC commands of this thread belong to. The pointer must + // stay valid until it is replaced; the RPC backend passes tensor->name. + GGML_API void ggml_trace_set_subject(const char * name, uint64_t uid); + + // Writes one event. `phase` is a free form category ("rpc.client", "sched", ...), `name` the + // event name. t0/t1 are microseconds from ggml_trace_time_us(); t1 == t0 makes it an instant. + // `fields` is appended verbatim inside the JSON object and may be NULL, for example + // "\"bytes\":128,\"cmd\":\"GRAPH_COMPUTE\"" + GGML_API void ggml_trace_event(const char * phase, const char * name, + int64_t t0, int64_t t1, const char * fields); + + // Same, with printf formatting for the extra fields. + GGML_API void ggml_trace_eventf(const char * phase, const char * name, + int64_t t0, int64_t t1, const char * fmt, ...); + + // Records the result of the four timestamp exchange with the peer into the header of this + // trace file, so the merge tool can put both nodes on one time line. + // t1 client sends, t2 peer receives, t3 peer replies, t4 client receives (microseconds) + GGML_API void ggml_trace_clock_offset(const char * peer, int64_t t1, int64_t t2, int64_t t3, int64_t t4); + + // GPU spans. + // + // A backend submit only queues the work, so host timestamps around it say nothing about when + // the kernels ran. If the backend exposes the timing hooks through its registry (CUDA does), + // these bracket the submit with two events on the compute stream and report them later on the + // host monotonic scale. ggml_trace_gpu_begin returns a tag, 0 if the backend cannot do it. + // Nothing here ever waits on the GPU: the spans are emitted by ggml_trace_gpu_flush once both + // of their events have completed, which the caller does from a point where it is idle anyway. + GGML_API uint64_t ggml_trace_gpu_begin(ggml_backend_t backend, const char * name); + GGML_API void ggml_trace_gpu_end (ggml_backend_t backend, uint64_t tag); + GGML_API void ggml_trace_gpu_flush(void); + + // Escapes a string for the JSON output. Returns `dst`. + GGML_API const char * ggml_trace_escape(char * dst, size_t dst_size, const char * src); + +#ifdef __cplusplus +} +#endif diff --git a/ggml/src/CMakeLists.txt b/ggml/src/CMakeLists.txt index 96535b49fa8..4f53912ae64 100644 --- a/ggml/src/CMakeLists.txt +++ b/ggml/src/CMakeLists.txt @@ -195,12 +195,14 @@ add_library(ggml-base ../include/ggml-backend.h ../include/ggml-cpp.h ../include/ggml-opt.h + ../include/ggml-trace.h ../include/gguf.h ggml.c ggml.cpp ggml-alloc.c ggml-backend.cpp ggml-backend-meta.cpp + ggml-trace.cpp ggml-opt.cpp ggml-threading.cpp ggml-threading.h diff --git a/ggml/src/ggml-backend.cpp b/ggml/src/ggml-backend.cpp index e519bdf50a1..34df336b8f0 100644 --- a/ggml/src/ggml-backend.cpp +++ b/ggml/src/ggml-backend.cpp @@ -10,6 +10,7 @@ #include "ggml-backend.h" #include "ggml-backend-impl.h" +#include "ggml-trace.h" #include "ggml-alloc.h" #include "ggml-impl.h" @@ -491,11 +492,26 @@ void ggml_backend_tensor_copy(const struct ggml_tensor * src, struct ggml_tensor #ifndef NDEBUG GGML_LOG_DEBUG("%s: warning: slow copy from %s to %s\n", __func__, ggml_backend_buffer_name(src->buffer), ggml_backend_buffer_name(dst->buffer)); #endif // NDEBUG + // The staging path taken between two backends that cannot copy to each other directly. + // For a layer split over RPC this is the hidden state going GPU -> host -> peer, so the + // trace breaks it into the host allocation, the read back and the send. size_t nbytes = ggml_nbytes(src); + const int64_t t0 = ggml_trace_flag ? ggml_trace_time_us() : 0; void * data = malloc(nbytes); + const int64_t t1 = ggml_trace_flag ? ggml_trace_time_us() : 0; ggml_backend_tensor_get(src, data, 0, nbytes); + const int64_t t2 = ggml_trace_flag ? ggml_trace_time_us() : 0; ggml_backend_tensor_set(dst, data, 0, nbytes); + const int64_t t3 = ggml_trace_flag ? ggml_trace_time_us() : 0; free(data); + if (ggml_trace_flag) { + ggml_trace_eventf("sched", "copy_stage", t0, ggml_trace_time_us(), + "\"tensor\":\"%s\",\"bytes\":%zu,\"src\":\"%s\",\"dst\":\"%s\"," + "\"malloc_us\":%lld,\"get_us\":%lld,\"set_us\":%lld", + src->name, nbytes, + ggml_backend_buffer_name(src->buffer), ggml_backend_buffer_name(dst->buffer), + (long long) (t1 - t0), (long long) (t2 - t1), (long long) (t3 - t2)); + } } } @@ -1608,6 +1624,9 @@ static enum ggml_status ggml_backend_sched_compute_splits(ggml_backend_sched_t s int split_backend_id = split->backend_id; ggml_backend_t split_backend = sched->backends[split_backend_id]; + const int64_t t_split0 = ggml_trace_flag ? ggml_trace_time_us() : 0; + int64_t t_inputs = t_split0; + // ensure the previous split's async work has completed before we start // this split, the allocator may have reused buffer regions across splits if (split->n_inputs == 0 && prev_backend_id >= 0 && prev_backend_id != split_backend_id) { @@ -1741,6 +1760,12 @@ static enum ggml_status ggml_backend_sched_compute_splits(ggml_backend_sched_t s } } + if (ggml_trace_flag) { t_inputs = ggml_trace_time_us(); } + + // GPU span around the submit of this split, so the trace shows when the kernels really + // ran and not only when the launch returned + const uint64_t gpu_tag = ggml_trace_flag ? ggml_trace_gpu_begin(split_backend, ggml_backend_name(split_backend)) : 0; + if (!sched->callback_eval) { enum ggml_status ec = ggml_backend_graph_compute_async(split_backend, &split->graph); if (ec != GGML_STATUS_SUCCESS) { @@ -1780,11 +1805,26 @@ static enum ggml_status ggml_backend_sched_compute_splits(ggml_backend_sched_t s } } + if (ggml_trace_flag) { + ggml_trace_gpu_end(split_backend, gpu_tag); + } + // record the event of this split if (sched->events[split_backend_id][sched->cur_copy] != NULL) { ggml_backend_event_record(sched->events[split_backend_id][sched->cur_copy], split_backend); } + if (ggml_trace_flag) { + // note: the submit is asynchronous on most backends, so t1 is when the work was + // queued, not when it finished; the GPU rows in the merged trace say that + ggml_trace_eventf("sched", "split", t_split0, ggml_trace_time_us(), + "\"split\":%d,\"n_splits\":%d,\"backend\":\"%s\",\"n_inputs\":%d," + "\"n_nodes\":%d,\"inputs_us\":%lld,\"gpu_tag\":%llu", + split_id, sched->n_splits, ggml_backend_name(split_backend), + split->n_inputs, split->graph.n_nodes, + (long long) (t_inputs - t_split0), (unsigned long long) gpu_tag); + } + prev_backend_id = split_backend_id; } @@ -1980,6 +2020,10 @@ void ggml_backend_sched_synchronize(ggml_backend_sched_t sched) { for (int i = 0; i < sched->n_backends; i++) { ggml_backend_synchronize(sched->backends[i]); } + if (ggml_trace_flag) { + // everything is idle here, so this is where the completed GPU spans are collected + ggml_trace_gpu_flush(); + } if (!sched->is_alloc) { // if the graph is not already allocated, always use copy 0 after a synchronization // this ensures that during generation the same copy is used every time, diff --git a/ggml/src/ggml-cuda/ggml-cuda.cu b/ggml/src/ggml-cuda/ggml-cuda.cu index 2456f7dcc62..4974129e447 100644 --- a/ggml/src/ggml-cuda/ggml-cuda.cu +++ b/ggml/src/ggml-cuda/ggml-cuda.cu @@ -5472,6 +5472,117 @@ static ggml_backend_feature * ggml_backend_cuda_get_features(ggml_backend_reg_t GGML_UNUSED(reg); } +// ----------------------------------------------------------------------------- +// GPU timing marks for the event tracer (see ggml/include/ggml-trace.h) +// +// The RPC server needs to know when the kernels of a graph really ran on the GPU, not when the +// launch returned, so it brackets ggml_backend_graph_compute with two CUDA events recorded on +// the backend's compute stream. The events are resolved later, without blocking, against one +// anchor event whose completion time on the host was measured once, which puts the GPU marks on +// the same monotonic microsecond scale as every other event in the trace. +// +// Reached through ggml_backend_reg_get_proc_address, so the RPC backend does not have to link +// against CUDA. +// ----------------------------------------------------------------------------- + +struct ggml_cuda_trace_mark { + uint64_t tag; + int kind; + cudaEvent_t event; +}; + +struct ggml_cuda_trace_state { + std::mutex mutex; + std::vector pending; + std::vector spare; + cudaEvent_t anchor = nullptr; + int64_t anchor_us = 0; + int device = -1; +}; + +static ggml_cuda_trace_state & ggml_cuda_trace() { + static ggml_cuda_trace_state state; + return state; +} + +// records a mark on the compute stream of `backend`; kind 0 = start of a span, 1 = end +extern "C" GGML_BACKEND_API void ggml_backend_cuda_trace_mark(ggml_backend_t backend, uint64_t tag, int kind); +extern "C" void ggml_backend_cuda_trace_mark(ggml_backend_t backend, uint64_t tag, int kind) { + ggml_backend_cuda_context * cuda_ctx = (ggml_backend_cuda_context *) backend->context; + ggml_cuda_trace_state & st = ggml_cuda_trace(); + + std::lock_guard lock(st.mutex); + + if (st.anchor == nullptr) { + ggml_cuda_set_device(cuda_ctx->device); + if (cudaEventCreate(&st.anchor) != cudaSuccess) { + st.anchor = nullptr; + return; + } + st.device = cuda_ctx->device; + // the host time of the instant the anchor completed on the GPU; every later mark is + // reported as anchor_us + elapsed(anchor, mark) + cudaEventRecord(st.anchor, cuda_ctx->stream()); + cudaEventSynchronize(st.anchor); + st.anchor_us = ggml_time_us(); + } + + // elapsed time is only defined between events of the same device + if (cuda_ctx->device != st.device) { + return; + } + + cudaEvent_t event = nullptr; + if (!st.spare.empty()) { + event = st.spare.back(); + st.spare.pop_back(); + } else { + if (cudaEventCreate(&event) != cudaSuccess) { + return; + } + } + + if (cudaEventRecord(event, cuda_ctx->stream()) != cudaSuccess) { + st.spare.push_back(event); + return; + } + + st.pending.push_back({ tag, kind, event }); +} + +// collects the marks whose events have completed, without waiting for any of them. Returns the +// number written; the caller loops until it gets less than `max`. +extern "C" GGML_BACKEND_API int ggml_backend_cuda_trace_poll(uint64_t * tags, int * kinds, int64_t * t_us, int max); +extern "C" int ggml_backend_cuda_trace_poll(uint64_t * tags, int * kinds, int64_t * t_us, int max) { + ggml_cuda_trace_state & st = ggml_cuda_trace(); + + std::lock_guard lock(st.mutex); + if (st.anchor == nullptr) { + return 0; + } + + int n = 0; + size_t keep = 0; + for (size_t i = 0; i < st.pending.size(); i++) { + ggml_cuda_trace_mark & mark = st.pending[i]; + if (n < max && cudaEventQuery(mark.event) == cudaSuccess) { + float ms = 0.0f; + if (cudaEventElapsedTime(&ms, st.anchor, mark.event) == cudaSuccess) { + tags [n] = mark.tag; + kinds[n] = mark.kind; + t_us [n] = st.anchor_us + (int64_t)(ms * 1000.0f); + n++; + } + st.spare.push_back(mark.event); + } else { + st.pending[keep++] = mark; + } + } + st.pending.resize(keep); + + return n; +} + static void * ggml_backend_cuda_reg_get_proc_address(ggml_backend_reg_t reg, const char * name) { GGML_UNUSED(reg); if (strcmp(name, "ggml_backend_comm_init") == 0) { @@ -5492,6 +5603,12 @@ static void * ggml_backend_cuda_reg_get_proc_address(ggml_backend_reg_t reg, con if (strcmp(name, "ggml_backend_get_features") == 0) { return (void *)ggml_backend_cuda_get_features; } + if (strcmp(name, "ggml_backend_cuda_trace_mark") == 0) { + return (void *)ggml_backend_cuda_trace_mark; + } + if (strcmp(name, "ggml_backend_cuda_trace_poll") == 0) { + return (void *)ggml_backend_cuda_trace_poll; + } return nullptr; } diff --git a/ggml/src/ggml-cuda/vendors/hip.h b/ggml/src/ggml-cuda/vendors/hip.h index 9aa558f3f4c..b9f0646b50f 100644 --- a/ggml/src/ggml-cuda/vendors/hip.h +++ b/ggml/src/ggml-cuda/vendors/hip.h @@ -60,7 +60,10 @@ #define cudaErrorMemoryAllocation hipErrorOutOfMemory #define cudaErrorPeerAccessAlreadyEnabled hipErrorPeerAccessAlreadyEnabled #define cudaErrorPeerAccessNotEnabled hipErrorPeerAccessNotEnabled +#define cudaEventCreate hipEventCreate #define cudaEventCreateWithFlags hipEventCreateWithFlags +#define cudaEventElapsedTime hipEventElapsedTime +#define cudaEventQuery hipEventQuery #define cudaEventDisableTiming hipEventDisableTiming #define cudaEventRecord hipEventRecord #define cudaEventSynchronize hipEventSynchronize diff --git a/ggml/src/ggml-cuda/vendors/musa.h b/ggml/src/ggml-cuda/vendors/musa.h index 6d725c7ec19..ce7732c1968 100644 --- a/ggml/src/ggml-cuda/vendors/musa.h +++ b/ggml/src/ggml-cuda/vendors/musa.h @@ -48,7 +48,10 @@ #define cudaErrorMemoryAllocation musaErrorMemoryAllocation #define cudaErrorPeerAccessAlreadyEnabled musaErrorPeerAccessAlreadyEnabled #define cudaErrorPeerAccessNotEnabled musaErrorPeerAccessNotEnabled +#define cudaEventCreate musaEventCreate #define cudaEventCreateWithFlags musaEventCreateWithFlags +#define cudaEventElapsedTime musaEventElapsedTime +#define cudaEventQuery musaEventQuery #define cudaEventDisableTiming musaEventDisableTiming #define cudaEventRecord musaEventRecord #define cudaEventSynchronize musaEventSynchronize diff --git a/ggml/src/ggml-rpc/ggml-rpc.cpp b/ggml/src/ggml-rpc/ggml-rpc.cpp index ad739590674..8680dcac2c4 100644 --- a/ggml/src/ggml-rpc/ggml-rpc.cpp +++ b/ggml/src/ggml-rpc/ggml-rpc.cpp @@ -3,6 +3,7 @@ #include "ggml-backend-impl.h" #include "ggml-cpp.h" #include "transport.h" +#include "ggml-trace.h" #include #include @@ -72,9 +73,37 @@ enum rpc_cmd { RPC_CMD_DEVICE_COUNT, RPC_CMD_GRAPH_RECOMPUTE, RPC_CMD_MEMSET_TENSOR, + // clock alignment for the event tracer: the client sends it once per connection, right after + // HELLO, and only while tracing is on. The server always answers it. + RPC_CMD_TRACE_SYNC, RPC_CMD_COUNT, }; +static const char * rpc_cmd_name(enum rpc_cmd cmd) { + switch (cmd) { + case RPC_CMD_ALLOC_BUFFER: return "ALLOC_BUFFER"; + case RPC_CMD_GET_ALIGNMENT: return "GET_ALIGNMENT"; + case RPC_CMD_GET_MAX_SIZE: return "GET_MAX_SIZE"; + case RPC_CMD_BUFFER_GET_BASE: return "BUFFER_GET_BASE"; + case RPC_CMD_FREE_BUFFER: return "FREE_BUFFER"; + case RPC_CMD_BUFFER_CLEAR: return "BUFFER_CLEAR"; + case RPC_CMD_SET_TENSOR: return "SET_TENSOR"; + case RPC_CMD_SET_TENSOR_HASH: return "SET_TENSOR_HASH"; + case RPC_CMD_GET_TENSOR: return "GET_TENSOR"; + case RPC_CMD_COPY_TENSOR: return "COPY_TENSOR"; + case RPC_CMD_GRAPH_COMPUTE: return "GRAPH_COMPUTE"; + case RPC_CMD_GET_DEVICE_MEMORY: return "GET_DEVICE_MEMORY"; + case RPC_CMD_INIT_TENSOR: return "INIT_TENSOR"; + case RPC_CMD_GET_ALLOC_SIZE: return "GET_ALLOC_SIZE"; + case RPC_CMD_HELLO: return "HELLO"; + case RPC_CMD_DEVICE_COUNT: return "DEVICE_COUNT"; + case RPC_CMD_GRAPH_RECOMPUTE: return "GRAPH_RECOMPUTE"; + case RPC_CMD_MEMSET_TENSOR: return "MEMSET_TENSOR"; + case RPC_CMD_TRACE_SYNC: return "TRACE_SYNC"; + default: return "UNKNOWN"; + } +} + static_assert(RPC_CMD_HELLO == 14, "RPC_CMD_HELLO must be always 14"); // Try RPC_CMD_SET_TENSOR_HASH first when data size is larger than this threshold @@ -92,6 +121,13 @@ struct rpc_msg_hello_rsp { uint8_t conn_caps[RPC_CONN_CAPS_SIZE]; }; +// the peer's monotonic clock at the moment it received the request (t2) and at the moment it +// replied (t3); with the client's t1 and t4 that is the usual four timestamp offset estimate +struct rpc_msg_trace_sync_rsp { + int64_t t2; + int64_t t3; +}; + struct rpc_msg_device_count_rsp { uint32_t device_count; }; @@ -249,14 +285,51 @@ static uint64_t fnv_hash(const uint8_t * data, size_t len) { return hash; } +// ----------------------------------------------------------------------------- +// rpc-server side of the event tracer +// +// One record per command served: when the serve thread started waiting for it, when its opcode +// and its payload arrived, when the handler ran, and when the reply left. The message helpers +// below fill in the parts they are in a position to see, so the individual command handlers stay +// untouched. +// ----------------------------------------------------------------------------- + +struct rpc_server_trace { + bool active = false; + uint8_t cmd = 0; + int64_t t_wait0 = 0; + int64_t t_recv0 = 0; + int64_t t_recv1 = 0; + int64_t t_exec0 = 0; + int64_t t_exec1 = 0; + int64_t t_send0 = 0; + int64_t t_send1 = 0; + size_t bytes_in = 0; + size_t bytes_out = 0; + uint64_t gpu_tag = 0; + int n_nodes = -1; + int device = -1; +}; + +static thread_local rpc_server_trace tls_srv; + static bool send_msg(socket_ptr sock, const void * msg, size_t msg_size) { + if (ggml_trace_flag && tls_srv.active) { + tls_srv.t_exec1 = ggml_trace_time_us(); + tls_srv.t_send0 = tls_srv.t_exec1; + tls_srv.bytes_out = msg_size + sizeof(uint64_t); + } if (!sock->send_data(&msg_size, sizeof(msg_size))) { return false; } if (!sock->send_data(msg, msg_size)) { return false; } - return sock->flush(); + const bool ok = sock->flush(); + if (ggml_trace_flag && tls_srv.active) { + tls_srv.t_send1 = ggml_trace_time_us(); + } + return ok; } static bool recv_msg(socket_ptr sock, void * msg, size_t msg_size) { @@ -267,7 +340,13 @@ static bool recv_msg(socket_ptr sock, void * msg, size_t msg_size) { if (size != msg_size) { return false; } - return sock->recv_data(msg, msg_size); + const bool ok = sock->recv_data(msg, msg_size); + if (ggml_trace_flag && tls_srv.active) { + tls_srv.t_recv1 = ggml_trace_time_us(); + tls_srv.t_exec0 = tls_srv.t_recv1; + tls_srv.bytes_in = msg_size + sizeof(uint64_t) + 1; + } + return ok; } static bool recv_msg(socket_ptr sock, std::vector & input) { @@ -281,7 +360,13 @@ static bool recv_msg(socket_ptr sock, std::vector & input) { GGML_LOG_ERROR("Failed to allocate input buffer of size %" PRIu64 "\n", size); return false; } - return sock->recv_data(input.data(), size); + const bool ok = sock->recv_data(input.data(), size); + if (ggml_trace_flag && tls_srv.active) { + tls_srv.t_recv1 = ggml_trace_time_us(); + tls_srv.t_exec0 = tls_srv.t_recv1; + tls_srv.bytes_in = size + sizeof(uint64_t) + 1; + } + return ok; } static bool parse_endpoint(const std::string & endpoint, std::string & host, int & port) { @@ -315,9 +400,29 @@ static bool send_rpc_cmd_locked(socket_ptr sock, enum rpc_cmd cmd, const void * return sock->flush(); } +// A command carries 1 byte of opcode and 8 bytes of length on the wire in addition to its +// payload, and a reply carries 8 bytes of length. The trace counts those, so the byte totals add +// up to what the link actually moved. +static const size_t RPC_CMD_HEADER_BYTES = 1 + sizeof(uint64_t); +static const size_t RPC_RSP_HEADER_BYTES = sizeof(uint64_t); + static bool send_rpc_cmd(socket_ptr sock, enum rpc_cmd cmd, const void * input, size_t input_size) { + const int64_t t_enq = ggml_trace_flag ? ggml_trace_time_us() : 0; + std::lock_guard lock(sock->conn.mtx_send); - return send_rpc_cmd_locked(sock, cmd, input, input_size); + + const int64_t t_send0 = ggml_trace_flag ? ggml_trace_time_us() : 0; + const bool status = send_rpc_cmd_locked(sock, cmd, input, input_size); + + if (ggml_trace_flag) { + const int64_t t_send1 = ggml_trace_time_us(); + ggml_trace_eventf("rpc.client", rpc_cmd_name(cmd), t_enq, t_send1, + "\"t_send0\":%lld,\"t_send1\":%lld,\"bytes_out\":%zu,\"bytes_in\":0,\"reply\":0,\"ok\":%d", + (long long) t_send0, (long long) t_send1, + input_size + RPC_CMD_HEADER_BYTES, status ? 1 : 0); + } + + return status; } // Reserves this thread's place in the response order of a connection. The server answers the @@ -350,39 +455,84 @@ struct rpc_response_ticket { // RPC request : | rpc_cmd (1 byte) | request_size (8 bytes) | request_data (request_size bytes) | // RPC response: | response_size (8 bytes) | response_data (response_size bytes) | static bool send_rpc_cmd(socket_ptr sock, enum rpc_cmd cmd, const void * input, size_t input_size, void * output, size_t output_size) { + // t_enq the calling thread reaches the dispatcher + // t_send0 the connection is ours, the first byte goes out + // t_send1 the last byte of the request is flushed + // t_wait our turn in the reply order has come + // t_recv0 the first bytes of the reply are in + // t_recv1 the reply is complete + const int64_t t_enq = ggml_trace_flag ? ggml_trace_time_us() : 0; + int64_t t_send0 = 0, t_send1 = 0, t_wait = 0, t_recv0 = 0, t_recv1 = 0; + + auto trace = [&](int ok) { + if (ggml_trace_flag) { + ggml_trace_eventf("rpc.client", rpc_cmd_name(cmd), t_enq, t_recv1 ? t_recv1 : ggml_trace_time_us(), + "\"t_send0\":%lld,\"t_send1\":%lld,\"t_wait\":%lld,\"t_recv0\":%lld,\"t_recv1\":%lld," + "\"bytes_out\":%zu,\"bytes_in\":%zu,\"reply\":1,\"ok\":%d", + (long long) t_send0, (long long) t_send1, (long long) t_wait, + (long long) t_recv0, (long long) t_recv1, + input_size + RPC_CMD_HEADER_BYTES, output_size + RPC_RSP_HEADER_BYTES, ok); + } + }; + std::unique_ptr ticket; bool failed = false; { std::lock_guard lock(sock->conn.mtx_send); + if (ggml_trace_flag) { t_send0 = ggml_trace_time_us(); } ticket.reset(new rpc_response_ticket(sock->conn)); if (!send_rpc_cmd_locked(sock, cmd, input, input_size)) { // still take our turn, so the ticket is released in order and no later waiter is // woken with a response that is not theirs failed = true; } + if (ggml_trace_flag) { t_send1 = ggml_trace_time_us(); } } if (failed) { ticket->wait(); + trace(0); return false; } // the response is read outside mtx_send, so the other threads can keep submitting ticket->wait(); + if (ggml_trace_flag) { t_wait = ggml_trace_time_us(); } uint64_t out_size; if (!sock->recv_data(&out_size, sizeof(out_size))) { + trace(0); return false; } + if (ggml_trace_flag) { t_recv0 = ggml_trace_time_us(); } if (out_size != output_size) { + trace(0); return false; } if (!sock->recv_data(output, output_size)) { + trace(0); return false; } + if (ggml_trace_flag) { t_recv1 = ggml_trace_time_us(); } + trace(1); return true; } +// Four timestamp clock alignment, once per connection. Written into the trace header of the +// client so the merge tool can put the peer's events on the client's time line. +static void rpc_trace_sync(const std::shared_ptr & sock, const std::string & endpoint) { + rpc_msg_trace_sync_rsp response = {}; + + const int64_t t1 = ggml_trace_time_us(); + if (!send_rpc_cmd(sock, RPC_CMD_TRACE_SYNC, nullptr, 0, &response, sizeof(response))) { + GGML_LOG_ERROR("%s: peer %s does not support the trace clock sync\n", __func__, endpoint.c_str()); + return; + } + const int64_t t4 = ggml_trace_time_us(); + + ggml_trace_clock_offset(endpoint.c_str(), t1, response.t2, response.t3, t4); +} + // RPC client-side implementation // Performs HELLO handshake with transport auto-negotiation. @@ -435,6 +585,10 @@ static std::shared_ptr get_socket(const std::string & endpoint) { if (!negotiate_hello(sock)) { return nullptr; } + ggml_trace_open(nullptr, "rpc-client"); + if (ggml_trace_flag) { + rpc_trace_sync(sock, endpoint); + } LOG_DBG("[%s] connected to %s\n", __func__, endpoint.c_str()); sockets[endpoint] = sock; return sock; @@ -538,6 +692,7 @@ static void ggml_backend_rpc_buffer_memset_tensor( static void ggml_backend_rpc_buffer_set_tensor(ggml_backend_buffer_t buffer, ggml_tensor * tensor, const void * data, size_t offset, size_t size) { ggml_backend_rpc_buffer_context * ctx = (ggml_backend_rpc_buffer_context *)buffer->context; + if (ggml_trace_flag) { ggml_trace_set_subject(tensor->name, 0); } rpc_tensor rpc_tensor = serialize_tensor(tensor); if (size > HASH_THRESHOLD) { rpc_msg_set_tensor_hash_req request; @@ -564,6 +719,7 @@ static void ggml_backend_rpc_buffer_set_tensor(ggml_backend_buffer_t buffer, ggm static void ggml_backend_rpc_buffer_get_tensor(ggml_backend_buffer_t buffer, const ggml_tensor * tensor, void * data, size_t offset, size_t size) { ggml_backend_rpc_buffer_context * ctx = (ggml_backend_rpc_buffer_context *)buffer->context; + if (ggml_trace_flag) { ggml_trace_set_subject(tensor->name, 0); } rpc_msg_get_tensor_req request; request.tensor = serialize_tensor(tensor); request.offset = offset; @@ -785,6 +941,11 @@ static enum ggml_status ggml_backend_rpc_graph_compute(ggml_backend_t backend, g auto sock = get_socket(rpc_ctx->endpoint); + const int64_t t_enq = ggml_trace_flag ? ggml_trace_time_us() : 0; + if (ggml_trace_flag) { + ggml_trace_set_subject(cgraph->nodes[cgraph->n_nodes - 1]->name, cgraph->uid); + } + // The graph stored by RPC_CMD_GRAPH_COMPUTE lives on the server per connection and device, // and one connection is shared by every backend of this endpoint - including the backends of // other llama_contexts. So the uid of the last graph sent has to be tracked per connection, @@ -792,20 +953,40 @@ static enum ggml_status ggml_backend_rpc_graph_compute(ggml_backend_t backend, g // the graph another context stored in between. std::unique_lock lock(sock->conn.mtx_send); + const int64_t t_send0 = ggml_trace_flag ? ggml_trace_time_us() : 0; + auto & last_uid = sock->conn.last_graph_uid[rpc_ctx->device]; if (cgraph->uid != 0 && last_uid == cgraph->uid) { rpc_msg_graph_recompute_req request; request.device = rpc_ctx->device; bool status = send_rpc_cmd_locked(sock, RPC_CMD_GRAPH_RECOMPUTE, &request, sizeof(request)); RPC_STATUS_ASSERT(status); + if (ggml_trace_flag) { + const int64_t t_send1 = ggml_trace_time_us(); + ggml_trace_eventf("rpc.client", "GRAPH_RECOMPUTE", t_enq, t_send1, + "\"t_send0\":%lld,\"t_send1\":%lld,\"bytes_out\":%zu,\"bytes_in\":0," + "\"reply\":0,\"ok\":1,\"n_nodes\":%d,\"dev\":%u,\"serialize_us\":0", + (long long) t_send0, (long long) t_send1, + sizeof(request) + RPC_CMD_HEADER_BYTES, cgraph->n_nodes, rpc_ctx->device); + } return GGML_STATUS_SUCCESS; } last_uid = cgraph->uid; std::vector input; serialize_graph(rpc_ctx->device, cgraph, input); + const int64_t t_ser = ggml_trace_flag ? ggml_trace_time_us() : 0; bool status = send_rpc_cmd_locked(sock, RPC_CMD_GRAPH_COMPUTE, input.data(), input.size()); RPC_STATUS_ASSERT(status); + if (ggml_trace_flag) { + const int64_t t_send1 = ggml_trace_time_us(); + ggml_trace_eventf("rpc.client", "GRAPH_COMPUTE", t_enq, t_send1, + "\"t_send0\":%lld,\"t_send1\":%lld,\"bytes_out\":%zu,\"bytes_in\":0," + "\"reply\":0,\"ok\":1,\"n_nodes\":%d,\"dev\":%u,\"serialize_us\":%lld", + (long long) t_ser, (long long) t_send1, + input.size() + RPC_CMD_HEADER_BYTES, cgraph->n_nodes, rpc_ctx->device, + (long long) (t_ser - t_send0)); + } return GGML_STATUS_SUCCESS; } @@ -1524,7 +1705,15 @@ bool rpc_server::graph_compute(const std::vector & input) { graph->use_counts[hash_pos] = tensor_ptrs.at(id)->use_count; } } + if (ggml_trace_flag) { + tls_srv.n_nodes = (int) n_nodes; + tls_srv.device = (int) device; + tls_srv.gpu_tag = ggml_trace_gpu_begin(backends[device], "GRAPH_COMPUTE"); + } ggml_status status = ggml_backend_graph_compute(backends[device], graph); + if (ggml_trace_flag) { + ggml_trace_gpu_end(backends[device], tls_srv.gpu_tag); + } GGML_ASSERT(status == GGML_STATUS_SUCCESS && "Unsuccessful graph computations are not supported with RPC"); stored_graphs[device].graph = graph; return true; @@ -1540,7 +1729,15 @@ bool rpc_server::graph_recompute(const rpc_msg_graph_recompute_req & request) { } ggml_cgraph * graph = stored_graphs[device].graph; LOG_DBG("[%s] device: %u\n", __func__, device); + if (ggml_trace_flag) { + tls_srv.n_nodes = graph->n_nodes; + tls_srv.device = (int) device; + tls_srv.gpu_tag = ggml_trace_gpu_begin(backends[device], "GRAPH_RECOMPUTE"); + } ggml_status status = ggml_backend_graph_compute(backends[device], graph); + if (ggml_trace_flag) { + ggml_trace_gpu_end(backends[device], tls_srv.gpu_tag); + } GGML_ASSERT(status == GGML_STATUS_SUCCESS && "Unsuccessful graph computations are not supported with RPC"); return true; } @@ -1605,9 +1802,21 @@ static void rpc_serve_client(const std::vector & backends, const // Activate transport upgrade using client's caps sock->update_caps(req.conn_caps); while (true) { + int64_t t_wait0 = 0; + if (ggml_trace_flag) { + tls_srv = rpc_server_trace(); + ggml_trace_gpu_flush(); + t_wait0 = ggml_trace_time_us(); + } if (!sock->recv_data(&cmd, 1)) { break; } + if (ggml_trace_flag) { + tls_srv.active = true; + tls_srv.cmd = cmd; + tls_srv.t_wait0 = t_wait0; + tls_srv.t_recv0 = ggml_trace_time_us(); + } if (cmd >= RPC_CMD_COUNT) { // fail fast if the command is invalid GGML_LOG_ERROR("Unknown command: %d\n", cmd); @@ -1618,6 +1827,18 @@ static void rpc_serve_client(const std::vector & backends, const // HELLO command is handled above return; } + case RPC_CMD_TRACE_SYNC: { + rpc_msg_trace_sync_rsp response = {}; + if (!recv_msg(sock, nullptr, 0)) { + return; + } + response.t2 = ggml_trace_time_us(); + response.t3 = ggml_trace_time_us(); + if (!send_msg(sock, &response, sizeof(response))) { + return; + } + break; + } case RPC_CMD_DEVICE_COUNT: { if (!recv_msg(sock, nullptr, 0)) { return; @@ -1842,6 +2063,27 @@ static void rpc_serve_client(const std::vector & backends, const return; } } + if (ggml_trace_flag && tls_srv.active) { + if (tls_srv.t_exec1 == 0) { + // a command with no reply, GRAPH_COMPUTE and SET_TENSOR are the hot ones + tls_srv.t_exec1 = ggml_trace_time_us(); + } + ggml_trace_eventf("rpc.server", rpc_cmd_name((enum rpc_cmd) cmd), + tls_srv.t_recv0, tls_srv.t_send1 ? tls_srv.t_send1 : tls_srv.t_exec1, + "\"t_wait0\":%lld,\"t_recv0\":%lld,\"t_recv1\":%lld,\"t_exec0\":%lld," + "\"t_exec1\":%lld,\"t_send0\":%lld,\"t_send1\":%lld," + "\"bytes_in\":%zu,\"bytes_out\":%zu,\"n_nodes\":%d,\"dev\":%d,\"gpu_tag\":%llu", + (long long) tls_srv.t_wait0, (long long) tls_srv.t_recv0, + (long long) tls_srv.t_recv1, (long long) tls_srv.t_exec0, + (long long) tls_srv.t_exec1, (long long) tls_srv.t_send0, + (long long) tls_srv.t_send1, + tls_srv.bytes_in, tls_srv.bytes_out, tls_srv.n_nodes, tls_srv.device, + (unsigned long long) tls_srv.gpu_tag); + tls_srv.active = false; + } + } + if (ggml_trace_flag) { + ggml_trace_gpu_flush(); } } diff --git a/ggml/src/ggml-trace.cpp b/ggml/src/ggml-trace.cpp new file mode 100644 index 00000000000..7894761921b --- /dev/null +++ b/ggml/src/ggml-trace.cpp @@ -0,0 +1,420 @@ +#include "ggml-trace.h" +#include "ggml-impl.h" +#include "ggml-backend-impl.h" + +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include + +#ifdef _WIN32 +# include +# define GGML_TRACE_GETPID() _getpid() +#else +# include +# define GGML_TRACE_GETPID() getpid() +#endif + +int ggml_trace_flag = 0; + +namespace { + +struct trace_state { + std::mutex mtx; + FILE * f = nullptr; + std::string role; + int64_t t_open = 0; + int64_t n_unflushed = 0; + + ~trace_state() { + if (f) { + fflush(f); + fclose(f); + f = nullptr; + } + } +}; + +trace_state & state() { + static trace_state s; + return s; +} + +std::atomic g_next_tid{0}; + +thread_local int tls_tid = -1; +thread_local int tls_group = -1; +thread_local const char * tls_subject = nullptr; +thread_local uint64_t tls_uid = 0; +thread_local std::string tls_line; + +// a traced process is usually stopped with a signal at the end of a run, and the tail of the +// stdio buffer would be lost, so the file is flushed every so many lines. At the rate a decode +// step produces events this is a handful of flushes per second. +const int64_t TRACE_FLUSH_EVERY = 128; + +// one line is built in the calling thread and handed to the file under the lock +void emit(const std::string & line) { + trace_state & s = state(); + + std::lock_guard lock(s.mtx); + if (s.f == nullptr) { + return; + } + fwrite(line.data(), 1, line.size(), s.f); + if (++s.n_unflushed >= TRACE_FLUSH_EVERY) { + s.n_unflushed = 0; + fflush(s.f); + } +} + +void append_escaped(std::string & out, const char * src) { + if (src == nullptr) { + return; + } + for (const char * p = src; *p; ++p) { + const unsigned char c = (unsigned char) *p; + switch (c) { + case '"': out += "\\\""; break; + case '\\': out += "\\\\"; break; + case '\n': out += "\\n"; break; + case '\r': out += "\\r"; break; + case '\t': out += "\\t"; break; + default: + if (c < 0x20) { + char buf[8]; + snprintf(buf, sizeof(buf), "\\u%04x", c); + out += buf; + } else { + out += (char) c; + } + } + } +} + +void append_i64(std::string & out, int64_t v) { + char buf[24]; + snprintf(buf, sizeof(buf), "%lld", (long long) v); + out += buf; +} + +} // namespace + +int64_t ggml_trace_time_us(void) { + return ggml_time_us(); +} + +int ggml_trace_tid(void) { + if (tls_tid < 0) { + tls_tid = g_next_tid.fetch_add(1); + } + return tls_tid; +} + +void ggml_trace_set_group(int group) { + tls_group = group; +} + +int ggml_trace_get_group(void) { + return tls_group; +} + +void ggml_trace_set_subject(const char * name, uint64_t uid) { + tls_subject = name; + tls_uid = uid; +} + +const char * ggml_trace_escape(char * dst, size_t dst_size, const char * src) { + if (dst == nullptr || dst_size == 0) { + return dst; + } + std::string tmp; + append_escaped(tmp, src); + const size_t n = tmp.size() < dst_size - 1 ? tmp.size() : dst_size - 1; + memcpy(dst, tmp.data(), n); + dst[n] = '\0'; + return dst; +} + +int ggml_trace_open(const char * path, const char * role) { + trace_state & s = state(); + + std::lock_guard lock(s.mtx); + if (s.f != nullptr) { + return 1; + } + if (path == nullptr || path[0] == '\0') { + path = getenv("GGML_RPC_TRACE"); + } + if (path == nullptr || path[0] == '\0') { + return 0; + } + + s.f = fopen(path, "wb"); + if (s.f == nullptr) { + GGML_LOG_ERROR("%s: cannot open trace file %s\n", __func__, path); + return 0; + } + s.role = role != nullptr ? role : "unknown"; + s.t_open = ggml_time_us(); + + char host[256] = ""; +#ifndef _WIN32 + if (gethostname(host, sizeof(host) - 1) != 0) { + host[0] = '\0'; + } +#endif + + std::string line = "{\"header\":1,\"role\":\""; + append_escaped(line, s.role.c_str()); + line += "\",\"host\":\""; + append_escaped(line, host); + line += "\",\"pid\":"; + append_i64(line, GGML_TRACE_GETPID()); + line += ",\"t_open_us\":"; + append_i64(line, s.t_open); + line += ",\"wall_us\":"; + // CLOCK_REALTIME at the same instant, only a sanity check for the merge tool + { + struct timespec ts; + timespec_get(&ts, TIME_UTC); + append_i64(line, (int64_t) ts.tv_sec * 1000000 + ts.tv_nsec / 1000); + } + line += "}\n"; + + fwrite(line.data(), 1, line.size(), s.f); + fflush(s.f); + + ggml_trace_flag = 1; + + GGML_LOG_INFO("%s: tracing to %s (role %s)\n", __func__, path, s.role.c_str()); + return 1; +} + +void ggml_trace_close(void) { + trace_state & s = state(); + + std::lock_guard lock(s.mtx); + ggml_trace_flag = 0; + if (s.f != nullptr) { + fflush(s.f); + fclose(s.f); + s.f = nullptr; + } +} + +void ggml_trace_clock_offset(const char * peer, int64_t t1, int64_t t2, int64_t t3, int64_t t4) { + if (!ggml_trace_flag) { + return; + } + // NTP style: the peer clock is ahead of ours by offset, the round trip is delay + const int64_t offset = ((t2 - t1) + (t3 - t4)) / 2; + const int64_t delay = (t4 - t1) - (t3 - t2); + + std::string line = "{\"clock_offset\":1,\"peer\":\""; + append_escaped(line, peer); + line += "\",\"t1\":"; append_i64(line, t1); + line += ",\"t2\":"; append_i64(line, t2); + line += ",\"t3\":"; append_i64(line, t3); + line += ",\"t4\":"; append_i64(line, t4); + line += ",\"offset_us\":"; append_i64(line, offset); + line += ",\"delay_us\":"; append_i64(line, delay); + line += "}\n"; + + emit(line); +} + +void ggml_trace_event(const char * phase, const char * name, int64_t t0, int64_t t1, const char * fields) { + if (!ggml_trace_flag) { + return; + } + + std::string & line = tls_line; + line.clear(); + line += "{\"ph\":\""; + append_escaped(line, phase); + line += "\",\"n\":\""; + append_escaped(line, name); + line += "\",\"t0\":"; + append_i64(line, t0); + line += ",\"t1\":"; + append_i64(line, t1); + line += ",\"tid\":"; + append_i64(line, ggml_trace_tid()); + if (tls_group >= 0) { + line += ",\"grp\":"; + append_i64(line, tls_group); + } + if (tls_subject != nullptr) { + line += ",\"subj\":\""; + append_escaped(line, tls_subject); + line += "\""; + } + if (tls_uid != 0) { + line += ",\"uid\":"; + append_i64(line, (int64_t) tls_uid); + } + if (fields != nullptr && fields[0] != '\0') { + line += ","; + line += fields; + } + line += "}\n"; + + emit(line); +} + +void ggml_trace_eventf(const char * phase, const char * name, int64_t t0, int64_t t1, const char * fmt, ...) { + if (!ggml_trace_flag) { + return; + } + + char fields[1024]; + fields[0] = '\0'; + if (fmt != nullptr) { + va_list args; + va_start(args, fmt); + vsnprintf(fields, sizeof(fields), fmt, args); + va_end(args); + } + + ggml_trace_event(phase, name, t0, t1, fields); +} + +// ----------------------------------------------------------------------------- +// GPU spans +// +// The timing hooks are looked up once through the backend registry, so ggml-base does not have +// to link against any GPU runtime. A backend that does not offer them simply has no GPU rows in +// the trace. +// ----------------------------------------------------------------------------- + +typedef void (*ggml_trace_gpu_mark_t)(ggml_backend_t backend, uint64_t tag, int kind); +typedef int (*ggml_trace_gpu_poll_t)(uint64_t * tags, int * kinds, int64_t * t_us, int max); + +namespace { + +struct trace_gpu_state { + std::mutex mutex; + // a scheduler runs over several backends and only some of them offer the hooks, so the answer + // is kept per registry: `probed` is every registry already asked, `supported` those that said + // yes. Handing a backend to the mark function of another backend's registry would be fatal. + std::unordered_set probed; + std::unordered_set supported; + ggml_trace_gpu_mark_t mark = nullptr; + ggml_trace_gpu_poll_t poll = nullptr; + std::unordered_map starts; + std::unordered_map names; + uint64_t next_tag = 1; +}; + +trace_gpu_state & gpu() { + static trace_gpu_state s; + return s; +} + +// true when `backend` can record GPU marks; caller holds gpu().mutex +bool gpu_probe(ggml_backend_t backend) { + trace_gpu_state & st = gpu(); + + ggml_backend_dev_t dev = ggml_backend_get_device(backend); + if (dev == nullptr) { + return false; + } + ggml_backend_reg_t reg = ggml_backend_dev_backend_reg(dev); + if (reg == nullptr) { + return false; + } + if (!st.probed.insert((const void *) reg).second) { + return st.supported.count((const void *) reg) != 0; + } + + ggml_trace_gpu_mark_t mark = (ggml_trace_gpu_mark_t) ggml_backend_reg_get_proc_address(reg, "ggml_backend_cuda_trace_mark"); + ggml_trace_gpu_poll_t poll = (ggml_trace_gpu_poll_t) ggml_backend_reg_get_proc_address(reg, "ggml_backend_cuda_trace_poll"); + if (mark == nullptr || poll == nullptr) { + GGML_LOG_INFO("ggml_trace: %s has no GPU timing hook, its events are host side only\n", + ggml_backend_dev_name(dev)); + return false; + } + st.mark = mark; + st.poll = poll; + st.supported.insert((const void *) reg); + return true; +} + +} // namespace + +uint64_t ggml_trace_gpu_begin(ggml_backend_t backend, const char * name) { + trace_gpu_state & st = gpu(); + + std::lock_guard lock(st.mutex); + if (!gpu_probe(backend)) { + return 0; + } + const uint64_t tag = st.next_tag++; + st.names[tag] = name != nullptr ? name : "gpu"; + st.mark(backend, tag, 0); + return tag; +} + +void ggml_trace_gpu_end(ggml_backend_t backend, uint64_t tag) { + trace_gpu_state & st = gpu(); + + std::lock_guard lock(st.mutex); + if (st.mark == nullptr || tag == 0) { + return; + } + // the tag is non zero only if this backend answered the probe + st.mark(backend, tag, 1); +} + +void ggml_trace_gpu_flush(void) { + trace_gpu_state & st = gpu(); + + std::vector names; + std::vector tags; + std::vector t0s; + std::vector t1s; + { + std::lock_guard lock(st.mutex); + if (st.poll == nullptr) { + return; + } + uint64_t tag_buf[64]; + int kind_buf[64]; + int64_t t_buf[64]; + int n = 0; + do { + n = st.poll(tag_buf, kind_buf, t_buf, 64); + for (int i = 0; i < n; i++) { + if (kind_buf[i] == 0) { + st.starts[tag_buf[i]] = t_buf[i]; + continue; + } + auto it = st.starts.find(tag_buf[i]); + if (it == st.starts.end()) { + continue; + } + auto nit = st.names.find(tag_buf[i]); + names.push_back(nit != st.names.end() ? nit->second : std::string("gpu")); + tags .push_back(tag_buf[i]); + t0s .push_back(it->second); + t1s .push_back(t_buf[i]); + st.starts.erase(it); + if (nit != st.names.end()) { + st.names.erase(nit); + } + } + } while (n == 64); + } + + for (size_t i = 0; i < names.size(); i++) { + ggml_trace_eventf("gpu", names[i].c_str(), t0s[i], t1s[i], + "\"gpu_tag\":%llu", (unsigned long long) tags[i]); + } +} diff --git a/scripts/rpc_trace/cpu_check.sh b/scripts/rpc_trace/cpu_check.sh new file mode 100755 index 00000000000..3baee399d6c --- /dev/null +++ b/scripts/rpc_trace/cpu_check.sh @@ -0,0 +1,89 @@ +#!/bin/bash +# Smoke test for the event tracer with no GPU involved: two local rpc-servers on the CPU backend +# and one llama-server splitting the layers over them, run once with the trace off and once with +# it on. Checks that the generated text is identical either way and that the merge tool accepts +# the result. +# +# scripts/rpc_trace/cpu_check.sh [outdir] +set -u + +BUILD=${1:?usage: cpu_check.sh [outdir]} +MODEL=${2:?usage: cpu_check.sh [outdir]} +OUT=${3:-/tmp/rpc_trace_cpu} + +BIN=$BUILD/bin +P1=${P1:-50111} +P2=${P2:-50112} +PORT=${PORT:-8197} + +mkdir -p "$OUT" +export LD_LIBRARY_PATH=$BIN +export LLAMA_ARG_OFFLINE=1 +export CUDA_VISIBLE_DEVICES= # CPU backend only + +pids=() +cleanup() { for p in "${pids[@]:-}"; do kill -9 "$p" 2>/dev/null; done; } +trap cleanup EXIT + +# cell (PEER_TRACE, if set, is the prefix of the peer trace files) +cell() { + local tag=$1; shift + rm -f "$OUT/$tag.out.txt" + local targs1=() targs2=() + if [ -n "${PEER_TRACE:-}" ]; then targs1=(--trace "$PEER_TRACE.1.jsonl"); targs2=(--trace "$PEER_TRACE.2.jsonl"); fi + "$BIN/ggml-rpc-server" -H 127.0.0.1 -p $P1 -t 4 ${targs1[@]+"${targs1[@]}"} > "$OUT/$tag.rpc1.log" 2>&1 & pids+=($!) + "$BIN/ggml-rpc-server" -H 127.0.0.1 -p $P2 -t 4 ${targs2[@]+"${targs2[@]}"} > "$OUT/$tag.rpc2.log" 2>&1 & pids+=($!) + sleep 3 + + "$BIN/llama-server" -m "$MODEL" -ngl 99 --host 127.0.0.1 --port $PORT --no-webui \ + -c 2048 --parallel 2 --rpc 127.0.0.1:$P1,127.0.0.1:$P2 --device RPC0,RPC1 -sm layer \ + --cache-ram 0 -t 4 > "$OUT/$tag.server.log" 2>&1 & local sp=$! + pids+=($sp) + for i in $(seq 1 300); do grep -q "listening on" "$OUT/$tag.server.log" && break; sleep 1; done + if ! grep -q "listening on" "$OUT/$tag.server.log"; then + echo "$tag: server failed to start"; tail -20 "$OUT/$tag.server.log"; return 1 + fi + + : > "$OUT/$tag.out.txt" + for p in "the capital of France is" "two plus two equals" "the colour of the sky is"; do + curl -s http://127.0.0.1:$PORT/completion -H 'Content-Type: application/json' \ + -d "{\"prompt\":\"$p\",\"n_predict\":32,\"temperature\":0,\"top_k\":1,\"seed\":1}" \ + | python3 -c 'import json,sys; print(json.load(sys.stdin)["content"])' >> "$OUT/$tag.out.txt" + done + + kill -TERM $sp 2>/dev/null + for i in $(seq 1 30); do kill -0 $sp 2>/dev/null || break; sleep 1; done + kill -9 $sp 2>/dev/null + sleep 1 + for p in "${pids[@]}"; do kill -TERM "$p" 2>/dev/null; done + sleep 2 + for p in "${pids[@]}"; do kill -9 "$p" 2>/dev/null; done + pids=() +} + +echo "== trace off" +unset GGML_RPC_TRACE +cell off + +echo "== trace on" +export GGML_RPC_TRACE=$OUT/on.client.jsonl +PEER_TRACE=$OUT/on.peer cell on +unset GGML_RPC_TRACE + +echo +if cmp -s "$OUT/off.out.txt" "$OUT/on.out.txt"; then + echo "output identical with the trace off and on: PASS" +else + echo "output DIFFERS with the trace on: FAIL" + diff "$OUT/off.out.txt" "$OUT/on.out.txt" | head -20 + exit 1 +fi + +for f in "$OUT/on.client.jsonl" "$OUT/on.peer.1.jsonl" "$OUT/on.peer.2.jsonl"; do + if [ ! -s "$f" ]; then echo "missing or empty trace $f: FAIL"; exit 1; fi + echo "$(basename "$f"): $(wc -l < "$f") lines" +done + +python3 "$(dirname "$0")/merge.py" "$OUT/on.client.jsonl" "$OUT/on.peer.1.jsonl" "$OUT/on.peer.2.jsonl" \ + --chrome "$OUT/on.chrome.json" --summary "$OUT/on.summary.txt" || exit 1 +cat "$OUT/on.summary.txt" diff --git a/scripts/rpc_trace/device_idle.py b/scripts/rpc_trace/device_idle.py new file mode 100755 index 00000000000..8a864a678fd --- /dev/null +++ b/scripts/rpc_trace/device_idle.py @@ -0,0 +1,213 @@ +#!/usr/bin/env python3 +"""Where the idle time of ONE device goes, on a two node layer split. + +merge.py answers "where does a decode step go", per pipeline group. This answers a different +question, and it is the one that decides what a fix would have to look like: + + the bottleneck GPU is busy B percent of the window. The other 100-B percent is idle. Is it + idle while the OTHER device is computing, or while NEITHER device is computing? + +Those have different fixes. Idle while the other device computes is a scheduling problem: the +pipeline groups are not offset, or a group is waiting at a synchronisation point, so the work +that should have covered this device was somewhere else. Idle while neither device computes is a +host problem: something on the CPU is between the two devices and nothing can run anywhere. + +The split is reported separately for the prefill phase and the decode phase of the cell, because +a serving cell spends its first seconds prefilling every slot and a device that is idle there is +idle for a completely different reason than one that is idle in steady decode. + +Usage: device_idle.py client.jsonl peer.jsonl [--out report.txt] +""" + +import argparse +import sys +import os +from collections import defaultdict + +sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) +from merge import load, union, union_len, clip, gaps, Index # noqa: E402 + + +def phase_windows(client): + """(prefill, decode) lists of (t0,t1) iteration windows, per group. + + An iteration is prefill when it submitted more tokens than it had slots processing: in decode + every slot contributes exactly one token, so n_tokens <= n_slots. The classification is per + iteration and not per time range, because the groups do not enter decode together. + """ + subs = defaultdict(list) # grp -> [(t0,t1,n_tokens)] + for e in client.events: + if e.get("ph") == "server" and e.get("n") == "submit": + subs[e.get("grp", 0)].append((e["t0"], e["t1"], e.get("n1", 0))) + for g in subs: + subs[g].sort() + + pre, dec = defaultdict(list), defaultdict(list) + for e in client.events: + if e.get("ph") != "server" or e.get("n") != "iteration": + continue + g = e.get("grp", 0) + n_slots = e.get("n1", 0) + toks = [n for a, b, n in subs.get(g, []) if a >= e["t0"] and b <= e["t1"]] + (pre if (toks and max(toks) > max(n_slots, 1)) else dec)[g].append((e["t0"], e["t1"])) + return pre, dec + + +def report(files, client, out): + servers = [f for f in files if f.role == "rpc-server"] + + def shift(f, e): + return (e["t0"] - f.offset_us, e["t1"] - f.offset_us) + + busy = { + "local": union([shift(client, e) for e in client.events if e.get("ph") == "gpu"]), + "peer": union([shift(f, e) for f in servers for e in f.events if e.get("ph") == "gpu"]), + } + if not busy["local"] or not busy["peer"]: + out.write("one of the two devices has no GPU spans; nothing to decompose\n") + return + + pre, dec = phase_windows(client) + groups = sorted(set(list(pre.keys()) + list(dec.keys()))) + + all_iters = [iv for g in groups for iv in pre[g] + dec[g]] + w0 = min(a for a, _ in all_iters) + w1 = max(b for _, b in all_iters) + + # Host phases, for attributing the stretches in which neither device computes. + # server/iteration is the PARENT span of every other server phase, so it is not a candidate: + # it would win every attribution and say nothing. What it does not cover inside an iteration + # is reported as "inside an iteration, untraced", which is a real answer and a different one. + host = {} + for e in client.events: + if e.get("ph") in ("server", "sched", "llama") and e.get("n") != "iteration": + host.setdefault("%s/%s" % (e.get("ph"), e.get("n")), []).append((e["t0"], e["t1"], e)) + host = {k: Index(v) for k, v in host.items()} + iter_ix = Index([(e["t0"], e["t1"], e) for e in client.events + if e.get("ph") == "server" and e.get("n") == "iteration"]) + + # phase boundary: prefill of the cell is everything up to the last prefill iteration + pre_all = union([iv for g in groups for iv in pre[g]]) + dec_all = union([iv for g in groups for iv in dec[g]]) + t_pre_end = max((b for _, b in pre_all), default=w0) + + phases = [("whole window", w0, w1), + ("prefill phase", w0, t_pre_end), + ("decode phase", t_pre_end, w1)] + + out.write("window %.3f s, %d groups, prefill ends %.3f s in " + "(%.1f%% of the window)\n" % ( + (w1 - w0) / 1e6, len(groups), (t_pre_end - w0) / 1e6, + 100.0 * (t_pre_end - w0) / max(w1 - w0, 1))) + out.write("prefill iterations %d, decode iterations %d\n\n" % ( + sum(len(pre[g]) for g in groups), sum(len(dec[g]) for g in groups))) + + for pname, p0, p1 in phases: + span = p1 - p0 + if span <= 0: + continue + out.write("=== %s: %.3f s\n" % (pname, span / 1e6)) + for dev in ("local", "peer"): + other = "peer" if dev == "local" else "local" + b = clip(busy[dev], p0, p1) + ob = Index([(a, c, None) for a, c in busy[other]]) + idle = gaps(b, p0, p1) + t_busy = union_len(b) + t_idle = union_len(idle) + + covered = 0.0 + n_sched, n_host = 0, 0 + len_sched, len_host = [], [] + attr = defaultdict(float) + by_grp = defaultdict(float) + for g0, g1 in idle: + c = ob.covered(g0, g1) + covered += c + if c > 0.5 * (g1 - g0): + n_sched += 1 + len_sched.append(g1 - g0) + else: + n_host += 1 + len_host.append(g1 - g0) + # the stretches of this hole in which the other device is ALSO idle + dead = gaps(clip(busy[other], g0, g1), g0, g1) + for d0, d1 in dead: + best, bestc = None, 0.0 + for name, ix in host.items(): + cv = ix.covered(d0, d1) + if cv > bestc: + bestc, best = cv, name + if best is not None: + attr[best] += bestc + rest = (d1 - d0) - bestc + if rest > 0: + inside = iter_ix.covered(d0, d1) + attr["inside an iteration, untraced"] += min(rest, inside) + attr["between iterations"] += max(rest - inside, 0.0) + for g in groups: + if union_len(clip(dec[g] + pre[g], d0, d1)) > 0.5 * (d1 - d0): + by_grp["group %d live" % g] += d1 - d0 + + out.write(" %-5s busy %6.2f%% idle %6.2f%% " + "(idle while %s computes %6.2f%%, idle with neither computing %6.2f%%)\n" + % (dev, 100.0 * t_busy / span, 100.0 * t_idle / span, other, + 100.0 * covered / span, 100.0 * (t_idle - covered) / span)) + out.write(" %d idle stretches: %d mostly-covered (median %.2f ms), " + "%d mostly-dead (median %.2f ms)\n" + % (len(idle), n_sched, _med(len_sched) / 1000.0, + n_host, _med(len_host) / 1000.0)) + top = sorted(attr.items(), key=lambda kv: -kv[1])[:6] + if top and top[0][1] > 0: + out.write(" neither computing, by host phase: %s\n" + % ", ".join("%s %.2f%%" % (k, 100.0 * v / span) + for k, v in top if v > 0)) + n_steps = sum(len(clip(dec[g] + pre[g], p0, p1)) for g in groups) + if n_steps: + out.write(" per group-step in this phase: idle %.2f ms " + "(%.2f ms while %s computes, %.2f ms with neither)\n" + % (t_idle / n_steps / 1000.0, covered / n_steps / 1000.0, + other, (t_idle - covered) / n_steps / 1000.0)) + out.write("\n") + + # how the two groups sit relative to each other: the overlap of their GPU demand + out.write("=== group offset in the decode phase\n") + for g in groups: + d = clip(dec[g], t_pre_end, w1) + out.write(" group %d: %d decode iterations, median %.1f ms, " + "covering %.1f%% of the decode phase\n" + % (g, len(d), _med([b - a for a, b in d]) / 1000.0, + 100.0 * union_len(d) / max(w1 - t_pre_end, 1))) + if len(groups) == 2: + a = clip(dec[groups[0]], t_pre_end, w1) + b = clip(dec[groups[1]], t_pre_end, w1) + both = union_len(a) + union_len(b) - union_len(a + b) + out.write(" the two groups are inside an iteration at the same time for %.1f%% " + "of the decode phase\n" % (100.0 * both / max(w1 - t_pre_end, 1))) + + +def _med(xs): + if not xs: + return 0.0 + xs = sorted(xs) + return xs[len(xs) // 2] + + +def main(): + ap = argparse.ArgumentParser() + ap.add_argument("traces", nargs="+") + ap.add_argument("--out") + args = ap.parse_args() + files, client = load(args.traces) + out = open(args.out, "w") if args.out else sys.stdout + try: + if client is None: + out.write("no client trace\n") + else: + report(files, client, out) + finally: + if args.out: + out.close() + + +if __name__ == "__main__": + main() diff --git a/scripts/rpc_trace/gpu_trace.sh b/scripts/rpc_trace/gpu_trace.sh new file mode 100755 index 00000000000..27893967a07 --- /dev/null +++ b/scripts/rpc_trace/gpu_trace.sh @@ -0,0 +1,141 @@ +#!/bin/bash +# Traced layer split cells on the Spark pair. Each configuration is run twice, once with the +# tracer off and once with it on, so the overhead of the tracer is measured and not assumed. +set -u +D=/home/nvidianew/temp/wt_trace +O=$D/bench; S=$O/samples +mkdir -p $O $S +BIN=$D/build/bin +PEER=192.168.200.13 +PEERDIR=/home/nvidianew/temp/wt_trace_bin +M=/home/nvidianew/.cache/huggingface/hub/models--unsloth--Qwen3.8-27B-GGUF/snapshots/4ca720788d1e01f1bff70c033e0d0028fd02e502/Qwen3.8-27B-UD-Q4_K_XL.gguf +RPCPORT=50052; PORT=8188; PY=/home/nvidianew/temp/llamacpp_pipe/tvenv/bin/python +LOG=$O/gpu_trace.log +CONC=${CONC:-32}; NTG=${NTG:-256}; NPP=${NPP:-128} +SSH="ssh -o BatchMode=yes -o ConnectTimeout=8 -o ControlMaster=auto -o ControlPath=/tmp/trace_ssh_%h -o ControlPersist=900 nvidianew@$PEER" +say() { echo "[$(date +%H:%M:%S)] $*" | tee -a "$LOG"; } + +thermals() { + local l=$(cat /sys/class/thermal/thermal_zone*/temp | sort -rn | head -1) + local r=$($SSH 'cat /sys/class/thermal/thermal_zone*/temp | sort -rn | head -1') + echo "$l $r" +} +guard() { + while true; do + read a b < <(thermals) + if [ "$a" -gt 80000 ] || [ "$b" -gt 80000 ]; then + say " thermal $a/$b over 80000, waiting 5 min"; sleep 300 + else + say " thermal ok local=$a peer=$b"; return 0 + fi + done +} +memcheck() { + local lf=$(free -g | awk '/^Mem:/{print $7}') + local rf=$($SSH "free -g | awk '/^Mem:/{print \$7}'") + say " free mem local=${lf}G peer=${rf}G" + if [ "$lf" -lt 20 ] || [ "$rf" -lt 20 ]; then say " !!! not enough free memory, abort"; exit 1; fi +} + +export LD_LIBRARY_PATH=$BIN LLAMA_ARG_OFFLINE=1 + +SRVPID=""; PEERPID="" + +# Kill every rpc-server this script started on the peer. By pid, and then by a match scoped to +# our own port, because a server left behind holds the port and the next cell silently talks to +# it instead ("Failed to create server socket" in its log and an empty trace). +stop_peer() { + # note: matched with pgrep -x on the binary name and then on the port in /proc, never with + # pgrep -f or pkill -f, whose pattern also matches the remote shell running it + $SSH "kill -TERM $PEERPID 2>/dev/null; sleep 2; kill -9 $PEERPID 2>/dev/null; + for p in \$(pgrep -x ggml-rpc-server 2>/dev/null); do + if tr '\\0' ' ' < /proc/\$p/cmdline 2>/dev/null | grep -q -- '-p $RPCPORT'; then + kill -9 \$p 2>/dev/null + fi + done; true" 2>/dev/null + PEERPID="" + # do not return until the port is free again + for i in $(seq 1 30); do + timeout 2 bash -c "/dev/null || return 0 + sleep 1 + done + say " !!! peer port $RPCPORT still busy" +} +trap '[ -n "$SRVPID" ] && kill -9 $SRVPID 2>/dev/null; stop_peer' EXIT + +# cell [extra server args...] +cell() { + local tag=$1; local dev=$2; local trace=$3; shift 3 + guard; memcheck + + local ptrace="" + [ "$trace" = 1 ] && ptrace="--trace /tmp/trace_${tag}_peer.jsonl" + stop_peer + # note: no setsid here. $! would then be the pid of setsid and the server, its child, would + # survive every kill. -n so ssh does not hold the terminal open waiting for stdin. + PEERPID=$($SSH -n "cd $PEERDIR && LD_LIBRARY_PATH=$PEERDIR nohup ./ggml-rpc-server -H 0.0.0.0 -p $RPCPORT $ptrace > /tmp/trace_rpc_$tag.log 2>&1 < /dev/null & echo \$!") + for i in $(seq 1 60); do timeout 2 bash -c "/dev/null && break; sleep 1; done + if $SSH -n "grep -q 'Failed to create server socket' /tmp/trace_rpc_$tag.log" 2>/dev/null; then + say " !!! peer rpc-server for $tag could not bind $RPCPORT, aborting"; exit 1 + fi + say " peer rpc-server $PEERPID up ($tag)" + + ( exec nvidia-smi --query-gpu=utilization.gpu,clocks.sm,temperature.gpu,power.draw --format=csv,noheader -lms 100 | while IFS= read -r l; do echo "$(date +%s.%N),$l"; done > "$S/${tag}_local.csv" ) & local SL=$! + ( exec $SSH 'exec nvidia-smi --query-gpu=utilization.gpu,clocks.sm,temperature.gpu,power.draw --format=csv,noheader -lms 100 | while IFS= read -r l; do echo "$(date +%s.%N),$l"; done' > "$S/${tag}_peer.csv" ) & local SP=$! + + local tracenv=() + [ "$trace" = 1 ] && tracenv=(GGML_RPC_TRACE=$O/${tag}_client.jsonl) + + env ${tracenv[@]+"${tracenv[@]}"} $BIN/llama-server -m "$M" -ngl 99 -fa on --host 127.0.0.1 --port $PORT --no-webui --slots \ + -c 16384 --parallel 32 --rpc $PEER:$RPCPORT --device $dev -sm layer --cache-ram 0 -t 6 "$@" \ + > "$O/$tag.server.log" 2>&1 & + SRVPID=$! + for i in $(seq 1 900); do grep -q "listening on" "$O/$tag.server.log" && break; sleep 1; done + grep -q "listening on" "$O/$tag.server.log" || { say " !!! $tag failed to start"; tail -5 "$O/$tag.server.log" | tee -a "$LOG"; } + say " $tag up" + + local t0=$(date +%s.%N) + $PY /home/nvidianew/temp/userscale/bench_users.py --backend http://127.0.0.1:$PORT --label $tag \ + --conc $CONC --npp $NPP --ntg $NTG --reqs-per-client 1 --out "$O/$tag.bench.jsonl" > "$O/$tag.bench.log" 2>&1 + echo "$t0 $(date +%s.%N)" > "$O/$tag.window" + tail -1 "$O/$tag.bench.jsonl" | tee -a "$LOG" + + kill -TERM $SRVPID 2>/dev/null; for i in $(seq 1 120); do kill -0 $SRVPID 2>/dev/null || break; sleep 1; done + kill -9 $SRVPID 2>/dev/null; SRVPID="" + sleep 2 # let the peer flush the tail of its trace before it is stopped + stop_peer + say " peer compute apps after $tag: $($SSH -n 'nvidia-smi --query-compute-apps=pid,used_memory --format=csv,noheader | tr "\n" " "')" + for p in $SL $SP; do for c in $(pgrep -P $p 2>/dev/null); do pkill -P $c 2>/dev/null; kill $c 2>/dev/null; done; kill $p 2>/dev/null; done + + if [ "$trace" = 1 ]; then + scp -q -o ControlPath=/tmp/trace_ssh_%h nvidianew@$PEER:/tmp/trace_${tag}_peer.jsonl "$O/${tag}_peer.jsonl" || say " !!! no peer trace for $tag" + $PY $D/scripts/rpc_trace/merge.py "$O/${tag}_client.jsonl" "$O/${tag}_peer.jsonl" \ + --chrome "$O/${tag}.chrome.json" --summary "$O/${tag}.summary.txt" 2>>"$LOG" + say " --- $tag summary"; cat "$O/${tag}.summary.txt" | tee -a "$LOG" + fi + + $PY - "$O/$tag.window" "$S/${tag}_local.csv" "$S/${tag}_peer.csv" <<'PYEOF' | tee -a "$LOG" +import sys +t0,t1=[float(x) for x in open(sys.argv[1]).read().split()] +for name,path in (("local",sys.argv[2]),("peer",sys.argv[3])): + u=[];c=[];t=[] + for line in open(path): + p=line.strip().split(",") + if len(p)<5: continue + try: ts=float(p[0]) + except: continue + if tst1: continue + u.append(float(p[1].split()[0])); c.append(float(p[2].split()[0])); t.append(float(p[3])) + if u: print(" %s util=%.1f%% clocks.sm=%.0fMHz tmax=%.0fC n=%d"%(name,sum(u)/len(u),sum(c)/len(c),max(t),len(u))) + else: print(" %s no samples"%name) +PYEOF +} + +cell n1_cr_off CUDA0,RPC0 0 +cell n1_cr_on CUDA0,RPC0 1 +cell n1_rc_off RPC0,CUDA0 0 +cell n1_rc_on RPC0,CUDA0 1 +cell n2_rc_off RPC0,CUDA0 0 --pipeline-groups 2 +cell n2_rc_on RPC0,CUDA0 1 --pipeline-groups 2 + +say "gpu_trace done" diff --git a/scripts/rpc_trace/merge.py b/scripts/rpc_trace/merge.py new file mode 100644 index 00000000000..f0964010073 --- /dev/null +++ b/scripts/rpc_trace/merge.py @@ -0,0 +1,456 @@ +#!/usr/bin/env python3 +"""Merge the event traces of the nodes of an RPC layer split. + +Each process writes JSON lines (see ggml/include/ggml-trace.h): one header, then one object per +event. The client also writes the result of a four timestamp exchange with every peer it connects +to, which gives the offset between the two monotonic clocks. This tool puts every file on the +client's time line and produces + + * a Chrome trace (chrome://tracing, or https://ui.perfetto.dev) with one row per node, thread + and pipeline group, plus a row per GPU carrying the CUDA event timings, and + * a text summary per decode step: local compute, transfer, peer compute, logits return, + sampling, and the idle fraction of each GPU. + +Usage: + merge.py client.jsonl peer.jsonl --chrome trace.json --summary summary.txt +""" + +import argparse +import bisect +import json +import os +import sys +from collections import defaultdict + +# ---------------------------------------------------------------------------- reading + + +class TraceFile: + def __init__(self, path): + self.path = path + self.header = {} + self.offsets = [] # clock offset records written by the client + self.events = [] + self.offset_us = 0 # this file's clock minus the client's clock + + with open(path, "r", errors="replace") as f: + for line in f: + line = line.strip() + if not line or not line.startswith("{"): + continue + try: + rec = json.loads(line) + except json.JSONDecodeError: + # a trace of a process that was killed can end in a partial line + continue + if "header" in rec: + self.header = rec + elif "clock_offset" in rec: + self.offsets.append(rec) + elif "ph" in rec and "t0" in rec and "t1" in rec: + self.events.append(rec) + + @property + def role(self): + return self.header.get("role", "unknown") + + @property + def host(self): + return self.header.get("host", "") + + def label(self): + return "%s %s" % (self.host or "node", self.role) + + +def load(paths): + files = [TraceFile(p) for p in paths] + + clients = [f for f in files if f.role in ("llama-server", "rpc-client")] + servers = [f for f in files if f.role == "rpc-server"] + + if not clients: + # a trace of the peer alone is still useful, it just has no common time line + return files, None + + client = clients[0] + + # map each peer endpoint to its measured offset; the exchange is repeated per connection, the + # median is used so a single delayed reply does not move the alignment + by_host = defaultdict(list) + for rec in client.offsets: + host = rec.get("peer", "").split(":")[0] + by_host[host].append(rec.get("offset_us", 0)) + + for f in servers: + cand = None + for host, values in by_host.items(): + if host == f.host or f.host.startswith(host) or host.startswith(f.host): + cand = values + break + if cand is None and len(by_host) == 1: + cand = list(by_host.values())[0] + if cand is None: + sys.stderr.write( + "warning: no clock offset for %s, its events are left on their own clock\n" % f.path) + continue + cand = sorted(cand) + f.offset_us = cand[len(cand) // 2] + + return files, client + + +# ---------------------------------------------------------------------------- intervals + + +def union_len(intervals): + """total length covered by a list of (t0, t1)""" + if not intervals: + return 0 + intervals = sorted(intervals) + total = 0 + cur0, cur1 = intervals[0] + for t0, t1 in intervals[1:]: + if t0 > cur1: + total += cur1 - cur0 + cur0, cur1 = t0, t1 + else: + cur1 = max(cur1, t1) + total += cur1 - cur0 + return total + + +def union(intervals): + if not intervals: + return [] + intervals = sorted(intervals) + out = [list(intervals[0])] + for t0, t1 in intervals[1:]: + if t0 > out[-1][1]: + out.append([t0, t1]) + else: + out[-1][1] = max(out[-1][1], t1) + return [(a, b) for a, b in out] + + +def clip(intervals, w0, w1): + out = [] + for t0, t1 in intervals: + a, b = max(t0, w0), min(t1, w1) + if b > a: + out.append((a, b)) + return out + + +def gaps(intervals, w0, w1): + """the holes of a union inside [w0, w1]""" + out = [] + cur = w0 + for t0, t1 in union(clip(intervals, w0, w1)): + if t0 > cur: + out.append((cur, t0)) + cur = max(cur, t1) + if cur < w1: + out.append((cur, w1)) + return out + + +# ---------------------------------------------------------------------------- chrome trace + + +def chrome_trace(files, client): + out = [] + t_base = None + for f in files: + for e in f.events: + t = e["t0"] - f.offset_us + t_base = t if t_base is None else min(t_base, t) + if t_base is None: + t_base = 0 + + for pid, f in enumerate(files, start=1): + out.append({"ph": "M", "pid": pid, "tid": 0, "name": "process_name", + "args": {"name": f.label()}}) + out.append({"ph": "M", "pid": pid, "tid": 0, "name": "process_sort_index", + "args": {"sort_index": pid}}) + + gpu_rows = {} + named = set() + + for e in f.events: + t0 = e["t0"] - f.offset_us - t_base + t1 = e["t1"] - f.offset_us - t_base + cat = e.get("ph", "") + tid = e.get("tid", 0) + + if cat == "gpu": + # one row per device, well away from the host thread ids + key = e.get("n", "gpu") + if key not in gpu_rows: + gpu_rows[key] = 10000 + len(gpu_rows) + out.append({"ph": "M", "pid": pid, "tid": gpu_rows[key], "name": "thread_name", + "args": {"name": "GPU %s" % key}}) + tid = gpu_rows[key] + elif tid not in named: + named.add(tid) + label = "thread %d" % tid + if e.get("grp") is not None: + label = "group %d thread %d" % (e["grp"], tid) + out.append({"ph": "M", "pid": pid, "tid": tid, "name": "thread_name", + "args": {"name": label}}) + + args = {k: v for k, v in e.items() if k not in ("ph", "n", "t0", "t1", "tid")} + out.append({"ph": "X", "pid": pid, "tid": tid, "cat": cat, "name": e.get("n", "?"), + "ts": t0, "dur": max(t1 - t0, 0), "args": args}) + + # the phases inside one RPC command, as slices nested in the command + for name, a, b in sub_phases(e): + a -= f.offset_us + t_base + b -= f.offset_us + t_base + if b > a: + out.append({"ph": "X", "pid": pid, "tid": tid, "cat": cat + ".phase", + "name": name, "ts": a, "dur": b - a}) + + # nested slices must not start before their parent + out.sort(key=lambda e: (e.get("ts", -1), -e.get("dur", 0))) + return {"traceEvents": out, "displayTimeUnit": "ms"} + + +def sub_phases(e): + cat = e.get("ph", "") + if cat == "rpc.client": + t_send0 = e.get("t_send0", 0) + t_send1 = e.get("t_send1", 0) + res = [] + if t_send0: + res.append(("queue", e["t0"], t_send0)) + res.append(("send", t_send0, t_send1)) + if e.get("reply"): + res.append(("wait reply", t_send1, e.get("t_wait", t_send1))) + res.append(("read reply", e.get("t_wait", t_send1), e.get("t_recv1", t_send1))) + return [r for r in res if r[2] > r[1]] + if cat == "rpc.server": + res = [("receive", e.get("t_recv0", 0), e.get("t_recv1", 0)), + ("execute", e.get("t_exec0", 0), e.get("t_exec1", 0)), + ("reply", e.get("t_send0", 0), e.get("t_send1", 0))] + return [r for r in res if r[1] and r[2] > r[1]] + return [] + + +# ---------------------------------------------------------------------------- summary + +LOGITS_MIN_BYTES = 64 * 1024 + + +class Index: + """intervals sorted by start, with a bisect lookup of the ones overlapping a window""" + + def __init__(self, items): + # items: list of (t0, t1, payload) + self.items = sorted(items, key=lambda x: x[0]) + self.starts = [x[0] for x in self.items] + self.max_dur = max((x[1] - x[0] for x in self.items), default=0) + + def overlapping(self, w0, w1): + lo = bisect.bisect_left(self.starts, w0 - self.max_dur) + out = [] + for t0, t1, payload in self.items[lo:]: + if t0 >= w1: + break + if t1 > w0: + out.append((t0, t1, payload)) + return out + + def covered(self, w0, w1): + return union_len(clip([(a, b) for a, b, _ in self.overlapping(w0, w1)], w0, w1)) + + +def summarize(files, client, out): + servers = [f for f in files if f.role == "rpc-server"] + + def shift(f, e): + return (e["t0"] - f.offset_us, e["t1"] - f.offset_us) + + # GPU busy intervals of each node, on the client's clock + gpu_local = Index([shift(client, e) + (e,) for e in client.events if e.get("ph") == "gpu"]) + gpu_peer = Index([shift(f, e) + (e,) for f in servers for e in f.events if e.get("ph") == "gpu"]) + + iters = [e for e in client.events if e.get("ph") == "server" and e.get("n") == "iteration"] + if not iters: + out.write("no llama-server iterations in the trace\n") + return + + w0 = min(e["t0"] for e in iters) + w1 = max(e["t1"] for e in iters) + span = max(w1 - w0, 1) + + out.write("trace window %.3f s, %d decode steps\n" % (span / 1e6, len(iters))) + for f in files: + out.write(" %-28s %-14s offset %+.3f ms, %d events\n" + % (os.path.basename(f.path), f.label(), f.offset_us / 1000.0, len(f.events))) + out.write("\n") + + groups = sorted({e.get("grp", 0) for e in iters}) + + # one index per (category, name, group), so a step is a bisect and not a scan of the file + idx = {} + for e in client.events: + key = (e.get("ph"), e.get("n"), e.get("grp", 0)) + idx.setdefault(key, []).append((e["t0"], e["t1"], e)) + for key in list(idx): + idx[key] = Index(idx[key]) + + empty = Index([]) + + def get(cat, name, grp): + return idx.get((cat, name, grp), empty) + + # the RPC commands of one group, all command types together + rpc_by_grp = {} + for grp in groups: + rpc_by_grp[grp] = Index([(e["t0"], e["t1"], e) for e in client.events + if e.get("ph") == "rpc.client" and e.get("grp", 0) == grp]) + + rows = [] + for grp in groups: + steps = [e for e in iters if e.get("grp", 0) == grp] + acc = defaultdict(float) + n = 0 + wire_out = 0 + wire_in = 0 + worst = (0, "", 0) + idle_by = defaultdict(float) + host = [get(cat, name, grp) for cat, name in + (("server", "batch_build"), ("server", "submit"), ("server", "synchronize"), + ("server", "post_decode"), ("server", "sampling"), ("server", "result_send"), + ("llama", "graph_compute"), ("sched", "split"), ("sched", "copy_stage"))] + + for it in steps: + t0, t1 = it["t0"], it["t1"] + if t1 <= t0: + continue + n += 1 + + cmds = [e for _, _, e in rpc_by_grp[grp].overlapping(t0, t1)] + + send = [(e.get("t_send0", e["t0"]), e.get("t_send1", e["t1"])) for e in cmds] + recv = [(e.get("t_wait", 0), e.get("t_recv1", 0)) for e in cmds if e.get("reply")] + recv = [r for r in recv if r[0] and r[1] > r[0]] + logits = [(e["t0"], e.get("t_recv1", e["t1"])) for e in cmds + if e.get("n") == "GET_TENSOR" and e.get("bytes_in", 0) >= LOGITS_MIN_BYTES] + + wire_out += sum(e.get("bytes_out", 0) for e in cmds) + wire_in += sum(e.get("bytes_in", 0) for e in cmds) + + local_iv = clip([(a, b) for a, b, _ in gpu_local.overlapping(t0, t1)], t0, t1) + peer_iv = clip([(a, b) for a, b, _ in gpu_peer.overlapping(t0, t1)], t0, t1) + + acc["step"] += t1 - t0 + acc["build"] += get("server", "batch_build", grp).covered(t0, t1) + acc["submit"] += get("server", "submit", grp).covered(t0, t1) + acc["sync"] += get("server", "synchronize", grp).covered(t0, t1) + acc["post"] += get("server", "post_decode", grp).covered(t0, t1) + acc["sampling"] += get("server", "sampling", grp).covered(t0, t1) + acc["send"] += get("server", "result_send", grp).covered(t0, t1) + acc["local_gpu"] += union_len(local_iv) + acc["peer_gpu"] += union_len(peer_iv) + acc["transfer"] += union_len(clip(send + recv, t0, t1)) + acc["logits"] += union_len(clip(logits, t0, t1)) + acc["stage"] += get("sched", "copy_stage", grp).covered(t0, t1) + + # the stretches of the step in which neither GPU was busy, and what the host was + # doing in them: the single longest one, and the total attributed per phase + hole = gaps(local_iv + peer_iv, t0, t1) + acc["idle_both"] += union_len(hole) + for g0, g1 in hole: + name, cov = name_gap(host, g0, g1) + if g1 - g0 > worst[0]: + worst = (g1 - g0, "%s (%.0f%% of the gap)" % (name, 100.0 * cov / max(g1 - g0, 1)), g0) + idle_by[name] += cov + idle_by["unattributed"] += (g1 - g0) - cov + + if n == 0: + continue + rows.append((grp, n, acc, wire_out, wire_in, worst, idle_by)) + + hdr = ("group steps step_ms build submit sync post sampl send | " + "localGPU peerGPU transfer logits stage | idle_both") + out.write(hdr + "\n") + out.write("-" * len(hdr) + "\n") + for grp, n, acc, wo, wi, worst, idle_by in rows: + def ms(k): + return acc[k] / n / 1000.0 + out.write("%5d %5d %7.1f %6.1f %7.1f %6.1f %6.1f %6.1f %6.1f | " + "%8.1f %8.1f %9.1f %7.1f %6.1f | %9.1f\n" + % (grp, n, ms("step"), ms("build"), ms("submit"), ms("sync"), ms("post"), + ms("sampling"), ms("send"), ms("local_gpu"), ms("peer_gpu"), + ms("transfer"), ms("logits"), ms("stage"), ms("idle_both"))) + out.write("\n") + + for grp, n, acc, wo, wi, worst, idle_by in rows: + out.write("group %d: %.1f kB out and %.1f kB in per step over RPC; " + "biggest idle gap %.1f ms in %s\n" + % (grp, wo / n / 1024.0, wi / n / 1024.0, worst[0] / 1000.0, worst[1])) + top = sorted(idle_by.items(), key=lambda kv: -kv[1])[:4] + out.write(" idle with neither GPU busy, per step: %s\n" + % ", ".join("%s %.1f ms" % (k, v / n / 1000.0) for k, v in top if v > 0)) + + busy_local = gpu_local.covered(w0, w1) + busy_peer = gpu_peer.covered(w0, w1) + out.write("\nover the whole window: local GPU busy %.1f%% (idle %.1f%%), " + "peer GPU busy %.1f%% (idle %.1f%%)\n" + % (100.0 * busy_local / span, 100.0 * (1 - busy_local / span), + 100.0 * busy_peer / span, 100.0 * (1 - busy_peer / span))) + if not gpu_local.items: + out.write("note: no GPU spans on the client, so the local GPU row is empty " + "(CPU backend, or a build without the CUDA timing hook)\n") + if not gpu_peer.items: + out.write("note: no GPU spans from the peer, so the peer GPU row is empty\n") + + +def name_gap(indexes, g0, g1): + """what the host was doing during a stretch in which no GPU was busy""" + best = None + best_cov = 0 + for ix in indexes: + for a, b, e in ix.overlapping(g0, g1): + cov = min(b, g1) - max(a, g0) + if cov > best_cov: + best_cov = cov + best = e + if best is None: + return "nothing traced", 0 + return "%s/%s" % (best.get("ph"), best.get("n")), best_cov + + +# ---------------------------------------------------------------------------- main + + +def main(): + ap = argparse.ArgumentParser(description=__doc__, + formatter_class=argparse.RawDescriptionHelpFormatter) + ap.add_argument("traces", nargs="+", help="trace files, client first") + ap.add_argument("--chrome", help="write a Chrome trace here") + ap.add_argument("--summary", help="write the text summary here (default: stdout)") + args = ap.parse_args() + + files, client = load(args.traces) + + if args.chrome: + with open(args.chrome, "w") as f: + json.dump(chrome_trace(files, client), f) + sys.stderr.write("wrote %s (%d events)\n" + % (args.chrome, sum(len(t.events) for t in files))) + + out = open(args.summary, "w") if args.summary else sys.stdout + try: + if client is None: + out.write("no client trace given, nothing to align against\n") + else: + summarize(files, client, out) + finally: + if args.summary: + out.close() + + +if __name__ == "__main__": + main() diff --git a/scripts/rpc_trace/nonrpc_bracket.sh b/scripts/rpc_trace/nonrpc_bracket.sh new file mode 100755 index 00000000000..f4926d61747 --- /dev/null +++ b/scripts/rpc_trace/nonrpc_bracket.sh @@ -0,0 +1,41 @@ +#!/bin/bash +# Single GPU, no RPC backend involved: llama-batched-bench base/new/base plus one traced pass, +# and a greedy md5 with the trace off and on, to show that a workload that does not use RPC is +# not moved by the tracer. +set -u +W=/home/nvidianew/temp/wt_trace; B0=/home/nvidianew/temp/wt_base/build/bin; B1=$W/build/bin +O=$W/bench; mkdir -p $O +M=/home/nvidianew/.cache/huggingface/hub/models--unsloth--Qwen3.8-27B-GGUF/snapshots/4ca720788d1e01f1bff70c033e0d0028fd02e502/Qwen3.8-27B-UD-Q4_K_XL.gguf +say(){ echo "[$(date +%H:%M:%S)] $*" | tee -a $O/bracket.log; } + +for pass in base_1 new base_2 new_traced; do + case $pass in base_*) BIN=$B0;; new*) BIN=$B1;; esac + tenv=() + [ "$pass" = new_traced ] && tenv=(GGML_RPC_TRACE=$O/nonrpc_trace.jsonl) + say "=== batched-bench $pass ($BIN)" + env ${tenv[@]+"${tenv[@]}"} LD_LIBRARY_PATH=$BIN LLAMA_ARG_OFFLINE=1 \ + $BIN/llama-batched-bench -m "$M" -c 32768 -npp 512 -ntg 128 -npl 1,8,32 \ + -ngl 99 -fa on -t 6 > $O/bb_$pass.log 2>&1 + grep -E "^\|" $O/bb_$pass.log | tail -4 | tee -a $O/bracket.log +done + +for pass in base new new_traced; do + case $pass in base) BIN=$B0;; new*) BIN=$B1;; esac + tenv=() + [ "$pass" = new_traced ] && tenv=(GGML_RPC_TRACE=$O/nonrpc_greedy.jsonl) + PORT=8194 + env ${tenv[@]+"${tenv[@]}"} LD_LIBRARY_PATH=$BIN LLAMA_ARG_OFFLINE=1 \ + $BIN/llama-server -m "$M" -ngl 99 -fa on --host 127.0.0.1 --port $PORT \ + --no-webui -c 8192 --parallel 8 --cache-ram 0 -t 6 > $O/greedy_$pass.srv.log 2>&1 & + sp=$! + for i in $(seq 1 600); do grep -q "listening on" $O/greedy_$pass.srv.log && break; sleep 1; done + : > $O/greedy_$pass.txt + for p in "The capital of France is" "Explain gravity in one sentence." "def fibonacci(n):" "List three primes:" "Once upon a time"; do + curl -s http://127.0.0.1:$PORT/completion -H 'Content-Type: application/json' \ + -d "$(python3 -c "import json,sys; print(json.dumps({'prompt':sys.argv[1],'n_predict':48,'temperature':0,'top_k':1,'seed':1234,'cache_prompt':False}))" "$p")" \ + | python3 -c "import sys,json; print(json.load(sys.stdin)['content'])" >> $O/greedy_$pass.txt + done + kill -TERM $sp 2>/dev/null; for i in $(seq 1 60); do kill -0 $sp 2>/dev/null || break; sleep 1; done; kill -9 $sp 2>/dev/null + say "single GPU greedy $pass: $(md5sum < $O/greedy_$pass.txt)" +done +say "bracket done" diff --git a/src/llama-context.cpp b/src/llama-context.cpp index 0402044da6b..8a69a86fdfc 100644 --- a/src/llama-context.cpp +++ b/src/llama-context.cpp @@ -1,5 +1,7 @@ #include "llama-context.h" +#include "ggml-trace.h" + #include "ggml.h" #include "llama-arch.h" #include "llama-graph.h" @@ -702,11 +704,33 @@ void llama_context::sched_reserve() { __func__, (t_end_us - t_start_us)/1000.0, ggml_backend_sched_get_n_copies(sched.get())); } +// RAII span for the event tracer, one branch on ggml_trace_flag when tracing is off +struct llama_trace_scope { + const char * name; + int64_t t0; + int n0; + int n1; + + llama_trace_scope(const char * name, int n0, int n1) : + name(name), t0(ggml_trace_flag ? ggml_trace_time_us() : 0), n0(n0), n1(n1) {} + + ~llama_trace_scope() { + if (ggml_trace_flag) { + ggml_trace_eventf("llama", name, t0, ggml_trace_time_us(), "\"n0\":%d,\"n1\":%d", n0, n1); + } + } + + llama_trace_scope(const llama_trace_scope &) = delete; + llama_trace_scope & operator=(const llama_trace_scope &) = delete; +}; + void llama_context::synchronize() { if (!sched) { return; } + llama_trace_scope span("synchronize", (int) n_queued_tokens, 0); + ggml_backend_sched_synchronize(sched.get()); // FIXME: if multiple single tokens are evaluated without a synchronization, @@ -1647,6 +1671,8 @@ int llama_context::decode(const llama_batch & batch_inp) { return -1; } + llama_trace_scope span_decode("decode", batch_inp.n_tokens, 0); + const auto & vocab = model.vocab; const auto & hparams = model.hparams; @@ -2491,11 +2517,20 @@ ggml_status llama_context::graph_compute( set_n_threads_fn.second(set_n_threads_fn.first, n_threads); } + const int64_t t0 = ggml_trace_flag ? ggml_trace_time_us() : 0; + auto status = ggml_backend_sched_graph_compute_async(sched.get(), gf); if (status != GGML_STATUS_SUCCESS) { LLAMA_LOG_ERROR("%s: ggml_backend_sched_graph_compute_async failed with error %d\n", __func__, status); } + if (ggml_trace_flag) { + // the individual splits are traced by the scheduler itself, this is the whole submit + ggml_trace_eventf("llama", "graph_compute", t0, ggml_trace_time_us(), + "\"n_splits\":%d,\"n_nodes\":%d,\"batched\":%d", + ggml_backend_sched_get_n_splits(sched.get()), ggml_graph_n_nodes(gf), batched ? 1 : 0); + } + // fprintf(stderr, "splits: %d\n", ggml_backend_sched_get_n_splits(sched)); return status; diff --git a/tools/rpc/rpc-server.cpp b/tools/rpc/rpc-server.cpp index 08e68039141..fbf2cf5552b 100644 --- a/tools/rpc/rpc-server.cpp +++ b/tools/rpc/rpc-server.cpp @@ -1,4 +1,5 @@ #include "ggml-rpc.h" +#include "ggml-trace.h" #ifdef _WIN32 # define NOMINMAX # define DIRECTORY_SEPARATOR '\\' @@ -175,6 +176,7 @@ struct rpc_server_params { bool use_cache = false; int n_threads = std::max(1U, std::thread::hardware_concurrency()/2); std::vector devices; + std::string trace; }; static void print_usage(int /*argc*/, char ** argv, rpc_server_params params) { @@ -186,6 +188,7 @@ static void print_usage(int /*argc*/, char ** argv, rpc_server_params params) { fprintf(stderr, " -H, --host HOST host to bind to (default: %s)\n", params.host.c_str()); fprintf(stderr, " -p, --port PORT port to bind to (default: %d)\n", params.port); fprintf(stderr, " -c, --cache enable local file cache\n"); + fprintf(stderr, " --trace FILE write an event trace to FILE (same as GGML_RPC_TRACE)\n"); fprintf(stderr, "\n"); } @@ -231,6 +234,11 @@ static bool rpc_server_params_parse(int argc, char ** argv, rpc_server_params & if (params.port <= 0 || params.port > 65535) { return false; } + } else if (arg == "--trace") { + if (++i >= argc) { + return false; + } + params.trace = argv[i]; } else if (arg == "-c" || arg == "--cache") { params.use_cache = true; } else if (arg == "-h" || arg == "--help") { @@ -308,6 +316,8 @@ int main(int argc, char * argv[]) { fprintf(stderr, "\n"); } + ggml_trace_open(params.trace.empty() ? nullptr : params.trace.c_str(), "rpc-server"); + auto devices = get_devices(params); if (devices.empty()) { fprintf(stderr, "No devices found\n"); diff --git a/tools/server/server-context.cpp b/tools/server/server-context.cpp index 1723a55766e..5af06d21d03 100644 --- a/tools/server/server-context.cpp +++ b/tools/server/server-context.cpp @@ -17,6 +17,8 @@ #include "mtmd.h" #include "mtmd-helper.h" +#include "ggml-trace.h" + #include #include #include @@ -808,6 +810,27 @@ static int process_mtmd_chunk(const server_slot & slot, mtmd::batch_ptr & mbatch // With N > 1 the point is that while group A's batch is being computed on the second stage of a // layer split (the RPC peer), group B's batch can be computed on the first stage (the local GPU), // so both devices are busy instead of each idling half of every decode step. +// RAII span for the event tracer (ggml/include/ggml-trace.h). Off unless GGML_RPC_TRACE is set, +// and then one branch per call site. +struct server_trace_scope { + const char * name; + int64_t t0; + int n0; + int n1; + + server_trace_scope(const char * name, int n0, int n1) : + name(name), t0(ggml_trace_flag ? ggml_trace_time_us() : 0), n0(n0), n1(n1) {} + + ~server_trace_scope() { + if (ggml_trace_flag) { + ggml_trace_eventf("server", name, t0, ggml_trace_time_us(), "\"n0\":%d,\"n1\":%d", n0, n1); + } + } + + server_trace_scope(const server_trace_scope &) = delete; + server_trace_scope & operator=(const server_trace_scope &) = delete; +}; + // ----------------------------------------------------------------------------- // per-group host-path profiling, enabled with LLAMA_SERVER_PIPE_PROF=1 @@ -1917,6 +1940,10 @@ struct server_context_impl { // the decode loop of one pipeline group, only used when n_groups > 1 void group_loop(server_group & grp) { + // every event raised below this point, down to the individual RPC commands, is tagged + // with the group that caused it + ggml_trace_set_group(grp.id); + while (true) { if (groups_stop.load(std::memory_order_relaxed)) { return; @@ -3388,7 +3415,20 @@ struct server_context_impl { // to keep the shared task loop spinning } + if (ggml_trace_flag) { + ggml_trace_set_group(grp.id); + } + + int n_slots_processing = 0; + if (ggml_trace_flag) { + for (auto * slot : grp.slots) { + n_slots_processing += slot->is_processing() ? 1 : 0; + } + } + server_trace_scope span_iter("iteration", grp.id, n_slots_processing); + try { + server_trace_scope span_build("batch_build", grp.id, n_slots_processing); scoped_timer t(t_pre_decode, n_pre_decode); prof_timer tp(&grp.prof.t_pre, prof_on); pre_decode(grp); @@ -4289,10 +4329,12 @@ struct server_context_impl { } window(this, &grp, &lk); { + server_trace_scope span("submit", grp.id, batch_view.n_tokens); prof_timer ts(&grp.prof.t_submit, prof_on); ret = llama_decode(ctx_tgt, batch_view); } if (ret == 0 && has_output) { + server_trace_scope span("synchronize", grp.id, batch_view.n_tokens); prof_timer ts(&grp.prof.t_sync, prof_on); llama_synchronize(ctx_tgt); } @@ -4301,10 +4343,12 @@ struct server_context_impl { // note: the sync is done here too, so that the wait is also covered by the yield queue_tasks.yield_to_queue([&]() { { + server_trace_scope span("submit", grp.id, batch_view.n_tokens); prof_timer ts(&grp.prof.t_submit, prof_on); ret = llama_decode(ctx_tgt, batch_view); } if (ret == 0 && has_output) { + server_trace_scope span("synchronize", grp.id, batch_view.n_tokens); prof_timer ts(&grp.prof.t_sync, prof_on); llama_synchronize(ctx_tgt); } @@ -4416,6 +4460,8 @@ struct server_context_impl { } void post_decode(server_group & grp, int32_t n_batch_tokens, int32_t off, llama_batch & batch_view) { + server_trace_scope span_post("post_decode", grp.id, n_batch_tokens); + // shadow the single-context members, as update_slots() does auto * ctx_tgt = grp.ctx; auto & slots = grp.slots; @@ -4479,6 +4525,9 @@ struct server_context_impl { } if (to_sample.size() > 1) { + // one span for the whole pass, on this thread: the workers must not emit spans of + // their own, they would interleave and be counted several times over + server_trace_scope span("sampling", grp.id, (int) to_sample.size()); prof_timer ps(&grp.prof.t_sampl_par, prof_on); // resolve the first row on this thread: the first call after a decode may have to @@ -4552,6 +4601,7 @@ struct server_context_impl { id = slot.pre_sampled; slot.pre_sampled = LLAMA_TOKEN_NULL; } else { + server_trace_scope span("sampling", grp.id, slot.id); scoped_timer timer(t_sampl, n_sampl); prof_timer ps(&grp.prof.t_sampl, prof_on); id = common_sampler_sample(slot.smpl.get(), slot.ctx_tgt, tok_idx); @@ -4587,18 +4637,22 @@ struct server_context_impl { populate_token_probs(slot, result, slot.task->params.post_sampling_probs, params_base.special, tok_idx); } - bool keep_going; { - prof_timer pt(&grp.prof.t_proc, prof_on); - keep_going = process_token(result, slot); - } - if (!keep_going) { - // release slot because of stop condition - slot.print_timings(); - send_final_response(slot); - slot.release(); + server_trace_scope span("result_send", grp.id, slot.id); - return; + bool keep_going; + { + prof_timer pt(&grp.prof.t_proc, prof_on); + keep_going = process_token(result, slot); + } + if (!keep_going) { + // release slot because of stop condition + slot.print_timings(); + send_final_response(slot); + slot.release(); + + return; + } } slot.print_timings_tg(); diff --git a/tools/server/server.cpp b/tools/server/server.cpp index 45c8d7f7005..99edd7dbf03 100644 --- a/tools/server/server.cpp +++ b/tools/server/server.cpp @@ -1,4 +1,5 @@ #include "server-context.h" +#include "ggml-trace.h" #include "server-http.h" #include "server-models.h" #include "server-cors-proxy.h" @@ -133,6 +134,9 @@ static server_http_context::handler_t ex_wrapper(server_http_context::handler_t int llama_server(int argc, char ** argv) { std::setlocale(LC_NUMERIC, "C"); + // opens the event trace if GGML_RPC_TRACE names a file, otherwise this is a no-op + ggml_trace_open(nullptr, "llama-server"); + #ifndef _WIN32 // Ignore SIGPIPE so the server does not crash if a child (MCP server, tools runtime) exits while we are writing to its stdin signal(SIGPIPE, SIG_IGN);