From 3befea7801a08af023979787da052c96b5b58a7a Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?P=C3=A9ter=20Szil=C3=A1gyi?= Date: Fri, 13 Oct 2017 10:32:31 +0300 Subject: [PATCH] eth, trie: implement fake-peer leaf-sync --- cmd/geth/chaincmd.go | 2 +- core/state/sync_test.go | 90 +++++++++++------ eth/api_test.go | 2 +- eth/downloader/downloader.go | 40 +++++--- eth/downloader/downloader_test.go | 20 ++++ eth/downloader/fakepeer.go | 57 +++++++++++ eth/downloader/metrics.go | 2 + eth/downloader/peer.go | 77 ++++++++++++-- eth/downloader/statesync.go | 150 ++++++++++++++++++++-------- eth/downloader/types.go | 12 +++ eth/handler_test.go | 2 +- eth/helper_test.go | 2 +- eth/peer.go | 8 ++ trie/sync.go | 160 ++++++++++++++++++------------ trie/sync_test.go | 25 ++--- trie/trie.go | 31 ++++-- 16 files changed, 502 insertions(+), 178 deletions(-) diff --git a/cmd/geth/chaincmd.go b/cmd/geth/chaincmd.go index 4a9a7b11b5..5af7427957 100644 --- a/cmd/geth/chaincmd.go +++ b/cmd/geth/chaincmd.go @@ -309,7 +309,7 @@ func copyDb(ctx *cli.Context) error { return err } peer := downloader.NewFakePeer("local", db, hc, dl) - if err = dl.RegisterPeer("local", 63, peer); err != nil { + if err = dl.RegisterPeer("local", 64, peer); err != nil { return err } // Synchronise with the simulated peer diff --git a/core/state/sync_test.go b/core/state/sync_test.go index 3be8384a08..973645ac49 100644 --- a/core/state/sync_test.go +++ b/core/state/sync_test.go @@ -19,6 +19,7 @@ package state import ( "bytes" "math/big" + "math/rand" "testing" "github.com/ethereum/go-ethereum/common" @@ -146,16 +147,23 @@ func testIterativeStateSync(t *testing.T, batch int) { queue := append([]common.Hash{}, sched.Missing(batch)...) for len(queue) > 0 { - results := make([]trie.SyncResult, len(queue)) + results := make([]*trie.SyncResult, len(queue)) for i, hash := range queue { - data, err := srcMem.Get(hash.Bytes()) - if err != nil { - t.Fatalf("failed to retrieve node data for %x: %v", hash, err) + if rand.Int()%2 == 0 { + data, err := srcMem.Get(hash.Bytes()) + if err != nil { + t.Fatalf("failed to retrieve node data for %x: %v", hash, err) + } + results[i] = &trie.SyncResult{Data: data} + } else { + trie, _ := trie.New(common.Hash{}, srcMem) + results[i], _, _ = trie.FetchData(hash, 8192) } - results[i] = trie.SyncResult{Hash: hash, Data: data} } - if _, index, err := sched.Process(results); err != nil { - t.Fatalf("failed to process result #%d: %v", index, err) + for index, result := range results { + if _, _, _, err := sched.Process(result); 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) @@ -179,16 +187,23 @@ func TestIterativeDelayedStateSync(t *testing.T) { queue := append([]common.Hash{}, sched.Missing(0)...) for len(queue) > 0 { // Sync only half of the scheduled nodes - results := make([]trie.SyncResult, len(queue)/2+1) + results := make([]*trie.SyncResult, len(queue)/2+1) for i, hash := range queue[:len(results)] { - data, err := srcMem.Get(hash.Bytes()) - if err != nil { - t.Fatalf("failed to retrieve node data for %x: %v", hash, err) + if rand.Int()%2 == 0 { + data, err := srcMem.Get(hash.Bytes()) + if err != nil { + t.Fatalf("failed to retrieve node data for %x: %v", hash, err) + } + results[i] = &trie.SyncResult{Data: data} + } else { + trie, _ := trie.New(common.Hash{}, srcMem) + results[i], _, _ = trie.FetchData(hash, 8192) } - results[i] = trie.SyncResult{Hash: hash, Data: data} } - if _, index, err := sched.Process(results); err != nil { - t.Fatalf("failed to process result #%d: %v", index, err) + for index, result := range results { + if _, _, _, err := sched.Process(result); 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) @@ -219,17 +234,25 @@ func testIterativeRandomStateSync(t *testing.T, batch int) { } for len(queue) > 0 { // Fetch all the queued nodes in a random order - results := make([]trie.SyncResult, 0, len(queue)) + results := make([]*trie.SyncResult, 0, len(queue)) for hash := range queue { - data, err := srcMem.Get(hash.Bytes()) - if err != nil { - t.Fatalf("failed to retrieve node data for %x: %v", hash, err) + if rand.Int()%2 == 0 { + data, err := srcMem.Get(hash.Bytes()) + if err != nil { + t.Fatalf("failed to retrieve node data for %x: %v", hash, err) + } + results = append(results, &trie.SyncResult{Data: data}) + } else { + trie, _ := trie.New(common.Hash{}, srcMem) + request, _, _ := trie.FetchData(hash, 8192) + results = append(results, request) } - results = append(results, trie.SyncResult{Hash: hash, Data: data}) } // Feed the retrieved results back and queue new tasks - if _, index, err := sched.Process(results); err != nil { - t.Fatalf("failed to process result #%d: %v", index, err) + for index, result := range results { + if _, _, _, err := sched.Process(result); 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) @@ -259,7 +282,7 @@ func TestIterativeRandomDelayedStateSync(t *testing.T) { } for len(queue) > 0 { // Sync only half of the scheduled nodes, even those in random order - results := make([]trie.SyncResult, 0, len(queue)/2+1) + results := make([]*trie.SyncResult, 0, len(queue)/2+1) for hash := range queue { delete(queue, hash) @@ -267,15 +290,17 @@ func TestIterativeRandomDelayedStateSync(t *testing.T) { if err != nil { t.Fatalf("failed to retrieve node data for %x: %v", hash, err) } - results = append(results, trie.SyncResult{Hash: hash, Data: data}) + results = append(results, &trie.SyncResult{Data: data}) if len(results) >= cap(results) { break } } // Feed the retrieved results back and queue new tasks - if _, index, err := sched.Process(results); err != nil { - t.Fatalf("failed to process result #%d: %v", index, err) + for index, result := range results { + if _, _, _, err := sched.Process(result); 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) @@ -304,24 +329,25 @@ func TestIncompleteStateSync(t *testing.T) { queue := append([]common.Hash{}, sched.Missing(1)...) for len(queue) > 0 { // Fetch a batch of state nodes - results := make([]trie.SyncResult, len(queue)) + results := make([]*trie.SyncResult, len(queue)) for i, hash := range queue { data, err := srcMem.Get(hash.Bytes()) if err != nil { t.Fatalf("failed to retrieve node data for %x: %v", hash, err) } - results[i] = trie.SyncResult{Hash: hash, Data: data} + results[i] = &trie.SyncResult{Data: data} } // Process each of the state nodes - if _, index, err := sched.Process(results); err != nil { - t.Fatalf("failed to process result #%d: %v", index, err) + for index, result := range results { + _, _, hash, err := sched.Process(result) + if err != nil { + t.Fatalf("failed to process result #%d: %v", index, err) + } + added = append(added, hash) } 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) - } // Check that all known sub-tries added so far are complete or missing entirely. checkSubtries: for _, hash := range added { diff --git a/eth/api_test.go b/eth/api_test.go index 49ce386886..515e24a440 100644 --- a/eth/api_test.go +++ b/eth/api_test.go @@ -31,7 +31,7 @@ var dumper = spew.ConfigState{Indent: " "} func TestStorageRangeAt(t *testing.T) { // Create a state where account 0x010000... has a few storage entries. var ( - db, _ = ethdb.NewMemDatabase() + db = ethdb.NewMemDatabase() state, _ = state.New(common.Hash{}, state.NewDatabase(db)) addr = common.Address{0x01} keys = []common.Hash{ // hashes of Keys of storage diff --git a/eth/downloader/downloader.go b/eth/downloader/downloader.go index 5782d4cf53..c7fae81b4c 100644 --- a/eth/downloader/downloader.go +++ b/eth/downloader/downloader.go @@ -34,17 +34,20 @@ import ( "github.com/ethereum/go-ethereum/event" "github.com/ethereum/go-ethereum/log" "github.com/ethereum/go-ethereum/params" + "github.com/ethereum/go-ethereum/trie" "github.com/rcrowley/go-metrics" ) var ( - MaxHashFetch = 512 // Amount of hashes to be fetched per retrieval request - MaxBlockFetch = 128 // Amount of blocks to be fetched per retrieval request - MaxHeaderFetch = 192 // Amount of block headers to be fetched per retrieval request - MaxSkeletonSize = 128 // Number of header fetches to need for a skeleton assembly - MaxBodyFetch = 128 // Amount of block bodies to be fetched per retrieval request - MaxReceiptFetch = 256 // Amount of transaction receipts to allow fetching per request - MaxStateFetch = 384 // Amount of node state values to allow fetching per request + MaxHashFetch = 512 // Amount of hashes to be fetched per retrieval request + MaxBlockFetch = 128 // Amount of blocks to be fetched per retrieval request + MaxHeaderFetch = 192 // Amount of block headers to be fetched per retrieval request + MaxSkeletonSize = 128 // Number of header fetches to need for a skeleton assembly + MaxBodyFetch = 128 // Amount of block bodies to be fetched per retrieval request + MaxReceiptFetch = 256 // Amount of transaction receipts to allow fetching per request + MaxStateFetch = 384 // Amount of node state values to allow fetching per request + MinTrieFetch = 100 * 1024 // Amount of state leaf data to fetch at minimum per request + MaxTrieFetch = 2048 * 1024 // Amount of state leaf data to allow fetching per request MaxForkAncestry = 3 * params.EpochDuration // Maximum chain reorganisation rttMinEstimate = 2 * time.Second // Minimum round-trip time to target for download requests @@ -57,7 +60,7 @@ var ( qosConfidenceCap = 10 // Number of peers above which not to modify RTT confidence qosTuningImpact = 0.25 // Impact that a new tuning target has on the previous value - maxQueuedHeaders = 32 * 1024 // [eth/62] Maximum number of headers to queue for import (DOS protection) + maxQueuedHeaders = 32 * 1024 // Maximum number of headers to queue for import (DOS protection) maxHeadersProcess = 2048 // Number of header download results to import at once into the chain maxResultsProcess = 2048 // Number of content download results to import at once into the chain @@ -137,6 +140,7 @@ type Downloader struct { stateSyncStart chan *stateSync trackStateReq chan *stateReq stateCh chan dataPack // [eth/63] Channel receiving inbound node state data + trieCh chan dataPack // [eth/64] Channel receiving inbound state trie data // Cancellation and termination cancelPeer string // Identifier of the peer currently being used as the master (cancel on drop) @@ -225,6 +229,7 @@ func New(mode SyncMode, stateDb ethdb.Database, mux *event.TypeMux, chain BlockC headerProcCh: make(chan []*types.Header, 1), quitCh: make(chan struct{}), stateCh: make(chan dataPack), + trieCh: make(chan dataPack), stateSyncStart: make(chan *stateSync), trackStateReq: make(chan *stateReq), } @@ -471,9 +476,9 @@ func (d *Downloader) syncWithPeer(p *peerConnection, hash common.Hash, td *big.I } fetchers := []func() error{ - func() error { return d.fetchHeaders(p, origin+1) }, // Headers are always retrieved - func() error { return d.fetchBodies(origin + 1) }, // Bodies are retrieved during normal and fast sync - func() error { return d.fetchReceipts(origin + 1) }, // Receipts are retrieved during fast sync + //func() error { return d.fetchHeaders(p, origin+1) }, // Headers are always retrieved + //func() error { return d.fetchBodies(origin + 1) }, // Bodies are retrieved during normal and fast sync + //func() error { return d.fetchReceipts(origin + 1) }, // Receipts are retrieved during fast sync func() error { return d.processHeaders(origin+1, td) }, } if d.mode == FastSync { @@ -1470,25 +1475,30 @@ func (d *Downloader) commitPivotBlock(result *fetchResult) error { // DeliverHeaders injects a new batch of block headers received from a remote // node into the download schedule. -func (d *Downloader) DeliverHeaders(id string, headers []*types.Header) (err error) { +func (d *Downloader) DeliverHeaders(id string, headers []*types.Header) error { return d.deliver(id, d.headerCh, &headerPack{id, headers}, headerInMeter, headerDropMeter) } // DeliverBodies injects a new batch of block bodies received from a remote node. -func (d *Downloader) DeliverBodies(id string, transactions [][]*types.Transaction, uncles [][]*types.Header) (err error) { +func (d *Downloader) DeliverBodies(id string, transactions [][]*types.Transaction, uncles [][]*types.Header) error { return d.deliver(id, d.bodyCh, &bodyPack{id, transactions, uncles}, bodyInMeter, bodyDropMeter) } // DeliverReceipts injects a new batch of receipts received from a remote node. -func (d *Downloader) DeliverReceipts(id string, receipts [][]*types.Receipt) (err error) { +func (d *Downloader) DeliverReceipts(id string, receipts [][]*types.Receipt) error { return d.deliver(id, d.receiptCh, &receiptPack{id, receipts}, receiptInMeter, receiptDropMeter) } // DeliverNodeData injects a new batch of node state data received from a remote node. -func (d *Downloader) DeliverNodeData(id string, data [][]byte) (err error) { +func (d *Downloader) DeliverNodeData(id string, data [][]byte) error { return d.deliver(id, d.stateCh, &statePack{id, data}, stateInMeter, stateDropMeter) } +// DeliverTrie injects a (possibly partial) sub-trie received from a remote node. +func (d *Downloader) DeliverTries(id string, res []*trie.SyncResult) error { + return d.deliver(id, d.trieCh, &triePack{id, res}, trieInMeter, trieDropMeter) +} + // deliver injects a new batch of data received from a remote node. func (d *Downloader) deliver(id string, destCh chan dataPack, packet dataPack, inMeter, dropMeter metrics.Meter) (err error) { // Update the delivery metrics for both good and failed deliveries diff --git a/eth/downloader/downloader_test.go b/eth/downloader/downloader_test.go index a656e1a1eb..7627bcccba 100644 --- a/eth/downloader/downloader_test.go +++ b/eth/downloader/downloader_test.go @@ -608,6 +608,23 @@ func (dlp *downloadTesterPeer) RequestNodeData(hashes []common.Hash) error { return nil } +// RequestTries constructs a getNodeData method associated with a particular +// peer in the download tester. The returned function can be used to retrieve +// batches of node state data from the particularly requested peer. +func (dlp *downloadTesterPeer) RequestTries(roots []common.Hash, limit common.StorageSize) error { + dlp.waitDelay() + + dlp.dl.lock.RLock() + defer dlp.dl.lock.RUnlock() + + if data, err := dlp.dl.peerDb.Get(roots[0].Bytes()); err == nil { + if !dlp.dl.peerMissingStates[dlp.id][roots[0]] { + go dlp.dl.downloader.DeliverTries(dlp.id, []*trie.SyncResult{{Data: data}}) + } + } + return nil +} + // assertOwnChain checks if the local chain contains the correct number of items // of the various chain components. func assertOwnChain(t *testing.T, tester *downloadTester, length int) { @@ -1690,6 +1707,9 @@ func (ftp *floodingTestPeer) RequestReceipts(hashes []common.Hash) error { func (ftp *floodingTestPeer) RequestNodeData(hashes []common.Hash) error { return ftp.peer.RequestNodeData(hashes) } +func (ftp *floodingTestPeer) RequestTries(roots []common.Hash, limit common.StorageSize) error { + return ftp.peer.RequestTries(roots, limit) +} func (ftp *floodingTestPeer) RequestHeadersByNumber(from uint64, count, skip int, reverse bool) error { deliveriesDone := make(chan struct{}, 500) diff --git a/eth/downloader/fakepeer.go b/eth/downloader/fakepeer.go index ebdb9c3342..df4645caf8 100644 --- a/eth/downloader/fakepeer.go +++ b/eth/downloader/fakepeer.go @@ -17,12 +17,18 @@ package downloader import ( + "fmt" "math/big" + "sync/atomic" + "time" "github.com/ethereum/go-ethereum/common" "github.com/ethereum/go-ethereum/core" "github.com/ethereum/go-ethereum/core/types" "github.com/ethereum/go-ethereum/ethdb" + "github.com/ethereum/go-ethereum/rlp" + "github.com/ethereum/go-ethereum/trie" + "github.com/golang/snappy" ) // FakePeer is a mock downloader peer that operates on a local database instance @@ -146,6 +152,18 @@ func (p *FakePeer) RequestReceipts(hashes []common.Hash) error { return nil } +var cnt int32 +var siz int32 + +func init() { + go func() { + for { + time.Sleep(time.Second) + //fmt.Println("Packets:", atomic.LoadInt32(&cnt), "Bytes (snappy):", atomic.LoadInt32(&siz)) + } + }() +} + // RequestNodeData implements downloader.Peer, returning a batch of state trie // nodes corresponding to the specified trie hashes. func (p *FakePeer) RequestNodeData(hashes []common.Hash) error { @@ -155,6 +173,45 @@ func (p *FakePeer) RequestNodeData(hashes []common.Hash) error { data = append(data, entry) } } + + atomic.AddInt32(&cnt, 1) + blob, _ := rlp.EncodeToBytes(data) + atomic.AddInt32(&siz, int32(len(snappy.Encode(nil, blob)))) + p.dl.DeliverNodeData(p.id, data) return nil } + +// RequestTries implements downloader.Peer, returning the (potentially partial) +// leaves of a state sub-trie. +func (p *FakePeer) RequestTries(roots []common.Hash, limit common.StorageSize) error { + start := time.Now() + + t, err := trie.New(common.Hash{}, p.db) + if err != nil { + return err + } + results := make([]*trie.SyncResult, 0, len(roots)) + for _, root := range roots { + res, size, err := t.FetchData(root, limit, time.Second-time.Since(start)) + if err != nil { + return err + } + results = append(results, res) + + if size >= limit || time.Since(start) > time.Second { + break + } + limit -= size + } + + atomic.AddInt32(&cnt, 1) + blob, _ := rlp.EncodeToBytes(results) + size := len(snappy.Encode(nil, blob)) + atomic.AddInt32(&siz, int32(size)) + + fmt.Printf("Packet #%d: %d bytes (%d total) in %v\n", atomic.LoadInt32(&cnt), size, atomic.LoadInt32(&siz), time.Since(start)) + + p.dl.DeliverTries(p.id, results) + return nil +} diff --git a/eth/downloader/metrics.go b/eth/downloader/metrics.go index 58764ccf06..99f80aedad 100644 --- a/eth/downloader/metrics.go +++ b/eth/downloader/metrics.go @@ -40,4 +40,6 @@ var ( stateInMeter = metrics.NewMeter("eth/downloader/states/in") stateDropMeter = metrics.NewMeter("eth/downloader/states/drop") + trieInMeter = metrics.NewMeter("eth/downloader/tries/in") + trieDropMeter = metrics.NewMeter("eth/downloader/tries/drop") ) diff --git a/eth/downloader/peer.go b/eth/downloader/peer.go index e638744ea5..650e2be9db 100644 --- a/eth/downloader/peer.go +++ b/eth/downloader/peer.go @@ -53,11 +53,13 @@ type peerConnection struct { blockIdle int32 // Current block activity state of the peer (idle = 0, active = 1) receiptIdle int32 // Current receipt activity state of the peer (idle = 0, active = 1) stateIdle int32 // Current node data activity state of the peer (idle = 0, active = 1) + trieIdle int32 // Current trie activity state of the peer (idle = 0, active = 1) headerThroughput float64 // Number of headers measured to be retrievable per second blockThroughput float64 // Number of blocks (bodies) measured to be retrievable per second receiptThroughput float64 // Number of receipts measured to be retrievable per second stateThroughput float64 // Number of node data pieces measured to be retrievable per second + trieThroughput float64 // Amount of trie leaves measured to be retrievable per second rtt time.Duration // Request round trip time to track responsiveness (QoS) @@ -65,6 +67,7 @@ type peerConnection struct { blockStarted time.Time // Time instance when the last block (body) fetch was started receiptStarted time.Time // Time instance when the last receipt fetch was started stateStarted time.Time // Time instance when the last node data fetch was started + trieStarted time.Time // Time instance when the last node data fetch was started lacking map[common.Hash]struct{} // Set of hashes not to request (didn't have previously) @@ -88,6 +91,7 @@ type Peer interface { RequestBodies([]common.Hash) error RequestReceipts([]common.Hash) error RequestNodeData([]common.Hash) error + RequestTries([]common.Hash, common.StorageSize) error } // lightPeerWrapper wraps a LightPeer struct, stubbing out the Peer-only methods. @@ -111,6 +115,9 @@ func (w *lightPeerWrapper) RequestReceipts([]common.Hash) error { func (w *lightPeerWrapper) RequestNodeData([]common.Hash) error { panic("RequestNodeData not supported in light client mode sync") } +func (w *lightPeerWrapper) RequestTries([]common.Hash, common.StorageSize) error { + panic("RequestTries not supported in light client mode sync") +} // newPeerConnection creates a new downloader peer. func newPeerConnection(id string, version int, peer Peer, logger log.Logger) *peerConnection { @@ -134,11 +141,13 @@ func (p *peerConnection) Reset() { atomic.StoreInt32(&p.blockIdle, 0) atomic.StoreInt32(&p.receiptIdle, 0) atomic.StoreInt32(&p.stateIdle, 0) + atomic.StoreInt32(&p.trieIdle, 0) p.headerThroughput = 0 p.blockThroughput = 0 p.receiptThroughput = 0 p.stateThroughput = 0 + p.trieThroughput = 0 p.lacking = make(map[common.Hash]struct{}) } @@ -222,6 +231,23 @@ func (p *peerConnection) FetchNodeData(hashes []common.Hash) error { return nil } +// FetchTries sends a trie retrieval request to the remote peer. +func (p *peerConnection) FetchTries(roots []common.Hash, limit common.StorageSize) error { + // Sanity check the protocol version + if p.version < 64 { + panic(fmt.Sprintf("trie fetch [eth/64+] requested on eth/%d", p.version)) + } + // Short circuit if the peer is already fetching + if !atomic.CompareAndSwapInt32(&p.trieIdle, 0, 1) { + return errAlreadyFetching + } + p.trieStarted = time.Now() + + go p.peer.RequestTries(roots, limit) + + return nil +} + // SetHeadersIdle sets the peer to idle, allowing it to execute new header retrieval // requests. Its estimated header retrieval throughput is updated with that measured // just now. @@ -257,6 +283,13 @@ func (p *peerConnection) SetNodeDataIdle(delivered int) { p.setIdle(p.stateStarted, delivered, &p.stateThroughput, &p.stateIdle) } +// SetTrieIdle sets the peer to idle, allowing it to execute new trie retrieval +// requests. Its estimated state retrieval throughput is updated with that +// measured just now. +func (p *peerConnection) SetTrieIdle(delivered int) { + p.setIdle(p.trieStarted, delivered, &p.trieThroughput, &p.trieIdle) +} + // setIdle sets the peer to idle, allowing it to execute new retrieval requests. // Its estimated retrieval throughput is updated with that measured just now. func (p *peerConnection) setIdle(started time.Time, delivered int, throughput *float64, idle *int32) { @@ -278,10 +311,17 @@ func (p *peerConnection) setIdle(started time.Time, delivered int, throughput *f *throughput = (1-measurementImpact)*(*throughput) + measurementImpact*measured p.rtt = time.Duration((1-measurementImpact)*float64(p.rtt) + measurementImpact*float64(elapsed)) - p.log.Trace("Peer throughput measurements updated", - "hps", p.headerThroughput, "bps", p.blockThroughput, - "rps", p.receiptThroughput, "sps", p.stateThroughput, - "miss", len(p.lacking), "rtt", p.rtt) + if p.version <= 63 { + p.log.Trace("Peer throughput measurements updated", + "hps", p.headerThroughput, "bps", p.blockThroughput, + "rps", p.receiptThroughput, "sps", p.stateThroughput, + "miss", len(p.lacking), "rtt", p.rtt) + } else { + p.log.Trace("Peer throughput measurements updated", + "hps", p.headerThroughput, "bps", p.blockThroughput, + "rps", p.receiptThroughput, "tps", p.trieThroughput, + "miss", len(p.lacking), "rtt", p.rtt) + } } // HeaderCapacity retrieves the peers header download allowance based on its @@ -320,6 +360,15 @@ func (p *peerConnection) NodeDataCapacity(targetRTT time.Duration) int { return int(math.Min(1+math.Max(1, p.stateThroughput*float64(targetRTT)/float64(time.Second)), float64(MaxStateFetch))) } +// TrieCapacity retrieves the peers trie download allowance based on its +// previously discovered throughput. +func (p *peerConnection) TrieCapacity(targetRTT time.Duration) int { + p.lock.RLock() + defer p.lock.RUnlock() + + return int(math.Min(1+math.Max(1, p.trieThroughput*float64(targetRTT)/float64(time.Second)), float64(MaxTrieFetch))) +} + // MarkLacking appends a new entity to the set of items (blocks, receipts, states) // that a peer is known not to have (i.e. have been requested before). If the // set reaches its maximum allowed capacity, items are randomly dropped off. @@ -400,7 +449,7 @@ func (ps *peerSet) Register(p *peerConnection) error { return errAlreadyRegistered } if len(ps.peers) > 0 { - p.headerThroughput, p.blockThroughput, p.receiptThroughput, p.stateThroughput = 0, 0, 0, 0 + p.headerThroughput, p.blockThroughput, p.receiptThroughput, p.stateThroughput, p.trieThroughput = 0, 0, 0, 0, 0 for _, peer := range ps.peers { peer.lock.RLock() @@ -408,12 +457,14 @@ func (ps *peerSet) Register(p *peerConnection) error { p.blockThroughput += peer.blockThroughput p.receiptThroughput += peer.receiptThroughput p.stateThroughput += peer.stateThroughput + p.trieThroughput += peer.trieThroughput peer.lock.RUnlock() } p.headerThroughput /= float64(len(ps.peers)) p.blockThroughput /= float64(len(ps.peers)) p.receiptThroughput /= float64(len(ps.peers)) p.stateThroughput /= float64(len(ps.peers)) + p.trieThroughput /= float64(len(ps.peers)) } ps.peers[p.id] = p ps.lock.Unlock() @@ -519,7 +570,21 @@ func (ps *peerSet) NodeDataIdlePeers() ([]*peerConnection, int) { defer p.lock.RUnlock() return p.stateThroughput } - return ps.idlePeers(63, 64, idle, throughput) + return ps.idlePeers(63, 63, idle, throughput) +} + +// TrieIdlePeers retrieves a flat list of all the currently trie-idle +// peers within the active peer set, ordered by their reputation. +func (ps *peerSet) TrieIdlePeers() ([]*peerConnection, int) { + idle := func(p *peerConnection) bool { + return atomic.LoadInt32(&p.trieIdle) == 0 + } + throughput := func(p *peerConnection) float64 { + p.lock.RLock() + defer p.lock.RUnlock() + return p.trieThroughput + } + return ps.idlePeers(64, 64, idle, throughput) } // idlePeers retrieves a flat list of all currently idle peers satisfying the diff --git a/eth/downloader/statesync.go b/eth/downloader/statesync.go index e0f7ed0cd7..ce40de3e00 100644 --- a/eth/downloader/statesync.go +++ b/eth/downloader/statesync.go @@ -32,18 +32,23 @@ import ( // stateReq represents a batch of state fetch requests groupped together into // a single data retrieval network packet. type stateReq struct { - items []common.Hash // Hashes of the state items to download - tasks map[common.Hash]*stateTask // Download tasks to track previous attempts - timeout time.Duration // Maximum round trip time for this to complete - timer *time.Timer // Timer to fire when the RTT timeout expires - peer *peerConnection // Peer that we're requesting from - response [][]byte // Response data of the peer (nil for timeouts) - dropped bool // Flag whether the peer dropped off early + items []common.Hash // Hashes of the state items to download (single one in leaf-mode) + limit common.StorageSize // Amount of trie-leaf data to download in leaf-mode + + tasks map[common.Hash]*stateTask // Download tasks to track previous attempts + timeout time.Duration // Maximum round trip time for this to complete + timer *time.Timer // Timer to fire when the RTT timeout expires + peer *peerConnection // Peer that we're requesting from + + nodeRes [][]byte // Response data of the node-mode peer (nil for timeouts) + trieRes []*trie.SyncResult // Response data of the leaf-mode peer (nil for timeouts) + + dropped bool // Flag whether the peer dropped off early } // timedOut returns if this request timed out. func (req *stateReq) timedOut() bool { - return req.response == nil + return req.nodeRes == nil && req.trieRes == nil } // stateSyncStats is a collection of progress stats to report during a state trie @@ -142,7 +147,21 @@ func (d *Downloader) runStateSync(s *stateSync) *stateSync { } // Finalize the request and queue up for processing req.timer.Stop() - req.response = pack.(*statePack).states + req.nodeRes = pack.(*statePack).states + + finished = append(finished, req) + delete(active, pack.PeerId()) + + case pack := <-d.trieCh: + // Discard any data not requested (or previsouly timed out) + req := active[pack.PeerId()] + if req == nil { + log.Debug("Unrequested trie data", "peer", pack.PeerId(), "len", pack.Items()) + continue + } + // Finalize the request and queue up for processing + req.timer.Stop() + req.trieRes = pack.(*triePack).results finished = append(finished, req) delete(active, pack.PeerId()) @@ -288,23 +307,48 @@ func (s *stateSync) loop() error { return errCancelStateFetch case req := <-s.deliver: - // Response, disconnect or timeout triggered, drop the peer if stalling - log.Trace("Received node data response", "peer", req.peer.id, "count", len(req.response), "dropped", req.dropped, "timeout", !req.dropped && req.timedOut()) - if len(req.items) <= 2 && !req.dropped && req.timedOut() { - // 2 items are the minimum requested, if even that times out, we've no use of - // this peer at the moment. - log.Warn("Stalling state sync, dropping peer", "peer", req.peer.id) - s.d.dropPeer(req.peer.id) - } - // Process all the received blobs and check for stale delivery - stale, err := s.process(req) - if err != nil { - log.Warn("Node data write error", "err", err) - return err - } - // The the delivery contains requested data, mark the node idle (otherwise it's a timed out delivery) - if !stale { - req.peer.SetNodeDataIdle(len(req.response)) + switch req.limit { + // If the request leaf limit is 0, we're in node-sync mode + case 0: + // Response, disconnect or timeout triggered, drop the peer if stalling + log.Trace("Received node data response", "peer", req.peer.id, "count", len(req.nodeRes), "dropped", req.dropped, "timeout", !req.dropped && req.timedOut()) + if len(req.items) <= 2 && !req.dropped && req.timedOut() { + // 2 items are the minimum requested, if even that times out, we've no use of + // this peer at the moment. + log.Warn("Stalling state sync, dropping peer", "peer", req.peer.id) + s.d.dropPeer(req.peer.id) + } + // Process all the received blobs and check for stale delivery + stale, err := s.process(req) + if err != nil { + log.Warn("Node data write error", "err", err) + return err + } + // The the delivery contains requested data, mark the node idle (otherwise it's a timed out delivery) + if !stale { + req.peer.SetNodeDataIdle(len(req.nodeRes)) + } + + // Otherwise we're in leaf-sync mode + default: + // Response, disconnect or timeout triggered, drop the peer if stalling + log.Trace("Received trie response", "peer", req.peer.id, "count", len(req.trieRes), "dropped", req.dropped, "timeout", !req.dropped && req.timedOut()) + if len(req.items) <= 2 && !req.dropped && req.timedOut() { + // 2 bytes are the minimum requested, if even that times out, we've no use of + // this peer at the moment. + log.Warn("Stalling trie sync, dropping peer", "peer", req.peer.id) + s.d.dropPeer(req.peer.id) + } + // Process all the received data and check for stale delivery + stale, err := s.process(req) + if err != nil { + log.Warn("Trie write error", "err", err) + return err + } + // The the delivery contains requested data, mark the node idle (otherwise it's a timed out delivery) + if !stale { + req.peer.SetTrieIdle(int(req.limit)) + } } } } @@ -321,7 +365,7 @@ func (s *stateSync) commit(force bool) error { if err := b.Write(); err != nil { return fmt.Errorf("DB write error: %v", err) } - s.updateStats(s.numUncommitted, 0, 0, time.Since(start)) + s.updateStats(s.numUncommitted, s.bytesUncommitted, 0, 0, time.Since(start)) s.numUncommitted = 0 s.bytesUncommitted = 0 return nil @@ -330,8 +374,25 @@ func (s *stateSync) commit(force bool) error { // assignTasks attempts to assing new tasks to all idle peers, either from the // batch currently being retried, or fetching new data from the trie sync itself. func (s *stateSync) assignTasks() { + // Iterate over all idle trie-sync capable peers and try to assign them state fetches + peers, _ := s.d.peers.TrieIdlePeers() + for _, p := range peers { + // Assign a batch of fetches proportional to the estimated latency/bandwidth + req := &stateReq{peer: p, timeout: s.d.requestTTL(), limit: common.StorageSize(p.TrieCapacity(s.d.requestRTT()))} + s.fillTasks(MaxStateFetch, req) + + // If the peer was assigned tasks to fetch, send the network request + if len(req.items) > 0 { + req.peer.log.Trace("Requesting new batch of data", "type", "trie", "limit", req.limit) + select { + case s.d.trackStateReq <- req: + req.peer.FetchTries(req.items, req.limit) + case <-s.cancel: + } + } + } // Iterate over all idle peers and try to assign them state fetches - peers, _ := s.d.peers.NodeDataIdlePeers() + peers, _ = s.d.peers.NodeDataIdlePeers() for _, p := range peers { // Assign a batch of fetches proportional to the estimated latency/bandwidth cap := p.NodeDataCapacity(s.d.requestRTT()) @@ -389,20 +450,31 @@ func (s *stateSync) process(req *stateReq) (bool, error) { defer func(start time.Time) { if duplicate > 0 || unexpected > 0 { - s.updateStats(0, duplicate, unexpected, time.Since(start)) + s.updateStats(0, 0, duplicate, unexpected, time.Since(start)) } }(time.Now()) - // Iterate over all the delivered data and inject one-by-one into the trie - progress, stale := false, len(req.response) > 0 + // Iterate over all the delivered data and inject into the trie + results := make([]*trie.SyncResult, 0, len(req.nodeRes)+1) + if req.trieRes != nil { + for _, res := range req.trieRes { + results = append(results, res) + } + } else { + for _, blob := range req.nodeRes { + results = append(results, &trie.SyncResult{Data: blob}) + } + } + progress, stale := false, len(req.nodeRes) > 0 || len(req.trieRes) > 0 + for _, res := range results { + items, bytes, hash, err := s.sched.Process(res) + + s.numUncommitted += items + s.bytesUncommitted += int(bytes) - for _, blob := range req.response { - prog, hash, err := s.sched.Process(&trie.SyncResult{Data: blob}) switch err { case nil: - s.numUncommitted++ - s.bytesUncommitted += len(blob) - progress = progress || prog + progress = progress || items > 0 case trie.ErrNotRequested: unexpected++ case trie.ErrAlreadyProcessed: @@ -428,7 +500,7 @@ func (s *stateSync) process(req *stateReq) (bool, error) { // If the node did deliver something, missing items may be due to a protocol // limit or a previous timeout + delayed delivery. Both cases should permit // the node to retry the missing items (to avoid single-peer stalls). - if len(req.response) > 0 || req.timedOut() { + if len(req.nodeRes) > 0 || len(req.trieRes) > 0 || req.timedOut() { delete(task.attempts, req.peer.id) } // If we've requested the node too many times already, it may be a malicious @@ -444,7 +516,7 @@ func (s *stateSync) process(req *stateReq) (bool, error) { // updateStats bumps the various state sync progress counters and displays a log // message for the user to see. -func (s *stateSync) updateStats(written, duplicate, unexpected int, duration time.Duration) { +func (s *stateSync) updateStats(written, bytes, duplicate, unexpected int, duration time.Duration) { s.d.syncStatsLock.Lock() defer s.d.syncStatsLock.Unlock() @@ -454,6 +526,6 @@ func (s *stateSync) updateStats(written, duplicate, unexpected int, duration tim s.d.syncStatsState.unexpected += uint64(unexpected) if written > 0 || duplicate > 0 || unexpected > 0 { - log.Info("Imported new state entries", "count", 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) + log.Info("Imported new state entries", "count", written, "bytes", bytes, "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) } } diff --git a/eth/downloader/types.go b/eth/downloader/types.go index 3f30ea9dd1..7b1ac1924c 100644 --- a/eth/downloader/types.go +++ b/eth/downloader/types.go @@ -20,6 +20,7 @@ import ( "fmt" "github.com/ethereum/go-ethereum/core/types" + "github.com/ethereum/go-ethereum/trie" ) // peerDropFn is a callback type for dropping a peer detected as malicious. @@ -77,3 +78,14 @@ type statePack struct { func (p *statePack) PeerId() string { return p.peerId } func (p *statePack) Items() int { return len(p.states) } func (p *statePack) Stats() string { return fmt.Sprintf("%d", len(p.states)) } + +// triePack is a subtrie returned by a peer (or a single node if the hash does +// not denote or properly define a state trie). +type triePack struct { + peerId string + results []*trie.SyncResult +} + +func (p *triePack) PeerId() string { return p.peerId } +func (p *triePack) Items() int { return len(p.results) } +func (p *triePack) Stats() string { return fmt.Sprintf("%d", len(p.results)) } diff --git a/eth/handler_test.go b/eth/handler_test.go index 88028a0514..185e028651 100644 --- a/eth/handler_test.go +++ b/eth/handler_test.go @@ -470,7 +470,7 @@ func testDAOChallenge(t *testing.T, localForked, remoteForked bool, timeout bool var ( evmux = new(event.TypeMux) pow = ethash.NewFaker() - db, _ = ethdb.NewMemDatabase() + db = ethdb.NewMemDatabase() config = ¶ms.ChainConfig{DAOForkBlock: big.NewInt(1), DAOForkSupport: localForked} gspec = &core.Genesis{Config: config} genesis = gspec.MustCommit(db) diff --git a/eth/helper_test.go b/eth/helper_test.go index b665531352..46fb85a4e0 100644 --- a/eth/helper_test.go +++ b/eth/helper_test.go @@ -53,7 +53,7 @@ func newTestProtocolManager(mode downloader.SyncMode, blocks int, generator func var ( evmux = new(event.TypeMux) engine = ethash.NewFaker() - db, _ = ethdb.NewMemDatabase() + db = ethdb.NewMemDatabase() gspec = &core.Genesis{ Config: params.TestChainConfig, Alloc: core.GenesisAlloc{testBank: {Balance: big.NewInt(1000000)}}, diff --git a/eth/peer.go b/eth/peer.go index 42ead53965..806f6e62dc 100644 --- a/eth/peer.go +++ b/eth/peer.go @@ -222,6 +222,14 @@ func (p *peer) RequestNodeData(hashes []common.Hash) error { return p2p.Send(p.rw, GetNodeDataMsg, hashes) } +// RequestTries fetches a sub-trie from a node's known state data, corresponding +// to the specified root hash. +func (p *peer) RequestTries(roots []common.Hash, limit common.StorageSize) error { + p.Log().Debug("Fetching sub-trie", "roots", roots, "limit", limit) + panic("not implemented") + //return p2p.Send(p.rw, GetNodeDataMsg, hashes) +} + // RequestReceipts fetches a batch of transaction receipts from a remote node. func (p *peer) RequestReceipts(hashes []common.Hash) error { p.Log().Debug("Fetching batch of receipts", "count", len(hashes)) diff --git a/trie/sync.go b/trie/sync.go index c9fd650681..0f220edf46 100644 --- a/trie/sync.go +++ b/trie/sync.go @@ -21,6 +21,9 @@ import ( "errors" "fmt" "hash" + "math" + "reflect" + "sync/atomic" "github.com/ethereum/go-ethereum/common" "github.com/ethereum/go-ethereum/crypto/sha3" @@ -89,6 +92,8 @@ type TrieSync struct { requests map[common.Hash]*request // Pending requests pertaining to a key hash queue *prque.Prque // Priority queue with the pending requests keccak hash.Hash // Keccak256 hasher to verify deliveries with + + nextId uint64 // Identifier component for the priority queue to split between same depths } // NewTrieSync creates a new trie data download scheduler. @@ -133,7 +138,7 @@ func (s *TrieSync) AddSubTrie(root common.Hash, depth int, parent common.Hash, c ancestor.deps++ req.parents = append(req.parents, ancestor) } - s.schedule(req) + s.schedule(req, false) } // AddRawEntry schedules the direct retrieval of a state entry that should not be @@ -166,7 +171,7 @@ func (s *TrieSync) AddRawEntry(hash common.Hash, depth int, parent common.Hash) ancestor.deps++ req.parents = append(req.parents, ancestor) } - s.schedule(req) + s.schedule(req, false) } // Missing retrieves the known missing nodes from the trie for retrieval. @@ -174,20 +179,21 @@ func (s *TrieSync) Missing(max int) []common.Hash { requests := []common.Hash{} for !s.queue.Empty() && (max == 0 || len(requests) < max) { hash := s.queue.PopItem().(common.Hash) - if s.requests[hash] != nil { + if req := s.requests[hash]; req != nil && req.data == nil { requests = append(requests, hash) + } else { + fmt.Printf(".") } } return requests } -// 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(result *SyncResult) (bool, common.Hash, error) { +// Process injects a batch of retrieved trie data, returning the number of nodes +// and bytes written, along with the hash of the node or sub-trie just processed. +func (s *TrieSync) Process(result *SyncResult) (int, common.StorageSize, common.Hash, error) { // If it's a plain or full sub-trie delivery, inject and return if len(result.Keys) == 0 && len(result.Proof) == 0 { - return s.processNode(result.Data) + return s.processNode(common.Hash{}, result.Data, false) } if len(result.Proof) == 0 { return s.processLeaves(result.Keys, result.Values) @@ -201,93 +207,101 @@ func (s *TrieSync) Process(result *SyncResult) (bool, common.Hash, error) { // processNode verifies and processes a trie node, returning if anything was // committed and the hash of the node injected. -func (s *TrieSync) processNode(blob []byte) (bool, common.Hash, error) { +func (s *TrieSync) processNode(hash common.Hash, blob []byte, ready bool) (int, common.StorageSize, common.Hash, error) { // Derive the hash of the result based on its content - var hash common.Hash - - s.keccak.Reset() - s.keccak.Write(blob) - s.keccak.Sum(hash[:0]) - + if hash == (common.Hash{}) { + s.keccak.Reset() + s.keccak.Write(blob) + s.keccak.Sum(hash[:0]) + } // If the item was not requested, bail out request := s.requests[hash] if request == nil { - return false, hash, ErrNotRequested + return 0, 0, hash, nil //ErrNotRequested } if request.data != nil { - return false, hash, nil // TODO(karalabe): Why not ErrAlreadyProcessed + return 0, 0, hash, ErrAlreadyProcessed } // If the item is a raw entry request, commit directly if request.raw { request.data = blob - s.commit(request) - return true, hash, nil + items, bytes := s.commit(request) + return items, bytes, hash, nil } // Decode and inject into the trie node, err := decodeNode(hash[:], blob, 0) if err != nil { - return false, hash, err + return 0, 0, hash, err } request.data = blob // Create and schedule a request for all the children nodes requests, err := s.children(request, node) if err != nil { - return false, hash, err + return 0, 0, hash, err } if len(requests) == 0 && request.deps == 0 { - s.commit(request) - return true, hash, nil + items, bytes := s.commit(request) + return items, bytes, hash, nil } request.deps += len(requests) for _, child := range requests { - s.schedule(child) + s.schedule(child, ready) } - return false, hash, nil + return 0, 0, hash, nil } // processLeaves reconstructs a sub-trie from the given key-value pairs, returning -// the root of the sub-trie or an error on failure. -func (s *TrieSync) processLeaves(keys [][]byte, values [][]byte) (bool, common.Hash, error) { +// the number of nodes and bytes written, along with the hash of the sub-trie just +// processed. +func (s *TrieSync) processLeaves(keys [][]byte, values [][]byte) (int, common.StorageSize, common.Hash, error) { // Inject all the leaves into a fresh trie and derive it's root hash db := ethdb.NewMemDatabase() trie, err := New(common.Hash{}, db) if err != nil { - return false, common.Hash{}, err + return 0, 0, common.Hash{}, err } for j := 0; j < len(keys); j++ { trie.Update(keys[j], values[j]) } root, err := trie.Commit() if err != nil { - return false, common.Hash{}, err + return 0, 0, common.Hash{}, err } // If the item was not requested, bail out request := s.requests[root] if request == nil { - return false, root, ErrNotRequested + return 0, 0, root, ErrNotRequested } if request.data != nil { - return false, root, ErrAlreadyProcessed + return 0, 0, root, ErrAlreadyProcessed } // Inject all key-values as is and complete the root - for _, key := range db.Keys() { - value, _ := db.Get(key) - if hash := common.BytesToHash(key); hash != root { - s.commitEntry(hash, value) - } else { - request.data = value + var ( + items int + bytes common.StorageSize + ) + it := trie.NodeIterator(nil) + for it.Next(true) { + if hash := it.Hash(); hash != (common.Hash{}) { + blob, _ := db.Get(hash[:]) + count, size, _, err := s.processNode(hash, blob, true) + + items += count + bytes += size + + if err != nil { + return items, bytes, root, err + } } } - s.commit(request) - - return true, root, nil + return items, bytes, root, nil } // processPartialLeaves reconstructs a sub-trie from the Merkle proof and the // available key-value pairs, commiting the available parts and scheduling the // missing items for future retrival. -func (s *TrieSync) processPartialLeaves(keys [][]byte, values [][]byte, proof [][]byte) (bool, common.Hash, error) { +func (s *TrieSync) processPartialLeaves(keys [][]byte, values [][]byte, proof [][]byte) (int, common.StorageSize, common.Hash, error) { // Derive the hash of the topmost proof var root common.Hash @@ -298,29 +312,29 @@ func (s *TrieSync) processPartialLeaves(keys [][]byte, values [][]byte, proof [] // If the item was not requested, bail out request := s.requests[root] if request == nil { - return false, root, ErrNotRequested + return 0, 0, root, ErrNotRequested } if request.data != nil { - return false, root, ErrAlreadyProcessed + return 0, 0, root, ErrAlreadyProcessed } // Decode the root node and schedule missing children node, err := decodeNode(root[:], proof[0], 0) if err != nil { - return false, root, err + return 0, 0, root, err } request.data = proof[0] requests, err := s.children(request, node) if err != nil { - return false, root, err + return 0, 0, root, err } if len(requests) == 0 && request.deps == 0 { - s.commit(request) - return true, root, nil + items, bytes := s.commit(request) + return items, bytes, root, nil } request.deps += len(requests) for _, child := range requests { - s.schedule(child) + s.schedule(child, false) } // Fulfill any children satisfied by the key-value pairs switch node := (node).(type) { @@ -328,15 +342,20 @@ func (s *TrieSync) processPartialLeaves(keys [][]byte, values [][]byte, proof [] // All keys must have the short node's path as a prefix for i, key := range keys { if !bytes.HasPrefix(key, node.Key) { - return false, root, fmt.Errorf("key mismatch at proof %x", proof[0]) + return 0, 0, root, fmt.Errorf("key mismatch at proof %x", proof[0]) } keys[i] = key[len(node.Key):] } // Recurse into the subtrie of the short node - commit, _, err := s.processPartialLeaves(keys, values, proof[1:]) - return commit, root, err + items, bytes, _, err := s.processPartialLeaves(keys, values, proof[1:]) + return items, bytes, root, err case *fullNode: + // Track the number of items and bytes written + var ( + items int + bytes common.StorageSize + ) // Split up the keyspace between the full node's children for i := 0; i < 17; i++ { if node.Children[i] != nil { @@ -350,21 +369,26 @@ func (s *TrieSync) processPartialLeaves(keys [][]byte, values [][]byte, proof [] if _, ok := node.Children[i].(hashNode); !ok { // If we're at the last node, process it as a partial trie if split == len(keys) && len(proof) != 1 { - commit, _, err := s.processPartialLeaves(keys[:split], values[:split], proof[1:]) - return commit, root, err + count, size, _, err := s.processPartialLeaves(keys[:split], values[:split], proof[1:]) + return items + count, bytes + size, root, err } // Otherwise we have a full sub-trie, parse in its entirety (if not already contained within the full node) - commit, _, err := s.processLeaves(keys[:split], values[:split]) + count, size, _, err := s.processLeaves(keys[:split], values[:split]) + + items += count + bytes += size + if err != nil { - return commit, root, err + return items, bytes, root, err } } keys = keys[split:] values = values[split:] } } + return items, bytes, root, nil } - return false, root, nil + return 0, 0, root, fmt.Errorf("unexpected node type: %v", reflect.TypeOf(node)) } // Commit flushes the data stored in the internal membatch out to persistent @@ -391,14 +415,16 @@ func (s *TrieSync) Pending() int { // schedule inserts a new state retrieval request into the fetch queue. If there // is already a pending request for this node, the new request will be discarded // and only a parent reference added to the old one. -func (s *TrieSync) schedule(req *request) { +func (s *TrieSync) schedule(req *request, ready bool) { // If we're already requesting this node, add a new reference and stop if old, ok := s.requests[req.hash]; ok { old.parents = append(old.parents, req.parents...) return } // Schedule the request for future retrieval - s.queue.Push(req.hash, float32(req.depth)) + if !ready { + s.queue.Push(req.hash, float32(req.depth)*math.MaxUint64+float32(math.MaxUint64-atomic.AddUint64(&s.nextId, 1))) + } s.requests[req.hash] = req } @@ -465,8 +491,13 @@ func (s *TrieSync) children(req *request, object node) ([]*request, error) { // commit finalizes a retrieval request and stores it into the membatch. If any // of the referencing parent requests complete due to this commit, they are also -// committed themselves. -func (s *TrieSync) commit(req *request) (err error) { +// committed themselves. The method returns the number of state items written to +// the membatch as well as their total data size. +func (s *TrieSync) commit(req *request) (int, common.StorageSize) { + var ( + items = 1 + bytes = common.StorageSize(len(req.data)) + ) // Write the node content to the membatch s.commitEntry(req.hash, req.data) delete(s.requests, req.hash) @@ -475,12 +506,13 @@ func (s *TrieSync) commit(req *request) (err error) { for _, parent := range req.parents { parent.deps-- if parent.deps == 0 { - if err := s.commit(parent); err != nil { - return err - } + count, size := s.commit(parent) + + items += count + bytes += size } } - return nil + return items, bytes } // commitEntry injects a raw database entry into the memory batch to be flushed diff --git a/trie/sync_test.go b/trie/sync_test.go index 18146ee3fa..59ab218ea8 100644 --- a/trie/sync_test.go +++ b/trie/sync_test.go @@ -20,6 +20,7 @@ import ( "bytes" "math/rand" "testing" + "time" "github.com/ethereum/go-ethereum/common" "github.com/ethereum/go-ethereum/ethdb" @@ -124,11 +125,11 @@ func testIterativeTrieSync(t *testing.T, batch int) { } results[i] = &SyncResult{Data: data} } else { - results[i], _ = srcTrie.FetchData(hash, 8192) + results[i], _, _ = srcTrie.FetchData(hash, 8192, 250*time.Millisecond) } } for index, result := range results { - if _, _, err := sched.Process(result); err != nil { + if _, _, _, err := sched.Process(result); err != nil { t.Fatalf("failed to process result #%d: %v", index, err) } } @@ -163,11 +164,11 @@ func TestIterativeDelayedTrieSync(t *testing.T) { } results[i] = &SyncResult{Data: data} } else { - results[i], _ = srcTrie.FetchData(hash, 8192) + results[i], _, _ = srcTrie.FetchData(hash, 8192, 250*time.Millisecond) } } for index, result := range results { - if _, _, err := sched.Process(result); err != nil { + if _, _, _, err := sched.Process(result); err != nil { t.Fatalf("failed to process result #%d: %v", index, err) } } @@ -209,13 +210,13 @@ func testIterativeRandomTrieSync(t *testing.T, batch int) { } results = append(results, &SyncResult{Data: data}) } else { - request, _ := srcTrie.FetchData(hash, 8192) + request, _, _ := srcTrie.FetchData(hash, 8192, 250*time.Millisecond) results = append(results, request) } } // Feed the retrieved results back and queue new tasks for index, result := range results { - if _, _, err := sched.Process(result); err != nil { + if _, _, _, err := sched.Process(result); err != nil { t.Fatalf("failed to process result #%d: %v", index, err) } } @@ -256,7 +257,7 @@ func TestIterativeRandomDelayedTrieSync(t *testing.T) { } results = append(results, &SyncResult{Data: data}) } else { - request, _ := srcTrie.FetchData(hash, 8192) + request, _, _ := srcTrie.FetchData(hash, 8192, 250*time.Millisecond) results = append(results, request) } if len(results) >= cap(results) { @@ -265,7 +266,7 @@ func TestIterativeRandomDelayedTrieSync(t *testing.T) { } // Feed the retrieved results back and queue new tasks for index, result := range results { - _, hash, err := sched.Process(result) + _, _, hash, err := sched.Process(result) if err != nil { t.Fatalf("failed to process result #%d: %v", index, err) } @@ -310,11 +311,11 @@ func TestDuplicateAvoidanceTrieSync(t *testing.T) { } results[i] = &SyncResult{Data: data} } else { - results[i], _ = srcTrie.FetchData(hash, 8192) + results[i], _, _ = srcTrie.FetchData(hash, 8192, 250*time.Millisecond) } } for index, result := range results { - if _, _, err := sched.Process(result); err != nil { + if _, _, _, err := sched.Process(result); err != nil { t.Fatalf("failed to process result #%d: %v", index, err) } } @@ -350,12 +351,12 @@ func TestIncompleteTrieSync(t *testing.T) { } results[i] = &SyncResult{Data: data} } else { - results[i], _ = srcTrie.FetchData(hash, 8192) + results[i], _, _ = srcTrie.FetchData(hash, 8192, 250*time.Millisecond) } } // Process each of the trie nodes for index, result := range results { - _, hash, err := sched.Process(result) + _, _, hash, err := sched.Process(result) if err != nil { t.Fatalf("failed to process result #%d: %v", index, err) } diff --git a/trie/trie.go b/trie/trie.go index f23a7d4630..2a9af6a650 100644 --- a/trie/trie.go +++ b/trie/trie.go @@ -20,10 +20,12 @@ package trie import ( "bytes" "fmt" + "time" "github.com/ethereum/go-ethereum/common" "github.com/ethereum/go-ethereum/crypto/sha3" "github.com/ethereum/go-ethereum/log" + "github.com/ethereum/go-ethereum/rlp" "github.com/rcrowley/go-metrics" ) @@ -506,7 +508,17 @@ func (t *Trie) hashRoot(db DatabaseWriter) (node, node, error) { // FetchData retrieves synchronization data from the trie to send to remote // nodes. -func (t *Trie) FetchData(hash common.Hash, limit common.StorageSize) (res *SyncResult, fail error) { +func (t *Trie) FetchData(hash common.Hash, limit common.StorageSize, timeout time.Duration) (res *SyncResult, size common.StorageSize, fail error) { + // If we have a snapshot of this exact data, return without iteration + if blob, _ := t.db.Get(append([]byte("S"), hash[:]...)); blob != nil { + result := new(SyncResult) + if err := rlp.DecodeBytes(blob, result); err == nil { + return result, common.StorageSize(len(blob)), nil + } + } + // Start measuring the time to allow aborting if the storage limit is too generous + start := time.Now() + // If this method panics, hash points to a non-iterable trie; return individual node defer func() { if r := recover(); r != nil { @@ -514,7 +526,7 @@ func (t *Trie) FetchData(hash common.Hash, limit common.StorageSize) (res *SyncR if err != nil { fail = err } else { - res = &SyncResult{Data: blob} + res, size = &SyncResult{Data: blob}, common.StorageSize(len(blob)) } } }() @@ -524,19 +536,26 @@ func (t *Trie) FetchData(hash common.Hash, limit common.StorageSize) (res *SyncR trie, _ := New(hash, t.db) it := NewIterator(trie.NodeIterator(nil)) - size := common.StorageSize(0) - for size < limit && it.Next() { + size = common.StorageSize(0) + for size < limit && time.Since(start) < timeout && it.Next() { result.Keys = append(result.Keys, common.CopyBytes(it.Key)) result.Values = append(result.Values, common.CopyBytes(it.Value)) size += common.StorageSize(len(it.Key) + len(it.Value)) } // If we've went past our data allowance, prove the partial data - if size >= limit { + if size >= limit || time.Since(start) >= timeout { result.Proof = it.Prove() if !it.Next() { result.Proof = nil // Overflowing item was the last after all } } - return result, nil + // If we have the entire storage-trie in memory, create a snapshot of it + if result.Proof == nil && len(result.Keys) > 0 && len(result.Keys[0]) == 32 { + if blob, err := rlp.EncodeToBytes(result); err == nil { + fmt.Printf(".") + t.db.Put(append([]byte("S"), hash[:]...), blob) + } + } + return result, size, nil }