diff --git a/app/dispatcher/default.go b/app/dispatcher/default.go index 863793fb..b66eef49 100644 --- a/app/dispatcher/default.go +++ b/app/dispatcher/default.go @@ -84,8 +84,9 @@ func (*DefaultDispatcher) Start() error { func (*DefaultDispatcher) Close() error { return nil } func (d *DefaultDispatcher) getLink(ctx context.Context) (*core.Link, *core.Link) { - uplinkReader, uplinkWriter := pipe.New() - downlinkReader, downlinkWriter := pipe.New() + opt := pipe.OptionsFromContext(ctx) + uplinkReader, uplinkWriter := pipe.New(opt...) + downlinkReader, downlinkWriter := pipe.New(opt...) inboundLink := &core.Link{ Reader: downlinkReader, diff --git a/app/policy/config.go b/app/policy/config.go index c5ecb065..9fe3036e 100644 --- a/app/policy/config.go +++ b/app/policy/config.go @@ -24,6 +24,10 @@ func defaultPolicy() *Policy { UplinkOnly: &Second{Value: uint32(p.Timeouts.UplinkOnly / time.Second)}, DownlinkOnly: &Second{Value: uint32(p.Timeouts.DownlinkOnly / time.Second)}, }, + Buffer: &Policy_Buffer{ + Enabled: p.Buffer.Enabled, + Size: p.Buffer.Size, + }, } } @@ -54,7 +58,8 @@ func (p *Policy) overrideWith(another *Policy) { // ToCorePolicy converts this Policy to core.Policy. func (p *Policy) ToCorePolicy() core.Policy { - var cp core.Policy + cp := core.DefaultPolicy() + if p.Timeout != nil { cp.Timeouts.ConnectionIdle = p.Timeout.ConnectionIdle.Duration() cp.Timeouts.Handshake = p.Timeout.Handshake.Duration() @@ -65,6 +70,10 @@ func (p *Policy) ToCorePolicy() core.Policy { cp.Stats.UserUplink = p.Stats.UserUplink cp.Stats.UserDownlink = p.Stats.UserDownlink } + if p.Buffer != nil { + cp.Buffer.Enabled = p.Buffer.Enabled + cp.Buffer.Size = p.Buffer.Size + } return cp } diff --git a/app/policy/config.pb.go b/app/policy/config.pb.go index 6cde3e14..64267f42 100644 --- a/app/policy/config.pb.go +++ b/app/policy/config.pb.go @@ -16,13 +16,35 @@ var _ = math.Inf const _ = proto.ProtoPackageIsVersion2 // please upgrade the proto package type Second struct { - Value uint32 `protobuf:"varint,1,opt,name=value" json:"value,omitempty"` + Value uint32 `protobuf:"varint,1,opt,name=value" json:"value,omitempty"` + XXX_NoUnkeyedLiteral struct{} `json:"-"` + XXX_unrecognized []byte `json:"-"` + XXX_sizecache int32 `json:"-"` } -func (m *Second) Reset() { *m = Second{} } -func (m *Second) String() string { return proto.CompactTextString(m) } -func (*Second) ProtoMessage() {} -func (*Second) Descriptor() ([]byte, []int) { return fileDescriptor0, []int{0} } +func (m *Second) Reset() { *m = Second{} } +func (m *Second) String() string { return proto.CompactTextString(m) } +func (*Second) ProtoMessage() {} +func (*Second) Descriptor() ([]byte, []int) { + return fileDescriptor_config_505638f2092d854e, []int{0} +} +func (m *Second) XXX_Unmarshal(b []byte) error { + return xxx_messageInfo_Second.Unmarshal(m, b) +} +func (m *Second) XXX_Marshal(b []byte, deterministic bool) ([]byte, error) { + return xxx_messageInfo_Second.Marshal(b, m, deterministic) +} +func (dst *Second) XXX_Merge(src proto.Message) { + xxx_messageInfo_Second.Merge(dst, src) +} +func (m *Second) XXX_Size() int { + return xxx_messageInfo_Second.Size(m) +} +func (m *Second) XXX_DiscardUnknown() { + xxx_messageInfo_Second.DiscardUnknown(m) +} + +var xxx_messageInfo_Second proto.InternalMessageInfo func (m *Second) GetValue() uint32 { if m != nil { @@ -32,14 +54,37 @@ func (m *Second) GetValue() uint32 { } type Policy struct { - Timeout *Policy_Timeout `protobuf:"bytes,1,opt,name=timeout" json:"timeout,omitempty"` - Stats *Policy_Stats `protobuf:"bytes,2,opt,name=stats" json:"stats,omitempty"` + Timeout *Policy_Timeout `protobuf:"bytes,1,opt,name=timeout" json:"timeout,omitempty"` + Stats *Policy_Stats `protobuf:"bytes,2,opt,name=stats" json:"stats,omitempty"` + Buffer *Policy_Buffer `protobuf:"bytes,3,opt,name=buffer" json:"buffer,omitempty"` + XXX_NoUnkeyedLiteral struct{} `json:"-"` + XXX_unrecognized []byte `json:"-"` + XXX_sizecache int32 `json:"-"` } -func (m *Policy) Reset() { *m = Policy{} } -func (m *Policy) String() string { return proto.CompactTextString(m) } -func (*Policy) ProtoMessage() {} -func (*Policy) Descriptor() ([]byte, []int) { return fileDescriptor0, []int{1} } +func (m *Policy) Reset() { *m = Policy{} } +func (m *Policy) String() string { return proto.CompactTextString(m) } +func (*Policy) ProtoMessage() {} +func (*Policy) Descriptor() ([]byte, []int) { + return fileDescriptor_config_505638f2092d854e, []int{1} +} +func (m *Policy) XXX_Unmarshal(b []byte) error { + return xxx_messageInfo_Policy.Unmarshal(m, b) +} +func (m *Policy) XXX_Marshal(b []byte, deterministic bool) ([]byte, error) { + return xxx_messageInfo_Policy.Marshal(b, m, deterministic) +} +func (dst *Policy) XXX_Merge(src proto.Message) { + xxx_messageInfo_Policy.Merge(dst, src) +} +func (m *Policy) XXX_Size() int { + return xxx_messageInfo_Policy.Size(m) +} +func (m *Policy) XXX_DiscardUnknown() { + xxx_messageInfo_Policy.DiscardUnknown(m) +} + +var xxx_messageInfo_Policy proto.InternalMessageInfo func (m *Policy) GetTimeout() *Policy_Timeout { if m != nil { @@ -55,18 +100,47 @@ func (m *Policy) GetStats() *Policy_Stats { return nil } -// Timeout is a message for timeout settings in various stages, in seconds. -type Policy_Timeout struct { - Handshake *Second `protobuf:"bytes,1,opt,name=handshake" json:"handshake,omitempty"` - ConnectionIdle *Second `protobuf:"bytes,2,opt,name=connection_idle,json=connectionIdle" json:"connection_idle,omitempty"` - UplinkOnly *Second `protobuf:"bytes,3,opt,name=uplink_only,json=uplinkOnly" json:"uplink_only,omitempty"` - DownlinkOnly *Second `protobuf:"bytes,4,opt,name=downlink_only,json=downlinkOnly" json:"downlink_only,omitempty"` +func (m *Policy) GetBuffer() *Policy_Buffer { + if m != nil { + return m.Buffer + } + return nil } -func (m *Policy_Timeout) Reset() { *m = Policy_Timeout{} } -func (m *Policy_Timeout) String() string { return proto.CompactTextString(m) } -func (*Policy_Timeout) ProtoMessage() {} -func (*Policy_Timeout) Descriptor() ([]byte, []int) { return fileDescriptor0, []int{1, 0} } +// Timeout is a message for timeout settings in various stages, in seconds. +type Policy_Timeout struct { + Handshake *Second `protobuf:"bytes,1,opt,name=handshake" json:"handshake,omitempty"` + ConnectionIdle *Second `protobuf:"bytes,2,opt,name=connection_idle,json=connectionIdle" json:"connection_idle,omitempty"` + UplinkOnly *Second `protobuf:"bytes,3,opt,name=uplink_only,json=uplinkOnly" json:"uplink_only,omitempty"` + DownlinkOnly *Second `protobuf:"bytes,4,opt,name=downlink_only,json=downlinkOnly" json:"downlink_only,omitempty"` + XXX_NoUnkeyedLiteral struct{} `json:"-"` + XXX_unrecognized []byte `json:"-"` + XXX_sizecache int32 `json:"-"` +} + +func (m *Policy_Timeout) Reset() { *m = Policy_Timeout{} } +func (m *Policy_Timeout) String() string { return proto.CompactTextString(m) } +func (*Policy_Timeout) ProtoMessage() {} +func (*Policy_Timeout) Descriptor() ([]byte, []int) { + return fileDescriptor_config_505638f2092d854e, []int{1, 0} +} +func (m *Policy_Timeout) XXX_Unmarshal(b []byte) error { + return xxx_messageInfo_Policy_Timeout.Unmarshal(m, b) +} +func (m *Policy_Timeout) XXX_Marshal(b []byte, deterministic bool) ([]byte, error) { + return xxx_messageInfo_Policy_Timeout.Marshal(b, m, deterministic) +} +func (dst *Policy_Timeout) XXX_Merge(src proto.Message) { + xxx_messageInfo_Policy_Timeout.Merge(dst, src) +} +func (m *Policy_Timeout) XXX_Size() int { + return xxx_messageInfo_Policy_Timeout.Size(m) +} +func (m *Policy_Timeout) XXX_DiscardUnknown() { + xxx_messageInfo_Policy_Timeout.DiscardUnknown(m) +} + +var xxx_messageInfo_Policy_Timeout proto.InternalMessageInfo func (m *Policy_Timeout) GetHandshake() *Second { if m != nil { @@ -97,14 +171,36 @@ func (m *Policy_Timeout) GetDownlinkOnly() *Second { } type Policy_Stats struct { - UserUplink bool `protobuf:"varint,1,opt,name=user_uplink,json=userUplink" json:"user_uplink,omitempty"` - UserDownlink bool `protobuf:"varint,2,opt,name=user_downlink,json=userDownlink" json:"user_downlink,omitempty"` + UserUplink bool `protobuf:"varint,1,opt,name=user_uplink,json=userUplink" json:"user_uplink,omitempty"` + UserDownlink bool `protobuf:"varint,2,opt,name=user_downlink,json=userDownlink" json:"user_downlink,omitempty"` + XXX_NoUnkeyedLiteral struct{} `json:"-"` + XXX_unrecognized []byte `json:"-"` + XXX_sizecache int32 `json:"-"` } -func (m *Policy_Stats) Reset() { *m = Policy_Stats{} } -func (m *Policy_Stats) String() string { return proto.CompactTextString(m) } -func (*Policy_Stats) ProtoMessage() {} -func (*Policy_Stats) Descriptor() ([]byte, []int) { return fileDescriptor0, []int{1, 1} } +func (m *Policy_Stats) Reset() { *m = Policy_Stats{} } +func (m *Policy_Stats) String() string { return proto.CompactTextString(m) } +func (*Policy_Stats) ProtoMessage() {} +func (*Policy_Stats) Descriptor() ([]byte, []int) { + return fileDescriptor_config_505638f2092d854e, []int{1, 1} +} +func (m *Policy_Stats) XXX_Unmarshal(b []byte) error { + return xxx_messageInfo_Policy_Stats.Unmarshal(m, b) +} +func (m *Policy_Stats) XXX_Marshal(b []byte, deterministic bool) ([]byte, error) { + return xxx_messageInfo_Policy_Stats.Marshal(b, m, deterministic) +} +func (dst *Policy_Stats) XXX_Merge(src proto.Message) { + xxx_messageInfo_Policy_Stats.Merge(dst, src) +} +func (m *Policy_Stats) XXX_Size() int { + return xxx_messageInfo_Policy_Stats.Size(m) +} +func (m *Policy_Stats) XXX_DiscardUnknown() { + xxx_messageInfo_Policy_Stats.DiscardUnknown(m) +} + +var xxx_messageInfo_Policy_Stats proto.InternalMessageInfo func (m *Policy_Stats) GetUserUplink() bool { if m != nil { @@ -120,14 +216,82 @@ func (m *Policy_Stats) GetUserDownlink() bool { return false } -type SystemPolicy struct { - Stats *SystemPolicy_Stats `protobuf:"bytes,1,opt,name=stats" json:"stats,omitempty"` +type Policy_Buffer struct { + Enabled bool `protobuf:"varint,1,opt,name=enabled" json:"enabled,omitempty"` + Size uint32 `protobuf:"varint,2,opt,name=size" json:"size,omitempty"` + XXX_NoUnkeyedLiteral struct{} `json:"-"` + XXX_unrecognized []byte `json:"-"` + XXX_sizecache int32 `json:"-"` } -func (m *SystemPolicy) Reset() { *m = SystemPolicy{} } -func (m *SystemPolicy) String() string { return proto.CompactTextString(m) } -func (*SystemPolicy) ProtoMessage() {} -func (*SystemPolicy) Descriptor() ([]byte, []int) { return fileDescriptor0, []int{2} } +func (m *Policy_Buffer) Reset() { *m = Policy_Buffer{} } +func (m *Policy_Buffer) String() string { return proto.CompactTextString(m) } +func (*Policy_Buffer) ProtoMessage() {} +func (*Policy_Buffer) Descriptor() ([]byte, []int) { + return fileDescriptor_config_505638f2092d854e, []int{1, 2} +} +func (m *Policy_Buffer) XXX_Unmarshal(b []byte) error { + return xxx_messageInfo_Policy_Buffer.Unmarshal(m, b) +} +func (m *Policy_Buffer) XXX_Marshal(b []byte, deterministic bool) ([]byte, error) { + return xxx_messageInfo_Policy_Buffer.Marshal(b, m, deterministic) +} +func (dst *Policy_Buffer) XXX_Merge(src proto.Message) { + xxx_messageInfo_Policy_Buffer.Merge(dst, src) +} +func (m *Policy_Buffer) XXX_Size() int { + return xxx_messageInfo_Policy_Buffer.Size(m) +} +func (m *Policy_Buffer) XXX_DiscardUnknown() { + xxx_messageInfo_Policy_Buffer.DiscardUnknown(m) +} + +var xxx_messageInfo_Policy_Buffer proto.InternalMessageInfo + +func (m *Policy_Buffer) GetEnabled() bool { + if m != nil { + return m.Enabled + } + return false +} + +func (m *Policy_Buffer) GetSize() uint32 { + if m != nil { + return m.Size + } + return 0 +} + +type SystemPolicy struct { + Stats *SystemPolicy_Stats `protobuf:"bytes,1,opt,name=stats" json:"stats,omitempty"` + XXX_NoUnkeyedLiteral struct{} `json:"-"` + XXX_unrecognized []byte `json:"-"` + XXX_sizecache int32 `json:"-"` +} + +func (m *SystemPolicy) Reset() { *m = SystemPolicy{} } +func (m *SystemPolicy) String() string { return proto.CompactTextString(m) } +func (*SystemPolicy) ProtoMessage() {} +func (*SystemPolicy) Descriptor() ([]byte, []int) { + return fileDescriptor_config_505638f2092d854e, []int{2} +} +func (m *SystemPolicy) XXX_Unmarshal(b []byte) error { + return xxx_messageInfo_SystemPolicy.Unmarshal(m, b) +} +func (m *SystemPolicy) XXX_Marshal(b []byte, deterministic bool) ([]byte, error) { + return xxx_messageInfo_SystemPolicy.Marshal(b, m, deterministic) +} +func (dst *SystemPolicy) XXX_Merge(src proto.Message) { + xxx_messageInfo_SystemPolicy.Merge(dst, src) +} +func (m *SystemPolicy) XXX_Size() int { + return xxx_messageInfo_SystemPolicy.Size(m) +} +func (m *SystemPolicy) XXX_DiscardUnknown() { + xxx_messageInfo_SystemPolicy.DiscardUnknown(m) +} + +var xxx_messageInfo_SystemPolicy proto.InternalMessageInfo func (m *SystemPolicy) GetStats() *SystemPolicy_Stats { if m != nil { @@ -137,14 +301,36 @@ func (m *SystemPolicy) GetStats() *SystemPolicy_Stats { } type SystemPolicy_Stats struct { - InboundUplink bool `protobuf:"varint,1,opt,name=inbound_uplink,json=inboundUplink" json:"inbound_uplink,omitempty"` - InboundDownlink bool `protobuf:"varint,2,opt,name=inbound_downlink,json=inboundDownlink" json:"inbound_downlink,omitempty"` + InboundUplink bool `protobuf:"varint,1,opt,name=inbound_uplink,json=inboundUplink" json:"inbound_uplink,omitempty"` + InboundDownlink bool `protobuf:"varint,2,opt,name=inbound_downlink,json=inboundDownlink" json:"inbound_downlink,omitempty"` + XXX_NoUnkeyedLiteral struct{} `json:"-"` + XXX_unrecognized []byte `json:"-"` + XXX_sizecache int32 `json:"-"` } -func (m *SystemPolicy_Stats) Reset() { *m = SystemPolicy_Stats{} } -func (m *SystemPolicy_Stats) String() string { return proto.CompactTextString(m) } -func (*SystemPolicy_Stats) ProtoMessage() {} -func (*SystemPolicy_Stats) Descriptor() ([]byte, []int) { return fileDescriptor0, []int{2, 0} } +func (m *SystemPolicy_Stats) Reset() { *m = SystemPolicy_Stats{} } +func (m *SystemPolicy_Stats) String() string { return proto.CompactTextString(m) } +func (*SystemPolicy_Stats) ProtoMessage() {} +func (*SystemPolicy_Stats) Descriptor() ([]byte, []int) { + return fileDescriptor_config_505638f2092d854e, []int{2, 0} +} +func (m *SystemPolicy_Stats) XXX_Unmarshal(b []byte) error { + return xxx_messageInfo_SystemPolicy_Stats.Unmarshal(m, b) +} +func (m *SystemPolicy_Stats) XXX_Marshal(b []byte, deterministic bool) ([]byte, error) { + return xxx_messageInfo_SystemPolicy_Stats.Marshal(b, m, deterministic) +} +func (dst *SystemPolicy_Stats) XXX_Merge(src proto.Message) { + xxx_messageInfo_SystemPolicy_Stats.Merge(dst, src) +} +func (m *SystemPolicy_Stats) XXX_Size() int { + return xxx_messageInfo_SystemPolicy_Stats.Size(m) +} +func (m *SystemPolicy_Stats) XXX_DiscardUnknown() { + xxx_messageInfo_SystemPolicy_Stats.DiscardUnknown(m) +} + +var xxx_messageInfo_SystemPolicy_Stats proto.InternalMessageInfo func (m *SystemPolicy_Stats) GetInboundUplink() bool { if m != nil { @@ -161,14 +347,36 @@ func (m *SystemPolicy_Stats) GetInboundDownlink() bool { } type Config struct { - Level map[uint32]*Policy `protobuf:"bytes,1,rep,name=level" json:"level,omitempty" protobuf_key:"varint,1,opt,name=key" protobuf_val:"bytes,2,opt,name=value"` - System *SystemPolicy `protobuf:"bytes,2,opt,name=system" json:"system,omitempty"` + Level map[uint32]*Policy `protobuf:"bytes,1,rep,name=level" json:"level,omitempty" protobuf_key:"varint,1,opt,name=key" protobuf_val:"bytes,2,opt,name=value"` + System *SystemPolicy `protobuf:"bytes,2,opt,name=system" json:"system,omitempty"` + XXX_NoUnkeyedLiteral struct{} `json:"-"` + XXX_unrecognized []byte `json:"-"` + XXX_sizecache int32 `json:"-"` } -func (m *Config) Reset() { *m = Config{} } -func (m *Config) String() string { return proto.CompactTextString(m) } -func (*Config) ProtoMessage() {} -func (*Config) Descriptor() ([]byte, []int) { return fileDescriptor0, []int{3} } +func (m *Config) Reset() { *m = Config{} } +func (m *Config) String() string { return proto.CompactTextString(m) } +func (*Config) ProtoMessage() {} +func (*Config) Descriptor() ([]byte, []int) { + return fileDescriptor_config_505638f2092d854e, []int{3} +} +func (m *Config) XXX_Unmarshal(b []byte) error { + return xxx_messageInfo_Config.Unmarshal(m, b) +} +func (m *Config) XXX_Marshal(b []byte, deterministic bool) ([]byte, error) { + return xxx_messageInfo_Config.Marshal(b, m, deterministic) +} +func (dst *Config) XXX_Merge(src proto.Message) { + xxx_messageInfo_Config.Merge(dst, src) +} +func (m *Config) XXX_Size() int { + return xxx_messageInfo_Config.Size(m) +} +func (m *Config) XXX_DiscardUnknown() { + xxx_messageInfo_Config.DiscardUnknown(m) +} + +var xxx_messageInfo_Config proto.InternalMessageInfo func (m *Config) GetLevel() map[uint32]*Policy { if m != nil { @@ -189,43 +397,50 @@ func init() { proto.RegisterType((*Policy)(nil), "v2ray.core.app.policy.Policy") proto.RegisterType((*Policy_Timeout)(nil), "v2ray.core.app.policy.Policy.Timeout") proto.RegisterType((*Policy_Stats)(nil), "v2ray.core.app.policy.Policy.Stats") + proto.RegisterType((*Policy_Buffer)(nil), "v2ray.core.app.policy.Policy.Buffer") proto.RegisterType((*SystemPolicy)(nil), "v2ray.core.app.policy.SystemPolicy") proto.RegisterType((*SystemPolicy_Stats)(nil), "v2ray.core.app.policy.SystemPolicy.Stats") proto.RegisterType((*Config)(nil), "v2ray.core.app.policy.Config") + proto.RegisterMapType((map[uint32]*Policy)(nil), "v2ray.core.app.policy.Config.LevelEntry") } -func init() { proto.RegisterFile("v2ray.com/core/app/policy/config.proto", fileDescriptor0) } +func init() { + proto.RegisterFile("v2ray.com/core/app/policy/config.proto", fileDescriptor_config_505638f2092d854e) +} -var fileDescriptor0 = []byte{ - // 478 bytes of a gzipped FileDescriptorProto - 0x1f, 0x8b, 0x08, 0x00, 0x00, 0x00, 0x00, 0x00, 0x02, 0xff, 0x8c, 0x93, 0xdf, 0x6e, 0xd3, 0x30, - 0x18, 0xc5, 0x95, 0x96, 0x66, 0xe3, 0x6b, 0xbb, 0x4d, 0x16, 0x93, 0x4a, 0x25, 0x60, 0xea, 0x34, - 0xd4, 0xdd, 0xb8, 0x52, 0x76, 0x03, 0x4c, 0x0c, 0x31, 0xfe, 0x48, 0x48, 0x20, 0x26, 0x97, 0x3f, - 0x82, 0x9b, 0xca, 0x73, 0x0c, 0x8b, 0xea, 0xd8, 0x56, 0xe2, 0x14, 0xe5, 0x35, 0x78, 0x8c, 0x3d, - 0x14, 0xd7, 0x3c, 0x06, 0x8a, 0xed, 0x2c, 0x1b, 0x5a, 0xb7, 0xde, 0x25, 0x47, 0xbf, 0x73, 0x74, - 0x3e, 0xdb, 0x1f, 0x3c, 0x5e, 0x44, 0x19, 0x2d, 0x31, 0x53, 0xe9, 0x84, 0xa9, 0x8c, 0x4f, 0xa8, - 0xd6, 0x13, 0xad, 0x44, 0xc2, 0xca, 0x09, 0x53, 0xf2, 0x47, 0xf2, 0x13, 0xeb, 0x4c, 0x19, 0x85, - 0xb6, 0x6b, 0x2e, 0xe3, 0x98, 0x6a, 0x8d, 0x1d, 0x33, 0x7a, 0x08, 0xe1, 0x94, 0x33, 0x25, 0x63, - 0x74, 0x0f, 0x3a, 0x0b, 0x2a, 0x0a, 0x3e, 0x08, 0x76, 0x82, 0x71, 0x9f, 0xb8, 0x9f, 0xd1, 0xdf, - 0x36, 0x84, 0x27, 0x16, 0x45, 0x2f, 0x60, 0xcd, 0x24, 0x29, 0x57, 0x85, 0xb1, 0x48, 0x37, 0xda, - 0xc3, 0xd7, 0x66, 0x62, 0xc7, 0xe3, 0x4f, 0x0e, 0x26, 0xb5, 0x0b, 0x3d, 0x85, 0x4e, 0x6e, 0xa8, - 0xc9, 0x07, 0x2d, 0x6b, 0xdf, 0xbd, 0xd9, 0x3e, 0xad, 0x50, 0xe2, 0x1c, 0xc3, 0xdf, 0x2d, 0x58, - 0xf3, 0x79, 0xe8, 0x10, 0xee, 0x9e, 0x51, 0x19, 0xe7, 0x67, 0x74, 0xce, 0x7d, 0x93, 0x07, 0x4b, - 0xa2, 0xdc, 0x68, 0xa4, 0xe1, 0xd1, 0x5b, 0xd8, 0x64, 0x4a, 0x4a, 0xce, 0x4c, 0xa2, 0xe4, 0x2c, - 0x89, 0x05, 0xf7, 0x6d, 0x6e, 0x89, 0xd8, 0x68, 0x5c, 0xef, 0x62, 0xc1, 0xd1, 0x11, 0x74, 0x0b, - 0x2d, 0x12, 0x39, 0x9f, 0x29, 0x29, 0xca, 0x41, 0x7b, 0x95, 0x0c, 0x70, 0x8e, 0x8f, 0x52, 0x94, - 0xe8, 0x18, 0xfa, 0xb1, 0xfa, 0x25, 0x9b, 0x84, 0x3b, 0xab, 0x24, 0xf4, 0x6a, 0x4f, 0x95, 0x31, - 0xfc, 0x00, 0x1d, 0x7b, 0x48, 0xe8, 0x11, 0x74, 0x8b, 0x9c, 0x67, 0x33, 0x97, 0x6f, 0xcf, 0x64, - 0x9d, 0x40, 0x25, 0x7d, 0xb6, 0x0a, 0xda, 0x85, 0xbe, 0x05, 0x6a, 0xbb, 0x9d, 0x79, 0x9d, 0xf4, - 0x2a, 0xf1, 0xb5, 0xd7, 0x46, 0xe7, 0x01, 0xf4, 0xa6, 0x65, 0x6e, 0x78, 0x7a, 0x71, 0xe1, 0xfe, - 0xbe, 0xdc, 0x21, 0xef, 0x2f, 0xeb, 0x76, 0xc9, 0x73, 0xf5, 0xd6, 0xbe, 0xd5, 0x05, 0xf7, 0x60, - 0x23, 0x91, 0xa7, 0xaa, 0x90, 0xf1, 0xd5, 0x8e, 0x7d, 0xaf, 0xfa, 0x9a, 0xfb, 0xb0, 0x55, 0x63, - 0xff, 0x35, 0xdd, 0xf4, 0xfa, 0x45, 0xd9, 0x3f, 0x01, 0x84, 0xaf, 0xec, 0xfb, 0x46, 0x47, 0xd0, - 0x11, 0x7c, 0xc1, 0xc5, 0x20, 0xd8, 0x69, 0x8f, 0xbb, 0xd1, 0x78, 0x49, 0x4d, 0x47, 0xe3, 0xf7, - 0x15, 0xfa, 0x46, 0x9a, 0xac, 0x24, 0xce, 0x86, 0x0e, 0x21, 0xcc, 0xed, 0x08, 0xb7, 0xbc, 0xcb, - 0xcb, 0x73, 0x12, 0x6f, 0x19, 0x7e, 0x05, 0x68, 0x12, 0xd1, 0x16, 0xb4, 0xe7, 0xbc, 0xf4, 0x1b, - 0x54, 0x7d, 0xa2, 0x83, 0x7a, 0xab, 0x6e, 0x7e, 0x65, 0x3e, 0xd5, 0xb1, 0xcf, 0x5a, 0x4f, 0x82, - 0xe3, 0xe7, 0x70, 0x9f, 0xa9, 0xf4, 0x7a, 0xfc, 0x24, 0xf8, 0x1e, 0xba, 0xaf, 0xf3, 0xd6, 0xf6, - 0x97, 0x88, 0xd0, 0x6a, 0xba, 0x8c, 0xe3, 0x97, 0x5a, 0xfb, 0xa4, 0xd3, 0xd0, 0x6e, 0xfd, 0xc1, - 0xbf, 0x00, 0x00, 0x00, 0xff, 0xff, 0xa9, 0x9f, 0x00, 0x31, 0x1f, 0x04, 0x00, 0x00, +var fileDescriptor_config_505638f2092d854e = []byte{ + // 523 bytes of a gzipped FileDescriptorProto + 0x1f, 0x8b, 0x08, 0x00, 0x00, 0x00, 0x00, 0x00, 0x02, 0xff, 0x8c, 0x94, 0xeb, 0x6a, 0x13, 0x41, + 0x14, 0xc7, 0xd9, 0x5c, 0x36, 0xf5, 0x24, 0xdb, 0x96, 0xc1, 0xc2, 0xba, 0xa0, 0x96, 0xd4, 0x4a, + 0xfa, 0x65, 0x03, 0x29, 0x88, 0x5a, 0xad, 0x18, 0x2f, 0x20, 0x28, 0x96, 0x89, 0x17, 0xf4, 0x4b, + 0xd8, 0xec, 0x9e, 0xd8, 0x25, 0x93, 0x99, 0x65, 0x2f, 0x91, 0xf5, 0x31, 0x7c, 0x8c, 0x3e, 0x54, + 0x9f, 0x45, 0x76, 0x2e, 0xa6, 0x95, 0x26, 0xf1, 0xdb, 0xcc, 0xe1, 0xf7, 0xff, 0x33, 0xff, 0xb3, + 0xe7, 0x2c, 0x3c, 0x5c, 0x0c, 0xd2, 0xa0, 0xf4, 0x43, 0x31, 0xef, 0x87, 0x22, 0xc5, 0x7e, 0x90, + 0x24, 0xfd, 0x44, 0xb0, 0x38, 0x2c, 0xfb, 0xa1, 0xe0, 0xd3, 0xf8, 0x87, 0x9f, 0xa4, 0x22, 0x17, + 0x64, 0xcf, 0x70, 0x29, 0xfa, 0x41, 0x92, 0xf8, 0x8a, 0xe9, 0xde, 0x03, 0x7b, 0x84, 0xa1, 0xe0, + 0x11, 0xb9, 0x0d, 0xcd, 0x45, 0xc0, 0x0a, 0x74, 0xad, 0x7d, 0xab, 0xe7, 0x50, 0x75, 0xe9, 0x5e, + 0x36, 0xc0, 0x3e, 0x93, 0x28, 0x79, 0x01, 0xad, 0x3c, 0x9e, 0xa3, 0x28, 0x72, 0x89, 0xb4, 0x07, + 0x87, 0xfe, 0x8d, 0x9e, 0xbe, 0xe2, 0xfd, 0x4f, 0x0a, 0xa6, 0x46, 0x45, 0x9e, 0x40, 0x33, 0xcb, + 0x83, 0x3c, 0x73, 0x6b, 0x52, 0x7e, 0xb0, 0x5e, 0x3e, 0xaa, 0x50, 0xaa, 0x14, 0xe4, 0x19, 0xd8, + 0x93, 0x62, 0x3a, 0xc5, 0xd4, 0xad, 0x4b, 0xed, 0x83, 0xf5, 0xda, 0xa1, 0x64, 0xa9, 0xd6, 0x78, + 0xbf, 0x6b, 0xd0, 0xd2, 0xaf, 0x21, 0x27, 0x70, 0xeb, 0x3c, 0xe0, 0x51, 0x76, 0x1e, 0xcc, 0x50, + 0xe7, 0xb8, 0xbb, 0xc2, 0x4c, 0x35, 0x86, 0x2e, 0x79, 0xf2, 0x16, 0x76, 0x42, 0xc1, 0x39, 0x86, + 0x79, 0x2c, 0xf8, 0x38, 0x8e, 0x18, 0xea, 0x2c, 0x1b, 0x2c, 0xb6, 0x97, 0xaa, 0x77, 0x11, 0x43, + 0x72, 0x0a, 0xed, 0x22, 0x61, 0x31, 0x9f, 0x8d, 0x05, 0x67, 0xa5, 0xce, 0xb4, 0xc1, 0x03, 0x94, + 0xe2, 0x23, 0x67, 0x25, 0x19, 0x82, 0x13, 0x89, 0x9f, 0x7c, 0xe9, 0xd0, 0xf8, 0x1f, 0x87, 0x8e, + 0xd1, 0x54, 0x1e, 0xde, 0x07, 0x68, 0xca, 0x16, 0x93, 0xfb, 0xd0, 0x2e, 0x32, 0x4c, 0xc7, 0xca, + 0x5f, 0xf6, 0x64, 0x8b, 0x42, 0x55, 0xfa, 0x2c, 0x2b, 0xe4, 0x00, 0x1c, 0x09, 0x18, 0xb9, 0xcc, + 0xbc, 0x45, 0x3b, 0x55, 0xf1, 0xb5, 0xae, 0x79, 0x8f, 0xc0, 0x56, 0x5d, 0x27, 0x2e, 0xb4, 0x90, + 0x07, 0x13, 0x86, 0x91, 0xf6, 0x32, 0x57, 0x42, 0xa0, 0x91, 0xc5, 0xbf, 0x54, 0xcf, 0x1c, 0x2a, + 0xcf, 0xdd, 0x0b, 0x0b, 0x3a, 0xa3, 0x32, 0xcb, 0x71, 0xfe, 0x77, 0xcc, 0xf4, 0x94, 0xa8, 0x8f, + 0x73, 0xb4, 0x2a, 0xd3, 0x15, 0xcd, 0xb5, 0x59, 0xf1, 0xbe, 0x99, 0x60, 0x87, 0xb0, 0x1d, 0xf3, + 0x89, 0x28, 0x78, 0x74, 0x3d, 0x9b, 0xa3, 0xab, 0x3a, 0xde, 0x11, 0xec, 0x1a, 0xec, 0x9f, 0x84, + 0x3b, 0xba, 0x6e, 0x42, 0x76, 0x2f, 0x2d, 0xb0, 0x5f, 0xc9, 0xad, 0x22, 0xa7, 0xd0, 0x64, 0xb8, + 0x40, 0xe6, 0x5a, 0xfb, 0xf5, 0x5e, 0x7b, 0xd0, 0x5b, 0xf1, 0x4c, 0x45, 0xfb, 0xef, 0x2b, 0xf4, + 0x0d, 0xcf, 0xd3, 0x92, 0x2a, 0x19, 0x39, 0x01, 0x3b, 0x93, 0x11, 0x36, 0x6c, 0xc3, 0xd5, 0x9c, + 0x54, 0x4b, 0xbc, 0xaf, 0x00, 0x4b, 0x47, 0xb2, 0x0b, 0xf5, 0x19, 0x96, 0x7a, 0x6f, 0xab, 0x23, + 0x39, 0x36, 0xbb, 0xbc, 0x7e, 0x3a, 0xb5, 0xab, 0x62, 0x9f, 0xd6, 0x1e, 0x5b, 0xc3, 0xe7, 0x70, + 0x27, 0x14, 0xf3, 0x9b, 0xf1, 0x33, 0xeb, 0xbb, 0xad, 0x4e, 0x17, 0xb5, 0xbd, 0x2f, 0x03, 0x1a, + 0x54, 0xe9, 0x52, 0xf4, 0x5f, 0x26, 0x89, 0x76, 0x9a, 0xd8, 0xf2, 0x5f, 0x73, 0xfc, 0x27, 0x00, + 0x00, 0xff, 0xff, 0x2c, 0x2e, 0xe6, 0xcf, 0x95, 0x04, 0x00, 0x00, } diff --git a/app/policy/config.proto b/app/policy/config.proto index cf64db86..12c5b84e 100644 --- a/app/policy/config.proto +++ b/app/policy/config.proto @@ -24,8 +24,14 @@ message Policy { bool user_downlink = 2; } + message Buffer { + bool enabled = 1; + uint32 size = 2; + } + Timeout timeout = 1; Stats stats = 2; + Buffer buffer = 3; } message SystemPolicy { diff --git a/app/proxyman/mux/mux.go b/app/proxyman/mux/mux.go index 1571a916..d6925c2c 100644 --- a/app/proxyman/mux/mux.go +++ b/app/proxyman/mux/mux.go @@ -51,7 +51,7 @@ func (m *ClientManager) Dispatch(ctx context.Context, link *core.Link) error { } } - client, err := NewClient(m.proxy, m.dialer, m) + client, err := NewClient(ctx, m.proxy, m.dialer, m) if err != nil { return newError("failed to create client").Base(err) } @@ -86,11 +86,13 @@ var muxCoolAddress = net.DomainAddress("v1.mux.cool") var muxCoolPort = net.Port(9527) // NewClient creates a new mux.Client. -func NewClient(p proxy.Outbound, dialer proxy.Dialer, m *ClientManager) (*Client, error) { +func NewClient(pctx context.Context, p proxy.Outbound, dialer proxy.Dialer, m *ClientManager) (*Client, error) { ctx := proxy.ContextWithTarget(context.Background(), net.TCPDestination(muxCoolAddress, muxCoolPort)) ctx, cancel := context.WithCancel(ctx) - uplinkReader, upLinkWriter := pipe.New() - downlinkReader, downlinkWriter := pipe.New() + + opts := pipe.OptionsFromContext(pctx) + uplinkReader, upLinkWriter := pipe.New(opts...) + downlinkReader, downlinkWriter := pipe.New(opts...) c := &Client{ sessionManager: NewSessionManager(), @@ -307,8 +309,9 @@ func (s *Server) Dispatch(ctx context.Context, dest net.Destination) (*core.Link return s.dispatcher.Dispatch(ctx, dest) } - uplinkReader, uplinkWriter := pipe.New() - downlinkReader, downlinkWriter := pipe.New() + opts := pipe.OptionsFromContext(ctx) + uplinkReader, uplinkWriter := pipe.New(opts...) + downlinkReader, downlinkWriter := pipe.New(opts...) worker := &ServerWorker{ dispatcher: s.dispatcher, diff --git a/app/proxyman/outbound/handler.go b/app/proxyman/outbound/handler.go index 327ec16b..1a338559 100644 --- a/app/proxyman/outbound/handler.go +++ b/app/proxyman/outbound/handler.go @@ -102,8 +102,9 @@ func (h *Handler) Dial(ctx context.Context, dest net.Destination) (internet.Conn newError("proxying to ", tag, " for dest ", dest).AtDebug().WithContext(ctx).WriteToLog() ctx = proxy.ContextWithTarget(ctx, dest) - uplinkReader, uplinkWriter := pipe.New() - downlinkReader, downlinkWriter := pipe.New() + opts := pipe.OptionsFromContext(ctx) + uplinkReader, uplinkWriter := pipe.New(opts...) + downlinkReader, downlinkWriter := pipe.New(opts...) go handler.Dispatch(ctx, &core.Link{Reader: uplinkReader, Writer: downlinkWriter}) return net.NewConnection(net.ConnectionInputMulti(uplinkWriter), net.ConnectionOutputMulti(downlinkReader)), nil diff --git a/policy.go b/policy.go index 6fff4dfd..c04e2f72 100644 --- a/policy.go +++ b/policy.go @@ -1,10 +1,12 @@ package core import ( + "context" "sync" "time" "v2ray.com/core/common" + "v2ray.com/core/common/platform" ) // TimeoutPolicy contains limits for connection timeout. @@ -27,6 +29,14 @@ type StatsPolicy struct { UserDownlink bool } +// BufferPolicy contains settings for internal buffer. +type BufferPolicy struct { + // Whether or not to enable internal buffer. + Enabled bool + // Size of internal buffer, in bytes. + Size uint32 +} + type SystemStatsPolicy struct { // Whether or not to enable stat counter for uplink traffic in inbound handlers. InboundUplink bool @@ -35,13 +45,15 @@ type SystemStatsPolicy struct { } type SystemPolicy struct { - Stats SystemStatsPolicy + Stats SystemStatsPolicy + Buffer BufferPolicy } // Policy is session based settings for controlling V2Ray requests. It contains various settings (or limits) that may differ for different users in the context. type Policy struct { Timeouts TimeoutPolicy // Timeout settings Stats StatsPolicy + Buffer BufferPolicy } // PolicyManager is a feature that provides Policy for the given user by its id or level. @@ -55,6 +67,28 @@ type PolicyManager interface { ForSystem() SystemPolicy } +var defaultBufferSize uint32 = 10 * 1024 * 1024 + +func init() { + const key = "v2ray.ray.buffer.size" + size := platform.EnvFlag{ + Name: key, + AltName: platform.NormalizeEnvName(key), + }.GetValueAsInt(10) + if size == 0 { + defaultBufferSize = 2147483647 + } else { + defaultBufferSize = uint32(size) * 1024 * 1024 + } +} + +func defaultBufferPolicy() BufferPolicy { + return BufferPolicy{ + Enabled: true, + Size: defaultBufferSize, + } +} + // DefaultPolicy returns the Policy when user is not specified. func DefaultPolicy() Policy { return Policy{ @@ -68,9 +102,28 @@ func DefaultPolicy() Policy { UserUplink: false, UserDownlink: false, }, + Buffer: defaultBufferPolicy(), } } +type policyKey int + +const ( + bufferPolicyKey policyKey = 0 +) + +func ContextWithBufferPolicy(ctx context.Context, p BufferPolicy) context.Context { + return context.WithValue(ctx, bufferPolicyKey, p) +} + +func BufferPolicyFromContext(ctx context.Context) BufferPolicy { + pPolicy := ctx.Value(bufferPolicyKey) + if pPolicy == nil { + return defaultBufferPolicy() + } + return pPolicy.(BufferPolicy) +} + type syncPolicyManager struct { sync.RWMutex PolicyManager diff --git a/proxy/dokodemo/dokodemo.go b/proxy/dokodemo/dokodemo.go index c8d29692..2ea04217 100644 --- a/proxy/dokodemo/dokodemo.go +++ b/proxy/dokodemo/dokodemo.go @@ -68,9 +68,11 @@ func (d *DokodemoDoor) Process(ctx context.Context, network net.Network, conn in return newError("unable to get destination") } + plcy := d.policy() ctx, cancel := context.WithCancel(ctx) - timer := signal.CancelAfterInactivity(ctx, cancel, d.policy().Timeouts.ConnectionIdle) + timer := signal.CancelAfterInactivity(ctx, cancel, plcy.Timeouts.ConnectionIdle) + ctx = core.ContextWithBufferPolicy(ctx, plcy.Buffer) link, err := dispatcher.Dispatch(ctx, dest) if err != nil { return newError("failed to dispatch request").Base(err) @@ -78,7 +80,7 @@ func (d *DokodemoDoor) Process(ctx context.Context, network net.Network, conn in requestDone := func() error { defer common.Close(link.Writer) - defer timer.SetTimeout(d.policy().Timeouts.DownlinkOnly) + defer timer.SetTimeout(plcy.Timeouts.DownlinkOnly) chunkReader := buf.NewReader(conn) @@ -90,7 +92,7 @@ func (d *DokodemoDoor) Process(ctx context.Context, network net.Network, conn in } responseDone := func() error { - defer timer.SetTimeout(d.policy().Timeouts.UplinkOnly) + defer timer.SetTimeout(plcy.Timeouts.UplinkOnly) var writer buf.Writer if network == net.Network_TCP { diff --git a/proxy/http/server.go b/proxy/http/server.go index 39ae609c..f8ef71a2 100755 --- a/proxy/http/server.go +++ b/proxy/http/server.go @@ -170,8 +170,11 @@ func (s *Server) handleConnect(ctx context.Context, request *http.Request, reade return newError("failed to write back OK response").Base(err) } + plcy := s.policy() ctx, cancel := context.WithCancel(ctx) - timer := signal.CancelAfterInactivity(ctx, cancel, s.policy().Timeouts.ConnectionIdle) + timer := signal.CancelAfterInactivity(ctx, cancel, plcy.Timeouts.ConnectionIdle) + + ctx = core.ContextWithBufferPolicy(ctx, plcy.Buffer) link, err := dispatcher.Dispatch(ctx, dest) if err != nil { return err diff --git a/proxy/shadowsocks/server.go b/proxy/shadowsocks/server.go index 0aa80784..ad8f2f37 100644 --- a/proxy/shadowsocks/server.go +++ b/proxy/shadowsocks/server.go @@ -168,6 +168,8 @@ func (s *Server) handleConnection(ctx context.Context, conn internet.Connection, ctx, cancel := context.WithCancel(ctx) timer := signal.CancelAfterInactivity(ctx, cancel, sessionPolicy.Timeouts.ConnectionIdle) + + ctx = core.ContextWithBufferPolicy(ctx, sessionPolicy.Buffer) link, err := dispatcher.Dispatch(ctx, dest) if err != nil { return err diff --git a/proxy/socks/server.go b/proxy/socks/server.go index e1cd9916..89752e58 100644 --- a/proxy/socks/server.go +++ b/proxy/socks/server.go @@ -130,13 +130,15 @@ func (s *Server) transport(ctx context.Context, reader io.Reader, writer io.Writ ctx, cancel := context.WithCancel(ctx) timer := signal.CancelAfterInactivity(ctx, cancel, s.policy().Timeouts.ConnectionIdle) + plcy := s.policy() + ctx = core.ContextWithBufferPolicy(ctx, plcy.Buffer) link, err := dispatcher.Dispatch(ctx, dest) if err != nil { return err } requestDone := func() error { - defer timer.SetTimeout(s.policy().Timeouts.DownlinkOnly) + defer timer.SetTimeout(plcy.Timeouts.DownlinkOnly) defer common.Close(link.Writer) v2reader := buf.NewReader(reader) @@ -148,7 +150,7 @@ func (s *Server) transport(ctx context.Context, reader io.Reader, writer io.Writ } responseDone := func() error { - defer timer.SetTimeout(s.policy().Timeouts.UplinkOnly) + defer timer.SetTimeout(plcy.Timeouts.UplinkOnly) v2writer := buf.NewWriter(writer) if err := buf.Copy(link.Reader, v2writer, buf.UpdateActivity(timer)); err != nil { diff --git a/proxy/vmess/inbound/inbound.go b/proxy/vmess/inbound/inbound.go index eff75d28..936d70f9 100644 --- a/proxy/vmess/inbound/inbound.go +++ b/proxy/vmess/inbound/inbound.go @@ -272,6 +272,8 @@ func (h *Handler) Process(ctx context.Context, network net.Network, connection i ctx, cancel := context.WithCancel(ctx) timer := signal.CancelAfterInactivity(ctx, cancel, sessionPolicy.Timeouts.ConnectionIdle) + + ctx = core.ContextWithBufferPolicy(ctx, sessionPolicy.Buffer) link, err := dispatcher.Dispatch(ctx, request.Destination()) if err != nil { return newError("failed to dispatch request to ", request.Destination()).Base(err) diff --git a/testing/scenarios/policy_test.go b/testing/scenarios/policy_test.go index 93e4de3e..2a0e8cb1 100644 --- a/testing/scenarios/policy_test.go +++ b/testing/scenarios/policy_test.go @@ -1,13 +1,17 @@ package scenarios import ( + "crypto/rand" "io" + "sync" "testing" "time" "v2ray.com/core" + "v2ray.com/core/app/log" "v2ray.com/core/app/policy" "v2ray.com/core/app/proxyman" + clog "v2ray.com/core/common/log" "v2ray.com/core/common/net" "v2ray.com/core/common/protocol" "v2ray.com/core/common/serial" @@ -167,3 +171,135 @@ func TestVMessClosing(t *testing.T) { CloseAllServers(servers) } + +func TestZeroBuffer(t *testing.T) { + assert := With(t) + + tcpServer := tcp.Server{ + MsgProcessor: xor, + } + dest, err := tcpServer.Start() + assert(err, IsNil) + defer tcpServer.Close() + + userID := protocol.NewID(uuid.New()) + serverPort := tcp.PickPort() + serverConfig := &core.Config{ + App: []*serial.TypedMessage{ + serial.ToTypedMessage(&policy.Config{ + Level: map[uint32]*policy.Policy{ + 0: { + Timeout: &policy.Policy_Timeout{ + UplinkOnly: &policy.Second{Value: 0}, + DownlinkOnly: &policy.Second{Value: 0}, + }, + Buffer: &policy.Policy_Buffer{ + Enabled: false, + }, + }, + }, + }), + }, + Inbound: []*core.InboundHandlerConfig{ + { + ReceiverSettings: serial.ToTypedMessage(&proxyman.ReceiverConfig{ + PortRange: net.SinglePortRange(serverPort), + Listen: net.NewIPOrDomain(net.LocalHostIP), + }), + ProxySettings: serial.ToTypedMessage(&inbound.Config{ + User: []*protocol.User{ + { + Account: serial.ToTypedMessage(&vmess.Account{ + Id: userID.String(), + AlterId: 64, + }), + }, + }, + }), + }, + }, + Outbound: []*core.OutboundHandlerConfig{ + { + ProxySettings: serial.ToTypedMessage(&freedom.Config{}), + }, + }, + } + + clientPort := tcp.PickPort() + clientConfig := &core.Config{ + App: []*serial.TypedMessage{ + serial.ToTypedMessage(&log.Config{ + ErrorLogLevel: clog.Severity_Debug, + ErrorLogType: log.LogType_Console, + }), + }, + Inbound: []*core.InboundHandlerConfig{ + { + ReceiverSettings: serial.ToTypedMessage(&proxyman.ReceiverConfig{ + PortRange: net.SinglePortRange(clientPort), + Listen: net.NewIPOrDomain(net.LocalHostIP), + }), + ProxySettings: serial.ToTypedMessage(&dokodemo.Config{ + Address: net.NewIPOrDomain(dest.Address), + Port: uint32(dest.Port), + NetworkList: &net.NetworkList{ + Network: []net.Network{net.Network_TCP}, + }, + }), + }, + }, + Outbound: []*core.OutboundHandlerConfig{ + { + ProxySettings: serial.ToTypedMessage(&outbound.Config{ + Receiver: []*protocol.ServerEndpoint{ + { + Address: net.NewIPOrDomain(net.LocalHostIP), + Port: uint32(serverPort), + User: []*protocol.User{ + { + Account: serial.ToTypedMessage(&vmess.Account{ + Id: userID.String(), + AlterId: 64, + SecuritySettings: &protocol.SecurityConfig{ + Type: protocol.SecurityType_AES128_GCM, + }, + }), + }, + }, + }, + }, + }), + }, + }, + } + + servers, err := InitializeServerConfigs(serverConfig, clientConfig) + assert(err, IsNil) + + var wg sync.WaitGroup + wg.Add(10) + for i := 0; i < 10; i++ { + go func() { + conn, err := net.DialTCP("tcp", nil, &net.TCPAddr{ + IP: []byte{127, 0, 0, 1}, + Port: int(clientPort), + }) + assert(err, IsNil) + + payload := make([]byte, 10240*1024) + rand.Read(payload) + + nBytes, err := conn.Write([]byte(payload)) + assert(err, IsNil) + assert(nBytes, Equals, len(payload)) + + response := readFrom(conn, time.Second*20, 10240*1024) + assert(response, Equals, xor([]byte(payload))) + assert(conn.Close(), IsNil) + wg.Done() + }() + } + wg.Wait() + + CloseAllServers(servers) +} diff --git a/transport/internet/http/dialer.go b/transport/internet/http/dialer.go index 6cbe336e..ae8753b4 100644 --- a/transport/internet/http/dialer.go +++ b/transport/internet/http/dialer.go @@ -83,7 +83,8 @@ func Dial(ctx context.Context, dest net.Destination) (internet.Connection, error return nil, err } - preader, pwriter := pipe.New(pipe.WithSizeLimit(20 * 1024)) + opts := pipe.OptionsFromContext(ctx) + preader, pwriter := pipe.New(opts...) breader := &buf.BufferedReader{Reader: preader} request := &http.Request{ Method: "PUT", diff --git a/transport/pipe/pipe.go b/transport/pipe/pipe.go index 6c70ba2c..40dfc510 100644 --- a/transport/pipe/pipe.go +++ b/transport/pipe/pipe.go @@ -1,7 +1,9 @@ package pipe import ( - "v2ray.com/core/common/platform" + "context" + + "v2ray.com/core" "v2ray.com/core/common/signal" ) @@ -19,10 +21,23 @@ func WithSizeLimit(limit int32) Option { } } +func OptionsFromContext(ctx context.Context) []Option { + var opt []Option + + bp := core.BufferPolicyFromContext(ctx) + if bp.Enabled { + opt = append(opt, WithSizeLimit(int32(bp.Size))) + } else { + opt = append(opt, WithoutSizeLimit()) + } + + return opt +} + // New creates a new Reader and Writer that connects to each other. func New(opts ...Option) (*Reader, *Writer) { p := &pipe{ - limit: defaultLimit, + limit: 0, readSignal: signal.NewNotifier(), writeSignal: signal.NewNotifier(), } @@ -48,18 +63,3 @@ func CloseError(v interface{}) { c.CloseError() } } - -var defaultLimit int32 = 10 * 1024 * 1024 - -func init() { - const raySizeEnvKey = "v2ray.ray.buffer.size" - size := platform.EnvFlag{ - Name: raySizeEnvKey, - AltName: platform.NormalizeEnvName(raySizeEnvKey), - }.GetValueAsInt(10) - if size == 0 { - defaultLimit = 2147483647 - } else { - defaultLimit = int32(size) * 1024 * 1024 - } -}