fix(smux): use stop chan to sync request fail status. #1000193713

Signed-off-by: Victor1319 <zengxuewei@oppo.com>

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 {

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:

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
 		}
 	}