fix race (#10)

* fix race

* up

* fix

Author: guonaihong

.github/workflows/go.yml

@@ -38,10 +38,12 @@ jobs:
       run: env GOARCH=386 go test -test.run=Test_Retry_sleep -v 
       #run: env GOARCH=386 go test -v -coverprofile='coverage.out' -covermode=count ./...
 
+    - name: Test-Race
+      run: env GOARCH=amd64 go test -v -race ./...
+
     - name: Test-amd64
       run: env GOARCH=amd64 go test -v -coverprofile='coverage.out' -covermode=count ./...
 
-
     - name: Upload Coverage report
       uses: codecov/codecov-action@v1
       with:

.gitignore

@@ -1,4 +1,5 @@
 *swp
+/coverage.out
 /cover.cov
 autobahn-testsuite
 autobahn-client-testsuite-darwin-arm64

conn.go

@@ -61,7 +61,7 @@ type Conn struct {
 	once   sync.Once
 
 	fr fixedreader.FixedReader
-	fw fixedwriter.FixedWriter
+	// fw fixedwriter.FixedWriter
 	bp bytespool.BytesPool // 实验某些特性加的字段
 
 	delayWrite
@@ -389,7 +389,8 @@ func (c *Conn) WriteMessage(op Opcode, writeBuf []byte) (err error) {
 		maskValue = rand.Uint32()
 	}
 
-	return frame.WriteFrame(&c.fw, c.c, writeBuf, true, rsv1, c.client, op, maskValue)
+	var fw fixedwriter.FixedWriter
+	return frame.WriteFrame(&fw, c.c, writeBuf, true, rsv1, c.client, op, maskValue)
 }
 
 func (c *Conn) SetWriteDeadline(t time.Time) error {
@@ -460,16 +461,17 @@ func (c *Conn) writeFragment(op Opcode, writeBuf []byte, maxFragment int /*单
 		maskValue = rand.Uint32()
 	}
 
+	var fw fixedwriter.FixedWriter
 	for len(writeBuf) > 0 {
 		if len(writeBuf) > maxFragment {
-			if err := frame.WriteFrame(&c.fw, c.c, writeBuf[:maxFragment], false, rsv1, c.client, op, maskValue); err != nil {
+			if err := frame.WriteFrame(&fw, c.c, writeBuf[:maxFragment], false, rsv1, c.client, op, maskValue); err != nil {
 				return err
 			}
 			writeBuf = writeBuf[maxFragment:]
 			op = Continuation
 			continue
 		}
-		return frame.WriteFrame(&c.fw, c.c, writeBuf, true, rsv1, c.client, op, maskValue)
+		return frame.WriteFrame(&fw, c.c, writeBuf, true, rsv1, c.client, op, maskValue)
 	}
 	return nil
 }
@@ -478,12 +480,12 @@ func (c *Conn) Close() (err error) {
 	c.once.Do(func() {
 		c.bp.Free()
 		err = c.c.Close()
+		c.delayMu.Lock()
 		if c.delayTimeout != nil {
 			c.delayTimeout.Stop()
-			c.delayMu.Lock()
 			c.delayBuf = nil
-			c.delayMu.Unlock()
 		}
+		c.delayMu.Unlock()
 		atomic.StoreInt32(&c.closed, 1)
 	})
 	return
@@ -508,7 +510,10 @@ func (c *Conn) writerDelayBufInner() (err error) {
 	return
 }
 
-// 延迟写消息, 对流量密集型的场景有用 或者开启tcp delay， WithClientTCPDelay WithServerTCPDelay
+// 对于流量场景这个版本推荐开启tcp delay 方法：WithClientTCPDelay() WithServerTCPDelay()
+
+// 该函数目前是研究性质的尝试
+// 延迟写消息, 对流量密集型的场景有用 或者开启tcp delay，
 // 1. 如果缓存的消息超过了多少条数
 // 2. 如果缓存的消费超过了多久的时间
 // 3. TODO: 最大缓存多少字节

conn_test.go

@@ -25,6 +25,7 @@ import (
 	"testing"
 	"time"
 
+	"github.com/antlabs/wsutil/fixedwriter"
 	"github.com/antlabs/wsutil/frame"
 	"github.com/antlabs/wsutil/opcode"
 )
@@ -225,7 +226,8 @@ func Test_ReadMessage(t *testing.T) {
 
 		// err = con.WriteMessage(Binary, []byte("hello"))
 		maskValue := rand.Uint32()
-		err = frame.WriteFrame(&con.fw, con.c, []byte("hello"), true, true, con.client, Binary, maskValue)
+		var fw fixedwriter.FixedWriter
+		err = frame.WriteFrame(&fw, con.c, []byte("hello"), true, true, con.client, Binary, maskValue)
 		if err != nil {
 			t.Error(err)
 		}
@@ -276,7 +278,8 @@ func Test_ReadMessage(t *testing.T) {
 
 		// err = con.WriteMessage(Binary, []byte("hello"))
 		maskValue := rand.Uint32()
-		err = frame.WriteFrame(&con.fw, con.c, []byte("hello"), true, true, con.client, Ping, maskValue)
+		var fw fixedwriter.FixedWriter
+		err = frame.WriteFrame(&fw, con.c, []byte("hello"), true, true, con.client, Ping, maskValue)
 		if err != nil {
 			t.Error(err)
 		}
@@ -397,12 +400,13 @@ func TestFragmentFrame(t *testing.T) {
 		// con.writeFragment(Ping, []byte("hello"), 1)
 
 		maskValue := rand.Uint32()
-		err = frame.WriteFrame(&con.fw, con.c, []byte("h"), false, false, con.client, Text, maskValue)
+		var fw fixedwriter.FixedWriter
+		err = frame.WriteFrame(&fw, con.c, []byte("h"), false, false, con.client, Text, maskValue)
 		if err != nil {
 			t.Error(err)
 		}
 		maskValue = rand.Uint32()
-		err = frame.WriteFrame(&con.fw, con.c, []byte{}, true, false, con.client, Text, maskValue)
+		err = frame.WriteFrame(&fw, con.c, []byte{}, true, false, con.client, Text, maskValue)
 		if err != nil {
 			t.Error(err)
 		}

upgrade.go

@@ -24,7 +24,6 @@ import (
 	"github.com/antlabs/wsutil/bufio2"
 	"github.com/antlabs/wsutil/bytespool"
 	"github.com/antlabs/wsutil/fixedreader"
-	"github.com/antlabs/wsutil/rsp"
 )
 
 type UpgradeServer struct {
@@ -73,7 +72,8 @@ func upgradeInner(w http.ResponseWriter, r *http.Request, conf *Config) (c *Conn
 		if !conf.disableBufioClearHack {
 			bufio2.ClearReadWriter(rw)
 		}
-		rsp.ClearRsp(w)
+		// TODO
+		// rsp.ClearRsp(w)
 		rw = nil
 	} else {
 		var rw *bufio.ReadWriter