txdag: support reset txdag reader when SetHead; (bnb-chain#31)

* txdag: support reset txdag reader when SetHead; * txdag: clean some useless logs; --------- Co-authored-by: galaio <[email protected]>
sunny2022da · Aug 14, 2024 · c68d6f7 · c68d6f7
1 parent 21c8936
commit c68d6f7
Show file tree

Hide file tree

Showing 2 changed files with 52 additions and 10 deletions.
diff --git a/core/blockchain.go b/core/blockchain.go
@@ -886,6 +886,10 @@ func (bc *BlockChain) setHeadBeyondRoot(head uint64, time uint64, root common.Ha
 	bc.miningStateCache.Purge()
 	bc.futureBlocks.Purge()
 
+	if bc.txDAGReader != nil {
+		bc.txDAGReader.Reset(head)
+	}
+
 	// Clear safe block, finalized block if needed
 	if safe := bc.CurrentSafeBlock(); safe != nil && head < safe.Number.Uint64() {
 		log.Warn("SetHead invalidated safe block")
@@ -1992,7 +1996,7 @@ func (bc *BlockChain) insertChain(chain types.Blocks, setHead bool) (int, error)
 			return it.index, err
 		}
 
-		if bc.enableTxDAG {
+		if bc.enableTxDAG && !bc.parallelExecution {
 			// compare input TxDAG when it enable in consensus
 			dag, err := statedb.ResolveTxDAG(len(block.Transactions()), []common.Address{block.Coinbase(), params.OptimismBaseFeeRecipient, params.OptimismL1FeeRecipient})
 			if err == nil {
@@ -2917,9 +2921,10 @@ func writeTxDAGToFile(writeHandle *os.File, item TxDAGOutputItem) error {
 	return err
 }
 
-var TxDAGCacheSize = 10000
+var TxDAGCacheSize = uint64(10000)
 
 type TxDAGFileReader struct {
+	output  string
 	file    *os.File
 	scanner *bufio.Scanner
 	cache   map[uint64]types.TxDAG
@@ -2928,16 +2933,12 @@ type TxDAGFileReader struct {
 }
 
 func NewTxDAGFileReader(output string) (*TxDAGFileReader, error) {
-	file, err := os.Open(output)
+	reader := &TxDAGFileReader{output: output}
+	err := reader.openFile(output)
 	if err != nil {
 		return nil, err
 	}
-	scanner := bufio.NewScanner(file)
-	scanner.Buffer(make([]byte, 5*1024*1024), 5*1024*1024)
-	return &TxDAGFileReader{
-		file:    file,
-		scanner: scanner,
-	}, nil
+	return reader, nil
 }
 
 func (t *TxDAGFileReader) Close() {
@@ -2946,6 +2947,18 @@ func (t *TxDAGFileReader) Close() {
 	t.closeFile()
 }
 
+func (t *TxDAGFileReader) openFile(output string) error {
+	file, err := os.Open(output)
+	if err != nil {
+		return err
+	}
+	scanner := bufio.NewScanner(file)
+	scanner.Buffer(make([]byte, 5*1024*1024), 5*1024*1024)
+	t.file = file
+	t.scanner = scanner
+	return nil
+}
+
 func (t *TxDAGFileReader) closeFile() {
 	if t.scanner != nil {
 		t.scanner = nil
@@ -2991,7 +3004,7 @@ func (t *TxDAGFileReader) TxDAG(expect uint64) types.TxDAG {
 		}
 		t.cache[num] = dag
 		t.latest = num
-		if len(t.cache) >= TxDAGCacheSize {
+		if uint64(len(t.cache)) >= TxDAGCacheSize {
 			break
 		}
 	}
@@ -3005,6 +3018,21 @@ func (t *TxDAGFileReader) TxDAG(expect uint64) types.TxDAG {
 	return t.cache[expect]
 }
 
+func (t *TxDAGFileReader) Reset(number uint64) error {
+	t.lock.Lock()
+	defer t.lock.Unlock()
+	if t.latest-TxDAGCacheSize <= number {
+		return nil
+	}
+	t.closeFile()
+	if err := t.openFile(t.output); err != nil {
+		return err
+	}
+	t.latest = 0
+	t.cache = nil
+	return nil
+}
+
 func readTxDAGItemFromLine(line string) (uint64, types.TxDAG, error) {
 	tokens := strings.Split(line, ",")
 	if len(tokens) != 2 {

diff --git a/core/blockchain_test.go b/core/blockchain_test.go
@@ -4789,6 +4789,20 @@ func TestTxDAGFile_LargeRead(t *testing.T) {
 	for i := uint64(0); i < totalSize; i++ {
 		require.Equal(t, except[i], reader.TxDAG(i), i)
 	}
+
+	// test reset to genesis
+	err = reader.Reset(0)
+	require.NoError(t, err)
+	for i := uint64(0); i < totalSize; i++ {
+		require.Equal(t, except[i], reader.TxDAG(i), i)
+	}
+
+	// test reset skip
+	err = reader.Reset(totalSize - TxDAGCacheSize)
+	require.NoError(t, err)
+	for i := totalSize - TxDAGCacheSize; i < totalSize; i++ {
+		require.Equal(t, except[i], reader.TxDAG(i), i)
+	}
 }
 
 func makeEmptyPlainTxDAG(cnt int, flags ...uint8) *types.PlainTxDAG {