From f7bd7ca66b332b2c97c69b2c5312fcfba852f9c6 Mon Sep 17 00:00:00 2001
From: Pratik Patil <pratikspatil024@gmail.com>
Date: Fri, 19 Aug 2022 10:32:55 +0530
Subject: [PATCH] Txn prioritizer implemented using mutex map (#487)

* basic txn prioritizer implemented using mutex map

* Re-execute parallel tasks when there is a read in coinbase or burn address

* Re-execute parallel tasks when there is a read in coinbase or burn address

* using *sync.RWMutex{} in mutexMap

Co-authored-by: Jerry <jerrycgh@gmail.com>
---
 core/blockstm/executor.go        | 34 ++++++++++++++++++++++++--------
 core/parallel_state_processor.go |  6 ++++++
 2 files changed, 32 insertions(+), 8 deletions(-)

diff --git a/core/blockstm/executor.go b/core/blockstm/executor.go
index cb5019bbb2..de63bcf7bf 100644
--- a/core/blockstm/executor.go
+++ b/core/blockstm/executor.go
@@ -2,7 +2,9 @@ package blockstm
 
 import (
 	"fmt"
+	"sync"
 
+	"github.com/ethereum/go-ethereum/common"
 	"github.com/ethereum/go-ethereum/log"
 )
 
@@ -19,12 +21,14 @@ type ExecTask interface {
 	MVReadList() []ReadDescriptor
 	MVWriteList() []WriteDescriptor
 	MVFullWriteList() []WriteDescriptor
+	Sender() common.Address
 }
 
 type ExecVersionView struct {
-	ver Version
-	et  ExecTask
-	mvh *MVHashMap
+	ver    Version
+	et     ExecTask
+	mvh    *MVHashMap
+	sender common.Address
 }
 
 func (ev *ExecVersionView) Execute() (er ExecResult) {
@@ -55,6 +59,13 @@ func ExecuteParallel(tasks []ExecTask) (lastTxIO *TxnInputOutput, err error) {
 	chTasks := make(chan ExecVersionView, len(tasks))
 	chResults := make(chan ExecResult, len(tasks))
 	chDone := make(chan bool)
+	mutMap := map[common.Address]*sync.RWMutex{}
+
+	for _, t := range tasks {
+		if _, ok := mutMap[t.Sender()]; !ok {
+			mutMap[t.Sender()] = &sync.RWMutex{}
+		}
+	}
 
 	var cntExec, cntSuccess, cntAbort, cntTotalValidations, cntValidationFail int
 
@@ -65,8 +76,15 @@ func ExecuteParallel(tasks []ExecTask) (lastTxIO *TxnInputOutput, err error) {
 				select {
 				case task := <-t:
 					{
-						res := task.Execute()
-						chResults <- res
+						m := mutMap[task.sender]
+						if !m.TryLock() {
+							// why not this? -> chTasks <- task
+							t <- task
+						} else {
+							res := task.Execute()
+							chResults <- res
+							m.Unlock()
+						}
 					}
 				case <-chDone:
 					break Loop
@@ -88,7 +106,7 @@ func ExecuteParallel(tasks []ExecTask) (lastTxIO *TxnInputOutput, err error) {
 			cntExec++
 
 			log.Debug("blockstm", "bootstrap: proc", x, "executing task", tx)
-			chTasks <- ExecVersionView{ver: Version{tx, 0}, et: tasks[tx], mvh: mvh}
+			chTasks <- ExecVersionView{ver: Version{tx, 0}, et: tasks[tx], mvh: mvh, sender: tasks[tx].Sender()}
 		}
 	}
 
@@ -161,7 +179,7 @@ func ExecuteParallel(tasks []ExecTask) (lastTxIO *TxnInputOutput, err error) {
 		nextTx := execTasks.takeNextPending()
 		if nextTx != -1 {
 			cntExec++
-			chTasks <- ExecVersionView{ver: Version{nextTx, txIncarnations[nextTx]}, et: tasks[nextTx], mvh: mvh}
+			chTasks <- ExecVersionView{ver: Version{nextTx, txIncarnations[nextTx]}, et: tasks[nextTx], mvh: mvh, sender: tasks[nextTx].Sender()}
 		}
 
 		// do validations ...
@@ -221,7 +239,7 @@ func ExecuteParallel(tasks []ExecTask) (lastTxIO *TxnInputOutput, err error) {
 				cntExec++
 
 				log.Debug("blockstm", "# tx queued up", nextTx)
-				chTasks <- ExecVersionView{ver: Version{nextTx, txIncarnations[nextTx]}, et: tasks[nextTx], mvh: mvh}
+				chTasks <- ExecVersionView{ver: Version{nextTx, txIncarnations[nextTx]}, et: tasks[nextTx], mvh: mvh, sender: tasks[nextTx].Sender()}
 			}
 		}
 
diff --git a/core/parallel_state_processor.go b/core/parallel_state_processor.go
index 5fa7a4d56e..4ed3c9161d 100644
--- a/core/parallel_state_processor.go
+++ b/core/parallel_state_processor.go
@@ -67,6 +67,7 @@ type ExecutionTask struct {
 	result                     *ExecutionResult
 	shouldDelayFeeCal          *bool
 	shouldRerunWithoutFeeDelay bool
+	sender                     common.Address
 }
 
 func (task *ExecutionTask) Execute(mvh *blockstm.MVHashMap, incarnation int) (err error) {
@@ -129,6 +130,10 @@ func (task *ExecutionTask) MVFullWriteList() []blockstm.WriteDescriptor {
 	return task.statedb.MVFullWriteList()
 }
 
+func (task *ExecutionTask) Sender() common.Address {
+	return task.sender
+}
+
 // 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.
@@ -182,6 +187,7 @@ func (p *ParallelStateProcessor) Process(block *types.Block, statedb *state.Stat
 			blockContext:      bc,
 			evmConfig:         cfg,
 			shouldDelayFeeCal: &shouldDelayFeeCal,
+			sender:            msg.From(),
 		}
 
 		tasks = append(tasks, task)