policy for internal buffer size

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()
+	opt := pipe.OptionsFromContext(ctx)
-	downlinkReader, downlinkWriter := pipe.New()
+	uplinkReader, uplinkWriter := pipe.New(opt...)
+	downlinkReader, downlinkWriter := pipe.New(opt...)
 
 	inboundLink := &core.Link{
 		Reader: downlinkReader,

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
 }
 

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) Reset()         { *m = Second{} }
-func (m *Second) String() string            { return proto.CompactTextString(m) }
+func (m *Second) String() string { return proto.CompactTextString(m) }
-func (*Second) ProtoMessage()               {}
+func (*Second) ProtoMessage()    {}
-func (*Second) Descriptor() ([]byte, []int) { return fileDescriptor0, []int{0} }
+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"`
+	Timeout              *Policy_Timeout `protobuf:"bytes,1,opt,name=timeout" json:"timeout,omitempty"`
-	Stats   *Policy_Stats   `protobuf:"bytes,2,opt,name=stats" json:"stats,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) Reset()         { *m = Policy{} }
-func (m *Policy) String() string            { return proto.CompactTextString(m) }
+func (m *Policy) String() string { return proto.CompactTextString(m) }
-func (*Policy) ProtoMessage()               {}
+func (*Policy) ProtoMessage()    {}
-func (*Policy) Descriptor() ([]byte, []int) { return fileDescriptor0, []int{1} }
+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.
+func (m *Policy) GetBuffer() *Policy_Buffer {
-type Policy_Timeout struct {
+	if m != nil {
-	Handshake      *Second `protobuf:"bytes,1,opt,name=handshake" json:"handshake,omitempty"`
+		return m.Buffer
-	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"`
+	return nil
-	DownlinkOnly   *Second `protobuf:"bytes,4,opt,name=downlink_only,json=downlinkOnly" json:"downlink_only,omitempty"`
 }
 
-func (m *Policy_Timeout) Reset()                    { *m = Policy_Timeout{} }
+// Timeout is a message for timeout settings in various stages, in seconds.
-func (m *Policy_Timeout) String() string            { return proto.CompactTextString(m) }
+type Policy_Timeout struct {
-func (*Policy_Timeout) ProtoMessage()               {}
+	Handshake            *Second  `protobuf:"bytes,1,opt,name=handshake" json:"handshake,omitempty"`
-func (*Policy_Timeout) Descriptor() ([]byte, []int) { return fileDescriptor0, []int{1, 0} }
+	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"`
+	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"`
+	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) Reset()         { *m = Policy_Stats{} }
-func (m *Policy_Stats) String() string            { return proto.CompactTextString(m) }
+func (m *Policy_Stats) String() string { return proto.CompactTextString(m) }
-func (*Policy_Stats) ProtoMessage()               {}
+func (*Policy_Stats) ProtoMessage()    {}
-func (*Policy_Stats) Descriptor() ([]byte, []int) { return fileDescriptor0, []int{1, 1} }
+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 {
+type Policy_Buffer struct {
-	Stats *SystemPolicy_Stats `protobuf:"bytes,1,opt,name=stats" json:"stats,omitempty"`
+	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 *Policy_Buffer) Reset()         { *m = Policy_Buffer{} }
-func (m *SystemPolicy) String() string            { return proto.CompactTextString(m) }
+func (m *Policy_Buffer) String() string { return proto.CompactTextString(m) }
-func (*SystemPolicy) ProtoMessage()               {}
+func (*Policy_Buffer) ProtoMessage()    {}
-func (*SystemPolicy) Descriptor() ([]byte, []int) { return fileDescriptor0, []int{2} }
+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"`
+	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"`
+	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) Reset()         { *m = SystemPolicy_Stats{} }
-func (m *SystemPolicy_Stats) String() string            { return proto.CompactTextString(m) }
+func (m *SystemPolicy_Stats) String() string { return proto.CompactTextString(m) }
-func (*SystemPolicy_Stats) ProtoMessage()               {}
+func (*SystemPolicy_Stats) ProtoMessage()    {}
-func (*SystemPolicy_Stats) Descriptor() ([]byte, []int) { return fileDescriptor0, []int{2, 0} }
+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"`
+	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"`
+	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) Reset()         { *m = Config{} }
-func (m *Config) String() string            { return proto.CompactTextString(m) }
+func (m *Config) String() string { return proto.CompactTextString(m) }
-func (*Config) ProtoMessage()               {}
+func (*Config) ProtoMessage()    {}
-func (*Config) Descriptor() ([]byte, []int) { return fileDescriptor0, []int{3} }
+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,
+var fileDescriptor_config_505638f2092d854e = []byte{
-	0x18, 0xc5, 0x95, 0x96, 0x66, 0xe3, 0x6b, 0xbb, 0x4d, 0x16, 0x93, 0x4a, 0x25, 0x60, 0xea, 0x34,
+	// 523 bytes of a gzipped FileDescriptorProto
-	0xd4, 0xdd, 0xb8, 0x52, 0x76, 0x03, 0x4c, 0x0c, 0x31, 0xfe, 0x48, 0x48, 0x20, 0x26, 0x97, 0x3f,
+	0x1f, 0x8b, 0x08, 0x00, 0x00, 0x00, 0x00, 0x00, 0x02, 0xff, 0x8c, 0x94, 0xeb, 0x6a, 0x13, 0x41,
-	0x82, 0x9b, 0xca, 0x73, 0x0c, 0x8b, 0xea, 0xd8, 0x56, 0xe2, 0x14, 0xe5, 0x35, 0x78, 0x8c, 0x3d,
+	0x14, 0xc7, 0xd9, 0x5c, 0x36, 0xf5, 0x24, 0xdb, 0x96, 0xc1, 0xc2, 0xba, 0xa0, 0x96, 0xd4, 0x4a,
-	0x14, 0xd7, 0x3c, 0x06, 0x8a, 0xed, 0x2c, 0x1b, 0x5a, 0xb7, 0xde, 0x25, 0x47, 0xbf, 0x73, 0x74,
+	0xfa, 0x65, 0x03, 0x29, 0x88, 0x5a, 0xad, 0x18, 0x2f, 0x20, 0x28, 0x96, 0x89, 0x17, 0xf4, 0x4b,
-	0x3e, 0xdb, 0x1f, 0x3c, 0x5e, 0x44, 0x19, 0x2d, 0x31, 0x53, 0xe9, 0x84, 0xa9, 0x8c, 0x4f, 0xa8,
+	0xd8, 0xec, 0x9e, 0xd8, 0x25, 0x93, 0x99, 0x65, 0x2f, 0x91, 0xf5, 0x31, 0x7c, 0x8c, 0x3e, 0x54,
-	0xd6, 0x13, 0xad, 0x44, 0xc2, 0xca, 0x09, 0x53, 0xf2, 0x47, 0xf2, 0x13, 0xeb, 0x4c, 0x19, 0x85,
+	0x9f, 0x45, 0x76, 0x2e, 0xa6, 0x95, 0x26, 0xf1, 0xdb, 0xcc, 0xe1, 0xf7, 0xff, 0x33, 0xff, 0xb3,
-	0xb6, 0x6b, 0x2e, 0xe3, 0x98, 0x6a, 0x8d, 0x1d, 0x33, 0x7a, 0x08, 0xe1, 0x94, 0x33, 0x25, 0x63,
+	0xe7, 0x2c, 0x3c, 0x5c, 0x0c, 0xd2, 0xa0, 0xf4, 0x43, 0x31, 0xef, 0x87, 0x22, 0xc5, 0x7e, 0x90,
-	0x74, 0x0f, 0x3a, 0x0b, 0x2a, 0x0a, 0x3e, 0x08, 0x76, 0x82, 0x71, 0x9f, 0xb8, 0x9f, 0xd1, 0xdf,
+	0x24, 0xfd, 0x44, 0xb0, 0x38, 0x2c, 0xfb, 0xa1, 0xe0, 0xd3, 0xf8, 0x87, 0x9f, 0xa4, 0x22, 0x17,
-	0x36, 0x84, 0x27, 0x16, 0x45, 0x2f, 0x60, 0xcd, 0x24, 0x29, 0x57, 0x85, 0xb1, 0x48, 0x37, 0xda,
+	0x64, 0xcf, 0x70, 0x29, 0xfa, 0x41, 0x92, 0xf8, 0x8a, 0xe9, 0xde, 0x03, 0x7b, 0x84, 0xa1, 0xe0,
-	0xc3, 0xd7, 0x66, 0x62, 0xc7, 0xe3, 0x4f, 0x0e, 0x26, 0xb5, 0x0b, 0x3d, 0x85, 0x4e, 0x6e, 0xa8,
+	0x11, 0xb9, 0x0d, 0xcd, 0x45, 0xc0, 0x0a, 0x74, 0xad, 0x7d, 0xab, 0xe7, 0x50, 0x75, 0xe9, 0x5e,
-	0xc9, 0x07, 0x2d, 0x6b, 0xdf, 0xbd, 0xd9, 0x3e, 0xad, 0x50, 0xe2, 0x1c, 0xc3, 0xdf, 0x2d, 0x58,
+	0x36, 0xc0, 0x3e, 0x93, 0x28, 0x79, 0x01, 0xad, 0x3c, 0x9e, 0xa3, 0x28, 0x72, 0x89, 0xb4, 0x07,
-	0xf3, 0x79, 0xe8, 0x10, 0xee, 0x9e, 0x51, 0x19, 0xe7, 0x67, 0x74, 0xce, 0x7d, 0x93, 0x07, 0x4b,
+	0x87, 0xfe, 0x8d, 0x9e, 0xbe, 0xe2, 0xfd, 0x4f, 0x0a, 0xa6, 0x46, 0x45, 0x9e, 0x40, 0x33, 0xcb,
-	0xa2, 0xdc, 0x68, 0xa4, 0xe1, 0xd1, 0x5b, 0xd8, 0x64, 0x4a, 0x4a, 0xce, 0x4c, 0xa2, 0xe4, 0x2c,
+	0x83, 0x3c, 0x73, 0x6b, 0x52, 0x7e, 0xb0, 0x5e, 0x3e, 0xaa, 0x50, 0xaa, 0x14, 0xe4, 0x19, 0xd8,
-	0x89, 0x05, 0xf7, 0x6d, 0x6e, 0x89, 0xd8, 0x68, 0x5c, 0xef, 0x62, 0xc1, 0xd1, 0x11, 0x74, 0x0b,
+	0x93, 0x62, 0x3a, 0xc5, 0xd4, 0xad, 0x4b, 0xed, 0x83, 0xf5, 0xda, 0xa1, 0x64, 0xa9, 0xd6, 0x78,
-	0x2d, 0x12, 0x39, 0x9f, 0x29, 0x29, 0xca, 0x41, 0x7b, 0x95, 0x0c, 0x70, 0x8e, 0x8f, 0x52, 0x94,
+	0xbf, 0x6b, 0xd0, 0xd2, 0xaf, 0x21, 0x27, 0x70, 0xeb, 0x3c, 0xe0, 0x51, 0x76, 0x1e, 0xcc, 0x50,
-	0xe8, 0x18, 0xfa, 0xb1, 0xfa, 0x25, 0x9b, 0x84, 0x3b, 0xab, 0x24, 0xf4, 0x6a, 0x4f, 0x95, 0x31,
+	0xe7, 0xb8, 0xbb, 0xc2, 0x4c, 0x35, 0x86, 0x2e, 0x79, 0xf2, 0x16, 0x76, 0x42, 0xc1, 0x39, 0x86,
-	0xfc, 0x00, 0x1d, 0x7b, 0x48, 0xe8, 0x11, 0x74, 0x8b, 0x9c, 0x67, 0x33, 0x97, 0x6f, 0xcf, 0x64,
+	0x79, 0x2c, 0xf8, 0x38, 0x8e, 0x18, 0xea, 0x2c, 0x1b, 0x2c, 0xb6, 0x97, 0xaa, 0x77, 0x11, 0x43,
-	0x9d, 0x40, 0x25, 0x7d, 0xb6, 0x0a, 0xda, 0x85, 0xbe, 0x05, 0x6a, 0xbb, 0x9d, 0x79, 0x9d, 0xf4,
+	0x72, 0x0a, 0xed, 0x22, 0x61, 0x31, 0x9f, 0x8d, 0x05, 0x67, 0xa5, 0xce, 0xb4, 0xc1, 0x03, 0x94,
-	0x2a, 0xf1, 0xb5, 0xd7, 0x46, 0xe7, 0x01, 0xf4, 0xa6, 0x65, 0x6e, 0x78, 0x7a, 0x71, 0xe1, 0xfe,
+	0xe2, 0x23, 0x67, 0x25, 0x19, 0x82, 0x13, 0x89, 0x9f, 0x7c, 0xe9, 0xd0, 0xf8, 0x1f, 0x87, 0x8e,
-	0xbe, 0xdc, 0x21, 0xef, 0x2f, 0xeb, 0x76, 0xc9, 0x73, 0xf5, 0xd6, 0xbe, 0xd5, 0x05, 0xf7, 0x60,
+	0xd1, 0x54, 0x1e, 0xde, 0x07, 0x68, 0xca, 0x16, 0x93, 0xfb, 0xd0, 0x2e, 0x32, 0x4c, 0xc7, 0xca,
-	0x23, 0x91, 0xa7, 0xaa, 0x90, 0xf1, 0xd5, 0x8e, 0x7d, 0xaf, 0xfa, 0x9a, 0xfb, 0xb0, 0x55, 0x63,
+	0x5f, 0xf6, 0x64, 0x8b, 0x42, 0x55, 0xfa, 0x2c, 0x2b, 0xe4, 0x00, 0x1c, 0x09, 0x18, 0xb9, 0xcc,
-	0xff, 0x35, 0xdd, 0xf4, 0xfa, 0x45, 0xd9, 0x3f, 0x01, 0x84, 0xaf, 0xec, 0xfb, 0x46, 0x47, 0xd0,
+	0xbc, 0x45, 0x3b, 0x55, 0xf1, 0xb5, 0xae, 0x79, 0x8f, 0xc0, 0x56, 0x5d, 0x27, 0x2e, 0xb4, 0x90,
-	0x11, 0x7c, 0xc1, 0xc5, 0x20, 0xd8, 0x69, 0x8f, 0xbb, 0xd1, 0x78, 0x49, 0x4d, 0x47, 0xe3, 0xf7,
+	0x07, 0x13, 0x86, 0x91, 0xf6, 0x32, 0x57, 0x42, 0xa0, 0x91, 0xc5, 0xbf, 0x54, 0xcf, 0x1c, 0x2a,
-	0x15, 0xfa, 0x46, 0x9a, 0xac, 0x24, 0xce, 0x86, 0x0e, 0x21, 0xcc, 0xed, 0x08, 0xb7, 0xbc, 0xcb,
+	0xcf, 0xdd, 0x0b, 0x0b, 0x3a, 0xa3, 0x32, 0xcb, 0x71, 0xfe, 0x77, 0xcc, 0xf4, 0x94, 0xa8, 0x8f,
-	0xcb, 0x73, 0x12, 0x6f, 0x19, 0x7e, 0x05, 0x68, 0x12, 0xd1, 0x16, 0xb4, 0xe7, 0xbc, 0xf4, 0x1b,
+	0x73, 0xb4, 0x2a, 0xd3, 0x15, 0xcd, 0xb5, 0x59, 0xf1, 0xbe, 0x99, 0x60, 0x87, 0xb0, 0x1d, 0xf3,
-	0x54, 0x7d, 0xa2, 0x83, 0x7a, 0xab, 0x6e, 0x7e, 0x65, 0x3e, 0xd5, 0xb1, 0xcf, 0x5a, 0x4f, 0x82,
+	0x89, 0x28, 0x78, 0x74, 0x3d, 0x9b, 0xa3, 0xab, 0x3a, 0xde, 0x11, 0xec, 0x1a, 0xec, 0x9f, 0x84,
-	0xe3, 0xe7, 0x70, 0x9f, 0xa9, 0xf4, 0x7a, 0xfc, 0x24, 0xf8, 0x1e, 0xba, 0xaf, 0xf3, 0xd6, 0xf6,
+	0x3b, 0xba, 0x6e, 0x42, 0x76, 0x2f, 0x2d, 0xb0, 0x5f, 0xc9, 0xad, 0x22, 0xa7, 0xd0, 0x64, 0xb8,
-	0x97, 0x88, 0xd0, 0x6a, 0xba, 0x8c, 0xe3, 0x97, 0x5a, 0xfb, 0xa4, 0xd3, 0xd0, 0x6e, 0xfd, 0xc1,
+	0x40, 0xe6, 0x5a, 0xfb, 0xf5, 0x5e, 0x7b, 0xd0, 0x5b, 0xf1, 0x4c, 0x45, 0xfb, 0xef, 0x2b, 0xf4,
-	0xbf, 0x00, 0x00, 0x00, 0xff, 0xff, 0xa9, 0x9f, 0x00, 0x31, 0x1f, 0x04, 0x00, 0x00,
+	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,
 }

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 {

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()
+	opts := pipe.OptionsFromContext(ctx)
-	downlinkReader, downlinkWriter := pipe.New()
+	uplinkReader, uplinkWriter := pipe.New(opts...)
+	downlinkReader, downlinkWriter := pipe.New(opts...)
 
 	worker := &ServerWorker{
 		dispatcher: s.dispatcher,

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()
+				opts := pipe.OptionsFromContext(ctx)
-				downlinkReader, downlinkWriter := pipe.New()
+				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

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

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 {

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

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

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 {

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)

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

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",

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