diff --git a/depends/xtaci/smux/frame.go b/depends/xtaci/smux/frame.go index 74e188d59..467a0581c 100644 --- a/depends/xtaci/smux/frame.go +++ b/depends/xtaci/smux/frame.go @@ -26,8 +26,6 @@ const ( const ( // initial peer window guess, a slow-start initialPeerWindow = 262144 - - frameFailState = 1 ) const ( @@ -44,7 +42,6 @@ type Frame struct { cmd byte sid uint32 data []byte - state int32 } func newFrame(version byte, cmd byte, sid uint32) Frame { diff --git a/depends/xtaci/smux/session.go b/depends/xtaci/smux/session.go index 6d6388b85..583e7a27e 100644 --- a/depends/xtaci/smux/session.go +++ b/depends/xtaci/smux/session.go @@ -4,7 +4,6 @@ import ( "container/heap" "encoding/binary" "errors" - "fmt" "io" "net" "sync" @@ -28,6 +27,7 @@ type writeRequest struct { prio uint32 frame Frame result chan writeResult + stop chan struct{} } type writeResult struct { @@ -458,18 +458,12 @@ func (s *Session) sendLoop() { case request := <-s.writes: s.frameLock.Lock() - state := atomic.LoadInt32(&request.frame.state) - if state == frameFailState { - s.frameLock.Unlock() - - result := writeResult{ - n: 0, - err: fmt.Errorf("abnormal stream %d", request.frame.sid), - } - - request.result <- result + select { + case <-request.stop: close(request.result) + s.frameLock.Unlock() continue + default: } buf[0] = request.frame.ver @@ -518,12 +512,24 @@ func (s *Session) writeFrame(f Frame) (n int, err error) { } // 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, prio uint32) (n int, err error) { req := writeRequest{ prio: prio, frame: f, result: make(chan writeResult, 1), + stop: make(chan struct{}), } + + defer func() { + if err != nil { + s.frameLock.Lock() + close(req.stop) + s.frameLock.Unlock() + return + } + close(req.stop) + }() + select { case s.shaper <- req: case <-s.die: diff --git a/depends/xtaci/smux/stream.go b/depends/xtaci/smux/stream.go index dede667c2..a95fb56bc 100644 --- a/depends/xtaci/smux/stream.go +++ b/depends/xtaci/smux/stream.go @@ -329,9 +329,6 @@ func (s *Stream) Write(b []byte) (n int, err error) { s.numWritten++ sent += n if err != nil { - s.sess.frameLock.Lock() - atomic.StoreInt32(&frame.state, frameFailState) - s.sess.frameLock.Unlock() return sent, err } }