diff --git a/weed/command/volume.go b/weed/command/volume.go index 601849f285..ef0aa2d303 100644 --- a/weed/command/volume.go +++ b/weed/command/volume.go @@ -74,6 +74,7 @@ type VolumeServerOptions struct { ldbTimeout *int64 debug *bool debugPort *int + udsListen *string // UDS socket path for RDMA sidecar integration } func init() { @@ -114,6 +115,7 @@ func init() { v.readBufferSizeMB = cmdVolume.Flag.Int("readBufferSizeMB", 4, " larger values can optimize query performance but will increase some memory usage,Use with hasSlowRead normally.") v.debug = cmdVolume.Flag.Bool("debug", false, "serves runtime profiling data via pprof on the port specified by -debug.port") v.debugPort = cmdVolume.Flag.Int("debug.port", 6060, "http port for debugging") + v.udsListen = cmdVolume.Flag.String("uds.listen", "", "Unix domain socket path for RDMA sidecar locate API (e.g., /tmp/sra-volume.sock)") } var cmdVolume = &Command{ @@ -289,6 +291,17 @@ func (v VolumeServerOptions) startVolumeServer(volumeFolders, maxVolumeCounts, v // starting grpc server grpcS := v.startGrpcService(volumeServer) + // starting UDS server for RDMA sidecar integration + var udsServer *weed_server.UdsServer + if *v.udsListen != "" { + var err error + udsServer, err = weed_server.NewUdsServer(volumeServer, *v.udsListen) + if err != nil { + glog.Fatalf("failed to start UDS server: %v", err) + } + udsServer.Start() + } + // starting public http server var publicHttpDown httpdown.Server if v.isSeparatedPublicPort() { @@ -315,7 +328,7 @@ func (v VolumeServerOptions) startVolumeServer(volumeFolders, maxVolumeCounts, v time.Sleep(time.Duration(*v.preStopSeconds) * time.Second) } - shutdown(publicHttpDown, clusterHttpServer, grpcS, volumeServer) + shutdown(publicHttpDown, clusterHttpServer, grpcS, volumeServer, udsServer) stopChan <- true }) @@ -324,7 +337,7 @@ func (v VolumeServerOptions) startVolumeServer(volumeFolders, maxVolumeCounts, v select { case <-stopChan: case <-ctx.Done(): - shutdown(publicHttpDown, clusterHttpServer, grpcS, volumeServer) + shutdown(publicHttpDown, clusterHttpServer, grpcS, volumeServer, udsServer) } } else { select { @@ -334,7 +347,7 @@ func (v VolumeServerOptions) startVolumeServer(volumeFolders, maxVolumeCounts, v } -func shutdown(publicHttpDown httpdown.Server, clusterHttpServer httpdown.Server, grpcS *grpc.Server, volumeServer *weed_server.VolumeServer) { +func shutdown(publicHttpDown httpdown.Server, clusterHttpServer httpdown.Server, grpcS *grpc.Server, volumeServer *weed_server.VolumeServer, udsServer *weed_server.UdsServer) { // firstly, stop the public http service to prevent from receiving new user request if nil != publicHttpDown { @@ -352,6 +365,11 @@ func shutdown(publicHttpDown httpdown.Server, clusterHttpServer httpdown.Server, glog.V(0).Infof("graceful stop gRPC ...") grpcS.GracefulStop() + if udsServer != nil { + glog.V(0).Infof("stop UDS server ...") + udsServer.Stop() + } + volumeServer.Shutdown() pprof.StopCPUProfile() diff --git a/weed/server/volume_server_uds.go b/weed/server/volume_server_uds.go new file mode 100644 index 0000000000..2867cdec64 --- /dev/null +++ b/weed/server/volume_server_uds.go @@ -0,0 +1,228 @@ +package weed_server + +import ( + "encoding/binary" + "fmt" + "io" + "net" + "os" + "strings" + "sync" + + "github.com/seaweedfs/seaweedfs/weed/glog" + "github.com/seaweedfs/seaweedfs/weed/storage/needle" +) + +// UDS protocol constants +const ( + UdsRequestSize = 24 // fid(16) + version(4) + flags(4) + UdsResponseSize = 32 // status(1) + pad(3) + volume_id(4) + offset(8) + length(8) + reserved(8) +) + +// UDS status codes +const ( + UdsStatusOk uint8 = 0 + UdsStatusNotFound uint8 = 1 + UdsStatusError uint8 = 2 +) + +// UdsServer handles UDS locate requests for RDMA sidecar integration +type UdsServer struct { + vs *VolumeServer + listener net.Listener + stopChan chan struct{} + wg sync.WaitGroup +} + +// LocateRequest represents a UDS locate request +type LocateRequest struct { + Fid [16]byte // ASCII fid, null-padded + Version uint32 + Flags uint32 +} + +// LocateResponse represents a UDS locate response +type LocateResponse struct { + Status uint8 + _pad [3]byte + VolumeId uint32 + Offset uint64 + Length uint64 + _ [8]byte // reserved +} + +// NewUdsServer creates a new UDS server for the volume server +func NewUdsServer(vs *VolumeServer, socketPath string) (*UdsServer, error) { + // Remove existing socket file if exists + if err := os.Remove(socketPath); err != nil && !os.IsNotExist(err) { + return nil, fmt.Errorf("failed to remove existing socket: %w", err) + } + + listener, err := net.Listen("unix", socketPath) + if err != nil { + return nil, fmt.Errorf("failed to listen on %s: %w", socketPath, err) + } + + // Set socket permissions to allow access + if err := os.Chmod(socketPath, 0666); err != nil { + listener.Close() + return nil, fmt.Errorf("failed to chmod socket: %w", err) + } + + uds := &UdsServer{ + vs: vs, + listener: listener, + stopChan: make(chan struct{}), + } + + glog.V(0).Infof("UDS server listening on %s", socketPath) + return uds, nil +} + +// Start begins accepting connections +func (u *UdsServer) Start() { + u.wg.Add(1) + go func() { + defer u.wg.Done() + u.acceptLoop() + }() +} + +// Stop gracefully shuts down the UDS server +func (u *UdsServer) Stop() { + close(u.stopChan) + u.listener.Close() + u.wg.Wait() + glog.V(0).Infoln("UDS server stopped") +} + +func (u *UdsServer) acceptLoop() { + for { + conn, err := u.listener.Accept() + if err != nil { + select { + case <-u.stopChan: + return + default: + glog.V(0).Infof("UDS accept error: %v", err) + continue + } + } + + u.wg.Add(1) + go func(c net.Conn) { + defer u.wg.Done() + u.handleConnection(c) + }(conn) + } +} + +func (u *UdsServer) handleConnection(conn net.Conn) { + defer conn.Close() + + reqBuf := make([]byte, UdsRequestSize) + respBuf := make([]byte, UdsResponseSize) + + for { + // Read request + _, err := io.ReadFull(conn, reqBuf) + if err != nil { + if err != io.EOF { + glog.V(2).Infof("UDS read error: %v", err) + } + return + } + + // Parse request + var req LocateRequest + copy(req.Fid[:], reqBuf[0:16]) + req.Version = binary.LittleEndian.Uint32(reqBuf[16:20]) + req.Flags = binary.LittleEndian.Uint32(reqBuf[20:24]) + + // Handle request + resp := u.handleLocate(&req) + + // Serialize response + respBuf[0] = resp.Status + respBuf[1] = 0 + respBuf[2] = 0 + respBuf[3] = 0 + binary.LittleEndian.PutUint32(respBuf[4:8], resp.VolumeId) + binary.LittleEndian.PutUint64(respBuf[8:16], resp.Offset) + binary.LittleEndian.PutUint64(respBuf[16:24], resp.Length) + // reserved bytes 24-32 are zero + + // Write response + if _, err := conn.Write(respBuf); err != nil { + glog.V(2).Infof("UDS write error: %v", err) + return + } + } +} + +func (u *UdsServer) handleLocate(req *LocateRequest) *LocateResponse { + resp := &LocateResponse{} + + // Extract fid string (null-terminated) + fidBytes := req.Fid[:] + fidLen := 0 + for i, b := range fidBytes { + if b == 0 { + fidLen = i + break + } + fidLen = i + 1 + } + fid := string(fidBytes[:fidLen]) + fid = strings.TrimSpace(fid) + + if fid == "" { + resp.Status = UdsStatusError + return resp + } + + // Parse fid (format: "volumeId,needleIdCookie" e.g., "3,01637037") + fileId, err := needle.ParseFileIdFromString(fid) + if err != nil { + glog.V(2).Infof("UDS: invalid fid %q: %v", fid, err) + resp.Status = UdsStatusError + return resp + } + + volumeId := fileId.VolumeId + needleId := fileId.Key + + // Get volume + v := u.vs.store.GetVolume(volumeId) + if v == nil { + glog.V(2).Infof("UDS: volume %d not found for fid %s", volumeId, fid) + resp.Status = UdsStatusNotFound + return resp + } + + // Lookup needle in the volume's needle map + // We need to access the needle map through the volume + nv, ok := v.GetNeedle(needleId) + if !ok || nv.Offset.IsZero() { + glog.V(3).Infof("UDS: needle %s not found in volume %d", fid, volumeId) + resp.Status = UdsStatusNotFound + return resp + } + + // Check if deleted + if nv.Size.IsDeleted() { + glog.V(3).Infof("UDS: needle %s is deleted", fid) + resp.Status = UdsStatusNotFound + return resp + } + + // Return the needle header offset (not data offset) + // The reader will parse the header to find DataSize + resp.Status = UdsStatusOk + resp.VolumeId = uint32(volumeId) + resp.Offset = uint64(nv.Offset.ToActualOffset()) + resp.Length = uint64(nv.Size) + + glog.V(3).Infof("UDS: located %s -> vol=%d offset=%d size=%d", fid, volumeId, resp.Offset, resp.Length) + return resp +} diff --git a/weed/server/volume_server_uds_test.go b/weed/server/volume_server_uds_test.go new file mode 100644 index 0000000000..53b7f3ec53 --- /dev/null +++ b/weed/server/volume_server_uds_test.go @@ -0,0 +1,225 @@ +package weed_server + +import ( + "encoding/binary" + "io" + "net" + "os" + "path/filepath" + "testing" + "time" +) + +func TestLocateRequestResponseSize(t *testing.T) { + // Verify protocol sizes match spec + if UdsRequestSize != 24 { + t.Errorf("UdsRequestSize expected 24, got %d", UdsRequestSize) + } + if UdsResponseSize != 32 { + t.Errorf("UdsResponseSize expected 32, got %d", UdsResponseSize) + } +} + +func TestLocateRequestSerialization(t *testing.T) { + req := LocateRequest{ + Version: 1, + Flags: 0, + } + copy(req.Fid[:], "3,01637037") + + buf := make([]byte, UdsRequestSize) + copy(buf[0:16], req.Fid[:]) + binary.LittleEndian.PutUint32(buf[16:20], req.Version) + binary.LittleEndian.PutUint32(buf[20:24], req.Flags) + + // Verify fid is correctly placed + fid := string(buf[0:10]) + if fid != "3,01637037" { + t.Errorf("Expected fid='3,01637037', got '%s'", fid) + } + + // Verify version + version := binary.LittleEndian.Uint32(buf[16:20]) + if version != 1 { + t.Errorf("Expected version=1, got %d", version) + } +} + +func TestLocateResponseSerialization(t *testing.T) { + resp := &LocateResponse{ + Status: UdsStatusOk, + VolumeId: 3, + Offset: 1024, + Length: 4096, + } + + buf := make([]byte, UdsResponseSize) + buf[0] = resp.Status + binary.LittleEndian.PutUint32(buf[4:8], resp.VolumeId) + binary.LittleEndian.PutUint64(buf[8:16], resp.Offset) + binary.LittleEndian.PutUint64(buf[16:24], resp.Length) + + // Deserialize and verify + status := buf[0] + if status != UdsStatusOk { + t.Errorf("Expected status=0, got %d", status) + } + + volumeId := binary.LittleEndian.Uint32(buf[4:8]) + if volumeId != 3 { + t.Errorf("Expected volumeId=3, got %d", volumeId) + } + + offset := binary.LittleEndian.Uint64(buf[8:16]) + if offset != 1024 { + t.Errorf("Expected offset=1024, got %d", offset) + } + + length := binary.LittleEndian.Uint64(buf[16:24]) + if length != 4096 { + t.Errorf("Expected length=4096, got %d", length) + } +} + +func TestUdsStatusCodes(t *testing.T) { + if UdsStatusOk != 0 { + t.Errorf("UdsStatusOk expected 0, got %d", UdsStatusOk) + } + if UdsStatusNotFound != 1 { + t.Errorf("UdsStatusNotFound expected 1, got %d", UdsStatusNotFound) + } + if UdsStatusError != 2 { + t.Errorf("UdsStatusError expected 2, got %d", UdsStatusError) + } +} + +// TestUdsServerBasicConnection tests that we can create and connect to a UDS server +func TestUdsServerBasicConnection(t *testing.T) { + // Create temp socket path + tmpDir := t.TempDir() + socketPath := filepath.Join(tmpDir, "test.sock") + + // Create a mock volume server (nil is fine since we won't make real lookups) + uds, err := NewUdsServer(nil, socketPath) + if err != nil { + t.Fatalf("Failed to create UDS server: %v", err) + } + defer uds.Stop() + + uds.Start() + + // Give server time to start + time.Sleep(10 * time.Millisecond) + + // Try to connect + conn, err := net.Dial("unix", socketPath) + if err != nil { + t.Fatalf("Failed to connect to UDS server: %v", err) + } + defer conn.Close() + + // Connection successful +} + +// TestUdsServerProtocol tests the basic protocol exchange +func TestUdsServerProtocol(t *testing.T) { + // Create temp socket path + tmpDir := t.TempDir() + socketPath := filepath.Join(tmpDir, "test.sock") + + // Create a mock volume server (vs=nil means handleLocate will return NotFound) + uds, err := NewUdsServer(nil, socketPath) + if err != nil { + t.Fatalf("Failed to create UDS server: %v", err) + } + defer uds.Stop() + + uds.Start() + time.Sleep(10 * time.Millisecond) + + // Connect + conn, err := net.Dial("unix", socketPath) + if err != nil { + t.Fatalf("Failed to connect: %v", err) + } + defer conn.Close() + + // Send a request (will fail because vs is nil, but we're testing protocol) + reqBuf := make([]byte, UdsRequestSize) + copy(reqBuf[0:16], "3,01637037") + binary.LittleEndian.PutUint32(reqBuf[16:20], 1) // version + binary.LittleEndian.PutUint32(reqBuf[20:24], 0) // flags + + _, err = conn.Write(reqBuf) + if err != nil { + t.Fatalf("Failed to write request: %v", err) + } + + // Read response + respBuf := make([]byte, UdsResponseSize) + _, err = io.ReadFull(conn, respBuf) + if err != nil { + t.Fatalf("Failed to read response: %v", err) + } + + // Since vs is nil, we expect an error status (handleLocate will panic or return error) + // In this case, we just check we got a valid response back + status := respBuf[0] + if status != UdsStatusOk && status != UdsStatusNotFound && status != UdsStatusError { + t.Errorf("Unexpected status: %d", status) + } +} + +// TestUdsSocketCleanup tests that existing socket is removed +func TestUdsSocketCleanup(t *testing.T) { + tmpDir := t.TempDir() + socketPath := filepath.Join(tmpDir, "test.sock") + + // Create a dummy file at the socket path + if err := os.WriteFile(socketPath, []byte("dummy"), 0644); err != nil { + t.Fatalf("Failed to create dummy file: %v", err) + } + + // NewUdsServer should remove the existing file + uds, err := NewUdsServer(nil, socketPath) + if err != nil { + t.Fatalf("Failed to create UDS server (should have cleaned up old socket): %v", err) + } + uds.Stop() +} + +func TestFidParsing(t *testing.T) { + tests := []struct { + fid string + valid bool + }{ + {"3,01637037", true}, + {"1,abc123", true}, + {"", false}, + {"invalid", false}, + } + + for _, tt := range tests { + t.Run(tt.fid, func(t *testing.T) { + req := &LocateRequest{} + copy(req.Fid[:], tt.fid) + + // Extract fid string using same logic as handleLocate + fidBytes := req.Fid[:] + fidLen := 0 + for i, b := range fidBytes { + if b == 0 { + fidLen = i + break + } + fidLen = i + 1 + } + fid := string(fidBytes[:fidLen]) + + // For empty fid, we expect empty string + if tt.fid == "" && fid != "" { + t.Errorf("Expected empty fid, got '%s'", fid) + } + }) + } +} diff --git a/weed/storage/volume.go b/weed/storage/volume.go index dd8ecbdce1..73ed7732e9 100644 --- a/weed/storage/volume.go +++ b/weed/storage/volume.go @@ -12,6 +12,7 @@ import ( "github.com/seaweedfs/seaweedfs/weed/stats" "github.com/seaweedfs/seaweedfs/weed/storage/backend" "github.com/seaweedfs/seaweedfs/weed/storage/needle" + "github.com/seaweedfs/seaweedfs/weed/storage/needle_map" "github.com/seaweedfs/seaweedfs/weed/storage/super_block" "github.com/seaweedfs/seaweedfs/weed/storage/types" @@ -203,6 +204,17 @@ func (v *Volume) IndexFileSize() uint64 { return v.nm.IndexFileSize() } +// GetNeedle looks up a needle by its ID and returns the needle value containing offset and size. +// This is used by the UDS handler for RDMA sidecar integration. +func (v *Volume) GetNeedle(needleId types.NeedleId) (*needle_map.NeedleValue, bool) { + v.dataFileAccessLock.RLock() + defer v.dataFileAccessLock.RUnlock() + if v.nm == nil { + return nil, false + } + return v.nm.Get(needleId) +} + func (v *Volume) DiskType() types.DiskType { return v.location.DiskType }