Merge from remote branch

adapter/adapter.go

@@ -136,6 +136,8 @@ func (p *Proxy) URLTest(ctx context.Context, url string) (t uint16, err error) {
 			return http.ErrUseLastResponse
 		},
 	}
+	defer client.CloseIdleConnections()
+
 	resp, err := client.Do(req)
 	if err != nil {
 		return

adapter/outbound/direct.go

@@ -14,9 +14,7 @@ type Direct struct {
 
 // DialContext implements C.ProxyAdapter
 func (d *Direct) DialContext(ctx context.Context, metadata *C.Metadata) (C.Conn, error) {
-	address := net.JoinHostPort(metadata.String(), metadata.DstPort)
+	c, err := dialer.DialContext(ctx, "tcp", metadata.RemoteAddress())
-
-	c, err := dialer.DialContext(ctx, "tcp", address)
 	if err != nil {
 		return nil, err
 	}

adapter/provider/healthcheck.go

@@ -2,9 +2,9 @@ package provider
 
 import (
 	"context"
-	"sync"
 	"time"
 
+	"github.com/Dreamacro/clash/common/batch"
 	C "github.com/Dreamacro/clash/constant"
 
 	"go.uber.org/atomic"
@@ -59,20 +59,17 @@ func (hc *HealthCheck) touch() {
 }
 
 func (hc *HealthCheck) check() {
-	ctx, cancel := context.WithTimeout(context.Background(), defaultURLTestTimeout)
+	b, _ := batch.New(context.Background(), batch.WithConcurrencyNum(10))
-	wg := &sync.WaitGroup{}
-
 	for _, proxy := range hc.proxies {
-		wg.Add(1)
+		p := proxy
-
+		b.Go(p.Name(), func() (interface{}, error) {
-		go func(p C.Proxy) {
+			ctx, cancel := context.WithTimeout(context.Background(), defaultURLTestTimeout)
+			defer cancel()
 			p.URLTest(ctx, hc.url)
-			wg.Done()
+			return nil, nil
-		}(proxy)
+		})
 	}
-
+	b.Wait()
-	wg.Wait()
-	cancel()
 }
 
 func (hc *HealthCheck) close() {

common/batch/batch.go

@@ -0,0 +1,105 @@
+package batch
+
+import (
+	"context"
+	"sync"
+)
+
+type Option = func(b *Batch)
+
+type Result struct {
+	Value interface{}
+	Err   error
+}
+
+type Error struct {
+	Key string
+	Err error
+}
+
+func WithConcurrencyNum(n int) Option {
+	return func(b *Batch) {
+		q := make(chan struct{}, n)
+		for i := 0; i < n; i++ {
+			q <- struct{}{}
+		}
+		b.queue = q
+	}
+}
+
+// Batch similar to errgroup, but can control the maximum number of concurrent
+type Batch struct {
+	result map[string]Result
+	queue  chan struct{}
+	wg     sync.WaitGroup
+	mux    sync.Mutex
+	err    *Error
+	once   sync.Once
+	cancel func()
+}
+
+func (b *Batch) Go(key string, fn func() (interface{}, error)) {
+	b.wg.Add(1)
+	go func() {
+		defer b.wg.Done()
+		if b.queue != nil {
+			<-b.queue
+			defer func() {
+				b.queue <- struct{}{}
+			}()
+		}
+
+		value, err := fn()
+		if err != nil {
+			b.once.Do(func() {
+				b.err = &Error{key, err}
+				if b.cancel != nil {
+					b.cancel()
+				}
+			})
+		}
+
+		ret := Result{value, err}
+		b.mux.Lock()
+		defer b.mux.Unlock()
+		b.result[key] = ret
+	}()
+}
+
+func (b *Batch) Wait() *Error {
+	b.wg.Wait()
+	if b.cancel != nil {
+		b.cancel()
+	}
+	return b.err
+}
+
+func (b *Batch) WaitAndGetResult() (map[string]Result, *Error) {
+	err := b.Wait()
+	return b.Result(), err
+}
+
+func (b *Batch) Result() map[string]Result {
+	b.mux.Lock()
+	defer b.mux.Unlock()
+	copy := map[string]Result{}
+	for k, v := range b.result {
+		copy[k] = v
+	}
+	return copy
+}
+
+func New(ctx context.Context, opts ...Option) (*Batch, context.Context) {
+	ctx, cancel := context.WithCancel(ctx)
+
+	b := &Batch{
+		result: map[string]Result{},
+	}
+
+	for _, o := range opts {
+		o(b)
+	}
+
+	b.cancel = cancel
+	return b, ctx
+}

common/batch/batch_test.go

@@ -0,0 +1,83 @@
+package batch
+
+import (
+	"context"
+	"errors"
+	"strconv"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/assert"
+)
+
+func TestBatch(t *testing.T) {
+	b, _ := New(context.Background())
+
+	now := time.Now()
+	b.Go("foo", func() (interface{}, error) {
+		time.Sleep(time.Millisecond * 100)
+		return "foo", nil
+	})
+	b.Go("bar", func() (interface{}, error) {
+		time.Sleep(time.Millisecond * 150)
+		return "bar", nil
+	})
+	result, err := b.WaitAndGetResult()
+
+	assert.Nil(t, err)
+
+	duration := time.Since(now)
+	assert.Less(t, duration, time.Millisecond*200)
+	assert.Equal(t, 2, len(result))
+
+	for k, v := range result {
+		assert.NoError(t, v.Err)
+		assert.Equal(t, k, v.Value.(string))
+	}
+}
+
+func TestBatchWithConcurrencyNum(t *testing.T) {
+	b, _ := New(
+		context.Background(),
+		WithConcurrencyNum(3),
+	)
+
+	now := time.Now()
+	for i := 0; i < 7; i++ {
+		idx := i
+		b.Go(strconv.Itoa(idx), func() (interface{}, error) {
+			time.Sleep(time.Millisecond * 100)
+			return strconv.Itoa(idx), nil
+		})
+	}
+	result, _ := b.WaitAndGetResult()
+	duration := time.Since(now)
+	assert.Greater(t, duration, time.Millisecond*260)
+	assert.Equal(t, 7, len(result))
+
+	for k, v := range result {
+		assert.NoError(t, v.Err)
+		assert.Equal(t, k, v.Value.(string))
+	}
+}
+
+func TestBatchContext(t *testing.T) {
+	b, ctx := New(context.Background())
+
+	b.Go("error", func() (interface{}, error) {
+		time.Sleep(time.Millisecond * 100)
+		return nil, errors.New("test error")
+	})
+
+	b.Go("ctx", func() (interface{}, error) {
+		<-ctx.Done()
+		return nil, ctx.Err()
+	})
+
+	result, err := b.WaitAndGetResult()
+
+	assert.NotNil(t, err)
+	assert.Equal(t, "error", err.Key)
+
+	assert.Equal(t, ctx.Err(), result["ctx"].Err)
+}

common/pool/alloc.go

@@ -8,11 +8,7 @@ import (
 	"sync"
 )
 
-var defaultAllocator *Allocator
+var defaultAllocator = NewAllocator()
-
-func init() {
-	defaultAllocator = NewAllocator()
-}
 
 // Allocator for incoming frames, optimized to prevent overwriting after zeroing
 type Allocator struct {

component/process/process_linux.go

@@ -16,14 +16,14 @@ import (
 )
 
 // from https://github.com/vishvananda/netlink/blob/bca67dfc8220b44ef582c9da4e9172bf1c9ec973/nl/nl_linux.go#L52-L62
-func init() {
+var nativeEndian = func() binary.ByteOrder {
 	var x uint32 = 0x01020304
 	if *(*byte)(unsafe.Pointer(&x)) == 0x01 {
-		nativeEndian = binary.BigEndian
+		return binary.BigEndian
-	} else {
-		nativeEndian = binary.LittleEndian
 	}
-}
+
+	return binary.LittleEndian
+}()
 
 type SocketResolver func(network string, ip net.IP, srcPort int) (inode, uid int, err error)
 type ProcessNameResolver func(inode, uid int) (name string, err error)
@@ -40,8 +40,6 @@ const (
 	pathProc                = "/proc"
 )
 
-var nativeEndian binary.ByteOrder = binary.LittleEndian
-
 func findProcessName(network string, ip net.IP, srcPort int) (string, error) {
 	inode, uid, err := DefaultSocketResolver(network, ip, srcPort)
 	if err != nil {

constant/metadata.go

@@ -18,7 +18,8 @@ const (
 
 	HTTP Type = iota
 	HTTPCONNECT
-	SOCKS
+	SOCKS4
+	SOCKS5
 	REDIR
 	TPROXY
 	TUN
@@ -47,7 +48,9 @@ func (t Type) String() string {
 		return "HTTP"
 	case HTTPCONNECT:
 		return "HTTP Connect"
-	case SOCKS:
+	case SOCKS4:
+		return "Socks4"
+	case SOCKS5:
 		return "Socks5"
 	case REDIR:
 		return "Redir"

constant/path.go

@@ -9,21 +9,19 @@ import (
 const Name = "clash"
 
 // Path is used to get the configuration path
-var Path *path
+var Path = func() *path {
-
-type path struct {
-	homeDir    string
-	configFile string
-}
-
-func init() {
 	homeDir, err := os.UserHomeDir()
 	if err != nil {
 		homeDir, _ = os.Getwd()
 	}
 
 	homeDir = P.Join(homeDir, ".config", Name)
-	Path = &path{homeDir: homeDir, configFile: "config.yaml"}
+	return &path{homeDir: homeDir, configFile: "config.yaml"}
+}()
+
+type path struct {
+	homeDir    string
+	configFile string
 }
 
 // SetHomeDir is used to set the configuration path

hub/executor/executor.go

@@ -76,16 +76,14 @@ func ApplyConfig(cfg *config.Config, force bool) {
 	defer mux.Unlock()
 
 	updateUsers(cfg.Users)
-	updateDNS(cfg.DNS, cfg.General)
-	updateGeneral(cfg.General, force)
-	log.SetLevel(log.DEBUG)
 	updateProxies(cfg.Proxies, cfg.Providers)
 	updateRules(cfg.Rules)
 	updateHosts(cfg.Hosts)
-	updateExperimental(cfg)
 	updateProfile(cfg)
 	updateIPTables(cfg.DNS, cfg.General)
-	log.SetLevel(cfg.General.LogLevel)
+	updateDNS(cfg.DNS, cfg.General)
+	updateGeneral(cfg.General, force)
+	updateExperimental(cfg)
 }
 
 func GetGeneral() *config.General {
@@ -178,7 +176,6 @@ func updateRules(rules []C.Rule) {
 }
 
 func updateGeneral(general *config.General, force bool) {
-	log.SetLevel(log.DEBUG)
 	tunnel.SetMode(general.Mode)
 	resolver.DisableIPv6 = !general.IPv6
 
@@ -223,7 +220,7 @@ func updateGeneral(general *config.General, force bool) {
 	}
 
 	if err := P.ReCreateSocks(general.SocksPort, tcpIn, udpIn); err != nil {
-		log.Errorln("Start SOCKS5 server error: %s", err.Error())
+		log.Errorln("Start SOCKS server error: %s", err.Error())
 	}
 
 	if err := P.ReCreateRedir(general.RedirPort, tcpIn, udpIn); err != nil {
@@ -235,7 +232,7 @@ func updateGeneral(general *config.General, force bool) {
 	}
 
 	if err := P.ReCreateMixed(general.MixedPort, tcpIn, udpIn); err != nil {
-		log.Errorln("Start Mixed(http and socks5) server error: %s", err.Error())
+		log.Errorln("Start Mixed(http and socks) server error: %s", err.Error())
 	}
 
 	if err := P.ReCreateTun(general.Tun, tcpIn, udpIn); err != nil {

hub/route/server.go

@@ -3,6 +3,7 @@ package route
 import (
 	"bytes"
 	"encoding/json"
+	"net"
 	"net/http"
 	"strings"
 	"time"
@@ -81,10 +82,15 @@ func Start(addr string, secret string) {
 		})
 	}
 
-	log.Infoln("RESTful API listening at: %s", addr)
+	l, err := net.Listen("tcp", addr)
-	err := http.ListenAndServe(addr, r)
 	if err != nil {
-		log.Errorln("External controller error: %s", err.Error())
+		log.Errorln("External controller listen error: %s", err)
+		return
+	}
+	serverAddr = l.Addr().String()
+	log.Infoln("RESTful API listening at: %s", serverAddr)
+	if err = http.Serve(l, r); err != nil {
+		log.Errorln("External controller serve error: %s", err)
 	}
 }
 

listener/http/proxy.go

@@ -45,6 +45,7 @@ func HandleConn(c net.Conn, in chan<- C.ConnContext, cache *cache.Cache) {
 			if request.Method == http.MethodConnect {
 				resp = responseWith(200)
 				resp.Status = "Connection established"
+				resp.ContentLength = -1
 
 				if resp.Write(conn) != nil {
 					break // close connection

listener/http/server.go

@@ -10,7 +10,6 @@ import (
 
 type Listener struct {
 	listener net.Listener
-	address  string
 	closed   bool
 }
 
@@ -31,7 +30,6 @@ func NewWithAuthenticate(addr string, in chan<- C.ConnContext, authenticate bool
 
 	hl := &Listener{
 		listener: l,
-		address:  addr,
 	}
 	go func() {
 		for {
@@ -55,5 +53,5 @@ func (l *Listener) Close() {
 }
 
 func (l *Listener) Address() string {
-	return l.address
+	return l.listener.Addr().String()
 }

listener/listener.go

@@ -157,7 +157,7 @@ func ReCreateSocks(port int, tcpIn chan<- C.ConnContext, udpIn chan<- *inbound.P
 	socksListener = tcpListener
 	socksUDPListener = udpListener
 
-	log.Infoln("SOCKS5 proxy listening at: %s", socksListener.Address())
+	log.Infoln("SOCKS proxy listening at: %s", socksListener.Address())
 	return nil
 }
 
@@ -289,7 +289,7 @@ func ReCreateMixed(port int, tcpIn chan<- C.ConnContext, udpIn chan<- *inbound.P
 		return err
 	}
 
-	log.Infoln("Mixed(http+socks5) proxy listening at: %s", mixedListener.Address())
+	log.Infoln("Mixed(http+socks) proxy listening at: %s", mixedListener.Address())
 	return nil
 }
 

listener/mixed/mixed.go

@@ -9,12 +9,12 @@ import (
 	C "github.com/Dreamacro/clash/constant"
 	"github.com/Dreamacro/clash/listener/http"
 	"github.com/Dreamacro/clash/listener/socks"
+	"github.com/Dreamacro/clash/transport/socks4"
 	"github.com/Dreamacro/clash/transport/socks5"
 )
 
 type Listener struct {
 	listener net.Listener
-	address  string
 	closed   bool
 	cache    *cache.Cache
 }
@@ -25,7 +25,10 @@ func New(addr string, in chan<- C.ConnContext) (*Listener, error) {
 		return nil, err
 	}
 
-	ml := &Listener{l, addr, false, cache.New(30 * time.Second)}
+	ml := &Listener{
+		listener: l,
+		cache:    cache.New(30 * time.Second),
+	}
 	go func() {
 		for {
 			c, err := ml.listener.Accept()
@@ -48,7 +51,7 @@ func (l *Listener) Close() {
 }
 
 func (l *Listener) Address() string {
-	return l.address
+	return l.listener.Addr().String()
 }
 
 func handleConn(conn net.Conn, in chan<- C.ConnContext, cache *cache.Cache) {
@@ -58,10 +61,12 @@ func handleConn(conn net.Conn, in chan<- C.ConnContext, cache *cache.Cache) {
 		return
 	}
 
-	if head[0] == socks5.Version {
+	switch head[0] {
-		socks.HandleSocks(bufConn, in)
+	case socks4.Version:
-		return
+		socks.HandleSocks4(bufConn, in)
+	case socks5.Version:
+		socks.HandleSocks5(bufConn, in)
+	default:
+		http.HandleConn(bufConn, in, cache)
 	}
-
-	http.HandleConn(bufConn, in, cache)
 }

listener/redir/tcp.go

@@ -9,7 +9,6 @@ import (
 
 type Listener struct {
 	listener net.Listener
-	address  string
 	closed   bool
 }
 
@@ -18,7 +17,9 @@ func New(addr string, in chan<- C.ConnContext) (*Listener, error) {
 	if err != nil {
 		return nil, err
 	}
-	rl := &Listener{l, addr, false}
+	rl := &Listener{
+		listener: l,
+	}
 
 	go func() {
 		for {
@@ -42,7 +43,7 @@ func (l *Listener) Close() {
 }
 
 func (l *Listener) Address() string {
-	return l.address
+	return l.listener.Addr().String()
 }
 
 func handleRedir(conn net.Conn, in chan<- C.ConnContext) {

listener/socks/tcp.go

@@ -6,14 +6,15 @@ import (
 	"net"
 
 	"github.com/Dreamacro/clash/adapter/inbound"
+	N "github.com/Dreamacro/clash/common/net"
 	C "github.com/Dreamacro/clash/constant"
 	authStore "github.com/Dreamacro/clash/listener/auth"
+	"github.com/Dreamacro/clash/transport/socks4"
 	"github.com/Dreamacro/clash/transport/socks5"
 )
 
 type Listener struct {
 	listener net.Listener
-	address  string
 	closed   bool
 }
 
@@ -23,7 +24,9 @@ func New(addr string, in chan<- C.ConnContext) (*Listener, error) {
 		return nil, err
 	}
 
-	sl := &Listener{l, addr, false}
+	sl := &Listener{
+		listener: l,
+	}
 	go func() {
 		for {
 			c, err := l.Accept()
@@ -33,7 +36,7 @@ func New(addr string, in chan<- C.ConnContext) (*Listener, error) {
 				}
 				continue
 			}
-			go HandleSocks(c, in)
+			go handleSocks(c, in)
 		}
 	}()
 
@@ -46,10 +49,40 @@ func (l *Listener) Close() {
 }
 
 func (l *Listener) Address() string {
-	return l.address
+	return l.listener.Addr().String()
 }
 
-func HandleSocks(conn net.Conn, in chan<- C.ConnContext) {
+func handleSocks(conn net.Conn, in chan<- C.ConnContext) {
+	bufConn := N.NewBufferedConn(conn)
+	head, err := bufConn.Peek(1)
+	if err != nil {
+		conn.Close()
+		return
+	}
+
+	switch head[0] {
+	case socks4.Version:
+		HandleSocks4(bufConn, in)
+	case socks5.Version:
+		HandleSocks5(bufConn, in)
+	default:
+		conn.Close()
+	}
+}
+
+func HandleSocks4(conn net.Conn, in chan<- C.ConnContext) {
+	addr, _, err := socks4.ServerHandshake(conn, authStore.Authenticator())
+	if err != nil {
+		conn.Close()
+		return
+	}
+	if c, ok := conn.(*net.TCPConn); ok {
+		c.SetKeepAlive(true)
+	}
+	in <- inbound.NewSocket(socks5.ParseAddr(addr), conn, C.SOCKS4)
+}
+
+func HandleSocks5(conn net.Conn, in chan<- C.ConnContext) {
 	target, command, err := socks5.ServerHandshake(conn, authStore.Authenticator())
 	if err != nil {
 		conn.Close()
@@ -63,5 +96,5 @@ func HandleSocks(conn net.Conn, in chan<- C.ConnContext) {
 		io.Copy(ioutil.Discard, conn)
 		return
 	}
-	in <- inbound.NewSocket(target, conn, C.SOCKS)
+	in <- inbound.NewSocket(target, conn, C.SOCKS5)
 }

listener/socks/udp.go

@@ -13,7 +13,6 @@ import (
 
 type UDPListener struct {
 	packetConn net.PacketConn
-	address    string
 	closed     bool
 }
 
@@ -27,7 +26,9 @@ func NewUDP(addr string, in chan<- *inbound.PacketAdapter) (*UDPListener, error)
 		log.Warnln("Failed to Reuse UDP Address: %s", err)
 	}
 
-	sl := &UDPListener{l, addr, false}
+	sl := &UDPListener{
+		packetConn: l,
+	}
 	go func() {
 		for {
 			buf := pool.Get(pool.RelayBufferSize)
@@ -52,7 +53,7 @@ func (l *UDPListener) Close() error {
 }
 
 func (l *UDPListener) Address() string {
-	return l.address
+	return l.packetConn.LocalAddr().String()
 }
 
 func handleSocksUDP(pc net.PacketConn, in chan<- *inbound.PacketAdapter, buf []byte, addr net.Addr) {
@@ -69,7 +70,7 @@ func handleSocksUDP(pc net.PacketConn, in chan<- *inbound.PacketAdapter, buf []b
 		bufRef:  buf,
 	}
 	select {
-	case in <- inbound.NewPacket(target, packet, C.TPROXY):
+	case in <- inbound.NewPacket(target, packet, C.SOCKS5):
 	default:
 	}
 }

listener/tproxy/tproxy.go

@@ -10,7 +10,6 @@ import (
 
 type Listener struct {
 	listener net.Listener
-	address  string
 	closed   bool
 }
 
@@ -33,7 +32,6 @@ func New(addr string, in chan<- C.ConnContext) (*Listener, error) {
 
 	rl := &Listener{
 		listener: l,
-		address:  addr,
 	}
 
 	go func() {
@@ -58,7 +56,7 @@ func (l *Listener) Close() {
 }
 
 func (l *Listener) Address() string {
-	return l.address
+	return l.listener.Addr().String()
 }
 
 func (l *Listener) handleTProxy(conn net.Conn, in chan<- C.ConnContext) {

listener/tproxy/udp.go

@@ -11,7 +11,6 @@ import (
 
 type UDPListener struct {
 	packetConn net.PacketConn
-	address    string
 	closed     bool
 }
 
@@ -21,7 +20,9 @@ func NewUDP(addr string, in chan<- *inbound.PacketAdapter) (*UDPListener, error)
 		return nil, err
 	}
 
-	rl := &UDPListener{l, addr, false}
+	rl := &UDPListener{
+		packetConn: l,
+	}
 
 	c := l.(*net.UDPConn)
 
@@ -65,7 +66,7 @@ func (l *UDPListener) Close() error {
 }
 
 func (l *UDPListener) Address() string {
-	return l.address
+	return l.packetConn.LocalAddr().String()
 }
 
 func handlePacketConn(pc net.PacketConn, in chan<- *inbound.PacketAdapter, buf []byte, lAddr *net.UDPAddr, rAddr *net.UDPAddr) {

test/README.md

@@ -47,3 +47,13 @@ Prerequisite
 ```
 $ go test -p 1 -v
 ```
+
+benchmark (Linux)
+
+> Cannot represent the throughput of the protocol on your machine
+> but you can compare the corresponding throughput of the protocol on clash
+> (change chunkSize to measure the maximum throughput of clash on your machine)
+
+```
+$ go test -benchmem -run=^$ -bench .
+```

test/clash_test.go

@@ -15,6 +15,7 @@ import (
 	"testing"
 	"time"
 
+	"github.com/Dreamacro/clash/adapter/outbound"
 	C "github.com/Dreamacro/clash/constant"
 	"github.com/Dreamacro/clash/hub/executor"
 	"github.com/Dreamacro/clash/transport/socks5"
@@ -70,7 +71,7 @@ func init() {
 		}
 	}
 
-	c, err := client.NewClientWithOpts(client.FromEnv)
+	c, err := client.NewClientWithOpts(client.FromEnv, client.WithAPIVersionNegotiation())
 	if err != nil {
 		panic(err)
 	}
@@ -619,6 +620,42 @@ func testSuit(t *testing.T, proxy C.ProxyAdapter) {
 	assert.NoError(t, testPacketConnTimeout(t, pc))
 }
 
+func benchmarkProxy(b *testing.B, proxy C.ProxyAdapter) {
+	l, err := net.Listen("tcp", ":10001")
+	if err != nil {
+		assert.FailNow(b, err.Error())
+	}
+
+	go func() {
+		c, err := l.Accept()
+		if err != nil {
+			assert.FailNow(b, err.Error())
+		}
+
+		io.Copy(io.Discard, c)
+		c.Close()
+	}()
+
+	chunkSize := int64(16 * 1024)
+	chunk := make([]byte, chunkSize)
+	conn, err := proxy.DialContext(context.Background(), &C.Metadata{
+		Host:     localIP.String(),
+		DstPort:  "10001",
+		AddrType: socks5.AtypDomainName,
+	})
+	if err != nil {
+		assert.FailNow(b, err.Error())
+	}
+
+	b.SetBytes(chunkSize)
+	b.ResetTimer()
+	for i := 0; i < b.N; i++ {
+		if _, err := conn.Write(chunk); err != nil {
+			assert.FailNow(b, err.Error())
+		}
+	}
+}
+
 func TestClash_Basic(t *testing.T) {
 	basic := `
 mixed-port: 10000
@@ -633,3 +670,8 @@ log-level: silent
 	time.Sleep(waitTime)
 	testPingPongWithSocksPort(t, 10000)
 }
+
+func Benchmark_Direct(b *testing.B) {
+	proxy := outbound.NewDirect()
+	benchmarkProxy(b, proxy)
+}

test/docker_test.go

@@ -11,7 +11,7 @@ import (
 var isDarwin = false
 
 func startContainer(cfg *container.Config, hostCfg *container.HostConfig, name string) (string, error) {
-	c, err := client.NewClientWithOpts(client.FromEnv)
+	c, err := client.NewClientWithOpts(client.FromEnv, client.WithAPIVersionNegotiation())
 	if err != nil {
 		return "", err
 	}
@@ -34,7 +34,7 @@ func startContainer(cfg *container.Config, hostCfg *container.HostConfig, name s
 }
 
 func cleanContainer(id string) error {
-	c, err := client.NewClientWithOpts(client.FromEnv)
+	c, err := client.NewClientWithOpts(client.FromEnv, client.WithAPIVersionNegotiation())
 	if err != nil {
 		return err
 	}

test/go.mod

@@ -3,24 +3,17 @@ module clash-test
 go 1.16
 
 require (
-	github.com/Dreamacro/clash v1.6.1-0.20210516120541-06fdd3abe0ab
+	github.com/Dreamacro/clash v1.6.5
-	github.com/Microsoft/go-winio v0.4.16 // indirect
+	github.com/Microsoft/go-winio v0.5.0 // indirect
-	github.com/containerd/containerd v1.4.4 // indirect
+	github.com/containerd/containerd v1.5.3 // indirect
-	github.com/docker/distribution v2.7.1+incompatible // indirect
+	github.com/docker/docker v20.10.7+incompatible
-	github.com/docker/docker v20.10.6+incompatible
 	github.com/docker/go-connections v0.4.0
-	github.com/docker/go-units v0.4.0 // indirect
-	github.com/gogo/protobuf v1.3.2 // indirect
-	github.com/google/go-cmp v0.5.5 // indirect
 	github.com/gorilla/mux v1.8.0 // indirect
-	github.com/miekg/dns v1.1.42
+	github.com/miekg/dns v1.1.43
-	github.com/moby/term v0.0.0-20201216013528-df9cb8a40635 // indirect
+	github.com/moby/term v0.0.0-20210619224110-3f7ff695adc6 // indirect
 	github.com/morikuni/aec v1.0.0 // indirect
-	github.com/opencontainers/go-digest v1.0.0 // indirect
-	github.com/opencontainers/image-spec v1.0.1 // indirect
 	github.com/stretchr/testify v1.7.0
-	golang.org/x/net v0.0.0-20210510120150-4163338589ed
+	golang.org/x/net v0.0.0-20210614182718-04defd469f4e
-	golang.org/x/time v0.0.0-20210220033141-f8bda1e9f3ba // indirect
+	golang.org/x/time v0.0.0-20210611083556-38a9dc6acbc6 // indirect
-	google.golang.org/grpc v1.36.0 // indirect
+	google.golang.org/grpc v1.39.0 // indirect
-	gotest.tools/v3 v3.0.3 // indirect
 )

test/snell_test.go

@@ -119,3 +119,40 @@ func TestClash_Snell(t *testing.T) {
 	time.Sleep(waitTime)
 	testSuit(t, proxy)
 }
+
+func Benchmark_Snell(b *testing.B) {
+	cfg := &container.Config{
+		Image:        ImageSnell,
+		ExposedPorts: defaultExposedPorts,
+		Cmd:          []string{"-c", "/config.conf"},
+	}
+	hostCfg := &container.HostConfig{
+		PortBindings: defaultPortBindings,
+		Binds:        []string{fmt.Sprintf("%s:/config.conf", C.Path.Resolve("snell-http.conf"))},
+	}
+
+	id, err := startContainer(cfg, hostCfg, "snell-http")
+	if err != nil {
+		assert.FailNow(b, err.Error())
+	}
+
+	b.Cleanup(func() {
+		cleanContainer(id)
+	})
+
+	proxy, err := outbound.NewSnell(outbound.SnellOption{
+		Name:   "snell",
+		Server: localIP.String(),
+		Port:   10002,
+		Psk:    "password",
+		ObfsOpts: map[string]interface{}{
+			"mode": "http",
+		},
+	})
+	if err != nil {
+		assert.FailNow(b, err.Error())
+	}
+
+	time.Sleep(waitTime)
+	benchmarkProxy(b, proxy)
+}

test/ss_test.go

@@ -170,3 +170,38 @@ func TestClash_ShadowsocksV2RayPlugin(t *testing.T) {
 	time.Sleep(waitTime)
 	testSuit(t, proxy)
 }
+
+func Benchmark_Shadowsocks(b *testing.B) {
+	cfg := &container.Config{
+		Image:        ImageShadowsocks,
+		Env:          []string{"SS_MODULE=ss-server", "SS_CONFIG=-s 0.0.0.0 -u -v -p 10002 -m aes-256-gcm -k FzcLbKs2dY9mhL"},
+		ExposedPorts: defaultExposedPorts,
+	}
+	hostCfg := &container.HostConfig{
+		PortBindings: defaultPortBindings,
+	}
+
+	id, err := startContainer(cfg, hostCfg, "ss")
+	if err != nil {
+		assert.FailNow(b, err.Error())
+	}
+
+	b.Cleanup(func() {
+		cleanContainer(id)
+	})
+
+	proxy, err := outbound.NewShadowSocks(outbound.ShadowSocksOption{
+		Name:     "ss",
+		Server:   localIP.String(),
+		Port:     10002,
+		Password: "FzcLbKs2dY9mhL",
+		Cipher:   "aes-256-gcm",
+		UDP:      true,
+	})
+	if err != nil {
+		assert.FailNow(b, err.Error())
+	}
+
+	time.Sleep(waitTime)
+	benchmarkProxy(b, proxy)
+}

test/trojan_test.go

@@ -92,3 +92,43 @@ func TestClash_TrojanGrpc(t *testing.T) {
 	time.Sleep(waitTime)
 	testSuit(t, proxy)
 }
+
+func Benchmark_Trojan(b *testing.B) {
+	cfg := &container.Config{
+		Image:        ImageTrojan,
+		ExposedPorts: defaultExposedPorts,
+	}
+	hostCfg := &container.HostConfig{
+		PortBindings: defaultPortBindings,
+		Binds: []string{
+			fmt.Sprintf("%s:/config/config.json", C.Path.Resolve("trojan.json")),
+			fmt.Sprintf("%s:/path/to/certificate.crt", C.Path.Resolve("example.org.pem")),
+			fmt.Sprintf("%s:/path/to/private.key", C.Path.Resolve("example.org-key.pem")),
+		},
+	}
+
+	id, err := startContainer(cfg, hostCfg, "trojan")
+	if err != nil {
+		assert.FailNow(b, err.Error())
+	}
+
+	b.Cleanup(func() {
+		cleanContainer(id)
+	})
+
+	proxy, err := outbound.NewTrojan(outbound.TrojanOption{
+		Name:           "trojan",
+		Server:         localIP.String(),
+		Port:           10002,
+		Password:       "password",
+		SNI:            "example.org",
+		SkipCertVerify: true,
+		UDP:            true,
+	})
+	if err != nil {
+		assert.FailNow(b, err.Error())
+	}
+
+	time.Sleep(waitTime)
+	benchmarkProxy(b, proxy)
+}

test/util_other_test.go

@@ -0,0 +1,12 @@
+// +build !darwin
+
+package main
+
+import (
+	"errors"
+	"net"
+)
+
+func defaultRouteIP() (net.IP, error) {
+	return nil, errors.New("not supported")
+}

test/vmess_test.go

@@ -345,3 +345,41 @@ func TestClash_VmessGrpc(t *testing.T) {
 	time.Sleep(waitTime)
 	testSuit(t, proxy)
 }
+
+func Benchmark_Vmess(b *testing.B) {
+	configPath := C.Path.Resolve("vmess-aead.json")
+
+	cfg := &container.Config{
+		Image:        ImageVmess,
+		ExposedPorts: defaultExposedPorts,
+	}
+	hostCfg := &container.HostConfig{
+		PortBindings: defaultPortBindings,
+		Binds:        []string{fmt.Sprintf("%s:/etc/v2ray/config.json", configPath)},
+	}
+
+	id, err := startContainer(cfg, hostCfg, "vmess-aead")
+	if err != nil {
+		assert.FailNow(b, err.Error())
+	}
+
+	b.Cleanup(func() {
+		cleanContainer(id)
+	})
+
+	proxy, err := outbound.NewVmess(outbound.VmessOption{
+		Name:    "vmess",
+		Server:  localIP.String(),
+		Port:    10002,
+		UUID:    "b831381d-6324-4d53-ad4f-8cda48b30811",
+		Cipher:  "auto",
+		AlterID: 0,
+		UDP:     true,
+	})
+	if err != nil {
+		assert.FailNow(b, err.Error())
+	}
+
+	time.Sleep(waitTime)
+	benchmarkProxy(b, proxy)
+}

transport/socks4/socks4.go

@@ -0,0 +1,195 @@
+package socks4
+
+import (
+	"bytes"
+	"encoding/binary"
+	"errors"
+	"io"
+	"net"
+	"strconv"
+
+	"github.com/Dreamacro/clash/component/auth"
+)
+
+const Version = 0x04
+
+type Command = uint8
+
+const (
+	CmdConnect Command = 0x01
+	CmdBind    Command = 0x02
+)
+
+type Code = uint8
+
+const (
+	RequestGranted          Code = 90
+	RequestRejected         Code = 91
+	RequestIdentdFailed     Code = 92
+	RequestIdentdMismatched Code = 93
+)
+
+var (
+	errVersionMismatched   = errors.New("version code mismatched")
+	errCommandNotSupported = errors.New("command not supported")
+	errIPv6NotSupported    = errors.New("IPv6 not supported")
+
+	ErrRequestRejected         = errors.New("request rejected or failed")
+	ErrRequestIdentdFailed     = errors.New("request rejected because SOCKS server cannot connect to identd on the client")
+	ErrRequestIdentdMismatched = errors.New("request rejected because the client program and identd report different user-ids")
+	ErrRequestUnknownCode      = errors.New("request failed with unknown code")
+)
+
+func ServerHandshake(rw io.ReadWriter, authenticator auth.Authenticator) (addr string, command Command, err error) {
+	var req [8]byte
+	if _, err = io.ReadFull(rw, req[:]); err != nil {
+		return
+	}
+
+	if req[0] != Version {
+		err = errVersionMismatched
+		return
+	}
+
+	if command = req[1]; command != CmdConnect {
+		err = errCommandNotSupported
+		return
+	}
+
+	var (
+		dstIP   = req[4:8] // [4]byte
+		dstPort = req[2:4] // [2]byte
+	)
+
+	var (
+		host   string
+		port   string
+		code   uint8
+		userID []byte
+	)
+	if userID, err = readUntilNull(rw); err != nil {
+		return
+	}
+
+	if isReservedIP(dstIP) {
+		var target []byte
+		if target, err = readUntilNull(rw); err != nil {
+			return
+		}
+		host = string(target)
+	}
+
+	port = strconv.Itoa(int(binary.BigEndian.Uint16(dstPort)))
+	if host != "" {
+		addr = net.JoinHostPort(host, port)
+	} else {
+		addr = net.JoinHostPort(net.IP(dstIP).String(), port)
+	}
+
+	// SOCKS4 only support USERID auth.
+	if authenticator == nil || authenticator.Verify(string(userID), "") {
+		code = RequestGranted
+	} else {
+		code = RequestIdentdMismatched
+	}
+
+	var reply [8]byte
+	reply[0] = 0x00 // reply code
+	reply[1] = code // result code
+	copy(reply[4:8], dstIP)
+	copy(reply[2:4], dstPort)
+
+	_, err = rw.Write(reply[:])
+	return
+}
+
+func ClientHandshake(rw io.ReadWriter, addr string, command Command, userID string) (err error) {
+	host, portStr, err := net.SplitHostPort(addr)
+	if err != nil {
+		return err
+	}
+
+	port, err := strconv.ParseUint(portStr, 10, 16)
+	if err != nil {
+		return err
+	}
+
+	ip := net.ParseIP(host)
+	if ip == nil /* HOST */ {
+		ip = net.IPv4(0, 0, 0, 1).To4()
+	} else if ip.To4() == nil /* IPv6 */ {
+		return errIPv6NotSupported
+	}
+
+	dstIP := ip.To4()
+
+	req := &bytes.Buffer{}
+	req.WriteByte(Version)
+	req.WriteByte(command)
+	binary.Write(req, binary.BigEndian, uint16(port))
+	req.Write(dstIP)
+	req.WriteString(userID)
+	req.WriteByte(0) /* NULL */
+
+	if isReservedIP(dstIP) /* SOCKS4A */ {
+		req.WriteString(host)
+		req.WriteByte(0) /* NULL */
+	}
+
+	if _, err = rw.Write(req.Bytes()); err != nil {
+		return err
+	}
+
+	var resp [8]byte
+	if _, err = io.ReadFull(rw, resp[:]); err != nil {
+		return err
+	}
+
+	if resp[0] != 0x00 {
+		return errVersionMismatched
+	}
+
+	switch resp[1] {
+	case RequestGranted:
+		return nil
+	case RequestRejected:
+		return ErrRequestRejected
+	case RequestIdentdFailed:
+		return ErrRequestIdentdFailed
+	case RequestIdentdMismatched:
+		return ErrRequestIdentdMismatched
+	default:
+		return ErrRequestUnknownCode
+	}
+}
+
+// For version 4A, if the client cannot resolve the destination host's
+// domain name to find its IP address, it should set the first three bytes
+// of DSTIP to NULL and the last byte to a non-zero value. (This corresponds
+// to IP address 0.0.0.x, with x nonzero. As decreed by IANA  -- The
+// Internet Assigned Numbers Authority -- such an address is inadmissible
+// as a destination IP address and thus should never occur if the client
+// can resolve the domain name.)
+func isReservedIP(ip net.IP) bool {
+	subnet := net.IPNet{
+		IP:   net.IPv4zero,
+		Mask: net.IPv4Mask(0xff, 0xff, 0xff, 0x00),
+	}
+
+	return !ip.IsUnspecified() && subnet.Contains(ip)
+}
+
+func readUntilNull(r io.Reader) ([]byte, error) {
+	var buf = &bytes.Buffer{}
+	var data [1]byte
+
+	for {
+		if _, err := r.Read(data[:]); err != nil {
+			return nil, err
+		}
+		if data[0] == 0 {
+			return buf.Bytes(), nil
+		}
+		buf.WriteByte(data[0])
+	}
+}

tunnel/tunnel.go

@@ -224,9 +224,9 @@ func handleUDPConn(packet *inbound.PacketAdapter) {
 		rawPc, err := proxy.DialUDP(metadata)
 		if err != nil {
 			if rule == nil {
-				log.Warnln("[UDP] dial %s to %s error: %s", proxy.Name(), metadata.String(), err.Error())
+				log.Warnln("[UDP] dial %s to %s error: %s", proxy.Name(), metadata.RemoteAddress(), err.Error())
 			} else {
-				log.Warnln("[UDP] dial %s (match %s/%s) to %s error: %s", proxy.Name(), rule.RuleType().String(), rule.Payload(), metadata.String(), err.Error())
+				log.Warnln("[UDP] dial %s (match %s/%s) to %s error: %s", proxy.Name(), rule.RuleType().String(), rule.Payload(), metadata.RemoteAddress(), err.Error())
 			}
 			return
 		}
@@ -235,13 +235,13 @@ func handleUDPConn(packet *inbound.PacketAdapter) {
 
 		switch true {
 		case rule != nil:
-			log.Infoln("[UDP] %s(%s) --> %s:%s match %s(%s) %s using %s", metadata.SourceAddress(), metadata.Process, metadata.String(), metadata.DstPort, rule.RuleType().String(), rule.Payload(), rule.NetWork().String(), rawPc.Chains().String())
+			log.Infoln("[UDP] %s(%s) --> %s:%s match %s(%s) %s using %s", metadata.SourceAddress(), metadata.Process, metadata.RemoteAddress(), metadata.DstPort, rule.RuleType().String(), rule.Payload(), rule.NetWork().String(), rawPc.Chains().String())
 		case mode == Global:
-			log.Infoln("[UDP] %s(%s) --> %s using GLOBAL", metadata.SourceAddress(), metadata.Process, metadata.String())
+			log.Infoln("[UDP] %s(%s) --> %s using GLOBAL", metadata.SourceAddress(), metadata.Process, metadata.RemoteAddress())
 		case mode == Direct:
-			log.Infoln("[UDP] %s(%s) --> %s using DIRECT", metadata.SourceAddress(), metadata.Process, metadata.String())
+			log.Infoln("[UDP] %s(%s) --> %s using DIRECT", metadata.SourceAddress(), metadata.Process, metadata.RemoteAddress())
 		default:
-			log.Infoln("[UDP] %s(%s) --> %s doesn't match any rule using DIRECT", metadata.SourceAddress(), metadata.Process, metadata.String())
+			log.Infoln("[UDP] %s(%s) --> %s doesn't match any rule using DIRECT", metadata.SourceAddress(), metadata.Process, metadata.RemoteAddress())
 		}
 
 		go handleUDPToLocal(packet.UDPPacket, pc, key, fAddr)
@@ -274,9 +274,9 @@ func handleTCPConn(ctx C.ConnContext) {
 	remoteConn, err := proxy.Dial(metadata)
 	if err != nil {
 		if rule == nil {
-			log.Warnln("[TCP] dial %s to %s error: %s", proxy.Name(), metadata.String(), err.Error())
+			log.Warnln("[TCP] dial %s to %s error: %s", proxy.Name(), metadata.RemoteAddress(), err.Error())
 		} else {
-			log.Warnln("[TCP] dial %s (match %s/%s) to %s error: %s", proxy.Name(), rule.RuleType().String(), rule.Payload(), metadata.String(), err.Error())
+			log.Warnln("[TCP] dial %s (match %s/%s) to %s error: %s", proxy.Name(), rule.RuleType().String(), rule.Payload(), metadata.RemoteAddress(), err.Error())
 		}
 		return
 	}
@@ -285,13 +285,13 @@ func handleTCPConn(ctx C.ConnContext) {
 
 	switch true {
 	case rule != nil:
-		log.Infoln("[TCP] %s(%s) --> %s:%s match %s(%s) %s using %s", metadata.SourceAddress(), metadata.Process, metadata.String(), metadata.DstPort, rule.RuleType().String(), rule.Payload(), rule.NetWork().String(), remoteConn.Chains().String())
+		log.Infoln("[TCP] %s(%s) --> %s:%s match %s(%s) %s using %s", metadata.SourceAddress(), metadata.Process, metadata.RemoteAddress(), metadata.DstPort, rule.RuleType().String(), rule.Payload(), rule.NetWork().String(), remoteConn.Chains().String())
 	case mode == Global:
-		log.Infoln("[TCP] %s(%s) --> %s using GLOBAL", metadata.SourceAddress(), metadata.Process, metadata.String())
+		log.Infoln("[TCP] %s(%s) --> %s using GLOBAL", metadata.SourceAddress(), metadata.Process, metadata.RemoteAddress())
 	case mode == Direct:
-		log.Infoln("[TCP] %s(%s) --> %s using DIRECT", metadata.SourceAddress(), metadata.Process, metadata.String())
+		log.Infoln("[TCP] %s(%s) --> %s using DIRECT", metadata.SourceAddress(), metadata.Process, metadata.RemoteAddress())
 	default:
-		log.Infoln("[TCP] %s(%s) --> %s doesn't match any rule using DIRECT", metadata.SourceAddress(), metadata.Process, metadata.String())
+		log.Infoln("[TCP] %s(%s) --> %s doesn't match any rule using DIRECT", metadata.SourceAddress(), metadata.Process, metadata.RemoteAddress())
 	}
 
 	handleSocket(ctx, remoteConn)