From da4f06ec120419f7f238e027b403b0d22ca68975 Mon Sep 17 00:00:00 2001 From: Chris Lu Date: Tue, 18 Aug 2026 21:32:23 -0700 Subject: [PATCH] Give the local Unix socket gRPC transport room to breathe (#10824) * Give the local Unix socket gRPC transport room to breathe Unix socket buffers default small and never autotune: 208KB on Linux, 8KB on macOS. Once the buffer cannot absorb what gRPC's loopyWriter emits for the in-flight streams the writer blocks on Write, and since v1.82.1 grpc-go counts per-RPC bookkeeping toward its control-buffer throttle, so both peers stop reading and the connection deadlocks for good. weed mini wedged at roughly 320 concurrent S3 PUTs with every filer RPC parked in waitOnHeader and no handler running. Force 8MB on both ends of the sockets we open. Best effort, since a kernel may clamp it lower; that only lowers the concurrency this survives. TCP loopback never hit this because its buffers start large and grow. * Set the buffer on accepted connections too Linux does not carry the listener's SO_SNDBUF onto sockets returned by accept, so only the dialing half was getting the headroom: measured 8388608 on the dialed side against the 212992 default on the accepted side. Wrap the listener and re-apply per connection. macOS inherits either way, which is why this did not show up locally. --- weed/pb/grpc_client_server.go | 20 +++++++++++-- weed/pb/local_socket_buffers.go | 39 +++++++++++++++++++++++++ weed/pb/local_socket_buffers_windows.go | 13 +++++++++ 3 files changed, 70 insertions(+), 2 deletions(-) create mode 100644 weed/pb/local_socket_buffers.go create mode 100644 weed/pb/local_socket_buffers_windows.go diff --git a/weed/pb/grpc_client_server.go b/weed/pb/grpc_client_server.go index f1c5bbcc9..878775b9b 100644 --- a/weed/pb/grpc_client_server.go +++ b/weed/pb/grpc_client_server.go @@ -150,11 +150,13 @@ func ServeGrpcOnLocalSocket(grpcServer *grpc.Server, grpcPort int) { if err := os.Remove(socketPath); err != nil && !os.IsNotExist(err) { glog.Warningf("Failed to remove old gRPC socket %s: %v", socketPath, err) } - listener, err := net.Listen("unix", socketPath) + lc := net.ListenConfig{Control: setLocalSocketBuffers} + listener, err := lc.Listen(context.Background(), "unix", socketPath) if err != nil { glog.Errorf("Failed to listen on gRPC Unix socket %s: %v", socketPath, err) return } + listener = &localSocketListener{Listener: listener} glog.V(0).Infof("gRPC also listening on Unix socket %s", socketPath) go func() { if err := grpcServer.Serve(listener); err != nil && err != grpc.ErrServerStopped { @@ -164,6 +166,20 @@ func ServeGrpcOnLocalSocket(grpcServer *grpc.Server, grpcPort int) { }() } +// localSocketListener re-applies the buffer sizes to every accepted connection. +type localSocketListener struct { + net.Listener +} + +func (l *localSocketListener) Accept() (net.Conn, error) { + c, err := l.Listener.Accept() + if err != nil { + return nil, err + } + applyLocalSocketBuffers(c) + return c, nil +} + func NewGrpcServer(opts ...grpc.ServerOption) *grpc.Server { var options []grpc.ServerOption options = append(options, @@ -209,7 +225,7 @@ func GrpcDial(ctx context.Context, address string, waitForReady bool, opts ...gr // Route through Unix socket if one is registered for this address's port if socketPath := resolveLocalGrpcSocket(address); socketPath != "" { options = append(options, grpc.WithContextDialer(func(ctx context.Context, _ string) (net.Conn, error) { - var d net.Dialer + d := net.Dialer{Control: setLocalSocketBuffers} return d.DialContext(ctx, "unix", socketPath) })) } else { diff --git a/weed/pb/local_socket_buffers.go b/weed/pb/local_socket_buffers.go new file mode 100644 index 000000000..3f7f426f3 --- /dev/null +++ b/weed/pb/local_socket_buffers.go @@ -0,0 +1,39 @@ +//go:build !windows + +package pb + +import ( + "net" + "syscall" +) + +// Unix socket buffers default small and never autotune (208KB on Linux, 8KB on +// macOS). Once a buffer cannot absorb what gRPC's loopyWriter emits for the +// in-flight streams the writer blocks, both peers latch grpc-go's control-buffer +// throttle, and the connection deadlocks for good. Best effort: a value the OS +// clamps down only lowers the concurrency this survives. +const localSocketBufBytes = 8 << 20 + +func setLocalSocketBufferFD(fd uintptr) { + syscall.SetsockoptInt(int(fd), syscall.SOL_SOCKET, syscall.SO_SNDBUF, localSocketBufBytes) + syscall.SetsockoptInt(int(fd), syscall.SOL_SOCKET, syscall.SO_RCVBUF, localSocketBufBytes) +} + +func setLocalSocketBuffers(_, _ string, c syscall.RawConn) error { + return c.Control(setLocalSocketBufferFD) +} + +// applyLocalSocketBuffers tunes an already-established connection. Linux does +// not carry the listener's buffer sizes onto accepted sockets, so the server +// half needs setting explicitly or only the dialing side gets the headroom. +func applyLocalSocketBuffers(c net.Conn) { + sc, ok := c.(syscall.Conn) + if !ok { + return + } + raw, err := sc.SyscallConn() + if err != nil { + return + } + raw.Control(setLocalSocketBufferFD) +} diff --git a/weed/pb/local_socket_buffers_windows.go b/weed/pb/local_socket_buffers_windows.go new file mode 100644 index 000000000..b847c5c1a --- /dev/null +++ b/weed/pb/local_socket_buffers_windows.go @@ -0,0 +1,13 @@ +//go:build windows + +package pb + +import ( + "net" + "syscall" +) + +// Windows never registers local Unix sockets, so there is nothing to tune. +func setLocalSocketBuffers(_, _ string, _ syscall.RawConn) error { return nil } + +func applyLocalSocketBuffers(_ net.Conn) {}