github/kcptun
4
0
Fork 0
mirror of https://github.com/xtaci/kcptun.git synced 2024-04-21 12:32:32 +00:00
Code Issues Releases Wiki Activity

rawcopy leads mistakes in smux in v2

Browse Source
This commit is contained in:
xtaci
2019-12-29 18:31:26 +08:00
parent 34bc48fd64
commit fc57c98098
5 changed files with 7 additions and 126 deletions
Download Patch File Download Diff File Expand all files Collapse all files
client/main.go
+3 -6
View File
@@ -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++
}
}
generic/copy.go
+1 -17
View File
@@ -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)
generic/rawcopy_unix.go
-83
View File
@@ -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
}
generic/rawcopy_windows.go
-14
View File
@@ -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)
}
server/main.go
+3 -6
View File
@@ -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())
}