diff --git a/cmd/vgwrdma/main.go b/cmd/vgwrdma/main.go index 29de5ec1..22e02103 100644 --- a/cmd/vgwrdma/main.go +++ b/cmd/vgwrdma/main.go @@ -16,13 +16,12 @@ package main import ( "context" + "errors" "fmt" "log" - "math" "net/http" _ "net/http/pprof" "os" - "strings" "sync" "github.com/gofiber/fiber/v3" @@ -35,6 +34,7 @@ import ( "github.com/versity/versitygw/debuglogger" "github.com/versity/versitygw/embedgw" "github.com/versity/versitygw/internal/netutil" + "github.com/versity/versitygw/internal/rdmamode" "github.com/versity/versitygw/rdma" "github.com/versity/versitygw/rdma/rcroutes" "github.com/versity/versitygw/rdma/rcserver" @@ -111,6 +111,7 @@ var ( socketPerm string rdmaIP string rcGidHint string + rdmaRCEnable bool rdmaPort uint poolBufSize int poolBufCount int @@ -242,11 +243,16 @@ documentation can be found in the GitHub wiki.`, ctx.IsSet("rdma-cq-depth") || ctx.IsSet("rdma-retry-count") - // Only commands that actually start a gateway need --rdma-ip; admin, - // utils, help, and version subcommands print output and exit without - // ever calling runGateway. - if gatewayCommands[ctx.Args().First()] && strings.TrimSpace(rdmaIP) == "" { - return fmt.Errorf("rdma-ip is required") + // Only commands that actually start a gateway need an + // RDMA path; admin, utils, help, and version + // subcommands print output and exit without ever + // calling runGateway. Either the cuObject v1 address + // or the RC v2 flag enables one. + if gatewayCommands[ctx.Args().First()] { + v1On, v2On := rdmamode.Mode(rdmaIP, rdmaRCEnable) + if !v1On && !v2On { + return fmt.Errorf("either --rdma-ip or --rdma-rc-enable is required") + } } return nil }, @@ -886,6 +892,12 @@ func initFlags() []cli.Flag { EnvVars: []string{"VGW_RC_GID_HINT"}, Destination: &rcGidHint, }, + &cli.BoolFlag{ + Name: "rdma-rc-enable", + Usage: "enable the hipobj-rc-v2 RC control routes and data plane (independent of --rdma-ip)", + EnvVars: []string{"VGW_RDMA_RC_ENABLE"}, + Destination: &rdmaRCEnable, + }, &cli.UintFlag{ Name: "rdma-port", Usage: "port for RDMA listener", @@ -967,27 +979,36 @@ func runGateway(ctx context.Context, be backend.Backend) error { if gwcli.CopyObjectThreshold < 1 { return fmt.Errorf("copy-object-threshold must be positive") } - if rdmaPort > 65535 { - return fmt.Errorf("rdma-port %d is out of range (0-65535)", rdmaPort) - } - if rdmaRetryCount > 7 { - return fmt.Errorf("rdma-retry-count %d is out of range (0-7)", rdmaRetryCount) - } - if poolBufSize <= 0 { - return fmt.Errorf("pool-buf-size %d must be positive", poolBufSize) - } - if poolBufCount <= 0 { - return fmt.Errorf("pool-buf-count %d must be positive", poolBufCount) - } - if rdmaTunablesSet && rdmaNumDCIs <= 0 { - return fmt.Errorf("rdma-num-dcis %d must be positive", rdmaNumDCIs) - } - if rdmaTunablesSet && rdmaCQDepth > math.MaxUint32 { - return fmt.Errorf("rdma-cq-depth %d exceeds maximum %d", rdmaCQDepth, uint32(math.MaxUint32)) + v1On, v2On := rdmamode.Mode(rdmaIP, rdmaRCEnable) + if v1On { + // v1-only settings are irrelevant when the cuObject + // backend is not running; stale environment values must + // not block v2-only startup. + v1s := rdmamode.V1Settings{ + Port: rdmaPort, + RetryCount: rdmaRetryCount, + PoolBufSize: poolBufSize, + PoolBufCnt: poolBufCount, + TunablesSet: rdmaTunablesSet, + NumDCIs: rdmaNumDCIs, + CQDepth: rdmaCQDepth, + } + if msg := rdmamode.V1ValidationError(v1s); msg != "" { + return errors.New(msg) + } } + // The gateway command owns the input backend from here on: + // every later failure path must close the whole chain it has + // built so far, exactly once. The closure reads be at run + // time, so the rollback covers the v1 chain and the RC + // wrapper once those layers are added; the once guard keeps + // the shared close with the RunVersityGW lifecycle single. + be = rdmamode.WrapShutdownOnce(be) + defer func() { be.Shutdown() }() + var s3Opts []s3api.Option - if rdmaIP != "" { + if v1On { rdma.ConfigureTelemetry(debug) var tunables *rdma.RDMATunables @@ -1010,7 +1031,11 @@ func runGateway(ctx context.Context, be backend.Backend) error { if err != nil { return err } - be = cuserverBackend + // The v1 layer has no idempotent Shutdown of its own; put + // the once owner around the completed chain so the + // rollback closure and the RunVersityGW lifecycle both + // stop at the same single close. + be = rdmamode.WrapShutdownOnce(cuserverBackend) s3Opts = append(s3Opts, s3api.WithMiddleware("/", cumiddleware.CuObjMiddleware)) } @@ -1116,11 +1141,13 @@ func runGateway(ctx context.Context, be backend.Backend) error { BuildTime: BuildTime, } - if rdmaIP != "" { + if v2On { // RC data plane: build the IAM service the gateway will // use so the control routes authenticate against the // same account store, then start the session server and - // mount the hipobj-rc-v2 routes on the S3 port. + // mount the hipobj-rc-v2 routes on the S3 port. Enabled + // by --rdma-rc-enable independently of the cuObject + // --rdma-ip flag. iamSvc, err := auth.New(cfg.IamOpts()) if err != nil { return fmt.Errorf("setup iam for rdma routes: %w", err) @@ -1148,7 +1175,11 @@ func runGateway(ctx context.Context, be backend.Backend) error { if err != nil { return err } - defer rcSvc.Close() + // Close the RC service before the backend chain so the + // control routes drain first; the wrapper also gives the + // rollback closure and the RunVersityGW lifecycle a + // single, ordered owner of both steps. + be = rdmamode.WrapBackendShutdownAfterRC(be, rcSvc) rcVerify := middlewares.VerifyV4Signature( middlewares.RootUserConfig{ diff --git a/internal/rdmamode/rdmamode.go b/internal/rdmamode/rdmamode.go new file mode 100644 index 00000000..a5b40aca --- /dev/null +++ b/internal/rdmamode/rdmamode.go @@ -0,0 +1,121 @@ +// Copyright 2026 Versity Software +// 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. + +// Package rdmamode resolves which RDMA paths a gateway runs and +// orders their shutdown. It is deliberately free of cgo so the +// behavior is testable on any build platform. +package rdmamode + +import ( + "fmt" + "math" + "strings" + "sync" + + "github.com/versity/versitygw/backend" +) + +// Mode resolves which RDMA paths run: the cuObject v1 backend +// follows the legacy --rdma-ip contract, and the hipobj-rc-v2 +// control routes follow --rdma-rc-enable. Neither flag implies +// the other. +func Mode(rdmaIP string, rcEnable bool) (v1, v2 bool) { + return strings.TrimSpace(rdmaIP) != "", rcEnable +} + +// V1Settings carries the cuObject v1 tunings that are validated +// only while the v1 backend runs. +type V1Settings struct { + Port uint + RetryCount uint + PoolBufSize int + PoolBufCnt int + TunablesSet bool + NumDCIs int + CQDepth uint +} + +// V1ValidationError describes the first invalid v1 setting, or +// the empty string when every setting is valid. Stale v1 values +// in the environment must not block v2-only or plain-S3 +// startup, so the gateway consults this only when v1 is on. +func V1ValidationError(s V1Settings) string { + switch { + case s.Port > 65535: + return fmt.Sprintf("rdma-port %d is out of range (0-65535)", s.Port) + case s.RetryCount > 7: + return fmt.Sprintf("rdma-retry-count %d is out of range (0-7)", s.RetryCount) + case s.PoolBufSize <= 0: + return fmt.Sprintf("pool-buf-size %d must be positive", s.PoolBufSize) + case s.PoolBufCnt <= 0: + return fmt.Sprintf("pool-buf-count %d must be positive", s.PoolBufCnt) + case s.TunablesSet && s.NumDCIs <= 0: + return fmt.Sprintf("rdma-num-dcis %d must be positive", s.NumDCIs) + case s.TunablesSet && s.CQDepth > math.MaxUint32: + return fmt.Sprintf("rdma-cq-depth %d exceeds the 32-bit limit", s.CQDepth) + default: + return "" + } +} + +// Closer is the close operation of the RC session service. It is +// idempotent. +type Closer interface{ Close() } + +// BackendShutdownAfterRC forwards a backend and closes the RC +// service before the wrapped backend shuts down. The RC handlers +// reference the backend and IAM service, so the RC service must +// stop accepting and drain before either dependency is closed by +// the gateway lifecycle. The whole shutdown, including the +// delegated backend call, runs exactly once: several closers +// (the gateway lifecycle and a deferred rollback in the startup +// path) may call Shutdown on the same instance, and the wrapped +// backends are not guaranteed to be idempotent. +type BackendShutdownAfterRC struct { + backend.Backend + rc Closer + closed sync.Once +} + +// Shutdown closes the RC service, then the wrapped backend, once. +func (b *BackendShutdownAfterRC) Shutdown() { + b.closed.Do(func() { + b.rc.Close() + b.Backend.Shutdown() + }) +} + +// ShutdownOnceBackend makes a backend Shutdown idempotent. The +// gateway startup path defers a rollback close while the gateway +// lifecycle also shuts its backend down; a bare backend has no +// idempotency guarantee of its own. +type ShutdownOnceBackend struct { + backend.Backend + once sync.Once +} + +// Shutdown closes the wrapped backend at most once. +func (b *ShutdownOnceBackend) Shutdown() { + b.once.Do(func() { b.Backend.Shutdown() }) +} + +// WrapShutdownOnce returns be with an idempotent Shutdown. +func WrapShutdownOnce(be backend.Backend) backend.Backend { + return &ShutdownOnceBackend{Backend: be} +} + +// WrapBackendShutdownAfterRC returns be with a Shutdown that +// closes rc first. +func WrapBackendShutdownAfterRC(be backend.Backend, rc Closer) backend.Backend { + return &BackendShutdownAfterRC{Backend: be, rc: rc} +} diff --git a/internal/rdmamode/rdmamode_test.go b/internal/rdmamode/rdmamode_test.go new file mode 100644 index 00000000..abeaa515 --- /dev/null +++ b/internal/rdmamode/rdmamode_test.go @@ -0,0 +1,258 @@ +// Copyright 2026 Versity Software +// 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. + +package rdmamode + +import ( + "math" + "runtime" + "sync" + "testing" + "time" + + "github.com/versity/versitygw/backend" +) + +func TestModeMatrix(t *testing.T) { + tests := []struct { + name string + rdmaIP string + rcEnable bool + wantV1, wantV2 bool + }{ + {"neither", "", false, false, false}, + {"v1 only", "192.0.2.1", false, true, false}, + {"v2 only", "", true, false, true}, + {"both", "192.0.2.1", true, true, true}, + {"v1 ip with whitespace", " ", false, false, false}, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + v1, v2 := Mode(tt.rdmaIP, tt.rcEnable) + if v1 != tt.wantV1 || v2 != tt.wantV2 { + t.Fatalf("Mode(%q, %v) = (%v, %v), want (%v, %v)", + tt.rdmaIP, tt.rcEnable, + v1, v2, tt.wantV1, tt.wantV2) + } + }) + } +} + +func TestV1ValidationCQDepthBoundary(t *testing.T) { + // The v1 tunable narrows to uint32; the boundary itself must + // pass and the first value beyond it must fail. + base := V1Settings{ + Port: 19100, RetryCount: 2, PoolBufSize: 1024, PoolBufCnt: 16, + NumDCIs: 8, TunablesSet: true, + } + base.CQDepth = math.MaxUint32 + if msg := V1ValidationError(base); msg != "" { + t.Fatalf("MaxUint32 rejected: %q", msg) + } + base.CQDepth = math.MaxUint32 + 1 + if msg := V1ValidationError(base); msg == "" { + t.Fatal("MaxUint32+1 accepted") + } +} + +func TestV1ValidationError(t *testing.T) { + if msg := V1ValidationError(V1Settings{ + Port: 99999, RetryCount: 99, PoolBufSize: -1, + }); msg == "" { + t.Fatal("expected an error for invalid settings") + } + if msg := V1ValidationError(V1Settings{ + Port: 19100, RetryCount: 2, PoolBufSize: 1024, PoolBufCnt: 16, + }); msg != "" { + t.Fatalf("valid settings reported %q", msg) + } +} + +// sequenceRC records when Close starts and finishes, so tests +// can prove the backend waits for it. +type sequenceRC struct { + mu sync.Mutex + started int + finished int + release chan struct{} +} + +func (r *sequenceRC) Close() { + r.mu.Lock() + r.started++ + r.mu.Unlock() + <-r.release + r.mu.Lock() + r.finished++ + r.mu.Unlock() +} + +func (r *sequenceRC) counts() (started, finished int) { + r.mu.Lock() + defer r.mu.Unlock() + return r.started, r.finished +} + +// sequenceBackend records Shutdown calls and whether the RC close +// had finished when each ran. +type sequenceBackend struct { + backend.BackendUnsupported + rc *sequenceRC + mu sync.Mutex + shutdowns int + rcDone bool +} + +func (b *sequenceBackend) Shutdown() { + started, finished := b.rc.counts() + b.mu.Lock() + defer b.mu.Unlock() + b.shutdowns++ + b.rcDone = started >= 1 && finished >= 1 +} + +func TestBackendShutdownClosesRCFirst(t *testing.T) { + rc := &sequenceRC{release: make(chan struct{})} + be := &sequenceBackend{rc: rc} + wrapped := WrapBackendShutdownAfterRC(be, rc) + + done := make(chan struct{}) + go func() { + wrapped.Shutdown() + close(done) + }() + + // While RC close is blocked, the backend must not shut down. + // Wait for the close to start; goroutine scheduling needs a + // moment even though the channel blocks it from finishing. + started := make(chan struct{}) + go func() { + for { + s, _ := rc.counts() + if s >= 1 { + close(started) + return + } + runtime.Gosched() + } + }() + select { + case <-started: + case <-time.After(5 * time.Second): + t.Fatal("RC close never started") + } + be.mu.Lock() + early := be.shutdowns + be.mu.Unlock() + if early != 0 { + t.Fatalf("backend shut down %d times before RC close finished", early) + } + + close(rc.release) + <-done + + be.mu.Lock() + shutdowns, rcDone := be.shutdowns, be.rcDone + be.mu.Unlock() + if shutdowns != 1 { + t.Fatalf("backend shutdown %d times, want 1", shutdowns) + } + if !rcDone { + t.Fatal("backend shut down before RC close finished") + } +} + +func TestBackendShutdownExactlyOnce(t *testing.T) { + // The gateway lifecycle and a startup rollback may both call + // Shutdown on the wrapped backend; every step must run once. + rc := &sequenceRC{release: make(chan struct{})} + close(rc.release) + be := &sequenceBackend{rc: rc} + wrapped := WrapBackendShutdownAfterRC(be, rc) + + var wg sync.WaitGroup + for i := 0; i < 4; i++ { + wg.Add(1) + go func() { + defer wg.Done() + wrapped.Shutdown() + }() + } + wg.Wait() + + s, f := rc.counts() + if s != 1 || f != 1 { + t.Fatalf("RC close started %d, finished %d; want 1, 1", s, f) + } + be.mu.Lock() + shutdowns := be.shutdowns + be.mu.Unlock() + if shutdowns != 1 { + t.Fatalf("backend shutdown %d times, want exactly 1", shutdowns) + } +} + +func TestShutdownOnceBackendClosesOnce(t *testing.T) { + // A rollback defer and the gateway lifecycle can both call + // Shutdown on a bare backend; the once wrapper must collapse + // them into a single close. + rc := &sequenceRC{release: make(chan struct{})} + close(rc.release) + be := &sequenceBackend{rc: rc} + wrapped := WrapShutdownOnce(be) + + var wg sync.WaitGroup + for i := 0; i < 3; i++ { + wg.Add(1) + go func() { + defer wg.Done() + wrapped.Shutdown() + }() + } + wg.Wait() + + be.mu.Lock() + shutdowns := be.shutdowns + be.mu.Unlock() + if shutdowns != 1 { + t.Fatalf("backend shutdown %d times, want 1", shutdowns) + } +} + +func TestRCWrapperChainsOntoOnceBackend(t *testing.T) { + // The gateway wraps the base backend once, then the RC + // wrapper on top. The RC close runs before the delegated + // shutdown, and repeated calls through either layer reach + // the base backend exactly once. + rc := &sequenceRC{release: make(chan struct{})} + close(rc.release) + base := &sequenceBackend{rc: rc} + onceWrapped := WrapShutdownOnce(base) + rcWrapped := WrapBackendShutdownAfterRC(onceWrapped, rc) + + rcWrapped.Shutdown() + rcWrapped.Shutdown() + onceWrapped.Shutdown() + + s, f := rc.counts() + if s != 1 || f != 1 { + t.Fatalf("RC close started %d, finished %d; want 1, 1", s, f) + } + base.mu.Lock() + shutdowns := base.shutdowns + base.mu.Unlock() + if shutdowns != 1 { + t.Fatalf("base backend shutdown %d times, want 1", shutdowns) + } +}