feat: allow non-ccc pipelines (#781)

This commit is contained in:
Ömer Faruk Irmak 2024-05-30 14:59:37 +03:00 committed by GitHub
parent aae829c406
commit a43879b738
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
4 changed files with 97 additions and 31 deletions

View file

@ -517,7 +517,11 @@ func (w *worker) startNewPipeline(timestamp int64) {
} }
w.currentPipelineStart = time.Now() w.currentPipelineStart = time.Now()
w.currentPipeline = pipeline.NewPipeline(w.chain, w.chain.GetVMConfig(), parentState, header, nextL1MsgIndex, w.getCCC()).WithBeforeTxHook(w.beforeTxHook) pipelineCCC := w.getCCC()
if !w.isRunning() {
pipelineCCC = nil
}
w.currentPipeline = pipeline.NewPipeline(w.chain, w.chain.GetVMConfig(), parentState, header, nextL1MsgIndex, pipelineCCC).WithBeforeTxHook(w.beforeTxHook)
deadline := time.Unix(int64(header.Time), 0) deadline := time.Unix(int64(header.Time), 0)
if w.chainConfig.Clique != nil && w.chainConfig.Clique.RelaxedPeriod { if w.chainConfig.Clique != nil && w.chainConfig.Clique.RelaxedPeriod {
@ -579,11 +583,19 @@ func (w *worker) startNewPipeline(timestamp int64) {
return return
} }
} }
// pipelineCCC was nil, so the block was built for RPC purposes only. Stop the pipeline immediately
// and update the pending block.
if pipelineCCC == nil {
w.currentPipeline.Stop()
}
} }
func (w *worker) handlePipelineResult(res *pipeline.Result) error { func (w *worker) handlePipelineResult(res *pipeline.Result) error {
if !w.isRunning() { // Rows being nil without an OverflowingTx means that block didn't go thru CCC,
if res != nil && res.FinalBlock != nil { // which means that we are not the sequencer. Do not attempt to commit.
if res != nil && res.Rows == nil && res.OverflowingTx == nil {
if res.FinalBlock != nil {
w.updateSnapshot(res.FinalBlock) w.updateSnapshot(res.FinalBlock)
} }
w.currentPipeline.Release() w.currentPipeline.Release()

View file

@ -1045,3 +1045,47 @@ func TestSealBlockAfterCliquePeriod(t *testing.T) {
t.Fatalf("timeout") t.Fatalf("timeout")
} }
} }
func TestPending(t *testing.T) {
var (
engine consensus.Engine
chainConfig *params.ChainConfig
db = rawdb.NewMemoryDatabase()
)
chainConfig = params.AllCliqueProtocolChanges
chainConfig.Clique = &params.CliqueConfig{Period: 1, Epoch: 30000}
chainConfig.Scroll.FeeVaultAddress = &common.Address{}
engine = clique.New(chainConfig.Clique, db)
w, b := newTestWorker(t, chainConfig, engine, db, 0)
defer w.close()
// This test chain imports the mined blocks.
b.genesis.MustCommit(db)
chain, _ := core.NewBlockChain(db, nil, b.chain.Config(), engine, vm.Config{
Debug: true,
Tracer: vm.NewStructLogger(&vm.LogConfig{EnableMemory: true, EnableReturnData: true})}, nil, nil)
defer chain.Stop()
// Define 3 transactions:
// A --> B (nonce: 0, gas: 20)
tx0, _ := types.SignTx(types.NewTransaction(b.txPool.Nonce(testBankAddress), testUserAddress, big.NewInt(100000000000000000), params.TxGas, big.NewInt(20*params.InitialBaseFee), nil), types.HomesteadSigner{}, testBankKey)
// A --> B (nonce: 1, gas: 5)
tx1, _ := types.SignTx(types.NewTransaction(b.txPool.Nonce(testBankAddress)+1, testUserAddress, big.NewInt(0), params.TxGas, big.NewInt(5*params.InitialBaseFee), nil), types.HomesteadSigner{}, testBankKey)
// B --> A (nonce: 0, gas: 20)
tx2, _ := types.SignTx(types.NewTransaction(b.txPool.Nonce(testUserAddress), testBankAddress, big.NewInt(0), params.TxGas, big.NewInt(20*params.InitialBaseFee), nil), types.HomesteadSigner{}, testUserKey)
// B --> A (nonce: 1, gas: 20)
tx3, _ := types.SignTx(types.NewTransaction(b.txPool.Nonce(testUserAddress)+1, testBankAddress, big.NewInt(0), params.TxGas, big.NewInt(20*params.InitialBaseFee), nil), types.HomesteadSigner{}, testUserKey)
// B --> A (nonce: 2, gas: 20)
tx4, _ := types.SignTx(types.NewTransaction(b.txPool.Nonce(testUserAddress)+2, testBankAddress, big.NewInt(0), params.TxGas, big.NewInt(20*params.InitialBaseFee), nil), types.HomesteadSigner{}, testUserKey)
// A --> B (nonce: 2, gas: 5)
tx5, _ := types.SignTx(types.NewTransaction(b.txPool.Nonce(testBankAddress)+2, testUserAddress, big.NewInt(0), params.TxGas, big.NewInt(5*params.InitialBaseFee), nil), types.HomesteadSigner{}, testBankKey)
b.txPool.AddRemotesSync([]*types.Transaction{tx0, tx1, tx2, tx3, tx4, tx5})
// start building pending block
w.startCh <- struct{}{}
time.Sleep(time.Second)
pending := w.pendingBlock()
assert.NotNil(t, pending)
assert.NotEmpty(t, pending.Transactions())
}

View file

@ -24,7 +24,7 @@ import (
const ( const (
VersionMajor = 5 // Major version component of the current release VersionMajor = 5 // Major version component of the current release
VersionMinor = 3 // Minor version component of the current release VersionMinor = 3 // Minor version component of the current release
VersionPatch = 30 // Patch version component of the current release VersionPatch = 31 // Patch version component of the current release
VersionMeta = "mainnet" // Version metadata to append to the version string VersionMeta = "mainnet" // Version metadata to append to the version string
) )

View file

@ -110,6 +110,14 @@ func (p *Pipeline) Start(deadline time.Time) error {
return nil return nil
} }
// Stop forces pipeline to stop its operation and return whatever progress it has so far
func (p *Pipeline) Stop() {
if p.txnQueue != nil {
close(p.txnQueue)
p.txnQueue = nil
}
}
func (p *Pipeline) TryPushTxns(txs types.OrderedTransactionSet, onFailingTxn func(txnIndex int, tx *types.Transaction, err error) bool) *Result { func (p *Pipeline) TryPushTxns(txs types.OrderedTransactionSet, onFailingTxn func(txnIndex int, tx *types.Transaction, err error) bool) *Result {
for { for {
tx := txs.Peek() tx := txs.Peek()
@ -127,8 +135,7 @@ func (p *Pipeline) TryPushTxns(txs types.OrderedTransactionSet, onFailingTxn fun
txs.Shift() txs.Shift()
default: default:
if errors.Is(err, ErrApplyStageDone) || onFailingTxn(p.txs.Len(), tx, err) { if errors.Is(err, ErrApplyStageDone) || onFailingTxn(p.txs.Len(), tx, err) {
close(p.txnQueue) p.Stop()
p.txnQueue = nil
return nil return nil
} }
@ -167,10 +174,7 @@ func (p *Pipeline) TryPushTxn(tx *types.Transaction) (*Result, error) {
// Release releases all resources related to the pipeline // Release releases all resources related to the pipeline
func (p *Pipeline) Release() { func (p *Pipeline) Release() {
if p.txnQueue != nil { p.Stop()
close(p.txnQueue)
p.txnQueue = nil
}
select { select {
case <-p.applyStageRespCh: case <-p.applyStageRespCh:
@ -303,7 +307,9 @@ type Result struct {
} }
func (p *Pipeline) cccStage(candidates <-chan *BlockCandidate, deadline time.Time) <-chan *Result { func (p *Pipeline) cccStage(candidates <-chan *BlockCandidate, deadline time.Time) <-chan *Result {
p.ccc.Reset() if p.ccc != nil {
p.ccc.Reset()
}
resultCh := make(chan *Result) resultCh := make(chan *Result)
var lastCandidate *BlockCandidate var lastCandidate *BlockCandidate
var lastAccRows *types.RowConsumption var lastAccRows *types.RowConsumption
@ -335,7 +341,7 @@ func (p *Pipeline) cccStage(candidates <-chan *BlockCandidate, deadline time.Tim
cccStart := time.Now() cccStart := time.Now()
var accRows *types.RowConsumption var accRows *types.RowConsumption
var err error var err error
if candidate != nil { if candidate != nil && p.ccc != nil {
accRows, err = p.ccc.ApplyTransaction(candidate.LastTrace) accRows, err = p.ccc.ApplyTransaction(candidate.LastTrace)
lastTxn := candidate.Txs[candidate.Txs.Len()-1] lastTxn := candidate.Txs[candidate.Txs.Len()-1]
cccTimer.UpdateSince(cccStart) cccTimer.UpdateSince(cccStart)
@ -352,6 +358,8 @@ func (p *Pipeline) cccStage(candidates <-chan *BlockCandidate, deadline time.Tim
lastCandidate = candidate lastCandidate = candidate
lastAccRows = accRows lastAccRows = accRows
} else if candidate != nil && p.ccc == nil {
lastCandidate = candidate
} }
// immediately close the block if deadline reached or apply stage is done // immediately close the block if deadline reached or apply stage is done
@ -381,29 +389,31 @@ func (p *Pipeline) traceAndApply(tx *types.Transaction) (*types.Receipt, *types.
return nil, nil, core.ErrGasLimitReached return nil, nil, core.ErrGasLimitReached
} }
// don't commit the state during tracing for circuit capacity checker, otherwise we cannot revert. if p.ccc != nil {
// and even if we don't commit the state, the `refund` value will still be correct, as explained in `CommitTransaction` // don't commit the state during tracing for circuit capacity checker, otherwise we cannot revert.
commitStateAfterApply := false // and even if we don't commit the state, the `refund` value will still be correct, as explained in `CommitTransaction`
snap := p.state.Snapshot() commitStateAfterApply := false
snap := p.state.Snapshot()
// 1. we have to check circuit capacity before `core.ApplyTransaction`, // 1. we have to check circuit capacity before `core.ApplyTransaction`,
// because if the tx can be successfully executed but circuit capacity overflows, it will be inconvenient to revert. // because if the tx can be successfully executed but circuit capacity overflows, it will be inconvenient to revert.
// 2. even if we don't commit to the state during the tracing (which means `clearJournalAndRefund` is not called during the tracing), // 2. even if we don't commit to the state during the tracing (which means `clearJournalAndRefund` is not called during the tracing),
// the `refund` value will still be correct, because: // the `refund` value will still be correct, because:
// 2.1 when starting handling the first tx, `state.refund` is 0 by default, // 2.1 when starting handling the first tx, `state.refund` is 0 by default,
// 2.2 after tracing, the state is either committed in `core.ApplyTransaction`, or reverted, so the `state.refund` can be cleared, // 2.2 after tracing, the state is either committed in `core.ApplyTransaction`, or reverted, so the `state.refund` can be cleared,
// 2.3 when starting handling the following txs, `state.refund` comes as 0 // 2.3 when starting handling the following txs, `state.refund` comes as 0
trace, err = tracing.NewTracerWrapper().CreateTraceEnvAndGetBlockTrace(p.chain.Config(), p.chain, p.chain.Engine(), p.chain.Database(), trace, err = tracing.NewTracerWrapper().CreateTraceEnvAndGetBlockTrace(p.chain.Config(), p.chain, p.chain.Engine(), p.chain.Database(),
p.state, p.parent, types.NewBlockWithHeader(&p.Header).WithBody([]*types.Transaction{tx}, nil), commitStateAfterApply) p.state, p.parent, types.NewBlockWithHeader(&p.Header).WithBody([]*types.Transaction{tx}, nil), commitStateAfterApply)
// `w.current.traceEnv.State` & `w.current.state` share a same pointer to the state, so only need to revert `w.current.state` // `w.current.traceEnv.State` & `w.current.state` share a same pointer to the state, so only need to revert `w.current.state`
// revert to snapshot for calling `core.ApplyMessage` again, (both `traceEnv.GetBlockTrace` & `core.ApplyTransaction` will call `core.ApplyMessage`) // revert to snapshot for calling `core.ApplyMessage` again, (both `traceEnv.GetBlockTrace` & `core.ApplyTransaction` will call `core.ApplyMessage`)
p.state.RevertToSnapshot(snap) p.state.RevertToSnapshot(snap)
if err != nil { if err != nil {
return nil, nil, err return nil, nil, err
}
} }
// create new snapshot for `core.ApplyTransaction` // create new snapshot for `core.ApplyTransaction`
snap = p.state.Snapshot() snap := p.state.Snapshot()
var receipt *types.Receipt var receipt *types.Receipt
receipt, err = core.ApplyTransaction(p.chain.Config(), p.chain, nil /* coinbase will default to chainConfig.Scroll.FeeVaultAddress */, p.gasPool, receipt, err = core.ApplyTransaction(p.chain.Config(), p.chain, nil /* coinbase will default to chainConfig.Scroll.FeeVaultAddress */, p.gasPool,