forks/go-ethereum
1
0
Fork 1
mirror of https://github.com/ethereum/go-ethereum.git synced 2026-08-20 10:52:25 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions

core, trie: enforce to use ethdb.Batch for node committing

Browse source
This commit is contained in:
rjl493456442 2019-08-09 08:34:28 +08:00
parent 061a6c972c
commit a387484d2e
3 changed files with 46 additions and 24 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

32
core/state/sync_test.go
View file

@ -136,7 +136,7 @@ func TestEmptyStateSync(t *testing.T) {
func TestIterativeStateSyncIndividual(t *testing.T) { testIterativeStateSync(t, 1) } func TestIterativeStateSyncIndividual(t *testing.T) { testIterativeStateSync(t, 1) }
func TestIterativeStateSyncBatched(t *testing.T) { testIterativeStateSync(t, 100) } func TestIterativeStateSyncBatched(t *testing.T) { testIterativeStateSync(t, 100) }
func testIterativeStateSync(t *testing.T, batch int) { func testIterativeStateSync(t *testing.T, count int) {
// Create a random state to copy // Create a random state to copy
srcDb, srcRoot, srcAccounts := makeTestState() srcDb, srcRoot, srcAccounts := makeTestState()
@ -144,7 +144,7 @@ func testIterativeStateSync(t *testing.T, batch int) {
dstDb := rawdb.NewMemoryDatabase() dstDb := rawdb.NewMemoryDatabase()
sched := NewStateSync(srcRoot, dstDb, trie.NewSyncBloom(1, dstDb)) sched := NewStateSync(srcRoot, dstDb, trie.NewSyncBloom(1, dstDb))
queue := append([]common.Hash{}, sched.Missing(batch)...) queue := append([]common.Hash{}, sched.Missing(count)...)
for len(queue) > 0 { for len(queue) > 0 {
results := make([]trie.SyncResult, len(queue)) results := make([]trie.SyncResult, len(queue))
for i, hash := range queue { for i, hash := range queue {
@ -157,10 +157,12 @@ func testIterativeStateSync(t *testing.T, batch int) {
if _, index, err := sched.Process(results); err != nil { if _, index, err := sched.Process(results); err != nil {
t.Fatalf("failed to process result #%d: %v", index, err) t.Fatalf("failed to process result #%d: %v", index, err)
} }
if err := sched.Commit(dstDb); err != nil { batch := dstDb.NewBatch()
if err := sched.Commit(batch); err != nil {
t.Fatalf("failed to commit data: %v", err) t.Fatalf("failed to commit data: %v", err)
} }
queue = append(queue[:0], sched.Missing(batch)...) batch.Write()
queue = append(queue[:0], sched.Missing(count)...)
} }
// Cross check that the two states are in sync // Cross check that the two states are in sync
checkStateAccounts(t, dstDb, srcRoot, srcAccounts) checkStateAccounts(t, dstDb, srcRoot, srcAccounts)
@ -190,9 +192,11 @@ func TestIterativeDelayedStateSync(t *testing.T) {
if _, index, err := sched.Process(results); err != nil { if _, index, err := sched.Process(results); err != nil {
t.Fatalf("failed to process result #%d: %v", index, err) t.Fatalf("failed to process result #%d: %v", index, err)
} }
if err := sched.Commit(dstDb); err != nil { batch := dstDb.NewBatch()
if err := sched.Commit(batch); err != nil {
t.Fatalf("failed to commit data: %v", err) t.Fatalf("failed to commit data: %v", err)
} }
batch.Write()
queue = append(queue[len(results):], sched.Missing(0)...) queue = append(queue[len(results):], sched.Missing(0)...)
} }
// Cross check that the two states are in sync // Cross check that the two states are in sync
@ -205,7 +209,7 @@ func TestIterativeDelayedStateSync(t *testing.T) {
func TestIterativeRandomStateSyncIndividual(t *testing.T) { testIterativeRandomStateSync(t, 1) } func TestIterativeRandomStateSyncIndividual(t *testing.T) { testIterativeRandomStateSync(t, 1) }
func TestIterativeRandomStateSyncBatched(t *testing.T) { testIterativeRandomStateSync(t, 100) } func TestIterativeRandomStateSyncBatched(t *testing.T) { testIterativeRandomStateSync(t, 100) }
func testIterativeRandomStateSync(t *testing.T, batch int) { func testIterativeRandomStateSync(t *testing.T, count int) {
// Create a random state to copy // Create a random state to copy
srcDb, srcRoot, srcAccounts := makeTestState() srcDb, srcRoot, srcAccounts := makeTestState()
@ -214,7 +218,7 @@ func testIterativeRandomStateSync(t *testing.T, batch int) {
sched := NewStateSync(srcRoot, dstDb, trie.NewSyncBloom(1, dstDb)) sched := NewStateSync(srcRoot, dstDb, trie.NewSyncBloom(1, dstDb))
queue := make(map[common.Hash]struct{}) queue := make(map[common.Hash]struct{})
for _, hash := range sched.Missing(batch) { for _, hash := range sched.Missing(count) {
queue[hash] = struct{}{} queue[hash] = struct{}{}
} }
for len(queue) > 0 { for len(queue) > 0 {
@ -231,11 +235,13 @@ func testIterativeRandomStateSync(t *testing.T, batch int) {
if _, index, err := sched.Process(results); err != nil { if _, index, err := sched.Process(results); err != nil {
t.Fatalf("failed to process result #%d: %v", index, err) t.Fatalf("failed to process result #%d: %v", index, err)
} }
if err := sched.Commit(dstDb); err != nil { batch := dstDb.NewBatch()
if err := sched.Commit(batch); err != nil {
t.Fatalf("failed to commit data: %v", err) t.Fatalf("failed to commit data: %v", err)
} }
batch.Write()
queue = make(map[common.Hash]struct{}) queue = make(map[common.Hash]struct{})
for _, hash := range sched.Missing(batch) { for _, hash := range sched.Missing(count) {
queue[hash] = struct{}{} queue[hash] = struct{}{}
} }
} }
@ -277,9 +283,11 @@ func TestIterativeRandomDelayedStateSync(t *testing.T) {
if _, index, err := sched.Process(results); err != nil { if _, index, err := sched.Process(results); err != nil {
t.Fatalf("failed to process result #%d: %v", index, err) t.Fatalf("failed to process result #%d: %v", index, err)
} }
if err := sched.Commit(dstDb); err != nil { batch := dstDb.NewBatch()
if err := sched.Commit(batch); err != nil {
t.Fatalf("failed to commit data: %v", err) t.Fatalf("failed to commit data: %v", err)
} }
batch.Write()
for _, hash := range sched.Missing(0) { for _, hash := range sched.Missing(0) {
queue[hash] = struct{}{} queue[hash] = struct{}{}
} }
@ -316,9 +324,11 @@ func TestIncompleteStateSync(t *testing.T) {
if _, index, err := sched.Process(results); err != nil { if _, index, err := sched.Process(results); err != nil {
t.Fatalf("failed to process result #%d: %v", index, err) t.Fatalf("failed to process result #%d: %v", index, err)
} }
if err := sched.Commit(dstDb); err != nil { batch := dstDb.NewBatch()
if err := sched.Commit(batch); err != nil {
t.Fatalf("failed to commit data: %v", err) t.Fatalf("failed to commit data: %v", err)
} }
batch.Write()
for _, result := range results { for _, result := range results {
added = append(added, result.Hash) added = append(added, result.Hash)
} }

2
trie/sync.go
View file

@ -222,7 +222,7 @@ func (s *Sync) Process(results []SyncResult) (bool, int, error) {
// Commit flushes the data stored in the internal membatch out to persistent // Commit flushes the data stored in the internal membatch out to persistent
// storage, returning any occurred error. // storage, returning any occurred error.
func (s *Sync) Commit(dbw ethdb.KeyValueWriter) error { func (s *Sync) Commit(dbw ethdb.Batch) error {
// Dump the membatch into a database dbw // Dump the membatch into a database dbw
for key, value := range s.membatch.batch { for key, value := range s.membatch.batch {
if err := dbw.Put(key[:], value); err != nil { if err := dbw.Put(key[:], value); err != nil {

36
trie/sync_test.go
View file

@ -105,7 +105,7 @@ func TestEmptySync(t *testing.T) {
func TestIterativeSyncIndividual(t *testing.T) { testIterativeSync(t, 1) } func TestIterativeSyncIndividual(t *testing.T) { testIterativeSync(t, 1) }
func TestIterativeSyncBatched(t *testing.T) { testIterativeSync(t, 100) } func TestIterativeSyncBatched(t *testing.T) { testIterativeSync(t, 100) }
func testIterativeSync(t *testing.T, batch int) { func testIterativeSync(t *testing.T, count int) {
// Create a random trie to copy // Create a random trie to copy
srcDb, srcTrie, srcData := makeTestTrie() srcDb, srcTrie, srcData := makeTestTrie()
@ -114,7 +114,7 @@ func testIterativeSync(t *testing.T, batch int) {
triedb := NewDatabase(diskdb) triedb := NewDatabase(diskdb)
sched := NewSync(srcTrie.Hash(), diskdb, nil, NewSyncBloom(1, diskdb)) sched := NewSync(srcTrie.Hash(), diskdb, nil, NewSyncBloom(1, diskdb))
queue := append([]common.Hash{}, sched.Missing(batch)...) queue := append([]common.Hash{}, sched.Missing(count)...)
for len(queue) > 0 { for len(queue) > 0 {
results := make([]SyncResult, len(queue)) results := make([]SyncResult, len(queue))
for i, hash := range queue { for i, hash := range queue {
@ -127,10 +127,12 @@ func testIterativeSync(t *testing.T, batch int) {
if _, index, err := sched.Process(results); err != nil { if _, index, err := sched.Process(results); err != nil {
t.Fatalf("failed to process result #%d: %v", index, err) t.Fatalf("failed to process result #%d: %v", index, err)
} }
if err := sched.Commit(diskdb); err != nil { batch := diskdb.NewBatch()
if err := sched.Commit(batch); err != nil {
t.Fatalf("failed to commit data: %v", err) t.Fatalf("failed to commit data: %v", err)
} }
queue = append(queue[:0], sched.Missing(batch)...) batch.Write()
queue = append(queue[:0], sched.Missing(count)...)
} }
// Cross check that the two tries are in sync // Cross check that the two tries are in sync
checkTrieContents(t, triedb, srcTrie.Hash().Bytes(), srcData) checkTrieContents(t, triedb, srcTrie.Hash().Bytes(), srcData)
@ -161,9 +163,11 @@ func TestIterativeDelayedSync(t *testing.T) {
if _, index, err := sched.Process(results); err != nil { if _, index, err := sched.Process(results); err != nil {
t.Fatalf("failed to process result #%d: %v", index, err) t.Fatalf("failed to process result #%d: %v", index, err)
} }
if err := sched.Commit(diskdb); err != nil { batch := diskdb.NewBatch()
if err := sched.Commit(batch); err != nil {
t.Fatalf("failed to commit data: %v", err) t.Fatalf("failed to commit data: %v", err)
} }
batch.Write()
queue = append(queue[len(results):], sched.Missing(10000)...) queue = append(queue[len(results):], sched.Missing(10000)...)
} }
// Cross check that the two tries are in sync // Cross check that the two tries are in sync
@ -176,7 +180,7 @@ func TestIterativeDelayedSync(t *testing.T) {
func TestIterativeRandomSyncIndividual(t *testing.T) { testIterativeRandomSync(t, 1) } func TestIterativeRandomSyncIndividual(t *testing.T) { testIterativeRandomSync(t, 1) }
func TestIterativeRandomSyncBatched(t *testing.T) { testIterativeRandomSync(t, 100) } func TestIterativeRandomSyncBatched(t *testing.T) { testIterativeRandomSync(t, 100) }
func testIterativeRandomSync(t *testing.T, batch int) { func testIterativeRandomSync(t *testing.T, count int) {
// Create a random trie to copy // Create a random trie to copy
srcDb, srcTrie, srcData := makeTestTrie() srcDb, srcTrie, srcData := makeTestTrie()
@ -186,7 +190,7 @@ func testIterativeRandomSync(t *testing.T, batch int) {
sched := NewSync(srcTrie.Hash(), diskdb, nil, NewSyncBloom(1, diskdb)) sched := NewSync(srcTrie.Hash(), diskdb, nil, NewSyncBloom(1, diskdb))
queue := make(map[common.Hash]struct{}) queue := make(map[common.Hash]struct{})
for _, hash := range sched.Missing(batch) { for _, hash := range sched.Missing(count) {
queue[hash] = struct{}{} queue[hash] = struct{}{}
} }
for len(queue) > 0 { for len(queue) > 0 {
@ -203,11 +207,13 @@ func testIterativeRandomSync(t *testing.T, batch int) {
if _, index, err := sched.Process(results); err != nil { if _, index, err := sched.Process(results); err != nil {
t.Fatalf("failed to process result #%d: %v", index, err) t.Fatalf("failed to process result #%d: %v", index, err)
} }
if err := sched.Commit(diskdb); err != nil { batch := diskdb.NewBatch()
if err := sched.Commit(batch); err != nil {
t.Fatalf("failed to commit data: %v", err) t.Fatalf("failed to commit data: %v", err)
} }
batch.Write()
queue = make(map[common.Hash]struct{}) queue = make(map[common.Hash]struct{})
for _, hash := range sched.Missing(batch) { for _, hash := range sched.Missing(count) {
queue[hash] = struct{}{} queue[hash] = struct{}{}
} }
} }
@ -248,9 +254,11 @@ func TestIterativeRandomDelayedSync(t *testing.T) {
if _, index, err := sched.Process(results); err != nil { if _, index, err := sched.Process(results); err != nil {
t.Fatalf("failed to process result #%d: %v", index, err) t.Fatalf("failed to process result #%d: %v", index, err)
} }
if err := sched.Commit(diskdb); err != nil { batch := diskdb.NewBatch()
if err := sched.Commit(batch); err != nil {
t.Fatalf("failed to commit data: %v", err) t.Fatalf("failed to commit data: %v", err)
} }
batch.Write()
for _, result := range results { for _, result := range results {
delete(queue, result.Hash) delete(queue, result.Hash)
} }
@ -293,9 +301,11 @@ func TestDuplicateAvoidanceSync(t *testing.T) {
if _, index, err := sched.Process(results); err != nil { if _, index, err := sched.Process(results); err != nil {
t.Fatalf("failed to process result #%d: %v", index, err) t.Fatalf("failed to process result #%d: %v", index, err)
} }
if err := sched.Commit(diskdb); err != nil { batch := diskdb.NewBatch()
if err := sched.Commit(batch); err != nil {
t.Fatalf("failed to commit data: %v", err) t.Fatalf("failed to commit data: %v", err)
} }
batch.Write()
queue = append(queue[:0], sched.Missing(0)...) queue = append(queue[:0], sched.Missing(0)...)
} }
// Cross check that the two tries are in sync // Cross check that the two tries are in sync
@ -329,9 +339,11 @@ func TestIncompleteSync(t *testing.T) {
if _, index, err := sched.Process(results); err != nil { if _, index, err := sched.Process(results); err != nil {
t.Fatalf("failed to process result #%d: %v", index, err) t.Fatalf("failed to process result #%d: %v", index, err)
} }
if err := sched.Commit(diskdb); err != nil { batch := diskdb.NewBatch()
if err := sched.Commit(batch); err != nil {
t.Fatalf("failed to commit data: %v", err) t.Fatalf("failed to commit data: %v", err)
} }
batch.Write()
for _, result := range results { for _, result := range results {
added = append(added, result.Hash) added = append(added, result.Hash)
} }