fix a data race in KCP

.github/ISSUE_TEMPLATE

@@ -1,7 +1,7 @@
 Please skip to the English section below if you don't write Chinese.
 
 中文：
-提交 Issue 之前请先阅读 [Issue 指引](https://www.v2ray.com/chapter_01/issue.html)，然后回答下面的问题，谢谢。
+提交 Issue 之前请先阅读 [Issue 指引](https://github.com/v2ray/v2ray-core/blob/master/.github/SUPPORT.md)，然后回答下面的问题，谢谢。
 除非特殊情况，请完整填写所有问题。不按模板发的 issue 将直接被关闭。
 
 1)  你正在使用哪个版本的 V2Ray？（如果服务器和客户端使用了不同版本，请注明）
@@ -36,7 +36,7 @@ Please skip to the English section below if you don't write Chinese.
     // 在这里附上客户端日志
 ```
 
-7) 请附上访问日志。在 Linux 中，日志通常在 `/var/log/v2ray/error.log` 文件中。
+7) 请附上访问日志。在 Linux 中，日志通常在 `/var/log/v2ray/access.log` 文件中。
 ```
     // 在这里附上服务器端日志
 ```
@@ -49,7 +49,7 @@ Please skip to the English section below if you don't write Chinese.
 Please remove the Chinese section above.
 
 English:
-Please read the [instruction](https://www.v2ray.com/en/get_started/issue.html) and answer the following questions before submitting your issue. Thank you.
+Please read the [instruction](https://github.com/v2ray/v2ray-core/blob/master/.github/SUPPORT.md) and answer the following questions before submitting your issue. Thank you.
 Please answer all the questions with enough information. All issues not following this template will be closed immediately.
 
 1) What version of V2Ray are you using (If you deploy different version on server and client, please explicitly point out)?

README.md

@@ -1,6 +1,6 @@
 # Project V
 
-[![Build Status][1]][2] [![codecov.io][3]][4] [![Go Report][5]][6] [![GoDoc][7]][8] [![codebeat][9]][10]
+[![Build Status][1]][2] [![codecov.io][3]][4] [![Go Report][5]][6] [![GoDoc][7]][8] [![codebeat][9]][10] [![Downloads][11]][12]
 
 [1]: https://travis-ci.org/v2ray/v2ray-core.svg?branch=master "Build Status badge"
 [2]: https://travis-ci.org/v2ray/v2ray-core "Travis-CI Build Status"
@@ -12,14 +12,13 @@
 [8]: https://godoc.org/v2ray.com/core "GoDoc"
 [9]: https://codebeat.co/badges/f2354ca8-3e24-463d-a2e3-159af73b2477 "Codebeat badge"
 [10]: https://codebeat.co/projects/github-com-v2ray-v2ray-core-master "Codebeat"
+[11]: https://img.shields.io/github/downloads/v2ray/v2ray-core/total.svg "All releases badge"
+[12]: https://github.com/v2ray/v2ray-core/releases/ "All releases number"
 
-V 是一个模块化的代理软件包，它的目标是提供常用的代理软件模块，简化网络代理软件的开发。
-
-[官方网站](https://www.v2ray.com/)
-
-V provides building blocks for network proxy development. Read our [Wiki](https://www.v2ray.com/en/index.html) for more information.
+Project V is a set of network tools that help you to build your own computer network. It secures your network connections and thus protects your privacy. See [our website](https://www.v2ray.com/) for more information.
 
 ## License
+
 [The MIT License (MIT)](https://raw.githubusercontent.com/v2ray/v2ray-core/master/LICENSE)
 
 ## Credits

app/log/log.go

@@ -9,6 +9,7 @@ import (
 	"v2ray.com/core/app/log/internal"
 	"v2ray.com/core/common"
 	"v2ray.com/core/common/errors"
+	"v2ray.com/core/common/log"
 )
 
 var (
@@ -51,15 +52,15 @@ func InitErrorLogger(file string) error {
 	return nil
 }
 
-func getLoggerAndPrefix(s errors.Severity) (internal.LogWriter, string) {
+func getLoggerAndPrefix(s log.Severity) (internal.LogWriter, string) {
 	switch s {
-	case errors.SeverityDebug:
+	case log.Severity_Debug:
 		return debugLogger, "[Debug]"
-	case errors.SeverityInfo:
+	case log.Severity_Info:
 		return infoLogger, "[Info]"
-	case errors.SeverityWarning:
+	case log.Severity_Warning:
 		return warningLogger, "[Warning]"
-	case errors.SeverityError:
+	case log.Severity_Error:
 		return errorLogger, "[Error]"
 	default:
 		return infoLogger, "[Info]"

app/policy/config.go

@@ -4,10 +4,12 @@ import (
 	"time"
 )
 
+// Duration converts Second to time.Duration.
 func (s *Second) Duration() time.Duration {
 	return time.Second * time.Duration(s.Value)
 }
 
+// OverrideWith overrides current Policy with another one.
 func (p *Policy) OverrideWith(another *Policy) {
 	if another.Timeout != nil {
 		if another.Timeout.Handshake != nil {

app/policy/manager/manager.go

@@ -7,10 +7,12 @@ import (
 	"v2ray.com/core/common"
 )
 
+// Instance is an instance of Policy manager.
 type Instance struct {
 	levels map[uint32]*policy.Policy
 }
 
+// New creates new Policy manager instance.
 func New(ctx context.Context, config *policy.Config) (*Instance, error) {
 	levels := config.Level
 	if levels == nil {

app/space.go

@@ -8,12 +8,14 @@ import (
 	"v2ray.com/core/common/event"
 )
 
+// Application is a component that runs in Space.
 type Application interface {
 	Interface() interface{}
 	Start() error
 	Close()
 }
 
+// CreateAppFromConfig creates an Application based on its config. Application must have been registered.
 func CreateAppFromConfig(ctx context.Context, config interface{}) (Application, error) {
 	application, err := common.CreateObject(ctx, config)
 	if err != nil {

app/web/config.pb.go

@@ -1,228 +0,0 @@
-package web
-
-import proto "github.com/golang/protobuf/proto"
-import fmt "fmt"
-import math "math"
-import v2ray_core_common_serial "v2ray.com/core/common/serial"
-
-// Reference imports to suppress errors if they are not otherwise used.
-var _ = proto.Marshal
-var _ = fmt.Errorf
-var _ = math.Inf
-
-// This is a compile-time assertion to ensure that this generated file
-// is compatible with the proto package it is being compiled against.
-// A compilation error at this line likely means your copy of the
-// proto package needs to be updated.
-const _ = proto.ProtoPackageIsVersion2 // please upgrade the proto package
-
-type FileServer struct {
-	Entry []*FileServer_Entry `protobuf:"bytes,1,rep,name=entry" json:"entry,omitempty"`
-}
-
-func (m *FileServer) Reset()                    { *m = FileServer{} }
-func (m *FileServer) String() string            { return proto.CompactTextString(m) }
-func (*FileServer) ProtoMessage()               {}
-func (*FileServer) Descriptor() ([]byte, []int) { return fileDescriptor0, []int{0} }
-
-func (m *FileServer) GetEntry() []*FileServer_Entry {
-	if m != nil {
-		return m.Entry
-	}
-	return nil
-}
-
-type FileServer_Entry struct {
-	// Types that are valid to be assigned to FileOrDir:
-	//	*FileServer_Entry_File
-	//	*FileServer_Entry_Directory
-	FileOrDir isFileServer_Entry_FileOrDir `protobuf_oneof:"FileOrDir"`
-	Path      string                       `protobuf:"bytes,3,opt,name=path" json:"path,omitempty"`
-}
-
-func (m *FileServer_Entry) Reset()                    { *m = FileServer_Entry{} }
-func (m *FileServer_Entry) String() string            { return proto.CompactTextString(m) }
-func (*FileServer_Entry) ProtoMessage()               {}
-func (*FileServer_Entry) Descriptor() ([]byte, []int) { return fileDescriptor0, []int{0, 0} }
-
-type isFileServer_Entry_FileOrDir interface {
-	isFileServer_Entry_FileOrDir()
-}
-
-type FileServer_Entry_File struct {
-	File string `protobuf:"bytes,1,opt,name=File,oneof"`
-}
-type FileServer_Entry_Directory struct {
-	Directory string `protobuf:"bytes,2,opt,name=Directory,oneof"`
-}
-
-func (*FileServer_Entry_File) isFileServer_Entry_FileOrDir()      {}
-func (*FileServer_Entry_Directory) isFileServer_Entry_FileOrDir() {}
-
-func (m *FileServer_Entry) GetFileOrDir() isFileServer_Entry_FileOrDir {
-	if m != nil {
-		return m.FileOrDir
-	}
-	return nil
-}
-
-func (m *FileServer_Entry) GetFile() string {
-	if x, ok := m.GetFileOrDir().(*FileServer_Entry_File); ok {
-		return x.File
-	}
-	return ""
-}
-
-func (m *FileServer_Entry) GetDirectory() string {
-	if x, ok := m.GetFileOrDir().(*FileServer_Entry_Directory); ok {
-		return x.Directory
-	}
-	return ""
-}
-
-func (m *FileServer_Entry) GetPath() string {
-	if m != nil {
-		return m.Path
-	}
-	return ""
-}
-
-// XXX_OneofFuncs is for the internal use of the proto package.
-func (*FileServer_Entry) XXX_OneofFuncs() (func(msg proto.Message, b *proto.Buffer) error, func(msg proto.Message, tag, wire int, b *proto.Buffer) (bool, error), func(msg proto.Message) (n int), []interface{}) {
-	return _FileServer_Entry_OneofMarshaler, _FileServer_Entry_OneofUnmarshaler, _FileServer_Entry_OneofSizer, []interface{}{
-		(*FileServer_Entry_File)(nil),
-		(*FileServer_Entry_Directory)(nil),
-	}
-}
-
-func _FileServer_Entry_OneofMarshaler(msg proto.Message, b *proto.Buffer) error {
-	m := msg.(*FileServer_Entry)
-	// FileOrDir
-	switch x := m.FileOrDir.(type) {
-	case *FileServer_Entry_File:
-		b.EncodeVarint(1<<3 | proto.WireBytes)
-		b.EncodeStringBytes(x.File)
-	case *FileServer_Entry_Directory:
-		b.EncodeVarint(2<<3 | proto.WireBytes)
-		b.EncodeStringBytes(x.Directory)
-	case nil:
-	default:
-		return fmt.Errorf("FileServer_Entry.FileOrDir has unexpected type %T", x)
-	}
-	return nil
-}
-
-func _FileServer_Entry_OneofUnmarshaler(msg proto.Message, tag, wire int, b *proto.Buffer) (bool, error) {
-	m := msg.(*FileServer_Entry)
-	switch tag {
-	case 1: // FileOrDir.File
-		if wire != proto.WireBytes {
-			return true, proto.ErrInternalBadWireType
-		}
-		x, err := b.DecodeStringBytes()
-		m.FileOrDir = &FileServer_Entry_File{x}
-		return true, err
-	case 2: // FileOrDir.Directory
-		if wire != proto.WireBytes {
-			return true, proto.ErrInternalBadWireType
-		}
-		x, err := b.DecodeStringBytes()
-		m.FileOrDir = &FileServer_Entry_Directory{x}
-		return true, err
-	default:
-		return false, nil
-	}
-}
-
-func _FileServer_Entry_OneofSizer(msg proto.Message) (n int) {
-	m := msg.(*FileServer_Entry)
-	// FileOrDir
-	switch x := m.FileOrDir.(type) {
-	case *FileServer_Entry_File:
-		n += proto.SizeVarint(1<<3 | proto.WireBytes)
-		n += proto.SizeVarint(uint64(len(x.File)))
-		n += len(x.File)
-	case *FileServer_Entry_Directory:
-		n += proto.SizeVarint(2<<3 | proto.WireBytes)
-		n += proto.SizeVarint(uint64(len(x.Directory)))
-		n += len(x.Directory)
-	case nil:
-	default:
-		panic(fmt.Sprintf("proto: unexpected type %T in oneof", x))
-	}
-	return n
-}
-
-type Server struct {
-	Domain   []string                               `protobuf:"bytes,1,rep,name=domain" json:"domain,omitempty"`
-	Settings *v2ray_core_common_serial.TypedMessage `protobuf:"bytes,2,opt,name=settings" json:"settings,omitempty"`
-}
-
-func (m *Server) Reset()                    { *m = Server{} }
-func (m *Server) String() string            { return proto.CompactTextString(m) }
-func (*Server) ProtoMessage()               {}
-func (*Server) Descriptor() ([]byte, []int) { return fileDescriptor0, []int{1} }
-
-func (m *Server) GetDomain() []string {
-	if m != nil {
-		return m.Domain
-	}
-	return nil
-}
-
-func (m *Server) GetSettings() *v2ray_core_common_serial.TypedMessage {
-	if m != nil {
-		return m.Settings
-	}
-	return nil
-}
-
-type Config struct {
-	Server []*Server `protobuf:"bytes,1,rep,name=server" json:"server,omitempty"`
-}
-
-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{2} }
-
-func (m *Config) GetServer() []*Server {
-	if m != nil {
-		return m.Server
-	}
-	return nil
-}
-
-func init() {
-	proto.RegisterType((*FileServer)(nil), "v2ray.core.app.web.FileServer")
-	proto.RegisterType((*FileServer_Entry)(nil), "v2ray.core.app.web.FileServer.Entry")
-	proto.RegisterType((*Server)(nil), "v2ray.core.app.web.Server")
-	proto.RegisterType((*Config)(nil), "v2ray.core.app.web.Config")
-}
-
-func init() { proto.RegisterFile("v2ray.com/core/app/web/config.proto", fileDescriptor0) }
-
-var fileDescriptor0 = []byte{
-	// 324 bytes of a gzipped FileDescriptorProto
-	0x1f, 0x8b, 0x08, 0x00, 0x00, 0x00, 0x00, 0x00, 0x02, 0xff, 0x6c, 0x91, 0xcf, 0x4a, 0xc3, 0x40,
-	0x10, 0x87, 0x4d, 0xff, 0x04, 0x33, 0xbd, 0x2d, 0x52, 0x42, 0x0f, 0x52, 0xaa, 0x48, 0x4f, 0x1b,
-	0x89, 0x9e, 0xc4, 0x8b, 0x6d, 0x15, 0x2f, 0xa2, 0xac, 0xa2, 0xe0, 0x41, 0xd9, 0xa4, 0x63, 0x5d,
-	0x68, 0xb2, 0xcb, 0x64, 0x69, 0xc9, 0x1b, 0x89, 0x4f, 0x29, 0xd9, 0x44, 0x2b, 0xda, 0x5b, 0x26,
-	0xf3, 0xfd, 0x76, 0x66, 0xbf, 0x85, 0x83, 0x55, 0x4c, 0xb2, 0xe4, 0xa9, 0xce, 0xa2, 0x54, 0x13,
-	0x46, 0xd2, 0x98, 0x68, 0x8d, 0x49, 0x94, 0xea, 0xfc, 0x4d, 0x2d, 0xb8, 0x21, 0x6d, 0x35, 0x63,
-	0xdf, 0x10, 0x21, 0x97, 0xc6, 0xf0, 0x35, 0x26, 0x83, 0xe3, 0x3f, 0xc1, 0x54, 0x67, 0x99, 0xce,
-	0xa3, 0x02, 0x49, 0xc9, 0x65, 0x64, 0x4b, 0x83, 0xf3, 0xd7, 0x0c, 0x8b, 0x42, 0x2e, 0xb0, 0x3e,
-	0x65, 0xf4, 0xe1, 0x01, 0x5c, 0xa9, 0x25, 0xde, 0x23, 0xad, 0x90, 0xd8, 0x19, 0x74, 0x31, 0xb7,
-	0x54, 0x86, 0xde, 0xb0, 0x3d, 0xee, 0xc5, 0x87, 0xfc, 0xff, 0x10, 0xbe, 0xc1, 0xf9, 0x65, 0xc5,
-	0x8a, 0x3a, 0x32, 0x78, 0x81, 0xae, 0xab, 0xd9, 0x1e, 0x74, 0x2a, 0x26, 0xf4, 0x86, 0xde, 0x38,
-	0xb8, 0xde, 0x11, 0xae, 0x62, 0xfb, 0x10, 0xcc, 0x14, 0x61, 0x6a, 0x35, 0x95, 0x61, 0xab, 0x69,
-	0x6d, 0x7e, 0x31, 0x06, 0x1d, 0x23, 0xed, 0x7b, 0xd8, 0xae, 0x5a, 0xc2, 0x7d, 0x4f, 0x7a, 0x10,
-	0x54, 0xd9, 0x5b, 0x9a, 0x29, 0x1a, 0xcd, 0xc1, 0x6f, 0xb6, 0xec, 0x83, 0x3f, 0xd7, 0x99, 0x54,
-	0xb9, 0x5b, 0x33, 0x10, 0x4d, 0xc5, 0x26, 0xb0, 0x5b, 0xa0, 0xb5, 0x2a, 0x5f, 0x14, 0x6e, 0x42,
-	0x2f, 0x3e, 0xfa, 0x7d, 0x81, 0xda, 0x06, 0xaf, 0x6d, 0xf0, 0x87, 0xca, 0xc6, 0x4d, 0x2d, 0x43,
-	0xfc, 0xe4, 0x46, 0xe7, 0xe0, 0x4f, 0x9d, 0x66, 0x16, 0x83, 0x5f, 0xb8, 0x79, 0x8d, 0x8c, 0xc1,
-	0x36, 0x19, 0xf5, 0x46, 0xa2, 0x21, 0x27, 0xa7, 0xd0, 0x4f, 0x75, 0xb6, 0x05, 0xbc, 0xf3, 0x9e,
-	0xdb, 0x6b, 0x4c, 0x3e, 0x5b, 0xec, 0x31, 0x16, 0xb2, 0xe4, 0xd3, 0xaa, 0x77, 0x61, 0x0c, 0x7f,
-	0xc2, 0x24, 0xf1, 0xdd, 0x5b, 0x9c, 0x7c, 0x05, 0x00, 0x00, 0xff, 0xff, 0xf9, 0x2a, 0xe1, 0x59,
-	0xf8, 0x01, 0x00, 0x00,
-}

app/web/config.proto

@@ -1,30 +0,0 @@
-syntax = "proto3";
-
-package v2ray.core.app.web;
-option csharp_namespace = "V2Ray.Core.App.Web";
-option go_package = "web";
-option java_package = "com.v2ray.core.app.web";
-option java_multiple_files = true;
-
-import "v2ray.com/core/common/serial/typed_message.proto";
-
-message FileServer {
-  message Entry {
-    oneof FileOrDir {
-      string File = 1;
-      string Directory = 2;
-    }
-    string path = 3;
-  }
-
-  repeated Entry entry = 1;
-}
-
-message Server {
-  repeated string domain = 1;
-  v2ray.core.common.serial.TypedMessage settings = 2;
-}
-
-message Config {
-  repeated Server server = 1;
-}

app/web/file_server.go

@@ -1 +0,0 @@
-package web

app/web/web.go

@@ -1,5 +0,0 @@
-package web
-
-type WebServer interface {
-	Handle()
-}

common/buf/multi_buffer.go

@@ -65,7 +65,7 @@ func (mb *MultiBuffer) AppendMulti(buf MultiBuffer) {
 	*mb = append(*mb, buf...)
 }
 
-// Copy copied the begining part of the MultiBuffer into the given byte array.
+// Copy copied the beginning part of the MultiBuffer into the given byte array.
 func (mb MultiBuffer) Copy(b []byte) int {
 	total := 0
 	for _, bb := range mb {
@@ -151,7 +151,7 @@ func (mb MultiBuffer) ToNetBuffers() net.Buffers {
 	return bs
 }
 
-// SliceBySize splits the begining of this MultiBuffer into another one, for at most size bytes.
+// SliceBySize splits the beginning of this MultiBuffer into another one, for at most size bytes.
 func (mb *MultiBuffer) SliceBySize(size int) MultiBuffer {
 	slice := NewMultiBufferCap(10)
 	sliceSize := 0

common/buf/reader.go

@@ -18,11 +18,14 @@ func NewBytesToBufferReader(reader io.Reader) Reader {
 	}
 }
 
+const mediumSize = 8 * 1024
+const largeSize = 64 * 1024
+
 func (r *BytesToBufferReader) readSmall() (MultiBuffer, error) {
 	b := New()
 	err := b.Reset(ReadFrom(r.Reader))
 	if b.IsFull() {
-		r.buffer = make([]byte, 32*1024)
+		r.buffer = make([]byte, mediumSize)
 	}
 	if !b.IsEmpty() {
 		return NewMultiBufferValue(b), nil
@@ -41,6 +44,9 @@ func (r *BytesToBufferReader) ReadMultiBuffer() (MultiBuffer, error) {
 	if nBytes > 0 {
 		mb := NewMultiBufferCap(nBytes/Size + 1)
 		mb.Write(r.buffer[:nBytes])
+		if nBytes == len(r.buffer) && len(r.buffer) == mediumSize {
+			r.buffer = make([]byte, largeSize)
+		}
 		return mb, nil
 	}
 	return nil, err

common/buf/reader_test.go

@@ -21,7 +21,11 @@ func TestAdaptiveReader(t *testing.T) {
 
 	b, err = reader.ReadMultiBuffer()
 	assert(err, IsNil)
-	assert(b.Len(), Equals, 32*1024)
+	assert(b.Len(), Equals, 8*1024)
+
+	b, err = reader.ReadMultiBuffer()
+	assert(err, IsNil)
+	assert(b.Len(), Equals, 64*1024)
 }
 
 func TestBytesReaderWriteTo(t *testing.T) {

common/buf/writer.go

@@ -93,6 +93,9 @@ func (w *BufferedWriter) WriteMultiBuffer(b MultiBuffer) error {
 	defer b.Release()
 
 	for !b.IsEmpty() {
+		if w.buffer == nil {
+			w.buffer = New()
+		}
 		if err := w.buffer.AppendSupplier(ReadFrom(&b)); err != nil {
 			return err
 		}
@@ -113,11 +116,7 @@ func (w *BufferedWriter) Flush() error {
 			return err
 		}
 
-		if w.buffered {
-			w.buffer = New()
-		} else {
-			w.buffer = nil
-		}
+		w.buffer = nil
 	}
 	return nil
 }

common/errors/errors.go

@@ -4,33 +4,24 @@ package errors
 import (
 	"strings"
 
+	"v2ray.com/core/common/log"
 	"v2ray.com/core/common/serial"
 )
 
-// Severity describes how severe the error is.
-type Severity int
-
-const (
-	SeverityDebug Severity = iota
-	SeverityInfo
-	SeverityWarning
-	SeverityError
-)
-
 type hasInnerError interface {
 	// Inner returns the underlying error of this one.
 	Inner() error
 }
 
 type hasSeverity interface {
-	Severity() Severity
+	Severity() log.Severity
 }
 
 // Error is an error object with underlying error.
 type Error struct {
 	message  []interface{}
 	inner    error
-	severity Severity
+	severity log.Severity
 	path     []string
 }
 
@@ -59,19 +50,19 @@ func (v *Error) Base(err error) *Error {
 	return v
 }
 
-func (v *Error) atSeverity(s Severity) *Error {
+func (v *Error) atSeverity(s log.Severity) *Error {
 	v.severity = s
 	return v
 }
 
-func (v *Error) Severity() Severity {
+func (v *Error) Severity() log.Severity {
 	if v.inner == nil {
 		return v.severity
 	}
 
 	if s, ok := v.inner.(hasSeverity); ok {
 		as := s.Severity()
-		if as > v.severity {
+		if as < v.severity {
 			return as
 		}
 	}
@@ -81,22 +72,22 @@ func (v *Error) Severity() Severity {
 
 // AtDebug sets the severity to debug.
 func (v *Error) AtDebug() *Error {
-	return v.atSeverity(SeverityDebug)
+	return v.atSeverity(log.Severity_Debug)
 }
 
 // AtInfo sets the severity to info.
 func (v *Error) AtInfo() *Error {
-	return v.atSeverity(SeverityInfo)
+	return v.atSeverity(log.Severity_Info)
 }
 
 // AtWarning sets the severity to warning.
 func (v *Error) AtWarning() *Error {
-	return v.atSeverity(SeverityWarning)
+	return v.atSeverity(log.Severity_Warning)
 }
 
 // AtError sets the severity to error.
 func (v *Error) AtError() *Error {
-	return v.atSeverity(SeverityError)
+	return v.atSeverity(log.Severity_Error)
 }
 
 // Path sets the path to the location where this error happens.
@@ -109,7 +100,7 @@ func (v *Error) Path(path ...string) *Error {
 func New(msg ...interface{}) *Error {
 	return &Error{
 		message:  msg,
-		severity: SeverityInfo,
+		severity: log.Severity_Info,
 	}
 }
 
@@ -128,9 +119,9 @@ func Cause(err error) error {
 	return err
 }
 
-func GetSeverity(err error) Severity {
+func GetSeverity(err error) log.Severity {
 	if s, ok := err.(hasSeverity); ok {
 		return s.Severity()
 	}
-	return SeverityInfo
+	return log.Severity_Info
 }

common/errors/errors_test.go

@@ -5,6 +5,7 @@ import (
 	"testing"
 
 	. "v2ray.com/core/common/errors"
+	"v2ray.com/core/common/log"
 	. "v2ray.com/ext/assert"
 )
 
@@ -12,17 +13,17 @@ func TestError(t *testing.T) {
 	assert := With(t)
 
 	err := New("TestError")
-	assert(GetSeverity(err), Equals, SeverityInfo)
+	assert(GetSeverity(err), Equals, log.Severity_Info)
 
 	err = New("TestError2").Base(io.EOF)
-	assert(GetSeverity(err), Equals, SeverityInfo)
+	assert(GetSeverity(err), Equals, log.Severity_Info)
 
 	err = New("TestError3").Base(io.EOF).AtWarning()
-	assert(GetSeverity(err), Equals, SeverityWarning)
+	assert(GetSeverity(err), Equals, log.Severity_Warning)
 
 	err = New("TestError4").Base(io.EOF).AtWarning()
 	err = New("TestError5").Base(err)
-	assert(GetSeverity(err), Equals, SeverityWarning)
+	assert(GetSeverity(err), Equals, log.Severity_Warning)
 	assert(err.Error(), HasSubstring, "EOF")
 }
 

common/log/access.go

@@ -0,0 +1,15 @@
+package log
+
+type AccessStatus string
+
+const (
+	AccessAccepted = AccessStatus("accepted")
+	AccessRejected = AccessStatus("rejected")
+)
+
+type AccessMessage struct {
+	From   interface{}
+	To     interface{}
+	Status AccessStatus
+	Reason interface{}
+}

common/log/log.go

@@ -0,0 +1,49 @@
+package log
+
+import (
+	"sync/atomic"
+	"unsafe"
+
+	"v2ray.com/core/common/serial"
+)
+
+type Message interface {
+	String() string
+}
+
+type Handler interface {
+	Handle(msg Message)
+}
+
+type noOpHandler byte
+
+func (noOpHandler) Handle(msg Message) {}
+
+type GeneralMessage struct {
+	Severity Severity
+	Content  interface{}
+}
+
+func (m *GeneralMessage) String() string {
+	return serial.Concat("[", m.Severity, "]: ", m.Content)
+}
+
+func Record(msg Message) {
+	h := (*Handler)(atomic.LoadPointer(&logHandler))
+	(*h).Handle(msg)
+}
+
+var (
+	logHandler unsafe.Pointer
+)
+
+func RegisterHandler(handler Handler) {
+	if handler == nil {
+		panic("Log handler is nil")
+	}
+	atomic.StorePointer(&logHandler, unsafe.Pointer(&handler))
+}
+
+func init() {
+	RegisterHandler(noOpHandler(0))
+}

common/log/log.pb.go

@@ -0,0 +1,68 @@
+package log
+
+import proto "github.com/golang/protobuf/proto"
+import fmt "fmt"
+import math "math"
+
+// Reference imports to suppress errors if they are not otherwise used.
+var _ = proto.Marshal
+var _ = fmt.Errorf
+var _ = math.Inf
+
+// This is a compile-time assertion to ensure that this generated file
+// is compatible with the proto package it is being compiled against.
+// A compilation error at this line likely means your copy of the
+// proto package needs to be updated.
+const _ = proto.ProtoPackageIsVersion2 // please upgrade the proto package
+
+type Severity int32
+
+const (
+	Severity_Unknown Severity = 0
+	Severity_Error   Severity = 1
+	Severity_Warning Severity = 2
+	Severity_Info    Severity = 3
+	Severity_Debug   Severity = 4
+)
+
+var Severity_name = map[int32]string{
+	0: "Unknown",
+	1: "Error",
+	2: "Warning",
+	3: "Info",
+	4: "Debug",
+}
+var Severity_value = map[string]int32{
+	"Unknown": 0,
+	"Error":   1,
+	"Warning": 2,
+	"Info":    3,
+	"Debug":   4,
+}
+
+func (x Severity) String() string {
+	return proto.EnumName(Severity_name, int32(x))
+}
+func (Severity) EnumDescriptor() ([]byte, []int) { return fileDescriptor0, []int{0} }
+
+func init() {
+	proto.RegisterEnum("v2ray.core.common.log.Severity", Severity_name, Severity_value)
+}
+
+func init() { proto.RegisterFile("v2ray.com/core/common/log/log.proto", fileDescriptor0) }
+
+var fileDescriptor0 = []byte{
+	// 178 bytes of a gzipped FileDescriptorProto
+	0x1f, 0x8b, 0x08, 0x00, 0x00, 0x00, 0x00, 0x00, 0x02, 0xff, 0xe2, 0x52, 0x2e, 0x33, 0x2a, 0x4a,
+	0xac, 0xd4, 0x4b, 0xce, 0xcf, 0xd5, 0x4f, 0xce, 0x2f, 0x4a, 0xd5, 0x4f, 0xce, 0xcf, 0xcd, 0xcd,
+	0xcf, 0xd3, 0xcf, 0xc9, 0x4f, 0x07, 0x61, 0xbd, 0x82, 0xa2, 0xfc, 0x92, 0x7c, 0x21, 0x51, 0x98,
+	0xa2, 0xa2, 0x54, 0x3d, 0x88, 0x02, 0xbd, 0x9c, 0xfc, 0x74, 0x2d, 0x17, 0x2e, 0x8e, 0xe0, 0xd4,
+	0xb2, 0xd4, 0xa2, 0xcc, 0x92, 0x4a, 0x21, 0x6e, 0x2e, 0xf6, 0xd0, 0xbc, 0xec, 0xbc, 0xfc, 0xf2,
+	0x3c, 0x01, 0x06, 0x21, 0x4e, 0x2e, 0x56, 0xd7, 0xa2, 0xa2, 0xfc, 0x22, 0x01, 0x46, 0x90, 0x78,
+	0x78, 0x62, 0x51, 0x5e, 0x66, 0x5e, 0xba, 0x00, 0x93, 0x10, 0x07, 0x17, 0x8b, 0x67, 0x5e, 0x5a,
+	0xbe, 0x00, 0x33, 0x48, 0x85, 0x4b, 0x6a, 0x52, 0x69, 0xba, 0x00, 0x8b, 0x93, 0x15, 0x97, 0x64,
+	0x72, 0x7e, 0xae, 0x1e, 0x56, 0x2b, 0x02, 0x18, 0xa3, 0x98, 0x73, 0xf2, 0xd3, 0x57, 0x31, 0x89,
+	0x86, 0x19, 0x05, 0x25, 0x56, 0xea, 0x39, 0x83, 0xa4, 0x9d, 0x21, 0xd2, 0x3e, 0xf9, 0xe9, 0x49,
+	0x6c, 0x60, 0xf7, 0x19, 0x03, 0x02, 0x00, 0x00, 0xff, 0xff, 0xfb, 0x69, 0xef, 0x0e, 0xc6, 0x00,
+	0x00, 0x00,
+}

common/log/log.proto

@@ -0,0 +1,16 @@
+syntax = "proto3";
+
+package v2ray.core.common.log;
+option csharp_namespace = "V2Ray.Core.Common.Log";
+option go_package = "log";
+option java_package = "com.v2ray.core.common.log";
+option java_multiple_files = true;
+
+enum Severity {
+  Unknown = 0;
+  Error = 1;
+  Warning = 2;
+  Info = 3;
+  Debug = 4;
+}
+

common/log/log_test.go

@@ -0,0 +1,32 @@
+package log_test
+
+import (
+	"testing"
+
+	"v2ray.com/core/common/log"
+	"v2ray.com/core/common/net"
+	. "v2ray.com/ext/assert"
+)
+
+type testLogger struct {
+	value string
+}
+
+func (l *testLogger) Handle(msg log.Message) {
+	l.value = msg.String()
+}
+
+func TestLogRecord(t *testing.T) {
+	assert := With(t)
+
+	var logger testLogger
+	log.RegisterHandler(&logger)
+
+	ip := "8.8.8.8"
+	log.Record(&log.GeneralMessage{
+		Severity: log.Severity_Error,
+		Content:  net.ParseAddress(ip),
+	})
+
+	assert(logger.value, Equals, "[Error]: "+ip)
+}

common/net/address.go

@@ -2,6 +2,7 @@ package net
 
 import (
 	"net"
+	"strings"
 
 	"v2ray.com/core/app/log"
 	"v2ray.com/core/common/predicate"
@@ -78,6 +79,7 @@ func ParseAddress(addr string) Address {
 	if lenAddr > 0 && addr[0] == '[' && addr[lenAddr-1] == ']' {
 		addr = addr[1 : lenAddr-1]
 	}
+	addr = strings.TrimSpace(addr)
 
 	ip := net.ParseIP(addr)
 	if ip != nil {

common/net/port.go

@@ -49,7 +49,7 @@ func (p Port) String() string {
 	return serial.Uint16ToString(p.Value())
 }
 
-// FromPort returns the begining port of this PortRange.
+// FromPort returns the beginning port of this PortRange.
 func (p PortRange) FromPort() Port {
 	return Port(p.From)
 }

common/signal/timer.go

@@ -26,7 +26,14 @@ func (t *ActivityTimer) SetTimeout(timeout time.Duration) {
 }
 
 func (t *ActivityTimer) run(ctx context.Context, cancel context.CancelFunc) {
-	ticker := time.NewTicker(<-t.timeout)
+	defer cancel()
+
+	timeout := <-t.timeout
+	if timeout == 0 {
+		return
+	}
+
+	ticker := time.NewTicker(timeout)
 	defer func() {
 		ticker.Stop()
 	}()
@@ -38,7 +45,6 @@ func (t *ActivityTimer) run(ctx context.Context, cancel context.CancelFunc) {
 			return
 		case timeout := <-t.timeout:
 			if timeout == 0 {
-				cancel()
 				return
 			}
 
@@ -51,7 +57,6 @@ func (t *ActivityTimer) run(ctx context.Context, cancel context.CancelFunc) {
 		case <-t.updated:
 		// Updated keep waiting.
 		default:
-			cancel()
 			return
 		}
 	}

core.go

@@ -18,7 +18,7 @@ import (
 )
 
 var (
-	version  = "3.1"
+	version  = "3.3"
 	build    = "Custom"
 	codename = "die Commanderin"
 	intro    = "An unified platform for anti-censorship."

main/config_json.go

@@ -18,6 +18,7 @@ func jsonToProto(input io.Reader) (*core.Config, error) {
 	cmd := exec.Command(v2ctl, "config")
 	cmd.Stdin = input
 	cmd.Stderr = os.Stderr
+	cmd.SysProcAttr = getSysProcAttr()
 
 	stdoutReader, err := cmd.StdoutPipe()
 	if err != nil {

main/config_json_other.go

@@ -0,0 +1,9 @@
+// +build !windows
+
+package main
+
+import "syscall"
+
+func getSysProcAttr() *syscall.SysProcAttr {
+	return nil
+}

main/config_json_windows.go

@@ -0,0 +1,11 @@
+// +build windows
+
+package main
+
+import "syscall"
+
+func getSysProcAttr() *syscall.SysProcAttr {
+	return &syscall.SysProcAttr{
+		HideWindow: true,
+	}
+}

proxy/shadowsocks/protocol.go

@@ -22,6 +22,7 @@ const (
 	AddrTypeDomain = 3
 )
 
+// ReadTCPSession reads a Shadowsocks TCP session from the given reader, returns its header and remaining parts.
 func ReadTCPSession(user *protocol.User, reader io.Reader) (*protocol.RequestHeader, buf.Reader, error) {
 	rawAccount, err := user.GetTypedAccount()
 	if err != nil {
@@ -136,6 +137,7 @@ func ReadTCPSession(user *protocol.User, reader io.Reader) (*protocol.RequestHea
 	return request, chunkReader, nil
 }
 
+// WriteTCPRequest writes Shadowsocks request into the given writer, and returns a writer for body.
 func WriteTCPRequest(request *protocol.RequestHeader, writer io.Writer) (buf.Writer, error) {
 	user := request.User
 	rawAccount, err := user.GetTypedAccount()

tools/release/config/vpoint_socks_vmess.json

@@ -13,105 +13,9 @@
     }
   },
   "outbound": {
-    "protocol": "vmess",
-    "settings": {
-      "vnext": [
-        {
-          "address": "v2ray.cool",
-          "port": 443,
-          "users": [
-            {
-              "id": "a3482e88-686a-4a58-8126-99c9df64b7bf",
-              "alterId": 64,
-              "security": "auto"
-            }
-          ]
-        }
-      ]
-    },
-    "mux": {
-      "enabled": true
-    },
-    "streamSettings": {
-      "network": "ws",
-      "security": "tls",
-      "tlsSettings": {
-        "serverName": "v2ray.cool",
-        "allowInsecure": true
-      },
-      "wsSettings": {
-        "path": "/v2ray.cool/"
-      }
-    }
-  },
-  "outboundDetour": [
-    {
-      "protocol": "freedom",
-      "settings": {},
-      "tag": "direct"
-    }
-  ],
-  "dns": {
-    "servers": [
-      "8.8.8.8",
-      "8.8.4.4",
-      "localhost"
-    ]
-  },
-  "routing": {
-    "strategy": "rules",
-    "settings": {
-      "domainStrategy": "IPIfNonMatch",
-      "rules": [
-        {
-          "type": "field",
-          "port": "1-52",
-          "outboundTag": "direct"
-        },
-        {
-          "type": "field",
-          "port": "54-79",
-          "outboundTag": "direct"
-        },
-        {
-          "type": "field",
-          "port": "81-442",
-          "outboundTag": "direct"
-        },
-        {
-          "type": "field",
-          "port": "444-65535",
-          "outboundTag": "direct"
-        },
-        {
-          "type": "field",
-          "domain": ["geosite:cn"],
-          "outboundTag": "direct"
-        },
-        {
-          "type": "field",
-          "ip": [
-            "0.0.0.0/8",
-            "10.0.0.0/8",
-            "100.64.0.0/10",
-            "127.0.0.0/8",
-            "169.254.0.0/16",
-            "172.16.0.0/12",
-            "192.0.0.0/24",
-            "192.0.2.0/24",
-            "192.168.0.0/16",
-            "198.18.0.0/15",
-            "198.51.100.0/24",
-            "203.0.113.0/24",
-            "::1/128",
-            "fc00::/7",
-            "fe80::/10",
-            "geoip:cn"
-          ],
-          "outboundTag": "direct"
-        }
-      ]
-    }
+    "protocol": "freedom",
+    "settings": {},
+    "tag": "direct"
   },
   "policy": {
     "levels": {

transport/internet/headers/srtp/srtp.go

@@ -25,7 +25,8 @@ func (s *SRTP) Write(b []byte) (int, error) {
 	return 4, nil
 }
 
-func NewSRTP(ctx context.Context, config interface{}) (interface{}, error) {
+// New returns a new SRTP instance based on the given config.
+func New(ctx context.Context, config interface{}) (interface{}, error) {
 	return &SRTP{
 		header: 0xB5E8,
 		number: dice.RollUint16(),
@@ -33,5 +34,5 @@ func NewSRTP(ctx context.Context, config interface{}) (interface{}, error) {
 }
 
 func init() {
-	common.Must(common.RegisterConfig((*Config)(nil), NewSRTP))
+	common.Must(common.RegisterConfig((*Config)(nil), New))
 }

transport/internet/headers/srtp/srtp_test.go

@@ -1,6 +1,7 @@
 package srtp_test
 
 import (
+	"context"
 	"testing"
 
 	"v2ray.com/core/common/buf"
@@ -12,7 +13,10 @@ func TestSRTPWrite(t *testing.T) {
 	assert := With(t)
 
 	content := []byte{'a', 'b', 'c', 'd', 'e', 'f', 'g'}
-	srtp := SRTP{}
+	srtpRaw, err := New(context.Background(), &Config{})
+	assert(err, IsNil)
+
+	srtp := srtpRaw.(*SRTP)
 
 	payload := buf.NewLocal(2048)
 	payload.AppendSupplier(srtp.Write)

transport/internet/headers/utp/utp.go

@@ -26,8 +26,8 @@ func (u *UTP) Write(b []byte) (int, error) {
 	return 4, nil
 }
 
-// NewUTP creates a new UTP header for the given config.
-func NewUTP(ctx context.Context, config interface{}) (interface{}, error) {
+// New creates a new UTP header for the given config.
+func New(ctx context.Context, config interface{}) (interface{}, error) {
 	return &UTP{
 		header:       1,
 		extension:    0,
@@ -36,5 +36,5 @@ func NewUTP(ctx context.Context, config interface{}) (interface{}, error) {
 }
 
 func init() {
-	common.Must(common.RegisterConfig((*Config)(nil), NewUTP))
+	common.Must(common.RegisterConfig((*Config)(nil), New))
 }

transport/internet/headers/utp/utp_test.go

@@ -1,6 +1,7 @@
 package utp_test
 
 import (
+	"context"
 	"testing"
 
 	"v2ray.com/core/common/buf"
@@ -12,7 +13,10 @@ func TestUTPWrite(t *testing.T) {
 	assert := With(t)
 
 	content := []byte{'a', 'b', 'c', 'd', 'e', 'f', 'g'}
-	utp := UTP{}
+	utpRaw, err := New(context.Background(), &Config{})
+	assert(err, IsNil)
+
+	utp := utpRaw.(*UTP)
 
 	payload := buf.NewLocal(2048)
 	payload.AppendSupplier(utp.Write)

transport/internet/headers/wechat/wechat.go

@@ -25,6 +25,7 @@ func (vc *VideoChat) Write(b []byte) (int, error) {
 	return 13, nil
 }
 
+// NewVideoChat returns a new VideoChat instance based on given config.
 func NewVideoChat(ctx context.Context, config interface{}) (interface{}, error) {
 	return &VideoChat{
 		sn: int(dice.RollUint16()),

transport/internet/headers/wechat/wechat_test.go

@@ -1,6 +1,7 @@
 package wechat_test
 
 import (
+	"context"
 	"testing"
 
 	"v2ray.com/core/common/buf"
@@ -11,7 +12,10 @@ import (
 func TestUTPWrite(t *testing.T) {
 	assert := With(t)
 
-	video := VideoChat{}
+	videoRaw, err := NewVideoChat(context.Background(), &VideoConfig{})
+	assert(err, IsNil)
+
+	video := videoRaw.(*VideoChat)
 
 	payload := buf.NewLocal(2048)
 	payload.AppendSupplier(video.Write)

transport/internet/kcp/connection.go

@@ -8,7 +8,6 @@ import (
 	"time"
 
 	"v2ray.com/core/app/log"
-	"v2ray.com/core/common"
 	"v2ray.com/core/common/buf"
 	"v2ray.com/core/common/predicate"
 )
@@ -38,7 +37,7 @@ const (
 	StatePeerClosed      State = 2 // Connection is closed on remote
 	StateTerminating     State = 3 // Connection is ready to be destroyed locally
 	StatePeerTerminating State = 4 // Connection is ready to be destroyed on remote
-	StateTerminated      State = 5 // Connection is detroyed.
+	StateTerminated      State = 5 // Connection is destroyed.
 )
 
 func nowMillisec() int64 {
@@ -137,23 +136,24 @@ func NewUpdater(interval uint32, shouldContinue predicate.Predicate, shouldTermi
 	return u
 }
 
-func (v *Updater) WakeUp() {
+func (u *Updater) WakeUp() {
 	select {
-	case v.notifier <- true:
+	case u.notifier <- true:
 	default:
 	}
 }
 
-func (v *Updater) Run() {
-	for <-v.notifier {
-		if v.shouldTerminate() {
+func (u *Updater) Run() {
+	for <-u.notifier {
+		if u.shouldTerminate() {
 			return
 		}
-		interval := v.Interval()
-		for v.shouldContinue() {
-			v.updateFunc()
-			time.Sleep(interval)
+		ticker := time.NewTicker(u.Interval())
+		for u.shouldContinue() {
+			u.updateFunc()
+			<-ticker.C
 		}
+		ticker.Stop()
 	}
 }
 
@@ -166,13 +166,14 @@ func (u *Updater) SetInterval(d time.Duration) {
 }
 
 type ConnMetadata struct {
-	LocalAddr  net.Addr
-	RemoteAddr net.Addr
+	LocalAddr    net.Addr
+	RemoteAddr   net.Addr
+	Conversation uint16
 }
 
 // Connection is a KCP connection over UDP.
 type Connection struct {
-	meta       *ConnMetadata
+	meta       ConnMetadata
 	closer     io.Closer
 	rd         time.Time
 	wd         time.Time // write deadline
@@ -181,7 +182,6 @@ type Connection struct {
 	dataOutput chan bool
 	Config     *Config
 
-	conv             uint16
 	state            State
 	stateBeginTime   uint32
 	lastIncomingTime uint32
@@ -200,11 +200,10 @@ type Connection struct {
 }
 
 // NewConnection create a new KCP connection between local and remote.
-func NewConnection(conv uint16, meta *ConnMetadata, writer PacketWriter, closer io.Closer, config *Config) *Connection {
-	log.Trace(newError("creating connection ", conv))
+func NewConnection(meta ConnMetadata, writer PacketWriter, closer io.Closer, config *Config) *Connection {
+	log.Trace(newError("creating connection ", meta.Conversation))
 
 	conn := &Connection{
-		conv:       conv,
 		meta:       meta,
 		closer:     closer,
 		since:      nowMillisec(),
@@ -280,24 +279,36 @@ func (v *Connection) ReadMultiBuffer() (buf.MultiBuffer, error) {
 			return nil, io.EOF
 		}
 
-		duration := time.Minute
-		if !v.rd.IsZero() {
-			duration = time.Until(v.rd)
-			if duration < 0 {
-				return nil, ErrIOTimeout
-			}
-		}
-
-		select {
-		case <-v.dataInput:
-		case <-time.After(duration):
-			if !v.rd.IsZero() && v.rd.Before(time.Now()) {
-				return nil, ErrIOTimeout
-			}
+		if err := v.waitForDataInput(); err != nil {
+			return nil, err
 		}
 	}
 }
 
+func (v *Connection) waitForDataInput() error {
+	if v.State() == StatePeerTerminating {
+		return io.EOF
+	}
+
+	duration := time.Minute
+	if !v.rd.IsZero() {
+		duration = time.Until(v.rd)
+		if duration < 0 {
+			return ErrIOTimeout
+		}
+	}
+
+	select {
+	case <-v.dataInput:
+	case <-time.After(duration):
+		if !v.rd.IsZero() && v.rd.Before(time.Now()) {
+			return ErrIOTimeout
+		}
+	}
+
+	return nil
+}
+
 // Read implements the Conn Read method.
 func (v *Connection) Read(b []byte) (int, error) {
 	if v == nil {
@@ -313,28 +324,32 @@ func (v *Connection) Read(b []byte) (int, error) {
 			return nBytes, nil
 		}
 
-		if v.State() == StatePeerTerminating {
-			return 0, io.EOF
-		}
-
-		duration := time.Minute
-		if !v.rd.IsZero() {
-			duration = time.Until(v.rd)
-			if duration < 0 {
-				return 0, ErrIOTimeout
-			}
-		}
-
-		select {
-		case <-v.dataInput:
-		case <-time.After(duration):
-			if !v.rd.IsZero() && v.rd.Before(time.Now()) {
-				return 0, ErrIOTimeout
-			}
+		if err := v.waitForDataInput(); err != nil {
+			return 0, err
 		}
 	}
 }
 
+func (v *Connection) waitForDataOutput() error {
+	duration := time.Minute
+	if !v.wd.IsZero() {
+		duration = time.Until(v.wd)
+		if duration < 0 {
+			return ErrIOTimeout
+		}
+	}
+
+	select {
+	case <-v.dataOutput:
+	case <-time.After(duration):
+		if !v.wd.IsZero() && v.wd.Before(time.Now()) {
+			return ErrIOTimeout
+		}
+	}
+
+	return nil
+}
+
 // Write implements io.Writer.
 func (v *Connection) Write(b []byte) (int, error) {
 	totalWritten := 0
@@ -344,35 +359,19 @@ func (v *Connection) Write(b []byte) (int, error) {
 			return totalWritten, io.ErrClosedPipe
 		}
 
-		for {
-			rb := v.sendingWorker.Push()
-			if rb == nil {
-				break
-			}
-			common.Must(rb.Reset(func(bb []byte) (int, error) {
-				return copy(bb[:v.mss], b[totalWritten:]), nil
-			}))
+		for v.sendingWorker.Push(func(bb []byte) (int, error) {
+			n := copy(bb[:v.mss], b[totalWritten:])
+			totalWritten += n
+			return n, nil
+		}) {
 			v.dataUpdater.WakeUp()
-			totalWritten += rb.Len()
 			if totalWritten == len(b) {
 				return totalWritten, nil
 			}
 		}
 
-		duration := time.Minute
-		if !v.wd.IsZero() {
-			duration = time.Until(v.wd)
-			if duration < 0 {
-				return totalWritten, ErrIOTimeout
-			}
-		}
-
-		select {
-		case <-v.dataOutput:
-		case <-time.After(duration):
-			if !v.wd.IsZero() && v.wd.Before(time.Now()) {
-				return totalWritten, ErrIOTimeout
-			}
+		if err := v.waitForDataOutput(); err != nil {
+			return totalWritten, err
 		}
 	}
 }
@@ -386,34 +385,17 @@ func (v *Connection) WriteMultiBuffer(mb buf.MultiBuffer) error {
 			return io.ErrClosedPipe
 		}
 
-		for {
-			rb := v.sendingWorker.Push()
-			if rb == nil {
-				break
-			}
-			common.Must(rb.Reset(func(bb []byte) (int, error) {
-				return mb.Read(bb[:v.mss])
-			}))
+		for v.sendingWorker.Push(func(bb []byte) (int, error) {
+			return mb.Read(bb[:v.mss])
+		}) {
 			v.dataUpdater.WakeUp()
 			if mb.IsEmpty() {
 				return nil
 			}
 		}
 
-		duration := time.Minute
-		if !v.wd.IsZero() {
-			duration = time.Until(v.wd)
-			if duration < 0 {
-				return ErrIOTimeout
-			}
-		}
-
-		select {
-		case <-v.dataOutput:
-		case <-time.After(duration):
-			if !v.wd.IsZero() && v.wd.Before(time.Now()) {
-				return ErrIOTimeout
-			}
+		if err := v.waitForDataOutput(); err != nil {
+			return err
 		}
 	}
 }
@@ -422,7 +404,7 @@ func (v *Connection) SetState(state State) {
 	current := v.Elapsed()
 	atomic.StoreInt32((*int32)(&v.state), int32(state))
 	atomic.StoreUint32(&v.stateBeginTime, current)
-	log.Trace(newError("#", v.conv, " entering state ", state, " at ", current).AtDebug())
+	log.Trace(newError("#", v.meta.Conversation, " entering state ", state, " at ", current).AtDebug())
 
 	switch state {
 	case StateReadyToClose:
@@ -561,7 +543,7 @@ func (v *Connection) Input(segments []Segment) {
 	atomic.StoreUint32(&v.lastIncomingTime, current)
 
 	for _, seg := range segments {
-		if seg.Conversation() != v.conv {
+		if seg.Conversation() != v.meta.Conversation {
 			break
 		}
 
@@ -618,7 +600,7 @@ func (v *Connection) flush() {
 	}
 
 	if v.State() == StateTerminating {
-		log.Trace(newError("#", v.conv, " sending terminating cmd.").AtDebug())
+		log.Trace(newError("#", v.meta.Conversation, " sending terminating cmd.").AtDebug())
 		v.Ping(current, CommandTerminate)
 
 		if current-atomic.LoadUint32(&v.stateBeginTime) > 8000 {
@@ -649,7 +631,7 @@ func (v *Connection) State() State {
 
 func (v *Connection) Ping(current uint32, cmd Command) {
 	seg := NewCmdOnlySegment()
-	seg.Conv = v.conv
+	seg.Conv = v.meta.Conversation
 	seg.Cmd = cmd
 	seg.ReceivinNext = v.receivingWorker.NextNumber()
 	seg.SendingNext = v.sendingWorker.FirstUnacknowledged()

transport/internet/kcp/connection_test.go

@@ -19,7 +19,7 @@ func (NoOpCloser) Close() error {
 func TestConnectionReadTimeout(t *testing.T) {
 	assert := With(t)
 
-	conn := NewConnection(1, &ConnMetadata{}, &KCPPacketWriter{
+	conn := NewConnection(ConnMetadata{Conversation: 1}, &KCPPacketWriter{
 		Writer: buf.DiscardBytes,
 	}, NoOpCloser(0), &Config{})
 	conn.SetReadDeadline(time.Now().Add(time.Second))

transport/internet/kcp/dialer.go

@@ -67,25 +67,19 @@ func DialKCP(ctx context.Context, dest net.Destination) (internet.Connection, er
 	}
 
 	conv := uint16(atomic.AddUint32(&globalConv, 1))
-	session := NewConnection(conv, &ConnMetadata{
-		LocalAddr:  rawConn.LocalAddr(),
-		RemoteAddr: rawConn.RemoteAddr(),
+	session := NewConnection(ConnMetadata{
+		LocalAddr:    rawConn.LocalAddr(),
+		RemoteAddr:   rawConn.RemoteAddr(),
+		Conversation: conv,
 	}, writer, rawConn, kcpSettings)
 
 	go fetchInput(ctx, rawConn, reader, session)
 
 	var iConn internet.Connection = session
 
-	if securitySettings := internet.SecuritySettingsFromContext(ctx); securitySettings != nil {
-		switch securitySettings := securitySettings.(type) {
-		case *v2tls.Config:
-			if dest.Address.Family().IsDomain() {
-				securitySettings.OverrideServerNameIfEmpty(dest.Address.Domain())
-			}
-			config := securitySettings.GetTLSConfig()
-			tlsConn := tls.Client(iConn, config)
-			iConn = tlsConn
-		}
+	if config := v2tls.ConfigFromContext(ctx, v2tls.WithDestination(dest)); config != nil {
+		tlsConn := tls.Client(iConn, config.GetTLSConfig())
+		iConn = tlsConn
 	}
 
 	return iConn, nil

transport/internet/kcp/listener.go

@@ -59,13 +59,11 @@ func NewListener(ctx context.Context, address net.Address, port net.Port, addCon
 		config:   kcpSettings,
 		addConn:  addConn,
 	}
-	securitySettings := internet.SecuritySettingsFromContext(ctx)
-	if securitySettings != nil {
-		switch securitySettings := securitySettings.(type) {
-		case *v2tls.Config:
-			l.tlsConfig = securitySettings.GetTLSConfig()
-		}
+
+	if config := v2tls.ConfigFromContext(ctx); config != nil {
+		l.tlsConfig = config.GetTLSConfig()
 	}
+
 	hub, err := udp.ListenUDP(address, port, udp.ListenOption{Callback: l.OnReceive, Concurrency: 2})
 	if err != nil {
 		return nil, err
@@ -124,9 +122,10 @@ func (v *Listener) OnReceive(payload *buf.Buffer, src net.Destination, originalD
 			Port: int(src.Port),
 		}
 		localAddr := v.hub.Addr()
-		conn = NewConnection(conv, &ConnMetadata{
-			LocalAddr:  localAddr,
-			RemoteAddr: remoteAddr,
+		conn = NewConnection(ConnMetadata{
+			LocalAddr:    localAddr,
+			RemoteAddr:   remoteAddr,
+			Conversation: conv,
 		}, &KCPPacketWriter{
 			Header:   v.header,
 			Security: v.security,

transport/internet/kcp/receiving.go

@@ -252,7 +252,7 @@ func (w *ReceivingWorker) Flush(current uint32) {
 
 func (w *ReceivingWorker) Write(seg Segment) error {
 	ackSeg := seg.(*AckSegment)
-	ackSeg.Conv = w.conn.conv
+	ackSeg.Conv = w.conn.meta.Conversation
 	ackSeg.ReceivingNext = w.nextNumber
 	ackSeg.ReceivingWindow = w.nextNumber + w.windowSize
 	if w.conn.State() == StateReadyToClose {

transport/internet/kcp/sending.go

@@ -3,6 +3,7 @@ package kcp
 import (
 	"sync"
 
+	"v2ray.com/core/common"
 	"v2ray.com/core/common/buf"
 )
 
@@ -284,23 +285,24 @@ func (v *SendingWorker) ProcessSegment(current uint32, seg *AckSegment, rto uint
 	}
 }
 
-func (v *SendingWorker) Push() *buf.Buffer {
+func (v *SendingWorker) Push(f buf.Supplier) bool {
 	v.Lock()
 	defer v.Unlock()
 
 	if v.window.IsFull() {
-		return nil
+		return false
 	}
 
 	b := v.window.Push(v.nextNumber)
 	v.nextNumber++
-	return b
+	common.Must(b.Reset(f))
+	return true
 }
 
 func (v *SendingWorker) Write(seg Segment) error {
 	dataSeg := seg.(*DataSegment)
 
-	dataSeg.Conv = v.conn.conv
+	dataSeg.Conv = v.conn.meta.Conversation
 	dataSeg.SendingNext = v.firstUnacknowledged
 	dataSeg.Option = 0
 	if v.conn.State() == StateReadyToClose {

transport/internet/tcp/dialer.go

@@ -19,22 +19,16 @@ func getTCPSettingsFromContext(ctx context.Context) *Config {
 }
 
 func Dial(ctx context.Context, dest net.Destination) (internet.Connection, error) {
-	log.Trace(newError("dailing TCP to ", dest))
+	log.Trace(newError("dialing TCP to ", dest))
 	src := internet.DialerSourceFromContext(ctx)
 
 	conn, err := internet.DialSystem(ctx, src, dest)
 	if err != nil {
 		return nil, err
 	}
-	if securitySettings := internet.SecuritySettingsFromContext(ctx); securitySettings != nil {
-		tlsConfig, ok := securitySettings.(*tls.Config)
-		if ok {
-			if dest.Address.Family().IsDomain() {
-				tlsConfig.OverrideServerNameIfEmpty(dest.Address.Domain())
-			}
-			config := tlsConfig.GetTLSConfig()
-			conn = tls.Client(conn, config)
-		}
+
+	if config := tls.ConfigFromContext(ctx, tls.WithDestination(dest)); config != nil {
+		conn = tls.Client(conn, config.GetTLSConfig())
 	}
 
 	tcpSettings := getTCPSettingsFromContext(ctx)

transport/internet/tcp/hub.go

@@ -37,12 +37,11 @@ func ListenTCP(ctx context.Context, address net.Address, port net.Port, addConn
 		config:   tcpSettings,
 		addConn:  addConn,
 	}
-	if securitySettings := internet.SecuritySettingsFromContext(ctx); securitySettings != nil {
-		tlsConfig, ok := securitySettings.(*tls.Config)
-		if ok {
-			l.tlsConfig = tlsConfig.GetTLSConfig()
-		}
+
+	if config := tls.ConfigFromContext(ctx); config != nil {
+		l.tlsConfig = config.GetTLSConfig()
 	}
+
 	if tcpSettings.HeaderSettings != nil {
 		headerConfig, err := tcpSettings.HeaderSettings.GetInstance()
 		if err != nil {

transport/internet/tls/config.go

@@ -1,9 +1,12 @@
 package tls
 
 import (
+	"context"
 	"crypto/tls"
 
 	"v2ray.com/core/app/log"
+	"v2ray.com/core/common/net"
+	"v2ray.com/core/transport/internet"
 )
 
 var (
@@ -42,8 +45,26 @@ func (c *Config) GetTLSConfig() *tls.Config {
 	return config
 }
 
-func (c *Config) OverrideServerNameIfEmpty(serverName string) {
-	if len(c.ServerName) == 0 {
-		c.ServerName = serverName
+type Option func(*Config)
+
+func WithDestination(dest net.Destination) Option {
+	return func(config *Config) {
+		if dest.Address.Family().IsDomain() && len(config.ServerName) == 0 {
+			config.ServerName = dest.Address.Domain()
+		}
 	}
 }
+
+func ConfigFromContext(ctx context.Context, opts ...Option) *Config {
+	securitySettings := internet.SecuritySettingsFromContext(ctx)
+	if securitySettings == nil {
+		return nil
+	}
+	if config, ok := securitySettings.(*Config); ok {
+		for _, opt := range opts {
+			opt(config)
+		}
+		return config
+	}
+	return nil
+}

transport/internet/websocket/dialer.go

@@ -35,22 +35,16 @@ func dialWebsocket(ctx context.Context, dest net.Destination) (net.Conn, error)
 		NetDial: func(network, addr string) (net.Conn, error) {
 			return internet.DialSystem(ctx, src, dest)
 		},
-		ReadBufferSize:   8 * 1024,
-		WriteBufferSize:  8 * 1024,
+		ReadBufferSize:   4 * 1024,
+		WriteBufferSize:  4 * 1024,
 		HandshakeTimeout: time.Second * 8,
 	}
 
 	protocol := "ws"
 
-	if securitySettings := internet.SecuritySettingsFromContext(ctx); securitySettings != nil {
-		tlsConfig, ok := securitySettings.(*tls.Config)
-		if ok {
-			protocol = "wss"
-			if dest.Address.Family().IsDomain() {
-				tlsConfig.OverrideServerNameIfEmpty(dest.Address.Domain())
-			}
-			dialer.TLSClientConfig = tlsConfig.GetTLSConfig()
-		}
+	if config := tls.ConfigFromContext(ctx, tls.WithDestination(dest)); config != nil {
+		protocol = "wss"
+		dialer.TLSClientConfig = config.GetTLSConfig()
 	}
 
 	host := dest.NetAddr()

transport/internet/websocket/hub.go

@@ -22,8 +22,8 @@ type requestHandler struct {
 }
 
 var upgrader = &websocket.Upgrader{
-	ReadBufferSize:   8 * 1024,
-	WriteBufferSize:  8 * 1024,
+	ReadBufferSize:   4 * 1024,
+	WriteBufferSize:  4 * 1024,
 	HandshakeTimeout: time.Second * 8,
 }
 
@@ -59,11 +59,8 @@ func ListenWS(ctx context.Context, address net.Address, port net.Port, addConn i
 		config:  wsSettings,
 		addConn: addConn,
 	}
-	if securitySettings := internet.SecuritySettingsFromContext(ctx); securitySettings != nil {
-		tlsConfig, ok := securitySettings.(*v2tls.Config)
-		if ok {
-			l.tlsConfig = tlsConfig.GetTLSConfig()
-		}
+	if config := v2tls.ConfigFromContext(ctx); config != nil {
+		l.tlsConfig = config.GetTLSConfig()
 	}
 
 	err := l.listenws(address, port)