From fc57c9809873e52972e278348e62312309fa68e3 Mon Sep 17 00:00:00 2001 From: xtaci Date: Sun, 29 Dec 2019 18:31:26 +0800 Subject: [PATCH] rawcopy leads mistakes in smux in v2 --- client/main.go | 9 ++--- generic/copy.go | 18 +-------- generic/rawcopy_unix.go | 83 -------------------------------------- generic/rawcopy_windows.go | 14 ------- server/main.go | 9 ++--- 5 files changed, 7 insertions(+), 126 deletions(-) delete mode 100644 generic/rawcopy_unix.go delete mode 100644 generic/rawcopy_windows.go diff --git a/client/main.go b/client/main.go index 3c5a4bc..25c32db 100644 --- a/client/main.go +++ b/client/main.go @@ -32,7 +32,7 @@ const ( var VERSION = "SELFBUILD" // handleClient aggregates connection p1 on mux with 'writeLock' -func handleClient(session *smux.Session, p1 net.Conn, ctrl *generic.CopyControl, quiet bool) { +func handleClient(session *smux.Session, p1 net.Conn, quiet bool) { logln := func(v ...interface{}) { if !quiet { log.Println(v...) @@ -52,7 +52,7 @@ func handleClient(session *smux.Session, p1 net.Conn, ctrl *generic.CopyControl, // start tunnel & wait for tunnel termination streamCopy := func(dst io.Writer, src io.ReadCloser) { - if _, err := generic.Copy(dst, src, ctrl); err != nil { + if _, err := generic.Copy(dst, src); err != nil { // report protocol error if err == smux.ErrInvalidProtocol { log.Println("smux", err, "in:", p1.RemoteAddr(), "out:", fmt.Sprint(p2.RemoteAddr(), "(", p2.ID(), ")")) @@ -425,13 +425,11 @@ func main() { muxes := make([]struct { session *smux.Session ttl time.Time - ctrl *generic.CopyControl // for control of memory in copying }, numconn) for k := range muxes { muxes[k].session = waitConn() muxes[k].ttl = time.Now().Add(time.Duration(config.AutoExpire) * time.Second) - muxes[k].ctrl = &generic.CopyControl{Buffer: make([]byte, bufSize)} } chScavenger := make(chan *smux.Session, 128) @@ -450,10 +448,9 @@ func main() { chScavenger <- muxes[idx].session muxes[idx].session = waitConn() muxes[idx].ttl = time.Now().Add(time.Duration(config.AutoExpire) * time.Second) - muxes[idx].ctrl = &generic.CopyControl{Buffer: make([]byte, bufSize)} } - go handleClient(muxes[idx].session, p1, muxes[idx].ctrl, config.Quiet) + go handleClient(muxes[idx].session, p1, config.Quiet) rr++ } } diff --git a/generic/copy.go b/generic/copy.go index 2964dcb..833c633 100644 --- a/generic/copy.go +++ b/generic/copy.go @@ -2,19 +2,12 @@ package generic import ( "io" - "net" - "sync" ) const bufSize = 4096 -type CopyControl struct { - Buffer []byte // shared buffer for copying controlled by mutex - sync.Mutex -} - // Memory optimized io.Copy function specified for this library -func Copy(dst io.Writer, src io.Reader, ctrl *CopyControl) (written int64, err error) { +func Copy(dst io.Writer, src io.Reader) (written int64, err error) { // If the reader has a WriteTo method, use it to do the copy. // Avoids an allocation and a copy. if wt, ok := src.(io.WriterTo); ok { @@ -25,15 +18,6 @@ func Copy(dst io.Writer, src io.Reader, ctrl *CopyControl) (written int64, err e return rt.ReadFrom(src) } - // if src is net.TCPConn, and dst is a multiplexed connection - // reading can be controlled by writable events of smux - // and make the reading serialized - if tcpconn, ok := src.(*net.TCPConn); ok { - if ctrl != nil { - return rawCopy(dst, tcpconn, ctrl) - } - } - // fallback to standard io.CopyBuffer buf := make([]byte, bufSize) return io.CopyBuffer(dst, src, buf) diff --git a/generic/rawcopy_unix.go b/generic/rawcopy_unix.go deleted file mode 100644 index 2de72d3..0000000 --- a/generic/rawcopy_unix.go +++ /dev/null @@ -1,83 +0,0 @@ -// +build aix darwin dragonfly freebsd linux netbsd openbsd solaris - -package generic - -import ( - "io" - "net" - - "syscall" -) - -// rawCopy can fan in N src into 1 dst with only 1 shared buffer -func rawCopy(dst io.Writer, src *net.TCPConn, ctrl *CopyControl) (written int64, err error) { - c, err := src.SyscallConn() - if err != nil { - return 0, err - } - - buf := ctrl.Buffer - var locked bool - for { - var er error - var nr int - rr := c.Read(func(s uintptr) bool { - // if the 'src' readable, acquire the shared lock first - // to make sure no other writers to 'dst' are blocked on dst.Write. - // With such design, we only need 1 buffer for a specific 'dst', - // especially when 'dst' is a multiplexed connection. - ctrl.Lock() - locked = true - nr, er = syscall.Read(int(s), buf) - if er == syscall.EAGAIN { - ctrl.Unlock() - locked = false - return false - } - // keep the lock on the shared buffer - // for the following dst.Write - return true - }) - - // read EOF - if nr == 0 && er == nil { - break - } - - if nr > 0 { - nw, ew := dst.Write(buf[0:nr]) - ctrl.Unlock() - locked = false - - if nw > 0 { - written += int64(nw) - } - if ew != nil { - err = ew - break - } - if nr != nw { - err = io.ErrShortWrite - break - } - } - if er != nil { - if er != io.EOF { - err = er - } - break - } - if rr != nil { - if rr != io.EOF { - err = rr - } - break - } - } - - if locked { - ctrl.Unlock() - } - - return written, err -} diff --git a/generic/rawcopy_windows.go b/generic/rawcopy_windows.go deleted file mode 100644 index 9e05050..0000000 --- a/generic/rawcopy_windows.go +++ /dev/null @@ -1,14 +0,0 @@ -// +build windows - -package generic - -import ( - "io" - "net" -) - -func rawCopy(dst io.Writer, src *net.TCPConn, ctrl *CopyControl) (written int64, err error) { - // fallback to standard io.CopyBuffer - buf := make([]byte, bufSize) - return io.CopyBuffer(dst, src, buf) -} diff --git a/server/main.go b/server/main.go index 24dca2c..a63fdb0 100644 --- a/server/main.go +++ b/server/main.go @@ -57,9 +57,6 @@ func handleMux(conn net.Conn, config *Config) { } defer mux.Close() - // copy to stream control - copyControl := &generic.CopyControl{Buffer: make([]byte, bufSize)} - for { stream, err := mux.AcceptStream() if err != nil { @@ -81,12 +78,12 @@ func handleMux(conn net.Conn, config *Config) { p1.Close() return } - handleClient(p1, p2, copyControl, config.Quiet) + handleClient(p1, p2, config.Quiet) }(stream) } } -func handleClient(p1 *smux.Stream, p2 net.Conn, ctrl *generic.CopyControl, quiet bool) { +func handleClient(p1 *smux.Stream, p2 net.Conn, quiet bool) { logln := func(v ...interface{}) { if !quiet { log.Println(v...) @@ -101,7 +98,7 @@ func handleClient(p1 *smux.Stream, p2 net.Conn, ctrl *generic.CopyControl, quiet // start tunnel & wait for tunnel termination streamCopy := func(dst io.Writer, src io.ReadCloser) { - if _, err := generic.Copy(dst, src, ctrl); err != nil { + if _, err := generic.Copy(dst, src); err != nil { if err == smux.ErrInvalidProtocol { log.Println("smux", err, "in:", fmt.Sprint(p1.RemoteAddr(), "(", p1.ID(), ")"), "out:", p2.RemoteAddr()) }