Repository: cometbft/cometbft (go)

Issue:
Delete usage of comet's libs/rand, make each gossip routine maintain its own rand instance.
<!-- < < < < < < < < < < < < < < < < < < < < < < < < < < < < < < < < < ☺ 
v                            ✰  Thanks for opening an issue! ✰    
v    Before smashing the submit button please review the template.
v    Word of caution: poorly thought-out proposals may be rejected 
v                     without deliberation 
☺ > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > >  -->

## Feature Request

### Summary

CometBFT's libs/rand was always a really questionable choice that never made sense to me. (Documented from before the cosmos hub launch: https://github.com/tendermint/tendermint/issues/2956 )

We should just delete all usage of our libs/rand, and make each gossiping thread maintain its own randomness instance. This will reduce mutex contention, and speedup core gossip routines.

Each rand instance has only 50% doing its actual rand operation. And we see `lockSlow` getting called, which per my understanding of the go runtime, means that this is actually a contended for mutex, hence this is justified.
![image](https://github.com/cometbft/cometbft/assets/6440154/169ffe6c-05b2-450e-8ca6-13789041c485)


Pull request:
perf: Make every gossip thread use its own randomness instance, reducing contention (backport #3006)
Closes #3005 

As noted in that issue, we currently are doing extra CPU overhead and mutex contention for getting a random number. This PR removes this overhead by making every performance sensitive thread have its own rand instance.

In a subsequent PR, we can cleanup all the testing usages, and likely just entirely delete our internal randomness package. I didn't do that in this PR to keep it straightforward to verify.

---

#### PR checklist

- [x] Tests written/updated
- [ ] Changelog entry added in `.changelog` (we use [unclog](https://github.com/informalsystems/unclog) to manage our changelog)
- [ ] Updated relevant documentation (`docs/` or `spec/`) and code comments
- [ ] Title follows the [Conventional Commits](https://www.conventionalcommits.org/en/v1.0.0/) spec
<hr>This is an automatic backport of pull request #3006 done by [Mergify](https://mergify.com).

Patch:
diff --git a/.changelog/unreleased/improvements/3006-use-thread-independent-randomness-in-gossip.md b/.changelog/unreleased/improvements/3006-use-thread-independent-randomness-in-gossip.md
new file mode 100644
index 000000000..290da0b1c
--- /dev/null
+++ b/.changelog/unreleased/improvements/3006-use-thread-independent-randomness-in-gossip.md
@@ -0,0 +1,2 @@
+- [`consensus`] Use an independent rng for gossip threads, reducing mutex contention.
+  ([\#3005](https://github.com/cometbft/cometbft/issues/3005)
diff --git a/internal/bits/bit_array.go b/internal/bits/bit_array.go
index 55224e162..4dad7a94e 100644
--- a/internal/bits/bit_array.go
+++ b/internal/bits/bit_array.go
@@ -4,12 +4,12 @@ import (
 	"encoding/binary"
 	"fmt"
 	"math/bits"
+	"math/rand"
 	"regexp"
 	"strings"
 	"sync"
 
 	cmtprotobits "github.com/cometbft/cometbft/api/cometbft/libs/bits/v1"
-	cmtrand "github.com/cometbft/cometbft/internal/rand"
 	cmtmath "github.com/cometbft/cometbft/libs/math"
 )
 
@@ -261,8 +261,8 @@ func (bA *BitArray) IsFull() bool {
 
 // PickRandom returns a random index for a set bit in the bit array.
 // If there is no such value, it returns 0, false.
-// It uses the global randomness in `random.go` to get this index.
-func (bA *BitArray) PickRandom() (int, bool) {
+// It uses the provided randomness to get this index.
+func (bA *BitArray) PickRandom(r *rand.Rand) (int, bool) {
 	if bA == nil {
 		return 0, false
 	}
@@ -273,7 +273,7 @@ func (bA *BitArray) PickRandom() (int, bool) {
 		bA.mtx.Unlock()
 		return 0, false
 	}
-	index := bA.getNthTrueIndex(cmtrand.Intn(numTrueIndices))
+	index := bA.getNthTrueIndex(r.Intn(numTrueIndices))
 	bA.mtx.Unlock()
 	if index == -1 {
 		return 0, false
diff --git a/internal/consensus/reactor.go b/internal/consensus/reactor.go
index 016fdb9ff..33b1ee99f 100644
--- a/internal/consensus/reactor.go
+++ b/internal/consensus/reactor.go
@@ -2,6 +2,7 @@ package consensus
 
 import (
 	"fmt"
+	"math/rand"
 	"reflect"
 	"sync"
 	"sync/atomic"
@@ -569,6 +570,7 @@ func (conR *Reactor) getRoundState() *cstypes.RoundState {
 
 func (conR *Reactor) gossipDataRoutine(peer p2p.Peer, ps *PeerState) {
 	logger := conR.Logger.With("peer", peer)
+	rng := cmtrand.NewStdlibRand()
 
 OUTER_LOOP:
 	for {
@@ -581,7 +583,7 @@ OUTER_LOOP:
 		// so we can reduce the amount of redundant block parts we send
 		if conR.conS.config.PeerGossipIntraloopSleepDuration > 0 {
 			// the config sets an upper bound for how long we sleep.
-			randDuration := cmtrand.Int63n(int64(conR.conS.config.PeerGossipIntraloopSleepDuration))
+			randDuration := rng.Int63n(int64(conR.conS.config.PeerGossipIntraloopSleepDuration))
 			time.Sleep(time.Duration(randDuration))
 		}
 
@@ -593,7 +595,7 @@ OUTER_LOOP:
 		// (Note these can match on hash so round doesn't matter)
 		// --------------------
 
-		if part, continueLoop := pickPartToSend(logger, conR.conS.blockStore, rs, ps, prs); part != nil {
+		if part, continueLoop := pickPartToSend(logger, conR.conS.blockStore, rs, ps, prs, rng); part != nil {
 			// part is not nil: we either succeed in sending it,
 			// or we were instructed not to sleep (busy-waiting)
 			if ps.SendPartSetHasPart(part, prs) || continueLoop {
@@ -624,6 +626,7 @@ OUTER_LOOP:
 
 func (conR *Reactor) gossipVotesRoutine(peer p2p.Peer, ps *PeerState) {
 	logger := conR.Logger.With("peer", peer)
+	rng := cmtrand.NewStdlibRand()
 
 	// Simple hack to throttle logs upon sleep.
 	sleeping := 0
@@ -639,7 +642,7 @@ OUTER_LOOP:
 		// so we can reduce the amount of redundant votes we send
 		if conR.conS.config.PeerGossipIntraloopSleepDuration > 0 {
 			// the config sets an upper bound for how long we sleep.
-			randDuration := cmtrand.Int63n(int64(conR.conS.config.PeerGossipIntraloopSleepDuration))
+			randDuration := rng.Int63n(int64(conR.conS.config.PeerGossipIntraloopSleepDuration))
 			time.Sleep(time.Duration(randDuration))
 		}
 
@@ -656,7 +659,7 @@ OUTER_LOOP:
 		// logger.Debug("gossipVotesRoutine", "rsHeight", rs.Height, "rsRound", rs.Round,
 		// "prsHeight", prs.Height, "prsRound", prs.Round, "prsStep", prs.Step)
 
-		if vote := pickVoteToSend(logger, conR.conS, rs, ps, prs); vote != nil {
+		if vote := pickVoteToSend(logger, conR.conS, rs, ps, prs, rng); vote != nil {
 			if ps.sendVoteSetHasVote(vote) {
 				continue OUTER_LOOP
 			}
@@ -789,10 +792,11 @@ func pickPartToSend(
 	rs *cstypes.RoundState,
 	ps *PeerState,
 	prs *cstypes.PeerRoundState,
+	rng *rand.Rand,
 ) (*types.Part, bool) {
 	// If peer has same part set header as us, send block parts
 	if rs.ProposalBlockParts.HasHeader(prs.ProposalBlockPartSetHeader) {
-		if index, ok := rs.ProposalBlockParts.BitArray().Sub(prs.ProposalBlockParts.Copy()).PickRandom(); ok {
+		if index, ok := rs.ProposalBlockParts.BitArray().Sub(prs.ProposalBlockParts.Copy()).PickRandom(rng); ok {
 			part := rs.ProposalBlockParts.GetPart(index)
 			// If sending this part fails, restart the OUTER_LOOP (busy-waiting).
 			return part, true
@@ -818,7 +822,8 @@ func pickPartToSend(
 			// continue the loop since prs is a copy and not affected by this initialization
 			return nil, true // continue OUTER_LOOP
 		}
-		part := pickPartForCatchup(heightLogger, rs, prs, blockStore)
+
+		part := pickPartForCatchup(heightLogger, rs, prs, blockStore, rng)
 		if part != nil {
 			// If sending this part fails, do not restart the OUTER_LOOP and sleep.
 			return part, false
@@ -833,8 +838,9 @@ func pickPartForCatchup(
 	rs *cstypes.RoundState,
 	prs *cstypes.PeerRoundState,
 	blockStore sm.BlockStore,
+	rng *rand.Rand,
 ) *types.Part {
-	index, ok := prs.ProposalBlockParts.Not().PickRandom()
+	index, ok := prs.ProposalBlockParts.Not().PickRandom(rng)
 	if !ok {
 		return nil
 	}
@@ -865,17 +871,18 @@ func pickVoteToSend(
 	rs *cstypes.RoundState,
 	ps *PeerState,
 	prs *cstypes.PeerRoundState,
+	rng *rand.Rand,
 ) *types.Vote {
 	// If height matches, then send LastCommit, Prevotes, Precommits.
 	if rs.Height == prs.Height {
 		heightLogger := logger.With("height", prs.Height)
-		return pickVoteCurrentHeight(heightLogger, rs, prs, ps)
+		return pickVoteCurrentHeight(heightLogger, rs, prs, ps, rng)
 	}
 
 	// Special catchup logic.
 	// If peer is lagging by height 1, send LastCommit.
 	if prs.Height != 0 && rs.Height == prs.Height+1 {
-		if vote := ps.PickVoteToSend(rs.LastCommit); vote != nil {
+		if vote := ps.PickVoteToSend(rs.LastCommit, rng); vote != nil {
 			logger.Debug("Picked rs.LastCommit to send", "height", prs.Height)
 			return vote
 		}
@@ -906,7 +913,8 @@ func pickVoteToSend(
 		if ec == nil {
 			return nil
 		}
-		if vote := ps.PickVoteToSend(ec); vote != nil {
+
+		if vote := ps.PickVoteToSend(ec, rng); vote != nil {
 			logger.Debug("Picked Catchup commit to send", "height", prs.Height)
 			return vote
 		}
@@ -919,10 +927,11 @@ func pickVoteCurrentHeight(
 	rs *cstypes.RoundState,
 	prs *cstypes.PeerRoundState,
 	ps *PeerState,
+	rng *rand.Rand,
 ) *types.Vote {
 	// If there are lastCommits to send...
 	if prs.Step == cstypes.RoundStepNewHeight {
-		if vote := ps.PickVoteToSend(rs.LastCommit); vote != nil {
+		if vote := ps.PickVoteToSend(rs.LastCommit, rng); vote != nil {
 			logger.Debug("Picked rs.LastCommit to send")
 			return vote
 		}
@@ -930,7 +939,7 @@ func pickVoteCurrentHeight(
 	// If there are POL prevotes to send...
 	if prs.Step <= cstypes.RoundStepPropose && prs.Round != -1 && prs.Round <= rs.Round && prs.ProposalPOLRound != -1 {
 		if polPrevotes := rs.Votes.Prevotes(prs.ProposalPOLRound); polPrevotes != nil {
-			if vote := ps.PickVoteToSend(polPrevotes); vote != nil {
+			if vote := ps.PickVoteToSend(polPrevotes, rng); vote != nil {
 				logger.Debug("Picked rs.Prevotes(prs.ProposalPOLRound) to send",
 					"round", prs.ProposalPOLRound)
 				return vote
@@ -939,21 +948,21 @@ func pickVoteCurrentHeight(
 	}
 	// If there are prevotes to send...
 	if prs.Step <= cstypes.RoundStepPrevoteWait && prs.Round != -1 && prs.Round <= rs.Round {
-		if vote := ps.PickVoteToSend(rs.Votes.Prevotes(prs.Round)); vote != nil {
+		if vote := ps.PickVoteToSend(rs.Votes.Prevotes(prs.Round), rng); vote != nil {
 			logger.Debug("Picked rs.Prevotes(prs.Round) to send", "round", prs.Round)
 			return vote
 		}
 	}
 	// If there are precommits to send...
 	if prs.Step <= cstypes.RoundStepPrecommitWait && prs.Round != -1 && prs.Round <= rs.Round {
-		if vote := ps.PickVoteToSend(rs.Votes.Precommits(prs.Round)); vote != nil {
+		if vote := ps.PickVoteToSend(rs.Votes.Precommits(prs.Round), rng); vote != nil {
 			logger.Debug("Picked rs.Precommits(prs.Round) to send", "round", prs.Round)
 			return vote
 		}
 	}
 	// If there are prevotes to send...Needed because of validBlock mechanism
 	if prs.Round != -1 && prs.Round <= rs.Round {
-		if vote := ps.PickVoteToSend(rs.Votes.Prevotes(prs.Round)); vote != nil {
+		if vote := ps.PickVoteToSend(rs.Votes.Prevotes(prs.Round), rng); vote != nil {
 			logger.Debug("Picked rs.Prevotes(prs.Round) to send", "round", prs.Round)
 			return vote
 		}
@@ -961,7 +970,7 @@ func pickVoteCurrentHeight(
 	// If there are POLPrevotes to send...
 	if prs.ProposalPOLRound != -1 {
 		if polPrevotes := rs.Votes.Prevotes(prs.ProposalPOLRound); polPrevotes != nil {
-			if vote := ps.PickVoteToSend(polPrevotes); vote != nil {
+			if vote := ps.PickVoteToSend(polPrevotes, rng); vote != nil {
 				logger.Debug("Picked rs.Prevotes(prs.ProposalPOLRound) to send",
 					"round", prs.ProposalPOLRound)
 				return vote
@@ -1258,7 +1267,7 @@ func (ps *PeerState) sendVoteSetHasVote(vote *types.Vote) bool {
 // PickVoteToSend picks a vote to send to the peer.
 // Returns true if a vote was picked.
 // NOTE: `votes` must be the correct Size() for the Height().
-func (ps *PeerState) PickVoteToSend(votes types.VoteSetReader) *types.Vote {
+func (ps *PeerState) PickVoteToSend(votes types.VoteSetReader, rng *rand.Rand) *types.Vote {
 	ps.mtx.Lock()
 	defer ps.mtx.Unlock()
 
@@ -1278,7 +1287,7 @@ func (ps *PeerState) PickVoteToSend(votes types.VoteSetReader) *types.Vote {
 	if psVotes == nil {
 		return nil // Not something worth sending
 	}
-	if index, ok := votes.BitArray().Sub(psVotes).PickRandom(); ok {
+	if index, ok := votes.BitArray().Sub(psVotes).PickRandom(rng); ok {
 		vote := votes.GetByIndex(int32(index))
 		if vote == nil {
 			ps.logger.Error("votes.GetByIndex returned nil", "votes", votes, "index", index)
diff --git a/internal/rand/random.go b/internal/rand/random.go
index 6197f0222..5a9f04bf8 100644
--- a/internal/rand/random.go
+++ b/internal/rand/random.go
@@ -37,14 +37,28 @@ func NewRand() *Rand {
 	return rand
 }
 
-func (r *Rand) init() {
+// Make a new stdlib rand source. Its up to the caller to ensure
+// that the rand source is not called in parallel.
+// The failure mode of calling the returned rand multiple times in parallel is
+// repeated values across threads.
+func NewStdlibRand() *mrand.Rand {
+	// G404: Use of weak random number generator (math/rand instead of crypto/rand)
+	//nolint:gosec
+	return mrand.New(mrand.NewSource(newSeed()))
+}
+
+func newSeed() int64 {
 	bz := cRandBytes(8)
 	var seed uint64
 	for i := 0; i < 8; i++ {
 		seed |= uint64(bz[i])
 		seed <<= 8
 	}
-	r.reset(int64(seed))
+	return int64(seed)
+}
+
+func (r *Rand) init() {
+	r.reset(newSeed())
 }
 
 func (r *Rand) reset(seed int64) {
@@ -302,6 +316,8 @@ func (r *Rand) Perm(n int) []int {
 // NOTE: This relies on the os's random number generator.
 // For real security, we should salt that with some seed.
 // See github.com/cometbft/cometbft/crypto for a more secure reader.
+// This function is thread safe, see:
+// https://stackoverflow.com/questions/75685374/is-golang-crypto-rand-thread-safe
 func cRandBytes(numBytes int) []byte {
 	b := make([]byte, numBytes)
 	_, err := crand.Read(b)


Test patch:
diff --git a/internal/bits/bit_array_test.go b/internal/bits/bit_array_test.go
index ea2d436dc..079dd6f10 100644
--- a/internal/bits/bit_array_test.go
+++ b/internal/bits/bit_array_test.go
@@ -4,7 +4,9 @@ import (
 	"bytes"
 	"encoding/json"
 	"fmt"
+	"math/rand"
 	"testing"
+	"time"
 
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
@@ -17,6 +19,7 @@ var (
 	empty64Bits = empty16Bits + empty16Bits + empty16Bits + empty16Bits
 	full16bits  = "xxxxxxxxxxxxxxxx"
 	full64bits  = full16bits + full16bits + full16bits + full16bits
+	grand       = rand.New(rand.NewSource(time.Now().UnixNano()))
 )
 
 func randBitArray(bits int) *BitArray {
@@ -140,7 +143,7 @@ func TestPickRandom(t *testing.T) {
 		var bitArr *BitArray
 		err := json.Unmarshal([]byte(tc.bA), &bitArr)
 		require.NoError(t, err)
-		_, ok := bitArr.PickRandom()
+		_, ok := bitArr.PickRandom(grand)
 		require.Equal(t, tc.ok, ok, "PickRandom got an unexpected result on input %s", tc.bA)
 	}
 }
@@ -395,6 +398,6 @@ func BenchmarkPickRandomBitArray(b *testing.B) {
 	require.NoError(b, err)
 	b.ResetTimer()
 	for i := 0; i < b.N; i++ {
-		_, _ = bitArr.PickRandom()
+		_, _ = bitArr.PickRandom(grand)
 	}
 }
diff --git a/internal/rand/random_test.go b/internal/rand/random_test.go
index d485a350a..91515413f 100644
--- a/internal/rand/random_test.go
+++ b/internal/rand/random_test.go
@@ -87,6 +87,24 @@ func TestRngConcurrencySafety(_ *testing.T) {
 	wg.Wait()
 }
 
+// Makes a new stdlib random instance 100 times concurrently.
+// Ensures that it is concurrent safe to create rand instances, and call independent rand
+// sources in parallel.
+func TestStdlibRngConcurrencySafety(_ *testing.T) {
+	var wg sync.WaitGroup
+	for i := 0; i < 100; i++ {
+		wg.Add(1)
+		go func() {
+			defer wg.Done()
+			r := NewStdlibRand()
+			_ = r.Uint64()
+			<-time.After(time.Millisecond * time.Duration(Intn(100)))
+			_ = r.Perm(3)
+		}()
+	}
+	wg.Wait()
+}
+
 func BenchmarkRandBytes10B(b *testing.B) {
 	benchmarkRandBytes(b, 10)
 }