Merge pull request #3184 from karalabe/import-gas-use

cmd/geth, code, eth/downloader: tune import logs and mem stats

cmd/geth/chaincmd.go

@@ -20,7 +20,9 @@ import (
 	"fmt"
 	"os"
 	"path/filepath"
+	"runtime"
 	"strconv"
+	"sync/atomic"
 	"time"
 
 	"github.com/ethereum/go-ethereum/cmd/utils"
@@ -85,37 +87,62 @@ func importChain(ctx *cli.Context) error {
 	chain, chainDb := utils.MakeChain(ctx, stack)
 	defer chainDb.Close()
 
+	// Start periodically gathering memory profiles
+	var peakMemAlloc, peakMemSys uint64
+	go func() {
+		stats := new(runtime.MemStats)
+		for {
+			runtime.ReadMemStats(stats)
+			if atomic.LoadUint64(&peakMemAlloc) < stats.Alloc {
+				atomic.StoreUint64(&peakMemAlloc, stats.Alloc)
+			}
+			if atomic.LoadUint64(&peakMemSys) < stats.Sys {
+				atomic.StoreUint64(&peakMemSys, stats.Sys)
+			}
+			time.Sleep(5 * time.Second)
+		}
+	}()
 	// Import the chain
 	start := time.Now()
 	if err := utils.ImportChain(chain, ctx.Args().First()); err != nil {
 		utils.Fatalf("Import error: %v", err)
 	}
-	fmt.Printf("Import done in %v.\n", time.Since(start))
+	fmt.Printf("Import done in %v.\n\n", time.Since(start))
 
-	if db, ok := chainDb.(*ethdb.LDBDatabase); ok {
-		// Output pre-compaction stats mostly to see the import trashing
-		stats, err := db.LDB().GetProperty("leveldb.stats")
-		if err != nil {
-			utils.Fatalf("Failed to read database stats: %v", err)
-		}
-		fmt.Println(stats)
-		fmt.Printf("Trie cache misses: %d\n", trie.CacheMisses())
-		fmt.Printf("Trie cache unloads: %d\n\n", trie.CacheUnloads())
+	// Output pre-compaction stats mostly to see the import trashing
+	db := chainDb.(*ethdb.LDBDatabase)
 
-		// Compact the entire database to more accurately measure disk io and print the stats
-		start = time.Now()
-		fmt.Println("Compacting entire database...")
-		if err = db.LDB().CompactRange(util.Range{}); err != nil {
-			utils.Fatalf("Compaction failed: %v", err)
-		}
-		fmt.Printf("Compaction done in %v.\n", time.Since(start))
+	stats, err := db.LDB().GetProperty("leveldb.stats")
+	if err != nil {
+		utils.Fatalf("Failed to read database stats: %v", err)
+	}
+	fmt.Println(stats)
+	fmt.Printf("Trie cache misses:  %d\n", trie.CacheMisses())
+	fmt.Printf("Trie cache unloads: %d\n\n", trie.CacheUnloads())
 
-		stats, err = db.LDB().GetProperty("leveldb.stats")
-		if err != nil {
-			utils.Fatalf("Failed to read database stats: %v", err)
-		}
-		fmt.Println(stats)
+	// Print the memory statistics used by the importing
+	mem := new(runtime.MemStats)
+	runtime.ReadMemStats(mem)
+
+	fmt.Printf("Object memory: %.3f MB current, %.3f MB peak\n", float64(mem.Alloc)/1024/1024, float64(atomic.LoadUint64(&peakMemAlloc))/1024/1024)
+	fmt.Printf("System memory: %.3f MB current, %.3f MB peak\n", float64(mem.Sys)/1024/1024, float64(atomic.LoadUint64(&peakMemSys))/1024/1024)
+	fmt.Printf("Allocations:   %.3f million\n", float64(mem.Mallocs)/1000000)
+	fmt.Printf("GC pause:      %v\n\n", time.Duration(mem.PauseTotalNs))
+
+	// Compact the entire database to more accurately measure disk io and print the stats
+	start = time.Now()
+	fmt.Println("Compacting entire database...")
+	if err = db.LDB().CompactRange(util.Range{}); err != nil {
+		utils.Fatalf("Compaction failed: %v", err)
 	}
+	fmt.Printf("Compaction done in %v.\n\n", time.Since(start))
+
+	stats, err = db.LDB().GetProperty("leveldb.stats")
+	if err != nil {
+		utils.Fatalf("Failed to read database stats: %v", err)
+	}
+	fmt.Println(stats)
+
 	return nil
 }
 

core/blockchain.go

@@ -774,7 +774,7 @@ func (self *BlockChain) InsertReceiptChain(blockChain types.Blocks, receiptChain
 	if stats.ignored > 0 {
 		ignored = fmt.Sprintf(" (%d ignored)", stats.ignored)
 	}
-	glog.V(logger.Info).Infof("imported %d receipts%s in %9v. #%d [%x… / %x…]", stats.processed, ignored, common.PrettyDuration(time.Since(start)), last.Number(), first.Hash().Bytes()[:4], last.Hash().Bytes()[:4])
+	glog.V(logger.Info).Infof("imported %d receipts in %9v. #%d [%x… / %x…]%s", stats.processed, common.PrettyDuration(time.Since(start)), last.Number(), first.Hash().Bytes()[:4], last.Hash().Bytes()[:4], ignored)
 
 	return 0, nil
 }
@@ -981,6 +981,7 @@ func (self *BlockChain) InsertChain(chain types.Blocks) (int, error) {
 
 		stats.processed++
 		if glog.V(logger.Info) {
+			stats.usedGas += usedGas.Uint64()
 			stats.report(chain, i)
 		}
 	}
@@ -993,6 +994,7 @@ func (self *BlockChain) InsertChain(chain types.Blocks) (int, error) {
 // insertStats tracks and reports on block insertion.
 type insertStats struct {
 	queued, processed, ignored int
+	usedGas                    uint64
 	lastIndex                  int
 	startTime                  time.Time
 }
@@ -1004,10 +1006,15 @@ const statsReportLimit = 8 * time.Second
 // report prints statistics if some number of blocks have been processed
 // or more than a few seconds have passed since the last message.
 func (st *insertStats) report(chain []*types.Block, index int) {
+	// Fetch the timings for the batch
 	var (
 		now     = time.Now()
 		elapsed = now.Sub(st.startTime)
 	)
+	if elapsed == 0 { // Yes Windows, I'm looking at you
+		elapsed = 1
+	}
+	// If we're at the last block of the batch or report period reached, log
 	if index == len(chain)-1 || elapsed >= statsReportLimit {
 		start, end := chain[st.lastIndex], chain[index]
 		txcount := countTransactions(chain[st.lastIndex : index+1])
@@ -1016,7 +1023,13 @@ func (st *insertStats) report(chain []*types.Block, index int) {
 		if st.queued > 0 || st.ignored > 0 {
 			extra = fmt.Sprintf(" (%d queued %d ignored)", st.queued, st.ignored)
 		}
-		glog.Infof("imported %d blocks%s, %5d txs in %9v. #%v [%x… / %x…]\n", st.processed, extra, txcount, common.PrettyDuration(elapsed), end.Number(), start.Hash().Bytes()[:4], end.Hash().Bytes()[:4])
+		hashes := ""
+		if st.processed > 1 {
+			hashes = fmt.Sprintf("%x… / %x…", start.Hash().Bytes()[:4], end.Hash().Bytes()[:4])
+		} else {
+			hashes = fmt.Sprintf("%x…", end.Hash().Bytes()[:4])
+		}
+		glog.Infof("imported %d blocks, %5d txs (%7.3f Mg) in %9v (%6.3f Mg/s). #%v [%s]%s", st.processed, txcount, float64(st.usedGas)/1000000, common.PrettyDuration(elapsed), float64(st.usedGas)*1000/float64(elapsed), end.Number(), hashes, extra)
 
 		*st = insertStats{startTime: now, lastIndex: index}
 	}

core/blockchain_test.go

@@ -436,7 +436,7 @@ func (bproc) ValidateState(block, parent *types.Block, state *state.StateDB, rec
 	return nil
 }
 func (bproc) Process(block *types.Block, statedb *state.StateDB, cfg vm.Config) (types.Receipts, vm.Logs, *big.Int, error) {
-	return nil, nil, nil, nil
+	return nil, nil, new(big.Int), nil
 }
 
 func makeHeaderChainWithDiff(genesis *types.Block, d []int, seed byte) []*types.Header {

eth/downloader/downloader.go

@@ -952,7 +952,7 @@ func (d *Downloader) fetchNodeData() error {
 
 				// Log a message to the user and return
 				if delivered > 0 {
-					glog.V(logger.Info).Infof("imported %d state entries in %9v: processed %d, pending at least %d", delivered, common.PrettyDuration(time.Since(start)), d.syncStatsStateDone, pending)
+					glog.V(logger.Info).Infof("imported %3d state entries in %9v: processed %d, pending at least %d", delivered, common.PrettyDuration(time.Since(start)), d.syncStatsStateDone, pending)
 				}
 			})
 		}