Skip to content

Commit dec4488

Browse files
floor-lickerfloor-licker
committed
feat: integrate ultra-low latency transaction feed with ~2.5μs end-to-end propagation including transaction event publishing on successful pool insertion, filtering by target address for selective feeds, comprehensive test suite validating filtering and multiple concurrent consumers, and benchmarks demonstrating 205ns publish operations with 2.5μs average propagation latency and 21μs maximum latency on Apple M4
1 parent 250b138 commit dec4488

File tree

3 files changed

+262
-2
lines changed

3 files changed

+262
-2
lines changed

‎core/txpool/fastfeed/feed.go‎

Lines changed: 11 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -102,7 +102,7 @@ func (f *TxFilter) Matches(event *TxEvent) bool {
102102

103103
// Check gas price
104104
if f.MinGasPrice > 0 {
105-
// Simple comparison of first 8 bytes as uint64
105+
// Convert last 8 bytes to uint64 (big-endian)
106106
gasPrice := uint64(event.GasPrice[24])<<56 |
107107
uint64(event.GasPrice[25])<<48 |
108108
uint64(event.GasPrice[26])<<40 |
@@ -111,7 +111,16 @@ func (f *TxFilter) Matches(event *TxEvent) bool {
111111
uint64(event.GasPrice[29])<<16 |
112112
uint64(event.GasPrice[30])<<8 |
113113
uint64(event.GasPrice[31])
114-
if gasPrice < f.MinGasPrice {
114+
// Only filter if we can reliably compare (if value fits in uint64)
115+
// Skip filtering for very large gas prices
116+
hasHigherBytes := false
117+
for i := 0; i < 24; i++ {
118+
if event.GasPrice[i] != 0 {
119+
hasHigherBytes = true
120+
break
121+
}
122+
}
123+
if !hasHigherBytes && gasPrice < f.MinGasPrice {
115124
return false
116125
}
117126
}

‎core/txpool/fastfeed_test.go‎

Lines changed: 246 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,246 @@
1+
// Copyright 2024 The go-ethereum Authors
2+
// This file is part of the go-ethereum library.
3+
//
4+
// The go-ethereum library is free software: you can redistribute it and/or modify
5+
// it under the terms of the GNU Lesser General Public License as published by
6+
// the Free Software Foundation, either version 3 of the License, or
7+
// (at your option) any later version.
8+
//
9+
// The go-ethereum library is distributed in the hope that it will be useful,
10+
// but WITHOUT ANY WARRANTY; without even the implied warranty of
11+
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
12+
// GNU Lesser General Public License for more details.
13+
//
14+
// You should have received a copy of the GNU Lesser General Public License
15+
// along with the go-ethereum library. If not, see <http://www.gnu.org/licenses/>.
16+
17+
package txpool
18+
19+
import (
20+
"math/big"
21+
"testing"
22+
"time"
23+
24+
"github.com/ethereum/go-ethereum/common"
25+
"github.com/ethereum/go-ethereum/core/txpool/fastfeed"
26+
"github.com/ethereum/go-ethereum/core/types"
27+
"github.com/ethereum/go-ethereum/crypto"
28+
"github.com/ethereum/go-ethereum/params"
29+
)
30+
31+
func TestTxFastFeedBasic(t *testing.T) {
32+
feed := fastfeed.NewTxFastFeed()
33+
34+
// Create a test transaction
35+
key, _ := crypto.GenerateKey()
36+
signer := types.LatestSigner(params.TestChainConfig)
37+
tx := types.MustSignNewTx(key, signer, &types.LegacyTx{
38+
Nonce: 0,
39+
GasPrice: big.NewInt(1000000000),
40+
Gas: 21000,
41+
To: &common.Address{1},
42+
Value: big.NewInt(1000000000000000000),
43+
})
44+
45+
// Subscribe
46+
sub, err := feed.Subscribe(nil)
47+
if err != nil {
48+
t.Fatalf("Failed to subscribe: %v", err)
49+
}
50+
defer sub.Unsubscribe()
51+
52+
// Publish transaction
53+
feed.Publish(tx, fastfeed.TxEventAdded)
54+
55+
// Receive event
56+
select {
57+
case event := <-sub.Events():
58+
if event.EventType != fastfeed.TxEventAdded {
59+
t.Errorf("Expected TxEventAdded, got %d", event.EventType)
60+
}
61+
expectedHash := tx.Hash()
62+
receivedHash := common.BytesToHash(event.Hash[:])
63+
if receivedHash != expectedHash {
64+
t.Errorf("Hash mismatch: expected %s, got %s", expectedHash.Hex(), receivedHash.Hex())
65+
}
66+
case <-time.After(100 * time.Millisecond):
67+
t.Fatal("Timeout waiting for event")
68+
}
69+
}
70+
71+
func TestTxFastFeedFiltering(t *testing.T) {
72+
feed := fastfeed.NewTxFastFeed()
73+
74+
// Create test transactions
75+
key, _ := crypto.GenerateKey()
76+
signer := types.LatestSigner(params.TestChainConfig)
77+
78+
targetAddr := common.Address{1}
79+
otherAddr := common.Address{2}
80+
81+
// Transaction to target address
82+
targetTx := types.MustSignNewTx(key, signer, &types.LegacyTx{
83+
Nonce: 0,
84+
GasPrice: big.NewInt(1000000000),
85+
Gas: 21000,
86+
To: &targetAddr,
87+
Value: big.NewInt(1000),
88+
})
89+
90+
// Transaction to other address
91+
otherTx := types.MustSignNewTx(key, signer, &types.LegacyTx{
92+
Nonce: 1,
93+
GasPrice: big.NewInt(1000000000),
94+
Gas: 21000,
95+
To: &otherAddr,
96+
Value: big.NewInt(2000),
97+
})
98+
99+
// Subscribe with address filter
100+
filter := &fastfeed.TxFilter{
101+
Addresses: map[common.Address]struct{}{
102+
targetAddr: {},
103+
},
104+
}
105+
sub, err := feed.Subscribe(filter)
106+
if err != nil {
107+
t.Fatalf("Failed to subscribe: %v", err)
108+
}
109+
defer sub.Unsubscribe()
110+
111+
// Publish both transactions
112+
feed.Publish(otherTx, fastfeed.TxEventAdded)
113+
feed.Publish(targetTx, fastfeed.TxEventAdded)
114+
115+
// Should only receive target address tx
116+
select {
117+
case event := <-sub.Events():
118+
receivedHash := common.BytesToHash(event.Hash[:])
119+
if receivedHash != targetTx.Hash() {
120+
t.Errorf("Expected target tx %s, got %s", targetTx.Hash().Hex(), receivedHash.Hex())
121+
}
122+
case <-time.After(100 * time.Millisecond):
123+
t.Fatal("Timeout waiting for filtered event")
124+
}
125+
126+
// Should not receive other address tx
127+
select {
128+
case event := <-sub.Events():
129+
t.Errorf("Unexpected event received: %s", common.BytesToHash(event.Hash[:]).Hex())
130+
case <-time.After(50 * time.Millisecond):
131+
// Expected timeout
132+
}
133+
}
134+
135+
func TestTxFastFeedMultipleConsumers(t *testing.T) {
136+
feed := fastfeed.NewTxFastFeed()
137+
138+
// Create test transaction
139+
key, _ := crypto.GenerateKey()
140+
signer := types.LatestSigner(params.TestChainConfig)
141+
tx := types.MustSignNewTx(key, signer, &types.LegacyTx{
142+
Nonce: 0,
143+
GasPrice: big.NewInt(1000000000),
144+
Gas: 21000,
145+
To: &common.Address{1},
146+
Value: big.NewInt(1000),
147+
})
148+
149+
// Create multiple subscribers
150+
const numSubs = 5
151+
subs := make([]*fastfeed.Subscription, numSubs)
152+
for i := 0; i < numSubs; i++ {
153+
sub, err := feed.Subscribe(nil)
154+
if err != nil {
155+
t.Fatalf("Failed to subscribe #%d: %v", i, err)
156+
}
157+
defer sub.Unsubscribe()
158+
subs[i] = sub
159+
}
160+
161+
// Publish transaction
162+
feed.Publish(tx, fastfeed.TxEventAdded)
163+
164+
// All subscribers should receive the event
165+
for i, sub := range subs {
166+
select {
167+
case event := <-sub.Events():
168+
receivedHash := common.BytesToHash(event.Hash[:])
169+
if receivedHash != tx.Hash() {
170+
t.Errorf("Subscriber %d: hash mismatch", i)
171+
}
172+
case <-time.After(100 * time.Millisecond):
173+
t.Fatalf("Subscriber %d: timeout waiting for event", i)
174+
}
175+
}
176+
}
177+
178+
func BenchmarkTxFastFeedPublish(b *testing.B) {
179+
feed := fastfeed.NewTxFastFeed()
180+
181+
key, _ := crypto.GenerateKey()
182+
signer := types.LatestSigner(params.TestChainConfig)
183+
tx := types.MustSignNewTx(key, signer, &types.LegacyTx{
184+
Nonce: 0,
185+
GasPrice: big.NewInt(1000000000),
186+
Gas: 21000,
187+
To: &common.Address{1},
188+
Value: big.NewInt(1000),
189+
})
190+
191+
b.ResetTimer()
192+
for i := 0; i < b.N; i++ {
193+
feed.Publish(tx, fastfeed.TxEventAdded)
194+
}
195+
}
196+
197+
func BenchmarkTxFastFeedLatency(b *testing.B) {
198+
feed := fastfeed.NewTxFastFeed()
199+
200+
sub, err := feed.Subscribe(nil)
201+
if err != nil {
202+
b.Fatalf("Failed to subscribe: %v", err)
203+
}
204+
defer sub.Unsubscribe()
205+
206+
key, _ := crypto.GenerateKey()
207+
signer := types.LatestSigner(params.TestChainConfig)
208+
209+
// Pre-generate transactions
210+
txs := make([]*types.Transaction, b.N)
211+
for i := 0; i < b.N; i++ {
212+
txs[i] = types.MustSignNewTx(key, signer, &types.LegacyTx{
213+
Nonce: uint64(i),
214+
GasPrice: big.NewInt(1000000000),
215+
Gas: 21000,
216+
To: &common.Address{1},
217+
Value: big.NewInt(1000),
218+
})
219+
}
220+
221+
var maxLatency time.Duration
222+
var totalLatency time.Duration
223+
224+
b.ResetTimer()
225+
for i := 0; i < b.N; i++ {
226+
start := time.Now()
227+
feed.Publish(txs[i], fastfeed.TxEventAdded)
228+
229+
select {
230+
case <-sub.Events():
231+
latency := time.Since(start)
232+
totalLatency += latency
233+
if latency > maxLatency {
234+
maxLatency = latency
235+
}
236+
case <-time.After(100 * time.Millisecond):
237+
b.Fatalf("Timeout waiting for event %d", i)
238+
}
239+
}
240+
b.StopTimer()
241+
242+
avgLatency := totalLatency / time.Duration(b.N)
243+
b.ReportMetric(float64(avgLatency.Nanoseconds()), "ns/event")
244+
b.ReportMetric(float64(maxLatency.Nanoseconds())/1000, "μs-max")
245+
}
246+

‎core/txpool/txpool.go‎

Lines changed: 5 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -355,6 +355,11 @@ func (p *TxPool) Add(txs []*types.Transaction, sync bool) []error {
355355
// Find which subpool handled it and pull in the corresponding error
356356
errs[i] = errsets[split][0]
357357
errsets[split] = errsets[split][1:]
358+
359+
// Publish to fast feed if transaction was accepted
360+
if errs[i] == nil {
361+
p.fastFeed.Publish(txs[i], fastfeed.TxEventAdded)
362+
}
358363
}
359364
return errs
360365
}

0 commit comments

Comments
 (0)