V1.0.0 rc1 osmo

.changelog/unreleased/breaking-changes/3184-remove-PeerSendBytesTotal-metric.md

@@ -0,0 +1,2 @@
+`[p2p]` Remove `PeerSendBytesTotal` metric as it is costly to track,
+and not that informative in debugging. ([\#3184](https://github.com/cometbft/cometbft/issues/3184))

.changelog/unreleased/improvements/3117-significantly-speedup-make-partset.md

@@ -1,2 +1,2 @@
-- `[types]` Significantly speedup types.MakePartSet and types.AddPart, which are used in creating a block proposal
-  ([\#3117](https://github.com/cometbft/cometbft/issues/3117))
+- [`types`] Significantly speedup types.MakePartSet and types.AddPart, which are used in creating a block proposal
+  ([\#3117](https://github.com/cometbft/cometbft/issues/3117)

.changelog/unreleased/improvements/3157-late-votes-dont-take-cs-mtx.md

@@ -0,0 +1,4 @@
+- [`consensus`] Check for late votes in the reactor, preventing them from entering the 
+single-threaded consensus logic. This change is a performance optimization that reduces the number 
+of redundant votes that are processed by the consensus logic.
+  ([\#3157](https://github.com/cometbft/cometbft/issues/3157)

.changelog/unreleased/improvements/3161-move-blockpart-redundancy-checks-into-reactor.md

@@ -0,0 +1,2 @@
+- [`consensus`] Move blockpart redundancy/lateness checks into the reactor, preventing late / duplicate parts from blocking consensus.
+  ([\#3161](https://github.com/cometbft/cometbft/issues/3161)

.changelog/unreleased/improvements/3211-make-cs-reactor-no-longer-takes-cs-locks.md

@@ -0,0 +1,4 @@
+- `[consensus]` Make the consensus reactor no longer have packets on receive take the consensus lock.
+Consensus will now update the reactor's view after every relevant change through the existing 
+synchronous event bus subscription.
+  ([\#3211](https://github.com/cometbft/cometbft/pull/3211))

.changelog/unreleased/improvements/3346-buffer-secret-conn-writs.md

@@ -0,0 +1,2 @@
+- `[p2p/secretconn]` Speedup secretconnection large writes, by buffering the write to the underlying connection.
+  ([\#3346](https://github.com/cometbft/cometbft/pull/3346))

.changelog/unreleased/improvements/3382-single-send-monitor-per-packet.md

@@ -0,0 +1,2 @@
+- `[p2p/conn]` Update send monitor, used for sending rate limiting, once per batch of packets sent
+  ([\#3382](https://github.com/cometbft/cometbft/pull/3382))

.changelog/unreleased/improvements/3403-remove-pool-buffer-usage-in-secretconn.md

@@ -0,0 +1,2 @@
+- `[p2p/conn]` Remove the usage of a synchronous pool of buffers in secret connection, storing instead the buffer in the connection struct. This reduces the synchronization primitive usage, speeding up the code.
+  ([\#3403](https://github.com/cometbft/cometbft/issues/3403))

CHANGELOG.md

@@ -1,6 +1,32 @@
 # CHANGELOG
 
+<<<<<<< HEAD
+<<<<<<< HEAD
 ## Unreleased
+=======
+=======
+## Unreleased (I think)
+
+* [#123](https://github.com/osmosis-labs/cometbft/pull/123)  perf(p2p/conn): Remove unneeded global pool buffers in secret connection #3403 
+* perf(p2p): Delete expensive debug log already slated for deletion #3412 
+* perf(p2p): Reduce the p2p metrics overhead. #3411 
+* commit f663bd35153b0b366c1e1e6b41e7f2dcff7963fd : one more debug log deletion
+* [#120](https://github.com/osmosis-labs/cometbft/pull/120) perf(consensus): Use TrySend for hasVote/HasBlockPart messages #3407 
+
+
+## v0.37.4-v25-osmo-11
+
+* [#117](https://github.com/osmosis-labs/cometbft/pull/117) fix(mempool)!: stop accepting TXs in the mempool if we can't keep up
+* [#118](https://github.com/osmosis-labs/cometbft/pull/118) perf: bring back stop recheck
+
+>>>>>>> d941920d2 (Secret conn remove pool buffer, align with upstream (#123))
+## v0.37.4-v25-osmo-10
+
+* [#115](https://github.com/osmosis-labs/cometbft/pull/115) perf(p2p/secretconn): Buffer secret connection writes (#3346)
+
+
+## v0.37.4-v25-osmo-9
+>>>>>>> 9f773defd ( perf(p2p/secretconn): Buffer secret connection writes #3346  (#115))
 
 *July 1, 2024*
 

crypto/merkle/bench_test.go

@@ -40,3 +40,25 @@ func BenchmarkInnerHash(b *testing.B) {
 		b.Fatal("Benchmark did not run!")
 	}
 }
+
+// Benchmark the time it takes to hash a 64kb leaf, which is the size of
+// a block part.
+// This helps determine whether its worth parallelizing this hash for the proposer.
+func BenchmarkLeafHash64kb(b *testing.B) {
+	b.ReportAllocs()
+	leaf := make([]byte, 64*1024)
+	hash := sha256.New()
+
+	for i := 0; i < b.N; i++ {
+		leaf[0] = byte(i)
+		got := leafHashOpt(hash, leaf)
+		if g, w := len(got), sha256.Size; g != w {
+			b.Fatalf("size discrepancy: got %d, want %d", g, w)
+		}
+		sink = got
+	}
+
+	if sink == nil {
+		b.Fatal("Benchmark did not run!")
+	}
+}

crypto/merkle/proof.go

@@ -4,6 +4,7 @@ import (
 	"bytes"
 	"errors"
 	"fmt"
+	"hash"
 
 	cmtcrypto "github.com/cometbft/cometbft/api/cometbft/crypto/v1"
 	"github.com/cometbft/cometbft/crypto/tmhash"
@@ -91,13 +92,14 @@ func (sp *Proof) Verify(rootHash []byte, leaf []byte) error {
 			Err: errors.New("negative proof index"),
 		}
 	}
-	leafHash := leafHash(leaf)
+	hash := tmhash.New()
+	leafHash := leafHashOpt(hash, leaf)
 	if !bytes.Equal(sp.LeafHash, leafHash) {
 		return ErrInvalidHash{
 			Err: fmt.Errorf("leaf %x, want %x", sp.LeafHash, leafHash),
 		}
 	}
-	computedHash, err := sp.computeRootHash()
+	computedHash, err := sp.computeRootHash(hash)
 	if err != nil {
 		return ErrInvalidHash{
 			Err: fmt.Errorf("compute root hash: %w", err),
@@ -111,9 +113,19 @@ func (sp *Proof) Verify(rootHash []byte, leaf []byte) error {
 	return nil
 }
 
+// Compute the root hash given a leaf hash.  Panics in case of errors.
+func (sp *Proof) ComputeRootHash() []byte {
+	computedHash, err := sp.computeRootHash(tmhash.New())
+	if err != nil {
+		panic(fmt.Errorf("ComputeRootHash errored %w", err))
+	}
+	return computedHash
+}
+
 // Compute the root hash given a leaf hash.
-func (sp *Proof) computeRootHash() ([]byte, error) {
+func (sp *Proof) computeRootHash(hash hash.Hash) ([]byte, error) {
 	return computeHashFromAunts(
+		hash,
 		sp.Index,
 		sp.Total,
 		sp.LeafHash,
@@ -200,7 +212,7 @@ func ProofFromProto(pb *cmtcrypto.Proof) (*Proof, error) {
 // Use the leafHash and innerHashes to get the root merkle hash.
 // If the length of the innerHashes slice isn't exactly correct, the result is nil.
 // Recursive impl.
-func computeHashFromAunts(index, total int64, leafHash []byte, innerHashes [][]byte) ([]byte, error) {
+func computeHashFromAunts(hash hash.Hash, index, total int64, leafHash []byte, innerHashes [][]byte) ([]byte, error) {
 	if index >= total || index < 0 || total <= 0 {
 		return nil, fmt.Errorf("invalid index %d and/or total %d", index, total)
 	}
@@ -218,18 +230,18 @@ func computeHashFromAunts(index, total int64, leafHash []byte, innerHashes [][]b
 		}
 		numLeft := getSplitPoint(total)
 		if index < numLeft {
-			leftHash, err := computeHashFromAunts(index, numLeft, leafHash, innerHashes[:len(innerHashes)-1])
+			leftHash, err := computeHashFromAunts(hash, index, numLeft, leafHash, innerHashes[:len(innerHashes)-1])
 			if err != nil {
 				return nil, err
 			}
 
-			return innerHash(leftHash, innerHashes[len(innerHashes)-1]), nil
+			return innerHashOpt(hash, leftHash, innerHashes[len(innerHashes)-1]), nil
 		}
-		rightHash, err := computeHashFromAunts(index-numLeft, total-numLeft, leafHash, innerHashes[:len(innerHashes)-1])
+		rightHash, err := computeHashFromAunts(hash, index-numLeft, total-numLeft, leafHash, innerHashes[:len(innerHashes)-1])
 		if err != nil {
 			return nil, err
 		}
-		return innerHash(innerHashes[len(innerHashes)-1], rightHash), nil
+		return innerHashOpt(hash, innerHashes[len(innerHashes)-1], rightHash), nil
 	}
 }
 
@@ -266,18 +278,22 @@ func (spn *ProofNode) FlattenAunts() [][]byte {
 // trails[0].Hash is the leaf hash for items[0].
 // trails[i].Parent.Parent....Parent == root for all i.
 func trailsFromByteSlices(items [][]byte) (trails []*ProofNode, root *ProofNode) {
+	return trailsFromByteSlicesInternal(tmhash.New(), items)
+}
+
+func trailsFromByteSlicesInternal(hash hash.Hash, items [][]byte) (trails []*ProofNode, root *ProofNode) {
 	// Recursive impl.
 	switch len(items) {
 	case 0:
 		return []*ProofNode{}, &ProofNode{emptyHash(), nil, nil, nil}
 	case 1:
-		trail := &ProofNode{leafHash(items[0]), nil, nil, nil}
+		trail := &ProofNode{leafHashOpt(hash, items[0]), nil, nil, nil}
 		return []*ProofNode{trail}, trail
 	default:
 		k := getSplitPoint(int64(len(items)))
-		lefts, leftRoot := trailsFromByteSlices(items[:k])
-		rights, rightRoot := trailsFromByteSlices(items[k:])
-		rootHash := innerHash(leftRoot.Hash, rightRoot.Hash)
+		lefts, leftRoot := trailsFromByteSlicesInternal(hash, items[:k])
+		rights, rightRoot := trailsFromByteSlicesInternal(hash, items[k:])
+		rootHash := innerHashOpt(hash, leftRoot.Hash, rightRoot.Hash)
 		root := &ProofNode{rootHash, nil, nil, nil}
 		leftRoot.Parent = root
 		leftRoot.Right = rightRoot

crypto/merkle/proof_value.go

@@ -104,7 +104,7 @@ func (op ValueOp) Run(args [][]byte) ([][]byte, error) {
 		}
 	}
 
-	rootHash, err := op.Proof.computeRootHash()
+	rootHash, err := op.Proof.computeRootHash(tmhash.New())
 	if err != nil {
 		return nil, err
 	}

docs/explanation/core/metrics.md

@@ -56,7 +56,6 @@ The following metrics are available:
 | p2p\_message\_receive\_bytes\_total        | Counter   | message\_type    | Number of bytes received from all peers per message type                                                                                   |
 | p2p\_peers                                 | Gauge     |                  | Number of peers node's connected to                                                                                                        |
 | p2p\_peer\_receive\_bytes\_total           | Counter   | peer\_id, chID   | Number of bytes per channel received from a given peer                                                                                     |
-| p2p\_peer\_send\_bytes\_total              | Counter   | peer\_id, chID   | Number of bytes per channel sent to a given peer                                                                                           |
 | p2p\_peer\_pending\_send\_bytes            | Gauge     | peer\_id         | Number of pending bytes to be sent to a given peer                                                                                         |
 | p2p\_num\_txs                              | Gauge     | peer\_id         | Number of transactions submitted by each peer\_id                                                                                          |
 | p2p\_pending\_send\_bytes                  | Gauge     | peer\_id         | Amount of data pending to be sent to peer                                                                                                  |

go.mod

@@ -14,7 +14,6 @@ require (
 	github.com/google/orderedcode v0.0.1
 	github.com/gorilla/websocket v1.5.3
 	github.com/lib/pq v1.10.9
-	github.com/libp2p/go-buffer-pool v0.1.0
 	github.com/minio/highwayhash v1.0.2
 	github.com/ory/dockertest v3.3.5+incompatible
 	github.com/pkg/errors v0.9.1

go.sum

@@ -257,8 +257,6 @@ github.com/kr/text v0.2.0 h1:5Nx0Ya0ZqY2ygV366QzturHI13Jq95ApcVaJBhpS+AY=
 github.com/kr/text v0.2.0/go.mod h1:eLer722TekiGuMkidMxC/pM04lWEeraHUUmBw8l2grE=
 github.com/lib/pq v1.10.9 h1:YXG7RB+JIjhP29X+OtkiDnYaXQwpS4JEWq7dtCCRUEw=
 github.com/lib/pq v1.10.9/go.mod h1:AlVN5x4E4T544tWzH6hKfbfQvm3HdbOxrmggDNAPY9o=
-github.com/libp2p/go-buffer-pool v0.1.0 h1:oK4mSFcQz7cTQIfqbe4MIj9gLW+mnanjyFtc6cdF0Y8=
-github.com/libp2p/go-buffer-pool v0.1.0/go.mod h1:N+vh8gMqimBzdKkSMVuydVDq+UV5QTWy5HSiZacSbPg=
 github.com/linxGnu/grocksdb v1.8.14 h1:HTgyYalNwBSG/1qCQUIott44wU5b2Y9Kr3z7SK5OfGQ=
 github.com/linxGnu/grocksdb v1.8.14/go.mod h1:QYiYypR2d4v63Wj1adOOfzglnoII0gLj3PNh4fZkcFA=
 github.com/magiconair/properties v1.8.7 h1:IeQXZAiQcpL9mgcAe1Nu6cX9LLw6ExEHKjN0VQdvPDY=

internal/autofile/group.go

@@ -16,6 +16,8 @@ import (
 	"github.com/cometbft/cometbft/libs/service"
 )
 
+var indexedFilePattern = regexp.MustCompile(`^.+\.([0-9]{3,})$`)
+
 const (
 	defaultGroupCheckDuration = 5000 * time.Millisecond
 	defaultHeadSizeLimit      = 10 * 1024 * 1024       // 10MB
@@ -59,7 +61,7 @@ type Group struct {
 	headBuf            *bufio.Writer
 	Dir                string // Directory that contains .Head
 	ticker             *time.Ticker
-	mtx                sync.Mutex
+	mtx                sync.RWMutex
 	headSizeLimit      int64
 	totalSizeLimit     int64
 	groupCheckDuration time.Duration
@@ -170,29 +172,29 @@ func (g *Group) Close() {
 
 // HeadSizeLimit returns the current head size limit.
 func (g *Group) HeadSizeLimit() int64 {
-	g.mtx.Lock()
-	defer g.mtx.Unlock()
+	g.mtx.RLock()
+	defer g.mtx.RUnlock()
 	return g.headSizeLimit
 }
 
 // TotalSizeLimit returns total size limit of the group.
 func (g *Group) TotalSizeLimit() int64 {
-	g.mtx.Lock()
-	defer g.mtx.Unlock()
+	g.mtx.RLock()
+	defer g.mtx.RUnlock()
 	return g.totalSizeLimit
 }
 
 // MaxIndex returns index of the last file in the group.
 func (g *Group) MaxIndex() int {
-	g.mtx.Lock()
-	defer g.mtx.Unlock()
+	g.mtx.RLock()
+	defer g.mtx.RUnlock()
 	return g.maxIndex
 }
 
 // MinIndex returns index of the first file in the group.
 func (g *Group) MinIndex() int {
-	g.mtx.Lock()
-	defer g.mtx.Unlock()
+	g.mtx.RLock()
+	defer g.mtx.RUnlock()
 	return g.minIndex
 }
 
@@ -219,8 +221,8 @@ func (g *Group) WriteLine(line string) error {
 
 // Buffered returns the size of the currently buffered data.
 func (g *Group) Buffered() int {
-	g.mtx.Lock()
-	defer g.mtx.Unlock()
+	g.mtx.RLock()
+	defer g.mtx.RUnlock()
 	return g.headBuf.Buffered()
 }
 
@@ -347,8 +349,8 @@ type GroupInfo struct {
 
 // Returns info after scanning all files in g.Head's dir.
 func (g *Group) ReadGroupInfo() GroupInfo {
-	g.mtx.Lock()
-	defer g.mtx.Unlock()
+	g.mtx.RLock()
+	defer g.mtx.RUnlock()
 	return g.readGroupInfo()
 }
 
@@ -380,7 +382,7 @@ func (g *Group) readGroupInfo() GroupInfo {
 		} else if strings.HasPrefix(fileInfo.Name(), headBase) {
 			fileSize := fileInfo.Size()
 			totalSize += fileSize
-			indexedFilePattern := regexp.MustCompile(`^.+\.([0-9]{3,})$`)
+
 			submatch := indexedFilePattern.FindSubmatch([]byte(fileInfo.Name()))
 			if len(submatch) != 0 {
 				// Matches
@@ -422,7 +424,7 @@ func filePathForIndex(headPath string, index int, maxIndex int) string {
 // GroupReader provides an interface for reading from a Group.
 type GroupReader struct {
 	*Group
-	mtx       sync.Mutex
+	mtx       sync.RWMutex
 	curIndex  int
 	curFile   *os.File
 	curReader *bufio.Reader

internal/consensus/invalid_test.go

@@ -51,7 +51,7 @@ func TestReactorInvalidPrecommit(t *testing.T) {
 	}
 	byzVal.mtx.Unlock()
 
-	// wait for a bunch of blocks
+	// wait for a bunch of blocks, from each validator
 	// TODO: make this tighter by ensuring the halt happens by block 2
 	for i := 0; i < 10; i++ {
 		timeoutWaitGroup(n, func(j int) {