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

@@ -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 {

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)
+}

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) {}