Skip to content

Commit 0f6910c

Browse files
ucwongucwong
committed
batch state prefetch
1 parent e70d2ea commit 0f6910c

3 files changed

Lines changed: 86 additions & 69 deletions

File tree

‎core/state_prefetcher.go‎

Lines changed: 46 additions & 30 deletions
Original file line numberDiff line numberDiff line change
@@ -17,13 +17,16 @@
1717
package core
1818

1919
import (
20+
"runtime"
2021
"sync/atomic"
2122

2223
"github.com/CortexFoundation/CortexTheseus/consensus"
2324
"github.com/CortexFoundation/CortexTheseus/core/state"
2425
"github.com/CortexFoundation/CortexTheseus/core/types"
2526
"github.com/CortexFoundation/CortexTheseus/core/vm"
2627
"github.com/CortexFoundation/CortexTheseus/params"
28+
29+
"golang.org/x/sync/errgroup"
2730
)
2831

2932
// statePrefetcher is a basic Prefetcher, which blindly executes a block on top
@@ -49,42 +52,55 @@ func newStatePrefetcher(config *params.ChainConfig, bc *BlockChain, engine conse
4952
// only goal is to pre-cache transaction signatures and state trie nodes.
5053
func (p *statePrefetcher) Prefetch(block *types.Block, statedb *state.StateDB, cfg vm.Config, interrupt *atomic.Bool) {
5154
var (
52-
header = block.Header()
53-
gaspool = new(GasPool).AddGas(block.GasLimit())
54-
blockContext = NewCVMBlockContext(header, p.bc, nil)
55-
cvm = vm.NewCVM(blockContext, statedb, p.config, cfg)
56-
signer = types.MakeSigner(p.config, header.Number, header.Time)
57-
quotaPool = NewQuotaPool(header.Quota)
55+
fails atomic.Int64
56+
header = block.Header()
57+
gaspool = new(GasPool).AddGas(block.GasLimit())
58+
signer = types.MakeSigner(p.config, header.Number, header.Time)
59+
workers errgroup.Group
60+
quotaPool = NewQuotaPool(header.Quota)
5861
)
62+
63+
workers.SetLimit(max(1, 4*runtime.NumCPU()/5)) // Aggressively run the prefetching
64+
5965
if err := quotaPool.SubQuota(header.QuotaUsed); err != nil {
6066
return
6167
}
68+
6269
// Iterate over and process the individual transactions
63-
byzantium := p.config.IsByzantium(block.Number())
6470
for i, tx := range block.Transactions() {
65-
// If block precaching was interrupted, abort
66-
if interrupt != nil && interrupt.Load() {
67-
return
68-
}
69-
// Convert the transaction into an executable message and pre-cache its sender
70-
msg, err := TransactionToMessage(tx, signer)
71-
if err != nil {
72-
return // Also invalid block, bail out
73-
}
74-
statedb.SetTxContext(tx.Hash(), i)
71+
stateCpy := statedb.Copy() // closure
72+
workers.Go(func() error {
73+
// If block precaching was interrupted, abort
74+
if interrupt != nil && interrupt.Load() {
75+
return nil
76+
}
77+
// We attempt to apply a transaction. The goal is not to execute
78+
// the transaction successfully, rather to warm up touched data slots.
79+
cvm := vm.NewCVM(NewCVMBlockContext(header, p.bc, nil), stateCpy, p.config, cfg)
7580

76-
// We attempt to apply a transaction. The goal is not to execute
77-
// the transaction successfully, rather to warm up touched data slots.
78-
if _, err := ApplyMessage(cvm, msg, gaspool, quotaPool); err != nil {
79-
return // Ugh, something went horribly wrong, bail out
80-
}
81-
// If we're pre-byzantium, pre-load trie nodes for the intermediate root
82-
if !byzantium {
83-
statedb.IntermediateRoot(true)
84-
}
85-
}
86-
// If were post-byzantium, pre-load trie nodes for the final root hash
87-
if byzantium {
88-
statedb.IntermediateRoot(true)
81+
// Convert the transaction into an executable message and pre-cache its sender
82+
msg, err := TransactionToMessage(tx, signer)
83+
if err != nil {
84+
fails.Add(1)
85+
return nil // Also invalid block, bail out
86+
}
87+
// Disable the nonce check
88+
msg.SkipNonceChecks = true
89+
90+
stateCpy.SetTxContext(tx.Hash(), i)
91+
92+
if _, err := ApplyMessage(cvm, msg, gaspool, quotaPool); err != nil {
93+
fails.Add(1)
94+
return nil // Ugh, something went horribly wrong, bail out
95+
}
96+
// If we're pre-byzantium, pre-load trie nodes for the intermediate root
97+
stateCpy.IntermediateRoot(true)
98+
99+
return nil
100+
})
89101
}
102+
workers.Wait()
103+
//blockPrefetchTxsValidMeter.Mark(int64(len(block.Transactions())) - fails.Load())
104+
//blockPrefetchTxsInvalidMeter.Mark(fails.Load())
105+
return
90106
}

‎core/state_quota.go‎

Lines changed: 3 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -33,7 +33,7 @@ var (
3333
big0 = big.NewInt(0)
3434
)
3535

36-
func (st *StateTransition) preQuotaCheck() error {
36+
func (st *stateTransition) preQuotaCheck() error {
3737
if st.uploading() {
3838
// log.Debug("state_transition", "uploading", st.uploading(), "st.state.GetNum(st.to())", st.state.GetNum(st.to()))
3939
if st.state.GetNum(st.to()).Cmp(big0) <= 0 {
@@ -73,7 +73,7 @@ func (st *StateTransition) preQuotaCheck() error {
7373
return nil
7474
}
7575

76-
func (st *StateTransition) quotaCalculate() (quota uint64, err error) {
76+
func (st *stateTransition) quotaCalculate() (quota uint64, err error) {
7777
if st.uploading() {
7878
cur := st.state.Upload(st.to()).Uint64()
7979
if cur > 0 {
@@ -126,7 +126,7 @@ func (st *StateTransition) quotaCalculate() (quota uint64, err error) {
126126
return
127127
}
128128

129-
func (st *StateTransition) modelGasCalculate(gu uint64) (uint64, error) {
129+
func (st *stateTransition) modelGasCalculate(gu uint64) (uint64, error) {
130130
for addr, mgas := range st.modelGas {
131131
if mgas > params.MODEL_GAS_UP_LIMIT {
132132
continue

‎core/state_transition.go‎

Lines changed: 37 additions & 36 deletions
Original file line numberDiff line numberDiff line change
@@ -93,8 +93,8 @@ The state transitioning model does all the necessary work to work out a valid ne
9393
6) Derive new state root
9494
*/
9595

96-
// StateTransition is the state of current tx in vm
97-
type StateTransition struct {
96+
// stateTransition is the state of current tx in vm
97+
type stateTransition struct {
9898
gp *GasPool
9999
qp *QuotaPool
100100
msg *Message
@@ -106,27 +106,27 @@ type StateTransition struct {
106106
}
107107

108108
type Message struct {
109-
To *common.Address
110-
From common.Address
111-
Nonce uint64
112-
Value *big.Int
113-
GasLimit uint64
114-
GasPrice *big.Int
115-
Data []byte
116-
SkipAccountChecks bool
109+
To *common.Address
110+
From common.Address
111+
Nonce uint64
112+
Value *big.Int
113+
GasLimit uint64
114+
GasPrice *big.Int
115+
Data []byte
116+
SkipNonceChecks bool
117117
}
118118

119119
// XXX Rename message to something less arbitrary?
120120
// TransactionToMessage converts a transaction into a Message.
121121
func TransactionToMessage(tx *types.Transaction, s types.Signer) (*Message, error) {
122122
msg := &Message{
123-
Nonce: tx.Nonce(),
124-
GasLimit: tx.Gas(),
125-
GasPrice: new(big.Int).Set(tx.GasPrice()),
126-
To: tx.To(),
127-
Value: tx.Value(),
128-
Data: tx.Data(),
129-
SkipAccountChecks: false,
123+
Nonce: tx.Nonce(),
124+
GasLimit: tx.Gas(),
125+
GasPrice: new(big.Int).Set(tx.GasPrice()),
126+
To: tx.To(),
127+
Value: tx.Value(),
128+
Data: tx.Data(),
129+
SkipNonceChecks: false,
130130
}
131131

132132
var err error
@@ -171,9 +171,9 @@ func IntrinsicGas(data []byte, contractCreation, upload, isHomestead, isEIP2028
171171
return gas, nil
172172
}
173173

174-
// NewStateTransition initialises and returns a new state transition object.
175-
func NewStateTransition(cvm *vm.CVM, msg *Message, gp *GasPool, qp *QuotaPool) *StateTransition {
176-
return &StateTransition{
174+
// newStateTransition initialises and returns a new state transition object.
175+
func newStateTransition(cvm *vm.CVM, msg *Message, gp *GasPool, qp *QuotaPool) *stateTransition {
176+
return &stateTransition{
177177
gp: gp,
178178
qp: qp,
179179
cvm: cvm,
@@ -191,18 +191,18 @@ func NewStateTransition(cvm *vm.CVM, msg *Message, gp *GasPool, qp *QuotaPool) *
191191
// state and would never be accepted within a block.
192192
func ApplyMessage(cvm *vm.CVM, msg *Message, gp *GasPool, qp *QuotaPool) (*ExecutionResult, error) {
193193
cvm.SetTxContext(NewCVMTxContext(msg))
194-
return NewStateTransition(cvm, msg, gp, qp).execute()
194+
return newStateTransition(cvm, msg, gp, qp).execute()
195195
}
196196

197197
// to returns the recipient of the message.
198-
func (st *StateTransition) to() common.Address {
198+
func (st *stateTransition) to() common.Address {
199199
if st.msg == nil || st.msg.To == nil /* contract creation */ {
200200
return common.Address{}
201201
}
202202
return *st.msg.To
203203
}
204204

205-
//func (st *StateTransition) useGas(amount uint64) error {
205+
//func (st *stateTransition) useGas(amount uint64) error {
206206
// if st.gasRemaining < amount {
207207
// return vm.ErrOutOfGas
208208
// }
@@ -211,7 +211,7 @@ func (st *StateTransition) to() common.Address {
211211
// return nil
212212
//}
213213

214-
func (st *StateTransition) buyGas() error {
214+
func (st *stateTransition) buyGas() error {
215215
mgval := new(big.Int).Mul(new(big.Int).SetUint64(st.msg.GasLimit), st.msg.GasPrice)
216216
if have, want := st.state.GetBalance(st.msg.From), mgval; have.Cmp(want) < 0 {
217217
return fmt.Errorf("%w: address %v have %v want %v gas %v price %v", errInsufficientBalanceForGas, st.msg.From.Hex(), have, want, st.msg.GasLimit, st.msg.GasPrice)
@@ -227,16 +227,17 @@ func (st *StateTransition) buyGas() error {
227227
}
228228

229229
// var confirmTime = params.CONFIRM_TIME * time.Second //-3600 * 24 * 30 * time.Second
230-
func (st *StateTransition) preCheck() error {
230+
func (st *stateTransition) preCheck() error {
231231
// Make sure this transaction's nonce is correct.
232-
if !st.msg.SkipAccountChecks {
233-
stNonce := st.state.GetNonce(st.msg.From)
234-
if msgNonce := st.msg.Nonce; stNonce < msgNonce {
235-
return fmt.Errorf("%w: address %v, tx: %d state: %d", ErrNonceTooHigh, st.msg.From.Hex(), msgNonce, stNonce)
232+
msg := st.msg
233+
if !msg.SkipNonceChecks {
234+
stNonce := st.state.GetNonce(msg.From)
235+
if msgNonce := msg.Nonce; stNonce < msgNonce {
236+
return fmt.Errorf("%w: address %v, tx: %d state: %d", ErrNonceTooHigh, msg.From.Hex(), msgNonce, stNonce)
236237
} else if stNonce > msgNonce {
237-
return fmt.Errorf("%w: address %v, tx: %d state: %d", ErrNonceTooLow, st.msg.From.Hex(), msgNonce, stNonce)
238+
return fmt.Errorf("%w: address %v, tx: %d state: %d", ErrNonceTooLow, msg.From.Hex(), msgNonce, stNonce)
238239
} else if stNonce+1 < stNonce {
239-
return fmt.Errorf("%w: address %v, nonce: %d", ErrNonceMax, st.msg.From.Hex(), stNonce)
240+
return fmt.Errorf("%w: address %v, nonce: %d", ErrNonceMax, msg.From.Hex(), stNonce)
240241
}
241242
}
242243

@@ -249,7 +250,7 @@ func (st *StateTransition) preCheck() error {
249250

250251
/*const interv = 5
251252
252-
func (st *StateTransition) TorrentSync(meta common.Address, dir string, errCh chan error) {
253+
func (st *stateTransition) TorrentSync(meta common.Address, dir string, errCh chan error) {
253254
street := big.NewInt(0).Sub(st.cvm.PeekNumber, st.cvm.BlockNumber)
254255
point := big.NewInt(time.Now().Add(confirmTime).Unix())
255256
if point.Cmp(st.cvm.Context.Time) > 0 || street.Cmp(big.NewInt(params.CONFIRM_BLOCKS)) > 0 {
@@ -292,7 +293,7 @@ func (st *StateTransition) TorrentSync(meta common.Address, dir string, errCh ch
292293
// execute will transition the state by applying the current message and
293294
// returning the result including the used gas. It returns an error if failed.
294295
// An error indicates a consensus issue.
295-
func (st *StateTransition) execute() (*ExecutionResult, error) {
296+
func (st *stateTransition) execute() (*ExecutionResult, error) {
296297
if err := st.preCheck(); err != nil {
297298
return nil, err
298299
}
@@ -416,11 +417,11 @@ func (st *StateTransition) execute() (*ExecutionResult, error) {
416417
}
417418

418419
// vote to model
419-
func (st *StateTransition) uploading() bool {
420+
func (st *stateTransition) uploading() bool {
420421
return st.msg != nil && st.msg.To != nil && st.msg.Value.Sign() == 0 && st.state.Uploading(st.to()) // && st.gasRemaining >= params.UploadGas
421422
}
422423

423-
func (st *StateTransition) refundGas() uint64 {
424+
func (st *stateTransition) refundGas() uint64 {
424425
// Apply refund counter, capped to half of the used gas.
425426
refund := st.gasUsed() / 2
426427
if refund > st.state.GetRefund() {
@@ -440,6 +441,6 @@ func (st *StateTransition) refundGas() uint64 {
440441
}
441442

442443
// gasUsed returns the amount of gas used up by the state transition.
443-
func (st *StateTransition) gasUsed() uint64 {
444+
func (st *stateTransition) gasUsed() uint64 {
444445
return st.initialGas - st.gasRemaining
445446
}

0 commit comments

Comments
 (0)