forks/go-ethereum
1
0
Fork 1
mirror of https://github.com/ethereum/go-ethereum.git synced 2026-08-18 18:02:24 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 1

core: use channel for promoted transactions to improve order during high volume

Browse source 
Also send via channel here too.
This commit is contained in:
Ryan Schneider 2018-04-23 14:24:52 -07:00
parent 66432f3821
commit 7b4f6c04dc
1 changed files with 22 additions and 12 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

34
core/tx_pool.go
View file

@ -192,6 +192,7 @@ type TxPool struct {
chainHeadSub event.Subscription chainHeadSub event.Subscription
signer types.Signer signer types.Signer
mu sync.RWMutex mu sync.RWMutex
promotedTxCh chan TxPreEvent
currentState *state.StateDB // Current state in the blockchain head currentState *state.StateDB // Current state in the blockchain head
pendingState *state.ManagedState // Pending state tracking virtual nonces pendingState *state.ManagedState // Pending state tracking virtual nonces
@ -219,16 +220,17 @@ func NewTxPool(config TxPoolConfig, chainconfig *params.ChainConfig, chain block
// Create the transaction pool with its initial settings // Create the transaction pool with its initial settings
pool := &TxPool{ pool := &TxPool{
config: config, config: config,
chainconfig: chainconfig, chainconfig: chainconfig,
chain: chain, chain: chain,
signer: types.NewEIP155Signer(chainconfig.ChainId), signer: types.NewEIP155Signer(chainconfig.ChainId),
pending: make(map[common.Address]*txList), pending: make(map[common.Address]*txList),
queue: make(map[common.Address]*txList), queue: make(map[common.Address]*txList),
beats: make(map[common.Address]time.Time), beats: make(map[common.Address]time.Time),
all: make(map[common.Hash]*types.Transaction), all: make(map[common.Hash]*types.Transaction),
chainHeadCh: make(chan ChainHeadEvent, chainHeadChanSize), chainHeadCh: make(chan ChainHeadEvent, chainHeadChanSize),
gasPrice: new(big.Int).SetUint64(config.PriceLimit), gasPrice: new(big.Int).SetUint64(config.PriceLimit),
promotedTxCh: make(chan TxPreEvent, config.GlobalSlots),
} }
pool.locals = newAccountSet(pool.signer) pool.locals = newAccountSet(pool.signer)
pool.priced = newTxPricedList(&pool.all) pool.priced = newTxPricedList(&pool.all)
@ -333,6 +335,10 @@ func (pool *TxPool) loop() {
} }
pool.mu.Unlock() pool.mu.Unlock()
} }
// Handle promoted transaction
case txPreEvent := <-pool.promotedTxCh:
pool.txFeed.Send(txPreEvent)
} }
} }
} }
@ -653,7 +659,9 @@ func (pool *TxPool) add(tx *types.Transaction, local bool) (bool, error) {
log.Trace("Pooled new executable transaction", "hash", hash, "from", from, "to", tx.To()) log.Trace("Pooled new executable transaction", "hash", hash, "from", from, "to", tx.To())
// We've directly injected a replacement transaction, notify subsystems // We've directly injected a replacement transaction, notify subsystems
go pool.txFeed.Send(TxPreEvent{tx}) go func(tx *types.Transaction) {
pool.promotedTxCh <- TxPreEvent{tx}
}(tx)
return old != nil, nil return old != nil, nil
} }
@ -747,7 +755,9 @@ func (pool *TxPool) promoteTx(addr common.Address, hash common.Hash, tx *types.T
pool.beats[addr] = time.Now() pool.beats[addr] = time.Now()
pool.pendingState.SetNonce(addr, tx.Nonce()+1) pool.pendingState.SetNonce(addr, tx.Nonce()+1)
go pool.txFeed.Send(TxPreEvent{tx}) go func(tx *types.Transaction) {
pool.promotedTxCh <- TxPreEvent{tx}
}(tx)
} }
// AddLocal enqueues a single transaction into the pool if it is valid, marking // AddLocal enqueues a single transaction into the pool if it is valid, marking