Synchronize with chain using parallel operations

blockbook.go

@@ -20,6 +20,7 @@ import (
 
 type Blockchain interface {
 	GetBestBlockHash() (string, error)
+	GetBestBlockHeight() (uint32, error)
 	GetBlockHash(height uint32) (string, error)
 	GetBlockHeader(hash string) (*bitcoin.BlockHeader, error)
 	GetBlock(hash string) (*bitcoin.Block, error)
@@ -154,7 +155,7 @@ func main() {
 			if err = db.GetTransactions(script, height, until, printResult); err != nil {
 				log.Fatalf("GetTransactions %v", err)
 			}
-		} else {
+		} else if !*resync {
 			if err = connectBlocksParallel(
 				rpc,
 				db,
@@ -225,57 +226,94 @@ func resyncIndex(chain Blockchain, index Index) error {
 		return nil
 	}
 
-	// If the local block is missing, we're indexing from the genesis block.
+	var header *bitcoin.BlockHeader
-	if local == "" {
+	if local != "" {
-		log.Printf("resync: genesis")
+		// Is local tip on the best chain?
-
+		header, err = chain.GetBlockHeader(local)
-		hash, err := chain.GetBlockHash(0)
+		forked := false
 		if err != nil {
-			return err
+			if e, ok := err.(*bitcoin.RPCError); ok && e.Message == "Block not found" {
-		}
+				forked = true
-		return connectBlocks(chain, index, hash)
+			} else {
-	}
+				return err
-
+			}
-	// Is local tip on the best chain?
-	header, err := chain.GetBlockHeader(local)
-	forked := false
-	if err != nil {
-		if e, ok := err.(*bitcoin.RPCError); ok && e.Message == "Block not found" {
-			forked = true
 		} else {
-			return err
+			if header.Confirmations < 0 {
+				forked = true
+			}
 		}
-	} else {
+
-		if header.Confirmations < 0 {
+		if forked {
-			forked = true
+			// find and disconnect forked blocks and then synchronize again
+			log.Printf("resync: local is forked")
+			var height uint32
+			for height = localBestHeight - 1; height >= 0; height-- {
+				local, err = index.GetBlockHash(height)
+				if err != nil {
+					return err
+				}
+				remote, err = chain.GetBlockHash(height)
+				if err != nil {
+					return err
+				}
+				if local == remote {
+					break
+				}
+			}
+			err = index.DisconnectBlocks(height+1, localBestHeight)
+			if err != nil {
+				return err
+			}
+			return resyncIndex(chain, index)
 		}
 	}
 
-	if forked {
+	startHeight := uint32(0)
-		log.Printf("resync: local is forked")
+	var hash string
-		var height uint32
+	if header != nil {
-		for height = localBestHeight - 1; height >= 0; height-- {
+		log.Printf("resync: local is behind")
-			local, err = index.GetBlockHash(height)
+		hash = header.Next
-			if err != nil {
+		startHeight = localBestHeight
-				return err
+	} else {
-			}
+		// If the local block is missing, we're indexing from the genesis block
-			remote, err = chain.GetBlockHash(height)
+		// or from the start block specified by flags
-			if err != nil {
+		if *blockHeight > 0 {
-				return err
+			startHeight = uint32(*blockHeight)
-			}
-			if local == remote {
-				break
-			}
 		}
-		err = index.DisconnectBlocks(height+1, localBestHeight)
+		log.Printf("resync: genesis from block %d", startHeight)
+		hash, err = chain.GetBlockHash(startHeight)
 		if err != nil {
 			return err
 		}
-		return resyncIndex(chain, index)
 	}
 
-	log.Printf("resync: local is behind")
+	// if parallel operation is enabled and the number of blocks to be connected is large,
-	return connectBlocks(chain, index, header.Next)
+	// use parallel routine to load majority of blocks
+	if *syncWorkers > 1 {
+		chainBestHeight, err := chain.GetBestBlockHeight()
+		if err != nil {
+			return err
+		}
+		if chainBestHeight-startHeight > uint32(*syncChunk) {
+			log.Printf("resync: parallel sync of blocks %d-%d", startHeight, chainBestHeight)
+			err = connectBlocksParallel(
+				chain,
+				index,
+				startHeight,
+				chainBestHeight,
+				*syncChunk,
+				*syncWorkers,
+			)
+			if err != nil {
+				return err
+			}
+			// after parallel load finish the sync using standard way,
+			// new blocks may have been created in the meantime
+			return resyncIndex(chain, index)
+		}
+	}
+
+	return connectBlocks(chain, index, hash)
 }
 
 func connectBlocks(