Refactor blockstm executor

core/blockstm/executor.go

@@ -3,7 +3,6 @@ package blockstm
 import (
 	"container/heap"
 	"fmt"
-	"sort"
 	"sync"
 	"time"
 
@@ -127,413 +126,402 @@ type ParallelExecutionResult struct {
 }
 
 const numGoProcs = 2
-const numSpeculativeProcs = 16
+const numSpeculativeProcs = 8
 
-// Max number of pre-validation to run per loop
-const preValidateLimit = 5
-
-// Max number of times a transaction (t) can be executed before its dependency is resolved to its previous tx (t-1)
-const maxIncarnation = 2
-
-// nolint: gocognit
-// A stateless executor that executes transactions in parallel
-func ExecuteParallel(tasks []ExecTask, profile bool) (ParallelExecutionResult, error) {
-	if len(tasks) == 0 {
-		return ParallelExecutionResult{MakeTxnInputOutput(len(tasks)), nil, nil}, nil
-	}
+type ParallelExecutor struct {
+	tasks []ExecTask
 
 	// Stores the execution statistics for each task
-	stats := make([][]uint64, 0, len(tasks))
-	statsMutex := sync.Mutex{}
+	stats      [][]uint64
+	statsMutex sync.Mutex
 
 	// Channel for tasks that should be prioritized
-	chTasks := make(chan ExecVersionView, len(tasks))
+	chTasks chan ExecVersionView
 
 	// Channel for speculative tasks
-	chSpeculativeTasks := make(chan struct{}, len(tasks))
-
-	// A priority queue that stores speculative tasks
-	specTaskQueue := NewSafePriorityQueue(len(tasks))
+	chSpeculativeTasks chan struct{}
 
 	// Channel to signal that the result of a transaction could be written to storage
-	chSettle := make(chan int, len(tasks))
+	specTaskQueue *SafePriorityQueue
+
+	// A priority queue that stores speculative tasks
+	chSettle chan int
 
 	// Channel to signal that a transaction has finished executing
-	chResults := make(chan struct{}, len(tasks))
+	chResults chan struct{}
 
 	// A priority queue that stores the transaction index of results, so we can validate the results in order
-	resultQueue := NewSafePriorityQueue(len(tasks))
+	resultQueue *SafePriorityQueue
 
 	// A wait group to wait for all settling tasks to finish
-	var settleWg sync.WaitGroup
+	settleWg sync.WaitGroup
 
 	// An integer that tracks the index of last settled transaction
-	lastSettled := -1
+	lastSettled int
 
 	// For a task that runs only after all of its preceding tasks have finished and passed validation,
 	// its result will be absolutely valid and therefore its validation could be skipped.
 	// This map stores the boolean value indicating whether a task satisfy this condition ( absolutely valid).
-	skipCheck := make(map[int]bool)
-
-	for i := 0; i < len(tasks); i++ {
-		skipCheck[i] = false
-	}
+	skipCheck map[int]bool
 
 	// Execution tasks stores the state of each execution task
-	execTasks := makeStatusManager(len(tasks))
+	execTasks taskStatusManager
 
 	// Validate tasks stores the state of each validation task
-	validateTasks := makeStatusManager(0)
+	validateTasks taskStatusManager
 
 	// Stats for debugging purposes
-	var cntExec, cntSuccess, cntAbort, cntTotalValidations, cntValidationFail int
+	cntExec, cntSuccess, cntAbort, cntTotalValidations, cntValidationFail int
 
-	diagExecSuccess := make([]int, len(tasks))
-	diagExecAbort := make([]int, len(tasks))
+	diagExecSuccess, diagExecAbort []int
 
-	// Initialize MVHashMap
-	mvh := MakeMVHashMap()
+	// Multi-version hash map
+	mvh *MVHashMap
 
 	// Stores the inputs and outputs of the last incardanotion of all transactions
-	lastTxIO := MakeTxnInputOutput(len(tasks))
+	lastTxIO *TxnInputOutput
 
 	// Tracks the incarnation number of each transaction
-	txIncarnations := make([]int, len(tasks))
+	txIncarnations []int
 
 	// A map that stores the estimated dependency of a transaction if it is aborted without any known dependency
-	estimateDeps := make(map[int][]int, len(tasks))
-
-	for i := 0; i < len(tasks); i++ {
-		estimateDeps[i] = make([]int, 0)
-	}
+	estimateDeps map[int][]int
 
 	// A map that records whether a transaction result has been speculatively validated
-	preValidated := make(map[int]bool, len(tasks))
+	preValidated map[int]bool
 
-	begin := time.Now()
+	// Time records when the parallel execution starts
+	begin time.Time
 
-	workerWg := sync.WaitGroup{}
-	workerWg.Add(numSpeculativeProcs + numGoProcs)
+	// Enable profiling
+	profile bool
+
+	// Worker wait group
+	workerWg sync.WaitGroup
+}
+
+func NewParallelExecutor(tasks []ExecTask, profile bool) *ParallelExecutor {
+	numTasks := len(tasks)
+
+	pe := &ParallelExecutor{
+		tasks:              tasks,
+		stats:              make([][]uint64, numTasks),
+		chTasks:            make(chan ExecVersionView, numTasks),
+		chSpeculativeTasks: make(chan struct{}, numTasks),
+		chSettle:           make(chan int, numTasks),
+		chResults:          make(chan struct{}, numTasks),
+		specTaskQueue:      NewSafePriorityQueue(numTasks),
+		resultQueue:        NewSafePriorityQueue(numTasks),
+		lastSettled:        -1,
+		skipCheck:          make(map[int]bool),
+		execTasks:          makeStatusManager(numTasks),
+		validateTasks:      makeStatusManager(0),
+		diagExecSuccess:    make([]int, numTasks),
+		diagExecAbort:      make([]int, numTasks),
+		mvh:                MakeMVHashMap(),
+		lastTxIO:           MakeTxnInputOutput(numTasks),
+		txIncarnations:     make([]int, numTasks),
+		estimateDeps:       make(map[int][]int),
+		preValidated:       make(map[int]bool),
+		begin:              time.Now(),
+		profile:            profile,
+	}
+
+	return pe
+}
+
+func (pe *ParallelExecutor) Prepare() {
+	prevSenderTx := make(map[common.Address]int)
+
+	for i, t := range pe.tasks {
+		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)
+		}
+
+		prevSenderTx[t.Sender()] = i
+	}
+
+	pe.workerWg.Add(numSpeculativeProcs + numGoProcs)
 
 	// Launch workers that execute transactions
 	for i := 0; i < numSpeculativeProcs+numGoProcs; i++ {
 		go func(procNum int) {
-			defer workerWg.Done()
+			defer pe.workerWg.Done()
 
 			doWork := func(task ExecVersionView) {
 				start := time.Duration(0)
-				if profile {
-					start = time.Since(begin)
+				if pe.profile {
+					start = time.Since(pe.begin)
 				}
 
 				res := task.Execute()
 
 				if res.err == nil {
-					mvh.FlushMVWriteSet(res.txAllOut)
+					pe.mvh.FlushMVWriteSet(res.txAllOut)
 				}
 
-				resultQueue.Push(res.ver.TxnIndex, res)
-				chResults <- struct{}{}
+				pe.resultQueue.Push(res.ver.TxnIndex, res)
+				pe.chResults <- struct{}{}
 
-				if profile {
-					end := time.Since(begin)
+				if pe.profile {
+					end := time.Since(pe.begin)
 
 					stat := []uint64{uint64(res.ver.TxnIndex), uint64(res.ver.Incarnation), uint64(start), uint64(end), uint64(procNum)}
 
-					statsMutex.Lock()
-					stats = append(stats, stat)
-					statsMutex.Unlock()
+					pe.statsMutex.Lock()
+					pe.stats = append(pe.stats, stat)
+					pe.statsMutex.Unlock()
 				}
 			}
 
 			if procNum < numSpeculativeProcs {
-				for range chSpeculativeTasks {
-					doWork(specTaskQueue.Pop().(ExecVersionView))
+				for range pe.chSpeculativeTasks {
+					doWork(pe.specTaskQueue.Pop().(ExecVersionView))
 				}
 			} else {
-				for task := range chTasks {
+				for task := range pe.chTasks {
 					doWork(task)
 				}
 			}
 		}(i)
 	}
 
-	// Launch a worker that settles valid transactions
-	settleWg.Add(len(tasks))
+	pe.settleWg.Add(len(pe.tasks))
 
 	go func() {
-		for t := range chSettle {
-			tasks[t].Settle()
-			settleWg.Done()
+		for t := range pe.chSettle {
+			pe.tasks[t].Settle()
+			pe.settleWg.Done()
 		}
 	}()
 
 	// bootstrap first execution
-	tx := execTasks.takeNextPending()
+	tx := pe.execTasks.takeNextPending()
 	if tx != -1 {
-		cntExec++
+		pe.cntExec++
 
-		chTasks <- ExecVersionView{ver: Version{tx, 0}, et: tasks[tx], mvh: mvh, sender: tasks[tx].Sender()}
+		pe.chTasks <- ExecVersionView{ver: Version{tx, 0}, et: pe.tasks[tx], mvh: pe.mvh, sender: pe.tasks[tx].Sender()}
 	}
-
-	// Before starting execution, going through each task to check their explicit dependencies (whether they are coming from the same account)
-	prevSenderTx := make(map[common.Address]int)
-
-	for i, t := range tasks {
-		if tx, ok := prevSenderTx[t.Sender()]; ok {
-			execTasks.addDependencies(tx, i)
-			execTasks.clearPending(i)
-		}
-
-		prevSenderTx[t.Sender()] = i
-	}
-
-	var res ExecResult
-
-	var err error
-
-	// Start main validation loop
-	// nolint:nestif
-	for range chResults {
-		res = resultQueue.Pop().(ExecResult)
-		tx := res.ver.TxnIndex
-
-		if res.err == nil {
-			lastTxIO.recordRead(tx, res.txIn)
-
-			if res.ver.Incarnation == 0 {
-				lastTxIO.recordWrite(tx, res.txOut)
-				lastTxIO.recordAllWrite(tx, res.txAllOut)
-			} else {
-				if res.txAllOut.hasNewWrite(lastTxIO.AllWriteSet(tx)) {
-					validateTasks.pushPendingSet(execTasks.getRevalidationRange(tx + 1))
-				}
-
-				prevWrite := lastTxIO.AllWriteSet(tx)
-
-				// Remove entries that were previously written but are no longer written
-
-				cmpMap := make(map[Key]bool)
-
-				for _, w := range res.txAllOut {
-					cmpMap[w.Path] = true
-				}
-
-				for _, v := range prevWrite {
-					if _, ok := cmpMap[v.Path]; !ok {
-						mvh.Delete(v.Path, tx)
-					}
-				}
-
-				lastTxIO.recordWrite(tx, res.txOut)
-				lastTxIO.recordAllWrite(tx, res.txAllOut)
-			}
-
-			validateTasks.pushPending(tx)
-			execTasks.markComplete(tx)
-			diagExecSuccess[tx]++
-			cntSuccess++
-
-			execTasks.removeDependency(tx)
-		} else if execErr, ok := res.err.(ErrExecAbortError); ok {
-
-			addedDependencies := false
-
-			if execErr.Dependency >= 0 {
-				l := len(estimateDeps[tx])
-				for l > 0 && estimateDeps[tx][l-1] > execErr.Dependency {
-					execTasks.removeDependency(estimateDeps[tx][l-1])
-					estimateDeps[tx] = estimateDeps[tx][:l-1]
-					l--
-				}
-				if txIncarnations[tx] < maxIncarnation {
-					addedDependencies = execTasks.addDependencies(execErr.Dependency, tx)
-				} else {
-					addedDependencies = execTasks.addDependencies(tx-1, tx)
-				}
-			} else {
-				estimate := 0
-
-				if len(estimateDeps[tx]) > 0 {
-					estimate = estimateDeps[tx][len(estimateDeps[tx])-1]
-				}
-				addedDependencies = execTasks.addDependencies(estimate, tx)
-				newEstimate := estimate + (estimate+tx)/2
-				if newEstimate >= tx {
-					newEstimate = tx - 1
-				}
-				estimateDeps[tx] = append(estimateDeps[tx], newEstimate)
-			}
-
-			execTasks.clearInProgress(tx)
-			if !addedDependencies {
-				execTasks.pushPending(tx)
-			}
-			txIncarnations[tx]++
-			diagExecAbort[tx]++
-			cntAbort++
-		} else {
-			err = res.err
-			break
-		}
-
-		// do validations ...
-		maxComplete := execTasks.maxAllComplete()
-
-		var toValidate []int
-
-		for validateTasks.minPending() <= maxComplete && validateTasks.minPending() >= 0 {
-			toValidate = append(toValidate, validateTasks.takeNextPending())
-		}
-
-		for i := 0; i < len(toValidate); i++ {
-			cntTotalValidations++
-
-			tx := toValidate[i]
-
-			if skipCheck[tx] || ValidateVersion(tx, lastTxIO, mvh) {
-				validateTasks.markComplete(tx)
-			} else {
-				cntValidationFail++
-				diagExecAbort[tx]++
-				for _, v := range lastTxIO.AllWriteSet(tx) {
-					mvh.MarkEstimate(v.Path, tx)
-				}
-				// 'create validation tasks for all transactions > tx ...'
-				validateTasks.pushPendingSet(execTasks.getRevalidationRange(tx + 1))
-				validateTasks.clearInProgress(tx) // clear in progress - pending will be added again once new incarnation executes
-
-				addedDependencies := false
-				if txIncarnations[tx] >= maxIncarnation {
-					addedDependencies = execTasks.addDependencies(tx-1, tx)
-				}
-
-				execTasks.clearComplete(tx)
-				if !addedDependencies {
-					execTasks.pushPending(tx)
-				}
-
-				preValidated[tx] = false
-				txIncarnations[tx]++
-			}
-		}
-
-		preValidateCount := 0
-		invalidated := []int{}
-
-		i := sort.SearchInts(validateTasks.pending, maxComplete+1)
-
-		for i < len(validateTasks.pending) && preValidateCount < preValidateLimit {
-			tx := validateTasks.pending[i]
-
-			if !preValidated[tx] {
-				cntTotalValidations++
-
-				if !ValidateVersion(tx, lastTxIO, mvh) {
-					cntValidationFail++
-					diagExecAbort[tx]++
-
-					invalidated = append(invalidated, tx)
-
-					if execTasks.checkComplete(tx) {
-						execTasks.clearComplete(tx)
-					}
-
-					if !execTasks.checkInProgress(tx) {
-						for _, v := range lastTxIO.AllWriteSet(tx) {
-							mvh.MarkEstimate(v.Path, tx)
-						}
-
-						validateTasks.pushPendingSet(execTasks.getRevalidationRange(tx + 1))
-
-						addedDependencies := false
-						if txIncarnations[tx] >= maxIncarnation {
-							addedDependencies = execTasks.addDependencies(tx-1, tx)
-						}
-
-						if !addedDependencies {
-							execTasks.pushPending(tx)
-						}
-					}
-
-					txIncarnations[tx]++
-
-					preValidated[tx] = false
-				} else {
-					preValidated[tx] = true
-				}
-				preValidateCount++
-			}
-
-			i++
-		}
-
-		for _, tx := range invalidated {
-			validateTasks.clearPending(tx)
-		}
-
-		// Settle transactions that have been validated to be correct and that won't be re-executed again
-		maxValidated := validateTasks.maxAllComplete()
-
-		for lastSettled < maxValidated {
-			lastSettled++
-			if execTasks.checkInProgress(lastSettled) || execTasks.checkPending(lastSettled) || execTasks.blockCount[lastSettled] >= 0 {
-				lastSettled--
-				break
-			}
-			chSettle <- lastSettled
-		}
-
-		if validateTasks.countComplete() == len(tasks) && execTasks.countComplete() == len(tasks) {
-			log.Debug("blockstm exec summary", "execs", cntExec, "success", cntSuccess, "aborts", cntAbort, "validations", cntTotalValidations, "failures", cntValidationFail, "#tasks/#execs", fmt.Sprintf("%.2f%%", float64(len(tasks))/float64(cntExec)*100))
-			break
-		}
-
-		// Send the next immediate pending transaction to be executed
-		if execTasks.minPending() != -1 && execTasks.minPending() == maxValidated+1 {
-			nextTx := execTasks.takeNextPending()
-			if nextTx != -1 {
-				cntExec++
-
-				skipCheck[nextTx] = true
-
-				chTasks <- ExecVersionView{ver: Version{nextTx, txIncarnations[nextTx]}, et: tasks[nextTx], mvh: mvh, sender: tasks[nextTx].Sender()}
-			}
-		}
-
-		// Send speculative tasks
-		for execTasks.peekPendingGE(maxValidated+3) != -1 || len(execTasks.inProgress) == 0 {
-			// We skip the next transaction to avoid the case where they all have conflicts and could not be
-			// scheduled for re-execution immediately even when it's their time to run, because they are already in
-			// speculative queue.
-			nextTx := execTasks.takePendingGE(maxValidated + 3)
-
-			if nextTx == -1 {
-				nextTx = execTasks.takeNextPending()
-			}
-
-			if nextTx != -1 {
-				cntExec++
-
-				task := ExecVersionView{ver: Version{nextTx, txIncarnations[nextTx]}, et: tasks[nextTx], mvh: mvh, sender: tasks[nextTx].Sender()}
-
-				specTaskQueue.Push(nextTx, task)
-				chSpeculativeTasks <- struct{}{}
-			}
-		}
-	}
-
-	close(chTasks)
-	close(chSpeculativeTasks)
-	workerWg.Wait()
-	close(chResults)
-	settleWg.Wait()
-	close(chSettle)
-
-	var dag DAG
-	if profile {
-		dag = BuildDAG(*lastTxIO)
-	}
-
-	return ParallelExecutionResult{lastTxIO, &stats, &dag}, err
+}
+
+// nolint: gocognit
+func (pe *ParallelExecutor) Step(res ExecResult) (result ParallelExecutionResult, err error) {
+	tx := res.ver.TxnIndex
+
+	if _, ok := res.err.(ErrExecAbortError); res.err != nil && !ok {
+		err = res.err
+		return
+	}
+
+	// nolint: nestif
+	if execErr, ok := res.err.(ErrExecAbortError); ok {
+		addedDependencies := false
+
+		if execErr.Dependency >= 0 {
+			l := len(pe.estimateDeps[tx])
+			for l > 0 && pe.estimateDeps[tx][l-1] > execErr.Dependency {
+				pe.execTasks.removeDependency(pe.estimateDeps[tx][l-1])
+				pe.estimateDeps[tx] = pe.estimateDeps[tx][:l-1]
+				l--
+			}
+
+			addedDependencies = pe.execTasks.addDependencies(execErr.Dependency, tx)
+		} else {
+			estimate := 0
+
+			if len(pe.estimateDeps[tx]) > 0 {
+				estimate = pe.estimateDeps[tx][len(pe.estimateDeps[tx])-1]
+			}
+			addedDependencies = pe.execTasks.addDependencies(estimate, tx)
+			newEstimate := estimate + (estimate+tx)/2
+			if newEstimate >= tx {
+				newEstimate = tx - 1
+			}
+			pe.estimateDeps[tx] = append(pe.estimateDeps[tx], newEstimate)
+		}
+
+		pe.execTasks.clearInProgress(tx)
+
+		if !addedDependencies {
+			pe.execTasks.pushPending(tx)
+		}
+		pe.txIncarnations[tx]++
+		pe.diagExecAbort[tx]++
+		pe.cntAbort++
+	} else {
+		pe.lastTxIO.recordRead(tx, res.txIn)
+
+		if res.ver.Incarnation == 0 {
+			pe.lastTxIO.recordWrite(tx, res.txOut)
+			pe.lastTxIO.recordAllWrite(tx, res.txAllOut)
+		} else {
+			if res.txAllOut.hasNewWrite(pe.lastTxIO.AllWriteSet(tx)) {
+				pe.validateTasks.pushPendingSet(pe.execTasks.getRevalidationRange(tx + 1))
+			}
+
+			prevWrite := pe.lastTxIO.AllWriteSet(tx)
+
+			// Remove entries that were previously written but are no longer written
+
+			cmpMap := make(map[Key]bool)
+
+			for _, w := range res.txAllOut {
+				cmpMap[w.Path] = true
+			}
+
+			for _, v := range prevWrite {
+				if _, ok := cmpMap[v.Path]; !ok {
+					pe.mvh.Delete(v.Path, tx)
+				}
+			}
+
+			pe.lastTxIO.recordWrite(tx, res.txOut)
+			pe.lastTxIO.recordAllWrite(tx, res.txAllOut)
+		}
+
+		pe.validateTasks.pushPending(tx)
+		pe.execTasks.markComplete(tx)
+		pe.diagExecSuccess[tx]++
+		pe.cntSuccess++
+
+		pe.execTasks.removeDependency(tx)
+	}
+
+	// do validations ...
+	maxComplete := pe.execTasks.maxAllComplete()
+
+	toValidate := make([]int, 0, 2)
+
+	for pe.validateTasks.minPending() <= maxComplete && pe.validateTasks.minPending() >= 0 {
+		toValidate = append(toValidate, pe.validateTasks.takeNextPending())
+	}
+
+	for i := 0; i < len(toValidate); i++ {
+		pe.cntTotalValidations++
+
+		tx := toValidate[i]
+
+		if pe.skipCheck[tx] || ValidateVersion(tx, pe.lastTxIO, pe.mvh) {
+			pe.validateTasks.markComplete(tx)
+		} else {
+			pe.cntValidationFail++
+			pe.diagExecAbort[tx]++
+			for _, v := range pe.lastTxIO.AllWriteSet(tx) {
+				pe.mvh.MarkEstimate(v.Path, tx)
+			}
+			// 'create validation tasks for all transactions > tx ...'
+			pe.validateTasks.pushPendingSet(pe.execTasks.getRevalidationRange(tx + 1))
+			pe.validateTasks.clearInProgress(tx) // clear in progress - pending will be added again once new incarnation executes
+
+			pe.execTasks.clearComplete(tx)
+			pe.execTasks.pushPending(tx)
+
+			pe.preValidated[tx] = false
+			pe.txIncarnations[tx]++
+		}
+	}
+
+	// Settle transactions that have been validated to be correct and that won't be re-executed again
+	maxValidated := pe.validateTasks.maxAllComplete()
+
+	for pe.lastSettled < maxValidated {
+		pe.lastSettled++
+		if pe.execTasks.checkInProgress(pe.lastSettled) || pe.execTasks.checkPending(pe.lastSettled) || pe.execTasks.isBlocked(pe.lastSettled) {
+			pe.lastSettled--
+			break
+		}
+		pe.chSettle <- pe.lastSettled
+	}
+
+	if pe.validateTasks.countComplete() == len(pe.tasks) && pe.execTasks.countComplete() == len(pe.tasks) {
+		log.Debug("blockstm exec summary", "execs", pe.cntExec, "success", pe.cntSuccess, "aborts", pe.cntAbort, "validations", pe.cntTotalValidations, "failures", pe.cntValidationFail, "#tasks/#execs", fmt.Sprintf("%.2f%%", float64(len(pe.tasks))/float64(pe.cntExec)*100))
+
+		close(pe.chTasks)
+		close(pe.chSpeculativeTasks)
+		pe.workerWg.Wait()
+		close(pe.chResults)
+		pe.settleWg.Wait()
+		close(pe.chSettle)
+
+		var dag DAG
+
+		if pe.profile {
+			dag = BuildDAG(*pe.lastTxIO)
+		}
+
+		return ParallelExecutionResult{pe.lastTxIO, &pe.stats, &dag}, err
+	}
+
+	// Send the next immediate pending transaction to be executed
+	if pe.execTasks.minPending() != -1 && pe.execTasks.minPending() == maxValidated+1 {
+		nextTx := pe.execTasks.takeNextPending()
+		if nextTx != -1 {
+			pe.cntExec++
+
+			pe.skipCheck[nextTx] = true
+
+			pe.chTasks <- ExecVersionView{ver: Version{nextTx, pe.txIncarnations[nextTx]}, et: pe.tasks[nextTx], mvh: pe.mvh, sender: pe.tasks[nextTx].Sender()}
+		}
+	}
+
+	// Send speculative tasks
+	for pe.execTasks.minPending() != -1 || len(pe.execTasks.inProgress) == 0 {
+		nextTx := pe.execTasks.takeNextPending()
+
+		if nextTx == -1 {
+			nextTx = pe.execTasks.takeNextPending()
+		}
+
+		if nextTx != -1 {
+			pe.cntExec++
+
+			task := ExecVersionView{ver: Version{nextTx, pe.txIncarnations[nextTx]}, et: pe.tasks[nextTx], mvh: pe.mvh, sender: pe.tasks[nextTx].Sender()}
+
+			pe.specTaskQueue.Push(nextTx, task)
+			pe.chSpeculativeTasks <- struct{}{}
+		}
+	}
+
+	return
+}
+
+type PropertyCheck func(*ParallelExecutor) error
+
+func executeParallelWithCheck(tasks []ExecTask, profile bool, check PropertyCheck) (result ParallelExecutionResult, err error) {
+	if len(tasks) == 0 {
+		return ParallelExecutionResult{MakeTxnInputOutput(len(tasks)), nil, nil}, nil
+	}
+
+	pe := NewParallelExecutor(tasks, profile)
+	pe.Prepare()
+
+	for range pe.chResults {
+		res := pe.resultQueue.Pop().(ExecResult)
+
+		result, err = pe.Step(res)
+
+		if err != nil {
+			return result, err
+		}
+
+		if check != nil {
+			err = check(pe)
+		}
+
+		if result.TxIO != nil || err != nil {
+			return result, err
+		}
+	}
+
+	return
+}
+
+func ExecuteParallel(tasks []ExecTask, profile bool) (result ParallelExecutionResult, err error) {
+	return executeParallelWithCheck(tasks, profile, func(pe *ParallelExecutor) error {
+		return nil
+	})
 }

core/blockstm/executor_test.go

@@ -36,7 +36,7 @@ type testExecTask struct {
 	nonce    int
 }
 
-type PathGenerator func(addr common.Address, j int, total int) Key
+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)
 
@@ -169,11 +169,11 @@ func longTailTimeGenerator(min time.Duration, max time.Duration, i int, j int) f
 	}
 }
 
-var randomPathGenerator = func(sender common.Address, j int, total int) Key {
-	return NewStateKey(sender, common.BigToHash((big.NewInt(int64(total)))))
+var randomPathGenerator = func(sender common.Address, i int, j int, total int) Key {
+	return NewStateKey(common.BigToAddress((big.NewInt(int64(i % 10)))), common.BigToHash((big.NewInt(int64(total)))))
 }
 
-var dexPathGenerator = func(sender common.Address, j int, total int) Key {
+var dexPathGenerator = func(sender common.Address, i int, j int, total int) Key {
 	if j == total-1 || j == 2 {
 		return NewSubpathKey(common.BigToAddress(big.NewInt(int64(0))), 1)
 	} else {
@@ -226,10 +226,10 @@ func taskFactory(numTask int, sender Sender, readsPerT int, writesPerT int, nonI
 		// Generate time and key path for each op except first two that are always read and write nonce
 		for j := 2; j < len(ops); j++ {
 			if ops[j].opType == readType {
-				ops[j].key = pathGenerator(s, j, len(ops))
+				ops[j].key = pathGenerator(s, i, j, len(ops))
 				ops[j].duration = readTime(i, j)
 			} else if ops[j].opType == writeType {
-				ops[j].key = pathGenerator(s, j, len(ops))
+				ops[j].key = pathGenerator(s, i, j, len(ops))
 				ops[j].duration = writeTime(i, j)
 			} else {
 				ops[j].duration = nonIOTime(i, j)
@@ -290,13 +290,64 @@ 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)
 }
 
-func runParallel(t *testing.T, tasks []ExecTask, validation func(TxnInputOutput) bool) time.Duration {
+func composeValidations(checks []PropertyCheck) PropertyCheck {
+	return func(pe *ParallelExecutor) error {
+		for _, check := range checks {
+			err := check(pe)
+			if err != nil {
+				return err
+			}
+		}
+
+		return nil
+	}
+}
+
+func checkNoStatusOverlap(pe *ParallelExecutor) error {
+	seen := make(map[int]string)
+
+	for _, tx := range pe.execTasks.complete {
+		seen[tx] = "complete"
+	}
+
+	for _, tx := range pe.execTasks.inProgress {
+		if v, ok := seen[tx]; ok {
+			return fmt.Errorf("tx %v is in both %v and inProgress", v, tx)
+		}
+
+		seen[tx] = "inProgress"
+	}
+
+	for _, tx := range pe.execTasks.pending {
+		if v, ok := seen[tx]; ok {
+			return fmt.Errorf("tx %v is in both %v complete and pending", v, tx)
+		}
+
+		seen[tx] = "pending"
+	}
+
+	return nil
+}
+
+func checkNoDroppedTx(pe *ParallelExecutor) error {
+	for i := 0; i < len(pe.tasks); i++ {
+		if !pe.execTasks.checkComplete(i) && !pe.execTasks.checkInProgress(i) && !pe.execTasks.checkPending(i) {
+			if !pe.execTasks.isBlocked(i) {
+				return fmt.Errorf("tx %v is not in any status and is not blocked by any other tx", i)
+			}
+		}
+	}
+
+	return nil
+}
+
+func runParallel(t *testing.T, tasks []ExecTask, validation PropertyCheck) time.Duration {
 	t.Helper()
 
 	start := time.Now()
-	results, _ := ExecuteParallel(tasks, false)
+	_, err := executeParallelWithCheck(tasks, false, validation)
 
-	txio := results.TxIO
+	assert.NoError(t, err, "error occur during parallel execution")
 
 	// Need to apply the final write set to storage
 
@@ -317,10 +368,6 @@ func runParallel(t *testing.T, tasks []ExecTask, validation func(TxnInputOutput)
 
 	duration := time.Since(start)
 
-	if validation != nil {
-		assert.True(t, validation(*txio))
-	}
-
 	return duration
 }
 
@@ -333,6 +380,8 @@ func TestLessConflicts(t *testing.T) {
 	numWrites := []int{20, 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) {
 		sender := func(i int) common.Address {
 			randomness := rand.Intn(10) + 10
@@ -340,7 +389,28 @@ func TestLessConflicts(t *testing.T) {
 		}
 		tasks, serialDuration := taskFactory(numTx, sender, numRead, numWrite, numNonIO, randomPathGenerator, readTime, writeTime, nonIOTime)
 
-		return runParallel(t, tasks, nil), serialDuration
+		return runParallel(t, tasks, checks), serialDuration
+	}
+
+	testExecutorComb(t, totalTxs, numReads, numWrites, numNonIO, taskRunner)
+}
+
+func TestZeroTx(t *testing.T) {
+	t.Parallel()
+	rand.Seed(0)
+
+	totalTxs := []int{0}
+	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) {
+		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
 	}
 
 	testExecutorComb(t, totalTxs, numReads, numWrites, numNonIO, taskRunner)
@@ -355,11 +425,13 @@ func TestAlternatingTx(t *testing.T) {
 	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) {
 		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, nil), serialDuration
+		return runParallel(t, tasks, checks), serialDuration
 	}
 
 	testExecutorComb(t, totalTxs, numReads, numWrites, numNonIO, taskRunner)
@@ -374,6 +446,8 @@ func TestMoreConflicts(t *testing.T) {
 	numWrites := []int{20, 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) {
 		sender := func(i int) common.Address {
 			randomness := rand.Intn(10) + 10
@@ -381,7 +455,7 @@ func TestMoreConflicts(t *testing.T) {
 		}
 		tasks, serialDuration := taskFactory(numTx, sender, numRead, numWrite, numNonIO, randomPathGenerator, readTime, writeTime, nonIOTime)
 
-		return runParallel(t, tasks, nil), serialDuration
+		return runParallel(t, tasks, checks), serialDuration
 	}
 
 	testExecutorComb(t, totalTxs, numReads, numWrites, numNonIO, taskRunner)
@@ -396,12 +470,14 @@ func TestRandomTx(t *testing.T) {
 	numWrites := []int{20, 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) {
 		// 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)
 
-		return runParallel(t, tasks, nil), serialDuration
+		return runParallel(t, tasks, checks), serialDuration
 	}
 
 	testExecutorComb(t, totalTxs, numReads, numWrites, numNonIO, taskRunner)
@@ -416,6 +492,8 @@ func TestTxWithLongTailRead(t *testing.T) {
 	numWrites := []int{20, 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) {
 		sender := func(i int) common.Address {
 			randomness := rand.Intn(10) + 10
@@ -426,7 +504,7 @@ func TestTxWithLongTailRead(t *testing.T) {
 
 		tasks, serialDuration := taskFactory(numTx, sender, numRead, numWrite, numNonIO, randomPathGenerator, longTailReadTimer, writeTime, nonIOTime)
 
-		return runParallel(t, tasks, nil), serialDuration
+		return runParallel(t, tasks, checks), serialDuration
 	}
 
 	testExecutorComb(t, totalTxs, numReads, numWrites, numNonIO, taskRunner)
@@ -441,29 +519,27 @@ func TestDexScenario(t *testing.T) {
 	numWrites := []int{20, 100, 200}
 	numNonIO := []int{100, 500}
 
-	validation := func(txio TxnInputOutput) bool {
-		for i, inputs := range txio.inputs {
-			foundDep := false
-
-			for _, input := range inputs {
-				if input.V.TxnIndex == i-1 {
-					foundDep = true
+	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)
+					}
 				}
 			}
-
-			if !foundDep {
-				return false
-			}
 		}
 
-		return true
+		return nil
 	}
 
+	checks := composeValidations([]PropertyCheck{checkNoStatusOverlap, postValidation, checkNoDroppedTx})
+
 	taskRunner := func(numTx int, numRead int, numWrite int, numNonIO int) (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)
 
-		return runParallel(t, tasks, validation), serialDuration
+		return runParallel(t, tasks, checks), serialDuration
 	}
 
 	testExecutorComb(t, totalTxs, numReads, numWrites, numNonIO, taskRunner)

core/blockstm/status.go

@@ -12,10 +12,10 @@ func makeStatusManager(numTasks int) (t taskStatusManager) {
 	}
 
 	t.dependency = make(map[int]map[int]bool, numTasks)
-	t.blockCount = make(map[int]int, numTasks)
+	t.blocker = make(map[int]map[int]bool, numTasks)
 
 	for i := 0; i < numTasks; i++ {
-		t.blockCount[i] = -1
+		t.blocker[i] = make(map[int]bool)
 	}
 
 	return
@@ -26,7 +26,7 @@ type taskStatusManager struct {
 	inProgress []int
 	complete   []int
 	dependency map[int]map[int]bool
-	blockCount map[int]int
+	blocker    map[int]map[int]bool
 }
 
 func insertInList(l []int, v int) []int {
@@ -56,35 +56,6 @@ func (m *taskStatusManager) takeNextPending() int {
 	return x
 }
 
-func (m *taskStatusManager) peekPendingGE(n int) int {
-	x := sort.SearchInts(m.pending, n)
-	if x >= len(m.pending) {
-		return -1
-	}
-
-	return m.pending[x]
-}
-
-// Take a pending task whose transaction index is greater than or equal to the given tx index
-func (m *taskStatusManager) takePendingGE(n int) int {
-	x := sort.SearchInts(m.pending, n)
-	if x >= len(m.pending) {
-		return -1
-	}
-
-	v := m.pending[x]
-
-	if x < len(m.pending)-1 {
-		m.pending = append(m.pending[:x], m.pending[x+1:]...)
-	} else {
-		m.pending = m.pending[:x]
-	}
-
-	m.inProgress = insertInList(m.inProgress, v)
-
-	return v
-}
-
 func hasNoGap(l []int) bool {
 	return l[0]+len(l) == l[len(l)-1]+1
 }
@@ -106,11 +77,7 @@ func (m taskStatusManager) maxAllComplete() int {
 }
 
 func (m *taskStatusManager) pushPending(tx int) {
-	if !m.checkComplete(tx) && !m.checkInProgress(tx) {
-		m.pending = insertInList(m.pending, tx)
-	} else {
-		panic(fmt.Errorf("should not happen - clear complete or inProgress before pushing pending"))
-	}
+	m.pending = insertInList(m.pending, tx)
 }
 
 func removeFromList(l []int, v int, expect bool) []int {
@@ -155,15 +122,12 @@ func (m *taskStatusManager) addDependencies(blocker int, dependent int) bool {
 		return false
 	}
 
-	curBlocker := m.blockCount[dependent]
-
-	if curBlocker > blocker {
-		return true
-	}
+	curblockers := m.blocker[dependent]
 
 	if m.checkComplete(blocker) {
-		// Blocking blocker has already completed
-		m.blockCount[dependent] = -1
+		// Blocker has already completed
+		delete(curblockers, blocker)
+
 		return false
 	}
 
@@ -172,16 +136,21 @@ func (m *taskStatusManager) addDependencies(blocker int, dependent int) bool {
 	}
 
 	m.dependency[blocker][dependent] = true
-	m.blockCount[dependent] = blocker
+	curblockers[blocker] = true
 
 	return true
 }
 
+func (m *taskStatusManager) isBlocked(tx int) bool {
+	return len(m.blocker[tx]) > 0
+}
+
 func (m *taskStatusManager) removeDependency(tx int) {
 	if deps, ok := m.dependency[tx]; ok && len(deps) > 0 {
 		for k := range deps {
-			if m.blockCount[k] == tx {
-				m.blockCount[k] = -1
+			delete(m.blocker[k], tx)
+
+			if len(m.blocker[k]) == 0 {
 				if !m.checkComplete(k) && !m.checkPending(k) && !m.checkInProgress(k) {
 					m.pushPending(k)
 				}
@@ -243,9 +212,7 @@ func (m *taskStatusManager) pushPendingSet(set []int) {
 			m.clearComplete(v)
 		}
 
-		if !m.checkInProgress(v) {
-			m.pushPending(v)
-		}
+		m.pushPending(v)
 	}
 }
 

core/parallel_state_processor.go

@@ -153,6 +153,17 @@ func (task *ExecutionTask) Sender() common.Address {
 }
 
 func (task *ExecutionTask) Settle() {
+	defer func() {
+		if r := recover(); r != nil {
+			// In some rare cases, ApplyMVWriteSet will panic due to an index out of range error when calculating the
+			// address hash in sha3 module. Recover from panic and continue the execution.
+			// After recovery, block receipts or merckle root will be incorrect, but this is fine, because the block
+			// will be rejected and re-synced.
+			log.Info("Recovered from error", "Error:", r)
+			return
+		}
+	}()
+
 	task.finalStateDB.Prepare(task.tx.Hash(), task.index)
 
 	coinbase, _ := task.blockChain.Engine().Author(task.header)

core/state/statedb.go

@@ -275,11 +275,8 @@ func MVRead[T any](s *StateDB, k blockstm.Key, defaultV T, readStorage func(s *S
 		}
 	case blockstm.MVReadResultDependency:
 		{
-			if res.DepIdx() > s.dep {
-				s.dep = res.DepIdx()
-			}
+			s.dep = res.DepIdx()
 
-			// Return immediate to executor when we found a dependency
 			panic("Found dependency")
 		}
 	case blockstm.MVReadResultNone: