Revert revert abci: change client to use multi-reader mutexes (backport

tendermint#6306) (tendermint#6873)
scrtlabs · Nov 9, 2022 · bae62a7 · bae62a7
1 parent a6dd0d2
commit bae62a7
Show file tree

Hide file tree

Showing 37 changed files with 30 additions and 28 deletions.
diff --git a/DOCKER/build.sh b/DOCKER/build.sh
diff --git a/DOCKER/docker-entrypoint.sh b/DOCKER/docker-entrypoint.sh
diff --git a/DOCKER/push.sh b/DOCKER/push.sh
diff --git a/abci/client/client.go b/abci/client/client.go
@@ -81,7 +81,7 @@ type ReqRes struct {
 	*sync.WaitGroup
 	*types.Response // Not set atomically, so be sure to use WaitGroup.
 
-	mtx tmsync.Mutex
+	mtx tmsync.RWMutex
 
 	// callbackInvoked as a variable to track if the callback was already
 	// invoked during the regular execution of the request. This variable
@@ -138,8 +138,8 @@ func (r *ReqRes) InvokeCallback() {
 //
 // ref: https://github.com/tendermint/tendermint/issues/5439
 func (r *ReqRes) GetCallback() func(*types.Response) {
-	r.mtx.Lock()
-	defer r.mtx.Unlock()
+	r.mtx.RLock()
+	defer r.mtx.RUnlock()
 	return r.cb
 }
 

diff --git a/abci/client/grpc_client.go b/abci/client/grpc_client.go
@@ -27,7 +27,7 @@ type grpcClient struct {
 	conn     *grpc.ClientConn
 	chReqRes chan *ReqRes // dispatches "async" responses to callbacks *in order*, needed by mempool
 
-	mtx   tmsync.Mutex
+	mtx   tmsync.RWMutex
 	addr  string
 	err   error
 	resCb func(*types.Request, *types.Response) // listens to all callbacks
@@ -144,17 +144,17 @@ func (cli *grpcClient) StopForError(err error) {
 }
 
 func (cli *grpcClient) Error() error {
-	cli.mtx.Lock()
-	defer cli.mtx.Unlock()
+	cli.mtx.RLock()
+	defer cli.mtx.RUnlock()
 	return cli.err
 }
 
 // Set listener for all responses
 // NOTE: callback may get internally generated flush responses.
 func (cli *grpcClient) SetResponseCallback(resCb Callback) {
 	cli.mtx.Lock()
+	defer cli.mtx.Unlock()
 	cli.resCb = resCb
-	cli.mtx.Unlock()
 }
 
 //----------------------------------------

diff --git a/abci/client/local_client.go b/abci/client/local_client.go
@@ -15,7 +15,7 @@ var _ Client = (*localClient)(nil)
 type localClient struct {
 	service.BaseService
 
-	mtx *tmsync.Mutex
+	mtx *tmsync.RWMutex
 	types.Application
 	Callback
 }
@@ -26,22 +26,24 @@ var _ Client = (*localClient)(nil)
 // methods of the given app.
 //
 // Both Async and Sync methods ignore the given context.Context parameter.
-func NewLocalClient(mtx *tmsync.Mutex, app types.Application) Client {
+func NewLocalClient(mtx *tmsync.RWMutex, app types.Application) Client {
 	if mtx == nil {
-		mtx = new(tmsync.Mutex)
+		mtx = &tmsync.RWMutex{}
 	}
+
 	cli := &localClient{
 		mtx:         mtx,
 		Application: app,
 	}
+
 	cli.BaseService = *service.NewBaseService(nil, "localClient", cli)
 	return cli
 }
 
 func (app *localClient) SetResponseCallback(cb Callback) {
 	app.mtx.Lock()
+	defer app.mtx.Unlock()
 	app.Callback = cb
-	app.mtx.Unlock()
 }
 
 // TODO: change types.Application to include Error()?
@@ -65,8 +67,8 @@ func (app *localClient) EchoAsync(msg string) *ReqRes {
 }
 
 func (app *localClient) InfoAsync(req types.RequestInfo) *ReqRes {
-	app.mtx.Lock()
-	defer app.mtx.Unlock()
+	app.mtx.RLock()
+	defer app.mtx.RUnlock()
 
 	res := app.Application.Info(req)
 	return app.callback(
@@ -109,8 +111,8 @@ func (app *localClient) CheckTxAsync(req types.RequestCheckTx) *ReqRes {
 }
 
 func (app *localClient) QueryAsync(req types.RequestQuery) *ReqRes {
-	app.mtx.Lock()
-	defer app.mtx.Unlock()
+	app.mtx.RLock()
+	defer app.mtx.RUnlock()
 
 	res := app.Application.Query(req)
 	return app.callback(
@@ -218,8 +220,8 @@ func (app *localClient) EchoSync(msg string) (*types.ResponseEcho, error) {
 }
 
 func (app *localClient) InfoSync(req types.RequestInfo) (*types.ResponseInfo, error) {
-	app.mtx.Lock()
-	defer app.mtx.Unlock()
+	app.mtx.RLock()
+	defer app.mtx.RUnlock()
 
 	res := app.Application.Info(req)
 	return &res, nil
@@ -250,8 +252,8 @@ func (app *localClient) CheckTxSync(req types.RequestCheckTx) (*types.ResponseCh
 }
 
 func (app *localClient) QuerySync(req types.RequestQuery) (*types.ResponseQuery, error) {
-	app.mtx.Lock()
-	defer app.mtx.Unlock()
+	app.mtx.RLock()
+	defer app.mtx.RUnlock()
 
 	res := app.Application.Query(req)
 	return &res, nil

diff --git a/abci/client/socket_client.go b/abci/client/socket_client.go
@@ -34,7 +34,7 @@ type socketClient struct {
 	reqQueue   chan *ReqRes
 	flushTimer *timer.ThrottleTimer
 
-	mtx     tmsync.Mutex
+	mtx     tmsync.RWMutex
 	err     error
 	reqSent *list.List                            // list of requests sent, waiting for response
 	resCb   func(*types.Request, *types.Response) // called on all requests, if set.
@@ -99,8 +99,8 @@ func (cli *socketClient) OnStop() {
 
 // Error returns an error if the client was stopped abruptly.
 func (cli *socketClient) Error() error {
-	cli.mtx.Lock()
-	defer cli.mtx.Unlock()
+	cli.mtx.RLock()
+	defer cli.mtx.RUnlock()
 	return cli.err
 }
 
@@ -110,8 +110,8 @@ func (cli *socketClient) Error() error {
 // NOTE: callback may get internally generated flush responses.
 func (cli *socketClient) SetResponseCallback(resCb Callback) {
 	cli.mtx.Lock()
+	defer cli.mtx.Unlock()
 	cli.resCb = resCb
-	cli.mtx.Unlock()
 }
 
 //----------------------------------------

diff --git a/abci/tests/test_app/test.sh b/abci/tests/test_app/test.sh
diff --git a/abci/tests/test_cli/test.sh b/abci/tests/test_cli/test.sh
diff --git a/consensus/byzantine_test.go b/consensus/byzantine_test.go
@@ -64,7 +64,7 @@ func TestByzantinePrevoteEquivocation(t *testing.T) {
 		blockDB := dbm.NewMemDB()
 		blockStore := store.NewBlockStore(blockDB)
 
-		mtx := new(tmsync.Mutex)
+		mtx := new(tmsync.RWMutex)
 		// one for mempool, one for consensus
 		proxyAppConnCon := abcicli.NewLocalClient(mtx, app)
 		proxyAppConnConMem := abcicli.NewLocalClient(mtx, app)

diff --git a/consensus/common_test.go b/consensus/common_test.go
@@ -391,7 +391,7 @@ func newStateWithConfigAndBlockStore(
 	blockStore := store.NewBlockStore(blockDB)
 
 	// one for mempool, one for consensus
-	mtx := new(tmsync.Mutex)
+	mtx := new(tmsync.RWMutex)
 
 	proxyAppConnCon := abcicli.NewLocalClient(mtx, app)
 	proxyAppConnConMem := abcicli.NewLocalClient(mtx, app)

diff --git a/consensus/reactor_test.go b/consensus/reactor_test.go
@@ -156,7 +156,7 @@ func TestReactorWithEvidence(t *testing.T) {
 		blockDB := dbm.NewMemDB()
 		blockStore := store.NewBlockStore(blockDB)
 
-		mtx := new(tmsync.Mutex)
+		mtx := new(tmsync.RWMutex)
 		memplMetrics := mempl.NopMetrics()
 		// one for mempool, one for consensus
 		proxyAppConnCon := abcicli.NewLocalClient(mtx, app)

diff --git a/docs/post.sh b/docs/post.sh
diff --git a/docs/pre.sh b/docs/pre.sh
diff --git a/docs/tendermint-core-image.jpg b/docs/tendermint-core-image.jpg
diff --git a/networks/local/localnode/wrapper.sh b/networks/local/localnode/wrapper.sh
diff --git a/networks/remote/ansible/inventory/digital_ocean.py b/networks/remote/ansible/inventory/digital_ocean.py
diff --git a/proxy/client.go b/proxy/client.go
@@ -23,15 +23,15 @@ type ClientCreator interface {
 // local proxy uses a mutex on an in-proc app
 
 type localClientCreator struct {
-	mtx *tmsync.Mutex
+	mtx *tmsync.RWMutex
 	app types.Application
 }
 
 // NewLocalClientCreator returns a ClientCreator for the given app,
 // which will be running locally.
 func NewLocalClientCreator(app types.Application) ClientCreator {
 	return &localClientCreator{
-		mtx: new(tmsync.Mutex),
+		mtx: new(tmsync.RWMutex),
 		app: app,
 	}
 }

diff --git a/rpc/jsonrpc/test/integration_test.sh b/rpc/jsonrpc/test/integration_test.sh
diff --git a/scripts/authors.sh b/scripts/authors.sh
diff --git a/scripts/dist.sh b/scripts/dist.sh
diff --git a/scripts/get_nodejs.sh b/scripts/get_nodejs.sh
diff --git a/scripts/mockery_generate.sh b/scripts/mockery_generate.sh
diff --git a/scripts/proto-gen.sh b/scripts/proto-gen.sh
diff --git a/scripts/qa/reporting/latency_throughput.py b/scripts/qa/reporting/latency_throughput.py
diff --git a/spec/ivy-proofs/check_proofs.sh b/spec/ivy-proofs/check_proofs.sh
diff --git a/spec/ivy-proofs/count_lines.sh b/spec/ivy-proofs/count_lines.sh
diff --git a/spec/light-client/accountability/run.sh b/spec/light-client/accountability/run.sh
diff --git a/test/app/clean.sh b/test/app/clean.sh
diff --git a/test/app/counter_test.sh b/test/app/counter_test.sh
diff --git a/test/app/kvstore_test.sh b/test/app/kvstore_test.sh
diff --git a/test/app/test.sh b/test/app/test.sh
diff --git a/test/e2e/docker/entrypoint b/test/e2e/docker/entrypoint
diff --git a/test/e2e/docker/entrypoint-builtin b/test/e2e/docker/entrypoint-builtin
diff --git a/test/e2e/docker/entrypoint-maverick b/test/e2e/docker/entrypoint-maverick
diff --git a/test/e2e/run-multiple.sh b/test/e2e/run-multiple.sh
diff --git a/test/loadtime/basic.sh b/test/loadtime/basic.sh