From 7b4f6c04dc08bf729e13cbd59c2e398586f3bc5f Mon Sep 17 00:00:00 2001 From: Ryan Schneider Date: Mon, 23 Apr 2018 14:24:52 -0700 Subject: [PATCH 1/2] core: use channel for promoted transactions to improve order during high volume Also send via channel here too. --- core/tx_pool.go | 34 ++++++++++++++++++++++------------ 1 file changed, 22 insertions(+), 12 deletions(-) diff --git a/core/tx_pool.go b/core/tx_pool.go index 388b40058f..1157af50bd 100644 --- a/core/tx_pool.go +++ b/core/tx_pool.go @@ -192,6 +192,7 @@ type TxPool struct { chainHeadSub event.Subscription signer types.Signer mu sync.RWMutex + promotedTxCh chan TxPreEvent currentState *state.StateDB // Current state in the blockchain head 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 pool := &TxPool{ - config: config, - chainconfig: chainconfig, - chain: chain, - signer: types.NewEIP155Signer(chainconfig.ChainId), - pending: make(map[common.Address]*txList), - queue: make(map[common.Address]*txList), - beats: make(map[common.Address]time.Time), - all: make(map[common.Hash]*types.Transaction), - chainHeadCh: make(chan ChainHeadEvent, chainHeadChanSize), - gasPrice: new(big.Int).SetUint64(config.PriceLimit), + config: config, + chainconfig: chainconfig, + chain: chain, + signer: types.NewEIP155Signer(chainconfig.ChainId), + pending: make(map[common.Address]*txList), + queue: make(map[common.Address]*txList), + beats: make(map[common.Address]time.Time), + all: make(map[common.Hash]*types.Transaction), + chainHeadCh: make(chan ChainHeadEvent, chainHeadChanSize), + gasPrice: new(big.Int).SetUint64(config.PriceLimit), + promotedTxCh: make(chan TxPreEvent, config.GlobalSlots), } pool.locals = newAccountSet(pool.signer) pool.priced = newTxPricedList(&pool.all) @@ -333,6 +335,10 @@ func (pool *TxPool) loop() { } 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()) // 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 } @@ -747,7 +755,9 @@ func (pool *TxPool) promoteTx(addr common.Address, hash common.Hash, tx *types.T pool.beats[addr] = time.Now() 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 From 70c1f89caf18e6d9f11aef5ce4ee55628dc060f2 Mon Sep 17 00:00:00 2001 From: Felix Lange Date: Thu, 17 May 2018 13:09:48 +0200 Subject: [PATCH 2/2] core: send tx events synchronously --- core/tx_pool.go | 36 +++++++++++++----------------------- 1 file changed, 13 insertions(+), 23 deletions(-) diff --git a/core/tx_pool.go b/core/tx_pool.go index 1157af50bd..f6f2a9b356 100644 --- a/core/tx_pool.go +++ b/core/tx_pool.go @@ -192,7 +192,6 @@ type TxPool struct { chainHeadSub event.Subscription signer types.Signer mu sync.RWMutex - promotedTxCh chan TxPreEvent currentState *state.StateDB // Current state in the blockchain head pendingState *state.ManagedState // Pending state tracking virtual nonces @@ -220,17 +219,16 @@ func NewTxPool(config TxPoolConfig, chainconfig *params.ChainConfig, chain block // Create the transaction pool with its initial settings pool := &TxPool{ - config: config, - chainconfig: chainconfig, - chain: chain, - signer: types.NewEIP155Signer(chainconfig.ChainId), - pending: make(map[common.Address]*txList), - queue: make(map[common.Address]*txList), - beats: make(map[common.Address]time.Time), - all: make(map[common.Hash]*types.Transaction), - chainHeadCh: make(chan ChainHeadEvent, chainHeadChanSize), - gasPrice: new(big.Int).SetUint64(config.PriceLimit), - promotedTxCh: make(chan TxPreEvent, config.GlobalSlots), + config: config, + chainconfig: chainconfig, + chain: chain, + signer: types.NewEIP155Signer(chainconfig.ChainId), + pending: make(map[common.Address]*txList), + queue: make(map[common.Address]*txList), + beats: make(map[common.Address]time.Time), + all: make(map[common.Hash]*types.Transaction), + chainHeadCh: make(chan ChainHeadEvent, chainHeadChanSize), + gasPrice: new(big.Int).SetUint64(config.PriceLimit), } pool.locals = newAccountSet(pool.signer) pool.priced = newTxPricedList(&pool.all) @@ -335,10 +333,6 @@ func (pool *TxPool) loop() { } pool.mu.Unlock() } - - // Handle promoted transaction - case txPreEvent := <-pool.promotedTxCh: - pool.txFeed.Send(txPreEvent) } } } @@ -658,10 +652,8 @@ func (pool *TxPool) add(tx *types.Transaction, local bool) (bool, error) { log.Trace("Pooled new executable transaction", "hash", hash, "from", from, "to", tx.To()) - // We've directly injected a replacement transaction, notify subsystems - go func(tx *types.Transaction) { - pool.promotedTxCh <- TxPreEvent{tx} - }(tx) + // We've directly injected a replacement transaction, notify subsystems. + pool.txFeed.Send(TxPreEvent{tx}) return old != nil, nil } @@ -755,9 +747,7 @@ func (pool *TxPool) promoteTx(addr common.Address, hash common.Hash, tx *types.T pool.beats[addr] = time.Now() pool.pendingState.SetNonce(addr, tx.Nonce()+1) - go func(tx *types.Transaction) { - pool.promotedTxCh <- TxPreEvent{tx} - }(tx) + pool.txFeed.Send(TxPreEvent{tx}) } // AddLocal enqueues a single transaction into the pool if it is valid, marking