abci: Add relaxed local client synchronization models (backport #1141)

.changelog/unreleased/breaking-changes/1141-abci-advanced-proxy.md

@@ -0,0 +1,3 @@
+- `[proxy]` Expand `ClientCreator` interface to allow
+  for per-"connection" control of client creation
+  ([\#1141](https://github.com/cometbft/cometbft/pull/1141))

.changelog/v0.38.0/improvements/1141-abci-consensus-syncd-proxy.md

@@ -0,0 +1,4 @@
+- `[abci/client]` Add consensus-synchronized local client creator,
+  which only imposes a mutex on the consensus "connection", leaving
+  the concurrency of all other "connections" up to the application
+  ([\#1141](https://github.com/cometbft/cometbft/pull/1141))

.changelog/v0.38.0/improvements/1141-abci-unsync-proxy.md

@@ -0,0 +1,3 @@
+- `[abci/client]` Add fully unsynchronized local client creator, which
+  imposes no mutexes on the application, leaving all handling of concurrency up
+  to the application ([\#1141](https://github.com/cometbft/cometbft/pull/1141))

abci/client/local_client.go

@@ -22,10 +22,12 @@ type localClient struct {
 
 var _ Client = (*localClient)(nil)
 
-// NewLocalClient creates a local client, which wraps the application interface that
-// Tendermint as the client will call to the application as the server. The only
-// difference, is that the local client has a global mutex which enforces serialization
-// of all the ABCI calls from Tendermint to the Application.
+// NewLocalClient creates a local client, which wraps the application interface
+// that Comet as the client will call to the application as the server.
+//
+// Concurrency control in each client instance is enforced by way of a single
+// mutex. If a mutex is not supplied (i.e. if mtx is nil), then one will be
+// created.
 func NewLocalClient(mtx *cmtsync.Mutex, app types.Application) Client {
 	if mtx == nil {
 		mtx = new(cmtsync.Mutex)
@@ -135,15 +137,17 @@ func (app *localClient) OfferSnapshot(ctx context.Context, req *types.RequestOff
 }
 
 func (app *localClient) LoadSnapshotChunk(ctx context.Context,
-	req *types.RequestLoadSnapshotChunk) (*types.ResponseLoadSnapshotChunk, error) {
+	req *types.RequestLoadSnapshotChunk,
+) (*types.ResponseLoadSnapshotChunk, error) {
 	app.mtx.Lock()
 	defer app.mtx.Unlock()
 
 	return app.Application.LoadSnapshotChunk(ctx, req)
 }
 
 func (app *localClient) ApplySnapshotChunk(ctx context.Context,
-	req *types.RequestApplySnapshotChunk) (*types.ResponseApplySnapshotChunk, error) {
+	req *types.RequestApplySnapshotChunk,
+) (*types.ResponseApplySnapshotChunk, error) {
 	app.mtx.Lock()
 	defer app.mtx.Unlock()
 

abci/client/unsync_local_client.go

@@ -0,0 +1,136 @@
+package abcicli
+
+import (
+	"context"
+	"sync"
+
+	types "github.com/cometbft/cometbft/abci/types"
+	"github.com/cometbft/cometbft/libs/service"
+)
+
+type unsyncLocalClient struct {
+	service.BaseService
+
+	types.Application
+
+	mtx sync.Mutex
+	Callback
+}
+
+var _ Client = (*unsyncLocalClient)(nil)
+
+// NewUnsyncLocalClient creates a local client, which wraps the application
+// interface that Comet as the client will call to the application as the
+// server.
+//
+// This differs from [NewLocalClient] in that it returns a client that only
+// maintains a mutex over the callback used by CheckTxAsync and not over the
+// application, leaving it up to the proxy to handle all concurrency. If the
+// proxy does not impose any concurrency restrictions, it is then left up to
+// the application to implement its own concurrency for the relevant group of
+// calls.
+func NewUnsyncLocalClient(app types.Application) Client {
+	cli := &unsyncLocalClient{
+		Application: app,
+	}
+	cli.BaseService = *service.NewBaseService(nil, "unsyncLocalClient", cli)
+	return cli
+}
+
+func (app *unsyncLocalClient) SetResponseCallback(cb Callback) {
+	app.mtx.Lock()
+	app.Callback = cb
+	app.mtx.Unlock()
+}
+
+func (app *unsyncLocalClient) CheckTxAsync(ctx context.Context, req *types.RequestCheckTx) (*ReqRes, error) {
+	res, err := app.Application.CheckTx(ctx, req)
+	if err != nil {
+		return nil, err
+	}
+	return app.callback(
+		types.ToRequestCheckTx(req),
+		types.ToResponseCheckTx(res),
+	), nil
+}
+
+func (app *unsyncLocalClient) callback(req *types.Request, res *types.Response) *ReqRes {
+	app.Callback(req, res)
+	rr := newLocalReqRes(req, res)
+	rr.callbackInvoked = true
+	return rr
+}
+
+//-------------------------------------------------------
+
+func (app *unsyncLocalClient) Error() error {
+	return nil
+}
+
+func (app *unsyncLocalClient) Flush(context.Context) error {
+	return nil
+}
+
+func (app *unsyncLocalClient) Echo(_ context.Context, msg string) (*types.ResponseEcho, error) {
+	return &types.ResponseEcho{Message: msg}, nil
+}
+
+func (app *unsyncLocalClient) Info(ctx context.Context, req *types.RequestInfo) (*types.ResponseInfo, error) {
+	return app.Application.Info(ctx, req)
+}
+
+func (app *unsyncLocalClient) CheckTx(ctx context.Context, req *types.RequestCheckTx) (*types.ResponseCheckTx, error) {
+	return app.Application.CheckTx(ctx, req)
+}
+
+func (app *unsyncLocalClient) Query(ctx context.Context, req *types.RequestQuery) (*types.ResponseQuery, error) {
+	return app.Application.Query(ctx, req)
+}
+
+func (app *unsyncLocalClient) Commit(ctx context.Context, req *types.RequestCommit) (*types.ResponseCommit, error) {
+	return app.Application.Commit(ctx, req)
+}
+
+func (app *unsyncLocalClient) InitChain(ctx context.Context, req *types.RequestInitChain) (*types.ResponseInitChain, error) {
+	return app.Application.InitChain(ctx, req)
+}
+
+func (app *unsyncLocalClient) ListSnapshots(ctx context.Context, req *types.RequestListSnapshots) (*types.ResponseListSnapshots, error) {
+	return app.Application.ListSnapshots(ctx, req)
+}
+
+func (app *unsyncLocalClient) OfferSnapshot(ctx context.Context, req *types.RequestOfferSnapshot) (*types.ResponseOfferSnapshot, error) {
+	return app.Application.OfferSnapshot(ctx, req)
+}
+
+func (app *unsyncLocalClient) LoadSnapshotChunk(ctx context.Context,
+	req *types.RequestLoadSnapshotChunk,
+) (*types.ResponseLoadSnapshotChunk, error) {
+	return app.Application.LoadSnapshotChunk(ctx, req)
+}
+
+func (app *unsyncLocalClient) ApplySnapshotChunk(ctx context.Context,
+	req *types.RequestApplySnapshotChunk,
+) (*types.ResponseApplySnapshotChunk, error) {
+	return app.Application.ApplySnapshotChunk(ctx, req)
+}
+
+func (app *unsyncLocalClient) PrepareProposal(ctx context.Context, req *types.RequestPrepareProposal) (*types.ResponsePrepareProposal, error) {
+	return app.Application.PrepareProposal(ctx, req)
+}
+
+func (app *unsyncLocalClient) ProcessProposal(ctx context.Context, req *types.RequestProcessProposal) (*types.ResponseProcessProposal, error) {
+	return app.Application.ProcessProposal(ctx, req)
+}
+
+func (app *unsyncLocalClient) ExtendVote(ctx context.Context, req *types.RequestExtendVote) (*types.ResponseExtendVote, error) {
+	return app.Application.ExtendVote(ctx, req)
+}
+
+func (app *unsyncLocalClient) VerifyVoteExtension(ctx context.Context, req *types.RequestVerifyVoteExtension) (*types.ResponseVerifyVoteExtension, error) {
+	return app.Application.VerifyVoteExtension(ctx, req)
+}
+
+func (app *unsyncLocalClient) FinalizeBlock(ctx context.Context, req *types.RequestFinalizeBlock) (*types.ResponseFinalizeBlock, error) {
+	return app.Application.FinalizeBlock(ctx, req)
+}

consensus/replay_stubs.go

@@ -61,7 +61,7 @@ func newMockProxyApp(finalizeBlockResponse *abci.ResponseFinalizeBlock) proxy.Ap
 	clientCreator := proxy.NewLocalClientCreator(&mockProxyApp{
 		finalizeBlockResponse: finalizeBlockResponse,
 	})
-	cli, _ := clientCreator.NewABCIClient()
+	cli, _ := clientCreator.NewABCIConsensusClient()
 	err := cli.Start()
 	if err != nil {
 		panic(err)

mempool/clist_mempool_test.go

@@ -67,7 +67,7 @@ func newMempoolWithApp(cc proxy.ClientCreator) (*CListMempool, cleanupFunc) {
 }
 
 func newMempoolWithAppAndConfig(cc proxy.ClientCreator, cfg *config.Config) (*CListMempool, cleanupFunc) {
-	appConnMem, _ := cc.NewABCIClient()
+	appConnMem, _ := cc.NewABCIMempoolClient()
 	appConnMem.SetLogger(log.TestingLogger().With("module", "abci-client", "connection", "mempool"))
 	err := appConnMem.Start()
 	if err != nil {
@@ -424,7 +424,7 @@ func TestSerialReap(t *testing.T) {
 	mp, cleanup := newMempoolWithApp(cc)
 	defer cleanup()
 
-	appConnCon, _ := cc.NewABCIClient()
+	appConnCon, _ := cc.NewABCIConsensusClient()
 	appConnCon.SetLogger(log.TestingLogger().With("module", "abci-client", "connection", "consensus"))
 	err := appConnCon.Start()
 	require.Nil(t, err)
@@ -629,7 +629,7 @@ func TestMempoolTxsBytes(t *testing.T) {
 	require.NoError(t, err)
 	assert.EqualValues(t, 10, mp.SizeBytes())
 
-	appConnCon, _ := cc.NewABCIClient()
+	appConnCon, _ := cc.NewABCIConsensusClient()
 	appConnCon.SetLogger(log.TestingLogger().With("module", "abci-client", "connection", "consensus"))
 	err = appConnCon.Start()
 	require.Nil(t, err)

proxy/app_conn_test.go

@@ -31,7 +31,7 @@ func TestEcho(t *testing.T) {
 	})
 
 	// Start client
-	cli, err := clientCreator.NewABCIClient()
+	cli, err := clientCreator.NewABCIMempoolClient()
 	if err != nil {
 		t.Fatalf("Error creating ABCI client: %v", err.Error())
 	}
@@ -72,7 +72,7 @@ func BenchmarkEcho(b *testing.B) {
 	})
 
 	// Start client
-	cli, err := clientCreator.NewABCIClient()
+	cli, err := clientCreator.NewABCIMempoolClient()
 	if err != nil {
 		b.Fatalf("Error creating ABCI client: %v", err.Error())
 	}