Skip to content
This repository has been archived by the owner on Aug 2, 2021. It is now read-only.

Commit

Permalink
swarm-network-rewrite conflicts with master (#589)
Browse files Browse the repository at this point in the history 
* core/types: avoid duplicating transactions on changing signer (#16435)

* core/state: cache missing storage entries (#16584)

* cmd/utils: point users to --syncmode under DEPRECATED (#16572)

Indicate that --light and --fast options are replaced by --syncmode

* trie: remove unused `buf` parameter (#16583)

* core, eth: fix tracer dirty finalization

* travis.yml: remove obsolete brew-cask install

* whisper: Golint fixes in whisper packages (#16637)

* vendor: fix leveldb crash when bigger than 1 TiB

* core: ensure local transactions aren't discarded as underpriced

This fixes an issue where local transactions are discarded as
underpriced when the pool and queue are full.

* evm/main: use blocknumber from genesis

* accounts: golint updates for this or self warning (#16627)

* tests: golint fixes for tests directory (#16640)

* trie: golint iterator fixes (#16639)

* internal: golint updates for this or self warning (#16634)

* core: golint updates for this or self warning (#16633)

* build: Add ldflags -s -w when building aar

Smaller size on mobile is always good.
Might also solve our maven central upload problem

* cmd/clef: documentation about setup (#16568)

clef: documentation about setup

* params: release geth 1.8.7

* VERSION, params: begin v1.8.8 release cycle

* log: changed if-else blocks to conform with golint (#16661)

* p2p: changed if-else blocks to conform with golint (#16660)

* les: changed if-else blocks to conform with golint (#16658)

* accounts: changed if-else blocks to conform with golint (#16654)

* rpc: golint error with context as last parameter (#16657)

* rpc/*: golint error with context as last parameter

* Update json.go

* metrics: golint updates for this or self warning (#16635)

* metrics/*: golint updates for this or self warning

* metrics/*: golint updates for this or self warning, updated pr from feedback

* consensus/ethash: fixed typo (#16665)

* event: golint updates for this or self warning (#16631)

* event/*: golint updates for this or self warning

* event/*: golint updates for this or self warning, pr updated per feedback

* eth: golint updates for this or self warning (#16632)

* eth/*:golint updates for this or self warning

* eth/*: golint updates for this or self warning, pr updated per feedback

* signer: fix golint errors (#16653)

* signer/*: golint fixes

Specifically naming and comment formatting for documentation

* signer/*: fixed naming error crashing build

* signer/*: corrected error

* signer/core: fix tiny error whitespace

* signer/rules: fix test refactor

* whisper/mailserver: pass init error to the caller (#16671)

* whisper/mailserver: pass init error to the caller

* whisper/mailserver: add returns to fmt.Errorf

* whisper/mailserver: check err in mailserver init test

* common: changed if-else blocks to conform with golint (#16656)

* mobile: add GetStatus Method for Receipt (#16598)

* core/rawdb: separate raw database access to own package (#16666)

* rlp: fix some golint warnings (#16659)

* p2p: fix some golint warnings (#16577)

* eth/filters: derive FilterCriteria from ethereum.FilterQuery (#16629)

* p2p/simulations/adapters: fix websocket log line parsing in exec adapter (#16667)

* build: specify the key to use when invoking gpg:sign-and-deploy-file (#16696)

* crypto: fix golint warnings (#16710)

* p2p: don't discard reason set by Disconnect (#16559)

Peer.run was discarding the reason for disconnection sent to the disc
channel by Disconnect.

* cmd: various golint fixes (#16700)

* cmd: various golint fixes

* cmd: update to pr change request

* cmd: update to pr change request

* eth: golint fixes to variable names (#16711)

* eth/filter: check nil pointer when unsubscribe (#16682)

* eth/filter: check nil pointer when unsubscribe

* eth/filters, accounts, rpc: abort system if subscribe failed

* eth/filter: add crit log before exit

* eth/filter, event: minor fixes

* whisper/shhclient: update call to shh_generateSymKeyFromPassword to pass a string (#16668)

* all: get rid of error when creating memory database (#16716)

* all: get rid of error when create mdb

* core: clean up variables definition

* all: inline mdb definition

* event: document select case slice use and add edge case test (#16680)

Feed keeps active subscription channels in a slice called 'f.sendCases'.
The Send method tracks the active cases in a local variable 'cases'
whose value is f.sendCases initially. 'cases' shrinks to a shorter
prefix of f.sendCases every time a send succeeds, moving the successful
case out of range of the active case list.

This can be confusing because the two slices share a backing array. Add
more comments to document what is going on. Also add a test for removing
a case that is in 'f.sentCases' but not 'cases'.

* travis: use Android NDK 16b (#16562)

* bmt: golint updates for this or self warning (#16628)

* bmt/*: golint updates for this or self warning

* Update bmt.go

* light: new CHT for mainnet and ropsten (#16736)

* params: release go-ethereum v1.8.8

* VERSION, params: start 1.8.9 release cycle

* accounts/abi: allow abi: tags when unpacking structs

Go code users can now tag event struct members with `abi:` to specify in what fields the event will be de-serialized.

See PR #16648 for details.

* travis: try to upgrade android builder to trusty

* p2p/enr: updates for discovery v4 compatibility (#16679)

This applies spec changes from ethereum/EIPs#1049 and adds support for
pluggable identity schemes.

Some care has been taken to make the "v4" scheme standalone. It uses
public APIs only and could be moved out of package enr at any time.

A couple of minor changes were needed to make identity schemes work:

- The sequence number is now updated in Set instead of when signing.
- Record is now copy-safe, i.e. calling Set on a shallow copy doesn't
  modify the record it was copied from.

* all: collate new transaction events together

* core, eth: minor txpool event cleanups

* travis, appveyor: bump Go release to 1.10.2

* core, consensus: fix some typos in comment code and output log

* eth: propagate blocks and transactions async

* trie: fixes to comply with golint (#16771)

* log: fixes for golint warnings (#16775)

* node: all golint warnings fixed (#16773)

* node: all golint warnings fixed

* node: rm per peter

* node: rm per peter

* vendor, ethdb: print warning log if leveldb is performing compaction (#16766)

* vendor: update leveldb package

* ethdb: print warning log if db is performing compaction

* ethdb: update annotation and log

* core/types: convert status type from uint to uint64 (#16784)

* trie: support proof generation from the iterator

* core/vm: fix typo in instructions.go (#16788)

* core: use a wrapped map to remove contention in `TxPool.Get`. (#16670)

* core: use a wrapped `map` and `sync.RWMutex` for `TxPool.all` to remove contention in `TxPool.Get`.

* core: Remove redundant `txLookup.Find` and improve comments on txLookup methods.
  • Loading branch information
@nonsense @gbalint
nonsense authored and gbalint committed May 24, 2018
1 parent 2aec12e commit 6f2d560
Show file tree
Hide file tree
Showing 7 changed files with 283 additions and 119 deletions.
21 changes: 11 additions & 10 deletions core/tx_list.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -397,13 +397,13 @@ func (h *priceHeap) Pop() interface{} {
// txPricedList is a price-sorted heap to allow operating on transactions pool
// contents in a price-incrementing way.
type txPricedList struct {
all *map[common.Hash]*types.Transaction // Pointer to the map of all transactions
items *priceHeap // Heap of prices of all the stored transactions
stales int // Number of stale price points to (re-heap trigger)
all *txLookup // Pointer to the map of all transactions
items *priceHeap // Heap of prices of all the stored transactions
stales int // Number of stale price points to (re-heap trigger)
}

// newTxPricedList creates a new price-sorted transaction heap.
func newTxPricedList(all *map[common.Hash]*types.Transaction) *txPricedList {
func newTxPricedList(all *txLookup) *txPricedList {
return &txPricedList{
all: all,
items: new(priceHeap),
Expand All @@ -425,12 +425,13 @@ func (l *txPricedList) Removed() {
return
}
// Seems we've reached a critical number of stale transactions, reheap
reheap := make(priceHeap, 0, len(*l.all))
reheap := make(priceHeap, 0, l.all.Count())

l.stales, l.items = 0, &reheap
for _, tx := range *l.all {
l.all.Range(func(hash common.Hash, tx *types.Transaction) bool {
*l.items = append(*l.items, tx)
}
return true
})
heap.Init(l.items)
}

Expand All @@ -443,7 +444,7 @@ func (l *txPricedList) Cap(threshold *big.Int, local *accountSet) types.Transact
for len(*l.items) > 0 {
// Discard stale transactions if found during cleanup
tx := heap.Pop(l.items).(*types.Transaction)
if _, ok := (*l.all)[tx.Hash()]; !ok {
if l.all.Get(tx.Hash()) == nil {
l.stales--
continue
}
Expand Down Expand Up @@ -475,7 +476,7 @@ func (l *txPricedList) Underpriced(tx *types.Transaction, local *accountSet) boo
// Discard stale price points if found at the heap start
for len(*l.items) > 0 {
head := []*types.Transaction(*l.items)[0]
if _, ok := (*l.all)[head.Hash()]; !ok {
if l.all.Get(head.Hash()) == nil {
l.stales--
heap.Pop(l.items)
continue
Expand All @@ -500,7 +501,7 @@ func (l *txPricedList) Discard(count int, local *accountSet) types.Transactions
for len(*l.items) > 0 && count > 0 {
// Discard stale transactions if found during cleanup
tx := heap.Pop(l.items).(*types.Transaction)
if _, ok := (*l.all)[tx.Hash()]; !ok {
if l.all.Get(tx.Hash()) == nil {
l.stales--
continue
}
Expand Down
130 changes: 96 additions & 34 deletions core/tx_pool.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -200,11 +200,11 @@ type TxPool struct {
locals *accountSet // Set of local transaction to exempt from eviction rules
journal *txJournal // Journal of local transaction to back up to disk

pending map[common.Address]*txList // All currently processable transactions
queue map[common.Address]*txList // Queued but non-processable transactions
beats map[common.Address]time.Time // Last heartbeat from each known account
all map[common.Hash]*types.Transaction // All transactions to allow lookups
priced *txPricedList // All transactions sorted by price
pending map[common.Address]*txList // All currently processable transactions
queue map[common.Address]*txList // Queued but non-processable transactions
beats map[common.Address]time.Time // Last heartbeat from each known account
all *txLookup // All transactions to allow lookups
priced *txPricedList // All transactions sorted by price

wg sync.WaitGroup // for shutdown sync

Expand All @@ -226,12 +226,12 @@ func NewTxPool(config TxPoolConfig, chainconfig *params.ChainConfig, chain block
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),
all: newTxLookup(),
chainHeadCh: make(chan ChainHeadEvent, chainHeadChanSize),
gasPrice: new(big.Int).SetUint64(config.PriceLimit),
}
pool.locals = newAccountSet(pool.signer)
pool.priced = newTxPricedList(&pool.all)
pool.priced = newTxPricedList(pool.all)
pool.reset(nil, chain.CurrentBlock().Header())

// If local transactions and journaling is enabled, load from disk
Expand Down Expand Up @@ -605,7 +605,7 @@ func (pool *TxPool) validateTx(tx *types.Transaction, local bool) error {
func (pool *TxPool) add(tx *types.Transaction, local bool) (bool, error) {
// If the transaction is already known, discard it
hash := tx.Hash()
if pool.all[hash] != nil {
if pool.all.Get(hash) != nil {
log.Trace("Discarding already known transaction", "hash", hash)
return false, fmt.Errorf("known transaction: %x", hash)
}
Expand All @@ -616,15 +616,15 @@ func (pool *TxPool) add(tx *types.Transaction, local bool) (bool, error) {
return false, err
}
// If the transaction pool is full, discard underpriced transactions
if uint64(len(pool.all)) >= pool.config.GlobalSlots+pool.config.GlobalQueue {
if uint64(pool.all.Count()) >= pool.config.GlobalSlots+pool.config.GlobalQueue {
// If the new transaction is underpriced, don't accept it
if !local && pool.priced.Underpriced(tx, pool.locals) {
log.Trace("Discarding underpriced transaction", "hash", hash, "price", tx.GasPrice())
underpricedTxCounter.Inc(1)
return false, ErrUnderpriced
}
// New transaction is better than our worse ones, make room for it
drop := pool.priced.Discard(len(pool.all)-int(pool.config.GlobalSlots+pool.config.GlobalQueue-1), pool.locals)
drop := pool.priced.Discard(pool.all.Count()-int(pool.config.GlobalSlots+pool.config.GlobalQueue-1), pool.locals)
for _, tx := range drop {
log.Trace("Discarding freshly underpriced transaction", "hash", tx.Hash(), "price", tx.GasPrice())
underpricedTxCounter.Inc(1)
Expand All @@ -642,11 +642,11 @@ func (pool *TxPool) add(tx *types.Transaction, local bool) (bool, error) {
}
// New transaction is better, replace old one
if old != nil {
delete(pool.all, old.Hash())
pool.all.Remove(old.Hash())
pool.priced.Removed()
pendingReplaceCounter.Inc(1)
}
pool.all[tx.Hash()] = tx
pool.all.Add(tx)
pool.priced.Put(tx)
pool.journalTx(from, tx)

Expand Down Expand Up @@ -689,12 +689,12 @@ func (pool *TxPool) enqueueTx(hash common.Hash, tx *types.Transaction) (bool, er
}
// Discard any previous transaction and mark this
if old != nil {
delete(pool.all, old.Hash())
pool.all.Remove(old.Hash())
pool.priced.Removed()
queuedReplaceCounter.Inc(1)
}
if pool.all[hash] == nil {
pool.all[hash] = tx
if pool.all.Get(hash) == nil {
pool.all.Add(tx)
pool.priced.Put(tx)
}
return old != nil, nil
Expand Down Expand Up @@ -726,22 +726,22 @@ func (pool *TxPool) promoteTx(addr common.Address, hash common.Hash, tx *types.T
inserted, old := list.Add(tx, pool.config.PriceBump)
if !inserted {
// An older transaction was better, discard this
delete(pool.all, hash)
pool.all.Remove(hash)
pool.priced.Removed()

pendingDiscardCounter.Inc(1)
return false
}
// Otherwise discard any previous transaction and mark this
if old != nil {
delete(pool.all, old.Hash())
pool.all.Remove(old.Hash())
pool.priced.Removed()

pendingReplaceCounter.Inc(1)
}
// Failsafe to work around direct pending inserts (tests)
if pool.all[hash] == nil {
pool.all[hash] = tx
if pool.all.Get(hash) == nil {
pool.all.Add(tx)
pool.priced.Put(tx)
}
// Set the potentially new pending nonce and notify any subsystems of the new tx
Expand Down Expand Up @@ -840,7 +840,7 @@ func (pool *TxPool) Status(hashes []common.Hash) []TxStatus {

status := make([]TxStatus, len(hashes))
for i, hash := range hashes {
if tx := pool.all[hash]; tx != nil {
if tx := pool.all.Get(hash); tx != nil {
from, _ := types.Sender(pool.signer, tx) // already validated
if pool.pending[from] != nil && pool.pending[from].txs.items[tx.Nonce()] != nil {
status[i] = TxStatusPending
Expand All @@ -855,24 +855,21 @@ func (pool *TxPool) Status(hashes []common.Hash) []TxStatus {
// Get returns a transaction if it is contained in the pool
// and nil otherwise.
func (pool *TxPool) Get(hash common.Hash) *types.Transaction {
pool.mu.RLock()
defer pool.mu.RUnlock()

return pool.all[hash]
return pool.all.Get(hash)
}

// removeTx removes a single transaction from the queue, moving all subsequent
// transactions back to the future queue.
func (pool *TxPool) removeTx(hash common.Hash, outofbound bool) {
// Fetch the transaction we wish to delete
tx, ok := pool.all[hash]
if !ok {
tx := pool.all.Get(hash)
if tx == nil {
return
}
addr, _ := types.Sender(pool.signer, tx) // already validated during insertion

// Remove it from the list of known transactions
delete(pool.all, hash)
pool.all.Remove(hash)
if outofbound {
pool.priced.Removed()
}
Expand Down Expand Up @@ -928,15 +925,15 @@ func (pool *TxPool) promoteExecutables(accounts []common.Address) {
for _, tx := range list.Forward(pool.currentState.GetNonce(addr)) {
hash := tx.Hash()
log.Trace("Removed old queued transaction", "hash", hash)
delete(pool.all, hash)
pool.all.Remove(hash)
pool.priced.Removed()
}
// Drop all transactions that are too costly (low balance or out of gas)
drops, _ := list.Filter(pool.currentState.GetBalance(addr), pool.currentMaxGas)
for _, tx := range drops {
hash := tx.Hash()
log.Trace("Removed unpayable queued transaction", "hash", hash)
delete(pool.all, hash)
pool.all.Remove(hash)
pool.priced.Removed()
queuedNofundsCounter.Inc(1)
}
Expand All @@ -952,7 +949,7 @@ func (pool *TxPool) promoteExecutables(accounts []common.Address) {
if !pool.locals.contains(addr) {
for _, tx := range list.Cap(int(pool.config.AccountQueue)) {
hash := tx.Hash()
delete(pool.all, hash)
pool.all.Remove(hash)
pool.priced.Removed()
queuedRateLimitCounter.Inc(1)
log.Trace("Removed cap-exceeding queued transaction", "hash", hash)
Expand Down Expand Up @@ -1001,7 +998,7 @@ func (pool *TxPool) promoteExecutables(accounts []common.Address) {
for _, tx := range list.Cap(list.Len() - 1) {
// Drop the transaction from the global pools too
hash := tx.Hash()
delete(pool.all, hash)
pool.all.Remove(hash)
pool.priced.Removed()

// Update the account nonce to the dropped transaction
Expand All @@ -1023,7 +1020,7 @@ func (pool *TxPool) promoteExecutables(accounts []common.Address) {
for _, tx := range list.Cap(list.Len() - 1) {
// Drop the transaction from the global pools too
hash := tx.Hash()
delete(pool.all, hash)
pool.all.Remove(hash)
pool.priced.Removed()

// Update the account nonce to the dropped transaction
Expand Down Expand Up @@ -1092,15 +1089,15 @@ func (pool *TxPool) demoteUnexecutables() {
for _, tx := range list.Forward(nonce) {
hash := tx.Hash()
log.Trace("Removed old pending transaction", "hash", hash)
delete(pool.all, hash)
pool.all.Remove(hash)
pool.priced.Removed()
}
// Drop all transactions that are too costly (low balance or out of gas), and queue any invalids back for later
drops, invalids := list.Filter(pool.currentState.GetBalance(addr), pool.currentMaxGas)
for _, tx := range drops {
hash := tx.Hash()
log.Trace("Removed unpayable pending transaction", "hash", hash)
delete(pool.all, hash)
pool.all.Remove(hash)
pool.priced.Removed()
pendingNofundsCounter.Inc(1)
}
Expand Down Expand Up @@ -1172,3 +1169,68 @@ func (as *accountSet) containsTx(tx *types.Transaction) bool {
func (as *accountSet) add(addr common.Address) {
as.accounts[addr] = struct{}{}
}

// txLookup is used internally by TxPool to track transactions while allowing lookup without
// mutex contention.
//
// Note, although this type is properly protected against concurrent access, it
// is **not** a type that should ever be mutated or even exposed outside of the
// transaction pool, since its internal state is tightly coupled with the pools
// internal mechanisms. The sole purpose of the type is to permit out-of-bound
// peeking into the pool in TxPool.Get without having to acquire the widely scoped
// TxPool.mu mutex.
type txLookup struct {
all map[common.Hash]*types.Transaction
lock sync.RWMutex
}

// newTxLookup returns a new txLookup structure.
func newTxLookup() *txLookup {
return &txLookup{
all: make(map[common.Hash]*types.Transaction),
}
}

// Range calls f on each key and value present in the map.
func (t *txLookup) Range(f func(hash common.Hash, tx *types.Transaction) bool) {
t.lock.RLock()
defer t.lock.RUnlock()

for key, value := range t.all {
if !f(key, value) {
break
}
}
}

// Get returns a transaction if it exists in the lookup, or nil if not found.
func (t *txLookup) Get(hash common.Hash) *types.Transaction {
t.lock.RLock()
defer t.lock.RUnlock()

return t.all[hash]
}

// Count returns the current number of items in the lookup.
func (t *txLookup) Count() int {
t.lock.RLock()
defer t.lock.RUnlock()

return len(t.all)
}

// Add adds a transaction to the lookup.
func (t *txLookup) Add(tx *types.Transaction) {
t.lock.Lock()
defer t.lock.Unlock()

t.all[tx.Hash()] = tx
}

// Remove removes a transaction from the lookup.
func (t *txLookup) Remove(hash common.Hash) {
t.lock.Lock()
defer t.lock.Unlock()

delete(t.all, hash)
}
Loading

0 comments on commit 6f2d560

Please sign in to comment.