From d53c2e7902c17a7aa8cab1f95c0d77dfe37c1b2f Mon Sep 17 00:00:00 2001 From: Pratik Patil Date: Fri, 23 Dec 2022 09:14:09 +0530 Subject: [PATCH] Block stm miner dependency (#561) * added support for dependencies (executor_tests) * added a function to get dependency map * getting all dependencies in the GetDep function * updated GetDep function * changed the type of AllDeps * added a function to get dependency map * updated GetDep function * generate and get dependencies from block producer * optimized getDep function * bug fix regarding txn index and dep structure * fixed gas bug * optimized getDep function * tests updated/added * few updates regarding dependencies * added channel to calculate the dependencies in a separate go routine * minor changes in the executor which uses latest changes of dependecies + removed metadata flag/argument * Use channel when metadata is available * small bug fix * getting reads and writes only when the transaction succeeds * fixed bug in adding dependencies * updated logic for delay/not delay * bug fix (shouldDelayFeeCal) in parallel state processor * lint fix * using EnableMVHashMap flag * fixed worker and stateProcessor and removed SetMVHashMapNil fumction from stateDB * addredded few comments and fixed bug in executor tests * commented executor tests with metadata (panic: test timed out after 5m0s) * added a check to check len(mvReadMapList) > 0 in miner * addressed comments, minor refactoring in dag.go * moved blockContext out of Execute and adding it in execution task * removed Author() from Settle() and added in execution task * not calling block.Header() again and again, using instead * removed EnableMVHashMap flag, and updated applyTransaction function * addressed comments * added unit test to check dependencies in the block header Co-authored-by: Jerry --- core/blockstm/dag.go | 54 ++++ core/blockstm/executor.go | 105 ++++++-- core/blockstm/executor_test.go | 415 +++++++++++++++++++++++++++++-- core/blockstm/txio.go | 12 + core/parallel_state_processor.go | 140 ++++++++--- core/state/statedb.go | 12 + core/state_processor.go | 43 +++- core/types/block.go | 5 + miner/worker.go | 114 ++++++++- miner/worker_test.go | 112 +++++++++ 10 files changed, 928 insertions(+), 84 deletions(-) diff --git a/core/blockstm/dag.go b/core/blockstm/dag.go index d122877fe9..47bd0685a3 100644 --- a/core/blockstm/dag.go +++ b/core/blockstm/dag.go @@ -14,6 +14,12 @@ type DAG struct { *dag.DAG } +type TxDep struct { + Index int + ReadList []ReadDescriptor + FullWriteList [][]WriteDescriptor +} + func HasReadDep(txFrom TxnOutput, txTo TxnInput) bool { reads := make(map[Key]bool) @@ -69,6 +75,54 @@ func BuildDAG(deps TxnInputOutput) (d DAG) { return } +func depsHelper(dependencies map[int]map[int]bool, txFrom TxnOutput, txTo TxnInput, i int, j int) map[int]map[int]bool { + if HasReadDep(txFrom, txTo) { + dependencies[i][j] = true + + for k := range dependencies[i] { + _, foundDep := dependencies[j][k] + + if foundDep { + delete(dependencies[i], k) + } + } + } + + return dependencies +} + +func UpdateDeps(deps map[int]map[int]bool, t TxDep) map[int]map[int]bool { + txTo := t.ReadList + + deps[t.Index] = map[int]bool{} + + for j := 0; j <= t.Index-1; j++ { + txFrom := t.FullWriteList[j] + + deps = depsHelper(deps, txFrom, txTo, t.Index, j) + } + + return deps +} + +func GetDep(deps TxnInputOutput) map[int]map[int]bool { + newDependencies := map[int]map[int]bool{} + + for i := 1; i < len(deps.inputs); i++ { + txTo := deps.inputs[i] + + newDependencies[i] = map[int]bool{} + + for j := 0; j <= i-1; j++ { + txFrom := deps.allOutputs[j] + + newDependencies = depsHelper(newDependencies, txFrom, txTo, i, j) + } + } + + return newDependencies +} + // Find the longest execution path in the DAG func (d DAG) LongestPath(stats map[int]ExecutionStat) ([]int, uint64) { prev := make(map[int]int, len(d.GetVertices())) diff --git a/core/blockstm/executor.go b/core/blockstm/executor.go index 18fd81c1c6..a086347610 100644 --- a/core/blockstm/executor.go +++ b/core/blockstm/executor.go @@ -26,6 +26,7 @@ type ExecTask interface { Hash() common.Hash Sender() common.Address Settle() + Dependencies() []int } type ExecVersionView struct { @@ -82,6 +83,34 @@ func (h *IntHeap) Pop() any { return x } +type SafeQueue interface { + Push(v int, d interface{}) + Pop() interface{} + Len() int +} + +type SafeFIFOQueue struct { + c chan interface{} +} + +func NewSafeFIFOQueue(capacity int) *SafeFIFOQueue { + return &SafeFIFOQueue{ + c: make(chan interface{}, capacity), + } +} + +func (q *SafeFIFOQueue) Push(v int, d interface{}) { + q.c <- d +} + +func (q *SafeFIFOQueue) Pop() interface{} { + return <-q.c +} + +func (q *SafeFIFOQueue) Len() int { + return len(q.c) +} + // A thread safe priority queue type SafePriorityQueue struct { m sync.Mutex @@ -122,9 +151,10 @@ func (pq *SafePriorityQueue) Len() int { } type ParallelExecutionResult struct { - TxIO *TxnInputOutput - Stats *map[int]ExecutionStat - Deps *DAG + TxIO *TxnInputOutput + Stats *map[int]ExecutionStat + Deps *DAG + AllDeps map[int]map[int]bool } const numGoProcs = 2 @@ -145,7 +175,7 @@ type ParallelExecutor struct { chSpeculativeTasks chan struct{} // Channel to signal that the result of a transaction could be written to storage - specTaskQueue *SafePriorityQueue + specTaskQueue SafeQueue // A priority queue that stores speculative tasks chSettle chan int @@ -154,7 +184,7 @@ type ParallelExecutor struct { chResults chan struct{} // A priority queue that stores the transaction index of results, so we can validate the results in order - resultQueue *SafePriorityQueue + resultQueue SafeQueue // A wait group to wait for all settling tasks to finish settleWg sync.WaitGroup @@ -211,9 +241,21 @@ type ExecutionStat struct { Worker int } -func NewParallelExecutor(tasks []ExecTask, profile bool) *ParallelExecutor { +func NewParallelExecutor(tasks []ExecTask, profile bool, metadata bool) *ParallelExecutor { numTasks := len(tasks) + var resultQueue SafeQueue + + var specTaskQueue SafeQueue + + if metadata { + resultQueue = NewSafeFIFOQueue(numTasks) + specTaskQueue = NewSafeFIFOQueue(numTasks) + } else { + resultQueue = NewSafePriorityQueue(numTasks) + specTaskQueue = NewSafePriorityQueue(numTasks) + } + pe := &ParallelExecutor{ tasks: tasks, stats: make(map[int]ExecutionStat, numTasks), @@ -221,8 +263,8 @@ func NewParallelExecutor(tasks []ExecTask, profile bool) *ParallelExecutor { chSpeculativeTasks: make(chan struct{}, numTasks), chSettle: make(chan int, numTasks), chResults: make(chan struct{}, numTasks), - specTaskQueue: NewSafePriorityQueue(numTasks), - resultQueue: NewSafePriorityQueue(numTasks), + specTaskQueue: specTaskQueue, + resultQueue: resultQueue, lastSettled: -1, skipCheck: make(map[int]bool), execTasks: makeStatusManager(numTasks), @@ -241,19 +283,36 @@ func NewParallelExecutor(tasks []ExecTask, profile bool) *ParallelExecutor { return pe } +// nolint: gocognit func (pe *ParallelExecutor) Prepare() { - prevSenderTx := make(map[common.Address]int) - for i, t := range pe.tasks { + clearPendingFlag := false + pe.skipCheck[i] = false pe.estimateDeps[i] = make([]int, 0) - if tx, ok := prevSenderTx[t.Sender()]; ok { - pe.execTasks.addDependencies(tx, i) - pe.execTasks.clearPending(i) - } + if len(t.Dependencies()) > 0 { + for _, val := range t.Dependencies() { + clearPendingFlag = true - prevSenderTx[t.Sender()] = i + pe.execTasks.addDependencies(val, i) + } + + if clearPendingFlag { + pe.execTasks.clearPending(i) + + clearPendingFlag = false + } + } else { + prevSenderTx := make(map[common.Address]int) + + if tx, ok := prevSenderTx[t.Sender()]; ok { + pe.execTasks.addDependencies(tx, i) + pe.execTasks.clearPending(i) + } + + prevSenderTx[t.Sender()] = i + } } pe.workerWg.Add(numSpeculativeProcs + numGoProcs) @@ -478,13 +537,13 @@ func (pe *ParallelExecutor) Step(res *ExecResult) (result ParallelExecutionResul pe.Close(true) - var dag DAG + var allDeps map[int]map[int]bool if pe.profile { - dag = BuildDAG(*pe.lastTxIO) + allDeps = GetDep(*pe.lastTxIO) } - return ParallelExecutionResult{pe.lastTxIO, &pe.stats, &dag}, err + return ParallelExecutionResult{pe.lastTxIO, &pe.stats, nil, allDeps}, err } // Send the next immediate pending transaction to be executed @@ -518,12 +577,12 @@ func (pe *ParallelExecutor) Step(res *ExecResult) (result ParallelExecutionResul type PropertyCheck func(*ParallelExecutor) error -func executeParallelWithCheck(tasks []ExecTask, profile bool, check PropertyCheck) (result ParallelExecutionResult, err error) { +func executeParallelWithCheck(tasks []ExecTask, profile bool, check PropertyCheck, metadata bool) (result ParallelExecutionResult, err error) { if len(tasks) == 0 { - return ParallelExecutionResult{MakeTxnInputOutput(len(tasks)), nil, nil}, nil + return ParallelExecutionResult{MakeTxnInputOutput(len(tasks)), nil, nil, nil}, nil } - pe := NewParallelExecutor(tasks, profile) + pe := NewParallelExecutor(tasks, profile, metadata) pe.Prepare() for range pe.chResults { @@ -547,6 +606,6 @@ func executeParallelWithCheck(tasks []ExecTask, profile bool, check PropertyChec return } -func ExecuteParallel(tasks []ExecTask, profile bool) (result ParallelExecutionResult, err error) { - return executeParallelWithCheck(tasks, profile, nil) +func ExecuteParallel(tasks []ExecTask, profile bool, metadata bool) (result ParallelExecutionResult, err error) { + return executeParallelWithCheck(tasks, profile, nil, metadata) } diff --git a/core/blockstm/executor_test.go b/core/blockstm/executor_test.go index c62e0ae9a4..686855b36a 100644 --- a/core/blockstm/executor_test.go +++ b/core/blockstm/executor_test.go @@ -19,6 +19,10 @@ type OpType int const readType = 0 const writeType = 1 const otherType = 2 +const greenTick = "✅" +const redCross = "❌" + +const threeRockets = "🚀🚀🚀" type Op struct { key Key @@ -28,30 +32,34 @@ type Op struct { } type testExecTask struct { - txIdx int - ops []Op - readMap map[Key]ReadDescriptor - writeMap map[Key]WriteDescriptor - sender common.Address - nonce int + txIdx int + ops []Op + readMap map[Key]ReadDescriptor + writeMap map[Key]WriteDescriptor + sender common.Address + nonce int + dependencies []int } type PathGenerator func(addr common.Address, i int, j int, total int) Key type TaskRunner func(numTx int, numRead int, numWrite int, numNonIO int) (time.Duration, time.Duration) +type TaskRunnerWithMetadata func(numTx int, numRead int, numWrite int, numNonIO int) (time.Duration, time.Duration, time.Duration) + type Timer func(txIdx int, opIdx int) time.Duration type Sender func(int) common.Address func NewTestExecTask(txIdx int, ops []Op, sender common.Address, nonce int) *testExecTask { return &testExecTask{ - txIdx: txIdx, - ops: ops, - readMap: make(map[Key]ReadDescriptor), - writeMap: make(map[Key]WriteDescriptor), - sender: sender, - nonce: nonce, + txIdx: txIdx, + ops: ops, + readMap: make(map[Key]ReadDescriptor), + writeMap: make(map[Key]WriteDescriptor), + sender: sender, + nonce: nonce, + dependencies: []int{}, } } @@ -157,6 +165,10 @@ func (t *testExecTask) Hash() common.Hash { return common.BytesToHash([]byte(fmt.Sprintf("%d", t.txIdx))) } +func (t *testExecTask) Dependencies() []int { + return t.dependencies +} + func randTimeGenerator(min time.Duration, max time.Duration) func(txIdx int, opIdx int) time.Duration { return func(txIdx int, opIdx int) time.Duration { return time.Duration(rand.Int63n(int64(max-min))) + min @@ -275,10 +287,10 @@ func testExecutorComb(t *testing.T, totalTxs []int, numReads []int, numWrites [] } total++ - performance := "✅" + performance := greenTick if execDuration >= expectedSerialDuration { - performance = "❌" + performance = redCross } fmt.Printf("exec duration %v, serial duration %v, time reduced %v %.2f%%, %v \n", execDuration, expectedSerialDuration, expectedSerialDuration-execDuration, float64(expectedSerialDuration-execDuration)/float64(expectedSerialDuration)*100, performance) @@ -294,6 +306,66 @@ func testExecutorComb(t *testing.T, totalTxs []int, numReads []int, numWrites [] fmt.Printf("Total exec duration: %v, total serial duration: %v, time reduced: %v, time reduced percent: %.2f%%\n", totalExecDuration, totalSerialDuration, totalSerialDuration-totalExecDuration, float64(totalSerialDuration-totalExecDuration)/float64(totalSerialDuration)*100) } +// nolint: gocognit +func testExecutorCombWithMetadata(t *testing.T, totalTxs []int, numReads []int, numWrites []int, numNonIOs []int, taskRunner TaskRunnerWithMetadata) { + t.Helper() + log.Root().SetHandler(log.LvlFilterHandler(log.LvlDebug, log.StreamHandler(os.Stderr, log.TerminalFormat(false)))) + + improved := 0 + improvedMetadata := 0 + rocket := 0 + total := 0 + + totalExecDuration := time.Duration(0) + totalExecDurationMetadata := time.Duration(0) + totalSerialDuration := time.Duration(0) + + for _, numTx := range totalTxs { + for _, numRead := range numReads { + for _, numWrite := range numWrites { + for _, numNonIO := range numNonIOs { + log.Info("Executing block", "numTx", numTx, "numRead", numRead, "numWrite", numWrite, "numNonIO", numNonIO) + execDuration, execDurationMetadata, expectedSerialDuration := taskRunner(numTx, numRead, numWrite, numNonIO) + + if execDuration < expectedSerialDuration { + improved++ + } + total++ + + performance := greenTick + + if execDuration >= expectedSerialDuration { + performance = redCross + + if execDurationMetadata <= expectedSerialDuration { + performance = threeRockets + rocket++ + } + } + + if execDuration >= execDurationMetadata { + improvedMetadata++ + } + + fmt.Printf("WITHOUT METADATA: exec duration %v, serial duration %v, time reduced %v %.2f%%, %v \n", execDuration, expectedSerialDuration, expectedSerialDuration-execDuration, float64(expectedSerialDuration-execDuration)/float64(expectedSerialDuration)*100, performance) + fmt.Printf("WITH METADATA: exec duration %v, exec duration with metadata %v, time reduced %v %.2f%%\n", execDuration, execDurationMetadata, execDuration-execDurationMetadata, float64(execDuration-execDurationMetadata)/float64(execDuration)*100) + + totalExecDuration += execDuration + totalExecDurationMetadata += execDurationMetadata + totalSerialDuration += expectedSerialDuration + } + } + } + } + + fmt.Println("\nImproved: ", improved, "Total: ", total, "success rate: ", float64(improved)/float64(total)*100) + fmt.Println("Metadata Better: ", improvedMetadata, "out of: ", total, "success rate: ", float64(improvedMetadata)/float64(total)*100) + fmt.Println("Rockets (Time of: metadata < serial < without metadata): ", rocket) + fmt.Printf("\nWithout metadata <> serial: Total exec duration: %v, total serial duration : %v, time reduced: %v, time reduced percent: %.2f%%\n", totalExecDuration, totalSerialDuration, totalSerialDuration-totalExecDuration, float64(totalSerialDuration-totalExecDuration)/float64(totalSerialDuration)*100) + fmt.Printf("With metadata <> serial: Total exec duration metadata: %v, total serial duration : %v, time reduced: %v, time reduced percent: %.2f%%\n", totalExecDurationMetadata, totalSerialDuration, totalSerialDuration-totalExecDurationMetadata, float64(totalSerialDuration-totalExecDurationMetadata)/float64(totalSerialDuration)*100) + fmt.Printf("Without metadata <> with metadata: Total exec duration: %v, total exec duration metadata: %v, time reduced: %v, time reduced percent: %.2f%%\n", totalExecDuration, totalExecDurationMetadata, totalExecDuration-totalExecDurationMetadata, float64(totalExecDuration-totalExecDurationMetadata)/float64(totalExecDuration)*100) +} + func composeValidations(checks []PropertyCheck) PropertyCheck { return func(pe *ParallelExecutor) error { for _, check := range checks { @@ -345,13 +417,14 @@ func checkNoDroppedTx(pe *ParallelExecutor) error { return nil } -func runParallel(t *testing.T, tasks []ExecTask, validation PropertyCheck) time.Duration { +// nolint: unparam +func runParallel(t *testing.T, tasks []ExecTask, validation PropertyCheck, metadata bool) time.Duration { t.Helper() profile := false start := time.Now() - result, err := executeParallelWithCheck(tasks, profile, validation) + result, err := executeParallelWithCheck(tasks, false, validation, metadata) if result.Deps != nil && profile { result.Deps.Report(*result.Stats, func(str string) { fmt.Println(str) }) @@ -381,6 +454,17 @@ func runParallel(t *testing.T, tasks []ExecTask, validation PropertyCheck) time. return duration } +func runParallelGetMetadata(t *testing.T, tasks []ExecTask, validation PropertyCheck) map[int]map[int]bool { + t.Helper() + + // fmt.Println("len(tasks)", len(tasks)) + res, err := executeParallelWithCheck(tasks, true, validation, false) + + assert.NoError(t, err, "error occur during parallel execution") + + return res.AllDeps +} + func TestLessConflicts(t *testing.T) { t.Parallel() rand.Seed(0) @@ -399,12 +483,58 @@ func TestLessConflicts(t *testing.T) { } tasks, serialDuration := taskFactory(numTx, sender, numRead, numWrite, numNonIO, randomPathGenerator, readTime, writeTime, nonIOTime) - return runParallel(t, tasks, checks), serialDuration + return runParallel(t, tasks, checks, false), serialDuration } testExecutorComb(t, totalTxs, numReads, numWrites, numNonIO, taskRunner) } +func TestLessConflictsWithMetadata(t *testing.T) { + t.Parallel() + rand.Seed(0) + + totalTxs := []int{300} + numReads := []int{100, 200} + numWrites := []int{100, 200} + numNonIOs := []int{100, 500} + + checks := composeValidations([]PropertyCheck{checkNoStatusOverlap, checkNoDroppedTx}) + + taskRunner := func(numTx int, numRead int, numWrite int, numNonIO int) (time.Duration, time.Duration, time.Duration) { + sender := func(i int) common.Address { + randomness := rand.Intn(10) + 10 + return common.BigToAddress(big.NewInt(int64(i % randomness))) + } + tasks, serialDuration := taskFactory(numTx, sender, numRead, numWrite, numNonIO, randomPathGenerator, readTime, writeTime, nonIOTime) + + parallelDuration := runParallel(t, tasks, checks, false) + + allDeps := runParallelGetMetadata(t, tasks, checks) + + newTasks := make([]ExecTask, 0, len(tasks)) + + for _, t := range tasks { + temp := t.(*testExecTask) + + keys := make([]int, len(allDeps[temp.txIdx])) + + i := 0 + + for k := range allDeps[temp.txIdx] { + keys[i] = k + i++ + } + + temp.dependencies = keys + newTasks = append(newTasks, temp) + } + + return parallelDuration, runParallel(t, newTasks, checks, true), serialDuration + } + + testExecutorCombWithMetadata(t, totalTxs, numReads, numWrites, numNonIOs, taskRunner) +} + func TestZeroTx(t *testing.T) { t.Parallel() rand.Seed(0) @@ -420,7 +550,7 @@ func TestZeroTx(t *testing.T) { sender := func(i int) common.Address { return common.BigToAddress(big.NewInt(int64(1))) } tasks, serialDuration := taskFactory(numTx, sender, numRead, numWrite, numNonIO, randomPathGenerator, readTime, writeTime, nonIOTime) - return runParallel(t, tasks, checks), serialDuration + return runParallel(t, tasks, checks, false), serialDuration } testExecutorComb(t, totalTxs, numReads, numWrites, numNonIO, taskRunner) @@ -441,12 +571,55 @@ func TestAlternatingTx(t *testing.T) { sender := func(i int) common.Address { return common.BigToAddress(big.NewInt(int64(i % 2))) } tasks, serialDuration := taskFactory(numTx, sender, numRead, numWrite, numNonIO, randomPathGenerator, readTime, writeTime, nonIOTime) - return runParallel(t, tasks, checks), serialDuration + return runParallel(t, tasks, checks, false), serialDuration } testExecutorComb(t, totalTxs, numReads, numWrites, numNonIO, taskRunner) } +func TestAlternatingTxWithMetadata(t *testing.T) { + t.Parallel() + rand.Seed(0) + + totalTxs := []int{200} + numReads := []int{20} + numWrites := []int{20} + numNonIO := []int{100} + + checks := composeValidations([]PropertyCheck{checkNoStatusOverlap, checkNoDroppedTx}) + + taskRunner := func(numTx int, numRead int, numWrite int, numNonIO int) (time.Duration, time.Duration, time.Duration) { + sender := func(i int) common.Address { return common.BigToAddress(big.NewInt(int64(i % 2))) } + tasks, serialDuration := taskFactory(numTx, sender, numRead, numWrite, numNonIO, randomPathGenerator, readTime, writeTime, nonIOTime) + + parallelDuration := runParallel(t, tasks, checks, false) + + allDeps := runParallelGetMetadata(t, tasks, checks) + + newTasks := make([]ExecTask, 0, len(tasks)) + + for _, t := range tasks { + temp := t.(*testExecTask) + + keys := make([]int, len(allDeps[temp.txIdx])) + + i := 0 + + for k := range allDeps[temp.txIdx] { + keys[i] = k + i++ + } + + temp.dependencies = keys + newTasks = append(newTasks, temp) + } + + return parallelDuration, runParallel(t, newTasks, checks, true), serialDuration + } + + testExecutorCombWithMetadata(t, totalTxs, numReads, numWrites, numNonIO, taskRunner) +} + func TestMoreConflicts(t *testing.T) { t.Parallel() rand.Seed(0) @@ -465,12 +638,58 @@ func TestMoreConflicts(t *testing.T) { } tasks, serialDuration := taskFactory(numTx, sender, numRead, numWrite, numNonIO, randomPathGenerator, readTime, writeTime, nonIOTime) - return runParallel(t, tasks, checks), serialDuration + return runParallel(t, tasks, checks, false), serialDuration } testExecutorComb(t, totalTxs, numReads, numWrites, numNonIO, taskRunner) } +func TestMoreConflictsWithMetadata(t *testing.T) { + t.Parallel() + rand.Seed(0) + + totalTxs := []int{300} + numReads := []int{100, 200} + numWrites := []int{100, 200} + numNonIO := []int{100, 500} + + checks := composeValidations([]PropertyCheck{checkNoStatusOverlap, checkNoDroppedTx}) + + taskRunner := func(numTx int, numRead int, numWrite int, numNonIO int) (time.Duration, time.Duration, time.Duration) { + sender := func(i int) common.Address { + randomness := rand.Intn(10) + 10 + return common.BigToAddress(big.NewInt(int64(i / randomness))) + } + tasks, serialDuration := taskFactory(numTx, sender, numRead, numWrite, numNonIO, randomPathGenerator, readTime, writeTime, nonIOTime) + + parallelDuration := runParallel(t, tasks, checks, false) + + allDeps := runParallelGetMetadata(t, tasks, checks) + + newTasks := make([]ExecTask, 0, len(tasks)) + + for _, t := range tasks { + temp := t.(*testExecTask) + + keys := make([]int, len(allDeps[temp.txIdx])) + + i := 0 + + for k := range allDeps[temp.txIdx] { + keys[i] = k + i++ + } + + temp.dependencies = keys + newTasks = append(newTasks, temp) + } + + return parallelDuration, runParallel(t, newTasks, checks, true), serialDuration + } + + testExecutorCombWithMetadata(t, totalTxs, numReads, numWrites, numNonIO, taskRunner) +} + func TestRandomTx(t *testing.T) { t.Parallel() rand.Seed(0) @@ -487,12 +706,56 @@ func TestRandomTx(t *testing.T) { sender := func(i int) common.Address { return common.BigToAddress(big.NewInt(int64(rand.Intn(10)))) } tasks, serialDuration := taskFactory(numTx, sender, numRead, numWrite, numNonIO, randomPathGenerator, readTime, writeTime, nonIOTime) - return runParallel(t, tasks, checks), serialDuration + return runParallel(t, tasks, checks, false), serialDuration } testExecutorComb(t, totalTxs, numReads, numWrites, numNonIO, taskRunner) } +func TestRandomTxWithMetadata(t *testing.T) { + t.Parallel() + rand.Seed(0) + + totalTxs := []int{300} + numReads := []int{100, 200} + numWrites := []int{100, 200} + numNonIO := []int{100, 500} + + checks := composeValidations([]PropertyCheck{checkNoStatusOverlap, checkNoDroppedTx}) + + taskRunner := func(numTx int, numRead int, numWrite int, numNonIO int) (time.Duration, time.Duration, time.Duration) { + // Randomly assign this tx to one of 10 senders + sender := func(i int) common.Address { return common.BigToAddress(big.NewInt(int64(rand.Intn(10)))) } + tasks, serialDuration := taskFactory(numTx, sender, numRead, numWrite, numNonIO, randomPathGenerator, readTime, writeTime, nonIOTime) + + parallelDuration := runParallel(t, tasks, checks, false) + + allDeps := runParallelGetMetadata(t, tasks, checks) + + newTasks := make([]ExecTask, 0, len(tasks)) + + for _, t := range tasks { + temp := t.(*testExecTask) + + keys := make([]int, len(allDeps[temp.txIdx])) + + i := 0 + + for k := range allDeps[temp.txIdx] { + keys[i] = k + i++ + } + + temp.dependencies = keys + newTasks = append(newTasks, temp) + } + + return parallelDuration, runParallel(t, newTasks, checks, true), serialDuration + } + + testExecutorCombWithMetadata(t, totalTxs, numReads, numWrites, numNonIO, taskRunner) +} + func TestTxWithLongTailRead(t *testing.T) { t.Parallel() rand.Seed(0) @@ -514,12 +777,61 @@ func TestTxWithLongTailRead(t *testing.T) { tasks, serialDuration := taskFactory(numTx, sender, numRead, numWrite, numNonIO, randomPathGenerator, longTailReadTimer, writeTime, nonIOTime) - return runParallel(t, tasks, checks), serialDuration + return runParallel(t, tasks, checks, false), serialDuration } testExecutorComb(t, totalTxs, numReads, numWrites, numNonIO, taskRunner) } +func TestTxWithLongTailReadWithMetadata(t *testing.T) { + t.Parallel() + rand.Seed(0) + + totalTxs := []int{300} + numReads := []int{100, 200} + numWrites := []int{100, 200} + numNonIO := []int{100, 500} + + checks := composeValidations([]PropertyCheck{checkNoStatusOverlap, checkNoDroppedTx}) + + taskRunner := func(numTx int, numRead int, numWrite int, numNonIO int) (time.Duration, time.Duration, time.Duration) { + sender := func(i int) common.Address { + randomness := rand.Intn(10) + 10 + return common.BigToAddress(big.NewInt(int64(i / randomness))) + } + + longTailReadTimer := longTailTimeGenerator(4*time.Microsecond, 12*time.Microsecond, 7, 10) + + tasks, serialDuration := taskFactory(numTx, sender, numRead, numWrite, numNonIO, randomPathGenerator, longTailReadTimer, writeTime, nonIOTime) + + parallelDuration := runParallel(t, tasks, checks, false) + + allDeps := runParallelGetMetadata(t, tasks, checks) + + newTasks := make([]ExecTask, 0, len(tasks)) + + for _, t := range tasks { + temp := t.(*testExecTask) + + keys := make([]int, len(allDeps[temp.txIdx])) + + i := 0 + + for k := range allDeps[temp.txIdx] { + keys[i] = k + i++ + } + + temp.dependencies = keys + newTasks = append(newTasks, temp) + } + + return parallelDuration, runParallel(t, newTasks, checks, true), serialDuration + } + + testExecutorCombWithMetadata(t, totalTxs, numReads, numWrites, numNonIO, taskRunner) +} + func TestDexScenario(t *testing.T) { t.Parallel() rand.Seed(0) @@ -549,8 +861,65 @@ func TestDexScenario(t *testing.T) { sender := func(i int) common.Address { return common.BigToAddress(big.NewInt(int64(i))) } tasks, serialDuration := taskFactory(numTx, sender, numRead, numWrite, numNonIO, dexPathGenerator, readTime, writeTime, nonIOTime) - return runParallel(t, tasks, checks), serialDuration + return runParallel(t, tasks, checks, false), serialDuration } testExecutorComb(t, totalTxs, numReads, numWrites, numNonIO, taskRunner) } + +func TestDexScenarioWithMetadata(t *testing.T) { + t.Parallel() + rand.Seed(0) + + totalTxs := []int{300} + numReads := []int{100, 200} + numWrites := []int{100, 200} + numNonIO := []int{100, 500} + + postValidation := func(pe *ParallelExecutor) error { + if pe.lastSettled == len(pe.tasks) { + for i, inputs := range pe.lastTxIO.inputs { + for _, input := range inputs { + if input.V.TxnIndex != i-1 { + return fmt.Errorf("Tx %d should depend on tx %d, but it actually depends on %d", i, i-1, input.V.TxnIndex) + } + } + } + } + + return nil + } + + checks := composeValidations([]PropertyCheck{checkNoStatusOverlap, postValidation, checkNoDroppedTx}) + + taskRunner := func(numTx int, numRead int, numWrite int, numNonIO int) (time.Duration, time.Duration, time.Duration) { + sender := func(i int) common.Address { return common.BigToAddress(big.NewInt(int64(i))) } + tasks, serialDuration := taskFactory(numTx, sender, numRead, numWrite, numNonIO, dexPathGenerator, readTime, writeTime, nonIOTime) + + parallelDuration := runParallel(t, tasks, checks, false) + + allDeps := runParallelGetMetadata(t, tasks, checks) + + newTasks := make([]ExecTask, 0, len(tasks)) + + for _, t := range tasks { + temp := t.(*testExecTask) + + keys := make([]int, len(allDeps[temp.txIdx])) + + i := 0 + + for k := range allDeps[temp.txIdx] { + keys[i] = k + i++ + } + + temp.dependencies = keys + newTasks = append(newTasks, temp) + } + + return parallelDuration, runParallel(t, newTasks, checks, true), serialDuration + } + + testExecutorCombWithMetadata(t, totalTxs, numReads, numWrites, numNonIO, taskRunner) +} diff --git a/core/blockstm/txio.go b/core/blockstm/txio.go index a08cf57d22..22541e0a78 100644 --- a/core/blockstm/txio.go +++ b/core/blockstm/txio.go @@ -80,3 +80,15 @@ func (io *TxnInputOutput) recordWrite(txId int, output []WriteDescriptor) { func (io *TxnInputOutput) recordAllWrite(txId int, output []WriteDescriptor) { io.allOutputs[txId] = output } + +func (io *TxnInputOutput) RecordReadAtOnce(inputs [][]ReadDescriptor) { + for ind, val := range inputs { + io.inputs[ind] = val + } +} + +func (io *TxnInputOutput) RecordAllWriteAtOnce(outputs [][]WriteDescriptor) { + for ind, val := range outputs { + io.allOutputs[ind] = val + } +} diff --git a/core/parallel_state_processor.go b/core/parallel_state_processor.go index c936f399cd..23457d5c60 100644 --- a/core/parallel_state_processor.go +++ b/core/parallel_state_processor.go @@ -75,6 +75,15 @@ type ExecutionTask struct { totalUsedGas *uint64 receipts *types.Receipts allLogs *[]*types.Log + + // length of dependencies -> 2 + k (k = a whole number) + // first 2 element in dependencies -> transaction index, and flag representing if delay is allowed or not + // (0 -> delay is not allowed, 1 -> delay is allowed) + // next k elements in dependencies -> transaction indexes on which transaction i is dependent on + dependencies []int + + blockContext vm.BlockContext + coinbase common.Address } func (task *ExecutionTask) Execute(mvh *blockstm.MVHashMap, incarnation int) (err error) { @@ -83,9 +92,7 @@ func (task *ExecutionTask) Execute(mvh *blockstm.MVHashMap, incarnation int) (er task.statedb.SetMVHashmap(mvh) task.statedb.SetIncarnation(incarnation) - blockContext := NewEVMBlockContext(task.header, task.blockChain, nil) - - evm := vm.NewEVM(blockContext, vm.TxContext{}, task.statedb, task.config, task.evmConfig) + evm := vm.NewEVM(task.blockContext, vm.TxContext{}, task.statedb, task.config, task.evmConfig) // Create a new context to be used in the EVM environment. txContext := NewEVMTxContext(task.msg) @@ -112,8 +119,8 @@ func (task *ExecutionTask) Execute(mvh *blockstm.MVHashMap, incarnation int) (er reads := task.statedb.MVReadMap() - if _, ok := reads[blockstm.NewSubpathKey(blockContext.Coinbase, state.BalancePath)]; ok { - log.Info("Coinbase is in MVReadMap", "address", blockContext.Coinbase) + if _, ok := reads[blockstm.NewSubpathKey(task.blockContext.Coinbase, state.BalancePath)]; ok { + log.Info("Coinbase is in MVReadMap", "address", task.blockContext.Coinbase) task.shouldRerunWithoutFeeDelay = true } @@ -157,6 +164,10 @@ func (task *ExecutionTask) Hash() common.Hash { return task.tx.Hash() } +func (task *ExecutionTask) Dependencies() []int { + return task.dependencies +} + func (task *ExecutionTask) Settle() { defer func() { if r := recover(); r != nil { @@ -171,9 +182,7 @@ func (task *ExecutionTask) Settle() { task.finalStateDB.Prepare(task.tx.Hash(), task.index) - coinbase, _ := task.blockChain.Engine().Author(task.header) - - coinbaseBalance := task.finalStateDB.GetBalance(coinbase) + coinbaseBalance := task.finalStateDB.GetBalance(task.coinbase) task.finalStateDB.ApplyMVWriteSet(task.statedb.MVWriteList()) @@ -186,7 +195,7 @@ func (task *ExecutionTask) Settle() { task.finalStateDB.AddBalance(task.result.BurntContractAddress, task.result.FeeBurnt) } - task.finalStateDB.AddBalance(coinbase, task.result.FeeTipped) + task.finalStateDB.AddBalance(task.coinbase, task.result.FeeTipped) output1 := new(big.Int).SetBytes(task.result.SenderInitBalance.Bytes()) output2 := new(big.Int).SetBytes(coinbaseBalance.Bytes()) @@ -196,7 +205,7 @@ func (task *ExecutionTask) Settle() { task.finalStateDB, task.msg.From(), - coinbase, + task.coinbase, task.result.FeeTipped, task.result.SenderInitBalance, @@ -267,6 +276,7 @@ func (p *ParallelStateProcessor) Process(block *types.Block, statedb *state.Stat blockNumber = block.Number() allLogs []*types.Log usedGas = new(uint64) + metadata bool ) // Mutate the block and state according to any hard-fork specs @@ -280,6 +290,14 @@ func (p *ParallelStateProcessor) Process(block *types.Block, statedb *state.Stat coinbase, _ := p.bc.Engine().Author(header) + deps, delayMap := GetDeps(block.Header().TxDependency) + + if block.Header().TxDependency != nil { + metadata = true + } + + blockContext := NewEVMBlockContext(header, p.bc, nil) + // Iterate over and process the individual transactions for i, tx := range block.Transactions() { msg, err := tx.AsMessage(types.MakeSigner(p.config, header.Number), header.BaseFee) @@ -290,37 +308,69 @@ func (p *ParallelStateProcessor) Process(block *types.Block, statedb *state.Stat cleansdb := statedb.Copy() - if msg.From() == coinbase { - shouldDelayFeeCal = false - } + if len(header.TxDependency) > 0 { + shouldDelayFeeCal = delayMap[i] - task := &ExecutionTask{ - msg: msg, - config: p.config, - gasLimit: block.GasLimit(), - blockNumber: blockNumber, - blockHash: blockHash, - tx: tx, - index: i, - cleanStateDB: cleansdb, - finalStateDB: statedb, - blockChain: p.bc, - header: header, - evmConfig: cfg, - shouldDelayFeeCal: &shouldDelayFeeCal, - sender: msg.From(), - totalUsedGas: usedGas, - receipts: &receipts, - allLogs: &allLogs, - } + task := &ExecutionTask{ + msg: msg, + config: p.config, + gasLimit: block.GasLimit(), + blockNumber: blockNumber, + blockHash: blockHash, + tx: tx, + index: i, + cleanStateDB: cleansdb, + finalStateDB: statedb, + blockChain: p.bc, + header: header, + evmConfig: cfg, + shouldDelayFeeCal: &shouldDelayFeeCal, + sender: msg.From(), + totalUsedGas: usedGas, + receipts: &receipts, + allLogs: &allLogs, + dependencies: deps[i], + blockContext: blockContext, + coinbase: coinbase, + } - tasks = append(tasks, task) + tasks = append(tasks, task) + } else { + if msg.From() == coinbase { + shouldDelayFeeCal = false + } + + task := &ExecutionTask{ + msg: msg, + config: p.config, + gasLimit: block.GasLimit(), + blockNumber: blockNumber, + blockHash: blockHash, + tx: tx, + index: i, + cleanStateDB: cleansdb, + finalStateDB: statedb, + blockChain: p.bc, + header: header, + evmConfig: cfg, + shouldDelayFeeCal: &shouldDelayFeeCal, + sender: msg.From(), + totalUsedGas: usedGas, + receipts: &receipts, + allLogs: &allLogs, + dependencies: nil, + blockContext: blockContext, + coinbase: coinbase, + } + + tasks = append(tasks, task) + } } backupStateDB := statedb.Copy() profile := false - result, err := blockstm.ExecuteParallel(tasks, profile) + result, err := blockstm.ExecuteParallel(tasks, false, metadata) if err == nil && profile { _, weight := result.Deps.LongestPath(*result.Stats) @@ -356,7 +406,7 @@ func (p *ParallelStateProcessor) Process(block *types.Block, statedb *state.Stat t.totalUsedGas = usedGas } - _, err = blockstm.ExecuteParallel(tasks, false) + _, err = blockstm.ExecuteParallel(tasks, false, metadata) break } @@ -372,3 +422,23 @@ func (p *ParallelStateProcessor) Process(block *types.Block, statedb *state.Stat return receipts, allLogs, *usedGas, nil } + +func GetDeps(txDependency [][]uint64) (map[int][]int, map[int]bool) { + deps := make(map[int][]int) + delayMap := make(map[int]bool) + + for i := 0; i <= len(txDependency)-1; i++ { + idx := int(txDependency[i][0]) + shouldDelay := txDependency[i][1] == 1 + + delayMap[idx] = shouldDelay + + deps[idx] = []int{} + + for j := 2; j <= len(txDependency[i])-1; j++ { + deps[idx] = append(deps[idx], int(txDependency[i][j])) + } + } + + return deps, delayMap +} diff --git a/core/state/statedb.go b/core/state/statedb.go index 4b74cb6153..b77bd9d763 100644 --- a/core/state/statedb.go +++ b/core/state/statedb.go @@ -179,6 +179,10 @@ func (sdb *StateDB) SetMVHashmap(mvhm *blockstm.MVHashMap) { sdb.dep = -1 } +func (sdb *StateDB) GetMVHashmap() *blockstm.MVHashMap { + return sdb.mvHashmap +} + func (s *StateDB) MVWriteList() []blockstm.WriteDescriptor { writes := make([]blockstm.WriteDescriptor, 0, len(s.writeMap)) @@ -227,6 +231,14 @@ func (s *StateDB) ensureWriteMap() { } } +func (s *StateDB) ClearReadMap() { + s.readMap = make(map[blockstm.Key]blockstm.ReadDescriptor) +} + +func (s *StateDB) ClearWriteMap() { + s.writeMap = make(map[blockstm.Key]blockstm.WriteDescriptor) +} + func (s *StateDB) HadInvalidRead() bool { return s.dep >= 0 } diff --git a/core/state_processor.go b/core/state_processor.go index d4c77ae410..7cfe613080 100644 --- a/core/state_processor.go +++ b/core/state_processor.go @@ -97,12 +97,51 @@ func applyTransaction(msg types.Message, config *params.ChainConfig, bc ChainCon txContext := NewEVMTxContext(msg) evm.Reset(txContext, statedb) - // Apply the transaction to the current state (included in the env). - result, err := ApplyMessage(evm, msg, gp) + var result *ExecutionResult + + var err error + + backupMVHashMap := statedb.GetMVHashmap() + + // pause recording read and write + statedb.SetMVHashmap(nil) + + coinbaseBalance := statedb.GetBalance(evm.Context.Coinbase) + + // resume recording read and write + statedb.SetMVHashmap(backupMVHashMap) + + result, err = ApplyMessageNoFeeBurnOrTip(evm, msg, gp) if err != nil { return nil, err } + // stop recording read and write + statedb.SetMVHashmap(nil) + + if evm.ChainConfig().IsLondon(blockNumber) { + statedb.AddBalance(result.BurntContractAddress, result.FeeBurnt) + } + + statedb.AddBalance(evm.Context.Coinbase, result.FeeTipped) + output1 := new(big.Int).SetBytes(result.SenderInitBalance.Bytes()) + output2 := new(big.Int).SetBytes(coinbaseBalance.Bytes()) + + // Deprecating transfer log and will be removed in future fork. PLEASE DO NOT USE this transfer log going forward. Parameters won't get updated as expected going forward with EIP1559 + // add transfer log + AddFeeTransferLog( + statedb, + + msg.From(), + evm.Context.Coinbase, + + result.FeeTipped, + result.SenderInitBalance, + coinbaseBalance, + output1.Sub(output1, result.FeeTipped), + output2.Add(output2, result.FeeTipped), + ) + // Update the state with pending changes. var root []byte if config.IsByzantium(blockNumber) { diff --git a/core/types/block.go b/core/types/block.go index d4451497af..0d91e543da 100644 --- a/core/types/block.go +++ b/core/types/block.go @@ -87,6 +87,11 @@ type Header struct { // BaseFee was added by EIP-1559 and is ignored in legacy headers. BaseFee *big.Int `json:"baseFeePerGas" rlp:"optional"` + // length of TxDependency -> n (n = number of transactions in the block) + // length of TxDependency[i] -> 2 + k (k = a whole number) + // first 2 element in TxDependency[i] -> transaction index, and flag representing if delay is allowed or not + // (0 -> delay is not allowed, 1 -> delay is allowed) + // next k elements in TxDependency[i] -> transaction indexes on which transaction i is dependent on TxDependency [][]uint64 `json:"txDependency" rlp:"optional"` /* diff --git a/miner/worker.go b/miner/worker.go index 797e7ea980..2d478c2761 100644 --- a/miner/worker.go +++ b/miner/worker.go @@ -35,6 +35,7 @@ import ( "github.com/ethereum/go-ethereum/consensus" "github.com/ethereum/go-ethereum/consensus/misc" "github.com/ethereum/go-ethereum/core" + "github.com/ethereum/go-ethereum/core/blockstm" "github.com/ethereum/go-ethereum/core/state" "github.com/ethereum/go-ethereum/core/types" "github.com/ethereum/go-ethereum/event" @@ -81,6 +82,10 @@ const ( // staleThreshold is the maximum depth of the acceptable stale block. staleThreshold = 7 + + // TODO: will be handled (and made mandatory) in a hardfork event + // when true, will get the transaction dependencies for parallel execution, also set in `state_processor.go` + EnableMVHashMap = true ) // environment is the worker's current environment and holds all @@ -918,7 +923,50 @@ func (w *worker) commitTransactions(env *environment, txs *types.TransactionsByP } var coalescedLogs []*types.Log + var depsMVReadList [][]blockstm.ReadDescriptor + + var depsMVFullWriteList [][]blockstm.WriteDescriptor + + var mvReadMapList []map[blockstm.Key]blockstm.ReadDescriptor + + var deps map[int]map[int]bool + + chDeps := make(chan blockstm.TxDep) + + var count int + + var depsWg sync.WaitGroup + + // create and add empty mvHashMap in statedb + if EnableMVHashMap { + depsMVReadList = [][]blockstm.ReadDescriptor{} + + depsMVFullWriteList = [][]blockstm.WriteDescriptor{} + + mvReadMapList = []map[blockstm.Key]blockstm.ReadDescriptor{} + + deps = map[int]map[int]bool{} + + chDeps = make(chan blockstm.TxDep) + + count = 0 + + depsWg.Add(1) + + go func(chDeps chan blockstm.TxDep) { + for t := range chDeps { + deps = blockstm.UpdateDeps(deps, t) + } + + depsWg.Done() + }(chDeps) + } + for { + if EnableMVHashMap { + env.state.AddEmptyMVHashMap() + } + // In the following three cases, we will interrupt the execution of the transaction. // (1) new head block event arrival, the interrupt signal is 1 // (2) worker start or restart, the interrupt signal is 1 @@ -986,7 +1034,23 @@ func (w *worker) commitTransactions(env *environment, txs *types.TransactionsByP // Everything ok, collect the logs and shift in the next transaction from the same account coalescedLogs = append(coalescedLogs, logs...) env.tcount++ - txs.Shift() + + if EnableMVHashMap { + depsMVReadList = append(depsMVReadList, env.state.MVReadList()) + depsMVFullWriteList = append(depsMVFullWriteList, env.state.MVFullWriteList()) + mvReadMapList = append(mvReadMapList, env.state.MVReadMap()) + + temp := blockstm.TxDep{ + Index: env.tcount - 1, + ReadList: depsMVReadList[count], + FullWriteList: depsMVFullWriteList, + } + + chDeps <- temp + count++ + + txs.Shift() + } case errors.Is(err, core.ErrTxTypeNotSupported): // Pop the unsupported transaction without shifting in the next from the account @@ -999,6 +1063,54 @@ func (w *worker) commitTransactions(env *environment, txs *types.TransactionsByP log.Debug("Transaction failed, account skipped", "hash", tx.Hash(), "err", err) txs.Shift() } + + if EnableMVHashMap { + env.state.ClearReadMap() + env.state.ClearWriteMap() + } + } + + // nolint:nestif + if EnableMVHashMap { + close(chDeps) + depsWg.Wait() + + if len(mvReadMapList) > 0 { + tempDeps := make([][]uint64, len(mvReadMapList)) + + // adding for txIdx = 0 + tempDeps[0] = []uint64{uint64(0)} + tempDeps[0] = append(tempDeps[0], 1) + + for j := range deps[0] { + tempDeps[0] = append(tempDeps[0], uint64(j)) + } + + for i := 1; i <= len(mvReadMapList)-1; i++ { + tempDeps[i] = []uint64{uint64(i)} + + reads := mvReadMapList[i-1] + + _, ok1 := reads[blockstm.NewSubpathKey(env.coinbase, state.BalancePath)] + _, ok2 := reads[blockstm.NewSubpathKey(common.HexToAddress(w.chainConfig.Bor.CalculateBurntContract(env.header.Number.Uint64())), state.BalancePath)] + + if ok1 || ok2 { + // 0 -> delay is not allowed + tempDeps[i] = append(tempDeps[i], 0) + } else { + // 1 -> delay is allowed + tempDeps[i] = append(tempDeps[i], 1) + } + + for j := range deps[i] { + tempDeps[i] = append(tempDeps[i], uint64(j)) + } + } + + env.header.TxDependency = tempDeps + } else { + env.header.TxDependency = nil + } } if !w.isRunning() && len(coalescedLogs) > 0 { diff --git a/miner/worker_test.go b/miner/worker_test.go index 011895c854..da40218a7d 100644 --- a/miner/worker_test.go +++ b/miner/worker_test.go @@ -714,3 +714,115 @@ func BenchmarkBorMining(b *testing.B) { } } } + +// uses core.NewParallelBlockChain to use the dependencies present in the block header +func BenchmarkBorMiningBlockSTMMetadata(b *testing.B) { + chainConfig := params.BorUnittestChainConfig + + ctrl := gomock.NewController(b) + defer ctrl.Finish() + + ethAPIMock := api.NewMockCaller(ctrl) + ethAPIMock.EXPECT().Call(gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any()).AnyTimes() + + spanner := bor.NewMockSpanner(ctrl) + spanner.EXPECT().GetCurrentValidators(gomock.Any(), gomock.Any(), gomock.Any()).Return([]*valset.Validator{ + { + ID: 0, + Address: TestBankAddress, + VotingPower: 100, + ProposerPriority: 0, + }, + }, nil).AnyTimes() + + heimdallClientMock := mocks.NewMockIHeimdallClient(ctrl) + heimdallClientMock.EXPECT().Close().Times(1) + + contractMock := bor.NewMockGenesisContract(ctrl) + + db, _, _ := NewDBForFakes(b) + + engine := NewFakeBor(b, db, chainConfig, ethAPIMock, spanner, heimdallClientMock, contractMock) + defer engine.Close() + + chainConfig.LondonBlock = big.NewInt(0) + + w, back, _ := NewTestWorker(b, chainConfig, engine, db, 0) + defer w.close() + + // This test chain imports the mined blocks. + db2 := rawdb.NewMemoryDatabase() + back.Genesis.MustCommit(db2) + + chain, _ := core.NewParallelBlockChain(db2, nil, back.chain.Config(), engine, vm.Config{}, nil, nil, nil) + defer chain.Stop() + + // Ignore empty commit here for less noise. + w.skipSealHook = func(task *task) bool { + return len(task.receipts) == 0 + } + + // fulfill tx pool + const ( + totalGas = testGas + params.TxGas + totalBlocks = 10 + ) + + var err error + + txInBlock := int(back.Genesis.GasLimit/totalGas) + 1 + + // a bit risky + for i := 0; i < 2*totalBlocks*txInBlock; i++ { + err = back.txPool.AddLocal(back.newRandomTx(true)) + if err != nil { + b.Fatal("while adding a local transaction", err) + } + + err = back.txPool.AddLocal(back.newRandomTx(false)) + if err != nil { + b.Fatal("while adding a remote transaction", err) + } + } + + // Wait for mined blocks. + sub := w.mux.Subscribe(core.NewMinedBlockEvent{}) + defer sub.Unsubscribe() + + b.ResetTimer() + + prev := uint64(time.Now().Unix()) + + // Start mining! + w.start() + + blockPeriod, ok := back.Genesis.Config.Bor.Period["0"] + if !ok { + blockPeriod = 1 + } + + for i := 0; i < totalBlocks; i++ { + select { + case ev := <-sub.Chan(): + block := ev.Data.(core.NewMinedBlockEvent).Block + + if _, err := chain.InsertChain([]*types.Block{block}); err != nil { + b.Fatalf("failed to insert new mined block %d: %v", block.NumberU64(), err) + } + + // check for dependencies + deps := block.TxDependency() + for i := 1; i < block.Transactions().Len(); i++ { + if deps[i][0] != uint64(i) || deps[i][1] != uint64(0) || deps[i][2] != uint64(i-1) || len(deps[i]) != 3 { + b.Fatalf("wrong dependency") + } + } + + b.Log("block", block.NumberU64(), "time", block.Time()-prev, "txs", block.Transactions().Len(), "gasUsed", block.GasUsed(), "gasLimit", block.GasLimit()) + + prev = block.Time() + case <-time.After(time.Duration(blockPeriod) * time.Second): + b.Fatalf("timeout") + } + } +}