Merge pull request #44 from filecoin-project/fix/client-reconnect

Fix client reconnect

Author: magik6k

client.go

@@ -32,7 +32,7 @@ var (
 
 	log = logging.Logger("rpc")
 
-	 _defaultHTTPClient = &http.Client{
+	_defaultHTTPClient = &http.Client{
 		Transport: &http.Transport{
 			Proxy: http.ProxyFromEnvironment,
 			DialContext: (&net.Dialer{

server.go

@@ -19,7 +19,7 @@ const (
 
 // RPCServer provides a jsonrpc 2.0 http server handler
 type RPCServer struct {
-	methods        map[string]rpcHandler
+	methods map[string]rpcHandler
 
 	// aliasedMethods contains a map of alias:original method names.
 	// These are used as fallbacks if a method is not found by the given method name.

websocket.go

@@ -51,6 +51,7 @@ type wsConn struct {
 	handler          *RPCServer
 	requests         <-chan clientRequest
 	pongs            chan struct{}
+	stopPings        func()
 	stop             <-chan struct{}
 	exiting          chan struct{}
 
@@ -511,6 +512,50 @@ func (c *wsConn) setupPings() func() {
 	}
 }
 
+// returns true if reconnected
+func (c *wsConn) tryReconnect(ctx context.Context) bool {
+	if c.connFactory == nil { // server side
+		return false
+	}
+
+	// connection dropped unexpectedly, do our best to recover it
+	c.closeInFlight()
+	c.closeChans()
+	c.incoming = make(chan io.Reader) // listen again for responses
+	go func() {
+		c.stopPings()
+
+		attempts := 0
+		var conn *websocket.Conn
+		for conn == nil {
+			time.Sleep(c.reconnectBackoff.next(attempts))
+			var err error
+			if conn, err = c.connFactory(); err != nil {
+				log.Debugw("websocket connection retry failed", "error", err)
+			}
+			select {
+			case <-ctx.Done():
+				break
+			default:
+				continue
+			}
+			attempts++
+		}
+
+		c.writeLk.Lock()
+		c.conn = conn
+		c.incomingErr = nil
+
+		c.stopPings = c.setupPings()
+
+		c.writeLk.Unlock()
+
+		go c.nextMessage()
+	}()
+
+	return true
+}
+
 func (c *wsConn) handleWsConn(ctx context.Context) {
 	c.incoming = make(chan io.Reader)
 	c.inflight = map[int64]clientRequest{}
@@ -530,8 +575,8 @@ func (c *wsConn) handleWsConn(ctx context.Context) {
 
 	// setup pings
 
-	stopPings := c.setupPings()
-	defer stopPings()
+	c.stopPings = c.setupPings()
+	defer c.stopPings()
 
 	var timeoutTimer *time.Timer
 	if c.timeout != 0 {
@@ -545,7 +590,10 @@ func (c *wsConn) handleWsConn(ctx context.Context) {
 		var timeoutCh <-chan time.Time
 		if timeoutTimer != nil {
 			if !timeoutTimer.Stop() {
-				<-timeoutTimer.C
+				select {
+				case <-timeoutTimer.C:
+				default:
+				}
 			}
 			timeoutTimer.Reset(c.timeout)
 
@@ -554,65 +602,30 @@ func (c *wsConn) handleWsConn(ctx context.Context) {
 
 		select {
 		case r, ok := <-c.incoming:
-			if !ok {
-				if c.incomingErr != nil {
-					if !websocket.IsCloseError(c.incomingErr, websocket.CloseNormalClosure) {
-						log.Debugw("websocket error", "error", c.incomingErr)
-						// connection dropped unexpectedly, do our best to recover it
-						c.closeInFlight()
-						c.closeChans()
-						c.incoming = make(chan io.Reader) // listen again for responses
-						go func() {
-							if c.connFactory == nil { // likely the server side, don't try to reconnect
-								return
-							}
-
-							stopPings()
-
-							attempts := 0
-							var conn *websocket.Conn
-							for conn == nil {
-								time.Sleep(c.reconnectBackoff.next(attempts))
-								var err error
-								if conn, err = c.connFactory(); err != nil {
-									log.Debugw("websocket connection retry failed", "error", err)
-								}
-								select {
-								case <-ctx.Done():
-									break
-								default:
-									continue
-								}
-								attempts++
-							}
-
-							c.writeLk.Lock()
-							c.conn = conn
-							c.incomingErr = nil
-
-							stopPings = c.setupPings()
-
-							c.writeLk.Unlock()
-
-							go c.nextMessage()
-						}()
-						continue
-					}
+			err := c.incomingErr
+
+			if ok {
+				// debug util - dump all messages to stderr
+				// r = io.TeeReader(r, os.Stderr)
+
+				var frame frame
+				err = json.NewDecoder(r).Decode(&frame)
+				if err == nil {
+					c.handleFrame(ctx, frame)
+					go c.nextMessage()
+					continue
 				}
-				return // remote closed
 			}
 
-			// debug util - dump all messages to stderr
-			// r = io.TeeReader(r, os.Stderr)
-
-			var frame frame
-			if err := json.NewDecoder(r).Decode(&frame); err != nil {
-				log.Error("handle me:", err)
-				return
+			if err == nil {
+				return // remote closed
 			}
 
-			c.handleFrame(ctx, frame)
-			go c.nextMessage()
+			log.Errorw("websocket error", "error", err)
+			// only client needs to reconnect
+			if !c.tryReconnect(ctx) {
+				return // failed to reconnect
+			}
 		case req := <-c.requests:
 			c.writeLk.Lock()
 			if req.req.ID != nil {
@@ -652,7 +665,12 @@ func (c *wsConn) handleWsConn(ctx context.Context) {
 			}
 			c.writeLk.Unlock()
 			log.Errorw("Connection timeout", "remote", c.conn.RemoteAddr())
-			return
+			// The server side does not perform the reconnect operation, so need to exit
+			if c.connFactory == nil {
+				return
+			}
+			// The client performs the reconnect operation, and if it exits it cannot start a handleWsConn again, so it does not need to exit
+			continue
 		case <-c.stop:
 			c.writeLk.Lock()
 			cmsg := websocket.FormatCloseMessage(websocket.CloseNormalClosure, "")