abci: modify Client interface and socket client (#5673)

`abci.Client`:

- Sync and Async methods now accept a context for cancellation
    * grpc client uses context to cancel both Sync and Async requests
    * local client ignores context parameter
    * socket client uses context to cancel Sync requests and to drop Async requests before sending them if context was cancelled prior to that

- Async methods return an error
    * socket client returns an error immediately if queue is full for Async requests
    * local client always returns nil error
    * grpc client returns an error if context was cancelled before we got response or the receiving queue had a space for response (do not confuse with the sending queue from the socket client)

- specify clients semantics in [doc.go](https://raw.githubusercontent.com/tendermint/tendermint/27112fffa62276bc016d56741f686f0f77931748/abci/client/doc.go)

`mempool.TxInfo`

- add optional `Context` to `TxInfo`, which can be used to cancel `CheckTx` request

Closes #5190

mempool/clist_mempool.go

@@ -3,6 +3,7 @@ package mempool
 import (
 	"bytes"
 	"container/list"
+	"context"
 	"crypto/sha256"
 	"fmt"
 	"sync"
@@ -185,7 +186,7 @@ func (mem *CListMempool) TxsBytes() int64 {
 
 // Lock() must be help by the caller during execution.
 func (mem *CListMempool) FlushAppConn() error {
-	return mem.proxyAppConn.FlushSync()
+	return mem.proxyAppConn.FlushSync(context.Background())
 }
 
 // XXX: Unsafe! Calling Flush may leave mempool in inconsistent state.
@@ -285,7 +286,16 @@ func (mem *CListMempool) CheckTx(tx types.Tx, cb func(*abci.Response), txInfo Tx
 		return ErrTxInCache
 	}
 
-	reqRes := mem.proxyAppConn.CheckTxAsync(abci.RequestCheckTx{Tx: tx})
+	ctx := context.Background()
+	if txInfo.Context != nil {
+		ctx = txInfo.Context
+	}
+
+	reqRes, err := mem.proxyAppConn.CheckTxAsync(ctx, abci.RequestCheckTx{Tx: tx})
+	if err != nil {
+		mem.cache.Remove(tx)
+		return err
+	}
 	reqRes.SetCallback(mem.reqResCb(tx, txInfo.SenderID, txInfo.SenderP2PID, cb))
 
 	return nil
@@ -634,17 +644,26 @@ func (mem *CListMempool) recheckTxs() {
 	mem.recheckCursor = mem.txs.Front()
 	mem.recheckEnd = mem.txs.Back()
 
+	ctx := context.Background()
+
 	// Push txs to proxyAppConn
 	// NOTE: globalCb may be called concurrently.
 	for e := mem.txs.Front(); e != nil; e = e.Next() {
 		memTx := e.Value.(*mempoolTx)
-		mem.proxyAppConn.CheckTxAsync(abci.RequestCheckTx{
+		_, err := mem.proxyAppConn.CheckTxAsync(ctx, abci.RequestCheckTx{
 			Tx:   memTx.tx,
 			Type: abci.CheckTxType_Recheck,
 		})
+		if err != nil {
+			// No need in retrying since memTx will be rechecked after next block.
+			mem.logger.Error("Can't check tx", "err", err)
+		}
 	}
 
-	mem.proxyAppConn.FlushAsync()
+	_, err := mem.proxyAppConn.FlushAsync(ctx)
+	if err != nil {
+		mem.logger.Error("Can't flush txs", "err", err)
+	}
 }
 
 //--------------------------------------------------------------------------------

mempool/clist_mempool_test.go

@@ -1,6 +1,7 @@
 package mempool
 
 import (
+	"context"
 	"crypto/rand"
 	"crypto/sha256"
 	"encoding/binary"
@@ -313,11 +314,12 @@ func TestSerialReap(t *testing.T) {
 	}
 
 	commitRange := func(start, end int) {
+		ctx := context.Background()
 		// Deliver some txs.
 		for i := start; i < end; i++ {
 			txBytes := make([]byte, 8)
 			binary.BigEndian.PutUint64(txBytes, uint64(i))
-			res, err := appConnCon.DeliverTxSync(abci.RequestDeliverTx{Tx: txBytes})
+			res, err := appConnCon.DeliverTxSync(ctx, abci.RequestDeliverTx{Tx: txBytes})
 			if err != nil {
 				t.Errorf("client error committing tx: %v", err)
 			}
@@ -326,7 +328,7 @@ func TestSerialReap(t *testing.T) {
 					res.Code, res.Data, res.Log)
 			}
 		}
-		res, err := appConnCon.CommitSync()
+		res, err := appConnCon.CommitSync(ctx)
 		if err != nil {
 			t.Errorf("client error committing: %v", err)
 		}
@@ -520,10 +522,11 @@ func TestMempoolTxsBytes(t *testing.T) {
 			t.Error(err)
 		}
 	})
-	res, err := appConnCon.DeliverTxSync(abci.RequestDeliverTx{Tx: txBytes})
+	ctx := context.Background()
+	res, err := appConnCon.DeliverTxSync(ctx, abci.RequestDeliverTx{Tx: txBytes})
 	require.NoError(t, err)
 	require.EqualValues(t, 0, res.Code)
-	res2, err := appConnCon.CommitSync()
+	res2, err := appConnCon.CommitSync(ctx)
 	require.NoError(t, err)
 	require.NotEmpty(t, res2.Data)
 

mempool/mempool.go

@@ -1,6 +1,7 @@
 package mempool
 
 import (
+	"context"
 	"fmt"
 
 	abci "github.com/tendermint/tendermint/abci/types"
@@ -98,6 +99,8 @@ type TxInfo struct {
 	SenderID uint16
 	// SenderP2PID is the actual p2p.ID of the sender, used e.g. for logging.
 	SenderP2PID p2p.ID
+	// Context is the optional context to cancel CheckTx
+	Context context.Context
 }
 
 //--------------------------------------------------------------------------------