perf(runtime): pool the local xray gRPC connection (#6916)

A client per runtime operation made a bulk delete pay a TCP and HTTP/2
handshake for every client. The local runtime now keeps one gRPC client
per API port and replaces it when a core restart moves the port; the
traffic poll keeps its own client, since StatsLastValues is unlocked.

A pooled channel keeps grpc-go's reconnect backoff after the core
restarts, so a call that fails with Unavailable drops the client and
dials again, as a client per call did.
This commit is contained in:
25f2002654
2026-10-24 23:00:09 +05:30
committed by GitHub
parent de21c128fc
commit f105132866
3 changed files with 186 additions and 4 deletions
+17 -4
View File
@@ -8,6 +8,9 @@ import (
"strings"
"sync"
"google.golang.org/grpc/codes"
"google.golang.org/grpc/status"
"github.com/mhsanaei/3x-ui/v3/internal/amneziawg"
"github.com/mhsanaei/3x-ui/v3/internal/amneziawgnet"
"github.com/mhsanaei/3x-ui/v3/internal/database/model"
@@ -41,12 +44,22 @@ func (l *Local) withAPI(fn func(api *xray.XrayAPI) error) error {
if port <= 0 {
return errors.New("local xray is not running")
}
var api xray.XrayAPI
if err := api.Init(port); err != nil {
api, err := localAPIForPort(port)
if err != nil {
return err
}
defer api.Close()
return fn(&api)
err = fn(api)
if status.Code(err) != codes.Unavailable {
return err
}
// A pooled channel keeps grpc-go's reconnect backoff after a core restart and fails fast for
// seconds; drop it and dial again, as a client per call did.
evictLocalAPIClient(api)
api, err = localAPIForPort(port)
if err != nil {
return err
}
return fn(api)
}
func (l *Local) AddInbound(_ context.Context, ib *model.Inbound) error {
+52
View File
@@ -0,0 +1,52 @@
package runtime
import (
"sync"
"github.com/mhsanaei/3x-ui/v3/internal/xray"
)
// One pooled gRPC client serves local runtime operations; never use it for traffic stats, whose
// unlocked StatsLastValues a second reader would corrupt, so the traffic poll keeps its own client.
var (
localAPIMu sync.Mutex
localAPIPort int
localAPIClient *xray.XrayAPI
)
// localAPIForPort returns the pooled client for the API port, replacing (and closing) it when a core
// restart moved the port; callers must not close it.
func localAPIForPort(port int) (*xray.XrayAPI, error) {
localAPIMu.Lock()
defer localAPIMu.Unlock()
if localAPIClient != nil {
if localAPIPort == port {
return localAPIClient, nil
}
localAPIClient.Close()
localAPIClient = nil
}
api := &xray.XrayAPI{}
if err := api.Init(port); err != nil {
return nil, err
}
localAPIClient = api
localAPIPort = port
return api, nil
}
// evictLocalAPIClient closes the pooled client so the next call dials a fresh connection, unless
// another call already replaced the one that failed.
func evictLocalAPIClient(api *xray.XrayAPI) {
localAPIMu.Lock()
defer localAPIMu.Unlock()
if localAPIClient != api {
return
}
localAPIClient.Close()
localAPIClient = nil
localAPIPort = 0
}
+117
View File
@@ -0,0 +1,117 @@
package runtime
import (
"context"
"net"
"strconv"
"testing"
"google.golang.org/grpc"
"google.golang.org/grpc/codes"
"google.golang.org/grpc/status"
"github.com/xtls/xray-core/app/proxyman/command"
"github.com/mhsanaei/3x-ui/v3/internal/xray"
)
// A client per operation made a bulk delete pay a TCP + HTTP/2 handshake per client, so the local
// runtime's client must stay open and be reused between calls.
func TestLocalWithAPIKeepsTheClientOpen(t *testing.T) {
port := 63211
local := NewLocal(LocalDeps{APIPort: func() int { return port }})
var api *xray.XrayAPI
if err := local.withAPI(func(a *xray.XrayAPI) error { api = a; return nil }); err != nil {
t.Fatalf("withAPI: %v", err)
}
if api.HandlerServiceClient == nil {
t.Fatal("the client was closed when the call returned; a pooled client must stay open")
}
}
func TestLocalWithAPIReusesOneClient(t *testing.T) {
port := 63212
local := NewLocal(LocalDeps{APIPort: func() int { return port }})
clients := make([]*xray.XrayAPI, 0, 3)
for range 3 {
if err := local.withAPI(func(a *xray.XrayAPI) error {
clients = append(clients, a)
return nil
}); err != nil {
t.Fatalf("withAPI: %v", err)
}
}
for i, c := range clients {
if c != clients[0] {
t.Fatalf("call %d got a different client; the local gRPC connection must be pooled", i)
}
}
}
// A core restart can move the API port, so the pool has to follow it and close
// the client it replaces instead of leaking it.
func TestLocalWithAPIEvictsTheClientWhenThePortChanges(t *testing.T) {
port := 63213
local := NewLocal(LocalDeps{APIPort: func() int { return port }})
var first *xray.XrayAPI
if err := local.withAPI(func(a *xray.XrayAPI) error { first = a; return nil }); err != nil {
t.Fatalf("withAPI: %v", err)
}
port = 63214
var second *xray.XrayAPI
if err := local.withAPI(func(a *xray.XrayAPI) error { second = a; return nil }); err != nil {
t.Fatalf("withAPI after the port change: %v", err)
}
if first == second {
t.Fatal("a changed API port must not reuse the old client")
}
if first.HandlerServiceClient != nil {
t.Fatal("the evicted client must be closed, not leaked")
}
}
type fakeHandlerServer struct {
command.UnimplementedHandlerServiceServer
}
func (fakeHandlerServer) RemoveInbound(context.Context, *command.RemoveInboundRequest) (*command.RemoveInboundResponse, error) {
return &command.RemoveInboundResponse{}, nil
}
// startFakeCore serves the core's handler service on port (0 picks a free one).
func startFakeCore(t *testing.T, port int) (int, func()) {
t.Helper()
listener, err := net.Listen("tcp", net.JoinHostPort("127.0.0.1", strconv.Itoa(port)))
if err != nil {
t.Fatalf("listen on %d: %v", port, err)
}
server := grpc.NewServer()
command.RegisterHandlerServiceServer(server, fakeHandlerServer{})
go func() { _ = server.Serve(listener) }()
return listener.Addr().(*net.TCPAddr).Port, server.Stop
}
// A pooled channel kept grpc-go's reconnect backoff after a core restart and failed fast for
// seconds; the first call after the restart must succeed, as it did with a client per call.
func TestLocalWithAPIRecoversAfterTheCoreRestarts(t *testing.T) {
port, stop := startFakeCore(t, 0)
local := NewLocal(LocalDeps{APIPort: func() int { return port }})
delInbound := func(api *xray.XrayAPI) error { return api.DelInbound("tag") }
if err := local.withAPI(delInbound); err != nil {
t.Fatalf("first call: %v", err)
}
stop()
if err := local.withAPI(delInbound); status.Code(err) != codes.Unavailable {
t.Fatalf("a call while the core is down: got %v, want Unavailable", err)
}
_, stopRestarted := startFakeCore(t, port)
t.Cleanup(stopRestarted)
if err := local.withAPI(delInbound); err != nil {
t.Fatalf("the first call after the core restarted must succeed: %v", err)
}
}