diff --git a/go.mod b/go.mod index 29df302..ebe2bc7 100644 --- a/go.mod +++ b/go.mod @@ -5,7 +5,7 @@ require ( github.com/pkg/errors v0.9.1 github.com/urfave/cli v1.22.12 github.com/xtaci/kcp-go/v5 v5.6.2 - github.com/xtaci/smux v1.5.22 + github.com/xtaci/smux v1.5.24 github.com/xtaci/tcpraw v1.2.25 golang.org/x/crypto v0.5.0 ) diff --git a/go.sum b/go.sum index f6b3b9b..65aa133 100644 --- a/go.sum +++ b/go.sum @@ -70,8 +70,10 @@ github.com/xtaci/kcp-go/v5 v5.6.2 h1:pSXMa5MOsb+EIZKe4sDBqlTExu2A/2Z+DFhoX2qtt2A github.com/xtaci/kcp-go/v5 v5.6.2/go.mod h1:LsinWoru+lWWJHb+EM9HeuqYxV6bb9rNcK12v67jYzQ= github.com/xtaci/lossyconn v0.0.0-20190602105132-8df528c0c9ae h1:J0GxkO96kL4WF+AIT3M4mfUVinOCPgf2uUWYFUzN0sM= github.com/xtaci/lossyconn v0.0.0-20190602105132-8df528c0c9ae/go.mod h1:gXtu8J62kEgmN++bm9BVICuT/e8yiLI2KFobd/TRFsE= -github.com/xtaci/smux v1.5.22 h1:W9AGvTnn7C9ORv2CsQ1QAeoKlsI/o6SUkFUN6sXYhdA= -github.com/xtaci/smux v1.5.22/go.mod h1:OMlQbT5vcgl2gb49mFkYo6SMf+zP3rcjcwQz7ZU7IGY= +github.com/xtaci/smux v1.5.23 h1:fKoWB0yoiOUJP6tdpWPm7Cg1+rMom83n1Jjtj2FrF1s= +github.com/xtaci/smux v1.5.23/go.mod h1:OMlQbT5vcgl2gb49mFkYo6SMf+zP3rcjcwQz7ZU7IGY= +github.com/xtaci/smux v1.5.24 h1:77emW9dtnOxxOQ5ltR+8BbsX1kzcOxQ5gB+aaV9hXOY= +github.com/xtaci/smux v1.5.24/go.mod h1:OMlQbT5vcgl2gb49mFkYo6SMf+zP3rcjcwQz7ZU7IGY= github.com/xtaci/tcpraw v1.2.25 h1:VDlqo0op17JeXBM6e2G9ocCNLOJcw9mZbobMbJjo0vk= github.com/xtaci/tcpraw v1.2.25/go.mod h1:dKyZ2V75s0cZ7cbgJYdxPvms7af0joIeOyx1GgJQbLk= github.com/yuin/goldmark v1.4.13/go.mod h1:6yULJ656Px+3vBD8DxQVa3kxgyrAnzto9xy5taEt/CY= diff --git a/vendor/github.com/xtaci/smux/session.go b/vendor/github.com/xtaci/smux/session.go index 3fad379..88015e3 100644 --- a/vendor/github.com/xtaci/smux/session.go +++ b/vendor/github.com/xtaci/smux/session.go @@ -14,6 +14,15 @@ import ( const ( defaultAcceptBacklog = 1024 maxShaperSize = 1024 + openCloseTimeout = 30 * time.Second // stream open/close timeout +) + +// define frame class +type CLASSID int + +const ( + CLSCTRL CLASSID = iota + CLSDATA ) var ( @@ -25,7 +34,7 @@ var ( ) type writeRequest struct { - prio uint32 + class CLASSID frame Frame seq uint32 result chan writeResult @@ -395,7 +404,7 @@ func (s *Session) keepalive() { for { select { case <-tickerPing.C: - s.writeFrameInternal(newFrame(byte(s.config.Version), cmdNOP, 0), tickerPing.C, 0) + s.writeFrameInternal(newFrame(byte(s.config.Version), cmdNOP, 0), tickerPing.C, CLSCTRL) s.notifyBucket() // force a signal to the recvLoop case <-tickerTimeout.C: if !atomic.CompareAndSwapInt32(&s.dataReady, 1, 0) { @@ -480,11 +489,6 @@ func (s *Session) sendLoop() { binary.LittleEndian.PutUint16(buf[2:], uint16(len(request.frame.data))) binary.LittleEndian.PutUint32(buf[4:], request.frame.sid) - // set timeout conn - if tconn, ok := s.conn.(interface{ SetWriteDeadline(t time.Time) error }); ok { - tconn.SetWriteDeadline(time.Now().Add(s.config.KeepAliveTimeout)) - } - if len(vec) > 0 { vec[0] = buf[:headerSize] vec[1] = request.frame.data @@ -519,13 +523,13 @@ func (s *Session) sendLoop() { // writeFrame writes the frame to the underlying connection // and returns the number of bytes written if successful func (s *Session) writeFrame(f Frame) (n int, err error) { - return s.writeFrameInternal(f, time.After(s.config.KeepAliveTimeout), 0) + return s.writeFrameInternal(f, time.After(openCloseTimeout), CLSCTRL) } // internal writeFrame version to support deadline used in keepalive -func (s *Session) writeFrameInternal(f Frame, deadline <-chan time.Time, prio uint32) (int, error) { +func (s *Session) writeFrameInternal(f Frame, deadline <-chan time.Time, class CLASSID) (int, error) { req := writeRequest{ - prio: prio, + class: class, frame: f, seq: atomic.AddUint32(&s.requestID, 1), result: make(chan writeResult, 1), diff --git a/vendor/github.com/xtaci/smux/shaper.go b/vendor/github.com/xtaci/smux/shaper.go index 35773ee..8d52ef7 100644 --- a/vendor/github.com/xtaci/smux/shaper.go +++ b/vendor/github.com/xtaci/smux/shaper.go @@ -6,8 +6,14 @@ func _itimediff(later, earlier uint32) int32 { type shaperHeap []writeRequest -func (h shaperHeap) Len() int { return len(h) } -func (h shaperHeap) Less(i, j int) bool { return _itimediff(h[j].seq, h[i].seq) > 0 } +func (h shaperHeap) Len() int { return len(h) } +func (h shaperHeap) Less(i, j int) bool { + if h[i].class != h[j].class { + return h[i].class < h[j].class + } + return _itimediff(h[j].seq, h[i].seq) > 0 +} + func (h shaperHeap) Swap(i, j int) { h[i], h[j] = h[j], h[i] } func (h *shaperHeap) Push(x interface{}) { *h = append(*h, x.(writeRequest)) } diff --git a/vendor/github.com/xtaci/smux/stream.go b/vendor/github.com/xtaci/smux/stream.go index 94e858e..0d7e045 100644 --- a/vendor/github.com/xtaci/smux/stream.go +++ b/vendor/github.com/xtaci/smux/stream.go @@ -255,7 +255,7 @@ func (s *Stream) sendWindowUpdate(consumed uint32) error { binary.LittleEndian.PutUint32(hdr[:], consumed) binary.LittleEndian.PutUint32(hdr[4:], uint32(s.sess.config.MaxStreamBuffer)) frame.data = hdr[:] - _, err := s.sess.writeFrameInternal(frame, deadline, 0) + _, err := s.sess.writeFrameInternal(frame, deadline, CLSDATA) return err } @@ -325,7 +325,7 @@ func (s *Stream) Write(b []byte) (n int, err error) { } frame.data = bts[:sz] bts = bts[sz:] - n, err := s.sess.writeFrameInternal(frame, deadline, 0) + n, err := s.sess.writeFrameInternal(frame, deadline, CLSDATA) s.numWritten++ sent += n if err != nil { @@ -393,7 +393,7 @@ func (s *Stream) writeV2(b []byte) (n int, err error) { } frame.data = bts[:sz] bts = bts[sz:] - n, err := s.sess.writeFrameInternal(frame, deadline, 0) + n, err := s.sess.writeFrameInternal(frame, deadline, CLSDATA) atomic.AddUint32(&s.numWritten, uint32(sz)) sent += n if err != nil { diff --git a/vendor/modules.txt b/vendor/modules.txt index aa81921..707622b 100644 --- a/vendor/modules.txt +++ b/vendor/modules.txt @@ -38,7 +38,7 @@ github.com/urfave/cli # github.com/xtaci/kcp-go/v5 v5.6.2 ## explicit; go 1.13 github.com/xtaci/kcp-go/v5 -# github.com/xtaci/smux v1.5.22 +# github.com/xtaci/smux v1.5.24 ## explicit; go 1.13 github.com/xtaci/smux # github.com/xtaci/tcpraw v1.2.25