feat: implement lock-free transaction fast feed with sub-100μs propagation target including lock-free SPMC ring buffer for zero-copy transaction events, binary fixed-size event layout for CPU cache efficiency, selective filtering by address and gas price, multiple concurrent consumer support with independent read positions, and integration foundation in txpool

core/txpool/fastfeed/feed.go

@@ -0,0 +1,289 @@
+// Copyright 2024 The go-ethereum Authors
+// This file is part of the go-ethereum library.
+//
+// The go-ethereum library is free software: you can redistribute it and/or modify
+// it under the terms of the GNU Lesser General Public License as published by
+// the Free Software Foundation, either version 3 of the License, or
+// (at your option) any later version.
+//
+// The go-ethereum library is distributed in the hope that it will be useful,
+// but WITHOUT ANY WARRANTY; without even the implied warranty of
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+// GNU Lesser General Public License for more details.
+//
+// You should have received a copy of the GNU Lesser General Public License
+// along with the go-ethereum library. If not, see <http://www.gnu.org/licenses/>.
+
+package fastfeed
+
+import (
+	"sync"
+	"sync/atomic"
+	"time"
+	"unsafe"
+
+	"errors"
+	
+	"github.com/ethereum/go-ethereum/common"
+	"github.com/ethereum/go-ethereum/core/types"
+	"github.com/ethereum/go-ethereum/log"
+)
+
+const (
+	// DefaultBufferSize is the default ring buffer capacity (must be power of 2)
+	DefaultBufferSize = 16384
+	
+	// MaxReaders is the maximum number of concurrent consumers
+	MaxReaders = 64
+	
+	// TxEventSize is the size of a transaction event in bytes
+	TxEventSize = 184 // 32 (hash) + 20 (from) + 20 (to) + 32 (value) + 32 (gasPrice) + 8 (nonce) + 8 (gas) + 4 (type) + 8 (timestamp) + 20 (padding)
+)
+
+// TxEventType represents the type of transaction event
+type TxEventType uint8
+
+const (
+	TxEventAdded TxEventType = iota
+	TxEventRemoved
+	TxEventReplaced
+)
+
+// TxEvent is a fixed-size transaction event optimized for zero-copy access.
+// Layout is designed for CPU cache efficiency and minimal memory access.
+type TxEvent struct {
+	Hash      [32]byte    // Transaction hash
+	From      [20]byte    // Sender address
+	To        [20]byte    // Recipient address (0x0 for contract creation)
+	Value     [32]byte    // Transfer value
+	GasPrice  [32]byte    // Gas price or maxFeePerGas for EIP-1559
+	Nonce     uint64      // Sender nonce
+	Gas       uint64      // Gas limit
+	Type      uint8       // Transaction type
+	EventType TxEventType // Event type (added/removed/replaced)
+	Timestamp uint64      // Event timestamp (nanoseconds)
+	_         [6]byte     // Padding for alignment
+}
+
+// TxFilter defines filtering criteria for transaction events.
+type TxFilter struct {
+	// Addresses to watch (empty = all addresses)
+	Addresses map[common.Address]struct{}
+	
+	// Contract methods to watch (first 4 bytes of calldata)
+	Methods map[[4]byte]struct{}
+	
+	// Minimum gas price filter
+	MinGasPrice uint64
+	
+	// Transaction types to include
+	Types map[uint8]struct{}
+}
+
+// Matches returns true if the transaction matches the filter.
+func (f *TxFilter) Matches(event *TxEvent) bool {
+	// Check addresses
+	if len(f.Addresses) > 0 {
+		fromAddr := common.BytesToAddress(event.From[:])
+		toAddr := common.BytesToAddress(event.To[:])
+		_, fromMatch := f.Addresses[fromAddr]
+		_, toMatch := f.Addresses[toAddr]
+		if !fromMatch && !toMatch {
+			return false
+		}
+	}
+	
+	// Check transaction type
+	if len(f.Types) > 0 {
+		if _, ok := f.Types[event.Type]; !ok {
+			return false
+		}
+	}
+	
+	// Check gas price
+	if f.MinGasPrice > 0 {
+		// Simple comparison of first 8 bytes as uint64
+		gasPrice := uint64(event.GasPrice[24])<<56 | 
+		           uint64(event.GasPrice[25])<<48 | 
+		           uint64(event.GasPrice[26])<<40 | 
+		           uint64(event.GasPrice[27])<<32 |
+		           uint64(event.GasPrice[28])<<24 | 
+		           uint64(event.GasPrice[29])<<16 | 
+		           uint64(event.GasPrice[30])<<8 | 
+		           uint64(event.GasPrice[31])
+		if gasPrice < f.MinGasPrice {
+			return false
+		}
+	}
+	
+	return true
+}
+
+// TxFastFeed is a high-performance transaction event feed using lock-free ring buffers.
+type TxFastFeed struct {
+	ring     *RingBuffer
+	mu       sync.RWMutex
+	filters  map[int]*TxFilter
+	nextID   int
+	enabled  atomic.Bool
+	
+	// Metrics
+	eventsPublished atomic.Uint64
+	eventsDropped   atomic.Uint64
+	lastPublish     atomic.Int64
+}
+
+// NewTxFastFeed creates a new fast transaction feed.
+func NewTxFastFeed() *TxFastFeed {
+	feed := &TxFastFeed{
+		ring:    NewRingBuffer(DefaultBufferSize, MaxReaders),
+		filters: make(map[int]*TxFilter),
+	}
+	feed.enabled.Store(true)
+	return feed
+}
+
+// Publish publishes a transaction event to all subscribers.
+func (f *TxFastFeed) Publish(tx *types.Transaction, eventType TxEventType) {
+	if !f.enabled.Load() {
+		return
+	}
+	
+	// Convert transaction to fixed-size event
+	event := f.txToEvent(tx, eventType)
+	
+	// Write to ring buffer
+	eventPtr := unsafe.Pointer(&event)
+	if !f.ring.Write(eventPtr) {
+		f.eventsDropped.Add(1)
+		log.Warn("Fast feed buffer full, event dropped", "hash", tx.Hash())
+		return
+	}
+	
+	f.eventsPublished.Add(1)
+	f.lastPublish.Store(time.Now().UnixNano())
+}
+
+// Subscribe creates a new subscription with optional filtering.
+func (f *TxFastFeed) Subscribe(filter *TxFilter) (*Subscription, error) {
+	f.mu.Lock()
+	defer f.mu.Unlock()
+	
+	if f.nextID >= MaxReaders {
+		return nil, ErrTooManySubscribers
+	}
+	
+	id := f.nextID
+	f.nextID++
+	
+	if filter != nil {
+		f.filters[id] = filter
+	}
+	
+	sub := &Subscription{
+		id:     id,
+		feed:   f,
+		events: make(chan *TxEvent, 256),
+		quit:   make(chan struct{}),
+	}
+	
+	// Reset reader position to current
+	f.ring.Reset(id)
+	
+	// Start event delivery goroutine
+	go sub.deliver()
+	
+	return sub, nil
+}
+
+// txToEvent converts a transaction to a fixed-size event.
+func (f *TxFastFeed) txToEvent(tx *types.Transaction, eventType TxEventType) TxEvent {
+	var event TxEvent
+	
+	// Hash
+	copy(event.Hash[:], tx.Hash().Bytes())
+	
+	// From (will be filled by caller if available)
+	// We don't compute sender here to avoid expensive ECDSA recovery
+	
+	// To
+	if to := tx.To(); to != nil {
+		copy(event.To[:], to.Bytes())
+	}
+	
+	// Value
+	if value := tx.Value(); value != nil {
+		copy(event.Value[:], value.Bytes())
+	}
+	
+	// Gas price
+	if gasPrice := tx.GasPrice(); gasPrice != nil {
+		copy(event.GasPrice[:], gasPrice.Bytes())
+	}
+	
+	// Other fields
+	event.Nonce = tx.Nonce()
+	event.Gas = tx.Gas()
+	event.Type = tx.Type()
+	event.EventType = eventType
+	event.Timestamp = uint64(time.Now().UnixNano())
+	
+	return event
+}
+
+// PublishWithSender publishes a transaction event with a known sender.
+func (f *TxFastFeed) PublishWithSender(tx *types.Transaction, from common.Address, eventType TxEventType) {
+	if !f.enabled.Load() {
+		return
+	}
+	
+	event := f.txToEvent(tx, eventType)
+	copy(event.From[:], from.Bytes())
+	
+	eventPtr := unsafe.Pointer(&event)
+	if !f.ring.Write(eventPtr) {
+		f.eventsDropped.Add(1)
+		log.Warn("Fast feed buffer full, event dropped", "hash", tx.Hash())
+		return
+	}
+	
+	f.eventsPublished.Add(1)
+	f.lastPublish.Store(time.Now().UnixNano())
+}
+
+// Enable enables the fast feed.
+func (f *TxFastFeed) Enable() {
+	f.enabled.Store(true)
+}
+
+// Disable disables the fast feed.
+func (f *TxFastFeed) Disable() {
+	f.enabled.Store(false)
+}
+
+// Stats returns feed statistics.
+type FeedStats struct {
+	BufferStats     BufferStats
+	EventsPublished uint64
+	EventsDropped   uint64
+	LastPublishNs   int64
+	Subscribers     int
+}
+
+// Stats returns current feed statistics.
+func (f *TxFastFeed) Stats() FeedStats {
+	f.mu.RLock()
+	subscribers := len(f.filters)
+	f.mu.RUnlock()
+	
+	return FeedStats{
+		BufferStats:     f.ring.Stats(),
+		EventsPublished: f.eventsPublished.Load(),
+		EventsDropped:   f.eventsDropped.Load(),
+		LastPublishNs:   f.lastPublish.Load(),
+		Subscribers:     subscribers,
+	}
+}
+
+var ErrTooManySubscribers = errors.New("too many subscribers")
+

core/txpool/fastfeed/ringbuffer.go

@@ -0,0 +1,215 @@
+// Copyright 2024 The go-ethereum Authors
+// This file is part of the go-ethereum library.
+//
+// The go-ethereum library is free software: you can redistribute it and/or modify
+// it under the terms of the GNU Lesser General Public License as published by
+// the Free Software Foundation, either version 3 of the License, or
+// (at your option) any later version.
+//
+// The go-ethereum library is distributed in the hope that it will be useful,
+// but WITHOUT ANY WARRANTY; without even the implied warranty of
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+// GNU Lesser General Public License for more details.
+//
+// You should have received a copy of the GNU Lesser General Public License
+// along with the go-ethereum library. If not, see <http://www.gnu.org/licenses/>.
+
+package fastfeed
+
+import (
+	"sync/atomic"
+	"unsafe"
+)
+
+// RingBuffer is a lock-free single-producer, multiple-consumer ring buffer
+// optimized for low-latency transaction propagation.
+type RingBuffer struct {
+	buffer   []unsafe.Pointer
+	capacity uint64
+	mask     uint64
+	
+	// Writer position (single producer)
+	writePos atomic.Uint64
+	
+	// Padding to prevent false sharing
+	_ [56]byte
+	
+	// Reader positions (multiple consumers)
+	// Each consumer maintains its own read position
+	readPositions []atomic.Uint64
+	maxReaders    int
+}
+
+// NewRingBuffer creates a new ring buffer with the given capacity.
+// Capacity must be a power of 2 for efficient masking.
+func NewRingBuffer(capacity int, maxReaders int) *RingBuffer {
+	if capacity&(capacity-1) != 0 {
+		panic("capacity must be a power of 2")
+	}
+	
+	rb := &RingBuffer{
+		buffer:        make([]unsafe.Pointer, capacity),
+		capacity:      uint64(capacity),
+		mask:          uint64(capacity - 1),
+		maxReaders:    maxReaders,
+		readPositions: make([]atomic.Uint64, maxReaders),
+	}
+	
+	return rb
+}
+
+// Write adds a new entry to the ring buffer.
+// Returns false if the buffer is full (slowest reader is too far behind).
+func (rb *RingBuffer) Write(data unsafe.Pointer) bool {
+	writePos := rb.writePos.Load()
+	
+	// Check if we would overwrite unread data
+	minReadPos := rb.getMinReadPos()
+	if writePos-minReadPos >= rb.capacity {
+		// Buffer is full, would overwrite unread data
+		return false
+	}
+	
+	// Write data
+	idx := writePos & rb.mask
+	atomic.StorePointer(&rb.buffer[idx], data)
+	
+	// Advance write position
+	rb.writePos.Store(writePos + 1)
+	
+	return true
+}
+
+// Read reads the next entry for the given reader ID.
+// Returns nil if no new data is available.
+func (rb *RingBuffer) Read(readerID int) unsafe.Pointer {
+	if readerID >= rb.maxReaders {
+		return nil
+	}
+	
+	readPos := rb.readPositions[readerID].Load()
+	writePos := rb.writePos.Load()
+	
+	// Check if data is available
+	if readPos >= writePos {
+		return nil
+	}
+	
+	// Check if data hasn't been overwritten
+	if writePos-readPos > rb.capacity {
+		// Data was overwritten, skip to oldest available
+		readPos = writePos - rb.capacity
+		rb.readPositions[readerID].Store(readPos)
+	}
+	
+	// Read data
+	idx := readPos & rb.mask
+	data := atomic.LoadPointer(&rb.buffer[idx])
+	
+	// Advance read position
+	rb.readPositions[readerID].Store(readPos + 1)
+	
+	return data
+}
+
+// Peek reads the next entry without advancing the read position.
+func (rb *RingBuffer) Peek(readerID int) unsafe.Pointer {
+	if readerID >= rb.maxReaders {
+		return nil
+	}
+	
+	readPos := rb.readPositions[readerID].Load()
+	writePos := rb.writePos.Load()
+	
+	if readPos >= writePos {
+		return nil
+	}
+	
+	if writePos-readPos > rb.capacity {
+		return nil
+	}
+	
+	idx := readPos & rb.mask
+	return atomic.LoadPointer(&rb.buffer[idx])
+}
+
+// Available returns the number of entries available to read for a given reader.
+func (rb *RingBuffer) Available(readerID int) int {
+	if readerID >= rb.maxReaders {
+		return 0
+	}
+	
+	readPos := rb.readPositions[readerID].Load()
+	writePos := rb.writePos.Load()
+	
+	if readPos >= writePos {
+		return 0
+	}
+	
+	available := writePos - readPos
+	if available > rb.capacity {
+		available = rb.capacity
+	}
+	
+	return int(available)
+}
+
+// getMinReadPos returns the minimum read position across all readers.
+func (rb *RingBuffer) getMinReadPos() uint64 {
+	min := rb.writePos.Load()
+	
+	for i := 0; i < rb.maxReaders; i++ {
+		pos := rb.readPositions[i].Load()
+		if pos < min {
+			min = pos
+		}
+	}
+	
+	return min
+}
+
+// Reset resets a reader's position to the current write position.
+// Useful for catching up when a reader falls too far behind.
+func (rb *RingBuffer) Reset(readerID int) {
+	if readerID < rb.maxReaders {
+		rb.readPositions[readerID].Store(rb.writePos.Load())
+	}
+}
+
+// Stats returns statistics about the ring buffer state.
+type BufferStats struct {
+	Capacity       int
+	WritePosition  uint64
+	ReadPositions  []uint64
+	MinReadPos     uint64
+	MaxLag         uint64
+}
+
+// Stats returns current buffer statistics.
+func (rb *RingBuffer) Stats() BufferStats {
+	writePos := rb.writePos.Load()
+	readPositions := make([]uint64, rb.maxReaders)
+	minReadPos := writePos
+	
+	for i := 0; i < rb.maxReaders; i++ {
+		pos := rb.readPositions[i].Load()
+		readPositions[i] = pos
+		if pos < minReadPos {
+			minReadPos = pos
+		}
+	}
+	
+	maxLag := uint64(0)
+	if writePos > minReadPos {
+		maxLag = writePos - minReadPos
+	}
+	
+	return BufferStats{
+		Capacity:      int(rb.capacity),
+		WritePosition: writePos,
+		ReadPositions: readPositions,
+		MinReadPos:    minReadPos,
+		MaxLag:        maxLag,
+	}
+}
+

core/txpool/fastfeed/subscription.go

@@ -0,0 +1,97 @@
+// Copyright 2024 The go-ethereum Authors
+// This file is part of the go-ethereum library.
+//
+// The go-ethereum library is free software: you can redistribute it and/or modify
+// it under the terms of the GNU Lesser General Public License as published by
+// the Free Software Foundation, either version 3 of the License, or
+// (at your option) any later version.
+//
+// The go-ethereum library is distributed in the hope that it will be useful,
+// but WITHOUT ANY WARRANTY; without even the implied warranty of
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+// GNU Lesser General Public License for more details.
+//
+// You should have received a copy of the GNU Lesser General Public License
+// along with the go-ethereum library. If not, see <http://www.gnu.org/licenses/>.
+
+package fastfeed
+
+import (
+	"errors"
+	"sync"
+)
+
+var (
+	ErrSubscriptionClosed = errors.New("subscription closed")
+)
+
+// Subscription represents a subscription to the transaction fast feed.
+type Subscription struct {
+	id     int
+	feed   *TxFastFeed
+	events chan *TxEvent
+	quit   chan struct{}
+	once   sync.Once
+}
+
+// Events returns the channel that delivers transaction events.
+func (s *Subscription) Events() <-chan *TxEvent {
+	return s.events
+}
+
+// Unsubscribe unsubscribes from the feed and releases resources.
+func (s *Subscription) Unsubscribe() {
+	s.once.Do(func() {
+		close(s.quit)
+		close(s.events)
+		
+		// Remove filter
+		s.feed.mu.Lock()
+		delete(s.feed.filters, s.id)
+		s.feed.mu.Unlock()
+	})
+}
+
+// deliver reads events from the ring buffer and delivers them to the subscription channel.
+func (s *Subscription) deliver() {
+	for {
+		select {
+		case <-s.quit:
+			return
+		default:
+		}
+		
+		// Try to read from ring buffer
+		eventPtr := s.feed.ring.Read(s.id)
+		if eventPtr == nil {
+			// No data available, yield
+			continue
+		}
+		
+		// Convert pointer to event
+		event := (*TxEvent)(eventPtr)
+		
+		// Apply filter if set
+		s.feed.mu.RLock()
+		filter, hasFilter := s.feed.filters[s.id]
+		s.feed.mu.RUnlock()
+		
+		if hasFilter && !filter.Matches(event) {
+			continue
+		}
+		
+		// Copy event to avoid data races
+		eventCopy := *event
+		
+		// Try to deliver to channel
+		select {
+		case s.events <- &eventCopy:
+		case <-s.quit:
+			return
+		default:
+			// Channel full, skip this event
+			// In production, might want to track skipped events
+		}
+	}
+}
+

core/txpool/txpool.go

@@ -25,6 +25,7 @@ import (
 	"github.com/ethereum/go-ethereum/common"
 	"github.com/ethereum/go-ethereum/core"
 	"github.com/ethereum/go-ethereum/core/state"
+	"github.com/ethereum/go-ethereum/core/txpool/fastfeed"
 	"github.com/ethereum/go-ethereum/core/types"
 	"github.com/ethereum/go-ethereum/event"
 	"github.com/ethereum/go-ethereum/log"
@@ -74,6 +75,9 @@ type TxPool struct {
 	term chan struct{}           // Termination channel to detect a closed pool
 
 	sync chan chan error // Testing / simulator channel to block until internal reset is done
+	
+	// Fast feed for low-latency transaction propagation
+	fastFeed *fastfeed.TxFastFeed
 }
 
 // New creates a new transaction pool to gather, sort and filter inbound
@@ -101,6 +105,7 @@ func New(gasTip uint64, chain BlockChain, subpools []SubPool) (*TxPool, error) {
 		quit:     make(chan chan error),
 		term:     make(chan struct{}),
 		sync:     make(chan chan error),
+		fastFeed: fastfeed.NewTxFastFeed(),
 	}
 	reserver := NewReservationTracker()
 	for i, subpool := range subpools {