diff --git a/infra/conf/grpc.go b/infra/conf/grpc.go index 429186b0..bd147bf0 100644 --- a/infra/conf/grpc.go +++ b/infra/conf/grpc.go @@ -14,6 +14,7 @@ type GRPCConfig struct { PermitWithoutStream bool `json:"permit_without_stream"` InitialWindowsSize int32 `json:"initial_windows_size"` UserAgent string `json:"user_agent"` + MultiConnections int32 `json:"multi_connections"` } func (g *GRPCConfig) Build() (proto.Message, error) { @@ -37,5 +38,6 @@ func (g *GRPCConfig) Build() (proto.Message, error) { PermitWithoutStream: g.PermitWithoutStream, InitialWindowsSize: g.InitialWindowsSize, UserAgent: g.UserAgent, + MultiConnections: g.MultiConnections, }, nil } diff --git a/transport/internet/grpc/config.pb.go b/transport/internet/grpc/config.pb.go index f6aecf47..6682bf8e 100644 --- a/transport/internet/grpc/config.pb.go +++ b/transport/internet/grpc/config.pb.go @@ -33,6 +33,7 @@ type Config struct { PermitWithoutStream bool `protobuf:"varint,6,opt,name=permit_without_stream,json=permitWithoutStream,proto3" json:"permit_without_stream,omitempty"` InitialWindowsSize int32 `protobuf:"varint,7,opt,name=initial_windows_size,json=initialWindowsSize,proto3" json:"initial_windows_size,omitempty"` UserAgent string `protobuf:"bytes,8,opt,name=user_agent,json=userAgent,proto3" json:"user_agent,omitempty"` + MultiConnections int32 `protobuf:"varint,9,opt,name=multi_connections,json=multiConnections,proto3" json:"multi_connections,omitempty"` } func (x *Config) Reset() { @@ -121,6 +122,13 @@ func (x *Config) GetUserAgent() string { return "" } +func (x *Config) GetMultiConnections() int32 { + if x != nil { + return x.MultiConnections + } + return 0 +} + var File_transport_internet_grpc_config_proto protoreflect.FileDescriptor var file_transport_internet_grpc_config_proto_rawDesc = []byte{ @@ -128,7 +136,7 @@ var file_transport_internet_grpc_config_proto_rawDesc = []byte{ 0x72, 0x6e, 0x65, 0x74, 0x2f, 0x67, 0x72, 0x70, 0x63, 0x2f, 0x63, 0x6f, 0x6e, 0x66, 0x69, 0x67, 0x2e, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x12, 0x25, 0x78, 0x72, 0x61, 0x79, 0x2e, 0x74, 0x72, 0x61, 0x6e, 0x73, 0x70, 0x6f, 0x72, 0x74, 0x2e, 0x69, 0x6e, 0x74, 0x65, 0x72, 0x6e, 0x65, 0x74, 0x2e, - 0x67, 0x72, 0x70, 0x63, 0x2e, 0x65, 0x6e, 0x63, 0x6f, 0x64, 0x69, 0x6e, 0x67, 0x22, 0xc2, 0x02, + 0x67, 0x72, 0x70, 0x63, 0x2e, 0x65, 0x6e, 0x63, 0x6f, 0x64, 0x69, 0x6e, 0x67, 0x22, 0xef, 0x02, 0x0a, 0x06, 0x43, 0x6f, 0x6e, 0x66, 0x69, 0x67, 0x12, 0x1c, 0x0a, 0x09, 0x61, 0x75, 0x74, 0x68, 0x6f, 0x72, 0x69, 0x74, 0x79, 0x18, 0x01, 0x20, 0x01, 0x28, 0x09, 0x52, 0x09, 0x61, 0x75, 0x74, 0x68, 0x6f, 0x72, 0x69, 0x74, 0x79, 0x12, 0x21, 0x0a, 0x0c, 0x73, 0x65, 0x72, 0x76, 0x69, 0x63, @@ -149,10 +157,13 @@ var file_transport_internet_grpc_config_proto_rawDesc = []byte{ 0x12, 0x69, 0x6e, 0x69, 0x74, 0x69, 0x61, 0x6c, 0x57, 0x69, 0x6e, 0x64, 0x6f, 0x77, 0x73, 0x53, 0x69, 0x7a, 0x65, 0x12, 0x1d, 0x0a, 0x0a, 0x75, 0x73, 0x65, 0x72, 0x5f, 0x61, 0x67, 0x65, 0x6e, 0x74, 0x18, 0x08, 0x20, 0x01, 0x28, 0x09, 0x52, 0x09, 0x75, 0x73, 0x65, 0x72, 0x41, 0x67, 0x65, - 0x6e, 0x74, 0x42, 0x33, 0x5a, 0x31, 0x67, 0x69, 0x74, 0x68, 0x75, 0x62, 0x2e, 0x63, 0x6f, 0x6d, - 0x2f, 0x78, 0x74, 0x6c, 0x73, 0x2f, 0x78, 0x72, 0x61, 0x79, 0x2d, 0x63, 0x6f, 0x72, 0x65, 0x2f, - 0x74, 0x72, 0x61, 0x6e, 0x73, 0x70, 0x6f, 0x72, 0x74, 0x2f, 0x69, 0x6e, 0x74, 0x65, 0x72, 0x6e, - 0x65, 0x74, 0x2f, 0x67, 0x72, 0x70, 0x63, 0x62, 0x06, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x33, + 0x6e, 0x74, 0x12, 0x2b, 0x0a, 0x11, 0x6d, 0x75, 0x6c, 0x74, 0x69, 0x5f, 0x63, 0x6f, 0x6e, 0x6e, + 0x65, 0x63, 0x74, 0x69, 0x6f, 0x6e, 0x73, 0x18, 0x09, 0x20, 0x01, 0x28, 0x05, 0x52, 0x10, 0x6d, + 0x75, 0x6c, 0x74, 0x69, 0x43, 0x6f, 0x6e, 0x6e, 0x65, 0x63, 0x74, 0x69, 0x6f, 0x6e, 0x73, 0x42, + 0x33, 0x5a, 0x31, 0x67, 0x69, 0x74, 0x68, 0x75, 0x62, 0x2e, 0x63, 0x6f, 0x6d, 0x2f, 0x78, 0x74, + 0x6c, 0x73, 0x2f, 0x78, 0x72, 0x61, 0x79, 0x2d, 0x63, 0x6f, 0x72, 0x65, 0x2f, 0x74, 0x72, 0x61, + 0x6e, 0x73, 0x70, 0x6f, 0x72, 0x74, 0x2f, 0x69, 0x6e, 0x74, 0x65, 0x72, 0x6e, 0x65, 0x74, 0x2f, + 0x67, 0x72, 0x70, 0x63, 0x62, 0x06, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x33, } var ( diff --git a/transport/internet/grpc/config.proto b/transport/internet/grpc/config.proto index fcaa2ed9..3c56eb16 100644 --- a/transport/internet/grpc/config.proto +++ b/transport/internet/grpc/config.proto @@ -12,4 +12,5 @@ message Config { bool permit_without_stream = 6; int32 initial_windows_size = 7; string user_agent = 8; + int32 multi_connections = 9; } diff --git a/transport/internet/grpc/dial.go b/transport/internet/grpc/dial.go index b8740dae..3df8a2a3 100644 --- a/transport/internet/grpc/dial.go +++ b/transport/internet/grpc/dial.go @@ -8,6 +8,7 @@ import ( "github.com/xtls/xray-core/common" c "github.com/xtls/xray-core/common/ctx" + "github.com/xtls/xray-core/common/dice" "github.com/xtls/xray-core/common/errors" "github.com/xtls/xray-core/common/net" "github.com/xtls/xray-core/common/session" @@ -42,8 +43,61 @@ type dialerConf struct { *internet.MemoryStreamConfig } +type globalDialers struct { + connMap map[dialerConf][]*grpc.ClientConn +} + +// getClientConn returns a client connection from the global dialer if the connections already reached the target number +// otherwise return nil +func (d *globalDialers) getClientConn(conf dialerConf) (*grpc.ClientConn, int) { + if d.connMap == nil { + d.connMap = make(map[dialerConf][]*grpc.ClientConn) + } + if d.connMap[conf] == nil { + d.connMap[conf] = []*grpc.ClientConn{} + } + + conns := d.connMap[conf] + + targetConnsNum := conf.MemoryStreamConfig.ProtocolSettings.(*Config).MultiConnections + if targetConnsNum > int32(len(conns)) { + return nil, 0 + } else { + index := dice.Roll(len(conns)) + return conns[index], index + } +} + +// addClientConn adds a client connection to the global dialer +func (d *globalDialers) addClientConn(conf dialerConf, conn *grpc.ClientConn) error { + if d.connMap == nil { + d.connMap = make(map[dialerConf][]*grpc.ClientConn) + } + if d.connMap[conf] == nil { + d.connMap[conf] = []*grpc.ClientConn{} + } + + conns := d.connMap[conf] + targetConnsNum := conf.MemoryStreamConfig.ProtocolSettings.(*Config).MultiConnections + if targetConnsNum <= int32(len(conns)) { + return errors.New("failed to add client connection beacuse reach the limit") + } else { + conns = append(conns, conn) + d.connMap[conf] = conns + return nil + } +} + +// updateClientConnWithIndex updates a client connection with the given index +// in the case if the clientConn is shutting down, replace it with the new one +func (d *globalDialers) updateClientConnWithIndex(conf dialerConf, conn *grpc.ClientConn, index int) error { + conns := d.connMap[conf] + conns[index] = conn + return nil +} + var ( - globalDialerMap map[dialerConf]*grpc.ClientConn + globalDialer globalDialers globalDialerAccess sync.Mutex ) @@ -77,15 +131,13 @@ func getGrpcClient(ctx context.Context, dest net.Destination, streamSettings *in globalDialerAccess.Lock() defer globalDialerAccess.Unlock() - if globalDialerMap == nil { - globalDialerMap = make(map[dialerConf]*grpc.ClientConn) - } tlsConfig := tls.ConfigFromStreamSettings(streamSettings) realityConfig := reality.ConfigFromStreamSettings(streamSettings) sockopt := streamSettings.SocketSettings grpcSettings := streamSettings.ProtocolSettings.(*Config) - if client, found := globalDialerMap[dialerConf{dest, streamSettings}]; found && client.GetState() != connectivity.Shutdown { + client, index := globalDialer.getClientConn(dialerConf{dest, streamSettings}) + if client != nil && client.GetState() != connectivity.Shutdown { return client, nil } @@ -183,6 +235,14 @@ func getGrpcClient(ctx context.Context, dest net.Destination, streamSettings *in gonet.JoinHostPort(grpcDestHost, dest.Port.String()), dialOptions..., ) - globalDialerMap[dialerConf{dest, streamSettings}] = conn + if client == nil { + err := globalDialer.addClientConn(dialerConf{dest, streamSettings}, conn) + if err != nil { + return nil, err + } + } + if client != nil && client.GetState() == connectivity.Shutdown { + globalDialer.updateClientConnWithIndex(dialerConf{dest, streamSettings}, conn, index) + } return conn, err }