From d782e622fc538ac0341680ab4c5cb79c96201494 Mon Sep 17 00:00:00 2001 From: Jihyeon Gim Date: Sat, 5 Sep 2026 19:46:00 +0900 Subject: [PATCH] rdma: add a log callback ABI to the RC data plane Wire C-side diagnostics (session reap, READY data phase outcome, init failures) through a sink callback so the gateway can surface them next to its own logs instead of losing them in stderr noise. The sink is a plain C function pointer installed once after init and valid until destroy: the Go side registers a fixed cgo trampoline (closures cannot cross the boundary), copies the message immediately per the lifetime contract, and never runs under the session map lock. Error-level lines keep the existing stderr output; --debug enables the level-2 diagnostic stream. --- cmd/vgwrdma/main.go | 1 + cuwrapper/rc/rc_server_abi.cpp | 57 ++++++++++++++++++++++++++++++++ cuwrapper/rc/rc_server_abi.h | 11 +++++++ rdma/rcserver/rcserver_linux.go | 58 +++++++++++++++++++++++++++++++++ rdma/rcserver/rcserver_stub.go | 1 + 5 files changed, 128 insertions(+) diff --git a/cmd/vgwrdma/main.go b/cmd/vgwrdma/main.go index d771d438..1779c312 100644 --- a/cmd/vgwrdma/main.go +++ b/cmd/vgwrdma/main.go @@ -1263,6 +1263,7 @@ func runGateway(ctx context.Context, be backend.Backend) error { MaxReadySlots: uint32(rcMaxReadySlots), TPrepMs: rcPrepTimeoutMs, TExecMs: rcExecTimeoutMs, + Debug: debug, }) if err != nil { return err diff --git a/cuwrapper/rc/rc_server_abi.cpp b/cuwrapper/rc/rc_server_abi.cpp index dfcfa2b1..c915b417 100644 --- a/cuwrapper/rc/rc_server_abi.cpp +++ b/cuwrapper/rc/rc_server_abi.cpp @@ -15,6 +15,8 @@ #include #include #include +#include +#include #include "rc_ibv_host.h" #include "v2_data_phase.h" @@ -89,6 +91,13 @@ struct rc_server { hipObj::DeviceHandle *device = nullptr; SessionTable table; rc_device_opts opts{}; + /* Diagnostic sink: null keeps stderr-only error reporting. + * Reads/writes are plain loads/stores; the sink is installed + * once at init time (before the reaper starts) and only + * cleared by destroy after the reaper joined, so no thread + * races an in-flight sink pointer swap. */ + rc_log_fn log_fn = nullptr; + void *log_ctx = nullptr; std::atomic epoch_counter{1}; /* resource accounting (global buckets; per-principal map). */ std::mutex acct_mtx; @@ -117,6 +126,22 @@ struct rc_server { namespace { +/* Emits a diagnostic line to the installed sink (level 0 keeps + * the stderr error stream intact by also printing there, so + * existing deployments do not lose the only log they had). + * Callers must not hold map_mtx/acct_mtx when calling. */ +void rcLog(const rc_server *srv, int level, const char *file, int line, + const char *fmt, ...) { + char buf[256]; + va_list ap; + va_start(ap, fmt); + vsnprintf(buf, sizeof(buf), fmt, ap); + va_end(ap); + if (level <= 0) fprintf(stderr, "%s\n", buf); + rc_log_fn fn = srv->log_fn; + if (fn) fn(srv->log_ctx, level, buf, file, line); +} + RcSession *findSession(rc_server *srv, const std::string &id) { auto it = srv->sessions_map.find(id); return it == srv->sessions_map.end() ? nullptr : it->second.get(); @@ -187,6 +212,16 @@ void reapSession(rc_server *srv, RcSession *s) { s->core.qp = conn.qp; /* null on success, survivor on failure */ s->core.cq = conn.cq; bool destroyed = q_ok && c_ok; + /* Terminal record for every session teardown path (expiry, + * CANCEL, and destroy); reap_pass may have missed the final + * state, so the last outcome observed at READY time travels + * with the log line. */ + rcLog(srv, 2, __FILE__, __LINE__, + "rc: session reaped id=%s op=%s target=%.96s staged=%llu " + "qp_destroyed=%d", + s->core.id.c_str(), s->core.op.c_str(), + s->core.target.c_str(), + (unsigned long long)s->staging_len, (int)destroyed); if (destroyed) { /* Same policy as releaseStaging: a failed dereg leaves the * MR registered against the shared PD, so the buffer stays @@ -257,6 +292,12 @@ std::string encodeReplyToken(hipObj::DeviceHandle *dh, uint32_t qpn) { extern "C" { +void rc_server_set_log_sink(rc_server *srv, rc_log_fn fn, void *ctx) { + if (!srv) return; + srv->log_fn = fn; + srv->log_ctx = ctx; +} + int rc_server_init(const rc_device_opts *opts, rc_server **out) { if (!opts || !out) return RC_E_ARG; if (!hipObj::ibv.ensureLoaded()) return RC_E_INTERNAL; @@ -309,12 +350,15 @@ int rc_server_init(const rc_device_opts *opts, rc_server **out) { } if (!ctx) { hipObj::ibv.free_device_list(devs); + fprintf(stderr, "rc: no verbs device matches gid_hint %.32s\n", + opts->gid_hint ? opts->gid_hint : ""); return RC_E_INTERNAL; } struct ibv_pd *pd = hipObj::ibv.alloc_pd(ctx); hipObj::ibv.free_device_list(devs); if (!pd) { hipObj::ibv.close_device(ctx); + fprintf(stderr, "rc: alloc_pd failed\n"); return RC_E_INTERNAL; } srv->device = new hipObj::DeviceHandle(); @@ -798,6 +842,19 @@ int rc_ready_transfer(rc_server *srv, const rc_ready_req *req, RcSession *after = findSession(srv, id); if (!after) return RC_E_NO_SESSION; + /* Keep the wire-level reason (poll status vs post failure vs + * timeout) alongside the outcome the response carries, so a + * VerifyFail is diagnosable without re-running the transfer. + * Logged with the lock dropped: the sink must not block under + * map_mtx. */ + int rlog = (r == hipObj::v2::DataPhaseResult::Ok) + ? 2 + : (r == hipObj::v2::DataPhaseResult::Busy ? 2 : 0); + rcLog(srv, rlog, __FILE__, __LINE__, + "rc: ready data phase session=%s op=%s outcome=%d bytes=%llu", + id.c_str(), after->core.op.c_str(), (int)r, + (unsigned long long)stats.bytes); + switch (r) { case hipObj::v2::DataPhaseResult::Ok: after->last_outcome = RC_READY_OK; diff --git a/cuwrapper/rc/rc_server_abi.h b/cuwrapper/rc/rc_server_abi.h index 55c7d191..4606b1a5 100644 --- a/cuwrapper/rc/rc_server_abi.h +++ b/cuwrapper/rc/rc_server_abi.h @@ -39,8 +39,19 @@ enum { typedef struct { const char *ptr; uint32_t len; } rc_str_in; +/* Diagnostic log sink. Called from RC server threads (request + * goroutines via cgo and the expiry reaper) without any server + * lock held. `msg` and `file` are only valid for the duration of + * the call; the sink must copy them if it needs them longer. + * `level` is 0 (error) or 2 (debug). Passing a null sink (or + * installing it after init) keeps stderr-only error reporting. */ +typedef void (*rc_log_fn)(void *ctx, int level, const char *msg, + const char *file, int line); + typedef struct rc_server rc_server; +void rc_server_set_log_sink(rc_server *srv, rc_log_fn fn, void *ctx); + /* Device selection: matching GID prefix when gid_hint is set, * otherwise the first verbs device. */ typedef struct { diff --git a/rdma/rcserver/rcserver_linux.go b/rdma/rcserver/rcserver_linux.go index b2f3b9a5..37c317bd 100644 --- a/rdma/rcserver/rcserver_linux.go +++ b/rdma/rcserver/rcserver_linux.go @@ -27,6 +27,12 @@ package rcserver #cgo LDFLAGS: -L${SRCDIR}/.. -l:librcserver.a -lstdc++ -ldl -lpthread #include "rc_server_abi.h" #include + +// Shim: a C function pointer that forwards into the exported Go +// sink. Go closures cannot be stored as C callbacks, so the +// level/file/line/msg are marshalled through this fixed trampoline. +extern void rcgo_log_sink(void *ctx, int level, char *msg, + char *file, int line); */ import "C" @@ -34,6 +40,7 @@ import ( "context" "errors" "fmt" + "log" "runtime" "sync" "sync/atomic" @@ -76,6 +83,10 @@ type DeviceOpts struct { // Concurrency slots (0 picks the C-side default 64/32). MaxReadySlots uint32 MaxStageSlots uint32 + // Debug enables level-2 (debug) RC diagnostics through the + // log sink; false keeps error-only reporting. Mirrors the + // gateway --debug flag. + Debug bool } // PrincipalID is the SHA-256 digest identifying the requester. @@ -174,6 +185,52 @@ func (s *RCSvc) Context() context.Context { return s.ctx } +// rcLogSink receives every diagnostic line the C server emits. +// It is stateless and process-global on purpose: the sink must be +// valid from init through destroy, independent of any single +// RCSvc instance. Lines are copied out immediately (the C msg +// buffer is only valid for the duration of the callback) and +// written to the standard error logger. rcLogMu serializes the +// copy because log.Logger is safe for concurrent use but keeps +// a single writer cheap. +var ( + rcLogMu sync.Mutex + rcLogLevel atomic.Int32 + rcLog = log.New(log.Writer(), "vgwrdma rc: ", 0) +) + +//export rcgo_log_sink +func rcgo_log_sink(_ unsafe.Pointer, level C.int, msg *C.char, + file *C.char, line C.int) { + // Level gating mirrors the C contract: 0 (error) always + // arrives here; 2 (debug) only when the service opted in. + if int32(level) > rcLogLevel.Load() { + return + } + m := C.GoString(msg) + f := "" + if file != nil { + f = C.GoString(file) + } + rcLogMu.Lock() + rcLog.Printf("%s (%s:%d) level=%d", m, f, int(line), int(level)) + rcLogMu.Unlock() +} + +// installLogSink wires the C server's diagnostics into the Go +// sink. debug selects level 2 (debug) versus level 0 (errors +// only); the C side keeps stderr error reporting when the sink +// is absent, so a nil install is a no-op by design. +func installLogSink(srv *C.rc_server, debug bool) { + if debug { + rcLogLevel.Store(2) + } else { + rcLogLevel.Store(0) + } + C.rc_server_set_log_sink(srv, + (*[0]byte)(C.rcgo_log_sink), nil) +} + // Init opens the verbs device and returns a service. func Init(opts DeviceOpts) (*RCSvc, error) { copts := C.rc_device_opts{ @@ -201,6 +258,7 @@ func Init(opts DeviceOpts) (*RCSvc, error) { if rc := C.rc_server_init(&copts, &srv); rc != C.RC_OK { return nil, fmt.Errorf("rcserver: init failed: status %d", int(rc)) } + installLogSink(srv, opts.Debug) ctx, cancel := context.WithCancel(context.Background()) return &RCSvc{srv: srv, ctx: ctx, cancel: cancel}, nil } diff --git a/rdma/rcserver/rcserver_stub.go b/rdma/rcserver/rcserver_stub.go index 6bb051e8..820858ca 100644 --- a/rdma/rcserver/rcserver_stub.go +++ b/rdma/rcserver/rcserver_stub.go @@ -37,6 +37,7 @@ type DeviceOpts struct { TExecMs uint64 MaxReadySlots uint32 MaxStageSlots uint32 + Debug bool } // PrincipalID is the SHA-256 digest identifying the requester.