mirror of
https://github.com/versity/versitygw.git
synced 2026-09-23 16:34:18 +00:00
fix: remove legacy rdma header descriptors
These headers are not used by the cuda cuObject client/server interactions, so are not needed in this server implementation.
This commit is contained in:
@@ -1,6 +1,6 @@
|
||||
# syntax=docker/dockerfile:1.7
|
||||
|
||||
ARG CUDA_TAG=13.3.0-devel-rockylinux9
|
||||
ARG CUDA_TAG=13.4.1-devel-rockylinux9
|
||||
FROM --platform=linux/amd64 nvidia/cuda:${CUDA_TAG}
|
||||
|
||||
# Install build prerequisites and cuObject server development packages.
|
||||
|
||||
@@ -121,9 +121,7 @@ func (cs *CuServer) PutObject(ctx context.Context, po s3response.PutObjectInput)
|
||||
// RDMA-accelerated path
|
||||
size, ok := cumiddleware.GetRDMASize(ctx)
|
||||
if !ok || size <= 0 {
|
||||
// Size comes from the legacy 3-header scheme or, for the combined
|
||||
// token scheme, the standard Content-Length header — neither was usable.
|
||||
return s3response.PutObjectOutput{}, fmt.Errorf("cuserver: RDMA PUT requires a positive size via the %s header or a standard Content-Length header", cumiddleware.HeaderRDMASize)
|
||||
return s3response.PutObjectOutput{}, fmt.Errorf("cuserver: RDMA PUT requires a positive Content-Length or RDMA token buffer size")
|
||||
}
|
||||
if size > int64(rdma.MaxTransferSize) {
|
||||
return s3response.PutObjectOutput{}, fmt.Errorf("cuserver: object size %d exceeds RDMA max %d", size, rdma.MaxTransferSize)
|
||||
|
||||
+7
-43
@@ -47,21 +47,11 @@ const (
|
||||
)
|
||||
|
||||
// Header names used by the cuObject client to pass RDMA descriptor info.
|
||||
//
|
||||
// HeaderRDMADescr/HeaderRDMASize/HeaderRDMARemoteAddr are a legacy 3-header
|
||||
// scheme used only by this repo's own test tools (cmd/cuobjtest,
|
||||
// cmd/rdmatest). The real cuObject/minio-cpp SDK does not send these; it
|
||||
// sends a single combined HeaderRDMAToken instead. Both schemes are
|
||||
// supported so existing test tooling keeps working.
|
||||
const (
|
||||
HeaderRDMADescr = "X-CuObj-RDMA-Descr"
|
||||
HeaderRDMASize = "X-CuObj-Content-Length"
|
||||
HeaderRDMARemoteAddr = "X-CuObj-Remote-Buf-Start"
|
||||
|
||||
// HeaderRDMAToken is the single combined header sent by the real
|
||||
// cuObject client SDK (e.g. minio-cpp). Its value is a colon-delimited
|
||||
// token per the cuObj RDMA descriptor protocol. The whole token is
|
||||
// passed through verbatim to the RDMA backend as the descriptor.
|
||||
// cuObject client. Its value is a colon-delimited token per the
|
||||
// cuObj RDMA descriptor protocol. The whole token is passed through
|
||||
// verbatim to the RDMA backend as the descriptor.
|
||||
//
|
||||
// This is the canonical definition of the wire format; the encoder in
|
||||
// cuwrapper/rdma_host_client_wrapper.cpp (build_token) must stay in
|
||||
@@ -99,16 +89,15 @@ const (
|
||||
// the fasthttp request context so the backend can retrieve them via the
|
||||
// GetRDMA* helper functions.
|
||||
//
|
||||
// If neither the legacy descriptor header nor the combined RDMA token
|
||||
// header is present, the request passes through unchanged — the backend
|
||||
// If the combined RDMA token header is not present, the request passes
|
||||
// through unchanged — the backend
|
||||
// will use the normal (non-RDMA) code path.
|
||||
//
|
||||
// If a descriptor is present but required fields are malformed, a 400 Bad
|
||||
// Request is returned immediately.
|
||||
func CuObjMiddleware(ctx fiber.Ctx) error {
|
||||
descr := ctx.Get(HeaderRDMADescr)
|
||||
token := ctx.Get(HeaderRDMAToken)
|
||||
if descr == "" && token == "" {
|
||||
if token == "" {
|
||||
return ctx.Next()
|
||||
}
|
||||
|
||||
@@ -116,32 +105,7 @@ func CuObjMiddleware(ctx fiber.Ctx) error {
|
||||
// the ctx.RequestCtx() call the versitygw controller uses to invoke the backend.
|
||||
rctx := ctx.RequestCtx()
|
||||
|
||||
if descr != "" {
|
||||
// Legacy 3-header scheme.
|
||||
rctx.SetUserValue(localKeyRDMADescr, descr)
|
||||
|
||||
if sizeStr := ctx.Get(HeaderRDMASize); sizeStr != "" {
|
||||
size, err := strconv.ParseInt(sizeStr, 10, 64)
|
||||
if err != nil || size <= 0 {
|
||||
return fiber.NewError(fiber.StatusBadRequest,
|
||||
HeaderRDMASize+": must be a positive integer")
|
||||
}
|
||||
rctx.SetUserValue(localKeyRDMASize, size)
|
||||
}
|
||||
|
||||
if addrStr := ctx.Get(HeaderRDMARemoteAddr); addrStr != "" {
|
||||
addr, err := strconv.ParseUint(addrStr, 10, 64)
|
||||
if err != nil {
|
||||
return fiber.NewError(fiber.StatusBadRequest,
|
||||
HeaderRDMARemoteAddr+": must be a non-negative integer")
|
||||
}
|
||||
rctx.SetUserValue(localKeyRDMARemoteStart, addr)
|
||||
}
|
||||
|
||||
return ctx.Next()
|
||||
}
|
||||
|
||||
// Combined token scheme (real cuObject client SDK). The descriptor
|
||||
// The descriptor
|
||||
// passed to the RDMA backend is the raw token string; the remote base
|
||||
// address is parsed from the token's first field, and the transfer size
|
||||
// normally comes from the standard Content-Length header (RDMA replaces
|
||||
|
||||
@@ -73,30 +73,6 @@ func TestCuObjMiddlewareNoHeadersPassesThrough(t *testing.T) {
|
||||
assert.Equal(t, http.StatusOK, resp.StatusCode)
|
||||
}
|
||||
|
||||
func TestCuObjMiddlewareLegacyHeaders(t *testing.T) {
|
||||
app := fiber.New()
|
||||
app.Use("*", CuObjMiddleware)
|
||||
app.Post("/", func(ctx fiber.Ctx) error {
|
||||
rctx := ctx.RequestCtx()
|
||||
descr, ok := GetRDMADescriptor(rctx)
|
||||
assert.True(t, ok)
|
||||
assert.Equal(t, "deadbeef", descr)
|
||||
size, ok := GetRDMASize(rctx)
|
||||
assert.True(t, ok)
|
||||
assert.Equal(t, int64(4096), size)
|
||||
assert.Equal(t, uint64(4660), GetRDMARemoteStart(rctx))
|
||||
return ctx.SendStatus(http.StatusOK)
|
||||
})
|
||||
|
||||
req := httptest.NewRequest(http.MethodPost, "/", nil)
|
||||
req.Header.Set(HeaderRDMADescr, "deadbeef")
|
||||
req.Header.Set(HeaderRDMASize, "4096")
|
||||
req.Header.Set(HeaderRDMARemoteAddr, "4660")
|
||||
resp, err := app.Test(req)
|
||||
assert.NoError(t, err)
|
||||
assert.Equal(t, http.StatusOK, resp.StatusCode)
|
||||
}
|
||||
|
||||
func TestCuObjMiddlewareTokenWithContentLength(t *testing.T) {
|
||||
app := fiber.New()
|
||||
app.Use("*", CuObjMiddleware)
|
||||
@@ -272,21 +248,3 @@ func TestRCTokenScheme16HexNoColonStill400(t *testing.T) {
|
||||
assert.NoError(t, err)
|
||||
assert.Equal(t, http.StatusBadRequest, resp.StatusCode)
|
||||
}
|
||||
|
||||
func TestLegacyDescriptorTakesPrecedenceOverRCToken(t *testing.T) {
|
||||
// When both the legacy descriptor header and an RC-shaped combined
|
||||
// token are present, the legacy path handles the request and the RC
|
||||
// gate must not fire.
|
||||
app, reached := newTestApp(t)
|
||||
req := httptest.NewRequest(http.MethodPost, "/", nil)
|
||||
req.Header.Set(HeaderRDMADescr, "legacy-descriptor")
|
||||
req.Header.Set(HeaderRDMAToken, rcToken(""))
|
||||
resp, err := app.Test(req)
|
||||
assert.NoError(t, err)
|
||||
assert.Equal(t, http.StatusOK, resp.StatusCode)
|
||||
select {
|
||||
case <-reached:
|
||||
default:
|
||||
t.Fatal("handler should have been reached via the legacy path")
|
||||
}
|
||||
}
|
||||
|
||||
@@ -43,11 +43,10 @@ import (
|
||||
// Session owns a host buffer registered for cuObject RDMA transfers via the
|
||||
// host-memory RDMA client. Session methods are not safe for concurrent use.
|
||||
type Session struct {
|
||||
client *hostclient.Client
|
||||
buf []byte
|
||||
token string
|
||||
size int
|
||||
remoteStart uint64
|
||||
client *hostclient.Client
|
||||
buf []byte
|
||||
token string
|
||||
size int
|
||||
}
|
||||
|
||||
// NewSession creates a host-memory RDMA session for a fixed transfer size.
|
||||
@@ -81,11 +80,10 @@ func NewSession(size int) (*Session, error) {
|
||||
}
|
||||
|
||||
return &Session{
|
||||
client: client,
|
||||
buf: buf,
|
||||
token: token,
|
||||
size: size,
|
||||
remoteStart: client.BufferAddr(),
|
||||
client: client,
|
||||
buf: buf,
|
||||
token: token,
|
||||
size: size,
|
||||
}, nil
|
||||
}
|
||||
|
||||
@@ -107,7 +105,7 @@ func (s *Session) Upload(base *s3lib.Client, bucket, key string, src []byte) err
|
||||
return fmt.Errorf("upload size mismatch: got %d bytes, want %d", len(src), s.size)
|
||||
}
|
||||
copy(s.buf, src)
|
||||
return doPut(base, bucket, key, int64(s.size), s.token, s.remoteStart)
|
||||
return doPut(base, bucket, key, int64(s.size), s.token)
|
||||
}
|
||||
|
||||
// Download performs a GET; the gateway RDMA-writes into the registered host
|
||||
@@ -119,7 +117,7 @@ func (s *Session) Download(base *s3lib.Client, bucket, key string, dst []byte) e
|
||||
for i := range s.buf {
|
||||
s.buf[i] = 0
|
||||
}
|
||||
if err := doGet(base, bucket, key, int64(s.size), s.token, s.remoteStart); err != nil {
|
||||
if err := doGet(base, bucket, key, int64(s.size), s.token); err != nil {
|
||||
return err
|
||||
}
|
||||
copy(dst, s.buf)
|
||||
|
||||
@@ -38,10 +38,9 @@ const (
|
||||
|
||||
// Session owns a CUDA buffer registered for cuObject RDMA token exchange.
|
||||
type Session struct {
|
||||
ctx *C.cuobj_client_ctx_t
|
||||
gpuBuf unsafe.Pointer
|
||||
size int
|
||||
remoteStart uint64
|
||||
ctx *C.cuobj_client_ctx_t
|
||||
gpuBuf unsafe.Pointer
|
||||
size int
|
||||
}
|
||||
|
||||
// NewSession creates a GPU-backed cuObject session for a fixed transfer size.
|
||||
@@ -82,7 +81,6 @@ func NewSession(size int) (*Session, error) {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
s.remoteStart = uint64(C.cuobj_client_ptr_to_u64(s.gpuBuf))
|
||||
return s, nil
|
||||
}
|
||||
|
||||
@@ -102,7 +100,7 @@ func (s *Session) Close() {
|
||||
}
|
||||
|
||||
// Upload copies src into the session GPU buffer and performs a PUT operation
|
||||
// using cuObject RDMA headers.
|
||||
// using a cuObject RDMA token.
|
||||
// src length must exactly match the size passed to NewSession.
|
||||
func (s *Session) Upload(base *s3lib.Client, bucket, key string, src []byte) error {
|
||||
if len(src) != s.size {
|
||||
@@ -116,7 +114,7 @@ func (s *Session) Upload(base *s3lib.Client, bucket, key string, src []byte) err
|
||||
return err
|
||||
}
|
||||
defer s.putToken(token)
|
||||
return doPut(base, bucket, key, int64(s.size), C.GoString(token), s.remoteStart)
|
||||
return doPut(base, bucket, key, int64(s.size), C.GoString(token))
|
||||
}
|
||||
|
||||
// Download performs a GET operation into the session GPU buffer and copies the
|
||||
@@ -134,7 +132,7 @@ func (s *Session) Download(base *s3lib.Client, bucket, key string, dst []byte) e
|
||||
return err
|
||||
}
|
||||
defer s.putToken(token)
|
||||
if err := doGet(base, bucket, key, int64(s.size), C.GoString(token), s.remoteStart); err != nil {
|
||||
if err := doGet(base, bucket, key, int64(s.size), C.GoString(token)); err != nil {
|
||||
return err
|
||||
}
|
||||
return s.copyD2H(dst)
|
||||
|
||||
@@ -16,7 +16,7 @@
|
||||
|
||||
// This file holds the S3 request plumbing shared by the GPU (session_linux.go)
|
||||
// and host-memory (session_host_linux.go) session implementations: both issue
|
||||
// a zero-byte PUT/GET carrying the cuObject RDMA descriptor headers, and let
|
||||
// a zero-byte PUT/GET carrying the cuObject RDMA token header, and let
|
||||
// the gateway perform the actual transfer via RDMA.
|
||||
package cuobjclient
|
||||
|
||||
@@ -37,10 +37,10 @@ import (
|
||||
"github.com/versity/versitygw/cumiddleware"
|
||||
)
|
||||
|
||||
func doPut(base *s3lib.Client, bucket, key string, size int64, descr string, remoteStart uint64) error {
|
||||
func doPut(base *s3lib.Client, bucket, key string, size int64, token string) error {
|
||||
var replyStatus string
|
||||
var transferredHeader string
|
||||
c := withRDMAHeaders(base, descr, size, remoteStart, &replyStatus, &transferredHeader)
|
||||
c := withRDMAHeaders(base, token, &replyStatus, &transferredHeader)
|
||||
_, err := c.PutObject(context.Background(), &s3lib.PutObjectInput{
|
||||
Bucket: aws.String(bucket),
|
||||
Key: aws.String(key),
|
||||
@@ -73,10 +73,10 @@ func doPut(base *s3lib.Client, bucket, key string, size int64, descr string, rem
|
||||
return nil
|
||||
}
|
||||
|
||||
func doGet(base *s3lib.Client, bucket, key string, size int64, descr string, remoteStart uint64) error {
|
||||
func doGet(base *s3lib.Client, bucket, key string, size int64, token string) error {
|
||||
var replyStatus string
|
||||
var transferredHeader string
|
||||
c := withRDMAHeaders(base, descr, size, remoteStart, &replyStatus, &transferredHeader)
|
||||
c := withRDMAHeaders(base, token, &replyStatus, &transferredHeader)
|
||||
out, err := c.GetObject(context.Background(), &s3lib.GetObjectInput{
|
||||
Bucket: aws.String(bucket),
|
||||
Key: aws.String(key),
|
||||
@@ -114,8 +114,8 @@ func doGet(base *s3lib.Client, bucket, key string, size int64, descr string, rem
|
||||
return err
|
||||
}
|
||||
|
||||
// withRDMAHeaders returns a client that adds the legacy RDMA descriptor,
|
||||
// size, and remote-address headers to every request. The headers are added
|
||||
// withRDMAHeaders returns a client that adds the RDMA token header to every
|
||||
// request. The header is added
|
||||
// via a Build-step middleware — which runs before the Finalize step that
|
||||
// signs the request — so SigV4 covers them in SignedHeaders; an
|
||||
// intermediary can no longer alter the RDMA controls without invalidating
|
||||
@@ -125,7 +125,7 @@ func doGet(base *s3lib.Client, bucket, key string, size int64, descr string, rem
|
||||
// otherwise checksum the empty HTTP body instead of the actual RDMA payload.
|
||||
// If replyStatus/transferred are non-nil, they are set to the response
|
||||
// HeaderRDMAReply and HeaderRDMABytesTransferred values (empty if absent).
|
||||
func withRDMAHeaders(base *s3lib.Client, descr string, size int64, remoteStart uint64, replyStatus, transferred *string) *s3lib.Client {
|
||||
func withRDMAHeaders(base *s3lib.Client, token string, replyStatus, transferred *string) *s3lib.Client {
|
||||
opts := base.Options()
|
||||
opts.RequestChecksumCalculation = aws.RequestChecksumCalculationWhenRequired
|
||||
opts.ResponseChecksumValidation = aws.ResponseChecksumValidationWhenRequired
|
||||
@@ -134,9 +134,7 @@ func withRDMAHeaders(base *s3lib.Client, descr string, size int64, remoteStart u
|
||||
func(ctx context.Context, in smithymiddleware.BuildInput, next smithymiddleware.BuildHandler) (
|
||||
smithymiddleware.BuildOutput, smithymiddleware.Metadata, error) {
|
||||
if req, ok := in.Request.(*smithyhttp.Request); ok {
|
||||
req.Header.Set(cumiddleware.HeaderRDMADescr, descr)
|
||||
req.Header.Set(cumiddleware.HeaderRDMASize, strconv.FormatInt(size, 10))
|
||||
req.Header.Set(cumiddleware.HeaderRDMARemoteAddr, strconv.FormatUint(remoteStart, 10))
|
||||
req.Header.Set(cumiddleware.HeaderRDMAToken, token)
|
||||
}
|
||||
return next.HandleBuild(ctx, in)
|
||||
}), smithymiddleware.Before); err != nil {
|
||||
|
||||
Reference in New Issue
Block a user