Merge pull request #101 from filecoin-project/fix/races

 Fix a number of races in websocket channel code paths

Author: magik6k

.circleci/config.yml

@@ -12,6 +12,11 @@ jobs:
     steps:
       - checkout
       - run: go test -v -timeout 10m ./...
+  test-race:
+    executor: golang
+    steps:
+      - checkout
+      - run: go test -race -v -timeout 10m ./...
   mod-tidy-check:
     executor: golang
     steps:
@@ -42,3 +47,4 @@ workflows:
       - lint-check
       - gofmt-check
       - test
+      - test-race

handler.go

@@ -239,7 +239,7 @@ func (s *handler) handleReader(ctx context.Context, r io.Reader, w io.Writer, rp
 			return
 		}
 
-		w.Write([]byte("["))
+		_, _ = w.Write([]byte("[")) // todo consider handling this error
 		for idx, req := range reqs {
 			if req.ID, err = normalizeID(req.ID); err != nil {
 				rpcError(wf, &req, rpcParseError, xerrors.Errorf("failed to parse ID: %w", err))
@@ -249,10 +249,10 @@ func (s *handler) handleReader(ctx context.Context, r io.Reader, w io.Writer, rp
 			s.handle(ctx, req, wf, rpcError, func(bool) {}, nil)
 
 			if idx != len(reqs)-1 {
-				w.Write([]byte(","))
+				_, _ = w.Write([]byte(",")) // todo consider handling this error
 			}
 		}
-		w.Write([]byte("]"))
+		_, _ = w.Write([]byte("]")) // todo consider handling this error
 	} else {
 		var req request
 		if err := json.NewDecoder(bufferedRequest).Decode(&req); err != nil {

rpc_test.go

@@ -160,16 +160,16 @@ func TestReconnection(t *testing.T) {
 	timer := time.NewTimer(captureDuration)
 
 	// record the number of connection attempts during this test
-	connectionAttempts := 1
+	connectionAttempts := int64(1)
 
 	closer, err := NewMergeClient(context.Background(), "ws://"+testServ.Listener.Addr().String(), "SimpleServerHandler", []interface{}{&rpcClient}, nil, func(c *Config) {
 		c.proxyConnFactory = func(f func() (*websocket.Conn, error)) func() (*websocket.Conn, error) {
 			return func() (*websocket.Conn, error) {
 				defer func() {
-					connectionAttempts++
+					atomic.AddInt64(&connectionAttempts, 1)
 				}()
 
-				if connectionAttempts > 1 {
+				if atomic.LoadInt64(&connectionAttempts) > 1 {
 					return nil, errors.New("simulates a failed reconnect attempt")
 				}
 
@@ -192,7 +192,7 @@ func TestReconnection(t *testing.T) {
 	<-timer.C
 
 	// do some math
-	attemptsPerSecond := int64(connectionAttempts) / int64(captureDuration/time.Second)
+	attemptsPerSecond := atomic.LoadInt64(&connectionAttempts) / int64(captureDuration/time.Second)
 
 	assert.Less(t, attemptsPerSecond, int64(50))
 }
@@ -677,7 +677,7 @@ func (h *ChanHandler) Sub(ctx context.Context, i int, eq int) (<-chan int, error
 				fmt.Println("ctxdone1", i, eq)
 				return
 			case <-wait:
-				fmt.Println("CONSUMED WAIT: ", i)
+				//fmt.Println("CONSUMED WAIT: ", i)
 			}
 
 			n += i
@@ -835,10 +835,11 @@ func TestChanServerClose(t *testing.T) {
 
 	tctx, tcancel := context.WithCancel(context.Background())
 
-	testServ := httptest.NewServer(rpcServer)
+	testServ := httptest.NewUnstartedServer(rpcServer)
 	testServ.Config.ConnContext = func(ctx context.Context, c net.Conn) context.Context {
 		return tctx
 	}
+	testServ.Start()
 
 	closer, err := NewClient(context.Background(), "ws://"+testServ.Listener.Addr().String(), "ChanHandler", &client, nil)
 	require.NoError(t, err)
@@ -966,10 +967,11 @@ func TestChanClientReceiveAll(t *testing.T) {
 
 	tctx, tcancel := context.WithCancel(context.Background())
 
-	testServ := httptest.NewServer(rpcServer)
+	testServ := httptest.NewUnstartedServer(rpcServer)
 	testServ.Config.ConnContext = func(ctx context.Context, c net.Conn) context.Context {
 		return tctx
 	}
+	testServ.Start()
 
 	closer, err := NewClient(context.Background(), "ws://"+testServ.Listener.Addr().String(), "ChanHandler", &client, nil)
 	require.NoError(t, err)
@@ -1005,6 +1007,11 @@ func TestChanClientReceiveAll(t *testing.T) {
 }
 
 func TestControlChanDeadlock(t *testing.T) {
+	_ = logging.SetLogLevel("rpc", "error")
+	defer func() {
+		_ = logging.SetLogLevel("rpc", "debug")
+	}()
+
 	for r := 0; r < 20; r++ {
 		testControlChanDeadlock(t)
 	}

websocket.go

@@ -82,7 +82,8 @@ type wsConn struct {
 	inflightLk sync.Mutex
 
 	// chanHandlers is a map of client-side channel handlers
-	chanHandlers map[uint64]func(m []byte, ok bool)
+	chanHandlersLk sync.Mutex
+	chanHandlers   map[uint64]*chanHandler
 
 	// ////
 	// Server related
@@ -99,6 +100,13 @@ type wsConn struct {
 	registerCh chan outChanReg
 }
 
+type chanHandler struct {
+	// take inside chanHandlersLk
+	lk sync.Mutex
+
+	cb func(m []byte, ok bool)
+}
+
 //                         //
 // WebSocket Message utils //
 //                         //
@@ -367,13 +375,20 @@ func (c *wsConn) handleChanMessage(frame frame) {
 		return
 	}
 
+	c.chanHandlersLk.Lock()
 	hnd, ok := c.chanHandlers[chid]
 	if !ok {
+		c.chanHandlersLk.Unlock()
 		log.Errorf("xrpc.ch.val: handler %d not found", chid)
 		return
 	}
 
-	hnd(params[1].data, true)
+	hnd.lk.Lock()
+	defer hnd.lk.Unlock()
+
+	c.chanHandlersLk.Unlock()
+
+	hnd.cb(params[1].data, true)
 }
 
 func (c *wsConn) handleChanClose(frame frame) {
@@ -389,15 +404,22 @@ func (c *wsConn) handleChanClose(frame frame) {
 		return
 	}
 
+	c.chanHandlersLk.Lock()
 	hnd, ok := c.chanHandlers[chid]
 	if !ok {
+		c.chanHandlersLk.Unlock()
 		log.Errorf("xrpc.ch.val: handler %d not found", chid)
 		return
 	}
 
+	hnd.lk.Lock()
+	defer hnd.lk.Unlock()
+
 	delete(c.chanHandlers, chid)
 
-	hnd(nil, false)
+	c.chanHandlersLk.Unlock()
+
+	hnd.cb(nil, false)
 }
 
 func (c *wsConn) handleResponse(frame frame) {
@@ -417,8 +439,12 @@ func (c *wsConn) handleResponse(frame frame) {
 			return
 		}
 
-		var chanCtx context.Context
-		chanCtx, c.chanHandlers[chid] = req.retCh()
+		chanCtx, chHnd := req.retCh()
+
+		c.chanHandlersLk.Lock()
+		c.chanHandlers[chid] = &chanHandler{cb: chHnd}
+		c.chanHandlersLk.Unlock()
+
 		go c.handleCtxAsync(chanCtx, frame.ID)
 	}
 
@@ -517,16 +543,28 @@ func (c *wsConn) closeInFlight() {
 	for _, cancel := range c.handling {
 		cancel()
 	}
+	c.handling = map[interface{}]context.CancelFunc{}
 	c.handlingLk.Unlock()
 
-	c.handling = map[interface{}]context.CancelFunc{}
 }
 
 func (c *wsConn) closeChans() {
+	c.chanHandlersLk.Lock()
+	defer c.chanHandlersLk.Unlock()
+
 	for chid := range c.chanHandlers {
 		hnd := c.chanHandlers[chid]
+
+		hnd.lk.Lock()
+
 		delete(c.chanHandlers, chid)
-		hnd(nil, false)
+
+		c.chanHandlersLk.Unlock()
+
+		hnd.cb(nil, false)
+
+		hnd.lk.Unlock()
+		c.chanHandlersLk.Lock()
 	}
 }
 
@@ -679,7 +717,7 @@ func (c *wsConn) handleWsConn(ctx context.Context) {
 	c.frameExecQueue = make(chan []byte, maxQueuedFrames)
 	c.inflight = map[interface{}]clientRequest{}
 	c.handling = map[interface{}]context.CancelFunc{}
-	c.chanHandlers = map[uint64]func(m []byte, ok bool){}
+	c.chanHandlers = map[uint64]*chanHandler{}
 	c.pongs = make(chan struct{}, 1)
 
 	c.registerCh = make(chan outChanReg)