diff --git a/core/blockstm/dag.go b/core/blockstm/dag.go index 3fe714a50e..d122877fe9 100644 --- a/core/blockstm/dag.go +++ b/core/blockstm/dag.go @@ -2,8 +2,8 @@ package blockstm import ( "fmt" - "sort" "strings" + "time" "github.com/heimdalr/dag" @@ -62,8 +62,6 @@ func BuildDAG(deps TxnInputOutput) (d DAG) { if err != nil { log.Warn("Failed to add edge", "from", txFromId, "to", txToId, "err", err) } - - break // once we add a 'backward' dep we can't execute before that transaction so no need to proceed } } } @@ -71,23 +69,67 @@ func BuildDAG(deps TxnInputOutput) (d DAG) { return } -func (d DAG) Report(out func(string)) { - roots := make([]int, 0) - rootIds := make([]string, 0) - rootIdMap := make(map[int]string, len(d.GetRoots())) +// 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())) - for k, i := range d.GetRoots() { - roots = append(roots, i.(int)) - rootIdMap[i.(int)] = k + for i := 0; i < len(d.GetVertices()); i++ { + prev[i] = -1 } - sort.Ints(roots) + pathWeights := make(map[int]uint64, len(d.GetVertices())) - for _, i := range roots { - rootIds = append(rootIds, rootIdMap[i]) + maxPath := 0 + maxPathWeight := uint64(0) + + idxToId := make(map[int]string, len(d.GetVertices())) + + for k, i := range d.GetVertices() { + idxToId[i.(int)] = k } - fmt.Println(roots) + for i := 0; i < len(idxToId); i++ { + parents, _ := d.GetParents(idxToId[i]) + + if len(parents) > 0 { + for _, p := range parents { + weight := pathWeights[p.(int)] + stats[i].End - stats[i].Start + if weight > pathWeights[i] { + pathWeights[i] = weight + prev[i] = p.(int) + } + } + } else { + pathWeights[i] = stats[i].End - stats[i].Start + } + + if pathWeights[i] > maxPathWeight { + maxPath = i + maxPathWeight = pathWeights[i] + } + } + + path := make([]int, 0) + for i := maxPath; i != -1; i = prev[i] { + path = append(path, i) + } + + // Reverse the path so the transactions are in the ascending order + for i, j := 0, len(path)-1; i < j; i, j = i+1, j-1 { + path[i], path[j] = path[j], path[i] + } + + return path, maxPathWeight +} + +func (d DAG) Report(stats map[int]ExecutionStat, out func(string)) { + longestPath, weight := d.LongestPath(stats) + + serialWeight := uint64(0) + + for i := 0; i < len(d.GetVertices()); i++ { + serialWeight += stats[i].End - stats[i].Start + } makeStrs := func(ints []int) (ret []string) { for _, v := range ints { @@ -97,29 +139,9 @@ func (d DAG) Report(out func(string)) { return } - maxDesc := 0 - maxDeps := 0 - totalDeps := 0 + out("Longest execution path:") + out(fmt.Sprintf("(%v) %v", len(longestPath), strings.Join(makeStrs(longestPath), "->"))) - for k, v := range roots { - ids := []int{v} - desc, _ := d.GetDescendants(rootIds[k]) - - for _, i := range desc { - ids = append(ids, i.(int)) - } - - sort.Ints(ids) - out(fmt.Sprintf("(%v) %v", len(ids), strings.Join(makeStrs(ids), "->"))) - - if len(desc) > maxDesc { - maxDesc = len(desc) - } - } - - numTx := len(d.DAG.GetVertices()) - out(fmt.Sprintf("max chain length: %v of %v (%v%%)", maxDesc+1, numTx, - fmt.Sprintf("%.1f", float64(maxDesc+1)*100.0/float64(numTx)))) - out(fmt.Sprintf("max dep count: %v of %v (%v%%)", maxDeps, totalDeps, - fmt.Sprintf("%.1f", float64(maxDeps)*100.0/float64(totalDeps)))) + out(fmt.Sprintf("Longest path ideal execution time: %v of %v (serial total), %v%%", time.Duration(weight), + time.Duration(serialWeight), fmt.Sprintf("%.1f", float64(weight)*100.0/float64(serialWeight)))) } diff --git a/core/blockstm/executor.go b/core/blockstm/executor.go index 0e2e8af137..18fd81c1c6 100644 --- a/core/blockstm/executor.go +++ b/core/blockstm/executor.go @@ -123,7 +123,7 @@ func (pq *SafePriorityQueue) Len() int { type ParallelExecutionResult struct { TxIO *TxnInputOutput - Stats *[][]uint64 + Stats *map[int]ExecutionStat Deps *DAG } @@ -133,8 +133,9 @@ const numSpeculativeProcs = 8 type ParallelExecutor struct { tasks []ExecTask - // Stores the execution statistics for each task - stats [][]uint64 + // Stores the execution statistics for the last incarnation of each task + stats map[int]ExecutionStat + statsMutex sync.Mutex // Channel for tasks that should be prioritized @@ -202,12 +203,20 @@ type ParallelExecutor struct { workerWg sync.WaitGroup } +type ExecutionStat struct { + TxIdx int + Incarnation int + Start uint64 + End uint64 + Worker int +} + func NewParallelExecutor(tasks []ExecTask, profile bool) *ParallelExecutor { numTasks := len(tasks) pe := &ParallelExecutor{ tasks: tasks, - stats: make([][]uint64, numTasks), + stats: make(map[int]ExecutionStat, numTasks), chTasks: make(chan ExecVersionView, numTasks), chSpeculativeTasks: make(chan struct{}, numTasks), chSettle: make(chan int, numTasks), @@ -272,10 +281,14 @@ func (pe *ParallelExecutor) Prepare() { if pe.profile { end := time.Since(pe.begin) - stat := []uint64{uint64(res.ver.TxnIndex), uint64(res.ver.Incarnation), uint64(start), uint64(end), uint64(procNum)} - pe.statsMutex.Lock() - pe.stats = append(pe.stats, stat) + pe.stats[res.ver.TxnIndex] = ExecutionStat{ + TxIdx: res.ver.TxnIndex, + Incarnation: res.ver.Incarnation, + Start: uint64(start), + End: uint64(end), + Worker: procNum, + } pe.statsMutex.Unlock() } } diff --git a/core/blockstm/executor_test.go b/core/blockstm/executor_test.go index bed60cdcd4..c62e0ae9a4 100644 --- a/core/blockstm/executor_test.go +++ b/core/blockstm/executor_test.go @@ -348,8 +348,14 @@ func checkNoDroppedTx(pe *ParallelExecutor) error { func runParallel(t *testing.T, tasks []ExecTask, validation PropertyCheck) time.Duration { t.Helper() + profile := false + start := time.Now() - _, err := executeParallelWithCheck(tasks, false, validation) + result, err := executeParallelWithCheck(tasks, profile, validation) + + if result.Deps != nil && profile { + result.Deps.Report(*result.Stats, func(str string) { fmt.Println(str) }) + } assert.NoError(t, err, "error occur during parallel execution") diff --git a/core/parallel_state_processor.go b/core/parallel_state_processor.go index 3a530852fa..c936f399cd 100644 --- a/core/parallel_state_processor.go +++ b/core/parallel_state_processor.go @@ -19,6 +19,7 @@ package core import ( "fmt" "math/big" + "time" "github.com/ethereum/go-ethereum/common" "github.com/ethereum/go-ethereum/consensus" @@ -29,6 +30,7 @@ import ( "github.com/ethereum/go-ethereum/core/vm" "github.com/ethereum/go-ethereum/crypto" "github.com/ethereum/go-ethereum/log" + "github.com/ethereum/go-ethereum/metrics" "github.com/ethereum/go-ethereum/params" ) @@ -247,6 +249,8 @@ func (task *ExecutionTask) Settle() { *task.allLogs = append(*task.allLogs, receipt.Logs...) } +var parallelizabilityTimer = metrics.NewRegisteredTimer("block/parallelizability", nil) + // Process processes the state changes according to the Ethereum rules by running // the transaction messages using the statedb and applying any rewards to both // the processor (coinbase) and any included uncles. @@ -314,7 +318,25 @@ func (p *ParallelStateProcessor) Process(block *types.Block, statedb *state.Stat } backupStateDB := statedb.Copy() - _, err := blockstm.ExecuteParallel(tasks, false) + + profile := false + result, err := blockstm.ExecuteParallel(tasks, profile) + + if err == nil && profile { + _, weight := result.Deps.LongestPath(*result.Stats) + + serialWeight := uint64(0) + + for i := 0; i < len(result.Deps.GetVertices()); i++ { + serialWeight += (*result.Stats)[i].End - (*result.Stats)[i].Start + } + + parallelizabilityTimer.Update(time.Duration(serialWeight * 100 / weight)) + + log.Info("Parallelizability", "Average (%)", parallelizabilityTimer.Mean()) + + log.Info("Parallelizability", "Histogram (%)", parallelizabilityTimer.Percentiles([]float64{0.001, 0.01, 0.05, 0.1, 0.25, 0.5, 0.75, 0.9, 0.95, 0.99, 0.999, 0.9999})) + } for _, task := range tasks { task := task.(*ExecutionTask)