mirror/erigon-pulse
1
0
Fork 0
mirror of https://gitlab.com/pulsechaincom/erigon-pulse.git synced 2025-01-01 00:31:21 +00:00
Code Issues Actions Packages Projects Releases Wiki Activity

Pool: started atomic (#61)

Browse Source 
* save

* save
This commit is contained in:
Alex Sharov 2021-09-08 12:30:37 +07:00 committed by GitHub
parent b4933b94d2
commit c4efc0ea3a
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
1 changed files with 10 additions and 9 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

19
txpool/pool.go
View File

@ -505,6 +505,7 @@ func (b *ByNonce) replaceOrInsert(mt *metaTx) *metaTx {
type TxPool struct { type TxPool struct {
lock *sync.RWMutex lock *sync.RWMutex
stared atomic.Bool
protocolBaseFee atomic.Uint64 protocolBaseFee atomic.Uint64
currentBaseFee atomic.Uint64 currentBaseFee atomic.Uint64
@ -669,7 +670,7 @@ func (p *TxPool) IsLocal(idHash []byte) bool {
return ok return ok
} }
func (p *TxPool) AddNewGoodPeer(peerID PeerID) { p.recentlyConnectedPeers.AddPeer(peerID) } func (p *TxPool) AddNewGoodPeer(peerID PeerID) { p.recentlyConnectedPeers.AddPeer(peerID) }
func (p *TxPool) Started() bool { return p.protocolBaseFee.Load() > 0 } func (p *TxPool) Started() bool { return p.stared.Load() }
// Best - returns top `n` elements of pending queue // Best - returns top `n` elements of pending queue
// id doesn't perform full copy of txs, hovewer underlying elements are immutable // id doesn't perform full copy of txs, hovewer underlying elements are immutable
@ -810,11 +811,10 @@ func (p *TxPool) AddLocals(ctx context.Context, newTxs TxSlots, tx kv.Tx) ([]Dis
return nil, err return nil, err
} }
protocolBaseFee, currentBaseFee := p.protocolBaseFee.Load(), p.currentBaseFee.Load() if !p.Started() {
if protocolBaseFee == 0 || currentBaseFee == 0 { return nil, fmt.Errorf("pool not started yet")
return nil, fmt.Errorf("non-zero base fee: %d,%d", protocolBaseFee, currentBaseFee)
} }
if err := onNewTxs(tx, p.senders, newTxs, protocolBaseFee, currentBaseFee, p.pending, p.baseFee, p.queued, p.byNonce, p.byHash, p.discardLocked); err != nil { if err := onNewTxs(tx, p.senders, newTxs, p.protocolBaseFee.Load(), p.currentBaseFee.Load(), p.pending, p.baseFee, p.queued, p.byNonce, p.byHash, p.discardLocked); err != nil {
return nil, err return nil, err
} }
@ -867,11 +867,11 @@ func (p *TxPool) processRemoteTxs(ctx context.Context, tx kv.Tx) error {
return err return err
} }
protocolBaseFee, currentBaseFee := p.protocolBaseFee.Load(), p.currentBaseFee.Load() if !p.stared.Load() {
if protocolBaseFee == 0 || currentBaseFee == 0 { return fmt.Errorf("txpool not started yet")
return fmt.Errorf("non-zero base fee: %d,%d", protocolBaseFee, currentBaseFee)
} }
if err := onNewTxs(tx, p.senders, newTxs, protocolBaseFee, currentBaseFee, p.pending, p.baseFee, p.queued, p.byNonce, p.byHash, p.discardLocked); err != nil {
if err := onNewTxs(tx, p.senders, newTxs, p.protocolBaseFee.Load(), p.currentBaseFee.Load(), p.pending, p.baseFee, p.queued, p.byNonce, p.byHash, p.discardLocked); err != nil {
return err return err
} }
@ -930,6 +930,7 @@ func (p *TxPool) setBaseFee(baseFee uint64) (uint64, uint64) {
p.protocolBaseFee.Store(calcProtocolBaseFee(baseFee)) p.protocolBaseFee.Store(calcProtocolBaseFee(baseFee))
p.currentBaseFee.Store(baseFee) p.currentBaseFee.Store(baseFee)
} }
p.stared.Store(true)
return p.protocolBaseFee.Load(), p.currentBaseFee.Load() return p.protocolBaseFee.Load(), p.currentBaseFee.Load()
} }