Browse Source

Revert "abci: change client to use multi-reader mutexes (#6306)" (backport #7106) (#7109)

pull/7205/head
mergify[bot] 3 years ago
committed by GitHub
parent
commit
a82cb7dcda
No known key found for this signature in database GPG Key ID: 4AEE18F83AFDEB23
9 changed files with 31 additions and 32 deletions
  1. +2
    -1
      CHANGELOG_PENDING.md
  2. +4
    -4
      abci/client/client.go
  3. +4
    -4
      abci/client/grpc_client.go
  4. +12
    -14
      abci/client/local_client.go
  5. +4
    -4
      abci/client/socket_client.go
  6. +1
    -1
      consensus/byzantine_test.go
  7. +1
    -1
      consensus/common_test.go
  8. +1
    -1
      consensus/reactor_test.go
  9. +2
    -2
      proxy/client.go

+ 2
- 1
CHANGELOG_PENDING.md View File

@ -27,4 +27,5 @@ Friendly reminder, we have a [bug bounty program](https://hackerone.com/tendermi
### BUG FIXES
- [\#7057](https://github.com/tendermint/tendermint/pull/7057) Import Postgres driver support for the psql indexer (@creachadair).
- [\#7057](https://github.com/tendermint/tendermint/pull/7057) Import Postgres driver support for the psql indexer (@creachadair).
- [\#7106](https://github.com/tendermint/tendermint/pull/7106) Revert mutex change to ABCI Clients (@tychoish).

+ 4
- 4
abci/client/client.go View File

@ -81,7 +81,7 @@ type ReqRes struct {
*sync.WaitGroup
*types.Response // Not set atomically, so be sure to use WaitGroup.
mtx tmsync.RWMutex
mtx tmsync.Mutex
done bool // Gets set to true once *after* WaitGroup.Done().
cb func(*types.Response) // A single callback that may be set.
}
@ -131,16 +131,16 @@ func (r *ReqRes) InvokeCallback() {
//
// ref: https://github.com/tendermint/tendermint/issues/5439
func (r *ReqRes) GetCallback() func(*types.Response) {
r.mtx.RLock()
defer r.mtx.RUnlock()
r.mtx.Lock()
defer r.mtx.Unlock()
return r.cb
}
// SetDone marks the ReqRes object as done.
func (r *ReqRes) SetDone() {
r.mtx.Lock()
defer r.mtx.Unlock()
r.done = true
r.mtx.Unlock()
}
func waitGroup1() (wg *sync.WaitGroup) {


+ 4
- 4
abci/client/grpc_client.go View File

@ -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.RWMutex
mtx tmsync.Mutex
addr string
err error
resCb func(*types.Request, *types.Response) // listens to all callbacks
@ -146,8 +146,8 @@ func (cli *grpcClient) StopForError(err error) {
}
func (cli *grpcClient) Error() error {
cli.mtx.RLock()
defer cli.mtx.RUnlock()
cli.mtx.Lock()
defer cli.mtx.Unlock()
return cli.err
}
@ -155,8 +155,8 @@ func (cli *grpcClient) Error() error {
// 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()
}
//----------------------------------------


+ 12
- 14
abci/client/local_client.go View File

@ -15,7 +15,7 @@ var _ Client = (*localClient)(nil)
type localClient struct {
service.BaseService
mtx *tmsync.RWMutex
mtx *tmsync.Mutex
types.Application
Callback
}
@ -26,24 +26,22 @@ var _ Client = (*localClient)(nil)
// methods of the given app.
//
// Both Async and Sync methods ignore the given context.Context parameter.
func NewLocalClient(mtx *tmsync.RWMutex, app types.Application) Client {
func NewLocalClient(mtx *tmsync.Mutex, app types.Application) Client {
if mtx == nil {
mtx = &tmsync.RWMutex{}
mtx = new(tmsync.Mutex)
}
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()?
@ -67,8 +65,8 @@ func (app *localClient) EchoAsync(msg string) *ReqRes {
}
func (app *localClient) InfoAsync(req types.RequestInfo) *ReqRes {
app.mtx.RLock()
defer app.mtx.RUnlock()
app.mtx.Lock()
defer app.mtx.Unlock()
res := app.Application.Info(req)
return app.callback(
@ -111,8 +109,8 @@ func (app *localClient) CheckTxAsync(req types.RequestCheckTx) *ReqRes {
}
func (app *localClient) QueryAsync(req types.RequestQuery) *ReqRes {
app.mtx.RLock()
defer app.mtx.RUnlock()
app.mtx.Lock()
defer app.mtx.Unlock()
res := app.Application.Query(req)
return app.callback(
@ -220,8 +218,8 @@ func (app *localClient) EchoSync(msg string) (*types.ResponseEcho, error) {
}
func (app *localClient) InfoSync(req types.RequestInfo) (*types.ResponseInfo, error) {
app.mtx.RLock()
defer app.mtx.RUnlock()
app.mtx.Lock()
defer app.mtx.Unlock()
res := app.Application.Info(req)
return &res, nil
@ -252,8 +250,8 @@ func (app *localClient) CheckTxSync(req types.RequestCheckTx) (*types.ResponseCh
}
func (app *localClient) QuerySync(req types.RequestQuery) (*types.ResponseQuery, error) {
app.mtx.RLock()
defer app.mtx.RUnlock()
app.mtx.Lock()
defer app.mtx.Unlock()
res := app.Application.Query(req)
return &res, nil


+ 4
- 4
abci/client/socket_client.go View File

@ -34,7 +34,7 @@ type socketClient struct {
reqQueue chan *ReqRes
flushTimer *timer.ThrottleTimer
mtx tmsync.RWMutex
mtx tmsync.Mutex
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.RLock()
defer cli.mtx.RUnlock()
cli.mtx.Lock()
defer cli.mtx.Unlock()
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()
}
//----------------------------------------


+ 1
- 1
consensus/byzantine_test.go View File

@ -59,7 +59,7 @@ func TestByzantinePrevoteEquivocation(t *testing.T) {
blockStore := store.NewBlockStore(blockDB)
// one for mempool, one for consensus
mtx := new(tmsync.RWMutex)
mtx := new(tmsync.Mutex)
proxyAppConnMem := abcicli.NewLocalClient(mtx, app)
proxyAppConnCon := abcicli.NewLocalClient(mtx, app)


+ 1
- 1
consensus/common_test.go View File

@ -389,7 +389,7 @@ func newStateWithConfigAndBlockStore(
blockStore := store.NewBlockStore(blockDB)
// one for mempool, one for consensus
mtx := new(tmsync.RWMutex)
mtx := new(tmsync.Mutex)
proxyAppConnMem := abcicli.NewLocalClient(mtx, app)
proxyAppConnCon := abcicli.NewLocalClient(mtx, app)


+ 1
- 1
consensus/reactor_test.go View File

@ -153,7 +153,7 @@ func TestReactorWithEvidence(t *testing.T) {
blockStore := store.NewBlockStore(blockDB)
// one for mempool, one for consensus
mtx := new(tmsync.RWMutex)
mtx := new(tmsync.Mutex)
proxyAppConnMem := abcicli.NewLocalClient(mtx, app)
proxyAppConnCon := abcicli.NewLocalClient(mtx, app)


+ 2
- 2
proxy/client.go View File

@ -21,7 +21,7 @@ type ClientCreator interface {
// local proxy uses a mutex on an in-proc app
type localClientCreator struct {
mtx *tmsync.RWMutex
mtx *tmsync.Mutex
app types.Application
}
@ -29,7 +29,7 @@ type localClientCreator struct {
// which will be running locally.
func NewLocalClientCreator(app types.Application) ClientCreator {
return &localClientCreator{
mtx: new(tmsync.RWMutex),
mtx: new(tmsync.Mutex),
app: app,
}
}


Loading…
Cancel
Save