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

save

Browse Source
This commit is contained in:
alex.sharov 2021-08-16 09:28:21 +07:00
parent c16fcc3a8d
commit ca4595a051
4 changed files with 34 additions and 20 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
txpool/fetch.go
View File

@ -412,7 +412,7 @@ func (f *Fetch) handleStateChanges(ctx context.Context, client remote.KVClient)
}
if err := f.coreDB.View(ctx, func(tx kv.Tx) error {
return f.pool.OnNewBlock(tx, diff, unwindTxs, minedTxs, req.ProtocolBaseFee, req.BlockHeight)
return f.pool.OnNewBlock(tx, diff, unwindTxs, minedTxs, req.ProtocolBaseFee, 0, req.BlockHeight)
}); err != nil {
log.Warn("onNewBlock", "err", err)
}

14
txpool/mocks_test.go
View File

@ -30,7 +30,7 @@ var _ Pool = &PoolMock{}
// IdHashKnownFunc: func(hash []byte) bool {
// panic("mock out the IdHashKnown method")
// },
// OnNewBlockFunc: func(db kv.Tx, stateChanges map[string]senderInfo, unwindTxs TxSlots, minedTxs TxSlots, protocolBaseFee uint64, blockHeight uint64) error {
// OnNewBlockFunc: func(db kv.Tx, stateChanges map[string]senderInfo, unwindTxs TxSlots, minedTxs TxSlots, protocolBaseFee uint64, pendingBaseFee uint64, blockHeight uint64) error {
// panic("mock out the OnNewBlock method")
// },
// }
@ -53,7 +53,7 @@ type PoolMock struct {
IdHashKnownFunc func(hash []byte) bool
// OnNewBlockFunc mocks the OnNewBlock method.
OnNewBlockFunc func(db kv.Tx, stateChanges map[string]senderInfo, unwindTxs TxSlots, minedTxs TxSlots, protocolBaseFee uint64, blockHeight uint64) error
OnNewBlockFunc func(db kv.Tx, stateChanges map[string]senderInfo, unwindTxs TxSlots, minedTxs TxSlots, protocolBaseFee uint64, pendingBaseFee uint64, blockHeight uint64) error
// calls tracks calls to the methods.
calls struct {
@ -91,6 +91,8 @@ type PoolMock struct {
MinedTxs TxSlots
// ProtocolBaseFee is the protocolBaseFee argument value.
ProtocolBaseFee uint64
// PendingBaseFee is the pendingBaseFee argument value.
PendingBaseFee uint64
// BlockHeight is the blockHeight argument value.
BlockHeight uint64
}
@ -240,13 +242,14 @@ func (mock *PoolMock) IdHashKnownCalls() []struct {
}
// OnNewBlock calls OnNewBlockFunc.
func (mock *PoolMock) OnNewBlock(db kv.Tx, stateChanges map[string]senderInfo, unwindTxs TxSlots, minedTxs TxSlots, protocolBaseFee uint64, blockHeight uint64) error {
func (mock *PoolMock) OnNewBlock(db kv.Tx, stateChanges map[string]senderInfo, unwindTxs TxSlots, minedTxs TxSlots, protocolBaseFee uint64, pendingBaseFee uint64, blockHeight uint64) error {
callInfo := struct {
Db kv.Tx
StateChanges map[string]senderInfo
UnwindTxs TxSlots
MinedTxs TxSlots
ProtocolBaseFee uint64
PendingBaseFee uint64
BlockHeight uint64
}{
Db: db,
@ -254,6 +257,7 @@ func (mock *PoolMock) OnNewBlock(db kv.Tx, stateChanges map[string]senderInfo, u
UnwindTxs: unwindTxs,
MinedTxs: minedTxs,
ProtocolBaseFee: protocolBaseFee,
PendingBaseFee: pendingBaseFee,
BlockHeight: blockHeight,
}
mock.lockOnNewBlock.Lock()
@ -265,7 +269,7 @@ func (mock *PoolMock) OnNewBlock(db kv.Tx, stateChanges map[string]senderInfo, u
)
return errOut
}
return mock.OnNewBlockFunc(db, stateChanges, unwindTxs, minedTxs, protocolBaseFee, blockHeight)
return mock.OnNewBlockFunc(db, stateChanges, unwindTxs, minedTxs, protocolBaseFee, pendingBaseFee, blockHeight)
}
// OnNewBlockCalls gets all the calls that were made to OnNewBlock.
@ -277,6 +281,7 @@ func (mock *PoolMock) OnNewBlockCalls() []struct {
UnwindTxs TxSlots
MinedTxs TxSlots
ProtocolBaseFee uint64
PendingBaseFee uint64
BlockHeight uint64
} {
var calls []struct {
@ -285,6 +290,7 @@ func (mock *PoolMock) OnNewBlockCalls() []struct {
UnwindTxs TxSlots
MinedTxs TxSlots
ProtocolBaseFee uint64
PendingBaseFee uint64
BlockHeight uint64
}
mock.lockOnNewBlock.RLock()

28
txpool/pool.go
View File

@ -41,7 +41,7 @@ type Pool interface {
IdHashKnown(hash []byte) bool
GetRlp(hash []byte) []byte
Add(db kv.Tx, newTxs TxSlots) error
OnNewBlock(db kv.Tx, stateChanges map[string]senderInfo, unwindTxs, minedTxs TxSlots, protocolBaseFee, blockHeight uint64) error
OnNewBlock(db kv.Tx, stateChanges map[string]senderInfo, unwindTxs, minedTxs TxSlots, protocolBaseFee, pendingBaseFee, blockHeight uint64) error
AddNewGoodPeer(peerID PeerID)
}
@ -318,7 +318,20 @@ func onNewTxs(senderInfo map[uint64]*senderInfo, newTxs TxSlots, protocolBaseFee
return nil
}
func (p *TxPool) OnNewBlock(coreDB kv.Tx, stateChanges map[string]senderInfo, unwindTxs, minedTxs TxSlots, protocolBaseFee, blockHeight uint64) error {
func (p *TxPool) setBaseFee(protocolBaseFee, pendingBaseFee uint64) (uint64, uint64) {
p.protocolBaseFee.Store(protocolBaseFee)
hasNewVal := pendingBaseFee > 0
if pendingBaseFee < protocolBaseFee {
pendingBaseFee = protocolBaseFee
}
if hasNewVal {
p.protocolBaseFee.Store(pendingBaseFee)
}
return protocolBaseFee, p.pendingBaseFee.Load()
}
func (p *TxPool) OnNewBlock(coreDB kv.Tx, stateChanges map[string]senderInfo, unwindTxs, minedTxs TxSlots, protocolBaseFee, pendingBaseFee, blockHeight uint64) error {
p.lock.Lock()
defer p.lock.Unlock()
if err := unwindTxs.Valid(); err != nil {
@ -329,11 +342,7 @@ func (p *TxPool) OnNewBlock(coreDB kv.Tx, stateChanges map[string]senderInfo, un
}
p.blockHeight.Store(blockHeight)
p.protocolBaseFee.Store(protocolBaseFee)
pendingBaseFee := p.pendingBaseFee.Load()
if pendingBaseFee < protocolBaseFee {
p.pendingBaseFee.Store(protocolBaseFee)
}
protocolBaseFee, pendingBaseFee = p.setBaseFee(protocolBaseFee, pendingBaseFee)
if err := setTxSenderID(coreDB, &p.senderID, p.senderIDs, p.senderInfo, unwindTxs); err != nil {
return err
@ -417,7 +426,7 @@ func setTxSenderID(coreDB kv.Tx, senderIDSequence *uint64, senderIDs map[string]
return nil
}
func onNewBlock(senderInfo map[uint64]*senderInfo, unwindTxs TxSlots, minedTxs []*TxSlot, protocolBaseFee, blockBaseFee uint64, pending, baseFee, queued *SubPool, byHash map[string]*metaTx, localsHistory *simplelru.LRU) error {
func onNewBlock(senderInfo map[uint64]*senderInfo, unwindTxs TxSlots, minedTxs []*TxSlot, protocolBaseFee, pendingBaseFee uint64, pending, baseFee, queued *SubPool, byHash map[string]*metaTx, localsHistory *simplelru.LRU) error {
for i := range unwindTxs.txs {
if unwindTxs.txs[i].senderID == 0 {
return fmt.Errorf("onNewBlock.unwindTxs: senderID can't be zero")
@ -473,7 +482,7 @@ func onNewBlock(senderInfo map[uint64]*senderInfo, unwindTxs TxSlots, minedTxs [
for i := range senderInfo {
// TODO: aggregate changed senders before call this func
onSenderChange(senderInfo[i], protocolBaseFee, blockBaseFee)
onSenderChange(senderInfo[i], protocolBaseFee, pendingBaseFee)
}
pending.EnforceInvariants()
@ -610,6 +619,7 @@ func onSenderChange(sender *senderInfo, protocolBaseFee, pendingBaseFee uint64)
// baseFee of the currently pending block. Set to 0 otherwise.
it.metaTx.subPool &^= EnoughFeeCapBlock
if it.metaTx.Tx.feeCap >= pendingBaseFee {
fmt.Printf("setttttt: %d,%d,%d\n", protocolBaseFee, pendingBaseFee, it.metaTx.Tx.feeCap)
it.metaTx.subPool |= EnoughFeeCapBlock
}

10
txpool/pool_fuzz_test.go
View File

@ -238,13 +238,11 @@ func FuzzOnNewBlocks11(f *testing.F) {
f.Add(u64[:], u64[:], u64[:], u64[:], sender[:], 10, 12)
f.Fuzz(func(t *testing.T, txNonce, values, tips, feeCap, sender []byte, protocolBaseFee1, pendingBaseFee1 uint8) {
t.Parallel()
protocolBaseFee, pendingBaseFee := uint64(protocolBaseFee1%16+1), uint64(pendingBaseFee1%16+1)
protocolBaseFee, pendingBaseFee := uint64(protocolBaseFee1%16+1), uint64(pendingBaseFee1%8+1)
if protocolBaseFee == 0 || pendingBaseFee == 0 {
t.Skip()
}
if pendingBaseFee < protocolBaseFee {
pendingBaseFee = protocolBaseFee
}
pendingBaseFee += protocolBaseFee
if len(sender) < 1+1+1 {
t.Skip()
}
@ -430,13 +428,13 @@ func FuzzOnNewBlocks11(f *testing.F) {
// go to first fork
//fmt.Printf("ll1: %d,%d,%d\n", pool.pending.Len(), pool.baseFee.Len(), pool.queued.Len())
unwindTxs, minedTxs1, p2pReceived, minedTxs2 := splitDataset(txs)
err = pool.OnNewBlock(nil, map[string]senderInfo{}, unwindTxs, minedTxs1, protocolBaseFee, 1)
err = pool.OnNewBlock(nil, map[string]senderInfo{}, unwindTxs, minedTxs1, protocolBaseFee, pendingBaseFee, 1)
assert.NoError(err)
check(unwindTxs, minedTxs1, "fork1")
checkNotify(unwindTxs, minedTxs1, "fork1")
// unwind everything and switch to new fork (need unwind mined now)
err = pool.OnNewBlock(nil, map[string]senderInfo{}, minedTxs1, minedTxs2, protocolBaseFee, 2)
err = pool.OnNewBlock(nil, map[string]senderInfo{}, minedTxs1, minedTxs2, protocolBaseFee, pendingBaseFee, 2)
assert.NoError(err)
check(minedTxs1, minedTxs2, "fork2")
checkNotify(minedTxs1, minedTxs2, "fork2")