Replace "only" with "xudpConcurrency" in Mux config

This commit is contained in:
RPRX 2023-04-10 10:36:07 +08:00 committed by GitHub
parent 29d7865d78
commit 24a2be43ef
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
6 changed files with 112 additions and 90 deletions

View file

@ -594,9 +594,9 @@ type MultiplexingConfig struct {
// Whether or not Mux is enabled. // Whether or not Mux is enabled.
Enabled bool `protobuf:"varint,1,opt,name=enabled,proto3" json:"enabled,omitempty"` Enabled bool `protobuf:"varint,1,opt,name=enabled,proto3" json:"enabled,omitempty"`
// Max number of concurrent connections that one Mux connection can handle. // Max number of concurrent connections that one Mux connection can handle.
Concurrency uint32 `protobuf:"varint,2,opt,name=concurrency,proto3" json:"concurrency,omitempty"` Concurrency int32 `protobuf:"varint,2,opt,name=concurrency,proto3" json:"concurrency,omitempty"`
// Both(0), TCP(net.Network_TCP), UDP(net.Network_UDP). // Transport XUDP in another Mux.
Only uint32 `protobuf:"varint,3,opt,name=only,proto3" json:"only,omitempty"` XudpConcurrency int32 `protobuf:"varint,3,opt,name=xudpConcurrency,proto3" json:"xudpConcurrency,omitempty"`
} }
func (x *MultiplexingConfig) Reset() { func (x *MultiplexingConfig) Reset() {
@ -638,16 +638,16 @@ func (x *MultiplexingConfig) GetEnabled() bool {
return false return false
} }
func (x *MultiplexingConfig) GetConcurrency() uint32 { func (x *MultiplexingConfig) GetConcurrency() int32 {
if x != nil { if x != nil {
return x.Concurrency return x.Concurrency
} }
return 0 return 0
} }
func (x *MultiplexingConfig) GetOnly() uint32 { func (x *MultiplexingConfig) GetXudpConcurrency() int32 {
if x != nil { if x != nil {
return x.Only return x.XudpConcurrency
} }
return 0 return 0
} }
@ -865,22 +865,23 @@ var file_app_proxyman_config_proto_rawDesc = []byte{
0x28, 0x0b, 0x32, 0x25, 0x2e, 0x78, 0x72, 0x61, 0x79, 0x2e, 0x61, 0x70, 0x70, 0x2e, 0x70, 0x72, 0x28, 0x0b, 0x32, 0x25, 0x2e, 0x78, 0x72, 0x61, 0x79, 0x2e, 0x61, 0x70, 0x70, 0x2e, 0x70, 0x72,
0x6f, 0x78, 0x79, 0x6d, 0x61, 0x6e, 0x2e, 0x4d, 0x75, 0x6c, 0x74, 0x69, 0x70, 0x6c, 0x65, 0x78, 0x6f, 0x78, 0x79, 0x6d, 0x61, 0x6e, 0x2e, 0x4d, 0x75, 0x6c, 0x74, 0x69, 0x70, 0x6c, 0x65, 0x78,
0x69, 0x6e, 0x67, 0x43, 0x6f, 0x6e, 0x66, 0x69, 0x67, 0x52, 0x11, 0x6d, 0x75, 0x6c, 0x74, 0x69, 0x69, 0x6e, 0x67, 0x43, 0x6f, 0x6e, 0x66, 0x69, 0x67, 0x52, 0x11, 0x6d, 0x75, 0x6c, 0x74, 0x69,
0x70, 0x6c, 0x65, 0x78, 0x53, 0x65, 0x74, 0x74, 0x69, 0x6e, 0x67, 0x73, 0x22, 0x64, 0x0a, 0x12, 0x70, 0x6c, 0x65, 0x78, 0x53, 0x65, 0x74, 0x74, 0x69, 0x6e, 0x67, 0x73, 0x22, 0x7a, 0x0a, 0x12,
0x4d, 0x75, 0x6c, 0x74, 0x69, 0x70, 0x6c, 0x65, 0x78, 0x69, 0x6e, 0x67, 0x43, 0x6f, 0x6e, 0x66, 0x4d, 0x75, 0x6c, 0x74, 0x69, 0x70, 0x6c, 0x65, 0x78, 0x69, 0x6e, 0x67, 0x43, 0x6f, 0x6e, 0x66,
0x69, 0x67, 0x12, 0x18, 0x0a, 0x07, 0x65, 0x6e, 0x61, 0x62, 0x6c, 0x65, 0x64, 0x18, 0x01, 0x20, 0x69, 0x67, 0x12, 0x18, 0x0a, 0x07, 0x65, 0x6e, 0x61, 0x62, 0x6c, 0x65, 0x64, 0x18, 0x01, 0x20,
0x01, 0x28, 0x08, 0x52, 0x07, 0x65, 0x6e, 0x61, 0x62, 0x6c, 0x65, 0x64, 0x12, 0x20, 0x0a, 0x0b, 0x01, 0x28, 0x08, 0x52, 0x07, 0x65, 0x6e, 0x61, 0x62, 0x6c, 0x65, 0x64, 0x12, 0x20, 0x0a, 0x0b,
0x63, 0x6f, 0x6e, 0x63, 0x75, 0x72, 0x72, 0x65, 0x6e, 0x63, 0x79, 0x18, 0x02, 0x20, 0x01, 0x28, 0x63, 0x6f, 0x6e, 0x63, 0x75, 0x72, 0x72, 0x65, 0x6e, 0x63, 0x79, 0x18, 0x02, 0x20, 0x01, 0x28,
0x0d, 0x52, 0x0b, 0x63, 0x6f, 0x6e, 0x63, 0x75, 0x72, 0x72, 0x65, 0x6e, 0x63, 0x79, 0x12, 0x12, 0x05, 0x52, 0x0b, 0x63, 0x6f, 0x6e, 0x63, 0x75, 0x72, 0x72, 0x65, 0x6e, 0x63, 0x79, 0x12, 0x28,
0x0a, 0x04, 0x6f, 0x6e, 0x6c, 0x79, 0x18, 0x03, 0x20, 0x01, 0x28, 0x0d, 0x52, 0x04, 0x6f, 0x6e, 0x0a, 0x0f, 0x78, 0x75, 0x64, 0x70, 0x43, 0x6f, 0x6e, 0x63, 0x75, 0x72, 0x72, 0x65, 0x6e, 0x63,
0x6c, 0x79, 0x2a, 0x23, 0x0a, 0x0e, 0x4b, 0x6e, 0x6f, 0x77, 0x6e, 0x50, 0x72, 0x6f, 0x74, 0x6f, 0x79, 0x18, 0x03, 0x20, 0x01, 0x28, 0x05, 0x52, 0x0f, 0x78, 0x75, 0x64, 0x70, 0x43, 0x6f, 0x6e,
0x63, 0x6f, 0x6c, 0x73, 0x12, 0x08, 0x0a, 0x04, 0x48, 0x54, 0x54, 0x50, 0x10, 0x00, 0x12, 0x07, 0x63, 0x75, 0x72, 0x72, 0x65, 0x6e, 0x63, 0x79, 0x2a, 0x23, 0x0a, 0x0e, 0x4b, 0x6e, 0x6f, 0x77,
0x0a, 0x03, 0x54, 0x4c, 0x53, 0x10, 0x01, 0x42, 0x55, 0x0a, 0x15, 0x63, 0x6f, 0x6d, 0x2e, 0x78, 0x6e, 0x50, 0x72, 0x6f, 0x74, 0x6f, 0x63, 0x6f, 0x6c, 0x73, 0x12, 0x08, 0x0a, 0x04, 0x48, 0x54,
0x72, 0x61, 0x79, 0x2e, 0x61, 0x70, 0x70, 0x2e, 0x70, 0x72, 0x6f, 0x78, 0x79, 0x6d, 0x61, 0x6e, 0x54, 0x50, 0x10, 0x00, 0x12, 0x07, 0x0a, 0x03, 0x54, 0x4c, 0x53, 0x10, 0x01, 0x42, 0x55, 0x0a,
0x50, 0x01, 0x5a, 0x26, 0x67, 0x69, 0x74, 0x68, 0x75, 0x62, 0x2e, 0x63, 0x6f, 0x6d, 0x2f, 0x78, 0x15, 0x63, 0x6f, 0x6d, 0x2e, 0x78, 0x72, 0x61, 0x79, 0x2e, 0x61, 0x70, 0x70, 0x2e, 0x70, 0x72,
0x74, 0x6c, 0x73, 0x2f, 0x78, 0x72, 0x61, 0x79, 0x2d, 0x63, 0x6f, 0x72, 0x65, 0x2f, 0x61, 0x70, 0x6f, 0x78, 0x79, 0x6d, 0x61, 0x6e, 0x50, 0x01, 0x5a, 0x26, 0x67, 0x69, 0x74, 0x68, 0x75, 0x62,
0x70, 0x2f, 0x70, 0x72, 0x6f, 0x78, 0x79, 0x6d, 0x61, 0x6e, 0xaa, 0x02, 0x11, 0x58, 0x72, 0x61, 0x2e, 0x63, 0x6f, 0x6d, 0x2f, 0x78, 0x74, 0x6c, 0x73, 0x2f, 0x78, 0x72, 0x61, 0x79, 0x2d, 0x63,
0x79, 0x2e, 0x41, 0x70, 0x70, 0x2e, 0x50, 0x72, 0x6f, 0x78, 0x79, 0x6d, 0x61, 0x6e, 0x62, 0x06, 0x6f, 0x72, 0x65, 0x2f, 0x61, 0x70, 0x70, 0x2f, 0x70, 0x72, 0x6f, 0x78, 0x79, 0x6d, 0x61, 0x6e,
0x70, 0x72, 0x6f, 0x74, 0x6f, 0x33, 0xaa, 0x02, 0x11, 0x58, 0x72, 0x61, 0x79, 0x2e, 0x41, 0x70, 0x70, 0x2e, 0x50, 0x72, 0x6f, 0x78,
0x79, 0x6d, 0x61, 0x6e, 0x62, 0x06, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x33,
} }
var ( var (

View file

@ -97,7 +97,7 @@ message MultiplexingConfig {
// Whether or not Mux is enabled. // Whether or not Mux is enabled.
bool enabled = 1; bool enabled = 1;
// Max number of concurrent connections that one Mux connection can handle. // Max number of concurrent connections that one Mux connection can handle.
uint32 concurrency = 2; int32 concurrency = 2;
// Both(0), TCP(net.Network_TCP), UDP(net.Network_UDP). // Transport XUDP in another Mux.
uint32 only = 3; int32 xudpConcurrency = 3;
} }

View file

@ -57,6 +57,7 @@ type Handler struct {
proxy proxy.Outbound proxy proxy.Outbound
outboundManager outbound.Manager outboundManager outbound.Manager
mux *mux.ClientManager mux *mux.ClientManager
xudp *mux.ClientManager
uplinkCounter stats.Counter uplinkCounter stats.Counter
downlinkCounter stats.Counter downlinkCounter stats.Counter
} }
@ -106,23 +107,49 @@ func NewHandler(ctx context.Context, config *core.OutboundHandlerConfig) (outbou
} }
if h.senderSettings != nil && h.senderSettings.MultiplexSettings != nil { if h.senderSettings != nil && h.senderSettings.MultiplexSettings != nil {
config := h.senderSettings.MultiplexSettings if config := h.senderSettings.MultiplexSettings; config.Enabled {
if config.Concurrency < 1 || config.Concurrency > 1024 { if config.Concurrency < 0 {
return nil, newError("invalid mux concurrency: ", config.Concurrency).AtWarning() h.mux = &mux.ClientManager{Enabled: false}
} }
if config.Concurrency == 0 {
config.Concurrency = 8 // same as before
}
if config.Concurrency > 0 {
h.mux = &mux.ClientManager{ h.mux = &mux.ClientManager{
Enabled: config.Enabled, Enabled: true,
Picker: &mux.IncrementalWorkerPicker{ Picker: &mux.IncrementalWorkerPicker{
Factory: &mux.DialingWorkerFactory{ Factory: &mux.DialingWorkerFactory{
Proxy: proxyHandler, Proxy: proxyHandler,
Dialer: h, Dialer: h,
Strategy: mux.ClientStrategy{ Strategy: mux.ClientStrategy{
MaxConcurrency: config.Concurrency, MaxConcurrency: uint32(config.Concurrency),
MaxConnection: 128, MaxConnection: 128,
}, },
}, },
}, },
Only: config.Only, }
}
if config.XudpConcurrency < 0 {
h.xudp = &mux.ClientManager{Enabled: false}
}
if config.XudpConcurrency == 0 {
h.xudp = nil // same as before
}
if config.XudpConcurrency > 0 {
h.xudp = &mux.ClientManager{
Enabled: true,
Picker: &mux.IncrementalWorkerPicker{
Factory: &mux.DialingWorkerFactory{
Proxy: proxyHandler,
Dialer: h,
Strategy: mux.ClientStrategy{
MaxConcurrency: uint32(config.XudpConcurrency),
MaxConnection: 128,
},
},
},
}
}
} }
} }
@ -137,16 +164,28 @@ func (h *Handler) Tag() string {
// Dispatch implements proxy.Outbound.Dispatch. // Dispatch implements proxy.Outbound.Dispatch.
func (h *Handler) Dispatch(ctx context.Context, link *transport.Link) { func (h *Handler) Dispatch(ctx context.Context, link *transport.Link) {
outbound := session.OutboundFromContext(ctx) if h.mux != nil {
if h.mux != nil && (h.mux.Enabled || session.MuxPreferedFromContext(ctx)) && test := func(err error) {
(h.mux.Only == 0 || (outbound != nil && h.mux.Only == uint32(outbound.Target.Network))) { if err != nil {
if err := h.mux.Dispatch(ctx, link); err != nil {
err := newError("failed to process mux outbound traffic").Base(err) err := newError("failed to process mux outbound traffic").Base(err)
session.SubmitOutboundErrorToOriginator(ctx, err) session.SubmitOutboundErrorToOriginator(ctx, err)
err.WriteToLog(session.ExportIDToError(ctx)) err.WriteToLog(session.ExportIDToError(ctx))
common.Interrupt(link.Writer) common.Interrupt(link.Writer)
} }
} else { }
if h.xudp != nil && session.OutboundFromContext(ctx).Target.Network == net.Network_UDP {
if !h.xudp.Enabled {
goto out
}
test(h.xudp.Dispatch(ctx, link))
return
}
if h.mux.Enabled {
test(h.mux.Dispatch(ctx, link))
return
}
}
out:
err := h.proxy.Process(ctx, link, h) err := h.proxy.Process(ctx, link, h)
if err != nil { if err != nil {
if errors.Is(err, io.EOF) || errors.Is(err, io.ErrClosedPipe) || errors.Is(err, context.Canceled) { if errors.Is(err, io.EOF) || errors.Is(err, io.ErrClosedPipe) || errors.Is(err, context.Canceled) {
@ -163,7 +202,6 @@ func (h *Handler) Dispatch(ctx context.Context, link *transport.Link) {
common.Must(common.Close(link.Writer)) common.Must(common.Close(link.Writer))
} }
common.Interrupt(link.Reader) common.Interrupt(link.Reader)
}
} }
// Address implements internet.Dialer. // Address implements internet.Dialer.

View file

@ -24,7 +24,6 @@ import (
type ClientManager struct { type ClientManager struct {
Enabled bool // wheather mux is enabled from user config Enabled bool // wheather mux is enabled from user config
Picker WorkerPicker Picker WorkerPicker
Only uint32
} }
func (m *ClientManager) Dispatch(ctx context.Context, link *transport.Link) error { func (m *ClientManager) Dispatch(ctx context.Context, link *transport.Link) error {

View file

@ -10,7 +10,6 @@ import (
"github.com/xtls/xray-core/app/dispatcher" "github.com/xtls/xray-core/app/dispatcher"
"github.com/xtls/xray-core/app/proxyman" "github.com/xtls/xray-core/app/proxyman"
"github.com/xtls/xray-core/app/stats" "github.com/xtls/xray-core/app/stats"
"github.com/xtls/xray-core/common/net"
"github.com/xtls/xray-core/common/serial" "github.com/xtls/xray-core/common/serial"
core "github.com/xtls/xray-core/core" core "github.com/xtls/xray-core/core"
"github.com/xtls/xray-core/transport/internet" "github.com/xtls/xray-core/transport/internet"
@ -110,34 +109,16 @@ func (c *SniffingConfig) Build() (*proxyman.SniffingConfig, error) {
type MuxConfig struct { type MuxConfig struct {
Enabled bool `json:"enabled"` Enabled bool `json:"enabled"`
Concurrency int16 `json:"concurrency"` Concurrency int16 `json:"concurrency"`
Only string `json:"only"` XudpConcurrency int16 `json:"xudpConcurrency"`
} }
// Build creates MultiplexingConfig, Concurrency < 0 completely disables mux. // Build creates MultiplexingConfig, Concurrency < 0 completely disables mux.
func (m *MuxConfig) Build() (*proxyman.MultiplexingConfig, error) { func (m *MuxConfig) Build() (*proxyman.MultiplexingConfig, error) {
if m.Concurrency < 0 { return &proxyman.MultiplexingConfig{
return nil, nil
}
if m.Concurrency == 0 {
m.Concurrency = 8
}
config := &proxyman.MultiplexingConfig{
Enabled: m.Enabled, Enabled: m.Enabled,
Concurrency: uint32(m.Concurrency), Concurrency: int32(m.Concurrency),
} XudpConcurrency: int32(m.XudpConcurrency),
}, nil
switch strings.ToLower(m.Only) {
case "", "both":
case "tcp":
config.Only = uint32(net.Network_TCP)
case "udp":
config.Only = uint32(net.Network_UDP)
default:
return nil, newError(`unknown "only": `, m.Only)
}
return config, nil
} }
type InboundDetourAllocationConfig struct { type InboundDetourAllocationConfig struct {

View file

@ -345,13 +345,16 @@ func TestMuxConfig_Build(t *testing.T) {
}}, }},
{"empty def", `{}`, &proxyman.MultiplexingConfig{ {"empty def", `{}`, &proxyman.MultiplexingConfig{
Enabled: false, Enabled: false,
Concurrency: 8, Concurrency: 0,
}}, }},
{"not enable", `{"enabled": false, "concurrency": 4}`, &proxyman.MultiplexingConfig{ {"not enable", `{"enabled": false, "concurrency": 4}`, &proxyman.MultiplexingConfig{
Enabled: false, Enabled: false,
Concurrency: 4, Concurrency: 4,
}}, }},
{"forbidden", `{"enabled": false, "concurrency": -1}`, nil}, {"forbidden", `{"enabled": false, "concurrency": -1}`, &proxyman.MultiplexingConfig{
Enabled: false,
Concurrency: -1,
}},
} }
for _, tt := range tests { for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) { t.Run(tt.name, func(t *testing.T) {