diff options
author | rjl493456442 <garyrong0905@gmail.com> | 2018-05-10 15:04:45 +0800 |
---|---|---|
committer | Péter Szilágyi <peterke@gmail.com> | 2018-05-18 16:46:44 +0800 |
commit | a2e43d28d01ef9642c7f6992b78b86bd0696c847 (patch) | |
tree | 2f5d3444071125e84155321db6fd79d941cfee0b /core | |
parent | 6286c255f16a914b39ffd3389cba154a53e66a13 (diff) | |
download | dexon-a2e43d28d01ef9642c7f6992b78b86bd0696c847.tar dexon-a2e43d28d01ef9642c7f6992b78b86bd0696c847.tar.gz dexon-a2e43d28d01ef9642c7f6992b78b86bd0696c847.tar.bz2 dexon-a2e43d28d01ef9642c7f6992b78b86bd0696c847.tar.lz dexon-a2e43d28d01ef9642c7f6992b78b86bd0696c847.tar.xz dexon-a2e43d28d01ef9642c7f6992b78b86bd0696c847.tar.zst dexon-a2e43d28d01ef9642c7f6992b78b86bd0696c847.zip |
all: collate new transaction events together
Diffstat (limited to 'core')
-rw-r--r-- | core/events.go | 4 | ||||
-rw-r--r-- | core/tx_journal.go | 32 | ||||
-rw-r--r-- | core/tx_pool.go | 32 | ||||
-rw-r--r-- | core/tx_pool_test.go | 31 |
4 files changed, 66 insertions, 33 deletions
diff --git a/core/events.go b/core/events.go index 6f404f612..fee34b39d 100644 --- a/core/events.go +++ b/core/events.go @@ -21,8 +21,8 @@ import ( "github.com/ethereum/go-ethereum/core/types" ) -// TxPreEvent is posted when a transaction enters the transaction pool. -type TxPreEvent struct{ Tx *types.Transaction } +// TxsPreEvent is posted when a batch of transactions enter the transaction pool. +type TxsPreEvent struct{ Txs types.Transactions } // PendingLogsEvent is posted pre mining and notifies of pending logs. type PendingLogsEvent struct { diff --git a/core/tx_journal.go b/core/tx_journal.go index e872d7b53..b344690b6 100644 --- a/core/tx_journal.go +++ b/core/tx_journal.go @@ -56,7 +56,7 @@ func newTxJournal(path string) *txJournal { // load parses a transaction journal dump from disk, loading its contents into // the specified pool. -func (journal *txJournal) load(add func(*types.Transaction) error) error { +func (journal *txJournal) load(add func([]*types.Transaction) []error) error { // Skip the parsing if the journal file doens't exist at all if _, err := os.Stat(journal.path); os.IsNotExist(err) { return nil @@ -76,7 +76,22 @@ func (journal *txJournal) load(add func(*types.Transaction) error) error { stream := rlp.NewStream(input, 0) total, dropped := 0, 0 - var failure error + // flush imports a batch of transactions and bump the appropriate progress counters + flush := func(txs types.Transactions) { + errs := add(txs) + for _, err := range errs { + if err != nil { + log.Debug("Failed to add journaled transaction", "err", err) + dropped++ + } + } + } + + var ( + failure error + txs types.Transactions + ) + for { // Parse the next transaction and terminate on error tx := new(types.Transaction) @@ -86,14 +101,17 @@ func (journal *txJournal) load(add func(*types.Transaction) error) error { } break } - // Import the transaction and bump the appropriate progress counters + txs = append(txs, tx) total++ - if err = add(tx); err != nil { - log.Debug("Failed to add journaled transaction", "err", err) - dropped++ - continue + if txs.Len() > 1024 { + flush(txs) + txs = types.Transactions{} } } + if txs.Len() > 0 { + flush(txs) + txs = types.Transactions{} + } log.Info("Loaded local transaction journal", "transactions", total, "dropped", dropped) return failure diff --git a/core/tx_pool.go b/core/tx_pool.go index 388b40058..f23828cb4 100644 --- a/core/tx_pool.go +++ b/core/tx_pool.go @@ -238,7 +238,7 @@ func NewTxPool(config TxPoolConfig, chainconfig *params.ChainConfig, chain block if !config.NoLocals && config.Journal != "" { pool.journal = newTxJournal(config.Journal) - if err := pool.journal.load(pool.AddLocal); err != nil { + if err := pool.journal.load(pool.AddLocals); err != nil { log.Warn("Failed to load transaction journal", "err", err) } if err := pool.journal.rotate(pool.local()); err != nil { @@ -444,9 +444,9 @@ func (pool *TxPool) Stop() { log.Info("Transaction pool stopped") } -// SubscribeTxPreEvent registers a subscription of TxPreEvent and +// SubscribeTxPreEvent registers a subscription of TxsPreEvent and // starts sending event to the given channel. -func (pool *TxPool) SubscribeTxPreEvent(ch chan<- TxPreEvent) event.Subscription { +func (pool *TxPool) SubscribeTxPreEvent(ch chan<- TxsPreEvent) event.Subscription { return pool.scope.Track(pool.txFeed.Subscribe(ch)) } @@ -653,7 +653,7 @@ 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 pool.txFeed.Send(TxsPreEvent{types.Transactions{tx}}) return old != nil, nil } @@ -715,7 +715,7 @@ func (pool *TxPool) journalTx(from common.Address, tx *types.Transaction) { // promoteTx adds a transaction to the pending (processable) list of transactions. // // Note, this method assumes the pool lock is held! -func (pool *TxPool) promoteTx(addr common.Address, hash common.Hash, tx *types.Transaction) { +func (pool *TxPool) promoteTx(addr common.Address, hash common.Hash, tx *types.Transaction) bool { // Try to insert the transaction into the pending queue if pool.pending[addr] == nil { pool.pending[addr] = newTxList(true) @@ -729,7 +729,7 @@ func (pool *TxPool) promoteTx(addr common.Address, hash common.Hash, tx *types.T pool.priced.Removed() pendingDiscardCounter.Inc(1) - return + return false } // Otherwise discard any previous transaction and mark this if old != nil { @@ -746,8 +746,7 @@ func (pool *TxPool) promoteTx(addr common.Address, hash common.Hash, tx *types.T // Set the potentially new pending nonce and notify any subsystems of the new tx pool.beats[addr] = time.Now() pool.pendingState.SetNonce(addr, tx.Nonce()+1) - - go pool.txFeed.Send(TxPreEvent{tx}) + return true } // AddLocal enqueues a single transaction into the pool if it is valid, marking @@ -907,6 +906,7 @@ func (pool *TxPool) removeTx(hash common.Hash, outofbound bool) { // future queue to the set of pending transactions. During this process, all // invalidated transactions (low nonce, low balance) are deleted. func (pool *TxPool) promoteExecutables(accounts []common.Address) { + var promotedTxs types.Transactions // Gather all the accounts potentially needing updates if accounts == nil { accounts = make([]common.Address, 0, len(pool.queue)) @@ -937,11 +937,16 @@ func (pool *TxPool) promoteExecutables(accounts []common.Address) { queuedNofundsCounter.Inc(1) } // Gather all executable transactions and promote them - for _, tx := range list.Ready(pool.pendingState.GetNonce(addr)) { + txs := list.Ready(pool.pendingState.GetNonce(addr)) + for _, tx := range txs { hash := tx.Hash() - log.Trace("Promoting queued transaction", "hash", hash) - pool.promoteTx(addr, hash, tx) + inserted := pool.promoteTx(addr, hash, tx) + if inserted { + log.Trace("Promoting queued transaction", "hash", hash) + promotedTxs = append(promotedTxs, tx) + } } + // Drop all transactions over the allowed limit if !pool.locals.contains(addr) { for _, tx := range list.Cap(int(pool.config.AccountQueue)) { @@ -957,6 +962,11 @@ func (pool *TxPool) promoteExecutables(accounts []common.Address) { delete(pool.queue, addr) } } + // Notify subsystem for new promoted transactions. + if promotedTxs.Len() > 0 { + pool.txFeed.Send(TxsPreEvent{promotedTxs}) + } + // If the pending limit is overflown, start equalizing allowances pending := uint64(0) for _, list := range pool.pending { diff --git a/core/tx_pool_test.go b/core/tx_pool_test.go index e7f52075e..c19581e3f 100644 --- a/core/tx_pool_test.go +++ b/core/tx_pool_test.go @@ -118,21 +118,26 @@ func validateTxPoolInternals(pool *TxPool) error { // validateEvents checks that the correct number of transaction addition events // were fired on the pool's event feed. -func validateEvents(events chan TxPreEvent, count int) error { - for i := 0; i < count; i++ { +func validateEvents(events chan TxsPreEvent, count int) error { + received := 0 + for { + if received == count { + break + } select { - case <-events: + case ev := <-events: + received += ev.Txs.Len() case <-time.After(time.Second): - return fmt.Errorf("event #%d not fired", i) + return fmt.Errorf("event #%d not fired", received) } } select { - case tx := <-events: - return fmt.Errorf("more than %d events fired: %v", count, tx.Tx) + case ev := <-events: + return fmt.Errorf("more than %d events fired: %v", count, ev.Txs) case <-time.After(50 * time.Millisecond): // This branch should be "default", but it's a data race between goroutines, - // reading the event channel and pushng into it, so better wait a bit ensuring + // reading the event channel and pushing into it, so better wait a bit ensuring // really nothing gets injected. } return nil @@ -669,7 +674,7 @@ func TestTransactionGapFilling(t *testing.T) { pool.currentState.AddBalance(account, big.NewInt(1000000)) // Keep track of transaction events to ensure all executables get announced - events := make(chan TxPreEvent, testTxPoolConfig.AccountQueue+5) + events := make(chan TxsPreEvent, testTxPoolConfig.AccountQueue+5) sub := pool.txFeed.Subscribe(events) defer sub.Unsubscribe() @@ -920,7 +925,7 @@ func TestTransactionPendingLimiting(t *testing.T) { pool.currentState.AddBalance(account, big.NewInt(1000000)) // Keep track of transaction events to ensure all executables get announced - events := make(chan TxPreEvent, testTxPoolConfig.AccountQueue+5) + events := make(chan TxsPreEvent, testTxPoolConfig.AccountQueue+5) sub := pool.txFeed.Subscribe(events) defer sub.Unsubscribe() @@ -1140,7 +1145,7 @@ func TestTransactionPoolRepricing(t *testing.T) { defer pool.Stop() // Keep track of transaction events to ensure all executables get announced - events := make(chan TxPreEvent, 32) + events := make(chan TxsPreEvent, 32) sub := pool.txFeed.Subscribe(events) defer sub.Unsubscribe() @@ -1327,7 +1332,7 @@ func TestTransactionPoolUnderpricing(t *testing.T) { defer pool.Stop() // Keep track of transaction events to ensure all executables get announced - events := make(chan TxPreEvent, 32) + events := make(chan TxsPreEvent, 32) sub := pool.txFeed.Subscribe(events) defer sub.Unsubscribe() @@ -1433,7 +1438,7 @@ func TestTransactionPoolStableUnderpricing(t *testing.T) { defer pool.Stop() // Keep track of transaction events to ensure all executables get announced - events := make(chan TxPreEvent, 32) + events := make(chan TxsPreEvent, 32) sub := pool.txFeed.Subscribe(events) defer sub.Unsubscribe() @@ -1495,7 +1500,7 @@ func TestTransactionReplacement(t *testing.T) { defer pool.Stop() // Keep track of transaction events to ensure all executables get announced - events := make(chan TxPreEvent, 32) + events := make(chan TxsPreEvent, 32) sub := pool.txFeed.Subscribe(events) defer sub.Unsubscribe() |