fix a data race in KCP

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
 }

core.go

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

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"
 )
@@ -360,16 +359,12 @@ 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
 			}
@@ -390,14 +385,9 @@ 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

transport/internet/kcp/dialer.go

@@ -77,16 +77,9 @@ func DialKCP(ctx context.Context, dest net.Destination) (internet.Connection, er
 
 	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

transport/internet/kcp/sending.go

@@ -3,6 +3,7 @@ package kcp
 import (
 	"sync"
 
+	"v2ray.com/core/common"
 	"v2ray.com/core/common/buf"
 )
 
@@ -284,17 +285,18 @@ 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 {

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)