From b4bf74ba9c367b5b640f7da14eb40bf2dad24a5e Mon Sep 17 00:00:00 2001 From: Sam Kleinman Date: Thu, 19 May 2022 18:16:34 +0200 Subject: [PATCH 1/4] abci: serialize semantics of abci client (#8578) Prior to this change, it was possible that two client calls could enqueue their requests in the response queue in a different order than they were processed by the sender goroutine. This violates the requirement that responses must be delivered in the same order they were enqueued. To avert this, make the sender goroutine responsible for enqueuing. Also, remove an unnecessary channel buffer. --- abci/client/socket_client.go | 13 ++++--------- 1 file changed, 4 insertions(+), 9 deletions(-) diff --git a/abci/client/socket_client.go b/abci/client/socket_client.go index aa4fdcbe9..8904d557d 100644 --- a/abci/client/socket_client.go +++ b/abci/client/socket_client.go @@ -17,12 +17,6 @@ import ( "github.com/tendermint/tendermint/libs/service" ) -const ( - // reqQueueSize is the max number of queued async requests. - // (memory: 256MB max assuming 1MB transactions) - reqQueueSize = 256 -) - // This is goroutine-safe, but users should beware that the application in // general is not meant to be interfaced with concurrent callers. type socketClient struct { @@ -48,7 +42,7 @@ var _ Client = (*socketClient)(nil) func NewSocketClient(logger log.Logger, addr string, mustConnect bool) Client { cli := &socketClient{ logger: logger, - reqQueue: make(chan *requestAndResponse, reqQueueSize), + reqQueue: make(chan *requestAndResponse), mustConnect: mustConnect, addr: addr, reqSent: list.New(), @@ -127,6 +121,8 @@ func (cli *socketClient) sendRequestsRoutine(ctx context.Context, conn io.Writer cli.stopForError(fmt.Errorf("flush buffer: %w", err)) return } + + cli.trackRequest(reqres) } } } @@ -158,7 +154,7 @@ func (cli *socketClient) recvResponseRoutine(ctx context.Context, conn io.Reader } } -func (cli *socketClient) willSendReq(reqres *requestAndResponse) { +func (cli *socketClient) trackRequest(reqres *requestAndResponse) { cli.mtx.Lock() defer cli.mtx.Unlock() @@ -199,7 +195,6 @@ func (cli *socketClient) doRequest(ctx context.Context, req *types.Request) (*ty } reqres := makeReqRes(req) - cli.willSendReq(reqres) select { case cli.reqQueue <- reqres: From 4a9bbe047f8dc762e9020b528a019899c781ddcc Mon Sep 17 00:00:00 2001 From: "M. J. Fromberger" Date: Thu, 19 May 2022 12:11:57 -0700 Subject: [PATCH 2/4] Fix lock sequencing in socket client request tracking. (#8581) * Fix lock sequencing in socket client request tracking. It is not safe to check base service state (IsRunning) while holding the lock for the client state. If we do, then during shutdown we may deadlock with the invocation of the OnStop handler, which the base service executes while holding the service lock. * Enqueue pending requests before sending them to the server. If we don't do this, the server can reply before the request lands in the queue. That will cause the receiver to terminate early for an unsolicited response. So enqueue first: This is safe because we're doing it in the same routine as services the channel, so we won't take another message till we are safely past that point. * Document what we did. * Fix socket paths in tests. --- abci/client/socket_client.go | 14 +++++++++----- internal/proxy/client_test.go | 6 +++--- 2 files changed, 12 insertions(+), 8 deletions(-) diff --git a/abci/client/socket_client.go b/abci/client/socket_client.go index 8904d557d..7dfcf76cc 100644 --- a/abci/client/socket_client.go +++ b/abci/client/socket_client.go @@ -112,6 +112,11 @@ func (cli *socketClient) sendRequestsRoutine(ctx context.Context, conn io.Writer case <-ctx.Done(): return case reqres := <-cli.reqQueue: + // N.B. We must enqueue before sending out the request, otherwise the + // server may reply before we do it, and the receiver will fail for an + // unsolicited reply. + cli.trackRequest(reqres) + if err := types.WriteMessage(reqres.Request, bw); err != nil { cli.stopForError(fmt.Errorf("write to buffer: %w", err)) return @@ -121,8 +126,6 @@ func (cli *socketClient) sendRequestsRoutine(ctx context.Context, conn io.Writer cli.stopForError(fmt.Errorf("flush buffer: %w", err)) return } - - cli.trackRequest(reqres) } } } @@ -155,13 +158,14 @@ func (cli *socketClient) recvResponseRoutine(ctx context.Context, conn io.Reader } func (cli *socketClient) trackRequest(reqres *requestAndResponse) { - cli.mtx.Lock() - defer cli.mtx.Unlock() - + // N.B. We must NOT hold the client state lock while checking this, or we + // may deadlock with shutdown. if !cli.IsRunning() { return } + cli.mtx.Lock() + defer cli.mtx.Unlock() cli.reqSent.PushBack(reqres) } diff --git a/internal/proxy/client_test.go b/internal/proxy/client_test.go index 09ac3f2c8..41a34bde7 100644 --- a/internal/proxy/client_test.go +++ b/internal/proxy/client_test.go @@ -58,7 +58,7 @@ func (app *appConnTest) Info(ctx context.Context, req *types.RequestInfo) (*type var SOCKET = "socket" func TestEcho(t *testing.T) { - sockPath := fmt.Sprintf("unix:///tmp/echo_%v.sock", tmrand.Str(6)) + sockPath := fmt.Sprintf("unix://%s/echo_%v.sock", t.TempDir(), tmrand.Str(6)) logger := log.NewNopLogger() client, err := abciclient.NewClient(logger, sockPath, SOCKET, true) if err != nil { @@ -98,7 +98,7 @@ func TestEcho(t *testing.T) { func BenchmarkEcho(b *testing.B) { b.StopTimer() // Initialize - sockPath := fmt.Sprintf("unix:///tmp/echo_%v.sock", tmrand.Str(6)) + sockPath := fmt.Sprintf("unix://%s/echo_%v.sock", b.TempDir(), tmrand.Str(6)) logger := log.NewNopLogger() client, err := abciclient.NewClient(logger, sockPath, SOCKET, true) if err != nil { @@ -146,7 +146,7 @@ func TestInfo(t *testing.T) { ctx, cancel := context.WithCancel(context.Background()) defer cancel() - sockPath := fmt.Sprintf("unix:///tmp/echo_%v.sock", tmrand.Str(6)) + sockPath := fmt.Sprintf("unix://%s/echo_%v.sock", t.TempDir(), tmrand.Str(6)) logger := log.NewNopLogger() client, err := abciclient.NewClient(logger, sockPath, SOCKET, true) if err != nil { From ad73e6da2f3d8f4cb00a0c4649ba06b941b40ec9 Mon Sep 17 00:00:00 2001 From: William Banfield <4561443+williambanfield@users.noreply.github.com> Date: Thu, 19 May 2022 15:35:30 -0400 Subject: [PATCH 3/4] consensus: update state from store before use in reactor (#8576) Closes: #8575 This PR aims to fix the `LastCommitRound can only be negative for initial height 0` issue we see in the e2e tests by initializing the `state` object before starting the receive routines in the consensus reactor. This is somewhat inelegant, but it should fix the issue. --- internal/consensus/reactor.go | 2 ++ internal/consensus/state.go | 18 +++++++++++------- 2 files changed, 13 insertions(+), 7 deletions(-) diff --git a/internal/consensus/reactor.go b/internal/consensus/reactor.go index 1a9d49057..501523339 100644 --- a/internal/consensus/reactor.go +++ b/internal/consensus/reactor.go @@ -216,6 +216,8 @@ func (r *Reactor) OnStart(ctx context.Context) error { if err := r.state.Start(ctx); err != nil { return err } + } else if err := r.state.updateStateFromStore(); err != nil { + return err } go r.updateRoundStateRoutine(ctx) diff --git a/internal/consensus/state.go b/internal/consensus/state.go index b016e2687..320042d30 100644 --- a/internal/consensus/state.go +++ b/internal/consensus/state.go @@ -122,9 +122,8 @@ type State struct { // store blocks and commits blockStore sm.BlockStore - stateStore sm.Store - initialStatePopulated bool - skipBootstrapping bool + stateStore sm.Store + skipBootstrapping bool // create and execute blocks blockExec *sm.BlockExecutor @@ -248,9 +247,6 @@ func NewState( } func (cs *State) updateStateFromStore() error { - if cs.initialStatePopulated { - return nil - } state, err := cs.stateStore.Load() if err != nil { return fmt.Errorf("loading state: %w", err) @@ -259,6 +255,15 @@ func (cs *State) updateStateFromStore() error { return nil } + eq, err := state.Equals(cs.state) + if err != nil { + return fmt.Errorf("comparing state: %w", err) + } + // if the new state is equivalent to the old state, we should not trigger a state update. + if eq { + return nil + } + // We have no votes, so reconstruct LastCommit from SeenCommit. if state.LastBlockHeight > 0 { cs.reconstructLastCommit(state) @@ -266,7 +271,6 @@ func (cs *State) updateStateFromStore() error { cs.updateToState(state) - cs.initialStatePopulated = true return nil } From 4786a5ffded3a59865772242266b84c736cf01ff Mon Sep 17 00:00:00 2001 From: "M. J. Fromberger" Date: Thu, 19 May 2022 15:23:14 -0700 Subject: [PATCH 4/4] Remove the periodically-scheduled Markdown link check. (#8580) The state of links in our documentation is now sufficiently good that we are running link checks during PRs. There is no longer any practical benefit to running the scheduled "global" check. Most of the errors it reports are rate limitations anyway (429). --- .github/workflows/linkchecker.yml | 12 ------------ 1 file changed, 12 deletions(-) delete mode 100644 .github/workflows/linkchecker.yml diff --git a/.github/workflows/linkchecker.yml b/.github/workflows/linkchecker.yml deleted file mode 100644 index e2ba80861..000000000 --- a/.github/workflows/linkchecker.yml +++ /dev/null @@ -1,12 +0,0 @@ -name: Check Markdown links -on: - schedule: - - cron: '* */24 * * *' -jobs: - markdown-link-check: - runs-on: ubuntu-latest - steps: - - uses: actions/checkout@v3 - - uses: creachadair/github-action-markdown-link-check@master - with: - folder-path: "docs"