refactor(mempool): use tmsync.Waker for available txs notifications

internal/consensus/state.go

@@ -768,6 +768,8 @@ func (cs *State) handleTimeout(
 }
 
 func (cs *State) handleTxsAvailable(ctx context.Context, stateData *StateData) {
+	// TODO: Change to trace
+	cs.logger.Debug("new transactions are available", "height", stateData.Height, "round", stateData.Round, "step", stateData.Step)
 	// We only need to do this for round 0.
 	if stateData.Round != 0 {
 		return

internal/consensus/vote_signer.go

@@ -34,8 +34,9 @@ func (s *voteSigner) signAddVote(
 	}
 	// If the node not in the validator set, do nothing.
 	if !stateData.Validators.HasProTxHash(s.privValidator.ProTxHash) {
-		s.logger.Error("do nothing, node %s is not a part of validator set %+v",
-			s.privValidator.ProTxHash.ShortString(), stateData.Validators)
+		s.logger.Error("do nothing, node is not a part of validator set",
+			"protxhash", s.privValidator.ProTxHash.ShortString(),
+			"validators", stateData.Validators)
 		return nil
 	}
 	keyVals := []any{"height", stateData.Height, "round", stateData.Round, "quorum_hash", stateData.Validators.QuorumHash}

internal/mempool/mempool.go

@@ -16,6 +16,7 @@ import (
 	"github.com/dashpay/tenderdash/config"
 	"github.com/dashpay/tenderdash/internal/libs/clist"
 	tmstrings "github.com/dashpay/tenderdash/internal/libs/strings"
+	tmsync "github.com/dashpay/tenderdash/internal/libs/sync"
 	"github.com/dashpay/tenderdash/libs/log"
 	"github.com/dashpay/tenderdash/types"
 )
@@ -48,7 +49,7 @@ type TxMempool struct {
 	// Synchronized fields, protected by mtx.
 	mtx                  *sync.RWMutex
 	notifiedTxsAvailable bool
-	txsAvailable         chan struct{} // one value sent per height when mempool is not empty
+	txsAvailable         *tmsync.Waker
 	preCheck             PreCheckFunc
 	postCheck            PostCheckFunc
 	height               int64 // the latest height passed to Update
@@ -146,12 +147,12 @@ func (txmp *TxMempool) EnableTxsAvailable() {
 	txmp.mtx.Lock()
 	defer txmp.mtx.Unlock()
 
-	txmp.txsAvailable = make(chan struct{}, 1)
+	txmp.txsAvailable = tmsync.NewWaker()
 }
 
 // TxsAvailable returns a channel which fires once for every height, and only
 // when transactions are available in the mempool. It is thread-safe.
-func (txmp *TxMempool) TxsAvailable() <-chan struct{} { return txmp.txsAvailable }
+func (txmp *TxMempool) TxsAvailable() <-chan struct{} { return txmp.txsAvailable.Sleep() }
 
 // CheckTx adds the given transaction to the mempool if it fits and passes the
 // application's ABCI CheckTx method.
@@ -771,8 +772,7 @@ func (txmp *TxMempool) recheckTransactions(ctx context.Context) {
 
 		// When recheck is complete, trigger a notification for more transactions.
 		_ = g.Wait()
-		txmp.mtx.Lock()
-		defer txmp.mtx.Unlock()
+
 		txmp.notifyTxsAvailable()
 	}()
 }
@@ -827,6 +827,11 @@ func (txmp *TxMempool) purgeExpiredTxs(blockHeight int64) {
 	}
 }
 
+// notifyTxsAvailable triggers a notification that transactions are available in
+// the mempool. It is a no-op if the mempool is empty or if a notification has
+// already been sent.
+//
+// No locking is required to call this method.
 func (txmp *TxMempool) notifyTxsAvailable() {
 	if txmp.Size() == 0 {
 		return // nothing to do
@@ -836,9 +841,6 @@ func (txmp *TxMempool) notifyTxsAvailable() {
 		// channel cap is 1, so this will send once
 		txmp.notifiedTxsAvailable = true
 
-		select {
-		case txmp.txsAvailable <- struct{}{}:
-		default:
-		}
+		txmp.txsAvailable.Wake()
 	}
 }

internal/mempool/reactor.go

@@ -29,7 +29,8 @@ type Reactor struct {
 
 	cfg     *config.MempoolConfig
 	mempool *TxMempool
-	ids     *IDs
+	// Peer IDs assigned for peers
+	ids *IDs
 
 	peerEvents p2p.PeerEventSubscriber
 	p2pClient  *client.Client