diff --git a/abci/client/grpc_client.go b/abci/client/grpc_client.go index 251939253..5890d5f05 100644 --- a/abci/client/grpc_client.go +++ b/abci/client/grpc_client.go @@ -52,6 +52,9 @@ func dialerFunc(ctx context.Context, addr string) (net.Conn, error) { } func (cli *grpcClient) OnStart(ctx context.Context) error { + timer := time.NewTimer(0) + defer timer.Stop() + RETRY_LOOP: for { conn, err := grpc.Dial(cli.addr, @@ -63,8 +66,13 @@ RETRY_LOOP: return err } cli.logger.Error(fmt.Sprintf("abci.grpcClient failed to connect to %v. Retrying...\n", cli.addr), "err", err) - time.Sleep(time.Second * dialRetryIntervalSeconds) - continue RETRY_LOOP + timer.Reset(time.Second * dialRetryIntervalSeconds) + select { + case <-ctx.Done(): + return ctx.Err() + case <-timer.C: + continue RETRY_LOOP + } } cli.logger.Info("Dialed server. Waiting for echo.", "addr", cli.addr) @@ -82,7 +90,13 @@ RETRY_LOOP: } cli.logger.Error("Echo failed", "err", err) - time.Sleep(time.Second * echoRetryIntervalSeconds) + timer.Reset(time.Second * echoRetryIntervalSeconds) + select { + case <-ctx.Done(): + return ctx.Err() + case <-timer.C: + continue ENSURE_CONNECTED + } } cli.client = client diff --git a/abci/client/socket_client.go b/abci/client/socket_client.go index dcf5fa519..b06fb6718 100644 --- a/abci/client/socket_client.go +++ b/abci/client/socket_client.go @@ -64,6 +64,8 @@ func (cli *socketClient) OnStart(ctx context.Context) error { err error conn net.Conn ) + timer := time.NewTimer(0) + defer timer.Stop() for { conn, err = tmnet.Connect(cli.addr) @@ -73,8 +75,15 @@ func (cli *socketClient) OnStart(ctx context.Context) error { } cli.logger.Error(fmt.Sprintf("abci.socketClient failed to connect to %v. Retrying after %vs...", cli.addr, dialRetryIntervalSeconds), "err", err) - time.Sleep(time.Second * dialRetryIntervalSeconds) - continue + + timer.Reset(time.Second * dialRetryIntervalSeconds) + select { + case <-ctx.Done(): + return ctx.Err() + case <-timer.C: + continue + } + } cli.conn = conn @@ -90,11 +99,7 @@ func (cli *socketClient) OnStop() { if cli.conn != nil { cli.conn.Close() } - - // this timeout is arbitrary. - ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) - defer cancel() - cli.drainQueue(ctx) + cli.drainQueue() } // Error returns an error if the client was stopped abruptly. @@ -113,12 +118,6 @@ func (cli *socketClient) sendRequestsRoutine(ctx context.Context, conn io.Writer case <-ctx.Done(): return case reqres := <-cli.reqQueue: - if ctx.Err() != nil { - return - } - - cli.willSendReq(reqres) - if err := types.WriteMessage(reqres.Request, bw); err != nil { cli.stopForError(fmt.Errorf("write to buffer: %w", err)) return @@ -162,6 +161,11 @@ func (cli *socketClient) recvResponseRoutine(ctx context.Context, conn io.Reader func (cli *socketClient) willSendReq(reqres *requestAndResponse) { cli.mtx.Lock() defer cli.mtx.Unlock() + + if !cli.IsRunning() { + return + } + cli.reqSent.PushBack(reqres) } @@ -189,6 +193,47 @@ func (cli *socketClient) didRecvResponse(res *types.Response) error { //---------------------------------------- +func (cli *socketClient) doRequest(ctx context.Context, req *types.Request) (*types.Response, error) { + if !cli.IsRunning() { + return nil, errors.New("client has stopped") + } + + reqres := makeReqRes(req) + cli.willSendReq(reqres) + + select { + case cli.reqQueue <- reqres: + case <-ctx.Done(): + return nil, fmt.Errorf("can't queue req: %w", ctx.Err()) + } + + select { + case <-reqres.signal: + if err := cli.Error(); err != nil { + return nil, err + } + + return reqres.Response, nil + case <-ctx.Done(): + return nil, ctx.Err() + } +} + +// drainQueue marks as complete and discards all remaining pending requests +// from the queue. +func (cli *socketClient) drainQueue() { + cli.mtx.Lock() + defer cli.mtx.Unlock() + + // mark all in-flight messages as resolved (they will get cli.Error()) + for req := cli.reqSent.Front(); req != nil; req = req.Next() { + reqres := req.Value.(*requestAndResponse) + reqres.markDone() + } +} + +//---------------------------------------- + func (cli *socketClient) Flush(ctx context.Context) error { _, err := cli.doRequest(ctx, types.ToRequestFlush()) if err != nil { @@ -319,58 +364,6 @@ func (cli *socketClient) FinalizeBlock(ctx context.Context, req types.RequestFin //---------------------------------------- -func (cli *socketClient) doRequest(ctx context.Context, req *types.Request) (*types.Response, error) { - reqres := makeReqRes(req) - - select { - case cli.reqQueue <- reqres: - case <-ctx.Done(): - return nil, fmt.Errorf("can't queue req: %w", ctx.Err()) - } - - select { - case <-reqres.signal: - if err := cli.Error(); err != nil { - return nil, err - } - - return reqres.Response, nil - case <-ctx.Done(): - return nil, ctx.Err() - } -} - -// drainQueue marks as complete and discards all remaining pending requests -// from the queue. -func (cli *socketClient) drainQueue(ctx context.Context) { - cli.mtx.Lock() - defer cli.mtx.Unlock() - - // mark all in-flight messages as resolved (they will get cli.Error()) - for req := cli.reqSent.Front(); req != nil; req = req.Next() { - reqres := req.Value.(*requestAndResponse) - reqres.markDone() - } - - // Mark all queued messages as resolved. - // - // TODO(creachadair): We can't simply range the channel, because it is never - // closed, and the writer continues to add work. - // See https://github.com/tendermint/tendermint/issues/6996. - for { - select { - case <-ctx.Done(): - return - case reqres := <-cli.reqQueue: - reqres.markDone() - default: - return - } - } -} - -//---------------------------------------- - func resMatchesReq(req *types.Request, res *types.Response) (ok bool) { switch req.Value.(type) { case *types.Request_Echo: