mirror of
https://github.com/ethereum/go-ethereum.git
synced 2026-07-28 07:36:44 +00:00
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 <jerrycgh@gmail.com>
This commit is contained in:
Pratik Patil
GitHub
parent
96e66e5256
commit
d53c2e7902
10 changed files with 928 additions and 84 deletions
|
|
@ -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()))
|
||||
|
|
|
|||
|
|
@ -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)
|
||||
}
|
||||
|
|
|
|||
|
|
@ -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)
|
||||
}
|
||||
|
|
|
|||
|
|
@ -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
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -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
|
||||
}
|
||||
|
|
|
|||
|
|
@ -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
|
||||
}
|
||||
|
|
|
|||
|
|
@ -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) {
|
||||
|
|
|
|||
|
|
@ -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"`
|
||||
|
||||
/*
|
||||
|
|
|
|||
114
miner/worker.go
114
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 {
|
||||
|
|
|
|||
|
|
@ -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")
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
|
|||
Loading…
Reference in a new issue