Merge pull request #1028 from ethereumjs/client-add-vm-execution-rebase

Client: add VM execution

Author: holgerd77

packages/client/bin/cli.ts

@@ -91,17 +91,21 @@ let logger: Logger | null = null
  * @param config
  */
 async function runNode(config: Config) {
-  const syncDataDir = config.getSyncDataDirectory()
-  fs.ensureDirSync(syncDataDir)
-  config.logger.info(`Sync data directory: ${syncDataDir}`)
+  const chainDataDir = config.getChainDataDirectory()
+  fs.ensureDirSync(chainDataDir)
+  const stateDataDir = config.getStateDataDirectory()
+  fs.ensureDirSync(stateDataDir)
+
+  config.logger.info(`Data directory: ${config.datadir}`)
 
   config.logger.info('Initializing Ethereumjs client...')
   if (config.lightserv) {
     config.logger.info(`Serving light peer requests`)
   }
   const client = new EthereumClient({
     config,
-    db: level(syncDataDir),
+    chainDB: level(chainDataDir),
+    stateDB: level(stateDataDir),
   })
   client.on('error', (err: any) => config.logger.error(err))
   client.on('listening', (details: any) => {
@@ -145,7 +149,7 @@ async function run() {
     common,
     syncmode: args.syncmode,
     lightserv: args.lightserv,
-    datadir: `${os.homedir()}/Library/Ethereum`,
+    datadir: `${os.homedir()}/Library/Ethereum/ethereumjs`,
     transports: args.transports,
     rpc: args.rpc,
     rpcport: args.rpcport,

packages/client/lib/blockchain/chain.ts

@@ -17,7 +17,7 @@ export interface ChainOptions {
   /**
    * Database to store blocks and metadata. Should be an abstract-leveldown compliant store.
    */
-  db?: LevelUp
+  chainDB?: LevelUp
 
   /**
    * Specify a blockchain which implements the Chain interface
@@ -79,7 +79,7 @@ export interface GenesisBlockParams {
 export class Chain extends EventEmitter {
   public config: Config
 
-  public db: LevelUp
+  public chainDB: LevelUp
   public blockchain: Blockchain
   public opened: boolean
 
@@ -107,13 +107,13 @@ export class Chain extends EventEmitter {
     this.blockchain =
       options.blockchain ??
       new Blockchain({
-        db: options.db,
+        db: options.chainDB,
         common: this.config.common,
         validateBlocks: false,
         validateConsensus: false,
       })
 
-    this.db = this.blockchain.db
+    this.chainDB = this.blockchain.db
     this.opened = false
   }
 

packages/client/lib/client.ts

@@ -13,7 +13,14 @@ export interface EthereumClientOptions {
    *
    * Default: Database created by the Blockchain class
    */
-  db?: LevelUp
+  chainDB?: LevelUp
+
+  /**
+   * Database to store the state. Should be an abstract-leveldown compliant store.
+   *
+   * Default: Database created by the Trie class
+   */
+  stateDB?: LevelUp
 
   /* List of bootnodes to use for discovery */
   bootnodes?: BootnodeLike[]
@@ -51,11 +58,12 @@ export default class EthereumClient extends events.EventEmitter {
       this.config.syncmode === 'full'
         ? new FullEthereumService({
             config: this.config,
-            db: options.db,
+            chainDB: options.chainDB,
+            stateDB: options.stateDB,
           })
         : new LightEthereumService({
             config: this.config,
-            db: options.db,
+            chainDB: options.chainDB,
           }),
     ]
     this.opened = false

packages/client/lib/config.ts

@@ -184,15 +184,25 @@ export class Config {
   }
 
   /**
-   * Returns the directory for storing the client sync data
+   * Returns the directory for storing the client chain data
    * based on syncmode and selected chain (subdirectory of 'datadir')
    */
-  getSyncDataDirectory(): string {
-    const syncDirName = this.syncmode === 'light' ? 'lightchaindata' : 'chaindata'
-    const chain = this.common.chainName()
-    const networkDirName = chain === 'mainnet' ? '' : `${chain}/`
+  getChainDataDirectory(): string {
+    const networkDirName = this.common.chainName()
+    const chainDataDirName = this.syncmode === 'light' ? 'lightchain' : 'chain'
 
-    const dataDir = `${this.datadir}/${networkDirName}ethereumjs/${syncDirName}`
+    const dataDir = `${this.datadir}/${networkDirName}/${chainDataDirName}`
+    return dataDir
+  }
+
+  /**
+   * Returns the directory for storing the client state data
+   * based selected chain (subdirectory of 'datadir')
+   */
+  getStateDataDirectory(): string {
+    const networkDirName = this.common.chainName()
+
+    const dataDir = `${this.datadir}/${networkDirName}/state`
     return dataDir
   }
 }

packages/client/lib/service/ethereumservice.ts

@@ -9,12 +9,15 @@ export interface EthereumServiceOptions extends ServiceOptions {
   chain?: Chain
 
   /* Blockchain database */
-  db?: LevelUp
+  chainDB?: LevelUp
+
+  /* State database */
+  stateDB?: LevelUp
 
   /* Sync retry interval in ms (default: 8000) */
   interval?: number
 
-  /* Protocol timeout in ms (default: 2000) */
+  /* Protocol timeout in ms (default: 4000) */
   timeout?: number
 }
 
@@ -39,7 +42,7 @@ export class EthereumService extends Service {
     this.flow = new FlowControl()
     this.chain = options.chain ?? new Chain(options)
     this.interval = options.interval ?? 8000
-    this.timeout = options.timeout ?? 2000
+    this.timeout = options.timeout ?? 4000
   }
 
   /**

packages/client/lib/service/fullethereumservice.ts

@@ -33,6 +33,7 @@ export class FullEthereumService extends EthereumService {
       config: this.config,
       pool: this.pool,
       chain: this.chain,
+      stateDB: options.stateDB,
       interval: this.interval,
     })
   }

packages/client/lib/sync/fetcher/blockfetcherbase.ts

@@ -35,7 +35,7 @@ export abstract class BlockFetcherBase<JobResult, StorageItem> extends Fetcher<
     super(options)
 
     this.chain = options.chain
-    this.maxPerRequest = options.maxPerRequest ?? 128
+    this.maxPerRequest = options.maxPerRequest ?? 50
     this.first = options.first
     this.count = options.count
   }

packages/client/lib/sync/fetcher/fetcher.ts

@@ -24,7 +24,7 @@ export interface FetcherOptions {
   /* Max write queue size (default: 16) */
   maxQueue?: number
 
-  /* Max items per request (default: 128) */
+  /* Max items per request (default: 50) */
   maxPerRequest?: number
 
   /* Retry interval in ms (default: 1000) */
@@ -74,7 +74,7 @@ export abstract class Fetcher<JobTask, JobResult, StorageItem> extends Readable
     this.interval = options.interval ?? 1000
     this.banTime = options.banTime ?? 60000
     this.maxQueue = options.maxQueue ?? 16
-    this.maxPerRequest = options.maxPerRequest ?? 128
+    this.maxPerRequest = options.maxPerRequest ?? 50
 
     this.in = new Heap({
       comparBefore: (

packages/client/lib/sync/fullsync.ts

@@ -1,34 +1,105 @@
 import { BN } from 'ethereumjs-util'
-import VM from '@ethereumjs/vm'
 import { Peer } from '../net/peer/peer'
 import { short } from '../util'
 import { Synchronizer, SynchronizerOptions } from './sync'
 import { BlockFetcher } from './fetcher'
 import { Block } from '@ethereumjs/block'
+import VM from '@ethereumjs/vm'
+import { DefaultStateManager } from '@ethereumjs/vm/dist/state'
+import { SecureTrie as Trie } from '@ethereumjs/trie'
 
 /**
  * Implements an ethereum full sync synchronizer
  * @memberof module:sync
  */
 export class FullSynchronizer extends Synchronizer {
+  private blockFetcher: BlockFetcher | null
+
   public vm: VM
+  public runningBlocks: boolean
 
-  private blockFetcher: BlockFetcher | null
+  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
 
     if (!this.config.vm) {
+      const trie = new Trie(this.stateDB)
+
+      const stateManager = new DefaultStateManager({
+        common: this.config.common,
+        trie,
+      })
+
       this.vm = new VM({
         common: this.config.common,
         blockchain: this.chain.blockchain,
+        stateManager,
       })
     } else {
       this.vm = this.config.vm
       //@ts-ignore blockchain has readonly property
       this.vm.blockchain = this.chain.blockchain
     }
+
+    this.runningBlocks = false
+    this.stopSyncing = false
+
+    const self = this
+    this.chain.on('updated', async function () {
+      // for some reason, if we use .on('updated', this.runBlocks), it runs in the context of the Chain and not in the FullSync context..?
+      await self.runBlocks()
+    })
+    // eslint-disable-next-line @typescript-eslint/no-floating-promises
+    this.chain.update()
+  }
+
+  /**
+   * This updates the VM once blocks were put in the VM
+   */
+  async runBlocks() {
+    if (!this.running || this.runningBlocks) {
+      return
+    }
+    this.runningBlocks = true
+    try {
+      let oldHead = Buffer.alloc(0)
+      let newHead = (await this.vm.blockchain.getHead()).hash()
+      while (!newHead.equals(oldHead) && !this.stopSyncing) {
+        oldHead = newHead
+        this.vmPromise = this.vm.runBlockchain(this.vm.blockchain, 1)
+        await this.vmPromise
+        const headBlock = await this.vm.blockchain.getHead()
+        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}`)
+          }
+        }
+      }
+    } finally {
+      this.runningBlocks = false
+    }
   }
 
   /**
@@ -167,6 +238,13 @@ export class FullSynchronizer extends Synchronizer {
    * @return {Promise}
    */
   async stop(): Promise<boolean> {
+    this.stopSyncing = true
+    if (this.vmPromise) {
+      // ensure that we wait that the VM finishes executing the block (and flushes the trie cache)
+      await this.vmPromise
+    }
+    await this.stateDB?.close()
+
     if (!this.running) {
       return false
     }
@@ -177,6 +255,7 @@ export class FullSynchronizer extends Synchronizer {
       delete this.blockFetcher
     }
     await super.stop()
+
     return true
   }
 }

packages/client/lib/sync/sync.ts

@@ -4,6 +4,7 @@ import { Peer } from '../net/peer/peer'
 import { FlowControl } from '../net/protocol'
 import { Config } from '../config'
 import { Chain } from '../blockchain'
+import { LevelUp } from 'levelup'
 
 export interface SynchronizerOptions {
   /* Config */
@@ -15,6 +16,9 @@ export interface SynchronizerOptions {
   /* Blockchain */
   chain: Chain
 
+  /* State database */
+  stateDB?: LevelUp
+
   /* Flow control manager */
   flow?: FlowControl
 
@@ -31,9 +35,10 @@ export abstract class Synchronizer extends EventEmitter {
 
   protected pool: PeerPool
   protected chain: Chain
+  protected stateDB?: LevelUp
   protected flow: FlowControl
   protected interval: number
-  protected running: boolean
+  public running: boolean
   protected forceSync: boolean
 
   /**
@@ -47,6 +52,7 @@ export abstract class Synchronizer extends EventEmitter {
 
     this.pool = options.pool
     this.chain = options.chain
+    this.stateDB = options.stateDB
     this.flow = options.flow ?? new FlowControl()
     this.interval = options.interval ?? 1000
     this.running = false
@@ -59,7 +65,7 @@ export abstract class Synchronizer extends EventEmitter {
     })
   }
 
-  abstract async sync(): Promise<boolean>
+  abstract sync(): Promise<boolean>
 
   /**
    * Returns synchronizer type