core, ethdb, trie: restart fast-sync bloom construction now and again

core/rawdb/table.go

@@ -67,6 +67,13 @@ func (t *table) NewIterator() ethdb.Iterator {
 	return t.NewIteratorWithPrefix(nil)
 }
 
+// NewIteratorWithStart creates a binary-alphabetical iterator over a subset of
+// database content starting at a particular initial key (or after, if it does
+// not exist).
+func (t *table) NewIteratorWithStart(start []byte) ethdb.Iterator {
+	return t.db.NewIteratorWithStart(start)
+}
+
 // NewIteratorWithPrefix creates a binary-alphabetical iterator over a subset
 // of database content with a particular key prefix.
 func (t *table) NewIteratorWithPrefix(prefix []byte) ethdb.Iterator {

ethdb/iterator.go

@@ -55,6 +55,11 @@ type Iteratee interface {
 	// contained within the key-value database.
 	NewIterator() Iterator
 
+	// NewIteratorWithStart creates a binary-alphabetical iterator over a subset of
+	// database content starting at a particular initial key (or after, if it does
+	// not exist).
+	NewIteratorWithStart(start []byte) Iterator
+
 	// NewIteratorWithPrefix creates a binary-alphabetical iterator over a subset
 	// of database content with a particular key prefix.
 	NewIteratorWithPrefix(prefix []byte) Iterator

ethdb/leveldb/leveldb.go

@@ -175,7 +175,14 @@ func (db *Database) NewBatch() ethdb.Batch {
 // NewIterator creates a binary-alphabetical iterator over the entire keyspace
 // contained within the leveldb database.
 func (db *Database) NewIterator() ethdb.Iterator {
-	return db.NewIteratorWithPrefix(nil)
+	return db.db.NewIterator(new(util.Range), nil)
+}
+
+// NewIteratorWithStart creates a binary-alphabetical iterator over a subset of
+// database content starting at a particular initial key (or after, if it does
+// not exist).
+func (db *Database) NewIteratorWithStart(start []byte) ethdb.Iterator {
+	return db.db.NewIterator(&util.Range{Start: start}, nil)
 }
 
 // NewIteratorWithPrefix creates a binary-alphabetical iterator over a subset

ethdb/memorydb/memorydb.go

@@ -132,7 +132,36 @@ func (db *Database) NewBatch() ethdb.Batch {
 // NewIterator creates a binary-alphabetical iterator over the entire keyspace
 // contained within the memory database.
 func (db *Database) NewIterator() ethdb.Iterator {
-	return db.NewIteratorWithPrefix(nil)
+	return db.NewIteratorWithStart(nil)
+}
+
+// NewIteratorWithStart creates a binary-alphabetical iterator over a subset of
+// database content starting at a particular initial key (or after, if it does
+// not exist).
+func (db *Database) NewIteratorWithStart(start []byte) ethdb.Iterator {
+	db.lock.RLock()
+	defer db.lock.RUnlock()
+
+	var (
+		st     = string(start)
+		keys   = make([]string, 0, len(db.db))
+		values = make([][]byte, 0, len(db.db))
+	)
+	// Collect the keys from the memory database corresponding to the given start
+	for key := range db.db {
+		if key >= st {
+			keys = append(keys, key)
+		}
+	}
+	// Sort the items and retrieve the associated values
+	sort.Strings(keys)
+	for _, key := range keys {
+		values = append(values, db.db[key])
+	}
+	return &iterator{
+		keys:   keys,
+		values: values,
+	}
 }
 
 // NewIteratorWithPrefix creates a binary-alphabetical iterator over a subset

trie/sync_bloom.go

@@ -92,19 +92,40 @@ func NewSyncBloom(memory uint64, database ethdb.Iteratee) *SyncBloom {
 
 // init iterates over the database, pushing every trie hash into the bloom filter.
 func (b *SyncBloom) init(database ethdb.Iteratee) {
+	// Iterate over the database, but restart every now and again to avoid holding
+	// a persistent snapshot since fast sync can push a ton of data concurrently,
+	// bloating the disk.
+	//
+	// Note, this is fine, because everything inserted into leveldb by fast sync is
+	// also pushed into the bloom directly, so we're not missing anything when the
+	// iterator is swapped out for a new one.
 	it := database.NewIterator()
-	defer it.Release()
 
-	start := time.Now()
+	var (
+		start = time.Now()
+		swap  = time.Now()
+	)
 	for it.Next() && atomic.LoadUint32(&b.closed) == 0 {
+		// If the database entry is a trie node, add it to the bloom
 		if key := it.Key(); len(key) == common.HashLength {
 			b.bloom.Add(syncBloomHasher(key))
 			bloomLoadMeter.Mark(1)
 		}
+		// If enough time elapsed since the last iterator swap, restart
+		if time.Since(swap) > 8*time.Second {
+			key := common.CopyBytes(it.Key())
+
+			it.Release()
+			it = database.NewIteratorWithStart(key)
+
+			log.Info("Initializing fast sync bloom", "items", b.bloom.N(), "errorrate", b.errorRate(), "elapsed", time.Since(start))
+			swap = time.Now()
 		}
-	log.Info("Initialized fast sync bloom", "items", b.bloom.N(), "errorrate", b.errorRate(), "elapsed", time.Since(start))
+	}
+	it.Release()
 
 	// Mark the bloom filter inited and return
+	log.Info("Initialized fast sync bloom", "items", b.bloom.N(), "errorrate", b.errorRate(), "elapsed", time.Since(start))
 	atomic.StoreUint32(&b.inited, 1)
 }