core, eth, trie: support batched trie sync db writes

core/state/sync.go

@@ -59,10 +59,16 @@ func (s *StateSync) Missing(max int) []common.Hash {
 }
 
 // Process injects a batch of retrieved trie nodes data, returning if something
-// was committed to the database and also the index of an entry if processing of
+// was committed to the memcache and also the index of an entry if processing of
 // it failed.
-func (s *StateSync) Process(list []trie.SyncResult, dbw trie.DatabaseWriter) (bool, int, error) {
+func (s *StateSync) Process(list []trie.SyncResult) (bool, int, error) {
-	return (*trie.TrieSync)(s).Process(list, dbw)
+	return (*trie.TrieSync)(s).Process(list)
+}
+
+// Commit flushes the data stored in the internal memcache out to persistent
+// storage, returning th enumber of items written and any occurred error.
+func (s *StateSync) Commit(dbw trie.DatabaseWriter) (int, error) {
+	return (*trie.TrieSync)(s).Commit(dbw)
 }
 
 // Pending returns the number of state entries currently pending for download.

core/state/sync_test.go

@@ -138,9 +138,12 @@ func testIterativeStateSync(t *testing.T, batch int) {
 			}
 			results[i] = trie.SyncResult{Hash: hash, Data: data}
 		}
-		if _, index, err := sched.Process(results, dstDb); err != nil {
+		if _, index, err := sched.Process(results); err != nil {
 			t.Fatalf("failed to process result #%d: %v", index, err)
 		}
+		if index, err := sched.Commit(dstDb); err != nil {
+			t.Fatalf("failed to commit data #%d: %v", index, err)
+		}
 		queue = append(queue[:0], sched.Missing(batch)...)
 	}
 	// Cross check that the two states are in sync
@@ -168,9 +171,12 @@ func TestIterativeDelayedStateSync(t *testing.T) {
 			}
 			results[i] = trie.SyncResult{Hash: hash, Data: data}
 		}
-		if _, index, err := sched.Process(results, dstDb); err != nil {
+		if _, index, err := sched.Process(results); err != nil {
 			t.Fatalf("failed to process result #%d: %v", index, err)
 		}
+		if index, err := sched.Commit(dstDb); err != nil {
+			t.Fatalf("failed to commit data #%d: %v", index, err)
+		}
 		queue = append(queue[len(results):], sched.Missing(0)...)
 	}
 	// Cross check that the two states are in sync
@@ -206,9 +212,12 @@ func testIterativeRandomStateSync(t *testing.T, batch int) {
 			results = append(results, trie.SyncResult{Hash: hash, Data: data})
 		}
 		// Feed the retrieved results back and queue new tasks
-		if _, index, err := sched.Process(results, dstDb); err != nil {
+		if _, index, err := sched.Process(results); err != nil {
 			t.Fatalf("failed to process result #%d: %v", index, err)
 		}
+		if index, err := sched.Commit(dstDb); err != nil {
+			t.Fatalf("failed to commit data #%d: %v", index, err)
+		}
 		queue = make(map[common.Hash]struct{})
 		for _, hash := range sched.Missing(batch) {
 			queue[hash] = struct{}{}
@@ -249,9 +258,12 @@ func TestIterativeRandomDelayedStateSync(t *testing.T) {
 			}
 		}
 		// Feed the retrieved results back and queue new tasks
-		if _, index, err := sched.Process(results, dstDb); err != nil {
+		if _, index, err := sched.Process(results); err != nil {
 			t.Fatalf("failed to process result #%d: %v", index, err)
 		}
+		if index, err := sched.Commit(dstDb); err != nil {
+			t.Fatalf("failed to commit data #%d: %v", index, err)
+		}
 		for _, hash := range sched.Missing(0) {
 			queue[hash] = struct{}{}
 		}
@@ -283,9 +295,12 @@ func TestIncompleteStateSync(t *testing.T) {
 			results[i] = trie.SyncResult{Hash: hash, Data: data}
 		}
 		// Process each of the state nodes
-		if _, index, err := sched.Process(results, dstDb); err != nil {
+		if _, index, err := sched.Process(results); err != nil {
 			t.Fatalf("failed to process result #%d: %v", index, err)
 		}
+		if index, err := sched.Commit(dstDb); err != nil {
+			t.Fatalf("failed to commit data #%d: %v", index, err)
+		}
 		for _, result := range results {
 			added = append(added, result.Hash)
 		}

eth/downloader/statesync.go

@@ -347,11 +347,11 @@ func (s *stateSync) fillTasks(n int, req *stateReq) {
 // delivered.
 func (s *stateSync) process(req *stateReq) (bool, error) {
 	// Collect processing stats and update progress if valid data was received
-	processed, duplicate, unexpected := 0, 0, 0
+	processed, written, duplicate, unexpected := 0, 0, 0, 0
 
 	defer func(start time.Time) {
-		if processed+duplicate+unexpected > 0 {
+		if processed+written+duplicate+unexpected > 0 {
-			s.updateStats(processed, duplicate, unexpected, time.Since(start))
+			s.updateStats(processed, written, duplicate, unexpected, time.Since(start))
 		}
 	}(time.Now())
 
@@ -379,10 +379,25 @@ func (s *stateSync) process(req *stateReq) (bool, error) {
 			stale = false
 		}
 	}
-	// If some data managed to hit the database, sync progressed so reset any pivot fail counter
+	// If some data managed to hit the database, flush and reset failure counters
-	if progress && atomic.LoadUint32(&s.d.fsPivotFails) > 1 {
+	if progress {
-		log.Trace("Fast-sync progressed, resetting fail counter", "previous", atomic.LoadUint32(&s.d.fsPivotFails))
+		// Flush any accumulated data out to disk
-		atomic.StoreUint32(&s.d.fsPivotFails, 1) // Don't ever reset to 0, as that will unlock the pivot block
+		batch := s.d.stateDB.NewBatch()
+
+		count, err := s.sched.Commit(batch)
+		if err != nil {
+			return stale, err
+		}
+		if err := batch.Write(); err != nil {
+			return stale, err
+		}
+		written = count
+
+		// If we're inside the critical section, reset fail counter since we progressed
+		if atomic.LoadUint32(&s.d.fsPivotFails) > 1 {
+			log.Trace("Fast-sync progressed, resetting fail counter", "previous", atomic.LoadUint32(&s.d.fsPivotFails))
+			atomic.StoreUint32(&s.d.fsPivotFails, 1) // Don't ever reset to 0, as that will unlock the pivot block
+		}
 	}
 	// Put unfulfilled tasks back into the retry queue
 	npeers := s.d.peers.Len()
@@ -409,28 +424,19 @@ func (s *stateSync) process(req *stateReq) (bool, error) {
 // peer into the state trie, returning whether anything useful was written or any
 // error occurred.
 func (s *stateSync) processNodeData(blob []byte) (bool, common.Hash, error) {
-	// Convert the delivered data blob into a trie sync result
 	res := trie.SyncResult{Data: blob}
 
 	s.keccak.Reset()
 	s.keccak.Write(blob)
 	s.keccak.Sum(res.Hash[:0])
 
-	// Process the trie node and write any finalized data
+	committed, _, err := s.sched.Process([]trie.SyncResult{res})
-	batch := s.d.stateDB.NewBatch()
-
-	committed, _, err := s.sched.Process([]trie.SyncResult{res}, batch)
-	if committed {
-		if werr := batch.Write(); werr != nil {
-			return false, common.Hash{}, werr
-		}
-	}
 	return committed, res.Hash, err
 }
 
 // updateStats bumps the various state sync progress counters and displays a log
 // message for the user to see.
-func (s *stateSync) updateStats(processed, duplicate, unexpected int, duration time.Duration) {
+func (s *stateSync) updateStats(processed, written, duplicate, unexpected int, duration time.Duration) {
 	s.d.syncStatsLock.Lock()
 	defer s.d.syncStatsLock.Unlock()
 
@@ -439,5 +445,5 @@ func (s *stateSync) updateStats(processed, duplicate, unexpected int, duration t
 	s.d.syncStatsState.duplicate += uint64(duplicate)
 	s.d.syncStatsState.unexpected += uint64(unexpected)
 
-	log.Info("Imported new state entries", "count", processed, "elapsed", common.PrettyDuration(duration), "processed", s.d.syncStatsState.processed, "pending", s.d.syncStatsState.pending, "retry", len(s.tasks), "duplicate", s.d.syncStatsState.duplicate, "unexpected", s.d.syncStatsState.unexpected)
+	log.Info("Imported new state entries", "count", processed, "flushed", written, "elapsed", common.PrettyDuration(duration), "processed", s.d.syncStatsState.processed, "pending", s.d.syncStatsState.pending, "retry", len(s.tasks), "duplicate", s.d.syncStatsState.duplicate, "unexpected", s.d.syncStatsState.unexpected)
 }

trie/sync.go

@@ -52,6 +52,21 @@ type SyncResult struct {
 	Data []byte      // Data content of the retrieved node
 }
 
+// SyncMemCache is an in-memory cache of successfully downloaded but not yet
+// persisted data items.
+type SyncMemCache struct {
+	cache map[common.Hash][]byte // In-memory memcache of recently ocmpleted items
+	order []common.Hash          // Order of completion to prevent out-of-order data loss
+}
+
+// newSyncMemCache allocates a new memory-cache for not-yet persisted trie nodes.
+func newSyncMemCache() *SyncMemCache {
+	return &SyncMemCache{
+		cache: make(map[common.Hash][]byte),
+		order: make([]common.Hash, 0, 256),
+	}
+}
+
 // TrieSyncLeafCallback is a callback type invoked when a trie sync reaches a
 // leaf node. It's used by state syncing to check if the leaf node requires some
 // further data syncing.
@@ -61,7 +76,8 @@ type TrieSyncLeafCallback func(leaf []byte, parent common.Hash) error
 // unknown trie hashes to retrieve, accepts node data associated with said hashes
 // and reconstructs the trie step by step until all is done.
 type TrieSync struct {
-	database DatabaseReader
+	database DatabaseReader           // Persistent database to check for existing entries
+	memcache *SyncMemCache            // Memory cache to avoid frequest database writes
 	requests map[common.Hash]*request // Pending requests pertaining to a key hash
 	queue    *prque.Prque             // Priority queue with the pending requests
 }
@@ -70,6 +86,7 @@ type TrieSync struct {
 func NewTrieSync(root common.Hash, database DatabaseReader, callback TrieSyncLeafCallback) *TrieSync {
 	ts := &TrieSync{
 		database: database,
+		memcache: newSyncMemCache(),
 		requests: make(map[common.Hash]*request),
 		queue:    prque.New(),
 	}
@@ -83,6 +100,9 @@ func (s *TrieSync) AddSubTrie(root common.Hash, depth int, parent common.Hash, c
 	if root == emptyRoot {
 		return
 	}
+	if _, ok := s.memcache.cache[root]; ok {
+		return
+	}
 	key := root.Bytes()
 	blob, _ := s.database.Get(key)
 	if local, err := decodeNode(key, blob, 0); local != nil && err == nil {
@@ -115,6 +135,9 @@ func (s *TrieSync) AddRawEntry(hash common.Hash, depth int, parent common.Hash)
 	if hash == emptyState {
 		return
 	}
+	if _, ok := s.memcache.cache[hash]; ok {
+		return
+	}
 	if blob, _ := s.database.Get(hash.Bytes()); blob != nil {
 		return
 	}
@@ -148,7 +171,7 @@ func (s *TrieSync) Missing(max int) []common.Hash {
 // Process injects a batch of retrieved trie nodes data, returning if something
 // was committed to the database and also the index of an entry if processing of
 // it failed.
-func (s *TrieSync) Process(results []SyncResult, dbw DatabaseWriter) (bool, int, error) {
+func (s *TrieSync) Process(results []SyncResult) (bool, int, error) {
 	committed := false
 
 	for i, item := range results {
@@ -163,7 +186,7 @@ func (s *TrieSync) Process(results []SyncResult, dbw DatabaseWriter) (bool, int,
 		// If the item is a raw entry request, commit directly
 		if request.raw {
 			request.data = item.Data
-			s.commit(request, dbw)
+			s.commit(request)
 			committed = true
 			continue
 		}
@@ -180,7 +203,7 @@ func (s *TrieSync) Process(results []SyncResult, dbw DatabaseWriter) (bool, int,
 			return committed, i, err
 		}
 		if len(requests) == 0 && request.deps == 0 {
-			s.commit(request, dbw)
+			s.commit(request)
 			committed = true
 			continue
 		}
@@ -192,6 +215,22 @@ func (s *TrieSync) Process(results []SyncResult, dbw DatabaseWriter) (bool, int,
 	return committed, 0, nil
 }
 
+// Commit flushes the data stored in the internal memcache out to persistent
+// storage, returning th enumber of items written and any occurred error.
+func (s *TrieSync) Commit(dbw DatabaseWriter) (int, error) {
+	// Dump the memcache into a database dbw
+	for i, key := range s.memcache.order {
+		if err := dbw.Put(key[:], s.memcache.cache[key]); err != nil {
+			return i, err
+		}
+	}
+	written := len(s.memcache.order)
+
+	// Drop the memcache data and return
+	s.memcache = newSyncMemCache()
+	return written, nil
+}
+
 // Pending returns the number of state entries currently pending for download.
 func (s *TrieSync) Pending() int {
 	return len(s.requests)
@@ -253,13 +292,17 @@ func (s *TrieSync) children(req *request, object node) ([]*request, error) {
 		// If the child references another node, resolve or schedule
 		if node, ok := (child.node).(hashNode); ok {
 			// Try to resolve the node from the local database
+			hash := common.BytesToHash(node)
+			if _, ok := s.memcache.cache[hash]; ok {
+				continue
+			}
 			blob, _ := s.database.Get(node)
 			if local, err := decodeNode(node[:], blob, 0); local != nil && err == nil {
 				continue
 			}
 			// Locally unknown node, schedule for retrieval
 			requests = append(requests, &request{
-				hash:     common.BytesToHash(node),
+				hash:     hash,
 				parents:  []*request{req},
 				depth:    child.depth,
 				callback: req.callback,
@@ -269,21 +312,21 @@ func (s *TrieSync) children(req *request, object node) ([]*request, error) {
 	return requests, nil
 }
 
-// commit finalizes a retrieval request and stores it into the database. If any
+// commit finalizes a retrieval request and stores it into the memcache. If any
 // of the referencing parent requests complete due to this commit, they are also
 // committed themselves.
-func (s *TrieSync) commit(req *request, dbw DatabaseWriter) (err error) {
+func (s *TrieSync) commit(req *request) (err error) {
-	// Write the node content to disk
+	// Write the node content to the memcache
-	if err := dbw.Put(req.hash[:], req.data); err != nil {
+	s.memcache.cache[req.hash] = req.data
-		return err
+	s.memcache.order = append(s.memcache.order, req.hash)
-	}
+
 	delete(s.requests, req.hash)
 
 	// Check all parents for completion
 	for _, parent := range req.parents {
 		parent.deps--
 		if parent.deps == 0 {
-			if err := s.commit(parent, dbw); err != nil {
+			if err := s.commit(parent); err != nil {
 				return err
 			}
 		}

trie/sync_test.go

@@ -122,9 +122,12 @@ func testIterativeTrieSync(t *testing.T, batch int) {
 			}
 			results[i] = SyncResult{hash, data}
 		}
-		if _, index, err := sched.Process(results, dstDb); err != nil {
+		if _, index, err := sched.Process(results); err != nil {
 			t.Fatalf("failed to process result #%d: %v", index, err)
 		}
+		if index, err := sched.Commit(dstDb); err != nil {
+			t.Fatalf("failed to commit data #%d: %v", index, err)
+		}
 		queue = append(queue[:0], sched.Missing(batch)...)
 	}
 	// Cross check that the two tries are in sync
@@ -152,9 +155,12 @@ func TestIterativeDelayedTrieSync(t *testing.T) {
 			}
 			results[i] = SyncResult{hash, data}
 		}
-		if _, index, err := sched.Process(results, dstDb); err != nil {
+		if _, index, err := sched.Process(results); err != nil {
 			t.Fatalf("failed to process result #%d: %v", index, err)
 		}
+		if index, err := sched.Commit(dstDb); err != nil {
+			t.Fatalf("failed to commit data #%d: %v", index, err)
+		}
 		queue = append(queue[len(results):], sched.Missing(10000)...)
 	}
 	// Cross check that the two tries are in sync
@@ -190,9 +196,12 @@ func testIterativeRandomTrieSync(t *testing.T, batch int) {
 			results = append(results, SyncResult{hash, data})
 		}
 		// Feed the retrieved results back and queue new tasks
-		if _, index, err := sched.Process(results, dstDb); err != nil {
+		if _, index, err := sched.Process(results); err != nil {
 			t.Fatalf("failed to process result #%d: %v", index, err)
 		}
+		if index, err := sched.Commit(dstDb); err != nil {
+			t.Fatalf("failed to commit data #%d: %v", index, err)
+		}
 		queue = make(map[common.Hash]struct{})
 		for _, hash := range sched.Missing(batch) {
 			queue[hash] = struct{}{}
@@ -231,9 +240,12 @@ func TestIterativeRandomDelayedTrieSync(t *testing.T) {
 			}
 		}
 		// Feed the retrieved results back and queue new tasks
-		if _, index, err := sched.Process(results, dstDb); err != nil {
+		if _, index, err := sched.Process(results); err != nil {
 			t.Fatalf("failed to process result #%d: %v", index, err)
 		}
+		if index, err := sched.Commit(dstDb); err != nil {
+			t.Fatalf("failed to commit data #%d: %v", index, err)
+		}
 		for _, result := range results {
 			delete(queue, result.Hash)
 		}
@@ -272,9 +284,12 @@ func TestDuplicateAvoidanceTrieSync(t *testing.T) {
 
 			results[i] = SyncResult{hash, data}
 		}
-		if _, index, err := sched.Process(results, dstDb); err != nil {
+		if _, index, err := sched.Process(results); err != nil {
 			t.Fatalf("failed to process result #%d: %v", index, err)
 		}
+		if index, err := sched.Commit(dstDb); err != nil {
+			t.Fatalf("failed to commit data #%d: %v", index, err)
+		}
 		queue = append(queue[:0], sched.Missing(0)...)
 	}
 	// Cross check that the two tries are in sync
@@ -304,9 +319,12 @@ func TestIncompleteTrieSync(t *testing.T) {
 			results[i] = SyncResult{hash, data}
 		}
 		// Process each of the trie nodes
-		if _, index, err := sched.Process(results, dstDb); err != nil {
+		if _, index, err := sched.Process(results); err != nil {
 			t.Fatalf("failed to process result #%d: %v", index, err)
 		}
+		if index, err := sched.Commit(dstDb); err != nil {
+			t.Fatalf("failed to commit data #%d: %v", index, err)
+		}
 		for _, result := range results {
 			added = append(added, result.Hash)
 		}