Merge pull request #1030 from ethereumjs/more-performant-trie-checkpointing-mechanism

Trie: Rework Checkpointing Mechanism

Author: holgerd77

packages/client/lib/sync/fullsync.ts

@@ -21,10 +21,6 @@ export class FullSynchronizer extends Synchronizer {
   private stopSyncing: boolean
   private vmPromise?: Promise<void>
 
-  // Tracking vars for log msg condensation on zero tx blocks
-  private NUM_ZERO_TXS_PER_LOG_MSG = 50
-  public zeroTxsBlockLogMsgCounter: number = 0
-
   constructor(options: SynchronizerOptions) {
     super(options)
     this.blockFetcher = null
@@ -68,9 +64,15 @@ export class FullSynchronizer extends Synchronizer {
       return
     }
     this.runningBlocks = true
+    let blockCounter = 0
+    let txCounter = 0
+    const NUM_BLOCKS_PER_LOG_MSG = 50
     try {
       let oldHead = Buffer.alloc(0)
-      let newHead = (await this.vm.blockchain.getHead()).hash()
+      const newHeadBlock = await this.vm.blockchain.getHead()
+      let newHead = newHeadBlock.hash()
+      const firstHeadBlock = newHeadBlock
+      let lastHeadBlock = newHeadBlock
       while (!newHead.equals(oldHead) && !this.stopSyncing) {
         oldHead = newHead
         this.vmPromise = this.vm.runBlockchain(this.vm.blockchain, 1)
@@ -79,27 +81,29 @@ export class FullSynchronizer extends Synchronizer {
         newHead = headBlock.hash()
         // check if we did run a new block:
         if (!newHead.equals(oldHead)) {
-          const number = headBlock.header.number.toNumber()
-          const hash = short(newHead)
-          const numTxs = headBlock.transactions.length
-          if (numTxs === 0) {
-            this.zeroTxsBlockLogMsgCounter += 1
-          }
-          if (
-            (numTxs > 0 && this.zeroTxsBlockLogMsgCounter > 0) ||
-            (numTxs === 0 && this.zeroTxsBlockLogMsgCounter >= this.NUM_ZERO_TXS_PER_LOG_MSG)
-          ) {
-            this.config.logger.info(`Processed ${this.zeroTxsBlockLogMsgCounter} blocks with 0 txs`)
-            this.zeroTxsBlockLogMsgCounter = 0
-          }
-          if (numTxs > 0) {
-            this.config.logger.info(`Executed block number=${number} hash=${hash} txs=${numTxs}`)
+          blockCounter += 1
+          txCounter += headBlock.transactions.length
+          lastHeadBlock = headBlock
+
+          if (blockCounter >= NUM_BLOCKS_PER_LOG_MSG) {
+            const firstNumber = firstHeadBlock.header.number.toNumber()
+            const firstHash = short(firstHeadBlock.hash())
+            const lastNumber = lastHeadBlock.header.number.toNumber()
+            const lastHash = short(lastHeadBlock.hash())
+            this.config.logger.info(
+              `Executed blocks count=${blockCounter} first=${firstNumber} hash=${firstHash} last=${lastNumber} hash=${lastHash} with txs=${txCounter}`
+            )
+            blockCounter = 0
+            txCounter = 0
           }
         }
       }
+    } catch (error) {
+      this.emit('error', error)
     } finally {
       this.runningBlocks = false
     }
+    return blockCounter
   }
 
   /**

packages/client/test/sync/fullsync.spec.ts

@@ -174,51 +174,28 @@ tape('[FullSynchronizer]', async (t) => {
     t.deepEqual(sync.vm.blockchain.getHead(), oldHead, 'should not modify blockchain on emtpy run')
 
     blockchain.getHead = td.func<any>()
-    td.when(blockchain.getHead()).thenResolve(
-      {
+    const getHeadResponse: any = []
+    for (let i = 2; i <= 100; i++) {
+      getHeadResponse.push({
         hash: () => {
-          return Buffer.from('hash1')
+          return Buffer.from(`hash${i}`)
         },
-        header: { number: new BN(1) },
-        transactions: [],
-      },
-      {
-        hash: () => {
-          return Buffer.from('hash2')
-        },
-        header: { number: new BN(2) },
-        transactions: [],
-      }
-    )
-    await sync.runBlocks()
-    t.equal(
-      sync.zeroTxsBlockLogMsgCounter,
-      1,
-      'should increase zero blocks counter on zero tx blocks'
-    )
+        header: { number: new BN(i) },
+        transactions: [i],
+      })
+    }
 
     td.when(blockchain.getHead()).thenResolve(
       {
         hash: () => {
-          return Buffer.from('hash1')
+          return Buffer.from('hash0')
         },
         header: { number: new BN(1) },
         transactions: [],
       },
-      {
-        hash: () => {
-          return Buffer.from('hash2')
-        },
-        header: { number: new BN(2) },
-        transactions: [1, 2, 3],
-      }
-    )
-    await sync.runBlocks()
-    t.equal(
-      sync.zeroTxsBlockLogMsgCounter,
-      0,
-      'should reset zero blocks counter on non-zero tx blocks'
+      ...getHeadResponse
     )
+    t.equal(await sync.runBlocks(), 49)
 
     t.end()
   })

packages/trie/package.json

@@ -21,7 +21,7 @@
     "tsc": "ethereumjs-config-tsc",
     "test": "npm run test:node && npm run test:browser",
     "test:browser": "npm run build && karma start karma.conf.js",
-    "test:node": "npm run build && tape -r ts-node/register test/*.ts"
+    "test:node": "tape -r ts-node/register test/*.ts"
   },
   "author": {
     "name": "mjbecze",

packages/trie/src/checkpointTrie.ts

@@ -1,47 +1,26 @@
 import { Trie as BaseTrie } from './baseTrie'
-import { ScratchReadStream } from './scratchReadStream'
-import { ScratchDB } from './scratch'
-import { DB } from './db'
-const WriteStream = require('level-ws')
 
 /**
  * Adds checkpointing to the {@link BaseTrie}
  */
 export class CheckpointTrie extends BaseTrie {
-  _mainDB: DB
-  _scratch: ScratchDB | null
-  _checkpoints: Buffer[]
-
   constructor(...args: any) {
     super(...args)
-    // Reference to main DB instance
-    this._mainDB = this.db
-    // DB instance used for checkpoints
-    this._scratch = null
-    // Roots of trie at the moment of checkpoint
-    this._checkpoints = []
   }
 
   /**
    * Is the trie during a checkpoint phase?
    */
   get isCheckpoint() {
-    return this._checkpoints.length > 0
+    return this.db.checkpoints.length > 0
   }
 
   /**
    * Creates a checkpoint that can later be reverted to or committed.
-   * After this is called, no changes to the trie will be permanently saved until `commit` is called.
-   * To override the checkpointing mechanism use `_maindb.put` to write directly write to db.
+   * After this is called, all changes can be reverted until `commit` is called.
    */
   checkpoint() {
-    const wasCheckpoint = this.isCheckpoint
-    this._checkpoints.push(this.root)
-
-    // Entering checkpoint mode is not necessary for nested checkpoints
-    if (!wasCheckpoint && this.isCheckpoint) {
-      this._enterCpMode()
-    }
+    this.db.checkpoint(this.root)
   }
 
   /**
@@ -55,13 +34,7 @@ export class CheckpointTrie extends BaseTrie {
     }
 
     await this.lock.wait()
-
-    this._checkpoints.pop()
-    // eslint-disable-next-line @typescript-eslint/no-unnecessary-condition
-    if (!this.isCheckpoint) {
-      await this._exitCpMode(true)
-    }
-
+    this.db.commit()
     this.lock.signal()
   }
 
@@ -71,15 +44,12 @@ export class CheckpointTrie extends BaseTrie {
    * parent checkpoint as current.
    */
   async revert(): Promise<void> {
-    await this.lock.wait()
-    // eslint-disable-next-line @typescript-eslint/no-unnecessary-condition
-    if (this.isCheckpoint) {
-      this.root = this._checkpoints.pop()!
-      // eslint-disable-next-line @typescript-eslint/no-unnecessary-condition
-      if (!this.isCheckpoint) {
-        await this._exitCpMode(false)
-      }
+    if (!this.isCheckpoint) {
+      throw new Error('trying to revert when not checkpointed')
     }
+
+    await this.lock.wait()
+    this.root = await this.db.revert()
     this.lock.signal()
   }
 
@@ -88,57 +58,11 @@ export class CheckpointTrie extends BaseTrie {
    * @param includeCheckpoints - If true and during a checkpoint, the copy will contain the checkpointing metadata and will use the same scratch as underlying db.
    */
   copy(includeCheckpoints = true): CheckpointTrie {
-    const db = this._mainDB.copy()
+    const db = this.db.copy()
     const trie = new CheckpointTrie(db._leveldb, this.root)
     if (includeCheckpoints && this.isCheckpoint) {
-      trie._checkpoints = this._checkpoints.slice()
-      trie._scratch = this._scratch!.copy()
-      trie.db = trie._scratch
+      trie.db.checkpoints = this.db.checkpoints.slice()
     }
     return trie
   }
-
-  /**
-   * Enter into checkpoint mode.
-   * @private
-   */
-  _enterCpMode() {
-    this._scratch = new ScratchDB(this._mainDB)
-    this.db = this._scratch
-  }
-
-  /**
-   * Exit from checkpoint mode.
-   * @private
-   */
-  async _exitCpMode(commitState: boolean): Promise<void> {
-    return new Promise((resolve) => {
-      const scratch = this._scratch as ScratchDB
-      this._scratch = null
-      this.db = this._mainDB
-
-      if (commitState) {
-        this._createScratchReadStream(scratch)
-          .pipe(WriteStream(this.db._leveldb))
-          .on('close', resolve)
-      } else {
-        process.nextTick(resolve)
-      }
-    })
-  }
-
-  /**
-   * Returns a `ScratchReadStream` based on the state updates
-   * since checkpoint.
-   * @private
-   */
-  _createScratchReadStream(scratchDb?: ScratchDB) {
-    const scratch = scratchDb || this._scratch
-    if (!scratch) {
-      throw new Error('No scratch found to use')
-    }
-    const trie = new BaseTrie(scratch._leveldb, this.root)
-    trie.db = scratch
-    return new ScratchReadStream(trie)
-  }
 }