fix buffer recycling

Fix typo

app/commander/outbound.go

@@ -27,7 +27,7 @@ func (l *OutboundListener) add(conn net.Conn) {
 func (l *OutboundListener) Accept() (net.Conn, error) {
 	select {
 	case <-l.done.C():
-		return nil, newError("listern closed")
+		return nil, newError("listen closed")
 	case c := <-l.buffer:
 		return c, nil
 	}
@@ -76,8 +76,6 @@ func (co *CommanderOutbound) Dispatch(ctx context.Context, r ray.OutboundRay) {
 	co.listener.add(c)
 	co.access.RUnlock()
 	<-closeSignal.Wait()
-
-	return
 }
 
 func (co *CommanderOutbound) Tag() string {

app/dispatcher/default.go

@@ -55,7 +55,7 @@ func (d *DefaultDispatcher) Dispatch(ctx context.Context, destination net.Destin
 	ctx = proxy.ContextWithTarget(ctx, destination)
 
 	outbound := ray.NewRay(ctx)
-	snifferList := proxyman.ProtocoSniffersFromContext(ctx)
+	snifferList := proxyman.ProtocolSniffersFromContext(ctx)
 	if destination.Address.Family().IsDomain() || len(snifferList) == 0 {
 		go d.routedDispatch(ctx, outbound, destination)
 	} else {
@@ -110,7 +110,7 @@ func (d *DefaultDispatcher) routedDispatch(ctx context.Context, outbound ray.Out
 				newError("taking detour [", tag, "] for [", destination, "]").WithContext(ctx).WriteToLog()
 				dispatcher = handler
 			} else {
-				newError("nonexisting tag: ", tag).AtWarning().WithContext(ctx).WriteToLog()
+				newError("non existing tag: ", tag).AtWarning().WithContext(ctx).WriteToLog()
 			}
 		} else {
 			newError("default route for ", destination).WithContext(ctx).WriteToLog()

app/dns/nameserver.go

@@ -177,6 +177,9 @@ func (s *UDPNameServer) QueryA(domain string) <-chan *ARecord {
 	b, err := msgToBuffer(msg)
 	if err != nil {
 		newError("failed to build A query for domain ", domain).Base(err).WriteToLog()
+		s.Lock()
+		delete(s.requests, id)
+		s.Unlock()
 		close(response)
 		return response
 	}

app/log/command/command.go

@@ -3,7 +3,8 @@ package command
 //go:generate go run $GOPATH/src/v2ray.com/core/common/errors/errorgen/main.go -pkg command -path App,Log,Command
 
 import (
-	context "golang.org/x/net/context"
+	"context"
+
 	grpc "google.golang.org/grpc"
 
 	"v2ray.com/core"

app/log/command/config.pb.go

@@ -5,7 +5,8 @@ import fmt "fmt"
 import math "math"
 
 import (
-	context "golang.org/x/net/context"
+	"context"
+
 	grpc "google.golang.org/grpc"
 )
 

app/proxyman/command/command.go

@@ -11,7 +11,7 @@ import (
 
 // InboundOperation is the interface for operations that applies to inbound handlers.
 type InboundOperation interface {
-	// ApplyInbound appliess this operation to the given inbound handler.
+	// ApplyInbound applies this operation to the given inbound handler.
 	ApplyInbound(context.Context, core.InboundHandler) error
 }
 

app/proxyman/command/command.pb.go

@@ -8,7 +8,8 @@ import v2ray_core_common_serial "v2ray.com/core/common/serial"
 import v2ray_core "v2ray.com/core"
 
 import (
-	context "golang.org/x/net/context"
+	"context"
+
 	grpc "google.golang.org/grpc"
 )
 

app/proxyman/mux/reader.go

@@ -51,12 +51,7 @@ func (r *PacketReader) ReadMultiBuffer() (buf.MultiBuffer, error) {
 		return nil, err
 	}
 
-	var b *buf.Buffer
-	if size <= buf.Size {
-		b = buf.New()
-	} else {
-		b = buf.NewLocal(int(size))
-	}
+	b := buf.NewSize(uint32(size))
 	if err := b.AppendSupplier(buf.ReadFullFrom(r.reader, int(size))); err != nil {
 		b.Release()
 		return nil, err
@@ -68,7 +63,7 @@ func (r *PacketReader) ReadMultiBuffer() (buf.MultiBuffer, error) {
 // StreamReader reads Mux frame as a stream.
 type StreamReader struct {
 	reader   *buf.BufferedReader
-	leftOver int
+	leftOver int32
 }
 
 // NewStreamReader creates a new StreamReader.
@@ -91,10 +86,10 @@ func (r *StreamReader) ReadMultiBuffer() (buf.MultiBuffer, error) {
 		if err != nil {
 			return nil, err
 		}
-		r.leftOver = int(size)
+		r.leftOver = int32(size)
 	}
 
-	mb, err := r.reader.ReadAtMost(r.leftOver)
-	r.leftOver -= mb.Len()
+	mb, err := r.reader.ReadAtMost(int(r.leftOver))
+	r.leftOver -= int32(mb.Len())
 	return mb, err
 }

app/proxyman/proxyman.go

@@ -17,7 +17,7 @@ func ContextWithProtocolSniffers(ctx context.Context, list []KnownProtocols) con
 	return context.WithValue(ctx, protocolsKey, list)
 }
 
-func ProtocoSniffersFromContext(ctx context.Context) []KnownProtocols {
+func ProtocolSniffersFromContext(ctx context.Context) []KnownProtocols {
 	if list, ok := ctx.Value(protocolsKey).([]KnownProtocols); ok {
 		return list
 	}

common/buf/buffer.go

@@ -12,11 +12,10 @@ type Supplier func([]byte) (int, error)
 // the buffer into an internal buffer pool, in order to recreate a buffer more
 // quickly.
 type Buffer struct {
-	v    []byte
-	pool Pool
+	v []byte
 
-	start int
-	end   int
+	start int32
+	end   int32
 }
 
 // Release recycles the buffer into an internal buffer pool.
@@ -24,11 +23,8 @@ func (b *Buffer) Release() {
 	if b == nil || b.v == nil {
 		return
 	}
-	if b.pool != nil {
-		b.pool.Free(b)
-	}
+	freeBytes(b.v)
 	b.v = nil
-	b.pool = nil
 	b.start = 0
 	b.end = 0
 }
@@ -48,25 +44,25 @@ func (b *Buffer) AppendBytes(bytes ...byte) int {
 // Append appends a byte array to the end of the buffer.
 func (b *Buffer) Append(data []byte) int {
 	nBytes := copy(b.v[b.end:], data)
-	b.end += nBytes
+	b.end += int32(nBytes)
 	return nBytes
 }
 
 // AppendSupplier appends the content of a BytesWriter to the buffer.
 func (b *Buffer) AppendSupplier(writer Supplier) error {
 	nBytes, err := writer(b.v[b.end:])
-	b.end += nBytes
+	b.end += int32(nBytes)
 	return err
 }
 
 // Byte returns the bytes at index.
 func (b *Buffer) Byte(index int) byte {
-	return b.v[b.start+index]
+	return b.v[b.start+int32(index)]
 }
 
 // SetByte sets the byte value at index.
 func (b *Buffer) SetByte(index int, value byte) {
-	b.v[b.start+index] = value
+	b.v[b.start+int32(index)] = value
 }
 
 // Bytes returns the content bytes of this Buffer.
@@ -78,7 +74,7 @@ func (b *Buffer) Bytes() []byte {
 func (b *Buffer) Reset(writer Supplier) error {
 	nBytes, err := writer(b.v)
 	b.start = 0
-	b.end = nBytes
+	b.end = int32(nBytes)
 	return err
 }
 
@@ -90,7 +86,7 @@ func (b *Buffer) BytesRange(from, to int) []byte {
 	if to < 0 {
 		to += b.Len()
 	}
-	return b.v[b.start+from : b.start+to]
+	return b.v[b.start+int32(from) : b.start+int32(to)]
 }
 
 // BytesFrom returns a slice of this Buffer starting from the given position.
@@ -98,7 +94,7 @@ func (b *Buffer) BytesFrom(from int) []byte {
 	if from < 0 {
 		from += b.Len()
 	}
-	return b.v[b.start+from : b.end]
+	return b.v[b.start+int32(from) : b.end]
 }
 
 // BytesTo returns a slice of this Buffer from start to the given position.
@@ -106,7 +102,7 @@ func (b *Buffer) BytesTo(to int) []byte {
 	if to < 0 {
 		to += b.Len()
 	}
-	return b.v[b.start : b.start+to]
+	return b.v[b.start : b.start+int32(to)]
 }
 
 // Slice cuts the buffer at the given position.
@@ -120,8 +116,8 @@ func (b *Buffer) Slice(from, to int) {
 	if to < from {
 		panic("Invalid slice")
 	}
-	b.end = b.start + to
-	b.start += from
+	b.end = b.start + int32(to)
+	b.start += int32(from)
 }
 
 // SliceFrom cuts the buffer at the given position.
@@ -129,7 +125,7 @@ func (b *Buffer) SliceFrom(from int) {
 	if from < 0 {
 		from += b.Len()
 	}
-	b.start += from
+	b.start += int32(from)
 }
 
 // Len returns the length of the buffer content.
@@ -137,7 +133,7 @@ func (b *Buffer) Len() int {
 	if b == nil {
 		return 0
 	}
-	return b.end - b.start
+	return int(b.end - b.start)
 }
 
 // IsEmpty returns true if the buffer is empty.
@@ -147,13 +143,13 @@ func (b *Buffer) IsEmpty() bool {
 
 // IsFull returns true if the buffer has no more room to grow.
 func (b *Buffer) IsFull() bool {
-	return b.end == len(b.v)
+	return b.end == int32(len(b.v))
 }
 
 // Write implements Write method in io.Writer.
 func (b *Buffer) Write(data []byte) (int, error) {
 	nBytes := copy(b.v[b.end:], data)
-	b.end += nBytes
+	b.end += int32(nBytes)
 	return nBytes, nil
 }
 
@@ -166,7 +162,7 @@ func (b *Buffer) Read(data []byte) (int, error) {
 	if nBytes == b.Len() {
 		b.Clear()
 	} else {
-		b.start += nBytes
+		b.start += int32(nBytes)
 	}
 	return nBytes, nil
 }
@@ -176,15 +172,16 @@ func (b *Buffer) String() string {
 	return string(b.Bytes())
 }
 
-// New creates a Buffer with 0 length and 8K capacity.
+// New creates a Buffer with 0 length and 2K capacity.
 func New() *Buffer {
-	return mediumPool.Allocate()
-}
-
-// NewLocal creates and returns a buffer with 0 length and given capacity on current thread.
-func NewLocal(size int) *Buffer {
 	return &Buffer{
-		v:    make([]byte, size),
-		pool: nil,
+		v: pool[0].Get().([]byte),
+	}
+}
+
+// NewSize creates and returns a buffer with 0 length and at least the given capacity.
+func NewSize(size uint32) *Buffer {
+	return &Buffer{
+		v: newBytes(size),
 	}
 }

common/buf/buffer_pool.go

@@ -4,49 +4,55 @@ import (
 	"sync"
 )
 
-// Pool provides functionality to generate and recycle buffers on demand.
-type Pool interface {
-	// Allocate either returns a unused buffer from the pool, or generates a new one from system.
-	Allocate() *Buffer
-	// Free recycles the given buffer.
-	Free(*Buffer)
-}
-
-// SyncPool is a buffer pool based on sync.Pool
-type SyncPool struct {
-	allocator *sync.Pool
-}
-
-// NewSyncPool creates a SyncPool with given buffer size.
-func NewSyncPool(bufferSize uint32) *SyncPool {
-	pool := &SyncPool{
-		allocator: &sync.Pool{
-			New: func() interface{} { return make([]byte, bufferSize) },
-		},
-	}
-	return pool
-}
-
-// Allocate implements Pool.Allocate().
-func (p *SyncPool) Allocate() *Buffer {
-	return &Buffer{
-		v:    p.allocator.Get().([]byte),
-		pool: p,
-	}
-}
-
-// Free implements Pool.Free().
-func (p *SyncPool) Free(buffer *Buffer) {
-	if buffer.v != nil {
-		p.allocator.Put(buffer.v)
-	}
-}
-
 const (
 	// Size of a regular buffer.
 	Size = 2 * 1024
 )
 
-var (
-	mediumPool Pool = NewSyncPool(Size)
+func createAllocFunc(size uint32) func() interface{} {
+	return func() interface{} {
+		return make([]byte, size)
+	}
+}
+
+const (
+	numPools  = 5
+	sizeMulti = 4
 )
+
+var (
+	pool     [numPools]sync.Pool
+	poolSize [numPools]uint32
+)
+
+func init() {
+	size := uint32(Size)
+	for i := 0; i < numPools; i++ {
+		pool[i] = sync.Pool{
+			New: createAllocFunc(size),
+		}
+		poolSize[i] = size
+		size *= sizeMulti
+	}
+}
+
+func newBytes(size uint32) []byte {
+	for idx, ps := range poolSize {
+		if size <= ps {
+			return pool[idx].Get().([]byte)
+		}
+	}
+	return make([]byte, size)
+}
+
+func freeBytes(b []byte) {
+	size := uint32(cap(b))
+	b = b[0:cap(b)]
+	for i := numPools - 1; i >= 0; i-- {
+		ps := poolSize[i]
+		if size >= ps {
+			pool[i].Put(b)
+			return
+		}
+	}
+}

common/buf/buffer_test.go

@@ -1,7 +1,6 @@
 package buf_test
 
 import (
-	"crypto/rand"
 	"testing"
 
 	. "v2ray.com/core/common/buf"
@@ -42,32 +41,6 @@ func TestBufferString(t *testing.T) {
 	assert(buffer.String(), Equals, "Test String")
 }
 
-func TestBufferWrite(t *testing.T) {
-	assert := With(t)
-
-	buffer := NewLocal(8)
-	nBytes, err := buffer.Write([]byte("abcd"))
-	assert(err, IsNil)
-	assert(nBytes, Equals, 4)
-	nBytes, err = buffer.Write([]byte("abcde"))
-	assert(err, IsNil)
-	assert(nBytes, Equals, 4)
-	assert(buffer.String(), Equals, "abcdabcd")
-}
-
-func TestSyncPool(t *testing.T) {
-	assert := With(t)
-
-	p := NewSyncPool(32)
-	b := p.Allocate()
-	assert(b.Len(), Equals, 0)
-
-	assert(b.AppendSupplier(ReadFrom(rand.Reader)), IsNil)
-	assert(b.Len(), Equals, 32)
-
-	b.Release()
-}
-
 func BenchmarkNewBuffer(b *testing.B) {
 	for i := 0; i < b.N; i++ {
 		buffer := New()
@@ -77,7 +50,7 @@ func BenchmarkNewBuffer(b *testing.B) {
 
 func BenchmarkNewLocalBuffer(b *testing.B) {
 	for i := 0; i < b.N; i++ {
-		buffer := NewLocal(Size)
+		buffer := NewSize(Size)
 		buffer.Release()
 	}
 }

common/buf/reader.go

@@ -19,14 +19,13 @@ func NewBytesToBufferReader(reader io.Reader) Reader {
 	}
 }
 
-const mediumSize = 8 * 1024
-const largeSize = 64 * 1024
+const xlSize = 128 * 1024
 
 func (r *BytesToBufferReader) readSmall() (MultiBuffer, error) {
 	b := New()
 	err := b.Reset(ReadFrom(r.Reader))
 	if b.IsFull() {
-		r.buffer = make([]byte, mediumSize)
+		r.buffer = newBytes(Size + 1)
 	}
 	if !b.IsEmpty() {
 		return NewMultiBufferValue(b), nil
@@ -45,11 +44,14 @@ 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)
+		if nBytes == len(r.buffer) && nBytes < xlSize {
+			freeBytes(r.buffer)
+			r.buffer = newBytes(uint32(nBytes) + 1)
 		}
 		return mb, nil
 	}
+	freeBytes(r.buffer)
+	r.buffer = nil
 	return nil, err
 }
 

common/buf/reader_test.go

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

common/crypto/auth.go

@@ -124,12 +124,7 @@ func (r *AuthenticationReader) ReadMultiBuffer() (buf.MultiBuffer, error) {
 		return nil, io.EOF
 	}
 
-	var b *buf.Buffer
-	if size <= buf.Size {
-		b = buf.New()
-	} else {
-		b = buf.NewLocal(size)
-	}
+	b := buf.NewSize(uint32(size))
 	if err := b.Reset(buf.ReadFullFrom(r.reader, size)); err != nil {
 		b.Release()
 		return nil, err

common/crypto/auth_test.go

@@ -24,13 +24,15 @@ func TestAuthenticationReaderWriter(t *testing.T) {
 	aead, err := cipher.NewGCM(block)
 	assert(err, IsNil)
 
-	rawPayload := make([]byte, 8192*10)
+	const payloadSize = 1024 * 80
+	rawPayload := make([]byte, payloadSize)
 	rand.Read(rawPayload)
 
-	payload := buf.NewLocal(8192 * 10)
+	payload := buf.NewSize(payloadSize)
 	payload.Append(rawPayload)
+	assert(payload.Len(), Equals, payloadSize)
 
-	cache := buf.NewLocal(160 * 1024)
+	cache := buf.NewSize(160 * 1024)
 	iv := make([]byte, 12)
 	rand.Read(iv)
 
@@ -45,7 +47,6 @@ func TestAuthenticationReaderWriter(t *testing.T) {
 	assert(writer.WriteMultiBuffer(buf.NewMultiBufferValue(payload)), IsNil)
 	assert(cache.Len(), Equals, 82658)
 	assert(writer.WriteMultiBuffer(buf.MultiBuffer{}), IsNil)
-	assert(err, IsNil)
 
 	reader := NewAuthenticationReader(&AEADAuthenticator{
 		AEAD: aead,
@@ -57,14 +58,16 @@ func TestAuthenticationReaderWriter(t *testing.T) {
 
 	var mb buf.MultiBuffer
 
-	for mb.Len() < len(rawPayload) {
+	for mb.Len() < payloadSize {
 		mb2, err := reader.ReadMultiBuffer()
 		assert(err, IsNil)
 
 		mb.AppendMulti(mb2)
 	}
 
-	mbContent := make([]byte, 8192*10)
+	assert(mb.Len(), Equals, payloadSize)
+
+	mbContent := make([]byte, payloadSize)
 	mb.Read(mbContent)
 	assert(mbContent, Equals, rawPayload)
 
@@ -83,7 +86,7 @@ func TestAuthenticationReaderWriterPacket(t *testing.T) {
 	aead, err := cipher.NewGCM(block)
 	assert(err, IsNil)
 
-	cache := buf.NewLocal(1024)
+	cache := buf.NewSize(1024)
 	iv := make([]byte, 12)
 	rand.Read(iv)
 

common/crypto/chunk_test.go

@@ -12,7 +12,7 @@ import (
 func TestChunkStreamIO(t *testing.T) {
 	assert := With(t)
 
-	cache := buf.NewLocal(8192)
+	cache := buf.NewSize(8192)
 
 	writer := NewChunkStreamWriter(PlainChunkSizeParser{}, cache)
 	reader := NewChunkStreamReader(PlainChunkSizeParser{}, cache)

common/net/address.go

@@ -22,7 +22,7 @@ var (
 )
 
 // AddressFamily is the type of address.
-type AddressFamily int
+type AddressFamily byte
 
 const (
 	// AddressFamilyIPv4 represents address as IPv4

common/net/destination.go

@@ -6,9 +6,9 @@ import (
 
 // Destination represents a network destination including address and protocol (tcp / udp).
 type Destination struct {
-	Network Network
-	Port    Port
 	Address Address
+	Port    Port
+	Network Network
 }
 
 // DestinationFromAddr generates a Destination from a net address.
@@ -56,7 +56,7 @@ func (d Destination) IsValid() bool {
 	return d.Network != Network_Unknown
 }
 
-// AsDestination converts current Enpoint into Destination.
+// AsDestination converts current Endpoint into Destination.
 func (p *Endpoint) AsDestination() Destination {
 	return Destination{
 		Network: p.Network,

common/platform/platform.go

@@ -53,7 +53,7 @@ func getExecutableDir() string {
 	return filepath.Dir(exec)
 }
 
-func getExecuableSubDir(dir string) func() string {
+func getExecutableSubDir(dir string) func() string {
 	return func() string {
 		return filepath.Join(getExecutableDir(), dir)
 	}
@@ -67,7 +67,7 @@ func GetAssetLocation(file string) string {
 
 func GetPluginDirectory() string {
 	const name = "v2ray.location.plugin"
-	pluginDir := EnvFlag{Name: name, AltName: NormalizeEnvName(name)}.GetValue(getExecuableSubDir("plugins"))
+	pluginDir := EnvFlag{Name: name, AltName: NormalizeEnvName(name)}.GetValue(getExecutableSubDir("plugins"))
 	return pluginDir
 }
 

common/protocol/address.go

@@ -61,7 +61,7 @@ func maybeIPPrefix(b byte) bool {
 
 func isValidDomain(d string) bool {
 	for _, c := range d {
-		if !((c >= '0' && c <= '9') || (c >= 'a' && c <= 'z') || (c >= 'A' && c <= 'Z') || c == '-' || c == '.') {
+		if !((c >= '0' && c <= '9') || (c >= 'a' && c <= 'z') || (c >= 'A' && c <= 'Z') || c == '-' || c == '.' || c == '_') {
 			return false
 		}
 	}
@@ -150,10 +150,8 @@ func (p *AddressParser) ReadAddressPort(buffer *buf.Buffer, input io.Reader) (ne
 }
 
 func (p *AddressParser) writePort(writer io.Writer, port net.Port) error {
-	if _, err := writer.Write(port.Bytes(nil)); err != nil {
-		return err
-	}
-	return nil
+	_, err := writer.Write(port.Bytes(nil))
+	return err
 }
 
 func (p *AddressParser) writeAddress(writer io.Writer, address net.Address) error {
@@ -172,7 +170,7 @@ func (p *AddressParser) writeAddress(writer io.Writer, address net.Address) erro
 		}
 	case net.AddressFamilyDomain:
 		domain := address.Domain()
-		if IsDomainTooLong(domain) {
+		if isDomainTooLong(domain) {
 			return newError("Super long domain is not supported: ", domain)
 		}
 		if _, err := writer.Write([]byte{tb, byte(len(domain))}); err != nil {

common/protocol/headers.go

@@ -85,6 +85,6 @@ func (sc *SecurityConfig) GetSecurityType() SecurityType {
 	return sc.Type
 }
 
-func IsDomainTooLong(domain string) bool {
+func isDomainTooLong(domain string) bool {
 	return len(domain) > 256
 }

common/signal/done.go

@@ -38,7 +38,7 @@ func (d *Done) Wait() {
 	<-d.c
 }
 
-// Close marks this Done 'done'. This method may be called mutliple times. All calls after first call will have no effect on its status.
+// Close marks this Done 'done'. This method may be called multiple times. All calls after first call will have no effect on its status.
 func (d *Done) Close() error {
 	d.access.Lock()
 	defer d.access.Unlock()

common/signal/exec.go

@@ -12,7 +12,7 @@ func executeAndFulfill(f func() error, done chan<- error) {
 	close(done)
 }
 
-// ExecuteAsync executes a function asychrously and return its result.
+// ExecuteAsync executes a function asynchronously and return its result.
 func ExecuteAsync(f func() error) <-chan error {
 	done := make(chan error, 1)
 	go executeAndFulfill(f, done)

common/signal/notifier.go

@@ -1,6 +1,6 @@
 package signal
 
-// Notifier is an utility for notifying changes. The change producer may notify changes multiple time, and the consumer may get notified asychronously.
+// Notifier is an utility for notifying changes. The change producer may notify changes multiple time, and the consumer may get notified asynchronously.
 type Notifier struct {
 	c chan struct{}
 }

config.proto

@@ -19,7 +19,7 @@ message Config {
 
   reserved 3;
 
-  // App configuration. Must be one in the app directory.
+  // App is for configurations of all features in V2Ray. A feature must implement the Feature interface, and its config type must be registered through common.RegisterConfig.
   repeated v2ray.core.common.serial.TypedMessage app = 4;
 
   // Transport settings.
@@ -30,19 +30,21 @@ message Config {
   repeated v2ray.core.common.serial.TypedMessage extension = 6;
 }
 
+// InboundHandlerConfig is the configuration for inbound handler.
 message InboundHandlerConfig {
-  // Tag of the inbound handler.
+  // Tag of the inbound handler. The tag must be unique among all inbound handlers
   string tag = 1;
-  // Settings for how this inbound proxy is handled. Must be ReceiverConfig above.
+  // Settings for how this inbound proxy is handled.
   v2ray.core.common.serial.TypedMessage receiver_settings = 2;
   // Settings for inbound proxy. Must be one of the inbound proxies.
   v2ray.core.common.serial.TypedMessage proxy_settings = 3;
 }
 
+// OutboundHandlerConfig is the configuration for outbound handler.
 message OutboundHandlerConfig {
   // Tag of this outbound handler.
   string tag = 1;
-  // Settings for how to dial connection for this outbound handler. Must be SenderConfig above.
+  // Settings for how to dial connection for this outbound handler.
   v2ray.core.common.serial.TypedMessage sender_settings = 2;
   // Settings for this outbound proxy. Must be one of the outbound proxies.
   v2ray.core.common.serial.TypedMessage proxy_settings = 3;

core.go

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

proxy/blackhole/config.go

@@ -27,7 +27,7 @@ func (*NoneResponse) WriteTo(buf.Writer) {}
 
 // WriteTo implements ResponseConfig.WriteTo().
 func (*HTTPResponse) WriteTo(writer buf.Writer) {
-	b := buf.NewLocal(512)
+	b := buf.New()
 	common.Must(b.AppendSupplier(serial.WriteString(http403response)))
 	writer.WriteMultiBuffer(buf.NewMultiBufferValue(b))
 }

proxy/shadowsocks/ota.go

@@ -70,22 +70,14 @@ func NewChunkReader(reader io.Reader, auth *Authenticator) *ChunkReader {
 }
 
 func (v *ChunkReader) ReadMultiBuffer() (buf.MultiBuffer, error) {
-	buffer := buf.New()
-	if err := buffer.AppendSupplier(buf.ReadFullFrom(v.reader, 2)); err != nil {
-		buffer.Release()
-		return nil, err
-	}
-	// There is a potential buffer overflow here. Large buffer is 64K bytes,
-	// while uin16 + 10 will be more than that
-	length := serial.BytesToUint16(buffer.BytesTo(2)) + AuthSize
-	if length > buf.Size {
-		// Theoretically the size of a chunk is 64K, but most Shadowsocks implementations used <4K buffer.
-		buffer.Release()
-		buffer = buf.NewLocal(int(length) + 128)
+	size, err := serial.ReadUint16(v.reader)
+	if err != nil {
+		return nil, newError("failed to read size")
 	}
+	size += AuthSize
 
-	buffer.Clear()
-	if err := buffer.AppendSupplier(buf.ReadFullFrom(v.reader, int(length))); err != nil {
+	buffer := buf.NewSize(uint32(size))
+	if err := buffer.AppendSupplier(buf.ReadFullFrom(v.reader, int(size))); err != nil {
 		buffer.Release()
 		return nil, err
 	}

proxy/shadowsocks/ota_test.go

@@ -24,11 +24,11 @@ func TestNormalChunkReading(t *testing.T) {
 func TestNormalChunkWriting(t *testing.T) {
 	assert := With(t)
 
-	buffer := buf.NewLocal(512)
+	buffer := buf.NewSize(512)
 	writer := NewChunkWriter(buffer, NewAuthenticator(ChunkKeyGenerator(
 		[]byte{21, 22, 23, 24, 25, 26, 27, 28, 29, 30, 31, 32, 33, 34, 35, 36})))
 
-	b := buf.NewLocal(256)
+	b := buf.NewSize(256)
 	b.Append([]byte{11, 12, 13, 14, 15, 16, 17, 18})
 	err := writer.WriteMultiBuffer(buf.NewMultiBufferValue(b))
 	assert(err, IsNil)

proxy/shadowsocks/protocol.go

@@ -35,7 +35,7 @@ func ReadTCPSession(user *protocol.User, reader io.Reader) (*protocol.RequestHea
 	}
 	account := rawAccount.(*MemoryAccount)
 
-	buffer := buf.NewLocal(512)
+	buffer := buf.New()
 	defer buffer.Release()
 
 	ivLen := account.Cipher.IVSize()
@@ -149,7 +149,7 @@ func WriteTCPRequest(request *protocol.RequestHeader, writer io.Writer) (buf.Wri
 		return nil, newError("failed to create encoding stream").Base(err).AtError()
 	}
 
-	header := buf.NewLocal(512)
+	header := buf.New()
 
 	if err := addrParser.WriteAddressPort(header, request.Address, request.Port); err != nil {
 		return nil, newError("failed to write address").Base(err)

proxy/shadowsocks/protocol_test.go

@@ -29,7 +29,7 @@ func TestUDPEncoding(t *testing.T) {
 		},
 	}
 
-	data := buf.NewLocal(256)
+	data := buf.NewSize(256)
 	data.AppendSupplier(serial.WriteString("test string"))
 	encodedData, err := EncodeUDPPacket(request, data.Bytes())
 	assert(err, IsNil)
@@ -39,6 +39,7 @@ func TestUDPEncoding(t *testing.T) {
 	assert(decodedData.Bytes(), Equals, data.Bytes())
 	assert(decodedRequest.Address, Equals, request.Address)
 	assert(decodedRequest.Port, Equals, request.Port)
+	assert(decodedRequest.Command, Equals, request.Command)
 }
 
 func TestTCPRequest(t *testing.T) {
@@ -118,6 +119,7 @@ func TestTCPRequest(t *testing.T) {
 		assert(err, IsNil)
 		assert(decodedRequest.Address, Equals, request.Address)
 		assert(decodedRequest.Port, Equals, request.Port)
+		assert(decodedRequest.Command, Equals, request.Command)
 
 		decodedData, err := reader.ReadMultiBuffer()
 		assert(err, IsNil)

proxy/socks/protocol.go

@@ -46,7 +46,9 @@ type ServerSession struct {
 }
 
 func (s *ServerSession) Handshake(reader io.Reader, writer io.Writer) (*protocol.RequestHeader, error) {
-	buffer := buf.NewLocal(512)
+	buffer := buf.New()
+	defer buffer.Release()
+
 	request := new(protocol.RequestHeader)
 
 	if err := buffer.AppendSupplier(buf.ReadFullFrom(reader, 2)); err != nil {
@@ -177,7 +179,7 @@ func (s *ServerSession) Handshake(reader io.Reader, writer io.Writer) (*protocol
 }
 
 func readUsernamePassword(reader io.Reader) (string, string, error) {
-	buffer := buf.NewLocal(512)
+	buffer := buf.New()
 	defer buffer.Release()
 
 	if err := buffer.Reset(buf.ReadFullFrom(reader, 2)); err != nil {
@@ -234,7 +236,9 @@ func writeSocks5AuthenticationResponse(writer io.Writer, version byte, auth byte
 }
 
 func writeSocks5Response(writer io.Writer, errCode byte, address net.Address, port net.Port) error {
-	buffer := buf.NewLocal(64)
+	buffer := buf.New()
+	defer buffer.Release()
+
 	buffer.AppendBytes(socks5Version, errCode, 0x00 /* reserved */)
 	if err := addrParser.WriteAddressPort(buffer, address, port); err != nil {
 		return err
@@ -245,7 +249,9 @@ func writeSocks5Response(writer io.Writer, errCode byte, address net.Address, po
 }
 
 func writeSocks4Response(writer io.Writer, errCode byte, address net.Address, port net.Port) error {
-	buffer := buf.NewLocal(32)
+	buffer := buf.New()
+	defer buffer.Release()
+
 	buffer.AppendBytes(0x00, errCode)
 	common.Must(buffer.AppendSupplier(serial.WriteUint16(port.Value())))
 	buffer.Append(address.IP())
@@ -282,6 +288,7 @@ func EncodeUDPPacket(request *protocol.RequestHeader, data []byte) (*buf.Buffer,
 	b := buf.New()
 	b.AppendBytes(0, 0, 0 /* Fragment */)
 	if err := addrParser.WriteAddressPort(b, request.Address, request.Port); err != nil {
+		b.Release()
 		return nil, err
 	}
 	b.Append(data)
@@ -338,7 +345,9 @@ func ClientHandshake(request *protocol.RequestHeader, reader io.Reader, writer i
 		authByte = byte(authPassword)
 	}
 
-	b := buf.NewLocal(512)
+	b := buf.New()
+	defer b.Release()
+
 	b.AppendBytes(socks5Version, 0x01, authByte)
 	if authByte == authPassword {
 		rawAccount, err := request.User.GetTypedAccount()

proxy/vmess/encoding/commands.go

@@ -32,7 +32,7 @@ func MarshalCommand(command interface{}, writer io.Writer) error {
 		return ErrUnknownCommand
 	}
 
-	buffer := buf.NewLocal(512)
+	buffer := buf.New()
 	defer buffer.Release()
 
 	err := factory.Marshal(command, buffer)

proxy/vmess/encoding/server.go

@@ -102,7 +102,12 @@ func NewServerSession(validator protocol.UserValidator, sessionHistory *SessionH
 
 func parseSecurityType(b byte) protocol.SecurityType {
 	if _, f := protocol.SecurityType_name[int32(b)]; f {
-		return protocol.SecurityType(b)
+		st := protocol.SecurityType(b)
+		// For backward compatibility.
+		if st == protocol.SecurityType_UNKNOWN {
+			st = protocol.SecurityType_LEGACY
+		}
+		return st
 	}
 	return protocol.SecurityType_UNKNOWN
 }

release/install-release.sh

@@ -206,7 +206,7 @@ startV2ray(){
 copyFile() {
     NAME=$1
     MANDATE=$2
-    ERROR=`cp "/tmp/v2ray/v2ray-${NEW_VER}-linux-${VDIS}/${NAME}" "/usr/bin/v2ray/${NAME}"`
+    ERROR=`cp "/tmp/v2ray/v2ray-${NEW_VER}-linux-${VDIS}/${NAME}" "/usr/bin/v2ray/${NAME}" 2>&1`
     if [[ $? -ne 0 ]]; then
         colorEcho ${YELLOW} "${ERROR}"
         if [ "$MANDATE" = true ]; then
@@ -251,7 +251,7 @@ installV2Ray(){
 }
 
 
-installInitScrip(){
+installInitScript(){
     SYSTEMCTL_CMD=$(command -v systemctl)
     SERVICE_CMD=$(command -v service)
 
@@ -371,11 +371,11 @@ main(){
             NEW_VER=`ls /tmp/v2ray |grep v2ray-v |cut -d "-" -f2`
         fi
     else
-        # dowload via network and extract
+        # download via network and extract
         installSoftware "curl"
         getVersion
         if [[ $? == 0 ]] && [[ "$FORCE" != "1" ]]; then
-            colorEcho ${GREEN} "Lastest version ${NEW_VER} is already installed."
+            colorEcho ${GREEN} "Latest version ${NEW_VER} is already installed."
             exit
         else
             colorEcho ${BLUE} "Installing V2Ray ${NEW_VER} on ${ARCH}"
@@ -389,7 +389,7 @@ main(){
         stopV2ray
     fi
     installV2Ray
-    installInitScrip
+    installInitScript
     if [[ ${V2RAY_RUNNING} -eq 1 ]];then
         colorEcho ${BLUE} "Restarting V2Ray service."
         startV2ray

testing/scenarios/command_test.go

@@ -116,7 +116,7 @@ func TestCommanderRemoveHandler(t *testing.T) {
 		assert(conn.Close(), IsNil)
 	}
 
-	cmdConn, err := grpc.Dial(fmt.Sprintf("127.0.0.1:%d", cmdPort), grpc.WithInsecure())
+	cmdConn, err := grpc.Dial(fmt.Sprintf("127.0.0.1:%d", cmdPort), grpc.WithInsecure(), grpc.WithBlock())
 	assert(err, IsNil)
 
 	hsClient := command.NewHandlerServiceClient(cmdConn)
@@ -296,7 +296,7 @@ func TestCommanderAddRemoveUser(t *testing.T) {
 		assert(conn.Close(), IsNil)
 	}
 
-	cmdConn, err := grpc.Dial(fmt.Sprintf("127.0.0.1:%d", cmdPort), grpc.WithInsecure())
+	cmdConn, err := grpc.Dial(fmt.Sprintf("127.0.0.1:%d", cmdPort), grpc.WithInsecure(), grpc.WithBlock())
 	assert(err, IsNil)
 
 	hsClient := command.NewHandlerServiceClient(cmdConn)

testing/scenarios/shadowsocks_test.go

@@ -615,6 +615,130 @@ func TestShadowsocksAES128GCMUDP(t *testing.T) {
 	CloseAllServers(servers)
 }
 
+func TestShadowsocksAES128GCMUDPMux(t *testing.T) {
+	assert := With(t)
+
+	udpServer := udp.Server{
+		MsgProcessor: xor,
+	}
+	dest, err := udpServer.Start()
+	assert(err, IsNil)
+	defer udpServer.Close()
+
+	account := serial.ToTypedMessage(&shadowsocks.Account{
+		Password:   "shadowsocks-password",
+		CipherType: shadowsocks.CipherType_AES_128_GCM,
+	})
+
+	serverPort := tcp.PickPort()
+	serverConfig := &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(serverPort),
+					Listen:    net.NewIPOrDomain(net.LocalHostIP),
+				}),
+				ProxySettings: serial.ToTypedMessage(&shadowsocks.ServerConfig{
+					UdpEnabled: false,
+					User: &protocol.User{
+						Account: account,
+						Level:   1,
+					},
+				}),
+			},
+		},
+		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_UDP},
+					},
+				}),
+			},
+		},
+		Outbound: []*core.OutboundHandlerConfig{
+			{
+				SenderSettings: serial.ToTypedMessage(&proxyman.SenderConfig{
+					MultiplexSettings: &proxyman.MultiplexingConfig{
+						Enabled:     true,
+						Concurrency: 8,
+					},
+				}),
+				ProxySettings: serial.ToTypedMessage(&shadowsocks.ClientConfig{
+					Server: []*protocol.ServerEndpoint{
+						{
+							Address: net.NewIPOrDomain(net.LocalHostIP),
+							Port:    uint32(serverPort),
+							User: []*protocol.User{
+								{
+									Account: account,
+								},
+							},
+						},
+					},
+				}),
+			},
+		},
+	}
+
+	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.DialUDP("udp", nil, &net.UDPAddr{
+				IP:   []byte{127, 0, 0, 1},
+				Port: int(clientPort),
+			})
+			assert(err, IsNil)
+
+			payload := make([]byte, 1024)
+			rand.Read(payload)
+
+			nBytes, err := conn.Write([]byte(payload))
+			assert(err, IsNil)
+			assert(nBytes, Equals, len(payload))
+
+			response := readFrom(conn, time.Second*5, 1024)
+			assert(response, Equals, xor([]byte(payload)))
+			assert(conn.Close(), IsNil)
+			wg.Done()
+		}()
+	}
+	wg.Wait()
+
+	CloseAllServers(servers)
+}
+
 func TestShadowsocksAES256GCMConformance(t *testing.T) {
 	assert := With(t)
 

transport/internet/config.go

@@ -10,7 +10,9 @@ var (
 )
 
 func RegisterProtocolConfigCreator(protocol TransportProtocol, creator ConfigCreator) error {
-	// TODO: check duplicate
+	if _, found := globalTransportConfigCreatorCache[protocol]; found {
+		return newError("protocol ", TransportProtocol_name[int32(protocol)], " is already registered").AtError()
+	}
 	globalTransportConfigCreatorCache[protocol] = creator
 	return nil
 }

transport/internet/headers/http/http_test.go

@@ -16,7 +16,7 @@ func TestReaderWriter(t *testing.T) {
 	assert := With(t)
 
 	cache := buf.New()
-	b := buf.NewLocal(256)
+	b := buf.NewSize(256)
 	b.AppendSupplier(serial.WriteString("abcd" + ENDING))
 	writer := NewHeaderWriter(b)
 	err := writer.Write(cache)

transport/internet/headers/srtp/srtp_test.go

@@ -18,7 +18,7 @@ func TestSRTPWrite(t *testing.T) {
 
 	srtp := srtpRaw.(*SRTP)
 
-	payload := buf.NewLocal(2048)
+	payload := buf.New()
 	payload.AppendSupplier(srtp.Write)
 	payload.Append(content)
 

transport/internet/headers/utp/utp_test.go

@@ -18,7 +18,7 @@ func TestUTPWrite(t *testing.T) {
 
 	utp := utpRaw.(*UTP)
 
-	payload := buf.NewLocal(2048)
+	payload := buf.New()
 	payload.AppendSupplier(utp.Write)
 	payload.Append(content)
 

transport/internet/headers/wechat/wechat_test.go

@@ -17,7 +17,7 @@ func TestUTPWrite(t *testing.T) {
 
 	video := videoRaw.(*VideoChat)
 
-	payload := buf.NewLocal(2048)
+	payload := buf.New()
 	payload.AppendSupplier(video.Write)
 
 	assert(payload.Len(), Equals, video.Size())

transport/internet/http/errors.generated.go

@@ -2,4 +2,6 @@ package http
 
 import "v2ray.com/core/common/errors"
 
-func newError(values ...interface{}) *errors.Error { return errors.New(values...).Path("Transport", "Internet", "HTTP") }
+func newError(values ...interface{}) *errors.Error {
+	return errors.New(values...).Path("Transport", "Internet", "HTTP")
+}

transport/internet/websocket/config.go

@@ -7,7 +7,7 @@ import (
 	"v2ray.com/core/transport/internet"
 )
 
-func (c *Config) GetNormailzedPath() string {
+func (c *Config) GetNormalizedPath() string {
 	path := c.Path
 	if len(path) == 0 {
 		return "/"

transport/internet/websocket/dialer.go

@@ -50,7 +50,7 @@ func dialWebsocket(ctx context.Context, dest net.Destination) (net.Conn, error)
 	if (protocol == "ws" && dest.Port == 80) || (protocol == "wss" && dest.Port == 443) {
 		host = dest.Address.String()
 	}
-	uri := protocol + "://" + host + wsSettings.GetNormailzedPath()
+	uri := protocol + "://" + host + wsSettings.GetNormalizedPath()
 
 	conn, resp, err := dialer.Dial(uri, wsSettings.GetRequestHeader())
 	if err != nil {

transport/internet/websocket/hub.go

@@ -92,7 +92,7 @@ func (ln *Listener) listenws(address net.Address, port net.Port) error {
 
 	go func() {
 		err := http.Serve(listener, &requestHandler{
-			path: ln.config.GetNormailzedPath(),
+			path: ln.config.GetNormalizedPath(),
 			ln:   ln,
 		})
 		if err != nil {

transport/ray/ray.go

@@ -13,7 +13,7 @@ type OutboundRay interface {
 
 	// OutboundOutput provides a stream to retrieve the response from the
 	// outbound connection. The outbound connection shall close the channel
-	// after all responses are receivced and put into the channel.
+	// after all responses are received and put into the channel.
 	OutboundOutput() OutputStream
 }
 
@@ -24,13 +24,13 @@ type InboundRay interface {
 	// is received and put into the channel.
 	InboundInput() OutputStream
 
-	// InboudBound provides a stream of data for the inbound connection to write
+	// InboundOutput provides a stream of data for the inbound connection to write
 	// as response. The inbound connection shall write all the data from the
 	// channel until it is closed.
 	InboundOutput() InputStream
 }
 
-// Ray is an internal tranport channel between inbound and outbound connection.
+// Ray is an internal transport channel between inbound and outbound connection.
 type Ray interface {
 	InboundRay
 	OutboundRay

v2ray.go

@@ -130,7 +130,7 @@ func (s *Instance) Start() error {
 }
 
 // RegisterFeature registers the given feature into V2Ray.
-// If feature is one of the following types, the corressponding feature in this Instance
+// If feature is one of the following types, the corresponding feature in this Instance
 // will be replaced: DNSClient, PolicyManager, Router, Dispatcher, InboundHandlerManager, OutboundHandlerManager.
 func (s *Instance) RegisterFeature(feature interface{}, instance Feature) error {
 	running := false