versitygw/cubackend/cuserver.go

// 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.

//go:build linux && amd64 && cgo

package cubackend

// Package backend implements the cuObject-accelerated storage backend.
// CuServer embeds the versitygw backend and overrides PutObject and
// GetObject to use RDMA transfers when the request contains a cuObject
// RDMA descriptor. All other S3 operations delegate directly to backend.

import (
	"bytes"
	"context"
	"fmt"
	"io"
	"net/http"

	"github.com/aws/aws-sdk-go-v2/service/s3"

	"github.com/versity/versitygw/backend"
	"github.com/versity/versitygw/cumiddleware"
	"github.com/versity/versitygw/debuglogger"
	"github.com/versity/versitygw/rdma"
	"github.com/versity/versitygw/rdma/bufferpool"
	"github.com/versity/versitygw/s3err"
	"github.com/versity/versitygw/s3response"
)

// CuServer is a versitygw Backend that embeds another backend and overrides
// PutObject/GetObject for RDMA-accelerated transfers via cuObjServer.
// All other S3 operations (list, head, delete, multipart, etc.) pass
// through to the embedded backend unchanged.
type CuServer struct {
	backend.Backend
	rdmaSrv *rdma.Server
	pool    *bufferpool.Pool
}

// New creates a CuServer backend. It creates a cuObjServer instance,
// starts the RDMA session, and pre-allocates a pool of RDMA-registered
// host buffers.
func New(opts CuServerOpts, be backend.Backend) (*CuServer, error) {
	rdmaSrv, err := rdma.NewServer(opts.RDMAIP, opts.RDMAPort, opts.RDMATunables)
	if err != nil {
		be.Shutdown()
		return nil, fmt.Errorf("cuserver: rdma server: %w", err)
	}

	// StartSession is a no-op when the library manages session start
	// internally (e.g. libcuobjserver v1.2.0 calls startRDMASession from
	// the cuObjServer constructor). It is kept here for forward compatibility
	// with library versions that require an explicit call.
	if err := rdmaSrv.StartSession(); err != nil {
		rdmaSrv.Close()
		be.Shutdown()
		return nil, fmt.Errorf("cuserver: rdma session: %w", err)
	}

	bufSize := opts.PoolBufSize
	if bufSize <= 0 {
		bufSize = rdma.MaxTransferSize
	}
	bufCount := opts.PoolBufCount
	if bufCount <= 0 {
		bufCount = 4
	}

	pool, err := bufferpool.NewPool(rdmaSrv, bufSize, bufCount)
	if err != nil {
		rdmaSrv.Close()
		be.Shutdown()
		return nil, fmt.Errorf("cuserver: buffer pool: %w", err)
	}

	return &CuServer{
		Backend: be,
		rdmaSrv: rdmaSrv,
		pool:    pool,
	}, nil
}

// String returns a human-readable identifier for the backend.
func (*CuServer) String() string {
	return "cuObject Server"
}

// Shutdown closes the buffer pool and RDMA server, then shuts down the
// embedded backend.
func (cs *CuServer) Shutdown() {
	cs.pool.Close()
	cs.rdmaSrv.Close()
	cs.Backend.Shutdown()
}

// PutObject handles S3 PUT requests. If the request context contains a
// cuObject RDMA descriptor, data is transferred via RDMA READ from the
// client's GPU memory into a local buffer, then written to the backend
// Without an RDMA descriptor, delegates directly to backend.
func (cs *CuServer) PutObject(ctx context.Context, po s3response.PutObjectInput) (s3response.PutObjectOutput, error) {
	descr, ok := cumiddleware.GetRDMADescriptor(ctx)
	if !ok {
		debuglogger.Logf("doing normal put")
		// Normal S3 path — no RDMA descriptor present
		return cs.Backend.PutObject(ctx, po)
	}

	debuglogger.Logf("doing RDMA put")
	// 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)
	}
	if size > int64(rdma.MaxTransferSize) {
		return s3response.PutObjectOutput{}, fmt.Errorf("cuserver: object size %d exceeds RDMA max %d", size, rdma.MaxTransferSize)
	}

	remoteStart := cumiddleware.GetRDMARemoteStart(ctx)
	debuglogger.Logf("RDMA PUT params: key=%q size=%d remoteStart=0x%x descr=%s", keyFromPtr(po.Key), size, remoteStart, descr)

	channelID, err := cs.rdmaSrv.AllocateChannel()
	if err != nil {
		return s3response.PutObjectOutput{}, fmt.Errorf("cuserver: alloc channel: %w", err)
	}
	defer cs.rdmaSrv.FreeChannel(channelID)

	buf, _, err := cs.pool.Acquire(ctx)
	if err != nil {
		return s3response.PutObjectOutput{}, fmt.Errorf("cuserver: acquire buffer: %w", err)
	}
	defer cs.pool.Release(buf)

	key := keyFromPtr(po.Key)

	// Stream RDMA chunks into a pipe so backend can begin writing before
	// the full transfer is complete.
	pr, pw := io.Pipe()
	po.Body = pr
	po.ContentLength = &size
	producerErrCh := make(chan error, 1)
	go func() {
		defer close(producerErrCh)

		local := buf.Slice()
		remaining := size
		offset := int64(0)

		for remaining > 0 {
			chunk := rdmaPutChunkSize(len(local), remaining)
			if chunk <= 0 {
				wrapped := fmt.Errorf("cuserver: invalid RDMA PUT buffer size %d", len(local))
				_ = pw.CloseWithError(wrapped)
				producerErrCh <- wrapped
				return
			}

			// RDMA READ current chunk from client into the start of local buffer.
			n, e := cs.rdmaSrv.HandlePut(key, buf, remoteStart+uint64(offset), chunk, descr, channelID)
			if e == nil && n != chunk {
				e = fmt.Errorf("short RDMA transfer: got %d bytes, want %d", n, chunk)
			}
			if e != nil {
				wrapped := fmt.Errorf("cuserver: RDMA PUT %q chunk offset=%d size=%d: %w", key, offset, chunk, e)
				_ = pw.CloseWithError(wrapped)
				producerErrCh <- wrapped
				return
			}

			if _, e := pw.Write(local[:int(chunk)]); e != nil {
				producerErrCh <- e
				return
			}

			offset += chunk
			remaining -= chunk
		}

		if e := pw.Close(); e != nil {
			producerErrCh <- e
			return
		}
		producerErrCh <- nil
	}()

	out, err := cs.Backend.PutObject(ctx, po)
	if err != nil {
		_ = pr.Close()
		if prodErr := <-producerErrCh; prodErr != nil {
			return s3response.PutObjectOutput{}, fmt.Errorf("cuserver: PUT %q failed (backend: %v; rdma/pipe: %w)", key, err, prodErr)
		}
		return s3response.PutObjectOutput{}, err
	}

	if prodErr := <-producerErrCh; prodErr != nil {
		return s3response.PutObjectOutput{}, prodErr
	}

	cumiddleware.SetRDMAReplyHeader(ctx, http.StatusOK, size)
	return out, nil
}

// UploadPart handles S3 multipart upload part requests. RDMA offload is not
// supported here: object data would travel out-of-band via RDMA, but the
// embedded backend's UploadPart still expects the part in the HTTP body.
// Decline the descriptor so the client retries over the normal HTTP path
// instead of silently uploading a wrong/empty part.
func (cs *CuServer) UploadPart(ctx context.Context, input *s3.UploadPartInput) (*s3.UploadPartOutput, error) {
	if _, ok := cumiddleware.GetRDMADescriptor(ctx); ok {
		return nil, s3err.GetAPIError(s3err.ErrNotImplemented)
	}
	return cs.Backend.UploadPart(ctx, input)
}

// GetObject handles S3 GET requests. If the request context contains a
// cuObject RDMA descriptor, the object is read from the backend
// into an RDMA buffer, then transferred via RDMA WRITE to the client's
// GPU memory. The HTTP response contains metadata only (no body).
// Without an RDMA descriptor, delegates directly to backend.
func (cs *CuServer) GetObject(ctx context.Context, input *s3.GetObjectInput) (*s3.GetObjectOutput, error) {
	descr, ok := cumiddleware.GetRDMADescriptor(ctx)
	if !ok {
		debuglogger.Logf("doing normal get")
		// Normal S3 path
		return cs.Backend.GetObject(ctx, input)
	}

	debuglogger.Logf("doing RDMA get")

	// First, get the object from backend (opens the file, returns metadata + Body)
	res, err := cs.Backend.GetObject(ctx, input)
	if err != nil {
		return nil, err
	}

	// Determine the actual content size
	var size int64
	if res.ContentLength != nil {
		size = *res.ContentLength
	}
	if size <= 0 {
		// No content to transfer — return metadata only
		return res, nil
	}
	if size > int64(rdma.MaxTransferSize) {
		if res.Body != nil {
			res.Body.Close()
		}
		return nil, fmt.Errorf("cuserver: object size %d exceeds RDMA max %d", size, rdma.MaxTransferSize)
	}
	if capSize, ok := cumiddleware.GetRDMASize(ctx); ok {
		if capSize <= 0 {
			if res.Body != nil {
				res.Body.Close()
			}
			return nil, fmt.Errorf("cuserver: RDMA GET requires a positive client capacity")
		}
		if size > capSize {
			if res.Body != nil {
				res.Body.Close()
			}
			return nil, fmt.Errorf("cuserver: object/range size %d exceeds RDMA client capacity %d", size, capSize)
		}
	}

	remoteStart := cumiddleware.GetRDMARemoteStart(ctx)

	channelID, err := cs.rdmaSrv.AllocateChannel()
	if err != nil {
		res.Body.Close()
		return nil, fmt.Errorf("cuserver: alloc channel: %w", err)
	}
	defer cs.rdmaSrv.FreeChannel(channelID)

	buf, slice, err := cs.pool.Acquire(ctx)
	if err != nil {
		res.Body.Close()
		return nil, fmt.Errorf("cuserver: acquire buffer: %w", err)
	}
	defer cs.pool.Release(buf)
	defer res.Body.Close()

	if len(slice) == 0 {
		return nil, fmt.Errorf("cuserver: invalid RDMA GET buffer size %d", len(slice))
	}

	key := keyFromPtr(input.Key)
	debuglogger.Logf("RDMA GET params: key=%q size=%d remoteStart=0x%x descr=%s", key, size, remoteStart, descr)

	remaining := size
	offset := int64(0)
	for remaining > 0 {
		chunk := rdmaPutChunkSize(len(slice), remaining)
		if chunk <= 0 {
			return nil, fmt.Errorf("cuserver: invalid RDMA GET buffer size %d", len(slice))
		}

		if _, err = io.ReadFull(res.Body, slice[:int(chunk)]); err != nil {
			return nil, fmt.Errorf("cuserver: read object %q chunk offset=%d size=%d into RDMA buffer: %w", key, offset, chunk, err)
		}

		n, err := cs.rdmaSrv.HandleGet(key, buf, remoteStart+uint64(offset), chunk, descr, channelID)
		if err == nil && n != chunk {
			err = fmt.Errorf("short RDMA transfer: got %d bytes, want %d", n, chunk)
		}
		if err != nil {
			return nil, fmt.Errorf("cuserver: RDMA GET %q chunk offset=%d size=%d: %w", key, offset, chunk, err)
		}

		offset += chunk
		remaining -= chunk
	}

	// Return metadata only — the data was already sent via RDMA.
	// Set Body to empty and clear ContentLength so the HTTP response
	// carries only headers.
	res.Body = io.NopCloser(bytes.NewReader(nil))
	zero := int64(0)
	res.ContentLength = &zero

	cumiddleware.SetRDMAReplyHeader(ctx, http.StatusOK, size)
	return res, nil
}

func keyFromPtr(p *string) string {
	if p == nil {
		return ""
	}
	return *p
}

// Ensure CuServer satisfies the Backend interface at compile time.
var _ backend.Backend = (*CuServer)(nil)

func rdmaPutChunkSize(bufSize int, remaining int64) int64 {
	if bufSize <= 0 || remaining <= 0 {
		return 0
	}

	chunkSize := int64(bufSize)
	if remaining < chunkSize {
		return remaining
	}

	return chunkSize
}