Merge pull request #186 from shadowsocks/feature-generic-cork

Feature generic cork

main.go

@@ -62,7 +62,7 @@ func main() {
 	flag.StringVar(&flags.PluginOpts, "plugin-opts", "", "Set SIP003 plugin options. (e.g., \"server;tls;host=mydomain.me\")")
 	flag.BoolVar(&flags.UDP, "udp", false, "(server-only) enable UDP support")
 	flag.BoolVar(&flags.TCP, "tcp", true, "(server-only) enable TCP support")
-	flag.BoolVar(&config.TCPCork, "tcpcork", false, "(client-only) enable TCP_CORK (Linux) or TCP_NOPUSH (BSD) for the first few packets")
+	flag.BoolVar(&config.TCPCork, "tcpcork", false, "coalesce writing first few packets")
 	flag.DurationVar(&config.UDPTimeout, "udptimeout", 5*time.Minute, "UDP tunnel timeout")
 	flag.Parse()
 

tcp.go

@@ -1,6 +1,7 @@
 package main
 
 import (
+	"bufio"
 	"io"
 	"io/ioutil"
 	"net"
@@ -71,9 +72,8 @@ func tcpLocal(addr, server string, shadow func(net.Conn) net.Conn, getAddr func(
 				return
 			}
 			defer rc.Close()
-			tc := rc.(*net.TCPConn)
 			if config.TCPCork {
-				timedCork(tc, 10*time.Millisecond)
+				rc = timedCork(rc, 10*time.Millisecond, 1280)
 			}
 			rc = shadow(rc)
 
@@ -112,11 +112,14 @@ func tcpRemote(addr string, shadow func(net.Conn) net.Conn) {
 
 		go func() {
 			defer c.Close()
+			if config.TCPCork {
+				c = timedCork(c, 10*time.Millisecond, 1280)
+			}
 			sc := shadow(c)
 
 			tgt, err := socks.ReadAddr(sc)
 			if err != nil {
-				logf("failed to get target address: %v", err)
+				logf("failed to get target address from %v: %v", c.RemoteAddr(), err)
 				// drain c to avoid leaking server behavioral features
 				// see https://www.ndss-symposium.org/ndss-paper/detecting-probe-resistant-proxies/
 				_, err = io.Copy(ioutil.Discard, c)
@@ -149,20 +152,59 @@ func tcpRemote(addr string, shadow func(net.Conn) net.Conn) {
 func relay(left, right net.Conn) error {
 	var err, err1 error
 	var wg sync.WaitGroup
-
+	var wait = 5 * time.Second
 	wg.Add(1)
 	go func() {
 		defer wg.Done()
 		_, err1 = io.Copy(right, left)
-		right.SetReadDeadline(time.Now()) // unblock read on right
+		right.SetReadDeadline(time.Now().Add(wait)) // unblock read on right
 	}()
 
 	_, err = io.Copy(left, right)
-	left.SetReadDeadline(time.Now()) // unblock read on left
+	left.SetReadDeadline(time.Now().Add(wait)) // unblock read on left
 	wg.Wait()
 
 	if err1 != nil {
 		err = err1
 	}
 	return err
 }
+
+type corkedConn struct {
+	net.Conn
+	bufw   *bufio.Writer
+	corked bool
+	delay  time.Duration
+	err    error
+	lock   sync.Mutex
+	once   sync.Once
+}
+
+func timedCork(c net.Conn, d time.Duration, bufSize int) net.Conn {
+	return &corkedConn{
+		Conn:   c,
+		bufw:   bufio.NewWriterSize(c, bufSize),
+		corked: true,
+		delay:  d,
+	}
+}
+
+func (w *corkedConn) Write(p []byte) (int, error) {
+	w.lock.Lock()
+	defer w.lock.Unlock()
+	if w.err != nil {
+		return 0, w.err
+	}
+	if w.corked {
+		w.once.Do(func() {
+			time.AfterFunc(w.delay, func() {
+				w.lock.Lock()
+				defer w.lock.Unlock()
+				w.corked = false
+				w.err = w.bufw.Flush()
+			})
+		})
+		return w.bufw.Write(p)
+	}
+	return w.Conn.Write(p)
+}

tcp_darwin.go

@@ -2,8 +2,6 @@ package main
 
 import (
 	"net"
-	"syscall"
-	"time"
 
 	"github.com/shadowsocks/go-shadowsocks2/pfutil"
 	"github.com/shadowsocks/go-shadowsocks2/socks"
@@ -24,18 +22,3 @@ func natLookup(c net.Conn) (socks.Addr, error) {
 	}
 	panic("not TCP connection")
 }
-
-func timedCork(c *net.TCPConn, d time.Duration) error {
-	rc, err := c.SyscallConn()
-	if err != nil {
-		return err
-	}
-	rc.Control(func(fd uintptr) { err = syscall.SetsockoptInt(int(fd), syscall.IPPROTO_TCP, syscall.TCP_NOPUSH, 1) })
-	if err != nil {
-		return err
-	}
-	time.AfterFunc(d, func() {
-		rc.Control(func(fd uintptr) { syscall.SetsockoptInt(int(fd), syscall.IPPROTO_TCP, syscall.TCP_NOPUSH, 0) })
-	})
-	return nil
-}

tcp_linux.go

@@ -2,8 +2,6 @@ package main
 
 import (
 	"net"
-	"syscall"
-	"time"
 
 	"github.com/shadowsocks/go-shadowsocks2/nfutil"
 	"github.com/shadowsocks/go-shadowsocks2/socks"
@@ -28,18 +26,3 @@ func redir6Local(addr, server string, shadow func(net.Conn) net.Conn) {
 	logf("TCP6 redirect %s <-> %s", addr, server)
 	tcpLocal(addr, server, shadow, func(c net.Conn) (socks.Addr, error) { return getOrigDst(c, true) })
 }
-
-func timedCork(c *net.TCPConn, d time.Duration) error {
-	rc, err := c.SyscallConn()
-	if err != nil {
-		return err
-	}
-	rc.Control(func(fd uintptr) { err = syscall.SetsockoptInt(int(fd), syscall.IPPROTO_TCP, syscall.TCP_CORK, 1) })
-	if err != nil {
-		return err
-	}
-	time.AfterFunc(d, func() {
-		rc.Control(func(fd uintptr) { syscall.SetsockoptInt(int(fd), syscall.IPPROTO_TCP, syscall.TCP_CORK, 0) })
-	})
-	return nil
-}

tcp_other.go

@@ -4,7 +4,6 @@ package main
 
 import (
 	"net"
-	"time"
 )
 
 func redirLocal(addr, server string, shadow func(net.Conn) net.Conn) {
@@ -14,5 +13,3 @@ func redirLocal(addr, server string, shadow func(net.Conn) net.Conn) {
 func redir6Local(addr, server string, shadow func(net.Conn) net.Conn) {
 	logf("TCP6 redirect not supported")
 }
-
-func timedCork(c *net.TCPConn, d time.Duration) error { return nil }