Add ability to calculate the longest execution path in a block

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))))
 }

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()
 				}
 			}

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")
 

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)