diff --git a/core/blockstm/executor.go b/core/blockstm/executor.go index b1c5770866..f0c05a7d94 100644 --- a/core/blockstm/executor.go +++ b/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 + }) } diff --git a/core/blockstm/executor_test.go b/core/blockstm/executor_test.go index 47c875007b..e7f6d685f3 100644 --- a/core/blockstm/executor_test.go +++ b/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) diff --git a/core/blockstm/status.go b/core/blockstm/status.go index f10957330c..7a5c895b7e 100644 --- a/core/blockstm/status.go +++ b/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) } } diff --git a/core/parallel_state_processor.go b/core/parallel_state_processor.go index 1267ede20b..871c38668f 100644 --- a/core/parallel_state_processor.go +++ b/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) diff --git a/core/state/statedb.go b/core/state/statedb.go index a650be1130..d10b5fd564 100644 --- a/core/state/statedb.go +++ b/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: