Merge pull request #36 from vulcanize/use-async-iter

Use concurrent node iterator

cmd/validateTrie.go

@@ -59,7 +59,8 @@ It can operate at three levels:
 }
 
 func validateTrie() {
-	v, err := newValidator()
+	numWorkers := viper.GetUint("validator.workers")
+	v, err := newValidator(numWorkers)
 	if err != nil {
 		logWithCommand.Fatal(err)
 	}
@@ -104,20 +105,20 @@ func validateTrie() {
 	logWithCommand.Debugf("groupcache stats %+v", stats)
 }
 
-func newValidator() (*validator.Validator, error) {
+func newValidator(workers uint) (*validator.Validator, error) {
 	ipfsPath := viper.GetString("ipfs.path")
 	if ipfsPath == "" {
 		db, err := validator.NewDB()
 		if err != nil {
 			logWithCommand.Fatal(err)
 		}
-		return validator.NewPGIPFSValidator(db), nil
+		return validator.NewPGIPFSValidator(db, workers), nil
 	}
 	bs, err := validator.InitIPFSBlockService(ipfsPath)
 	if err != nil {
 		return nil, err
 	}
-	return validator.NewIPFSValidator(bs), nil
+	return validator.NewIPFSValidator(bs, workers), nil
 }
 
 func init() {
@@ -128,10 +129,12 @@ func init() {
 	validateTrieCmd.PersistentFlags().String("storage-root", "", "Root of the storage trie we wish to validate; for storage validation")
 	validateTrieCmd.PersistentFlags().String("address", "", "Contract address for the storage trie we wish to validate; for storage validation")
 	validateTrieCmd.PersistentFlags().String("ipfs-path", "", "Path to IPFS repository; if provided operations move through the IPFS repo otherwise Postgres connection params are expected in the provided config")
+	validateTrieCmd.PersistentFlags().Int("workers", 4, "number of concurrent workers to use")
 
 	viper.BindPFlag("validator.stateRoot", validateTrieCmd.PersistentFlags().Lookup("state-root"))
 	viper.BindPFlag("validator.type", validateTrieCmd.PersistentFlags().Lookup("type"))
 	viper.BindPFlag("validator.storageRoot", validateTrieCmd.PersistentFlags().Lookup("storage-root"))
 	viper.BindPFlag("validator.address", validateTrieCmd.PersistentFlags().Lookup("address"))
+	viper.BindPFlag("validator.workers", validateTrieCmd.PersistentFlags().Lookup("workers"))
 	viper.BindPFlag("ipfs.path", validateTrieCmd.PersistentFlags().Lookup("ipfs-path"))
 }

go.mod

@@ -118,6 +118,8 @@ require (
 	github.com/klauspost/compress v1.15.1 // indirect
 	github.com/klauspost/cpuid/v2 v2.0.12 // indirect
 	github.com/koron/go-ssdp v0.0.2 // indirect
+	github.com/kr/text v0.2.0 // indirect
+	github.com/libp2p/go-addr-util v0.1.0 // indirect
 	github.com/libp2p/go-buffer-pool v0.0.2 // indirect
 	github.com/libp2p/go-cidranger v1.1.0 // indirect
 	github.com/libp2p/go-doh-resolver v0.4.0 // indirect
@@ -203,6 +205,9 @@ require (
 	github.com/tklauser/go-sysconf v0.3.5 // indirect
 	github.com/tklauser/numcpus v0.2.2 // indirect
 	github.com/wI2L/jsondiff v0.2.0 // indirect
+	github.com/tklauser/go-sysconf v0.3.5 // indirect
+	github.com/tklauser/numcpus v0.2.2 // indirect
+	github.com/vulcanize/go-eth-state-node-iterator v1.1.4
 	github.com/whyrusleeping/base32 v0.0.0-20170828182744-c30ac30633cc // indirect
 	github.com/whyrusleeping/cbor-gen v0.0.0-20210219115102-f37d292932f2 // indirect
 	github.com/whyrusleeping/chunker v0.0.0-20181014151217-fe64bd25879f // indirect