From 061a6c972ce3c3f8ed06430f4ae0ec2771a9f9f5 Mon Sep 17 00:00:00 2001 From: rjl493456442 Date: Thu, 8 Aug 2019 16:23:00 +0800 Subject: [PATCH] core, eth/downloader, trie: get rid of trie node order When we flush a batch of trie nodes into database during the state sync, we should guarantee that all children should be flushed before parent. Actually the trie nodes commit order is strict by: children -> parent. But when we flush all ready nodes into db, actually we don't need order anymore since (1) they are all ready nodes(no more dependency) (2) underlying database(e.g.) can provide write atomicity --- core/state/sync_test.go | 20 ++++++++++---------- eth/downloader/statesync.go | 2 +- trie/sync.go | 17 ++++++----------- trie/sync_test.go | 24 ++++++++++++------------ 4 files changed, 29 insertions(+), 34 deletions(-) diff --git a/core/state/sync_test.go b/core/state/sync_test.go index de098dce0f..efd23f78b3 100644 --- a/core/state/sync_test.go +++ b/core/state/sync_test.go @@ -157,8 +157,8 @@ func testIterativeStateSync(t *testing.T, batch int) { 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) + if err := sched.Commit(dstDb); err != nil { + t.Fatalf("failed to commit data: %v", err) } queue = append(queue[:0], sched.Missing(batch)...) } @@ -190,8 +190,8 @@ func TestIterativeDelayedStateSync(t *testing.T) { 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) + if err := sched.Commit(dstDb); err != nil { + t.Fatalf("failed to commit data: %v", err) } queue = append(queue[len(results):], sched.Missing(0)...) } @@ -231,8 +231,8 @@ func testIterativeRandomStateSync(t *testing.T, batch int) { 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) + if err := sched.Commit(dstDb); err != nil { + t.Fatalf("failed to commit data: %v", err) } queue = make(map[common.Hash]struct{}) for _, hash := range sched.Missing(batch) { @@ -277,8 +277,8 @@ func TestIterativeRandomDelayedStateSync(t *testing.T) { 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) + if err := sched.Commit(dstDb); err != nil { + t.Fatalf("failed to commit data: %v", err) } for _, hash := range sched.Missing(0) { queue[hash] = struct{}{} @@ -316,8 +316,8 @@ func TestIncompleteStateSync(t *testing.T) { 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) + if err := sched.Commit(dstDb); err != nil { + t.Fatalf("failed to commit data: %v", err) } for _, result := range results { added = append(added, result.Hash) diff --git a/eth/downloader/statesync.go b/eth/downloader/statesync.go index b422557d58..f875b3a84c 100644 --- a/eth/downloader/statesync.go +++ b/eth/downloader/statesync.go @@ -347,7 +347,7 @@ func (s *stateSync) commit(force bool) error { } start := time.Now() b := s.d.stateDB.NewBatch() - if written, err := s.sched.Commit(b); written == 0 || err != nil { + if err := s.sched.Commit(b); err != nil { return err } if err := b.Write(); err != nil { diff --git a/trie/sync.go b/trie/sync.go index 6f40b45a1e..4db9603122 100644 --- a/trie/sync.go +++ b/trie/sync.go @@ -57,14 +57,12 @@ type SyncResult struct { // persisted data items. type syncMemBatch struct { batch map[common.Hash][]byte // In-memory membatch of recently completed items - order []common.Hash // Order of completion to prevent out-of-order data loss } // newSyncMemBatch allocates a new memory-buffer for not-yet persisted trie nodes. func newSyncMemBatch() *syncMemBatch { return &syncMemBatch{ batch: make(map[common.Hash][]byte), - order: make([]common.Hash, 0, 256), } } @@ -223,20 +221,18 @@ func (s *Sync) Process(results []SyncResult) (bool, int, error) { } // Commit flushes the data stored in the internal membatch out to persistent -// storage, returning the number of items written and any occurred error. -func (s *Sync) Commit(dbw ethdb.KeyValueWriter) (int, error) { +// storage, returning any occurred error. +func (s *Sync) Commit(dbw ethdb.KeyValueWriter) error { // Dump the membatch into a database dbw - for i, key := range s.membatch.order { - if err := dbw.Put(key[:], s.membatch.batch[key]); err != nil { - return i, err + for key, value := range s.membatch.batch { + if err := dbw.Put(key[:], value); err != nil { + return err } s.bloom.Add(key[:]) } - written := len(s.membatch.order) // TODO(karalabe): could an order change improve write performance? - // Drop the membatch data and return s.membatch = newSyncMemBatch() - return written, nil + return nil } // Pending returns the number of state entries currently pending for download. @@ -330,7 +326,6 @@ func (s *Sync) children(req *request, object node) ([]*request, error) { func (s *Sync) commit(req *request) (err error) { // Write the node content to the membatch s.membatch.batch[req.hash] = req.data - s.membatch.order = append(s.membatch.order, req.hash) delete(s.requests, req.hash) diff --git a/trie/sync_test.go b/trie/sync_test.go index 0621bb4357..10030d8bae 100644 --- a/trie/sync_test.go +++ b/trie/sync_test.go @@ -127,8 +127,8 @@ func testIterativeSync(t *testing.T, batch int) { if _, index, err := sched.Process(results); err != nil { t.Fatalf("failed to process result #%d: %v", index, err) } - if index, err := sched.Commit(diskdb); err != nil { - t.Fatalf("failed to commit data #%d: %v", index, err) + if err := sched.Commit(diskdb); err != nil { + t.Fatalf("failed to commit data: %v", err) } queue = append(queue[:0], sched.Missing(batch)...) } @@ -161,8 +161,8 @@ func TestIterativeDelayedSync(t *testing.T) { if _, index, err := sched.Process(results); err != nil { t.Fatalf("failed to process result #%d: %v", index, err) } - if index, err := sched.Commit(diskdb); err != nil { - t.Fatalf("failed to commit data #%d: %v", index, err) + if err := sched.Commit(diskdb); err != nil { + t.Fatalf("failed to commit data: %v", err) } queue = append(queue[len(results):], sched.Missing(10000)...) } @@ -203,8 +203,8 @@ func testIterativeRandomSync(t *testing.T, batch int) { if _, index, err := sched.Process(results); err != nil { t.Fatalf("failed to process result #%d: %v", index, err) } - if index, err := sched.Commit(diskdb); err != nil { - t.Fatalf("failed to commit data #%d: %v", index, err) + if err := sched.Commit(diskdb); err != nil { + t.Fatalf("failed to commit data: %v", err) } queue = make(map[common.Hash]struct{}) for _, hash := range sched.Missing(batch) { @@ -248,8 +248,8 @@ func TestIterativeRandomDelayedSync(t *testing.T) { if _, index, err := sched.Process(results); err != nil { t.Fatalf("failed to process result #%d: %v", index, err) } - if index, err := sched.Commit(diskdb); err != nil { - t.Fatalf("failed to commit data #%d: %v", index, err) + if err := sched.Commit(diskdb); err != nil { + t.Fatalf("failed to commit data: %v", err) } for _, result := range results { delete(queue, result.Hash) @@ -293,8 +293,8 @@ func TestDuplicateAvoidanceSync(t *testing.T) { if _, index, err := sched.Process(results); err != nil { t.Fatalf("failed to process result #%d: %v", index, err) } - if index, err := sched.Commit(diskdb); err != nil { - t.Fatalf("failed to commit data #%d: %v", index, err) + if err := sched.Commit(diskdb); err != nil { + t.Fatalf("failed to commit data: %v", err) } queue = append(queue[:0], sched.Missing(0)...) } @@ -329,8 +329,8 @@ func TestIncompleteSync(t *testing.T) { if _, index, err := sched.Process(results); err != nil { t.Fatalf("failed to process result #%d: %v", index, err) } - if index, err := sched.Commit(diskdb); err != nil { - t.Fatalf("failed to commit data #%d: %v", index, err) + if err := sched.Commit(diskdb); err != nil { + t.Fatalf("failed to commit data: %v", err) } for _, result := range results { added = append(added, result.Hash)