From 980078d8228a53425ffbd392c36a8021d26c47a2 Mon Sep 17 00:00:00 2001 From: Jihyeon Gim Date: Sat, 5 Sep 2026 17:53:05 +0900 Subject: [PATCH 1/8] rdma: expose the RC data plane resource limits as gateway flags The hipobj-rc-v2 data plane started with its session, queue pair, staging and timeout limits hardcoded at the rcserver.Init call site, so operators could not size the RC plane for their hardware the way they can for the cuObject backend. Add one flag per limit plus the READY admission slot count, all defaulting to the values the gateway passes today, and validate them through a new rdmamode.V2ValidationError consulted only when the RC data plane is enabled, mirroring the stale-value handling of the v1 settings. Counts are parsed as uint64 and range-checked against the uint32 narrowing at the DeviceOpts boundary, and the timeouts carry an upper bound that keeps the nowMs + timeout deadline arithmetic in the C core from wrapping. --- cmd/vgwrdma/main.go | 108 ++++++++++++++++++++++++++--- internal/rdmamode/rdmamode.go | 52 ++++++++++++++ internal/rdmamode/rdmamode_test.go | 97 ++++++++++++++++++++++++++ 3 files changed, 249 insertions(+), 8 deletions(-) diff --git a/cmd/vgwrdma/main.go b/cmd/vgwrdma/main.go index 33306dd3..d771d438 100644 --- a/cmd/vgwrdma/main.go +++ b/cmd/vgwrdma/main.go @@ -120,6 +120,15 @@ var ( rdmaCQDepth uint rdmaRetryCount uint rdmaTunablesSet bool + rcMaxSessions uint64 + rcMaxUserSessions uint64 + rcMaxStagingBytes uint64 + rcMaxUserStagingBytes uint64 + rcMaxQPs uint64 + rcMaxUserQPs uint64 + rcMaxReadySlots uint64 + rcPrepTimeoutMs uint64 + rcExecTimeoutMs uint64 ) var ( @@ -898,6 +907,69 @@ func initFlags() []cli.Flag { EnvVars: []string{"VGW_RDMA_RC_ENABLE"}, Destination: &rdmaRCEnable, }, + &cli.Uint64Flag{ + Name: "rdma-rc-max-sessions", + Usage: "maximum concurrent hipobj-rc-v2 sessions (default 1024)", + EnvVars: []string{"VGW_RDMA_RC_MAX_SESSIONS"}, + Value: 1024, + Destination: &rcMaxSessions, + }, + &cli.Uint64Flag{ + Name: "rdma-rc-max-user-sessions", + Usage: "per-principal session limit for the hipobj-rc-v2 data plane (default 64)", + EnvVars: []string{"VGW_RDMA_RC_MAX_USER_SESSIONS"}, + Value: 64, + Destination: &rcMaxUserSessions, + }, + &cli.Uint64Flag{ + Name: "rdma-rc-max-staging-bytes", + Usage: "total staging buffer budget for hipobj-rc-v2 sessions in bytes (default 4294967296)", + EnvVars: []string{"VGW_RDMA_RC_MAX_STAGING_BYTES"}, + Value: 4 << 30, + Destination: &rcMaxStagingBytes, + }, + &cli.Uint64Flag{ + Name: "rdma-rc-max-user-staging-bytes", + Usage: "per-principal staging buffer budget for the hipobj-rc-v2 data plane in bytes (default 1073741824)", + EnvVars: []string{"VGW_RDMA_RC_MAX_USER_STAGING_BYTES"}, + Value: 1 << 30, + Destination: &rcMaxUserStagingBytes, + }, + &cli.Uint64Flag{ + Name: "rdma-rc-max-qps", + Usage: "maximum queue pairs for the hipobj-rc-v2 data plane (default 1024)", + EnvVars: []string{"VGW_RDMA_RC_MAX_QPS"}, + Value: 1024, + Destination: &rcMaxQPs, + }, + &cli.Uint64Flag{ + Name: "rdma-rc-max-user-qps", + Usage: "per-principal queue pair limit for the hipobj-rc-v2 data plane (default 16)", + EnvVars: []string{"VGW_RDMA_RC_MAX_USER_QPS"}, + Value: 16, + Destination: &rcMaxUserQPs, + }, + &cli.Uint64Flag{ + Name: "rdma-rc-max-ready-slots", + Usage: "concurrent READY transfers admitted by the hipobj-rc-v2 data plane (default 64)", + EnvVars: []string{"VGW_RDMA_RC_MAX_READY_SLOTS"}, + Value: 64, + Destination: &rcMaxReadySlots, + }, + &cli.Uint64Flag{ + Name: "rdma-rc-prep-timeout-ms", + Usage: "milliseconds a hipobj-rc-v2 session may wait for READY after PREPARE (default 100000)", + EnvVars: []string{"VGW_RDMA_RC_PREP_TIMEOUT_MS"}, + Value: 100000, + Destination: &rcPrepTimeoutMs, + }, + &cli.Uint64Flag{ + Name: "rdma-rc-exec-timeout-ms", + Usage: "milliseconds a hipobj-rc-v2 READY transfer may run (default 30000)", + EnvVars: []string{"VGW_RDMA_RC_EXEC_TIMEOUT_MS"}, + Value: 30000, + Destination: &rcExecTimeoutMs, + }, &cli.UintFlag{ Name: "rdma-port", Usage: "port for RDMA listener", @@ -1006,6 +1078,25 @@ func runGateway(ctx context.Context, be backend.Backend) error { return errors.New(msg) } } + if v2On { + // RC-only settings are irrelevant when the hipobj-rc-v2 + // data plane is not running; stale environment values + // must not block v1-only or plain-S3 startup. + v2s := rdmamode.V2Settings{ + MaxSessions: rcMaxSessions, + MaxUserSessions: rcMaxUserSessions, + MaxStagingBytes: rcMaxStagingBytes, + MaxUserStagingBytes: rcMaxUserStagingBytes, + MaxQPs: rcMaxQPs, + MaxUserQPs: rcMaxUserQPs, + MaxReadySlots: rcMaxReadySlots, + TPrepMs: rcPrepTimeoutMs, + TExecMs: rcExecTimeoutMs, + } + if msg := rdmamode.V2ValidationError(v2s); msg != "" { + return errors.New(msg) + } + } var s3Opts []s3api.Option if v1On { @@ -1163,14 +1254,15 @@ func runGateway(ctx context.Context, be backend.Backend) error { rcSvc, err := rcserver.Init(rcserver.DeviceOpts{ GidHint: rcGidHint, Port: 1, - MaxSessions: 1024, - MaxUserSessions: 64, - MaxStagingBytes: 4 << 30, - MaxUserStagingBytes: 1 << 30, - MaxQPs: 1024, - MaxUserQPs: 16, - TPrepMs: 100000, - TExecMs: 30000, + MaxSessions: uint32(rcMaxSessions), + MaxUserSessions: uint32(rcMaxUserSessions), + MaxStagingBytes: rcMaxStagingBytes, + MaxUserStagingBytes: rcMaxUserStagingBytes, + MaxQPs: uint32(rcMaxQPs), + MaxUserQPs: uint32(rcMaxUserQPs), + MaxReadySlots: uint32(rcMaxReadySlots), + TPrepMs: rcPrepTimeoutMs, + TExecMs: rcExecTimeoutMs, }) if err != nil { return err diff --git a/internal/rdmamode/rdmamode.go b/internal/rdmamode/rdmamode.go index a5b40aca..92cad9fa 100644 --- a/internal/rdmamode/rdmamode.go +++ b/internal/rdmamode/rdmamode.go @@ -21,6 +21,7 @@ import ( "math" "strings" "sync" + "time" "github.com/versity/versitygw/backend" ) @@ -68,6 +69,57 @@ func V1ValidationError(s V1Settings) string { } } +// V2Settings carries the hipobj-rc-v2 tunings that are validated +// only while the RC data plane runs. Counts arrive as uint64 from +// the CLI and narrow to uint32 at the DeviceOpts boundary, so the +// range is checked before the narrowing cast. Timeouts feed +// deadline arithmetic (nowMs + timeout), so an upper bound keeps +// the sum from wrapping. +type V2Settings struct { + MaxSessions uint64 + MaxUserSessions uint64 + MaxStagingBytes uint64 + MaxUserStagingBytes uint64 + MaxQPs uint64 + MaxUserQPs uint64 + MaxReadySlots uint64 + TPrepMs uint64 + TExecMs uint64 +} + +// v2TimeoutCeiling bounds the RC timeouts well below the uint64 +// wrap point of nowMs + timeout arithmetic in the C core. +const v2TimeoutCeiling = uint64(24 * time.Hour / time.Millisecond) + +// V2ValidationError describes the first invalid v2 setting, or +// the empty string when every setting is valid. Stale v2 values +// in the environment must not block v1-only or plain-S3 startup, +// so the gateway consults this only when v2 is on. +func V2ValidationError(s V2Settings) string { + switch { + case s.MaxSessions < 1 || s.MaxSessions > math.MaxUint32: + return fmt.Sprintf("rdma-rc-max-sessions %d is out of range (1-%d)", s.MaxSessions, uint64(math.MaxUint32)) + case s.MaxUserSessions < 1 || s.MaxUserSessions > math.MaxUint32: + return fmt.Sprintf("rdma-rc-max-user-sessions %d is out of range (1-%d)", s.MaxUserSessions, uint64(math.MaxUint32)) + case s.MaxStagingBytes < 1: + return fmt.Sprintf("rdma-rc-max-staging-bytes %d must be positive", s.MaxStagingBytes) + case s.MaxUserStagingBytes < 1: + return fmt.Sprintf("rdma-rc-max-user-staging-bytes %d must be positive", s.MaxUserStagingBytes) + case s.MaxQPs < 1 || s.MaxQPs > math.MaxUint32: + return fmt.Sprintf("rdma-rc-max-qps %d is out of range (1-%d)", s.MaxQPs, uint64(math.MaxUint32)) + case s.MaxUserQPs < 1 || s.MaxUserQPs > math.MaxUint32: + return fmt.Sprintf("rdma-rc-max-user-qps %d is out of range (1-%d)", s.MaxUserQPs, uint64(math.MaxUint32)) + case s.MaxReadySlots < 1 || s.MaxReadySlots > math.MaxUint32: + return fmt.Sprintf("rdma-rc-max-ready-slots %d is out of range (1-%d)", s.MaxReadySlots, uint64(math.MaxUint32)) + case s.TPrepMs < 1 || s.TPrepMs > v2TimeoutCeiling: + return fmt.Sprintf("rdma-rc-prep-timeout-ms %d is out of range (1-%d)", s.TPrepMs, v2TimeoutCeiling) + case s.TExecMs < 1 || s.TExecMs > v2TimeoutCeiling: + return fmt.Sprintf("rdma-rc-exec-timeout-ms %d is out of range (1-%d)", s.TExecMs, v2TimeoutCeiling) + default: + return "" + } +} + // Closer is the close operation of the RC session service. It is // idempotent. type Closer interface{ Close() } diff --git a/internal/rdmamode/rdmamode_test.go b/internal/rdmamode/rdmamode_test.go index abeaa515..3da8d035 100644 --- a/internal/rdmamode/rdmamode_test.go +++ b/internal/rdmamode/rdmamode_test.go @@ -256,3 +256,100 @@ func TestRCWrapperChainsOntoOnceBackend(t *testing.T) { t.Fatalf("base backend shutdown %d times, want 1", shutdowns) } } + +func v2Defaults() V2Settings { + return V2Settings{ + MaxSessions: 1024, + MaxUserSessions: 64, + MaxStagingBytes: 4 << 30, + MaxUserStagingBytes: 1 << 30, + MaxQPs: 1024, + MaxUserQPs: 16, + MaxReadySlots: 64, + TPrepMs: 100000, + TExecMs: 30000, + } +} + +func TestV2ValidationDefaultsPass(t *testing.T) { + // The defaults mirror the values the gateway passed before + // the flags existed, so a deployment that sets nothing must + // keep starting. + if msg := V2ValidationError(v2Defaults()); msg != "" { + t.Fatalf("defaults rejected: %q", msg) + } +} + +func TestV2ValidationRejectsZero(t *testing.T) { + // Every knob must be positive: counts narrow to uint32 and + // zero would disable a limit or a deadline in the C core. + for name, mutate := range map[string]func(*V2Settings){ + "max-sessions": func(s *V2Settings) { s.MaxSessions = 0 }, + "max-user-sessions": func(s *V2Settings) { s.MaxUserSessions = 0 }, + "max-staging-bytes": func(s *V2Settings) { s.MaxStagingBytes = 0 }, + "max-user-staging-bytes": func(s *V2Settings) { s.MaxUserStagingBytes = 0 }, + "max-qps": func(s *V2Settings) { s.MaxQPs = 0 }, + "max-user-qps": func(s *V2Settings) { s.MaxUserQPs = 0 }, + "max-ready-slots": func(s *V2Settings) { s.MaxReadySlots = 0 }, + "prep-timeout": func(s *V2Settings) { s.TPrepMs = 0 }, + "exec-timeout": func(s *V2Settings) { s.TExecMs = 0 }, + } { + s := v2Defaults() + mutate(&s) + if msg := V2ValidationError(s); msg == "" { + t.Fatalf("%s: zero accepted", name) + } + } +} + +func TestV2ValidationCountBoundaries(t *testing.T) { + // Counts arrive as uint64 and narrow to uint32 at the + // DeviceOpts boundary: MaxUint32 passes, the first value + // beyond it fails. + fields := []struct { + name string + field *uint64 + }{ + {"max-sessions", new(uint64)}, + {"max-user-sessions", new(uint64)}, + {"max-qps", new(uint64)}, + {"max-user-qps", new(uint64)}, + {"max-ready-slots", new(uint64)}, + } + for _, f := range fields { + base := v2Defaults() + f.field = &base.MaxSessions + switch f.name { + case "max-user-sessions": + f.field = &base.MaxUserSessions + case "max-qps": + f.field = &base.MaxQPs + case "max-user-qps": + f.field = &base.MaxUserQPs + case "max-ready-slots": + f.field = &base.MaxReadySlots + } + *f.field = math.MaxUint32 + if msg := V2ValidationError(base); msg != "" { + t.Fatalf("%s at MaxUint32 rejected: %q", f.name, msg) + } + *f.field = math.MaxUint32 + 1 + if msg := V2ValidationError(base); msg == "" { + t.Fatalf("%s beyond MaxUint32 accepted", f.name) + } + } +} + +func TestV2ValidationTimeoutCeiling(t *testing.T) { + // Timeouts feed nowMs + timeout deadline arithmetic in the + // C core, so values past the ceiling are refused. + base := v2Defaults() + base.TPrepMs = v2TimeoutCeiling + if msg := V2ValidationError(base); msg != "" { + t.Fatalf("ceiling rejected: %q", msg) + } + base.TPrepMs = v2TimeoutCeiling + 1 + if msg := V2ValidationError(base); msg == "" { + t.Fatal("ceiling+1 accepted") + } +} From d782e622fc538ac0341680ab4c5cb79c96201494 Mon Sep 17 00:00:00 2001 From: Jihyeon Gim Date: Sat, 5 Sep 2026 19:46:00 +0900 Subject: [PATCH 2/8] 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. From 06a34632ea5027334a699eaafaea3fec91b05788 Mon Sep 17 00:00:00 2001 From: Jihyeon Gim Date: Sat, 5 Sep 2026 19:50:29 +0900 Subject: [PATCH 3/8] rdma: name the deviceless init failures in the RC server The verbs loader and device enumeration failures returned RC_E_INTERNAL without any stderr trace, which made a VM or container without RDMA indistinguishable from a genuine library problem. Print the failing step so operators can tell the two apart at startup. --- cuwrapper/rc/rc_server_abi.cpp | 10 ++++++++-- 1 file changed, 8 insertions(+), 2 deletions(-) diff --git a/cuwrapper/rc/rc_server_abi.cpp b/cuwrapper/rc/rc_server_abi.cpp index c915b417..da2f1938 100644 --- a/cuwrapper/rc/rc_server_abi.cpp +++ b/cuwrapper/rc/rc_server_abi.cpp @@ -300,7 +300,10 @@ void rc_server_set_log_sink(rc_server *srv, rc_log_fn fn, void *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; + if (!hipObj::ibv.ensureLoaded()) { + fprintf(stderr, "rc: cannot load libibverbs (dlopen/dlsym failed)\n"); + return RC_E_INTERNAL; + } std::unique_ptr srv(new rc_server()); srv->opts = *opts; /* ibv port numbers are 1-based; treat an unset (0) port as 1 so @@ -316,7 +319,10 @@ int rc_server_init(const rc_device_opts *opts, rc_server **out) { int n = 0; struct ibv_device **devs = hipObj::ibv.get_device_list(&n); - if (!devs || n == 0) return RC_E_INTERNAL; + if (!devs || n == 0) { + fprintf(stderr, "rc: no RDMA devices found (ibv_get_device_list)\n"); + return RC_E_INTERNAL; + } struct ibv_device *chosen = devs[0]; /* GID hint: pick the first device/port whose GID starts with it. * Query with srv->opts.port, which the normalization above has From 09bbe9ccbda4d10795d822a40756198423c240ec Mon Sep 17 00:00:00 2001 From: Jihyeon Gim Date: Sat, 5 Sep 2026 19:53:28 +0900 Subject: [PATCH 4/8] rdma: resolve the inline-only verbs calls from the provider ops table ibv_poll_cq, ibv_post_send, and ibv_post_recv stopped being exported library symbols in modern rdma-core: verbs.h ships them as static inline wrappers that dispatch through cq->context->ops. dlsym therefore returned null for them and the loader rejected perfectly usable libraries, failing RC server init with a bare RC_E_INTERNAL on hosts with rdma-core 61+. Open the first device briefly, read the three function pointers from its context ops table, and close it again. The check now only requires symbols that actually exist in the library, and the failure mode for an ops-less provider is explicit. --- cuwrapper/rc/rc_ibv_host.cpp | 28 +++++++++++++++++++++++++++- 1 file changed, 27 insertions(+), 1 deletion(-) diff --git a/cuwrapper/rc/rc_ibv_host.cpp b/cuwrapper/rc/rc_ibv_host.cpp index 3ebcdadd..0d0c9cd9 100644 --- a/cuwrapper/rc/rc_ibv_host.cpp +++ b/cuwrapper/rc/rc_ibv_host.cpp @@ -75,7 +75,33 @@ bool IBVWrapper::ensureLoaded() { struct ibv_send_wr *, struct ibv_send_wr **)>(load("ibv_post_send")); loaded = funcs_.get_device_list != nullptr && funcs_.open_device != nullptr && funcs_.alloc_pd != nullptr && funcs_.create_qp != nullptr && - funcs_.modify_qp != nullptr && funcs_.poll_cq != nullptr; + funcs_.modify_qp != nullptr; + if (loaded) { + /* poll_cq/post_send/post_recv are static inline wrappers in + * modern verbs.h (they dispatch through cq->context->ops), so + * dlsym cannot find them on rdma-core 61+. Resolve them from + * the ops table of the first successfully opened context + * instead; every context from the same device shares these + * providers. */ + int n = 0; + struct ibv_device **devs = funcs_.get_device_list(&n); + struct ibv_context *probe = nullptr; + if (devs && n > 0) probe = funcs_.open_device(devs[0]); + if (devs) funcs_.free_device_list(devs); + if (!probe) { + fprintf(stderr, "rc: no RDMA device to resolve verbs ops\n"); + loaded = false; + } else { + funcs_.poll_cq = probe->ops.poll_cq; + funcs_.post_recv = probe->ops.post_recv; + funcs_.post_send = probe->ops.post_send; + funcs_.close_device(probe); + if (!funcs_.poll_cq || !funcs_.post_recv || !funcs_.post_send) { + fprintf(stderr, "rc: provider ops table incomplete\n"); + loaded = false; + } + } + } if (loaded) { /* Mirror into the member seam for direct ibv.x() calls. */ get_device_list = funcs_.get_device_list; From 34a3e4152ed8c00246f4ec5a44ad78d83d876a39 Mon Sep 17 00:00:00 2001 From: Jihyeon Gim Date: Sat, 5 Sep 2026 21:59:00 +0900 Subject: [PATCH 5/8] rdma: pin session strings passed to the RC cgo ABI The Prepare and ReadyTransfer wrappers embed string views built from Go heap strings inside request structs passed to C by pointer. The cgo pointer check rejects such requests when the string data is an unpinned Go heap pointer, so any live PREPARE or READY call with header-derived strings panicked at the call boundary and the route returned a 500. Constant strings passed the check because their data lives in read-only static storage, which is why standalone callers kept working while the gateway did not. Pin the string bytes with runtime.Pinner for the duration of the cgo call and drop the now redundant KeepAlive calls in those two wrappers. The other string-taking wrappers pass rc_str_in by value and are unaffected. Also add a deviceless cgo boundary regression test that calls the real Prepare wrapper with heap-backed interior-pointer strings and an invalid opcode, so C returns from argument validation before the server handle is touched. --- rdma/rcserver/rcserver_cgo_linux_test.go | 58 ++++++++++++++++++++++++ rdma/rcserver/rcserver_linux.go | 35 +++++++++----- 2 files changed, 82 insertions(+), 11 deletions(-) create mode 100644 rdma/rcserver/rcserver_cgo_linux_test.go diff --git a/rdma/rcserver/rcserver_cgo_linux_test.go b/rdma/rcserver/rcserver_cgo_linux_test.go new file mode 100644 index 00000000..6df4e6a9 --- /dev/null +++ b/rdma/rcserver/rcserver_cgo_linux_test.go @@ -0,0 +1,58 @@ +//go:build linux && cgo + +package rcserver + +import ( + "errors" + "strings" + "testing" + "unsafe" +) + +// heapSubstring returns a substring whose backing array is heap-allocated +// and whose data pointer is an interior pointer, mirroring how header +// strings reach the binding from the gateway routes. +func heapSubstring(value string) string { + const prefix = "prefix:" + backing := strings.Clone(prefix + value + ":suffix") + return backing[len(prefix) : len(prefix)+len(value)] +} + +// TestPrepareHeapStringsReachCValidation calls the real Prepare wrapper +// with heap-backed strings. The request uses an invalid opcode so the C +// entrypoint returns RC_E_ARG from its argument validation before the +// server handle is dereferenced; the dummy handle below is never touched. +// Before the pinnedStrIn fix this call panics at the cgo pointer check; +// after it, the C argument validation runs and the error surfaces. +func TestPrepareHeapStringsReachCValidation(t *testing.T) { + const literalTarget = "/bucket1/obj1" + heapTarget := heapSubstring(literalTarget) + heapToken := heapSubstring(strings.Repeat("0", 88)) + for _, tc := range []struct { + name, target, token string + }{ + {"literal_control", literalTarget, ""}, + {"heap_target", heapTarget, ""}, + {"heap_token", literalTarget, heapToken}, + {"both_heap", heapTarget, heapToken}, + {"empty_control", "", ""}, + } { + t.Run(tc.name, func(t *testing.T) { + var svc RCSvc + // Test-only opaque sentinel, not an initialized rc_server. + // Op=255 returns in C argument validation before the + // server pointer is dereferenced. + dummy := new(uint64) + *(*unsafe.Pointer)(unsafe.Pointer(&svc.srv)) = unsafe.Pointer(dummy) + resp, err := svc.Prepare(PrepareRequest{ + Op: 255, + Size: 1, + Target: tc.target, + ClientToken: tc.token, + }) + if resp != nil || !errors.Is(err, ErrArg) { + t.Fatalf("Prepare = (%v, %v), want (nil, ErrArg)", resp, err) + } + }) + } +} diff --git a/rdma/rcserver/rcserver_linux.go b/rdma/rcserver/rcserver_linux.go index 37c317bd..481170e6 100644 --- a/rdma/rcserver/rcserver_linux.go +++ b/rdma/rcserver/rcserver_linux.go @@ -371,18 +371,33 @@ func statusError(rc C.int) error { } } +// strIn borrows string bytes for a synchronous rc_str_in value argument. +// For a string view embedded in a request struct passed by pointer, use +// pinnedStrIn instead: the cgo pointer check rejects request structs that +// contain unpinned Go string data. C must not retain or modify the borrowed +// bytes. func strIn(s string) C.rc_str_in { if s == "" { return C.rc_str_in{ptr: nil, len: 0} } - // The pointer must stay alive across the cgo call; every - // wrapper keeps the string referenced until after the call. return C.rc_str_in{ ptr: (*C.char)(unsafe.Pointer(unsafe.StringData(s))), len: C.uint32_t(len(s)), } } +// pinnedStrIn builds a strIn view and pins the string bytes for the +// duration of the enclosing cgo call. The caller owns the Pinner and +// must defer Unpin before the first pinnedStrIn call. Pinning an +// interior pointer pins the whole backing allocation. +func pinnedStrIn(s string, pins *runtime.Pinner) C.rc_str_in { + in := strIn(s) + if in.ptr != nil { + pins.Pin(unsafe.Pointer(in.ptr)) + } + return in +} + // cStr reads a fixed C char array of length n into a Go string. func cStr(p *C.char, n C.uint32_t) string { if n == 0 { @@ -396,14 +411,13 @@ func (s *RCSvc) Prepare(req PrepareRequest) (*PrepareResponse, error) { if s.srv == nil { return nil, errors.New("rcserver: service closed") } + var pins runtime.Pinner + defer pins.Unpin() var principal C.rc_principal_id copy((*[32]byte)(unsafe.Pointer(&principal.id[0]))[:], req.Principal[:]) - var token C.rc_str_in - if req.ClientToken != "" { - token = strIn(req.ClientToken) - } - target := strIn(req.Target) + token := pinnedStrIn(req.ClientToken, &pins) + target := pinnedStrIn(req.Target, &pins) creq := C.rc_prepare_req{ principal: principal, op: C.uint8_t(req.Op), @@ -415,8 +429,6 @@ func (s *RCSvc) Prepare(req PrepareRequest) (*PrepareResponse, error) { client_token: token, } var cresp C.rc_prepare_resp - runtime.KeepAlive(req.ClientToken) - runtime.KeepAlive(req.Target) rc := C.rc_prepare(s.srv, &creq, &cresp) if rc != C.RC_OK { return nil, statusError(rc) @@ -510,9 +522,11 @@ func (s *RCSvc) ReadyTransfer(req ReadyRequest) (*ReadyResponse, error) { if s.srv == nil { return nil, errors.New("rcserver: service closed") } + var pins runtime.Pinner + defer pins.Unpin() var principal C.rc_principal_id copy((*[32]byte)(unsafe.Pointer(&principal.id[0]))[:], req.Principal[:]) - in := strIn(req.SessionID) + in := pinnedStrIn(req.SessionID, &pins) creq := C.rc_ready_req{ principal: principal, session_id: in, @@ -523,7 +537,6 @@ func (s *RCSvc) ReadyTransfer(req ReadyRequest) (*ReadyResponse, error) { } var cresp C.rc_ready_resp rc := C.rc_ready_transfer(s.srv, &creq, &cresp) - runtime.KeepAlive(req.SessionID) if rc != C.RC_OK { return nil, statusError(rc) } From f92d6d64b1836571e9e4c2323d223c9098b27f76 Mon Sep 17 00:00:00 2001 From: Jihyeon Gim Date: Sat, 5 Sep 2026 21:59:05 +0900 Subject: [PATCH 6/8] rdma: carry the session id in the reaped session record The session id only existed as the sessions map key; the session record itself kept an empty id string, so the terminal reap record logged an empty id for every expired, cancelled, or destroyed session. Copy the id into the record at creation time so teardown logs identify the session they describe. --- cuwrapper/rc/rc_server_abi.cpp | 4 ++++ 1 file changed, 4 insertions(+) diff --git a/cuwrapper/rc/rc_server_abi.cpp b/cuwrapper/rc/rc_server_abi.cpp index da2f1938..71e32c2a 100644 --- a/cuwrapper/rc/rc_server_abi.cpp +++ b/cuwrapper/rc/rc_server_abi.cpp @@ -613,6 +613,10 @@ int rc_prepare(rc_server *srv, const rc_prepare_req *req, srv->opts.t_prep_ms ? hipObj::v2::clockSource().nowMs() + srv->opts.t_prep_ms : 0; + /* The session record carries its own id copy: reap logging and the + * terminal teardown record read core.id, while the map key is the + * only other place the id lives. */ + rs.core.id = id; { std::lock_guard g(srv->map_mtx); rs.staging_buf = reinterpret_cast(buf); From c07f75a61212da1a9764118cf12aa3a865cde31c Mon Sep 17 00:00:00 2001 From: Jihyeon Gim Date: Sat, 5 Sep 2026 23:58:13 +0900 Subject: [PATCH 7/8] rdma: add a point-in-time session snapshot to the RC C ABI Expose rc_server_sessions_snapshot, which copies every live session into fixed rc_session_snapshot records under the map lock and invokes the callback once per record outside the lock. Each session records a monotonic creation timestamp, because the prepare/ready deadlines move as the session progresses and cannot serve as an age reference. The state byte combines the session state machine value with a reap-pending marker, so callers can distinguish sessions that are about to be reaped from healthy ones. Records whose op or target does not fit the fixed fields are skipped rather than truncated. --- cuwrapper/rc/rc_server_abi.cpp | 42 ++++++++++++++++++++++++++++++++++ cuwrapper/rc/rc_server_abi.h | 24 +++++++++++++++++++ 2 files changed, 66 insertions(+) diff --git a/cuwrapper/rc/rc_server_abi.cpp b/cuwrapper/rc/rc_server_abi.cpp index 71e32c2a..7d7c5e6a 100644 --- a/cuwrapper/rc/rc_server_abi.cpp +++ b/cuwrapper/rc/rc_server_abi.cpp @@ -43,6 +43,9 @@ struct RcSession { V2Session core; uint64_t epoch = 0; std::atomic next_nonce{1}; + /* Monotonic creation timestamp for session-age observability; + * deadlines cannot serve that role because READY moves them. */ + uint64_t created_ms = 0; /* staged metadata from finish_staging (GET) or finish_put. */ std::string etag; std::string version_id; @@ -613,6 +616,7 @@ int rc_prepare(rc_server *srv, const rc_prepare_req *req, srv->opts.t_prep_ms ? hipObj::v2::clockSource().nowMs() + srv->opts.t_prep_ms : 0; + rs.created_ms = hipObj::v2::clockSource().nowMs(); /* The session record carries its own id copy: reap logging and the * terminal teardown record read core.id, while the map key is the * only other place the id lives. */ @@ -757,6 +761,44 @@ int rc_session_info(rc_server *srv, rc_str_in session_id, return RC_OK; } +int rc_server_sessions_snapshot(rc_server *srv, rc_snapshot_cb cb, + void *ctx) { + if (!srv || !cb) return RC_E_ARG; + if (srv->closing.load()) return RC_E_INTERNAL; + + /* Fixed records so the vector can move without invalidating the + * string pointers inside; the copies own everything the callback + * reads, so delivery happens outside the map lock and cannot race + * the reaper moving or erasing entries. */ + std::vector recs; + uint64_t now = hipObj::v2::clockSource().nowMs(); + { + std::lock_guard g(srv->map_mtx); + recs.reserve(srv->sessions_map.size()); + for (const auto &kv : srv->sessions_map) { + const RcSession &s = *kv.second; + rc_session_snapshot r{}; + if (s.core.id.size() != 32) continue; /* live ids are 32 hex */ + if (s.core.op.size() >= sizeof(r.op)) continue; + if (s.core.target.size() > sizeof(r.target) - 1) continue; + memcpy(r.session_id, s.core.id.data(), 32); + r.session_id[32] = 0; + memcpy(r.op, s.core.op.data(), s.core.op.size()); + r.op[s.core.op.size()] = 0; + memcpy(r.target, s.core.target.data(), s.core.target.size()); + r.target[s.core.target.size()] = 0; + r.target_len = (uint32_t)s.core.target.size(); + r.state = (uint8_t)s.core.state; + if (s.reap_pending) r.state |= RC_SNAPSHOT_REAP_PENDING; + r.age_ms = s.created_ms ? now - s.created_ms : 0; + r.staging_bytes = s.staging_len; + recs.push_back(r); + } + } + for (const auto &r : recs) cb(&r, ctx); + return RC_OK; +} + int rc_ready_transfer(rc_server *srv, const rc_ready_req *req, rc_ready_resp *resp) { if (!srv || !req || !resp) return RC_E_ARG; diff --git a/cuwrapper/rc/rc_server_abi.h b/cuwrapper/rc/rc_server_abi.h index 4606b1a5..f852c19b 100644 --- a/cuwrapper/rc/rc_server_abi.h +++ b/cuwrapper/rc/rc_server_abi.h @@ -151,6 +151,30 @@ typedef struct { uint8_t op; } rc_session_info_resp; +/* Session observability snapshot. The record is a point-in-time copy; + * the fields are valid only inside the callback invocation. + * state combines the session state machine value with the reap-pending + * marker (RC_SNAPSHOT_REAP_PENDING) because CANCEL and expiry only set + * that marker without moving the state. */ +enum { RC_SNAPSHOT_REAP_PENDING = 0x80 }; + +typedef struct { + char session_id[33]; /* 32 hex + NUL */ + char op[4]; /* "GET" or "PUT" */ + char target[2048]; + uint32_t target_len; + uint8_t state; /* SessState value | RC_SNAPSHOT_REAP_PENDING */ + uint64_t age_ms; /* now - created_ms; 0 when the clock is absent */ + uint64_t staging_bytes; +} rc_session_snapshot; + +/* Copies every live session under the map lock into fixed records and + * invokes cb(rec, ctx) once per record outside the lock, in map order. + * A session whose op or target does not fit is skipped (record too + * small), not truncated. Returns RC_E_ARG for null arguments. */ +typedef void (*rc_snapshot_cb)(const rc_session_snapshot *rec, void *ctx); +int rc_server_sessions_snapshot(rc_server *srv, rc_snapshot_cb cb, void *ctx); + /* Lifecycle. destroy waits for active calls and the reaper. */ int rc_server_init(const rc_device_opts *opts, rc_server **out); void rc_server_destroy(rc_server *srv); From 6eb574dc1a97778cf6b89016a60aa6d4c5b7ca11 Mon Sep 17 00:00:00 2001 From: Jihyeon Gim Date: Sat, 5 Sep 2026 23:58:18 +0900 Subject: [PATCH 8/8] rdma: expose live RC sessions on the admin server Add a SessionsSnapshot view over the new C ABI entry point and serve it from the admin server as GET /rc-sessions. The admin server gains a WithAdminRoute option so an embedding binary can register extra admin routes that run with the same signature verification and admin checks as the built-in endpoints; vgwrdma registers the snapshot there when the RC feature is enabled. The route replies with the usual XML error surface so unsigned or non-admin requests get a 403 rather than a generic 500. Stub builds return a not supported error, keeping the build matrix unchanged. --- cmd/vgwrdma/main.go | 34 +++++++++++ embedgw/embedgw.go | 6 ++ rdma/rcroutes/admin_linux.go | 69 ++++++++++++++++++++++ rdma/rcroutes/admin_stub.go | 33 +++++++++++ rdma/rcserver/rcserver_cgo_linux_test.go | 15 +++++ rdma/rcserver/rcserver_linux.go | 73 ++++++++++++++++++++++++ rdma/rcserver/rcserver_stub.go | 26 +++++++++ s3api/admin-server.go | 32 +++++++++++ 8 files changed, 288 insertions(+) create mode 100644 rdma/rcroutes/admin_linux.go create mode 100644 rdma/rcroutes/admin_stub.go diff --git a/cmd/vgwrdma/main.go b/cmd/vgwrdma/main.go index 1779c312..e5d3bf4d 100644 --- a/cmd/vgwrdma/main.go +++ b/cmd/vgwrdma/main.go @@ -35,11 +35,14 @@ import ( "github.com/versity/versitygw/embedgw" "github.com/versity/versitygw/internal/netutil" "github.com/versity/versitygw/internal/rdmamode" + "github.com/versity/versitygw/metrics" "github.com/versity/versitygw/rdma" "github.com/versity/versitygw/rdma/rcroutes" "github.com/versity/versitygw/rdma/rcserver" "github.com/versity/versitygw/s3api" "github.com/versity/versitygw/s3api/middlewares" + "github.com/versity/versitygw/s3api/utils" + "github.com/versity/versitygw/s3err" ) var ( @@ -1296,6 +1299,37 @@ func runGateway(ctx context.Context, be backend.Backend) error { s3api.WithRoute("POST", "/.hipobj-rc/ready", rcAuth, rcH.Ready), s3api.WithRoute("POST", "/.hipobj-rc/cancel", rcAuth, rcH.Cancel), ) + // The RC session snapshot rides the standalone admin server + // (started when --admin-port is given) behind the same + // signature verification and admin role check as the other + // admin endpoints. + rcAdminAuth := func(ctx fiber.Ctx) error { + if err := rcVerify(ctx); err != nil { + return err + } + if err := middlewares.IsAdmin(metrics.ActionAdminListBuckets)(ctx); err != nil { + return err + } + return nil + } + // Admin routes answer with the same XML error surface as the + // built-in admin endpoints; the default fiber error handler + // would turn auth failures into a bare 500. + rcAdminRoute := func(ctx fiber.Ctx) error { + err := rcAdminAuth(ctx) + if err == nil { + err = rcH.AdminSnapshot(ctx) + } + if serr, ok := err.(s3err.S3Error); ok { + requestID, hostID := utils.EnsureRequestIDs(ctx) + return ctx.Status(serr.StatusCode()).Send( + serr.XMLBody(requestID, hostID)) + } + return err + } + cfg.AdminOptions = append(cfg.AdminOptions, + s3api.WithAdminRoute("GET", "/rc-sessions", rcAdminRoute, rcH.AdminSnapshot), + ) } else { cfg.S3Options = s3Opts } diff --git a/embedgw/embedgw.go b/embedgw/embedgw.go index cd227560..f7c6daa9 100644 --- a/embedgw/embedgw.go +++ b/embedgw/embedgw.go @@ -75,6 +75,11 @@ type Config struct { // control over the admin endpoint with optionally separate TLS certs. AdminPorts []string + // AdminOptions carries extra standalone-admin-server options from + // the embedding binary (e.g. additional admin routes). Only used + // when AdminPorts is non-empty. + AdminOptions []s3api.AdminOpt + // MaxConnections is the maximum number of concurrent TCP connections // accepted by the S3 API server. MaxConnections int @@ -873,6 +878,7 @@ func RunVersityGW(ctx context.Context, be backend.Backend, cfg *Config) error { admOpts := []s3api.AdminOpt{ s3api.WithAdminConcurrencyLimiter(cfg.AdminMaxConnections, cfg.AdminMaxRequests), } + admOpts = append(admOpts, cfg.AdminOptions...) if corsAllowOrigin != "" { admOpts = append(admOpts, s3api.WithAdminCORSAllowOrigin(corsAllowOrigin)) diff --git a/rdma/rcroutes/admin_linux.go b/rdma/rcroutes/admin_linux.go new file mode 100644 index 00000000..eb60f23e --- /dev/null +++ b/rdma/rcroutes/admin_linux.go @@ -0,0 +1,69 @@ +// Copyright 2026 Versity Software +// Copyright 2026 Gluesys Inc. and Jihyeon Gim +// This file is licensed under the Apache License, Version 2.0 +// (the "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +// Admin session snapshot route for the RC data plane. +//go:build linux && amd64 && cgo + +package rcroutes + +import ( + "github.com/gofiber/fiber/v3" + + "github.com/versity/versitygw/rdma/rcserver" +) + +// AdminSnapshot serves the RC session snapshot on the admin surface. +// The route is registered only when the RC plane is enabled and the +// standalone admin server runs; middleware before it has already +// verified the admin signature and role. +func (h *Handler) AdminSnapshot(ctx fiber.Ctx) error { + sessions, err := h.svc.SessionsSnapshot() + if err != nil { + return WriteRouteError(ctx, mapRcError(err)) + } + type sessionRecord struct { + SessionID string `json:"session_id"` + Op string `json:"op"` + Target string `json:"target"` + State string `json:"state"` + ReapPending bool `json:"reap_pending"` + AgeMs uint64 `json:"age_ms"` + StagingBytes uint64 `json:"staging_bytes"` + } + stateNames := map[uint8]string{ + rcserver.SnapshotStatePrepared: "prepared", + rcserver.SnapshotStatePublishing: "publishing", + rcserver.SnapshotStateTransferring: "transferring", + rcserver.SnapshotStateCompleting: "completing", + rcserver.SnapshotStateReaping: "reaping", + } + out := make([]sessionRecord, 0, len(sessions)) + for _, s := range sessions { + name, ok := stateNames[s.State] + if !ok { + name = "unknown" + } + out = append(out, sessionRecord{ + SessionID: s.SessionID, + Op: s.Op, + Target: s.Target, + State: name, + ReapPending: s.ReapPending, + AgeMs: s.AgeMs, + StagingBytes: s.StagingBytes, + }) + } + return ctx.JSON(fiber.Map{"sessions": out}) +} diff --git a/rdma/rcroutes/admin_stub.go b/rdma/rcroutes/admin_stub.go new file mode 100644 index 00000000..b6d2f742 --- /dev/null +++ b/rdma/rcroutes/admin_stub.go @@ -0,0 +1,33 @@ +// Copyright 2026 Versity Software +// Copyright 2026 Gluesys Inc. and Jihyeon Gim +// This file is licensed under the Apache License, Version 2.0 +// (the "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +// Admin session snapshot route (stub for platforms without RDMA +// support). +//go:build !(linux && amd64 && cgo) + +package rcroutes + +import ( + "errors" + + "github.com/gofiber/fiber/v3" +) + +// AdminSnapshot is a stub; the route is never registered without the +// RC data plane. +func (h *Handler) AdminSnapshot(ctx fiber.Ctx) error { + return fiber.NewError(fiber.StatusNotFound, + errors.New("rdma rc data plane not available").Error()) +} diff --git a/rdma/rcserver/rcserver_cgo_linux_test.go b/rdma/rcserver/rcserver_cgo_linux_test.go index 6df4e6a9..7a0c3c52 100644 --- a/rdma/rcserver/rcserver_cgo_linux_test.go +++ b/rdma/rcserver/rcserver_cgo_linux_test.go @@ -1,3 +1,18 @@ +// Copyright 2026 Versity Software +// Copyright 2026 Gluesys Inc. and Jihyeon Gim +// This file is licensed under the Apache License, Version 2.0 +// (the "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + //go:build linux && cgo package rcserver diff --git a/rdma/rcserver/rcserver_linux.go b/rdma/rcserver/rcserver_linux.go index 481170e6..2713c1c7 100644 --- a/rdma/rcserver/rcserver_linux.go +++ b/rdma/rcserver/rcserver_linux.go @@ -33,6 +33,7 @@ package rcserver // 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); +extern void rcgo_snapshot_cb(rc_session_snapshot *rec, void *ctx); */ import "C" @@ -517,6 +518,78 @@ func (s *RCSvc) SessionInfo(sessionID string, who PrincipalID) (*SessionInfo, er }, nil } +// SessionSnapshot describes a live RC session for observability. +type SessionSnapshot struct { + SessionID string + Op string + Target string + State uint8 // SessState value, may be ORed with ReapPending + ReapPending bool + AgeMs uint64 + StagingBytes uint64 +} + +// SnapshotState constants mirrored from the C ABI. +const ( + SnapshotStatePrepared = 0 + SnapshotStatePublishing = 1 + SnapshotStateTransferring = 2 + SnapshotStateCompleting = 3 + SnapshotStateReaping = 4 + SnapshotReapPending = 0x80 +) + +// snapshotReceiver is set by SessionsSnapshot for the duration of the +// C call; the fixed trampoline copies each record into it. The mutex +// covers the case of concurrent snapshot calls. +var ( + snapshotMu sync.Mutex + snapshotSink *[]SessionSnapshot +) + +//export rcgo_snapshot_cb +func rcgo_snapshot_cb(rec *C.rc_session_snapshot, _ unsafe.Pointer) { + sink := snapshotSink + if sink == nil || rec == nil { + return + } + st := uint8(rec.state) + *sink = append(*sink, SessionSnapshot{ + SessionID: C.GoString(&rec.session_id[0]), + Op: C.GoString(&rec.op[0]), + Target: C.GoStringN(&rec.target[0], C.int(rec.target_len)), + State: st &^ SnapshotReapPending, + ReapPending: st&SnapshotReapPending != 0, + AgeMs: uint64(rec.age_ms), + StagingBytes: uint64(rec.staging_bytes), + }) +} + +// SessionsSnapshot copies every live session into Go-owned records. +// The C side snapshots under its map lock and delivers the copies +// outside it, so the callback cannot block the data plane. +func (s *RCSvc) SessionsSnapshot() ([]SessionSnapshot, error) { + if s.srv == nil { + return nil, errors.New("rcserver: service closed") + } + if !s.TryEnter() { + return nil, ErrInternal + } + defer s.Leave() + + snapshotMu.Lock() + defer snapshotMu.Unlock() + var out []SessionSnapshot + snapshotSink = &out + defer func() { snapshotSink = nil }() + rc := C.rc_server_sessions_snapshot(s.srv, + (*[0]byte)(C.rcgo_snapshot_cb), nil) + if rc != C.RC_OK { + return nil, statusError(rc) + } + return out, nil +} + // ReadyTransfer runs the data phase (READY). func (s *RCSvc) ReadyTransfer(req ReadyRequest) (*ReadyResponse, error) { if s.srv == nil { diff --git a/rdma/rcserver/rcserver_stub.go b/rdma/rcserver/rcserver_stub.go index 820858ca..99e6b220 100644 --- a/rdma/rcserver/rcserver_stub.go +++ b/rdma/rcserver/rcserver_stub.go @@ -111,6 +111,27 @@ type SessionInfo struct { Target string } +// SessionSnapshot describes a live RC session for observability. +type SessionSnapshot struct { + SessionID string + Op string + Target string + State uint8 + ReapPending bool + AgeMs uint64 + StagingBytes uint64 +} + +// Snapshot state values mirrored from the cgo ABI. +const ( + SnapshotStatePrepared = 0 + SnapshotStatePublishing = 1 + SnapshotStateTransferring = 2 + SnapshotStateCompleting = 3 + SnapshotStateReaping = 4 + SnapshotReapPending = 0x80 +) + // RCSvc owns the RC session server (stub). type RCSvc struct{} @@ -152,6 +173,11 @@ func (s *RCSvc) SessionInfo(sessionID string, who PrincipalID) (*SessionInfo, er return nil, errNotSupported } +// SessionsSnapshot is a stub. +func (s *RCSvc) SessionsSnapshot() ([]SessionSnapshot, error) { + return nil, errNotSupported +} + // ReadyTransfer is a stub. func (s *RCSvc) ReadyTransfer(req ReadyRequest) (*ReadyResponse, error) { return nil, errNotSupported diff --git a/s3api/admin-server.go b/s3api/admin-server.go index d06b98e5..3be14529 100644 --- a/s3api/admin-server.go +++ b/s3api/admin-server.go @@ -42,6 +42,15 @@ type S3AdminServer struct { maxConnections int maxRequests int socketPerm os.FileMode + extraRoutes []adminRouteMount +} + +// adminRouteMount is a route registered on the admin app after the +// built-in admin router ran. +type adminRouteMount struct { + method string + path string + handlers []fiber.Handler } func NewAdminServer(be backend.Backend, root middlewares.RootUserConfig, region string, iam auth.IAMService, l s3log.AuditLogger, ctrl controllers.S3ApiController, opts ...AdminOpt) *S3AdminServer { @@ -95,6 +104,14 @@ func NewAdminServer(be backend.Backend, root middlewares.RootUserConfig, region server.router.Init(app, be, iam, l, root, region, server.debug, server.corsAllowOrigin) + for _, r := range server.extraRoutes { + args := make([]any, 0, len(r.handlers)) + for _, h := range r.handlers { + args = append(args, h) + } + app.Add([]string{r.method}, r.path, args[0], args[1:]...) + } + return server } @@ -136,6 +153,21 @@ func WithAdminSocketPerm(perm os.FileMode) AdminOpt { return func(s *S3AdminServer) { s.socketPerm = perm } } +// WithAdminRoute registers a route on the standalone admin server, +// after the built-in admin routes and their middleware chain. Use it +// for admin-surface endpoints that do not fit the S3 admin controller +// shape. +func WithAdminRoute(method, path string, handlers ...fiber.Handler) AdminOpt { + return func(s *S3AdminServer) { + copied := append([]fiber.Handler(nil), handlers...) + s.extraRoutes = append(s.extraRoutes, adminRouteMount{ + method: method, + path: path, + handlers: copied, + }) + } +} + // ServeMultiPort creates listeners for multiple port specifications and serves // on all of them simultaneously. This supports listening on multiple ports and/or // addresses (e.g., [":8080", "localhost:8081"]).