From f98de20f7e196650654c11ce8201e83fe225edfb Mon Sep 17 00:00:00 2001 From: "M. J. Fromberger" Date: Mon, 11 Jul 2022 16:34:05 -0700 Subject: [PATCH 1/4] p2p: ensure closed channels stop receiving service (#8979) Once these channels are closed, we should not continue to service them, as they will never again deliver nonzero values. --- internal/p2p/router.go | 11 ++++++++--- 1 file changed, 8 insertions(+), 3 deletions(-) diff --git a/internal/p2p/router.go b/internal/p2p/router.go index 56558a80f..55377169c 100644 --- a/internal/p2p/router.go +++ b/internal/p2p/router.go @@ -427,8 +427,10 @@ func (r *Router) routeChannel( ) { for { select { - case envelope := <-outCh: - if envelope.IsZero() { + case envelope, ok := <-outCh: + if !ok { + return + } else if envelope.IsZero() { continue } // Mark the envelope with the channel ID to allow sendPeer() to pass @@ -507,7 +509,10 @@ func (r *Router) routeChannel( } } - case peerError := <-errCh: + case peerError, ok := <-errCh: + if !ok { + return + } maxPeerCapacity := r.peerManager.HasMaxPeerCapacity() r.logger.Error("peer error", "peer", peerError.NodeID, From cb93d3b587cd85f4c918f30bbd90311f827760c0 Mon Sep 17 00:00:00 2001 From: "M. J. Fromberger" Date: Mon, 11 Jul 2022 18:06:49 -0700 Subject: [PATCH 2/4] mempool: don't log message type mismatch in the default callback (#8969) --- internal/mempool/v1/mempool.go | 10 ++++++---- 1 file changed, 6 insertions(+), 4 deletions(-) diff --git a/internal/mempool/v1/mempool.go b/internal/mempool/v1/mempool.go index 9c4122d8e..aa9ef69b0 100644 --- a/internal/mempool/v1/mempool.go +++ b/internal/mempool/v1/mempool.go @@ -462,6 +462,10 @@ func (txmp *TxMempool) Update( func (txmp *TxMempool) initialTxCallback(wtx *WrappedTx, res *abci.Response) { checkTxRes, ok := res.Value.(*abci.Response_CheckTx) if !ok { + txmp.logger.Error("mempool: received incorrect result type in CheckTx callback", + "expected", reflect.TypeOf(&abci.Response_CheckTx{}).Name(), + "got", reflect.TypeOf(res.Value).Name(), + ) return } @@ -630,10 +634,8 @@ func (txmp *TxMempool) insertTx(wtx *WrappedTx) { func (txmp *TxMempool) recheckTxCallback(req *abci.Request, res *abci.Response) { checkTxRes, ok := res.Value.(*abci.Response_CheckTx) if !ok { - txmp.logger.Error("mempool: received incorrect result type in CheckTx callback", - "expected", reflect.TypeOf(&abci.Response_CheckTx{}).Name(), - "got", reflect.TypeOf(res.Value).Name(), - ) + // Don't log this; this is the default callback and other response types + // can safely be ignored. return } From 9e64c95e56dad03c566cc6b9966e205d0d66b299 Mon Sep 17 00:00:00 2001 From: "M. J. Fromberger" Date: Tue, 12 Jul 2022 08:00:29 -0700 Subject: [PATCH 3/4] mempool: reduce lock contention during CheckTx (cleanup) (#8983) The way this was originally structured, we reacquired the lock after issuing the initial ABCI CheckTx call, only to immediately release it. Restructure the code so that this redundant acquire is no longer necessary. --- internal/mempool/v1/mempool.go | 109 +++++++++++++++++---------------- 1 file changed, 55 insertions(+), 54 deletions(-) diff --git a/internal/mempool/v1/mempool.go b/internal/mempool/v1/mempool.go index aa9ef69b0..f797568c9 100644 --- a/internal/mempool/v1/mempool.go +++ b/internal/mempool/v1/mempool.go @@ -177,69 +177,70 @@ func (txmp *TxMempool) CheckTx( // During the initial phase of CheckTx, we do not need to modify any state. // A transaction will not actually be added to the mempool until it survives // a call to the ABCI CheckTx method and size constraint checks. - txmp.mtx.RLock() - defer txmp.mtx.RUnlock() + height, err := func() (int64, error) { + txmp.mtx.RLock() + defer txmp.mtx.RUnlock() - // Reject transactions in excess of the configured maximum transaction size. - if len(tx) > txmp.config.MaxTxBytes { - return types.ErrTxTooLarge{Max: txmp.config.MaxTxBytes, Actual: len(tx)} - } - - // If a precheck hook is defined, call it before invoking the application. - if txmp.preCheck != nil { - if err := txmp.preCheck(tx); err != nil { - return types.ErrPreCheck{Reason: err} + // Reject transactions in excess of the configured maximum transaction size. + if len(tx) > txmp.config.MaxTxBytes { + return 0, types.ErrTxTooLarge{Max: txmp.config.MaxTxBytes, Actual: len(tx)} } - } - // Early exit if the proxy connection has an error. - if err := txmp.proxyAppConn.Error(); err != nil { + // If a precheck hook is defined, call it before invoking the application. + if txmp.preCheck != nil { + if err := txmp.preCheck(tx); err != nil { + return 0, types.ErrPreCheck{Reason: err} + } + } + + // Early exit if the proxy connection has an error. + if err := txmp.proxyAppConn.Error(); err != nil { + return 0, err + } + + txKey := tx.Key() + + // Check for the transaction in the cache. + if !txmp.cache.Push(tx) { + // If the cached transaction is also in the pool, record its sender. + if elt, ok := txmp.txByKey[txKey]; ok { + w := elt.Value.(*WrappedTx) + w.SetPeer(txInfo.SenderID) + } + return 0, types.ErrTxInCache + } + return txmp.height, nil + }() + if err != nil { return err } - txKey := tx.Key() - - // Check for the transaction in the cache. - if !txmp.cache.Push(tx) { - // If the cached transaction is also in the pool, record its sender. - if elt, ok := txmp.txByKey[txKey]; ok { - w := elt.Value.(*WrappedTx) - w.SetPeer(txInfo.SenderID) - } - return types.ErrTxInCache - } - // Initiate an ABCI CheckTx for this transaction. The callback is // responsible for adding the transaction to the pool if it survives. - return func() error { - // N.B.: We have to issue the call outside the lock. In a local client, - // even an "async" call invokes its callback immediately which will make - // the callback deadlock trying to acquire the same lock. This isn't a - // problem with out-of-process calls, but this has to work for both. - height := txmp.height - txmp.mtx.RUnlock() - defer txmp.mtx.RLock() - - reqRes, err := txmp.proxyAppConn.CheckTxAsync(ctx, abci.RequestCheckTx{Tx: tx}) - if err != nil { - txmp.cache.Remove(tx) - return err + // + // N.B.: We have to issue the call outside the lock. In a local client, + // even an "async" call invokes its callback immediately which will make + // the callback deadlock trying to acquire the same lock. This isn't a + // problem with out-of-process calls, but this has to work for both. + reqRes, err := txmp.proxyAppConn.CheckTxAsync(ctx, abci.RequestCheckTx{Tx: tx}) + if err != nil { + txmp.cache.Remove(tx) + return err + } + reqRes.SetCallback(func(res *abci.Response) { + wtx := &WrappedTx{ + tx: tx, + hash: tx.Key(), + timestamp: time.Now().UTC(), + height: height, } - reqRes.SetCallback(func(res *abci.Response) { - wtx := &WrappedTx{ - tx: tx, - hash: txKey, - timestamp: time.Now().UTC(), - height: height, - } - wtx.SetPeer(txInfo.SenderID) - txmp.initialTxCallback(wtx, res) - if cb != nil { - cb(res) - } - }) - return nil - }() + wtx.SetPeer(txInfo.SenderID) + txmp.initialTxCallback(wtx, res) + if cb != nil { + cb(res) + } + }) + return nil } // RemoveTxByKey removes the transaction with the specified key from the From 379096815648dfc9d007a33530016b9e58cef492 Mon Sep 17 00:00:00 2001 From: "M. J. Fromberger" Date: Tue, 12 Jul 2022 10:28:51 -0700 Subject: [PATCH 4/4] mempool: release lock during app connection flush (#8984) This case is symmetric to what we did for CheckTx calls, where we release the mempool mutex to ensure callbacks can fire during call setup. We also need this behaviour for application flush, for the same reason: The caller holds the lock by contract from the Mempool interface. --- internal/mempool/v1/mempool.go | 9 +++++++++ 1 file changed, 9 insertions(+) diff --git a/internal/mempool/v1/mempool.go b/internal/mempool/v1/mempool.go index f797568c9..67c4c2858 100644 --- a/internal/mempool/v1/mempool.go +++ b/internal/mempool/v1/mempool.go @@ -131,6 +131,15 @@ func (txmp *TxMempool) SizeBytes() int64 { return atomic.LoadInt64(&txmp.txsByte // The caller must hold an exclusive mempool lock (by calling txmp.Lock) before // calling FlushAppConn. func (txmp *TxMempool) FlushAppConn() error { + // N.B.: We have to issue the call outside the lock so that its callback can + // fire. It's safe to do this, the flush will block until complete. + // + // We could just not require the caller to hold the lock at all, but the + // semantics of the Mempool interface require the caller to hold it, and we + // can't change that without disrupting existing use. + txmp.mtx.Unlock() + defer txmp.mtx.Lock() + return txmp.proxyAppConn.FlushSync(context.Background()) }