From 14174a69e9b23141cd90267eeb9bc7efac9dab30 Mon Sep 17 00:00:00 2001 From: rjl493456442 Date: Fri, 25 Oct 2019 11:37:52 +0800 Subject: [PATCH] all: integrate CDN idea into light client --- cmd/geth/config.go | 3 - cmd/geth/main.go | 2 + cmd/geth/usage.go | 2 + cmd/utils/flags.go | 17 + .../selector/randomselect.go | 62 ++- .../selector/randomselect_test.go | 10 +- core/blockchain.go | 5 + eth/config.go | 10 +- eth/gen_config.go | 12 + ethclient/ethclient.go | 2 +- les/client.go | 10 +- les/client_handler.go | 4 +- les/distributor.go | 9 +- les/http_retriever.go | 486 ++++++++++++++++++ les/http_retriever_test.go | 94 ++++ les/odr.go | 130 +++-- les/odr_requests.go | 42 +- les/{retrieve.go => p2p_retrieve.go} | 25 +- les/server.go | 4 +- les/serverpool.go | 39 +- les/test_helper.go | 6 +- les/txrelay.go | 4 +- lescdn/chain.go | 43 +- lescdn/gzip.go | 16 + lescdn/lesleech/main.go | 16 + lescdn/misc.go | 141 +++++ lescdn/service.go | 31 +- lescdn/state.go | 20 +- light/cht_tiler.go | 368 +++++++++++++ light/odr.go | 81 ++- light/odr_test.go | 6 +- light/odr_util.go | 5 +- light/postprocess.go | 25 +- light/trie.go | 5 +- trie/iterator.go | 6 +- trie/proof.go | 41 ++ trie/trie.go | 84 ++- trie/trie_test.go | 51 ++ 38 files changed, 1733 insertions(+), 184 deletions(-) rename les/randselect.go => common/selector/randomselect.go (75%) rename les/randselect_test.go => common/selector/randomselect_test.go (92%) create mode 100644 les/http_retriever.go create mode 100644 les/http_retriever_test.go rename les/{retrieve.go => p2p_retrieve.go} (93%) create mode 100644 lescdn/misc.go create mode 100644 light/cht_tiler.go diff --git a/cmd/geth/config.go b/cmd/geth/config.go index 07006b575f..71e8355348 100644 --- a/cmd/geth/config.go +++ b/cmd/geth/config.go @@ -178,9 +178,6 @@ func makeFullNode(ctx *cli.Context) *node.Node { if cfg.Ethstats.URL != "" { utils.RegisterEthStatsService(stack, cfg.Ethstats.URL) } - // Add the light server CDN if requested - utils.RegisterLesCDNService(stack) - return stack } diff --git a/cmd/geth/main.go b/cmd/geth/main.go index 99ef78238f..b29429d044 100644 --- a/cmd/geth/main.go +++ b/cmd/geth/main.go @@ -101,6 +101,8 @@ var ( utils.UltraLightServersFlag, utils.UltraLightFractionFlag, utils.UltraLightOnlyAnnounceFlag, + utils.LesCDNURLFlag, + utils.LesCDNSwitchFlag, utils.WhitelistFlag, utils.CacheFlag, utils.CacheDatabaseFlag, diff --git a/cmd/geth/usage.go b/cmd/geth/usage.go index 6f3197b9c6..ea87618240 100644 --- a/cmd/geth/usage.go +++ b/cmd/geth/usage.go @@ -94,6 +94,8 @@ var AppHelpFlagGroups = []flagGroup{ utils.UltraLightServersFlag, utils.UltraLightFractionFlag, utils.UltraLightOnlyAnnounceFlag, + utils.LesCDNURLFlag, + utils.LesCDNSwitchFlag, }, }, { diff --git a/cmd/utils/flags.go b/cmd/utils/flags.go index 5ede9de434..6a97624243 100644 --- a/cmd/utils/flags.go +++ b/cmd/utils/flags.go @@ -287,6 +287,15 @@ var ( Name: "ulc.onlyannounce", Usage: "Ultra light server sends announcements only", } + LesCDNURLFlag = cli.StringFlag{ + Name: "light.cdn.url", + Usage: "The URL of les cdn which can speed up request serving", + } + LesCDNSwitchFlag = cli.Int64Flag{ + Name: "light.cdn.switch", + Usage: "The maximum waiting time on the p2p network before switch to CDN(count by ms)", + Value: 100, // Default cdn switch is 100ms + } // Ethash settings EthashCacheDirFlag = DirectoryFlag{ Name: "ethash.cachedir", @@ -1010,6 +1019,12 @@ func setLes(ctx *cli.Context, cfg *eth.Config) { if ctx.GlobalIsSet(UltraLightOnlyAnnounceFlag.Name) { cfg.UltraLightOnlyAnnounce = ctx.GlobalBool(UltraLightOnlyAnnounceFlag.Name) } + if ctx.GlobalIsSet(LesCDNURLFlag.Name) { + cfg.LesCDNURL = ctx.GlobalString(LesCDNURLFlag.Name) + } + if ctx.GlobalIsSet(LesCDNSwitchFlag.Name) { + cfg.LesCDNSwitch = time.Duration(ctx.GlobalInt64(LesCDNSwitchFlag.Name) * 1000 * 1000) + } } // makeDatabaseHandles raises out the number of allowed file handles per process @@ -1569,6 +1584,8 @@ func RegisterEthService(stack *node.Node, cfg *eth.Config) { } return fullNode, err }) + // Add the light server CDN if requested + RegisterLesCDNService(stack) } if err != nil { Fatalf("Failed to register the Ethereum service: %v", err) diff --git a/les/randselect.go b/common/selector/randomselect.go similarity index 75% rename from les/randselect.go rename to common/selector/randomselect.go index 8efe0c94d3..d7888a92cf 100644 --- a/les/randselect.go +++ b/common/selector/randomselect.go @@ -14,43 +14,30 @@ // You should have received a copy of the GNU Lesser General Public License // along with the go-ethereum library. If not, see . -package les +package selector -import ( - "math/rand" -) +import "math/rand" -// wrsItem interface should be implemented by any entries that are to be selected from -// a weightedRandomSelect set. Note that recalculating monotonously decreasing item -// weights on-demand (without constantly calling update) is allowed -type wrsItem interface { +// SelectedItem interface should be implemented by any entries that are to be selected from +// a WeightedRandomSelect set. Note that recalculating monotonously decreasing item +// weights on-demand (without constantly calling Update) is allowed +type SelectedItem interface { Weight() int64 } -// weightedRandomSelect is capable of weighted random selection from a set of items -type weightedRandomSelect struct { +// WeightedRandomSelect is capable of weighted random selection from a set of items +type WeightedRandomSelect struct { root *wrsNode - idx map[wrsItem]int + idx map[SelectedItem]int } -// newWeightedRandomSelect returns a new weightedRandomSelect structure -func newWeightedRandomSelect() *weightedRandomSelect { - return &weightedRandomSelect{root: &wrsNode{maxItems: wrsBranches}, idx: make(map[wrsItem]int)} -} - -// update updates an item's weight, adds it if it was non-existent or removes it if -// the new weight is zero. Note that explicitly updating decreasing weights is not necessary. -func (w *weightedRandomSelect) update(item wrsItem) { - w.setWeight(item, item.Weight()) -} - -// remove removes an item from the set -func (w *weightedRandomSelect) remove(item wrsItem) { - w.setWeight(item, 0) +// NewWeightedRandomSelect returns a new WeightedRandomSelect structure +func NewWeightedRandomSelect() *WeightedRandomSelect { + return &WeightedRandomSelect{root: &wrsNode{maxItems: wrsBranches}, idx: make(map[SelectedItem]int)} } // setWeight sets an item's weight to a specific value (removes it if zero) -func (w *weightedRandomSelect) setWeight(item wrsItem, weight int64) { +func (w *WeightedRandomSelect) setWeight(item SelectedItem, weight int64) { idx, ok := w.idx[item] if ok { w.root.setWeight(idx, weight) @@ -71,11 +58,22 @@ func (w *weightedRandomSelect) setWeight(item wrsItem, weight int64) { } } -// choose randomly selects an item from the set, with a chance proportional to its +// Update updates an item's weight, adds it if it was non-existent or removes it if +// the new weight is zero. Note that explicitly updating decreasing weights is not necessary. +func (w *WeightedRandomSelect) Update(item SelectedItem) { + w.setWeight(item, item.Weight()) +} + +// Remove removes an item from the set +func (w *WeightedRandomSelect) Remove(item SelectedItem) { + w.setWeight(item, 0) +} + +// Choose randomly selects an item from the set, with a chance proportional to its // current weight. If the weight of the chosen element has been decreased since the // last stored value, returns it with a newWeight/oldWeight chance, otherwise just // updates its weight and selects another one -func (w *weightedRandomSelect) choose() wrsItem { +func (w *WeightedRandomSelect) Choose() SelectedItem { for { if w.root.sumWeight == 0 { return nil @@ -103,7 +101,7 @@ type wrsNode struct { } // insert recursively inserts a new item to the tree and returns the item index -func (n *wrsNode) insert(item wrsItem, weight int64) int { +func (n *wrsNode) insert(item SelectedItem, weight int64) int { branch := 0 for n.items[branch] != nil && (n.level == 0 || n.items[branch].(*wrsNode).itemCnt == n.items[branch].(*wrsNode).maxItems) { branch++ @@ -154,12 +152,12 @@ func (n *wrsNode) setWeight(idx int, weight int64) int64 { return diff } -// choose recursively selects an item from the tree and returns it along with its weight -func (n *wrsNode) choose(val int64) (wrsItem, int64) { +// Choose recursively selects an item from the tree and returns it along with its weight +func (n *wrsNode) choose(val int64) (SelectedItem, int64) { for i, w := range n.weights { if val < w { if n.level == 0 { - return n.items[i].(wrsItem), n.weights[i] + return n.items[i].(SelectedItem), n.weights[i] } return n.items[i].(*wrsNode).choose(val) } diff --git a/les/randselect_test.go b/common/selector/randomselect_test.go similarity index 92% rename from les/randselect_test.go rename to common/selector/randomselect_test.go index 9ae7726ddd..29916a6d14 100644 --- a/les/randselect_test.go +++ b/common/selector/randomselect_test.go @@ -14,7 +14,7 @@ // You should have received a copy of the GNU Lesser General Public License // along with the go-ethereum library. If not, see . -package les +package selector import ( "math/rand" @@ -36,15 +36,15 @@ func (t *testWrsItem) Weight() int64 { func TestWeightedRandomSelect(t *testing.T) { testFn := func(cnt int) { - s := newWeightedRandomSelect() + s := NewWeightedRandomSelect() w := -1 list := make([]testWrsItem, cnt) for i := range list { list[i] = testWrsItem{idx: i, widx: &w} - s.update(&list[i]) + s.Update(&list[i]) } w = rand.Intn(cnt) - c := s.choose() + c := s.Choose() if c == nil { t.Errorf("expected item, got nil") } else { @@ -53,7 +53,7 @@ func TestWeightedRandomSelect(t *testing.T) { } } w = -2 - if s.choose() != nil { + if s.Choose() != nil { t.Errorf("expected nil, got item") } } diff --git a/core/blockchain.go b/core/blockchain.go index d7fcbd5e31..8b594b60d0 100644 --- a/core/blockchain.go +++ b/core/blockchain.go @@ -532,6 +532,11 @@ func (bc *BlockChain) StateCache() state.Database { return bc.stateCache } +// Database returns the disk database underpinning the blockchain instance. +func (bc *BlockChain) Database() ethdb.Database { + return bc.db +} + // Reset purges the entire blockchain, restoring it to its genesis state. func (bc *BlockChain) Reset() error { return bc.ResetWithGenesisBlock(bc.genesisBlock) diff --git a/eth/config.go b/eth/config.go index 2eaf21fbc3..a1d5103fa5 100644 --- a/eth/config.go +++ b/eth/config.go @@ -106,10 +106,12 @@ type Config struct { Whitelist map[uint64]common.Hash `toml:"-"` // Light client options - LightServ int `toml:",omitempty"` // Maximum percentage of time allowed for serving LES requests - LightIngress int `toml:",omitempty"` // Incoming bandwidth limit for light servers - LightEgress int `toml:",omitempty"` // Outgoing bandwidth limit for light servers - LightPeers int `toml:",omitempty"` // Maximum number of LES client peers + LightServ int `toml:",omitempty"` // Maximum percentage of time allowed for serving LES requests + LightIngress int `toml:",omitempty"` // Incoming bandwidth limit for light servers + LightEgress int `toml:",omitempty"` // Outgoing bandwidth limit for light servers + LightPeers int `toml:",omitempty"` // Maximum number of LES client peers + LesCDNURL string `toml:",omitempty"` // The URL of external les cdn. + LesCDNSwitch time.Duration `toml:",omitempty"` // Maximum waiting time on p2p network before switch to CDN // Ultra Light client options UltraLightServers []string `toml:",omitempty"` // List of trusted ultra light servers diff --git a/eth/gen_config.go b/eth/gen_config.go index 1c659c393c..e306c9dd9b 100644 --- a/eth/gen_config.go +++ b/eth/gen_config.go @@ -29,6 +29,8 @@ func (c Config) MarshalTOML() (interface{}, error) { LightIngress int `toml:",omitempty"` LightEgress int `toml:",omitempty"` LightPeers int `toml:",omitempty"` + LesCDNURL string `toml:",omitempty"` + LesCDNSwitch time.Duration `toml:",omitempty"` UltraLightServers []string `toml:",omitempty"` UltraLightFraction int `toml:",omitempty"` UltraLightOnlyAnnounce bool `toml:",omitempty"` @@ -65,6 +67,8 @@ func (c Config) MarshalTOML() (interface{}, error) { enc.LightIngress = c.LightIngress enc.LightEgress = c.LightEgress enc.LightPeers = c.LightPeers + enc.LesCDNURL = c.LesCDNURL + enc.LesCDNSwitch = c.LesCDNSwitch enc.UltraLightServers = c.UltraLightServers enc.UltraLightFraction = c.UltraLightFraction enc.UltraLightOnlyAnnounce = c.UltraLightOnlyAnnounce @@ -105,6 +109,8 @@ func (c *Config) UnmarshalTOML(unmarshal func(interface{}) error) error { LightIngress *int `toml:",omitempty"` LightEgress *int `toml:",omitempty"` LightPeers *int `toml:",omitempty"` + LesCDNURL *string `toml:",omitempty"` + LesCDNSwitch *time.Duration `toml:",omitempty"` UltraLightServers []string `toml:",omitempty"` UltraLightFraction *int `toml:",omitempty"` UltraLightOnlyAnnounce *bool `toml:",omitempty"` @@ -166,6 +172,12 @@ func (c *Config) UnmarshalTOML(unmarshal func(interface{}) error) error { if dec.LightPeers != nil { c.LightPeers = *dec.LightPeers } + if dec.LesCDNURL != nil { + c.LesCDNURL = *dec.LesCDNURL + } + if dec.LesCDNSwitch != nil { + c.LesCDNSwitch = *dec.LesCDNSwitch + } if dec.UltraLightServers != nil { c.UltraLightServers = dec.UltraLightServers } diff --git a/ethclient/ethclient.go b/ethclient/ethclient.go index a60d732231..6f279d539c 100644 --- a/ethclient/ethclient.go +++ b/ethclient/ethclient.go @@ -61,7 +61,7 @@ func (ec *Client) Close() { // Blockchain Access -// ChainId retrieves the current chain ID for transaction replay protection. +// ChainID retrieves the current chain ID for transaction replay protection. func (ec *Client) ChainID(ctx context.Context) (*big.Int, error) { var result hexutil.Big err := ec.c.CallContext(ctx, &result, "eth_chainId") diff --git a/les/client.go b/les/client.go index dfd0909778..1fed8169f3 100644 --- a/les/client.go +++ b/les/client.go @@ -51,7 +51,8 @@ type LightEthereum struct { peers *serverPeerSet reqDist *requestDistributor - retriever *retrieveManager + p2pRtr *p2pRetriever + httpRtr *httpRetriever odr *LesOdr relay *lesTxRelay handler *clientHandler @@ -100,10 +101,11 @@ func New(ctx *node.ServiceContext, config *eth.Config) (*LightEthereum, error) { bloomIndexer: eth.NewBloomIndexer(chainDb, params.BloomBitsBlocksClient, params.HelperTrieConfirmations), serverPool: newServerPool(chainDb, config.UltraLightServers), } - leth.retriever = newRetrieveManager(peers, leth.reqDist, leth.serverPool) - leth.relay = newLesTxRelay(peers, leth.retriever) + leth.p2pRtr = newRetrieveManager(peers, leth.reqDist, leth.serverPool) + leth.httpRtr = newHTTPRetriever(config.LesCDNURL, config.LesCDNSwitch, chainDb) + leth.relay = newLesTxRelay(peers, leth.p2pRtr) - leth.odr = NewLesOdr(chainDb, light.DefaultClientIndexerConfig, leth.retriever) + leth.odr = NewLesOdr(chainDb, light.DefaultClientIndexerConfig, leth.p2pRtr, leth.httpRtr) leth.chtIndexer = light.NewChtIndexer(chainDb, leth.odr, params.CHTFrequency, params.HelperTrieConfirmations) leth.bloomTrieIndexer = light.NewBloomTrieIndexer(chainDb, leth.odr, params.BloomBitsBlocksClient, params.BloomTrieFrequency) leth.odr.SetIndexers(leth.chtIndexer, leth.bloomTrieIndexer, leth.bloomIndexer) diff --git a/les/client_handler.go b/les/client_handler.go index d04574c8c7..b1f182d320 100644 --- a/les/client_handler.go +++ b/les/client_handler.go @@ -299,7 +299,7 @@ func (h *clientHandler) handleMsg(p *serverPeer) error { } case StopMsg: p.freeze() - h.backend.retriever.frozen(p) + h.backend.p2pRtr.frozen(p) p.Log().Debug("Service stopped") case ResumeMsg: var bv uint64 @@ -315,7 +315,7 @@ func (h *clientHandler) handleMsg(p *serverPeer) error { } // Deliver the received response to retriever. if deliverMsg != nil { - if err := h.backend.retriever.deliver(p, deliverMsg); err != nil { + if err := h.backend.p2pRtr.deliver(p, deliverMsg); err != nil { p.errCount++ if p.errCount > maxResponseErrors { return err diff --git a/les/distributor.go b/les/distributor.go index 4d2be1b8f9..cbd7b146a6 100644 --- a/les/distributor.go +++ b/les/distributor.go @@ -22,6 +22,7 @@ import ( "time" "github.com/ethereum/go-ethereum/common/mclock" + "github.com/ethereum/go-ethereum/common/selector" ) // requestDistributor implements a mechanism that distributes requests to @@ -194,7 +195,7 @@ func (d *requestDistributor) nextRequest() (distPeer, *distReq, time.Duration) { elem := d.reqQueue.Front() var ( bestWait time.Duration - sel *weightedRandomSelect + sel *selector.WeightedRandomSelect ) d.peerLock.RLock() @@ -219,9 +220,9 @@ func (d *requestDistributor) nextRequest() (distPeer, *distReq, time.Duration) { wait, bufRemain := peer.waitBefore(cost) if wait == 0 { if sel == nil { - sel = newWeightedRandomSelect() + sel = selector.NewWeightedRandomSelect() } - sel.update(selectPeerItem{peer: peer, req: req, weight: int64(bufRemain*1000000) + 1}) + sel.Update(selectPeerItem{peer: peer, req: req, weight: int64(bufRemain*1000000) + 1}) } else { if bestWait == 0 || wait < bestWait { bestWait = wait @@ -239,7 +240,7 @@ func (d *requestDistributor) nextRequest() (distPeer, *distReq, time.Duration) { } if sel != nil { - c := sel.choose().(selectPeerItem) + c := sel.Choose().(selectPeerItem) return c.peer, c.req, 0 } return nil, nil, bestWait diff --git a/les/http_retriever.go b/les/http_retriever.go new file mode 100644 index 0000000000..119461e117 --- /dev/null +++ b/les/http_retriever.go @@ -0,0 +1,486 @@ +// Copyright 2019 The go-ethereum Authors +// This file is part of the go-ethereum library. +// +// The go-ethereum library is free software: you can redistribute it and/or modify +// it under the terms of the GNU Lesser General Public License as published by +// the Free Software Foundation, either version 3 of the License, or +// (at your option) any later version. +// +// The go-ethereum library is distributed in the hope that it will be useful, +// but WITHOUT ANY WARRANTY; without even the implied warranty of +// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +// GNU Lesser General Public License for more details. +// +// You should have received a copy of the GNU Lesser General Public License +// along with the go-ethereum library. If not, see . + +package les + +import ( + "context" + "encoding/binary" + "errors" + "fmt" + "io/ioutil" + "net/http" + "strings" + "time" + + "github.com/ethereum/go-ethereum/common" + "github.com/ethereum/go-ethereum/core" + "github.com/ethereum/go-ethereum/core/rawdb" + "github.com/ethereum/go-ethereum/core/types" + "github.com/ethereum/go-ethereum/crypto" + "github.com/ethereum/go-ethereum/ethdb" + "github.com/ethereum/go-ethereum/light" + "github.com/ethereum/go-ethereum/rlp" + "github.com/ethereum/go-ethereum/trie" +) + +// RequestByHTTP sends a HTTP request to les CDN and validate all replies with +// local header. If the passing context is canceled, then function returns. +// +// todo(rjl493456442) the code organization is really ugly, need big refactor. +func (r *BlockRequest) RequestByHTTP(ctx context.Context, url string /* todo auth credential */, db ethdb.Database) error { + var ( + txs types.Transactions + uncles []*types.Header + pending int + errch = make(chan error) + ) + // Retrieve our stored header and validate block content against it + header := rawdb.ReadHeader(db, r.Hash, r.Number) + if header == nil { + return errHeaderUnavailable + } + // Send a http request if tx set is not empty. + if header.TxHash != types.EmptyRootHash { + pending += 1 + go func() { + res, err := httpDo(ctx, fmt.Sprintf("%s/chain/0x%x/transactions", url, r.Hash)) + if err != nil { + errch <- err + return + } + if err := rlp.DecodeBytes(res, &txs); err != nil { + errch <- err + return + } + if header.TxHash != types.DeriveSha(txs) { + errch <- errTxHashMismatch + return + } + errch <- nil + }() + } + // Send a http request if uncle set is not empty. + if header.UncleHash != types.EmptyUncleHash { + pending += 1 + go func() { + res, err := httpDo(ctx, fmt.Sprintf("%s/chain/0x%x/uncles", url, r.Hash)) + if err != nil { + errch <- err + return + } + if err := rlp.DecodeBytes(res, &uncles); err != nil { + errch <- err + return + } + if header.UncleHash != types.CalcUncleHash(uncles) { + errch <- errUncleHashMismatch + return + } + errch <- nil + }() + } + // Wait all retrieve workers and return any error. + for i := 0; i < pending; i++ { + if err := <-errch; err != nil { + return err + } + } + r.Lock.Lock() + defer r.Lock.Unlock() + r.Txs, r.Uncles = txs, uncles + return nil +} + +// RequestByHTTP sends a HTTP request to les CDN and validate all replies with +// local header. If the passing context is canceled, then function returns. +// +// todo(rjl493456442) the code organization is really ugly, need big refactor. +func (r *ReceiptsRequest) RequestByHTTP(ctx context.Context, url string /* todo auth credential */, db ethdb.Database) error { + var receipts types.Receipts + if r.Header == nil { + r.Header = rawdb.ReadHeader(db, r.Hash, r.Number) + } + if r.Header.ReceiptHash != types.EmptyRootHash { + res, err := httpDo(ctx, fmt.Sprintf("%s/chain/0x%x/receipts", url, r.Hash)) + if err != nil { + return err + } + if err := rlp.DecodeBytes(res, &receipts); err != nil { + return err + } + if r.Header.ReceiptHash != types.DeriveSha(receipts) { + return errReceiptHashMismatch + } + } + r.Lock.Lock() + defer r.Lock.Unlock() + r.Receipts = receipts + return nil +} + +// RequestByHTTP sends a HTTP request to les CDN and validate all replies with +// local header. If the passing context is canceled, then function returns. +// +// todo(rjl493456442) the code organization is really ugly, need big refactor. +func (r *CodeRequest) RequestByHTTP(ctx context.Context, url string /* todo auth credential */, db ethdb.Database) error { + res, err := httpDo(ctx, fmt.Sprintf("%s/state/0x%x?target=%d&limit=%d&barrier=%d", url, r.Hash, 1, 1, 1)) + if err != nil { + return err + } + var nodes [][]byte + if err := rlp.DecodeBytes(res, &nodes); err != nil { + return err + } + if len(nodes) != 1 { + return errInvalidEntryCount + } + // Verify the data and store if checks out + if hash := crypto.Keccak256Hash(nodes[0]); r.Hash != hash { + return errDataHashMismatch + } + r.Lock.Lock() + defer r.Lock.Unlock() + r.Data = nodes[0] + return nil +} + +// RequestByHTTP sends a HTTP request to les CDN and validate all replies with +// local header. If the passing context is canceled, then function returns. +// +// todo(rjl493456442) the code organization is really ugly, need big refactor. +func (r *TrieRequest) RequestByHTTP(ctx context.Context, url string /* todo auth credential */, db ethdb.Database) error { + res, err := httpDo(ctx, fmt.Sprintf("%s/state/0x%x?target=%d&limit=%d&barrier=%d", url, r.MissNodeHash, 16, 256, 2)) + if err != nil { + return err + } + // Decode all received response. + var nodes [][]byte + if err := rlp.DecodeBytes(res, &nodes); err != nil { + return err + } + // Validate the received sub trie. + proofDb := light.NewNodeSet() + for _, node := range nodes { + proofDb.Put(crypto.Keccak256(node), node) + } + if err := trie.VerifyTrie(r.MissNodeHash, proofDb, proofDb.KeyCount()); err != nil { + return err + } + r.Lock.Lock() + defer r.Lock.Unlock() + r.Proof = proofDb // Pass the validation, set it the result. + return nil +} + +// RequestByHTTPV1 sends a HTTP request to les CDN and validate all replies with +// local cht root. If the passing context is canceled, then function returns. +// +// todo(rjl493456442) the code organization is really ugly, need big refactor. +func (r *ChtRequest) RequestByHTTPV1(ctx context.Context, url string /* todo auth credential */, db ethdb.Database) error { + res, err := httpDo(ctx, fmt.Sprintf("%s/misc/cht/0x%x?&number=%d", url, r.ChtRoot, r.BlockNum)) + if err != nil { + return err + } + var list light.NodeList + if err := rlp.DecodeBytes(res, &list); err != nil { + return err + } + // Validate and resolve the response + var ( + proof = list.NodeSet() + key = make([]byte, 8) + ) + reads := &readTraceDB{db: proof} + binary.BigEndian.PutUint64(key[:], r.BlockNum) + value, _, err := trie.VerifyProof(r.ChtRoot, key, reads) + if err != nil { + return err + } + if len(reads.reads) != proof.KeyCount() { + return errUselessNodes + } + // Decode concrete cht node. + var chtNode light.ChtNode + if err := rlp.DecodeBytes(value, &chtNode); err != nil { + return err + } + // Request corresponding header now. + res, err = httpDo(ctx, fmt.Sprintf("%s/chain/0x%x/header", url, chtNode.Hash)) + if err != nil { + return err + } + var header *types.Header + if err := rlp.DecodeBytes(res, &header); err != nil { + return err + } + if header.Hash() != chtNode.Hash { + return errInvalidHeader + } + r.Lock.Lock() + defer r.Lock.Unlock() + r.Header, r.Td, r.Proof = header, chtNode.Td, proof + return nil +} + +// RequestByHTTPV2 sends a HTTP request to les CDN and validate all replies with +// local cht root. If the passing context is canceled, then function returns. +// +// todo(rjl493456442) the code organization is really ugly, need big refactor. +func (r *ChtRequest) RequestByHTTP(ctx context.Context, url string /* todo auth credential */, db ethdb.Database) error { + var ( + key [8]byte + chtNode light.ChtNode + + table = rawdb.NewTable(db, light.ChtTablePrefix) + triedb = trie.NewDatabase(table) + ) + binary.BigEndian.PutUint64(key[:], r.BlockNum) + resolve := func() (error, []*trie.TrieTrace) { + t, err := trie.NewTraceTrie(r.ChtRoot, triedb, &trie.TraceConfig{RecordHash: true, RecordPath: true}) + if err != nil { + return err, t.GetTraces() + } + blob, err := t.TryGet(key[:]) + if err != nil { + return err, t.GetTraces() + } + err = rlp.DecodeBytes(blob, &chtNode) + if err != nil { + return err, nil + } + return nil, nil + } + for { + err, traces := resolve() + if err == nil { + break + } + missError, ok := err.(*trie.MissingNodeError) + if !ok { + return err + } + levels := light.MaxTileLevels(r.Config.ChtSize) + var nodeHash common.Hash + switch { + case missError.NodeHash == r.ChtRoot: + nodeHash = r.ChtRoot + case (16-len(missError.Path))/2 >= levels: + nodeHash = r.ChtRoot + case ((16-len(missError.Path))-1)%2 == 0: + nodeHash = missError.NodeHash + default: + nodeHash = traces[len(traces)-1].Hash + } + // Send http request to fetch the missing tile. + res, err := httpDo(ctx, fmt.Sprintf("%s/misc/chtv2/0x%x", url, nodeHash)) + if err != nil { + return err + } + var tile [][]byte + err = rlp.DecodeBytes(res, &tile) + if err != nil { + return err + } + // Validate the received sub trie. + proofDb := light.NewNodeSet() + for _, node := range tile { + proofDb.Put(crypto.Keccak256(node), node) + } + if err := trie.VerifyTrie(nodeHash, proofDb, proofDb.KeyCount()); err != nil { + return err + } + proofDb.Store(table) // Push all verified nodes into the disk + } + if chtNode.Hash == (common.Hash{}) { + return errors.New("failed to retrieve cht node") + } + // Request corresponding header now. + res, err := httpDo(ctx, fmt.Sprintf("%s/chain/0x%x/header", url, chtNode.Hash)) + if err != nil { + return err + } + var header *types.Header + if err := rlp.DecodeBytes(res, &header); err != nil { + return err + } + if header.Hash() != chtNode.Hash { + return errInvalidHeader + } + r.Lock.Lock() + defer r.Lock.Unlock() + r.Header, r.Td = header, chtNode.Td + return nil +} + +// RequestByHTTP sends a HTTP request to les CDN and validate all replies with +// local bloom trie root. If the passing context is canceled, then function returns. +// +// todo(rjl493456442) the code organization is really ugly, need big refactor. +func (r *BloomRequest) RequestByHTTP(ctx context.Context, url string /* todo auth credential */, db ethdb.Database) error { + // Short circuit if request is empty + if len(r.SectionList) == 0 { + return nil + } + var ( + errch = make(chan error) + bits = make([][]byte, len(r.SectionList)) + proofs = make([]*light.NodeSet, len(r.SectionList)) + ) + for index, section := range r.SectionList { + go func(index int, section uint64) { + res, err := httpDo(ctx, fmt.Sprintf("%s/misc/bloomtrie/0x%x?bit=%d§ion=%d", url, r.BloomTrieRoot, r.BitIndex, section)) + if err != nil { + errch <- err + return + } + var list light.NodeList + if err := rlp.DecodeBytes(res, &list); err != nil { + errch <- err + return + } + // Validate and resolve the response + var ( + proof = list.NodeSet() + key = make([]byte, 10) + ) + reads := &readTraceDB{db: proof} + binary.BigEndian.PutUint16(key[:2], uint16(r.BitIndex)) + binary.BigEndian.PutUint64(key[2:], section) + value, _, err := trie.VerifyProof(r.BloomTrieRoot, key, reads) + if err != nil { + errch <- err + return + } + if len(reads.reads) != proof.KeyCount() { + errch <- errUselessNodes + } + bits[index], proofs[index] = value, proof + errch <- nil + }(index, section) + } + // Return error if any of sub request failed. + for i := 0; i < len(r.SectionList); i++ { + if err := <-errch; err != nil { + return err + } + } + // Mix all proofs into single one + mix := light.NewNodeSet() + for _, proof := range proofs { + proof.Store(mix) + } + r.Lock.Lock() + defer r.Lock.Unlock() + r.BloomBits, r.Proofs = bits, mix + return nil +} + +// RequestByHTTP sends a HTTP request to les CDN. Note for txstatus request +// there is no way to meaningfully validate the reply. If the passing context +// is canceled, then function returns. +// +// todo(rjl493456442) the code organization is really ugly, need big refactor. +func (r *TxStatusRequest) RequestByHTTP(ctx context.Context, url string /* todo auth credential */, db ethdb.Database) error { + var ( + hashes = r.Hashes + errch = make(chan error) + status = make([]light.TxStatus, len(hashes)) + ) + for index, hash := range hashes { + go func(index int, hash common.Hash) { + res, err := httpDo(ctx, fmt.Sprintf("%s/chain/0x%x/txstatus", url, hash)) + if err != nil { + errch <- err + return + } + var lookup rawdb.LegacyTxLookupEntry + if err := rlp.DecodeBytes(res, &lookup); err != nil { + errch <- err + return + } + status[index] = light.TxStatus{ + Status: core.TxStatusIncluded, + Lookup: &lookup, + } + errch <- nil + }(index, hash) + } + for i := 0; i < len(hashes); i++ { + if err := <-errch; err != nil { + return err + } + } + r.Lock.Lock() + defer r.Lock.Unlock() + r.Status = status + return nil +} + +// httpDo sends a http request based on the given URL, extracts the +// response and return. +func httpDo(ctx context.Context, url string) ([]byte, error) { + req, err := http.NewRequest("GET", url, nil) + if err != nil { + return nil, err + } + req = req.WithContext(ctx) + + res, err := http.DefaultClient.Do(req) + if err != nil { + return nil, err + } + blob, err := ioutil.ReadAll(res.Body) + if err != nil { + return nil, err + } + res.Body.Close() + return blob, nil +} + +type httpRetriever struct { + url string + trigger time.Duration + db ethdb.Database +} + +func newHTTPRetriever(url string, trigger time.Duration, db ethdb.Database) *httpRetriever { + // If external CDN is not configured, return an nil instance. + if url == "" { + return nil + } + // validate the given host url, adjust it if necessary. + for strings.HasSuffix(url, "/") { + url = url[:len(url)-1] + } + return &httpRetriever{ + url: url, + trigger: trigger, + db: db, + } +} + +// retrieve sends a request to specified CDN network and waits for an answer +// that is delivered and successfully validated by the validator callback. +// It returns when a valid answer is delivered or the context is cancelled. +func (r *httpRetriever) retrieve(ctx context.Context, req LesOdrRequest) error { + select { + case <-ctx.Done(): + return nil + case <-time.NewTimer(r.trigger).C: + } + return req.RequestByHTTP(ctx, r.url, r.db) +} diff --git a/les/http_retriever_test.go b/les/http_retriever_test.go new file mode 100644 index 0000000000..92d4f3e564 --- /dev/null +++ b/les/http_retriever_test.go @@ -0,0 +1,94 @@ +// Copyright 2019 The go-ethereum Authors +// This file is part of the go-ethereum library. +// +// The go-ethereum library is free software: you can redistribute it and/or modify +// it under the terms of the GNU Lesser General Public License as published by +// the Free Software Foundation, either version 3 of the License, or +// (at your option) any later version. +// +// The go-ethereum library is distributed in the hope that it will be useful, +// but WITHOUT ANY WARRANTY; without even the implied warranty of +// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +// GNU Lesser General Public License for more details. +// +// You should have received a copy of the GNU Lesser General Public License +// along with the go-ethereum library. If not, see . + +package les + +import ( + "context" + "testing" + "time" + + "github.com/ethereum/go-ethereum/common" + "github.com/ethereum/go-ethereum/core" + "github.com/ethereum/go-ethereum/lescdn" + "github.com/ethereum/go-ethereum/light" +) + +func TestHTTPRequest(t *testing.T) { + config := light.TestServerIndexerConfig + + waitIndexers := func(cIndexer, bIndexer, btIndexer *core.ChainIndexer) { + for { + cs, _, _ := cIndexer.Sections() + bts, _, _ := btIndexer.Sections() + if cs >= 1 && bts >= 1 { + break + } + time.Sleep(10 * time.Millisecond) + } + } + // Generate 512+4 blocks (totally 1 CHT sections) + server, client, tearDown := newClientServerEnv(t, int(config.ChtSize+config.ChtConfirms), 3, waitIndexers, nil, 0, false, true) + defer tearDown() + + // Start a CDN service for request handling + lesCDN := lescdn.New(server.backend.Blockchain()) + lesCDN.Start(nil) + + // Prepare CDN requests + blockOne := server.backend.Blockchain().GetBlockByNumber(1) + headerSection := server.backend.Blockchain().GetHeaderByNumber(config.BloomTrieSize - 1) + chtRoot := light.GetChtRoot(server.db, 0, headerSection.Hash()) + bloomTrieRoot := light.GetBloomTrieRoot(server.db, 0, headerSection.Hash()) + + // Prepare txstatus requests + var hashes []common.Hash + txs := blockOne.Transactions() + for _, tx := range txs { + hashes = append(hashes, tx.Hash()) + } + + // Prepare state reqeusts + blockHead := server.backend.Blockchain().CurrentBlock() + state, _ := server.backend.Blockchain().State() + codeHash := state.GetCodeHash(registrarAddr) + + stateTrieID := &light.TrieID{ + BlockHash: blockHead.Hash(), + BlockNumber: blockHead.NumberU64(), + } + accTrieID := &light.TrieID{ + BlockHash: blockHead.Hash(), + BlockNumber: blockHead.NumberU64(), + AccKey: registrarAddr.Bytes(), + } + var requests = []light.OdrRequest{ + &light.BlockRequest{Hash: blockOne.Hash(), Number: blockOne.NumberU64()}, + &light.ReceiptsRequest{Hash: blockOne.Hash(), Number: blockOne.NumberU64(), Header: blockOne.Header()}, + &light.TrieRequest{Id: stateTrieID, MissNodeHash: blockHead.Root()}, + &light.CodeRequest{Hash: codeHash, Id: accTrieID}, + &light.ChtRequest{ChtRoot: chtRoot, ChtNum: 0, Config: server.handler.server.iConfig, BlockNum: headerSection.Number.Uint64()}, + &light.BloomRequest{SectionList: []uint64{0}, BitIndex: 0, Config: server.handler.server.iConfig, BloomTrieNum: 0, BloomTrieRoot: bloomTrieRoot}, + &light.TxStatusRequest{Hashes: hashes}, + } + for _, request := range requests { + req := LesRequest(request) + err := req.RequestByHTTP(context.Background(), "http://localhost:8548", client.db) + if err != nil { + t.Fatalf("Failed to retrieve data via CDN: %v", err) + } + } +} diff --git a/les/odr.go b/les/odr.go index f8469cc103..901d9f5885 100644 --- a/les/odr.go +++ b/les/odr.go @@ -18,13 +18,23 @@ package les import ( "context" + "errors" + "sync" "time" - "github.com/ethereum/go-ethereum/common/mclock" + "github.com/ethereum/go-ethereum/common" "github.com/ethereum/go-ethereum/core" "github.com/ethereum/go-ethereum/ethdb" "github.com/ethereum/go-ethereum/light" "github.com/ethereum/go-ethereum/log" + "github.com/ethereum/go-ethereum/metrics" +) + +var ( + httpRequestGauge = metrics.NewRegisteredGauge("les/client/req/http/count", nil) + httpRequestTimer = metrics.NewRegisteredTimer("les/cleint/req/http/duration", nil) + p2pRequestGauge = metrics.NewRegisteredGauge("les/client/req/p2p/count", nil) + p2pRequestTimer = metrics.NewRegisteredTimer("les/cleint/req/p2p/duration", nil) ) // LesOdr implements light.OdrBackend @@ -32,15 +42,17 @@ type LesOdr struct { db ethdb.Database indexerConfig *light.IndexerConfig chtIndexer, bloomTrieIndexer, bloomIndexer *core.ChainIndexer - retriever *retrieveManager + p2pRtr *p2pRetriever + httpRtr *httpRetriever stop chan struct{} } -func NewLesOdr(db ethdb.Database, config *light.IndexerConfig, retriever *retrieveManager) *LesOdr { +func NewLesOdr(db ethdb.Database, config *light.IndexerConfig, p2prtr *p2pRetriever, httprtr *httpRetriever) *LesOdr { return &LesOdr{ db: db, indexerConfig: config, - retriever: retriever, + p2pRtr: p2prtr, + httpRtr: httprtr, stop: make(chan struct{}), } } @@ -100,35 +112,89 @@ type Msg struct { // Retrieve tries to fetch an object from the LES network. // If the network retrieval was successful, it stores the object in local db. -func (odr *LesOdr) Retrieve(ctx context.Context, req light.OdrRequest) (err error) { - lreq := LesRequest(req) +func (odr *LesOdr) Retrieve(ctx context.Context, req light.OdrRequest) error { + defer func(start time.Time) { + log.Debug("Retrieved data", "elapsed", common.PrettyDuration(time.Since(start))) + }(time.Now()) - reqID := genReqID() - rq := &distReq{ - getCost: func(dp distPeer) uint64 { - return lreq.GetCost(dp.(*serverPeer)) - }, - canSend: func(dp distPeer) bool { - p := dp.(*serverPeer) - if !p.onlyAnnounce { - return lreq.CanSend(p) - } - return false - }, - request: func(dp distPeer) func() { - p := dp.(*serverPeer) - cost := lreq.GetCost(p) - p.fcServer.QueuedRequest(reqID, cost) - return func() { lreq.Request(reqID, p) } - }, + var ( + count int + wg sync.WaitGroup + errorCh = make(chan error, 2) + + ctx1, cancelFn1 = context.WithCancel(ctx) + ctx2, cancelFn2 = context.WithCancel(ctx) + ) + // retrieve invokes given retrival action, update metrics no matter successful + // or not, return error via buffered channel. + retrieve := func(method string, action func() error, successCallback func(), gauge metrics.Gauge, timer metrics.Timer) { + defer wg.Done() + + defer func(start time.Time) { + gauge.Update(gauge.Value() + 1) + timer.UpdateSince(start) + log.Debug("Retrieved data", "method", method, "elasped", common.PrettyDuration(time.Since(start))) + }(time.Now()) + + err := action() + if err == nil { + successCallback() + } + errorCh <- err } - sent := mclock.Now() - if err = odr.retriever.retrieve(ctx, reqID, rq, func(p distPeer, msg *Msg) error { return lreq.Validate(odr.db, msg) }, odr.stop); err == nil { - // retrieved from network, store in db - req.StoreResult(odr.db) - requestRTT.Update(time.Duration(mclock.Now() - sent)) - } else { - log.Debug("Failed to retrieve data from network", "err", err) + // If p2p retriever is available, spin it up. + if odr.p2pRtr != nil { + wg.Add(1) + count += 1 + + reqID := genReqID() + rq := &distReq{ + getCost: func(dp distPeer) uint64 { return LesRequest(req).GetCost(dp.(*serverPeer)) }, + canSend: func(dp distPeer) bool { + p := dp.(*serverPeer) + if !p.onlyAnnounce { + return LesRequest(req).CanSend(p) + } + return false + }, + request: func(dp distPeer) func() { + p := dp.(*serverPeer) + cost := LesRequest(req).GetCost(p) + p.fcServer.QueuedRequest(reqID, cost) + return func() { LesRequest(req).Request(reqID, p) } + }, + } + go retrieve("p2p", func() error { + return odr.p2pRtr.retrieve(ctx1, reqID, rq, func(p distPeer, msg *Msg) error { return LesRequest(req).Validate(odr.db, msg) }, odr.stop) + }, func() { + cancelFn2() // Explicitly stop http retriever + }, p2pRequestGauge, p2pRequestTimer) } - return + // If http retriever is available, spin it up. + if odr.httpRtr != nil { + wg.Add(1) + count += 1 + go retrieve("http", func() error { + return odr.httpRtr.retrieve(ctx2, LesRequest(req)) + }, func() { + cancelFn1() // Explicitly stop p2p retriever + }, httpRequestGauge, httpRequestTimer) + } + if count == 0 { + return errors.New("no available retriever") + } + // Waiting the response. If any returned error is nil, regard data + // retreval successfully. + wg.Wait() + + var mix string + for i := 0; i < count; i++ { + if err := <-errorCh; err != nil { + mix = mix + ":" + err.Error() + } else { + req.StoreResult(odr.db) // retrieved from network, store in db + return nil + } + } + return errors.New(mix) } diff --git a/les/odr_requests.go b/les/odr_requests.go index 146da2213c..0572740f66 100644 --- a/les/odr_requests.go +++ b/les/odr_requests.go @@ -17,6 +17,7 @@ package les import ( + "context" "encoding/binary" "errors" "fmt" @@ -36,6 +37,7 @@ var ( errInvalidMessageType = errors.New("invalid message type") errInvalidEntryCount = errors.New("invalid number of response entries") errHeaderUnavailable = errors.New("header unavailable") + errInvalidHeader = errors.New("invalid header") errTxHashMismatch = errors.New("transaction hash mismatch") errUncleHashMismatch = errors.New("uncle hash mismatch") errReceiptHashMismatch = errors.New("receipt hash mismatch") @@ -50,6 +52,7 @@ type LesOdrRequest interface { CanSend(*serverPeer) bool Request(uint64, *serverPeer) error Validate(ethdb.Database, *Msg) error + RequestByHTTP(context.Context, string, ethdb.Database) error } func LesRequest(req light.OdrRequest) LesOdrRequest { @@ -120,12 +123,9 @@ func (r *BlockRequest) Validate(db ethdb.Database, msg *Msg) error { if header.UncleHash != types.CalcUncleHash(body.Uncles) { return errUncleHashMismatch } - // Validations passed, encode and store RLP - data, err := rlp.EncodeToBytes(body) - if err != nil { - return err - } - r.Rlp = data + r.Lock.Lock() + defer r.Lock.Unlock() + r.Txs, r.Uncles = body.Transactions, body.Uncles return nil } @@ -175,7 +175,8 @@ func (r *ReceiptsRequest) Validate(db ethdb.Database, msg *Msg) error { if r.Header.ReceiptHash != types.DeriveSha(receipt) { return errReceiptHashMismatch } - // Validations passed, store and return + r.Lock.Lock() + defer r.Lock.Unlock() r.Receipts = receipt return nil } @@ -231,6 +232,8 @@ func (r *TrieRequest) Validate(db ethdb.Database, msg *Msg) error { if len(reads.reads) != nodeSet.KeyCount() { return errUselessNodes } + r.Lock.Lock() + defer r.Lock.Unlock() r.Proof = nodeSet return nil } @@ -284,6 +287,8 @@ func (r *CodeRequest) Validate(db ethdb.Database, msg *Msg) error { if hash := crypto.Keccak256Hash(data); r.Hash != hash { return errDataHashMismatch } + r.Lock.Lock() + defer r.Lock.Unlock() r.Data = data return nil } @@ -397,10 +402,11 @@ func (r *ChtRequest) Validate(db ethdb.Database, msg *Msg) error { } } // Verifications passed, store and return + r.Lock.Lock() + defer r.Lock.Unlock() r.Header = header r.Proof = nodeSet r.Td = node.Td // For untrusted request, td here is nil, todo improve the les/2 protocol - return nil } @@ -414,7 +420,7 @@ type BloomRequest light.BloomRequest // GetCost returns the cost of the given ODR request according to the serving // peer's cost table (implementation of LesOdrRequest) func (r *BloomRequest) GetCost(peer *serverPeer) uint64 { - return peer.getRequestCost(GetHelperTrieProofsMsg, len(r.SectionIndexList)) + return peer.getRequestCost(GetHelperTrieProofsMsg, len(r.SectionList)) } // CanSend tells if a certain peer is suitable for serving the given request @@ -430,13 +436,13 @@ func (r *BloomRequest) CanSend(peer *serverPeer) bool { // Request sends an ODR request to the LES network (implementation of LesOdrRequest) func (r *BloomRequest) Request(reqID uint64, peer *serverPeer) error { - peer.Log().Debug("Requesting BloomBits", "bloomTrie", r.BloomTrieNum, "bitIdx", r.BitIdx, "sections", r.SectionIndexList) - reqs := make([]HelperTrieReq, len(r.SectionIndexList)) + peer.Log().Debug("Requesting BloomBits", "bloomTrie", r.BloomTrieNum, "bitIdx", r.BitIndex, "sections", r.SectionList) + reqs := make([]HelperTrieReq, len(r.SectionList)) var encNumber [10]byte - binary.BigEndian.PutUint16(encNumber[:2], uint16(r.BitIdx)) + binary.BigEndian.PutUint16(encNumber[:2], uint16(r.BitIndex)) - for i, sectionIdx := range r.SectionIndexList { + for i, sectionIdx := range r.SectionList { binary.BigEndian.PutUint64(encNumber[2:], sectionIdx) reqs[i] = HelperTrieReq{ Type: htBloomBits, @@ -451,7 +457,7 @@ func (r *BloomRequest) Request(reqID uint64, peer *serverPeer) error { // returns true and stores results in memory if the message was a valid reply // to the request (implementation of LesOdrRequest) func (r *BloomRequest) Validate(db ethdb.Database, msg *Msg) error { - log.Debug("Validating BloomBits", "bloomTrie", r.BloomTrieNum, "bitIdx", r.BitIdx, "sections", r.SectionIndexList) + log.Debug("Validating BloomBits", "bloomTrie", r.BloomTrieNum, "bitIdx", r.BitIndex, "sections", r.SectionList) // Ensure we have a correct message with a single proof element if msg.MsgType != MsgHelperTrieProofs { @@ -462,13 +468,13 @@ func (r *BloomRequest) Validate(db ethdb.Database, msg *Msg) error { nodeSet := proofs.NodeSet() reads := &readTraceDB{db: nodeSet} - r.BloomBits = make([][]byte, len(r.SectionIndexList)) + r.BloomBits = make([][]byte, len(r.SectionList)) // Verify the proofs var encNumber [10]byte - binary.BigEndian.PutUint16(encNumber[:2], uint16(r.BitIdx)) + binary.BigEndian.PutUint16(encNumber[:2], uint16(r.BitIndex)) - for i, idx := range r.SectionIndexList { + for i, idx := range r.SectionList { binary.BigEndian.PutUint64(encNumber[2:], idx) value, _, err := trie.VerifyProof(r.BloomTrieRoot, encNumber[:], reads) if err != nil { @@ -518,6 +524,8 @@ func (r *TxStatusRequest) Validate(db ethdb.Database, msg *Msg) error { if len(status) != len(r.Hashes) { return errInvalidEntryCount } + r.Lock.Lock() + defer r.Lock.Unlock() r.Status = status return nil } diff --git a/les/retrieve.go b/les/p2p_retrieve.go similarity index 93% rename from les/retrieve.go rename to les/p2p_retrieve.go index 5fa68b7456..265f3efdf1 100644 --- a/les/retrieve.go +++ b/les/p2p_retrieve.go @@ -34,9 +34,12 @@ var ( hardRequestTimeout = time.Second * 10 ) -// retrieveManager is a layer on top of requestDistributor which takes care of -// matching replies by request ID and handles timeouts and resends if necessary. -type retrieveManager struct { +// p2pRetriever is a layer on top of requestDistributor which sends the request +// to p2p network, takes care of matching replies by request ID and handles timeouts. +// +// Besides, if there exists a validation function for request, then retriever will +// run the function and resend the request to other peers if the reply is invalid. +type p2pRetriever struct { dist *requestDistributor peers *serverPeerSet serverPool peerSelector @@ -53,9 +56,9 @@ type peerSelector interface { adjustResponseTime(*poolEntry, time.Duration, bool) } -// sentReq represents a request sent and tracked by retrieveManager +// sentReq represents a request sent and tracked by p2pRetriever type sentReq struct { - rm *retrieveManager + rm *p2pRetriever req *distReq id uint64 validate validatorFunc @@ -99,8 +102,8 @@ const ( ) // newRetrieveManager creates the retrieve manager -func newRetrieveManager(peers *serverPeerSet, dist *requestDistributor, serverPool peerSelector) *retrieveManager { - return &retrieveManager{ +func newRetrieveManager(peers *serverPeerSet, dist *requestDistributor, serverPool peerSelector) *p2pRetriever { + return &p2pRetriever{ peers: peers, dist: dist, serverPool: serverPool, @@ -112,7 +115,7 @@ func newRetrieveManager(peers *serverPeerSet, dist *requestDistributor, serverPo // that is delivered through the deliver function and successfully validated by the // validator callback. It returns when a valid answer is delivered or the context is // cancelled. -func (rm *retrieveManager) retrieve(ctx context.Context, reqID uint64, req *distReq, val validatorFunc, shutdown chan struct{}) error { +func (rm *p2pRetriever) retrieve(ctx context.Context, reqID uint64, req *distReq, val validatorFunc, shutdown chan struct{}) error { sentReq := rm.sendReq(reqID, req, val) select { case <-sentReq.stopCh: @@ -126,7 +129,7 @@ func (rm *retrieveManager) retrieve(ctx context.Context, reqID uint64, req *dist // sendReq starts a process that keeps trying to retrieve a valid answer for a // request from any suitable peers until stopped or succeeded. -func (rm *retrieveManager) sendReq(reqID uint64, req *distReq, val validatorFunc) *sentReq { +func (rm *p2pRetriever) sendReq(reqID uint64, req *distReq, val validatorFunc) *sentReq { r := &sentReq{ rm: rm, req: req, @@ -163,7 +166,7 @@ func (rm *retrieveManager) sendReq(reqID uint64, req *distReq, val validatorFunc } // deliver is called by the LES protocol manager to deliver reply messages to waiting requests -func (rm *retrieveManager) deliver(peer distPeer, msg *Msg) error { +func (rm *p2pRetriever) deliver(peer distPeer, msg *Msg) error { rm.lock.RLock() req, ok := rm.sentReqs[msg.ReqID] rm.lock.RUnlock() @@ -176,7 +179,7 @@ func (rm *retrieveManager) deliver(peer distPeer, msg *Msg) error { // frozen is called by the LES protocol manager when a server has suspended its service and we // should not expect an answer for the requests already sent there -func (rm *retrieveManager) frozen(peer distPeer) { +func (rm *p2pRetriever) frozen(peer distPeer) { rm.lock.RLock() defer rm.lock.RUnlock() diff --git a/les/server.go b/les/server.go index f72f31321a..eb447d42a9 100644 --- a/les/server.go +++ b/les/server.go @@ -125,6 +125,7 @@ func NewLesServer(e *eth.Ethereum, config *eth.Config) (*LesServer, error) { "chtroot", checkpoint.CHTRoot, "bloomroot", checkpoint.BloomRoot) } srv.chtIndexer.Start(e.BlockChain()) + return srv, nil } @@ -213,7 +214,8 @@ func (s *LesServer) SetBloomBitsIndexer(bloomIndexer *core.ChainIndexer) { bloomIndexer.AddChildIndexer(s.bloomTrieIndexer) } -// SetClient sets the rpc client and starts running checkpoint contract if it is not yet watched. +// SetContractBackend sets the rpc client and starts running checkpoint contract +// if it is not yet watched. func (s *LesServer) SetContractBackend(backend bind.ContractBackend) { if s.oracle == nil { return diff --git a/les/serverpool.go b/les/serverpool.go index ec99a2d982..2d98c012af 100644 --- a/les/serverpool.go +++ b/les/serverpool.go @@ -28,6 +28,7 @@ import ( "time" "github.com/ethereum/go-ethereum/common/mclock" + "github.com/ethereum/go-ethereum/common/selector" "github.com/ethereum/go-ethereum/crypto" "github.com/ethereum/go-ethereum/ethdb" "github.com/ethereum/go-ethereum/log" @@ -129,7 +130,7 @@ type serverPool struct { adjustStats chan poolStatAdjust knownQueue, newQueue poolEntryQueue - knownSelect, newSelect *weightedRandomSelect + knownSelect, newSelect *selector.WeightedRandomSelect knownSelected, newSelected int fastDiscover bool connCh chan *connReq @@ -152,8 +153,8 @@ func newServerPool(db ethdb.Database, ulcServers []string) *serverPool { disconnCh: make(chan *disconnReq), registerCh: make(chan *registerReq), closeCh: make(chan struct{}), - knownSelect: newWeightedRandomSelect(), - newSelect: newWeightedRandomSelect(), + knownSelect: selector.NewWeightedRandomSelect(), + newSelect: selector.NewWeightedRandomSelect(), fastDiscover: true, trustedNodes: parseTrustedNodes(ulcServers), } @@ -402,8 +403,8 @@ func (pool *serverPool) eventLoop() { entry.lastConnected = addr entry.addr = make(map[string]*poolEntryAddress) entry.addr[addr.strKey()] = addr - entry.addrSelect = *newWeightedRandomSelect() - entry.addrSelect.update(addr) + entry.addrSelect = *selector.NewWeightedRandomSelect() + entry.addrSelect.Update(addr) req.result <- entry } @@ -459,7 +460,7 @@ func (pool *serverPool) findOrNewNode(node *enode.Node) *poolEntry { entry = &poolEntry{ node: node, addr: make(map[string]*poolEntryAddress), - addrSelect: *newWeightedRandomSelect(), + addrSelect: *selector.NewWeightedRandomSelect(), shortRetry: shortRetryCnt, } pool.entries[node.ID()] = entry @@ -477,7 +478,7 @@ func (pool *serverPool) findOrNewNode(node *enode.Node) *poolEntry { entry.addr[addr.strKey()] = addr } addr.lastSeen = now - entry.addrSelect.update(addr) + entry.addrSelect.Update(addr) if !entry.known { pool.newQueue.setLatest(entry) } @@ -505,7 +506,7 @@ func (pool *serverPool) loadNodes() { pool.entries[e.node.ID()] = e if pool.trustedNodes[e.node.ID()] == nil { pool.knownQueue.setLatest(e) - pool.knownSelect.update((*knownEntry)(e)) + pool.knownSelect.Update((*knownEntry)(e)) } } } @@ -556,8 +557,8 @@ func (pool *serverPool) saveNodes() { // Note that it is called by the new/known queues from which the entry has already // been removed so removing it from the queues is not necessary. func (pool *serverPool) removeEntry(entry *poolEntry) { - pool.newSelect.remove((*discoveredEntry)(entry)) - pool.knownSelect.remove((*knownEntry)(entry)) + pool.newSelect.Remove((*discoveredEntry)(entry)) + pool.knownSelect.Remove((*knownEntry)(entry)) entry.removed = true delete(pool.entries, entry.node.ID()) } @@ -586,8 +587,8 @@ func (pool *serverPool) setRetryDial(entry *poolEntry) { // updateCheckDial is called when an entry can potentially be dialed again. It updates // its selection weights and checks if new dials can/should be made. func (pool *serverPool) updateCheckDial(entry *poolEntry) { - pool.newSelect.update((*discoveredEntry)(entry)) - pool.knownSelect.update((*knownEntry)(entry)) + pool.newSelect.Update((*discoveredEntry)(entry)) + pool.knownSelect.Update((*knownEntry)(entry)) pool.checkDial() } @@ -596,7 +597,7 @@ func (pool *serverPool) updateCheckDial(entry *poolEntry) { func (pool *serverPool) checkDial() { fillWithKnownSelects := !pool.fastDiscover for pool.knownSelected < targetKnownSelect { - entry := pool.knownSelect.choose() + entry := pool.knownSelect.Choose() if entry == nil { fillWithKnownSelects = false break @@ -604,7 +605,7 @@ func (pool *serverPool) checkDial() { pool.dial((*poolEntry)(entry.(*knownEntry)), true) } for pool.knownSelected+pool.newSelected < targetServerCount { - entry := pool.newSelect.choose() + entry := pool.newSelect.Choose() if entry == nil { break } @@ -615,7 +616,7 @@ func (pool *serverPool) checkDial() { // is over, we probably won't find more in the near future so select more // known entries if possible for pool.knownSelected < targetServerCount { - entry := pool.knownSelect.choose() + entry := pool.knownSelect.Choose() if entry == nil { break } @@ -636,7 +637,7 @@ func (pool *serverPool) dial(entry *poolEntry, knownSelected bool) { } else { pool.newSelected++ } - addr := entry.addrSelect.choose().(*poolEntryAddress) + addr := entry.addrSelect.Choose().(*poolEntryAddress) log.Debug("Dialing new peer", "lesaddr", entry.node.ID().String()+"@"+addr.strKey(), "set", len(entry.addr), "known", knownSelected) entry.dialed = addr go func() { @@ -684,7 +685,7 @@ type poolEntry struct { addr map[string]*poolEntryAddress node *enode.Node lastConnected, dialed *poolEntryAddress - addrSelect weightedRandomSelect + addrSelect selector.WeightedRandomSelect lastDiscovered mclock.AbsTime known, knownSelected, trusted bool @@ -734,8 +735,8 @@ func (e *poolEntry) DecodeRLP(s *rlp.Stream) error { e.node = enode.NewV4(pubkey, entry.IP, int(entry.Port), int(entry.Port)) e.addr = make(map[string]*poolEntryAddress) e.addr[addr.strKey()] = addr - e.addrSelect = *newWeightedRandomSelect() - e.addrSelect.update(addr) + e.addrSelect = *selector.NewWeightedRandomSelect() + e.addrSelect.Update(addr) e.lastConnected = addr e.connectStats = entry.CStat e.delayStats = entry.DStat diff --git a/les/test_helper.go b/les/test_helper.go index d9ffe32db2..fa9aadb206 100644 --- a/les/test_helper.go +++ b/les/test_helper.go @@ -209,8 +209,8 @@ func newTestClientHandler(backend *backends.SimulatedBackend, odr *LesOdr, index closeCh: make(chan struct{}), }, peers: peers, - reqDist: odr.retriever.dist, - retriever: odr.retriever, + reqDist: odr.p2pRtr.dist, + p2pRtr: odr.p2pRtr, odr: odr, engine: engine, blockchain: chain, @@ -493,7 +493,7 @@ func newClientServerEnv(t *testing.T, blocks int, protocol int, callback indexer } dist := newRequestDistributor(speers, clock) rm := newRetrieveManager(speers, dist, nil) - odr := NewLesOdr(cdb, light.TestClientIndexerConfig, rm) + odr := NewLesOdr(cdb, light.TestClientIndexerConfig, rm, nil) sindexers := testIndexers(sdb, nil, light.TestServerIndexerConfig) cIndexers := testIndexers(cdb, odr, light.TestClientIndexerConfig) diff --git a/les/txrelay.go b/les/txrelay.go index 595c4d5808..4b225abe24 100644 --- a/les/txrelay.go +++ b/les/txrelay.go @@ -38,10 +38,10 @@ type lesTxRelay struct { lock sync.RWMutex stop chan struct{} - retriever *retrieveManager + retriever *p2pRetriever } -func newLesTxRelay(ps *serverPeerSet, retriever *retrieveManager) *lesTxRelay { +func newLesTxRelay(ps *serverPeerSet, retriever *p2pRetriever) *lesTxRelay { r := &lesTxRelay{ txSent: make(map[common.Hash]*ltrInfo), txPending: make(map[common.Hash]struct{}), diff --git a/lescdn/chain.go b/lescdn/chain.go index 1a71dc1362..014a7ae299 100644 --- a/lescdn/chain.go +++ b/lescdn/chain.go @@ -1,3 +1,19 @@ +// Copyright 2019 The go-ethereum Authors +// This file is part of the go-ethereum library. +// +// The go-ethereum library is free software: you can redistribute it and/or modify +// it under the terms of the GNU Lesser General Public License as published by +// the Free Software Foundation, either version 3 of the License, or +// (at your option) any later version. +// +// The go-ethereum library is distributed in the hope that it will be useful, +// but WITHOUT ANY WARRANTY; without even the implied warranty of +// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +// GNU Lesser General Public License for more details. +// +// You should have received a copy of the GNU Lesser General Public License +// along with the go-ethereum library. If not, see . + package lescdn import ( @@ -6,6 +22,7 @@ import ( "github.com/ethereum/go-ethereum/common" "github.com/ethereum/go-ethereum/common/hexutil" + "github.com/ethereum/go-ethereum/log" ) // serveChain is responsible for serving HTTP requests for chain data. @@ -30,7 +47,8 @@ func (s *Service) serveChainItem(hash common.Hash) http.Handler { case "header": // Retrieve the header and attempt to return it if header := s.chain.GetHeaderByHash(hash); header != nil { - reply(w, header) + replyAndCache(w, header) + log.Debug("Served chain item", "type", "header", "hash", hash) return } // Header not found, error out appropriately @@ -40,7 +58,8 @@ func (s *Service) serveChainItem(hash common.Hash) http.Handler { case "uncles": // Retrieve the block and attempt to return the uncles if block := s.chain.GetBlockByHash(hash); block != nil { - reply(w, block.Uncles()) + replyAndCache(w, block.Uncles()) + log.Debug("Served chain item", "type", "uncles", "hash", hash) return } // Block not found, error out appropriately @@ -50,7 +69,8 @@ func (s *Service) serveChainItem(hash common.Hash) http.Handler { case "transactions": // Retrieve the block and attempt to return the transactions if block := s.chain.GetBlockByHash(hash); block != nil { - reply(w, block.Transactions()) + replyAndCache(w, block.Transactions()) + log.Debug("Served chain item", "type", "transactions", "hash", hash) return } // Block not found, error out appropriately @@ -60,12 +80,27 @@ func (s *Service) serveChainItem(hash common.Hash) http.Handler { case "receipts": // Retrieve the receipts and attempt to return them if receipts := s.chain.GetReceiptsByHash(hash); receipts != nil { - reply(w, receipts) + replyAndCache(w, receipts) + log.Debug("Served chain item", "type", "receipts", "hash", hash) return } // Receipts not found, error out appropriately http.Error(w, http.StatusText(http.StatusNotFound), http.StatusNotFound) return + + case "txstatus": + // Retrieve the tx lookup and attempt to return them + if lookup := s.chain.GetTransactionLookup(hash); lookup != nil { + reply(w, lookup) // Can't cache in HTTP layer, since it's still mutable. + log.Debug("Served chain item", "type", "txstatus", "hash", hash) + return + } + // Tx lookup not found, error out appropriately + // Note in theory we should check txpool for the pending transaction, + // But it will increase lots of pressure to txpool, and also the status + // of transaction in pool is not suitable to cache in CDN. + http.Error(w, http.StatusText(http.StatusNotFound), http.StatusNotFound) + return } http.Error(w, http.StatusText(http.StatusNotFound), http.StatusNotFound) }) diff --git a/lescdn/gzip.go b/lescdn/gzip.go index a21d4beb73..92396c8d4c 100644 --- a/lescdn/gzip.go +++ b/lescdn/gzip.go @@ -1,3 +1,19 @@ +// Copyright 2019 The go-ethereum Authors +// This file is part of the go-ethereum library. +// +// The go-ethereum library is free software: you can redistribute it and/or modify +// it under the terms of the GNU Lesser General Public License as published by +// the Free Software Foundation, either version 3 of the License, or +// (at your option) any later version. +// +// The go-ethereum library is distributed in the hope that it will be useful, +// but WITHOUT ANY WARRANTY; without even the implied warranty of +// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +// GNU Lesser General Public License for more details. +// +// You should have received a copy of the GNU Lesser General Public License +// along with the go-ethereum library. If not, see . + package lescdn import ( diff --git a/lescdn/lesleech/main.go b/lescdn/lesleech/main.go index 0c763a64bb..564a9d3f51 100644 --- a/lescdn/lesleech/main.go +++ b/lescdn/lesleech/main.go @@ -1,3 +1,19 @@ +// Copyright 2019 The go-ethereum Authors +// This file is part of the go-ethereum library. +// +// The go-ethereum library is free software: you can redistribute it and/or modify +// it under the terms of the GNU Lesser General Public License as published by +// the Free Software Foundation, either version 3 of the License, or +// (at your option) any later version. +// +// The go-ethereum library is distributed in the hope that it will be useful, +// but WITHOUT ANY WARRANTY; without even the implied warranty of +// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +// GNU Lesser General Public License for more details. +// +// You should have received a copy of the GNU Lesser General Public License +// along with the go-ethereum library. If not, see . + package main import ( diff --git a/lescdn/misc.go b/lescdn/misc.go new file mode 100644 index 0000000000..f2538672fd --- /dev/null +++ b/lescdn/misc.go @@ -0,0 +1,141 @@ +// Copyright 2019 The go-ethereum Authors +// This file is part of the go-ethereum library. +// +// The go-ethereum library is free software: you can redistribute it and/or modify +// it under the terms of the GNU Lesser General Public License as published by +// the Free Software Foundation, either version 3 of the License, or +// (at your option) any later version. +// +// The go-ethereum library is distributed in the hope that it will be useful, +// but WITHOUT ANY WARRANTY; without even the implied warranty of +// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +// GNU Lesser General Public License for more details. +// +// You should have received a copy of the GNU Lesser General Public License +// along with the go-ethereum library. If not, see . + +package lescdn + +import ( + "encoding/binary" + "fmt" + "net/http" + "strconv" + + "github.com/ethereum/go-ethereum/common" + "github.com/ethereum/go-ethereum/common/hexutil" + "github.com/ethereum/go-ethereum/core/rawdb" + "github.com/ethereum/go-ethereum/light" + "github.com/ethereum/go-ethereum/log" + "github.com/ethereum/go-ethereum/trie" +) + +// serveMisc is responsible for serving HTTP requests for miscellaneous data. +func (s *Service) serveMisc(w http.ResponseWriter, r *http.Request) { + switch shift(&r.URL.Path) { + case "cht": + s.serveCHT(w, r) + case "chtv2": + s.serveCHTV2(w, r) + case "bloomtrie": + s.serveBloomTrie(w, r) + } +} + +// serveCHT serves the CHT request and caches the result via HTTP layer. +// +// The format of request is: +// misc/cht/?number=%d +func (s *Service) serveCHT(w http.ResponseWriter, r *http.Request) { + var number uint64 + if numbers, ok := r.URL.Query()["number"]; ok { + number, _ = strconv.ParseUint(numbers[0], 0, 64) + } + root, err := hexutil.Decode(shift(&r.URL.Path)) + if err != nil { + http.Error(w, fmt.Sprintf("invalid cht root: %v", err), http.StatusBadRequest) + return + } + if len(root) != common.HashLength { + http.Error(w, fmt.Sprintf("invalid cht root: length %d != %d", len(root), common.HashLength), http.StatusBadRequest) + return + } + db := s.chain.Database() + t, err := trie.New(common.BytesToHash(root), trie.NewDatabaseWithCache(rawdb.NewTable(db, light.ChtTablePrefix), 1)) + if err != nil { + http.Error(w, fmt.Sprintf("invalid cht trie: %v", err), http.StatusBadRequest) + } + // Generate merkle proof based on the user request + proof := light.NewNodeSet() + var key [8]byte + binary.BigEndian.PutUint64(key[:], number) + if err := t.Prove(key[:], 0, proof); err != nil { + http.Error(w, fmt.Sprintf("failed to generate proof path: %v", err), http.StatusBadRequest) + } + replyAndCache(w, proof.NodeList()) // Done, cache it in http layer. + log.Debug("Served cht request", "chtRoot", common.BytesToHash(root), "number", number) +} + +// serveCHTV2 serves the CHT request and caches the result via HTTP layer. +// +// The format of request is: +// misc/chtv2/ +func (s *Service) serveCHTV2(w http.ResponseWriter, r *http.Request) { + root, err := hexutil.Decode(shift(&r.URL.Path)) + if err != nil { + http.Error(w, fmt.Sprintf("invalid cht root: %v", err), http.StatusBadRequest) + return + } + if len(root) != common.HashLength { + http.Error(w, fmt.Sprintf("invalid cht root: length %d != %d", len(root), common.HashLength), http.StatusBadRequest) + return + } + db := s.chain.Database() + tiles, err := light.ReadCHTTile(db, common.BytesToHash(root)) + if err != nil { + http.Error(w, fmt.Sprintf("invalid cht request: %v", err), http.StatusBadRequest) + } + replyAndCache(w, tiles) +} + +// serveBloomTrie serves the bloomTrie request and caches the result +// via HTTP layer. +// +// The format of request is: +// misc/bloomtrie/?bit=%d§ion=%d +func (s *Service) serveBloomTrie(w http.ResponseWriter, r *http.Request) { + var ( + bit uint64 + section uint64 + ) + if bits, ok := r.URL.Query()["bit"]; ok { + bit, _ = strconv.ParseUint(bits[0], 0, 64) + } + if sections, ok := r.URL.Query()["section"]; ok { + section, _ = strconv.ParseUint(sections[0], 0, 64) + } + root, err := hexutil.Decode(shift(&r.URL.Path)) + if err != nil { + http.Error(w, fmt.Sprintf("invalid bloom trie root: %v", err), http.StatusBadRequest) + return + } + if len(root) != common.HashLength { + http.Error(w, fmt.Sprintf("invalid bloom trie root: length %d != %d", len(root), common.HashLength), http.StatusBadRequest) + return + } + db := s.chain.Database() + t, err := trie.New(common.BytesToHash(root), trie.NewDatabaseWithCache(rawdb.NewTable(db, light.BloomTrieTablePrefix), 1)) + if err != nil { + http.Error(w, fmt.Sprintf("invalid bloom trie: %v", err), http.StatusBadRequest) + } + // Generate merkle proof based on the user request + proof := light.NewNodeSet() + var key [10]byte + binary.BigEndian.PutUint16(key[:2], uint16(bit)) + binary.BigEndian.PutUint64(key[2:], section) + if err := t.Prove(key[:], 0, proof); err != nil { + http.Error(w, fmt.Sprintf("failed to generate proof path: %v", err), http.StatusBadRequest) + } + replyAndCache(w, proof.NodeList()) // Done, cache it in http layer. + log.Debug("Served bloom trie request", "bloomRoot", common.BytesToHash(root), "bit", bit, "section", section) +} diff --git a/lescdn/service.go b/lescdn/service.go index 16994ad7ed..3750beca1f 100644 --- a/lescdn/service.go +++ b/lescdn/service.go @@ -1,3 +1,19 @@ +// Copyright 2019 The go-ethereum Authors +// This file is part of the go-ethereum library. +// +// The go-ethereum library is free software: you can redistribute it and/or modify +// it under the terms of the GNU Lesser General Public License as published by +// the Free Software Foundation, either version 3 of the License, or +// (at your option) any later version. +// +// The go-ethereum library is distributed in the hope that it will be useful, +// but WITHOUT ANY WARRANTY; without even the implied warranty of +// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +// GNU Lesser General Public License for more details. +// +// You should have received a copy of the GNU Lesser General Public License +// along with the go-ethereum library. If not, see . + package lescdn import ( @@ -58,6 +74,10 @@ func (s *Service) ServeHTTP(w http.ResponseWriter, r *http.Request) { case "state": s.serveState(w, r) return + + case "misc": + s.serveMisc(w, r) + return } http.Error(w, http.StatusText(http.StatusNotFound), http.StatusNotFound) } @@ -79,11 +99,18 @@ func shift(p *string) string { return head } -// reply marshals a value into the response stream via RLP, also setting caching +// replyAndCache marshals a value into the response stream via RLP, also setting caching // to indefinite. -func reply(w http.ResponseWriter, v interface{}) { +func replyAndCache(w http.ResponseWriter, v interface{}) { w.Header().Set("Cache-Control", "max-age=31536000") // 1 year cache expiry if err := rlp.Encode(w, v); err != nil { http.Error(w, err.Error(), http.StatusInternalServerError) } } + +// reply marshals a value into the response stream via RLP but not caches it. +func reply(w http.ResponseWriter, v interface{}) { + if err := rlp.Encode(w, v); err != nil { + http.Error(w, err.Error(), http.StatusInternalServerError) + } +} diff --git a/lescdn/state.go b/lescdn/state.go index b4f8761d04..525665fafd 100644 --- a/lescdn/state.go +++ b/lescdn/state.go @@ -1,3 +1,19 @@ +// Copyright 2019 The go-ethereum Authors +// This file is part of the go-ethereum library. +// +// The go-ethereum library is free software: you can redistribute it and/or modify +// it under the terms of the GNU Lesser General Public License as published by +// the Free Software Foundation, either version 3 of the License, or +// (at your option) any later version. +// +// The go-ethereum library is distributed in the hope that it will be useful, +// but WITHOUT ANY WARRANTY; without even the implied warranty of +// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +// GNU Lesser General Public License for more details. +// +// You should have received a copy of the GNU Lesser General Public License +// along with the go-ethereum library. If not, see . + package lescdn import ( @@ -8,6 +24,7 @@ import ( "github.com/ethereum/go-ethereum/common" "github.com/ethereum/go-ethereum/common/hexutil" "github.com/ethereum/go-ethereum/common/prque" + "github.com/ethereum/go-ethereum/log" "github.com/ethereum/go-ethereum/trie" ) @@ -105,7 +122,8 @@ func (s *Service) serveState(w http.ResponseWriter, r *http.Request) { } else { nodes = append(nodes, merges...) } - reply(w, nodes) + replyAndCache(w, nodes) + log.Debug("Served state request", "root", common.BytesToHash(root), "target", cutTileTarget, "limit", curTileLimit, "barrier", curTileBarrier) } // makeIdealTile gathers trie nodes and assembles an ideal tile: one that barely diff --git a/light/cht_tiler.go b/light/cht_tiler.go new file mode 100644 index 0000000000..1ebcb87618 --- /dev/null +++ b/light/cht_tiler.go @@ -0,0 +1,368 @@ +// Copyright 2019 The go-ethereum Authors +// This file is part of the go-ethereum library. +// +// The go-ethereum library is free software: you can redistribute it and/or modify +// it under the terms of the GNU Lesser General Public License as published by +// the Free Software Foundation, either version 3 of the License, or +// (at your option) any later version. +// +// The go-ethereum library is distributed in the hope that it will be useful, +// but WITHOUT ANY WARRANTY; without even the implied warranty of +// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +// GNU Lesser General Public License for more details. +// +// You should have received a copy of the GNU Lesser General Public License +// along with the go-ethereum library. If not, see . + +package light + +import ( + "encoding/binary" + "errors" + "sync" + "time" + + "github.com/ethereum/go-ethereum/common" + "github.com/ethereum/go-ethereum/core/rawdb" + "github.com/ethereum/go-ethereum/ethdb" + "github.com/ethereum/go-ethereum/log" + "github.com/ethereum/go-ethereum/rlp" + "github.com/ethereum/go-ethereum/trie" +) + +var ( + // headTiledSection tracks the latest known tiled section index. + headTiledSection = []byte("LastSection") + tablePrefix = "cht-tile-v10" // tablePrefix is the namespace of tile database. + tilePrefix = []byte("t") // tilePrefix + level(uint8) + position(uint64 big endian) -> tile + fullnodeChildren = 16 // Each completed full node has 16 children(not include the value of itself) + nibbleLen = 16 // The length of nibbles of cht path(term is not included). + + // levelDivisors is the divisors of each tile level. levelDivisors can be used + // for calculating tile number in each level. + // + // e.g. if the size of section is N, the level0 tile number is N/16, level1 tile + // number is N/4096. + // + // We only maintain three level divisors here, since it's enough. + levelDivisors = []int{16, 4096, 1048576} +) + +var errNoCommittedCHT = errors.New("no committed cht for tiles generation") + +// chtTiler is reponsible for creating CHT tiles whenever a new section +// is committed. +type chtTiler struct { + size uint64 // The number of records in one section + levels int // The number of levels we can build immutable tiles + db ethdb.Database // The main database used to store all other data. + table ethdb.Database // The database used to store all tiles relative records + chtTable ethdb.Database // The database used to store all cht nodes. + + taskCh chan uint64 + wg sync.WaitGroup + closeCh chan struct{} +} + +func newCHTTiler(db ethdb.Database, size uint64, sectionCount uint64) *chtTiler { + // Ensure we can build some complete tiles. + if size < uint64(fullnodeChildren) { + return nil + } + tiler := &chtTiler{ + size: size, + levels: MaxTileLevels(size), + db: db, + table: rawdb.NewTable(db, tablePrefix), + chtTable: rawdb.NewTable(db, ChtTablePrefix), + taskCh: make(chan uint64, 64), + closeCh: make(chan struct{}), + } + tiler.wg.Add(1) + go tiler.run(sectionCount) + return tiler +} + +func (t *chtTiler) run(sectionCount uint64) { + defer t.wg.Done() + defer log.Debug("chtTiler stopped") + + var ( + // head is the lastest known tiled section index, nil means no one. + head = readHeadSection(t.table) + + // taskQueue contains all un-processed sections. + taskQueue []uint64 + ) + // Generate initial tiling tasks. + if head == nil { + for i := uint64(0); i < sectionCount; i++ { + taskQueue = append(taskQueue, i) + } + } else { + // Double check whether some committed sections have + // been reverted. If so, re-run the tile task. + for i := uint64(0); i <= *head; i++ { + num := (i+1)*t.size - 1 + hash := rawdb.ReadCanonicalHash(t.db, num) + if tile, _ := readCHTTile(t.table, GetChtRoot(t.db, num, hash)); len(tile) == 0 { + taskQueue = append(taskQueue, i) + log.Debug("Readd commited section for tiling", "section", i) + } + } + for i := *head + 1; i < sectionCount; i++ { + taskQueue = append(taskQueue, i) + } + } + // newTrie initialises the cht trie of given section. + newTrie := func(section uint64) (*trie.Trie, error) { + // Calculate the number of the last block in the specified section + number := (section+1)*t.size - 1 + hash := rawdb.ReadCanonicalHash(t.db, number) + + root := GetChtRoot(t.db, section, hash) + if root == (common.Hash{}) { + return nil, errNoCommittedCHT + } + t, err := trie.New(root, trie.NewDatabaseWithCache(t.chtTable, 1)) + if err != nil { + return nil, err + } + return t, nil + } + // createTiles creates tiles for new generated cht branches in new section. + createTiles := func(section uint64) error { + defer func(start time.Time) { + log.Info("Created tiles", "section", section, "elasped", common.PrettyDuration(time.Since(start))) + }(time.Now()) + + var iter trie.NodeIterator + if section == 0 { + curTrie, err := newTrie(section) + if err != nil { + return err + } + iter = curTrie.NodeIterator(nil) // Create a iterator for traversing the whole CHT + } else { + prevTrie, err := newTrie(section - 1) + if err != nil { + return err + } + prevIter := prevTrie.NodeIterator(nil) + curTrie, err := newTrie(section) + if err != nil { + return err + } + curIter := curTrie.NodeIterator(nil) + iter, _ = trie.NewDifferenceIterator(prevIter, curIter) // Create a diff iterator for traversing the diff + } + // The concrete algorithm for tiling CHT. + // + // We divide CHT into several tiles, each has 16 trie nodes + 1 parent + // (except the topmost one). We can also call tile as node group. + // + // For a section, there are "size" number records, specifically in CHT + // the size is 32768. So that we will have 2048 level0 tiles, 8 level1 + // tiles. + // + // All generated tiles are stored in the local database, the key is root + // of tile. Also all tiles(except topmost one) are immutable, since they + // are children of a complete full node. + // + // The order of trie traverse here is deep first. So the concrete tiling + // algorithm is pretty simple. + // If the path length > 14, pack all trie nodes into level0 tile + // If the path length > 12, pack all trie nodes into level1 tile. + // For all other nodes, pack them into topmost tile. + type topnode struct { + path []byte + blob []byte + } + var ( + tophashes []common.Hash + topnodes = make(map[common.Hash]topnode) + + tiles = make([][][]byte, t.levels) + heads = make([]common.Hash, t.levels) + triedb = trie.NewDatabaseWithCache(t.chtTable, 1) + ) + pack := func(index int, hash common.Hash) error { + node, err := triedb.Node(iter.Hash()) + if err != nil { + return err + } + // Record the head trie node hash if tile is empty. + // Becase order of trie traverse here is deep first, + // we can always get parent of tile packed first. + if index < t.levels && len(tiles[index]) == 0 { + heads[index] = hash + } + tiles[index] = append(tiles[index], node) + if index < t.levels && len(tiles[index]) == fullnodeChildren+1 { + writeCHTTile(t.table, heads[index], tiles[index]) + tiles[index] = nil + } + return nil + } + for iter.Next(true) { + path := iter.Path() + // Ignore the value node here, since it will be + // embedded in the parent short node. + if len(path) == nibbleLen+1 { + continue + } + // If in this level we can't assmble a complete tile, + // pack all of them into topmost level. + if (nibbleLen-len(path))/2 >= t.levels { + node, err := triedb.Node(iter.Hash()) + if err != nil { + return err + } + topnodes[iter.Hash()] = topnode{path: common.CopyBytes(iter.Path()), blob: node} + tophashes = append(tophashes, iter.Hash()) + continue + } + // For the bottom level, we can assemble some complete + // tiles, pack nodes into corresponding tiles. + level := (nibbleLen - len(path)) / 2 + if err := pack(level, iter.Hash()); err != nil { + return err + } + } + // Genrate the topmost tile and persist. + hash := rawdb.ReadCanonicalHash(t.db, (section+1)*t.size-1) + root := GetChtRoot(t.db, section, hash) + + // Fetch all old but immutable children and add into the topmost tile. + curhashes, curnodes := tophashes, topnodes + for { + var hashes []common.Hash + var nodes = make(map[common.Hash]topnode) + for _, h := range curhashes { + node := curnodes[h] + trie.IterateRefs(node.blob, func(path []byte, hash common.Hash) error { + path = append(node.path, path...) + if _, exist := topnodes[hash]; !exist && (nibbleLen-len(path))/2 >= t.levels { + blob, err := triedb.Node(hash) + if err != nil { + return err + } + nodes[hash] = topnode{path: path, blob: blob} + hashes = append(hashes, hash) + } + return nil + }) + } + // Nothing to expand + if len(hashes) == 0 { + break + } + curhashes, curnodes = hashes, nodes + for _, h := range hashes { + topnodes[h] = nodes[h] + } + } + var toptile [][]byte + for _, node := range topnodes { + toptile = append(toptile, node.blob) + } + writeCHTTile(t.table, root, toptile) + writeHeadSection(t.table, section) + return nil + } + runTask := func() { + for len(taskQueue) > 0 { + createTiles(taskQueue[0]) + taskQueue = taskQueue[1:] + } + } + for { + runTask() + select { + case section := <-t.taskCh: + taskQueue = append(taskQueue, section) + case <-t.closeCh: + return + } + } +} + +func (t *chtTiler) commit(section uint64) { + select { + case t.taskCh <- section: + case <-t.closeCh: + } +} + +func (t *chtTiler) close() { + close(t.closeCh) + t.wg.Wait() +} + +// readHeadSection reads the last known tiled section index from database. +func readHeadSection(db ethdb.KeyValueReader) *uint64 { + data, _ := db.Get(headTiledSection) + if len(data) != 8 { + return nil + } + number := binary.BigEndian.Uint64(data) + return &number +} + +// writeHeadSection writes the lastest known tiled section index to database. +func writeHeadSection(db ethdb.KeyValueWriter, section uint64) { + var enc [8]byte + binary.BigEndian.PutUint64(enc[:], section) + if err := db.Put(headTiledSection, enc[:]); err != nil { + log.Crit("Failed to store head tiled section", "err", err) + } +} + +// readCHTTile retrieves the relative tiles from database. +func readCHTTile(db ethdb.KeyValueReader, tileKey common.Hash) ([][]byte, error) { + var key []byte + key = append(key, tilePrefix...) + key = append(key, tileKey.Bytes()...) + enc, err := db.Get(key) + if err != nil { + return nil, err + } + var tiles [][]byte + err = rlp.DecodeBytes(enc, &tiles) + if err != nil { + return nil, err + } + return tiles, nil +} + +// writeCHTTile writes generated tiles into the database. +func writeCHTTile(db ethdb.KeyValueWriter, tileKey common.Hash, tile [][]byte) { + var key []byte + key = append(key, tilePrefix...) + key = append(key, tileKey.Bytes()...) + enc, err := rlp.EncodeToBytes(tile) + if err != nil { + log.Crit("Failed to rlp encode the tile blob", "err", err) + } + if err := db.Put(key, enc[:]); err != nil { + log.Crit("Failed to store tile", "err", err) + } +} + +// ReadCHTTile retrieves the relative tiles from database based on the given key. +func ReadCHTTile(db ethdb.Database, tileKey common.Hash) ([][]byte, error) { + table := rawdb.NewTable(db, tablePrefix) + return readCHTTile(table, tileKey) +} + +// MaxTileLevels calculates the levels of immutable tiles. +func MaxTileLevels(size uint64) int { + var levels int + for i := 1; i <= len(levelDivisors); i++ { + if size/uint64(levelDivisors[i-1]) == 0 { + levels = i - 1 + break + } + } + return levels +} diff --git a/light/odr.go b/light/odr.go index 907712ede7..1fbe40528e 100644 --- a/light/odr.go +++ b/light/odr.go @@ -20,12 +20,14 @@ import ( "context" "errors" "math/big" + "sync" "github.com/ethereum/go-ethereum/common" "github.com/ethereum/go-ethereum/core" "github.com/ethereum/go-ethereum/core/rawdb" "github.com/ethereum/go-ethereum/core/types" "github.com/ethereum/go-ethereum/ethdb" + "github.com/ethereum/go-ethereum/rlp" ) // NoOdr is the default context passed to an ODR capable function when the ODR @@ -82,22 +84,31 @@ func StorageTrieID(state *TrieID, addrHash, root common.Hash) *TrieID { // TrieRequest is the ODR request type for state/storage trie entries type TrieRequest struct { - OdrRequest - Id *TrieID - Key []byte + // Request fields + Id *TrieID + Key []byte + MissNodeHash common.Hash + + // Response fields + Lock sync.Mutex Proof *NodeSet } // StoreResult stores the retrieved data in local database func (req *TrieRequest) StoreResult(db ethdb.Database) { - req.Proof.Store(db) + if req.Proof != nil { + req.Proof.Store(db) + } } // CodeRequest is the ODR request type for retrieving contract code type CodeRequest struct { - OdrRequest + // Request fields Id *TrieID // references storage trie of the account Hash common.Hash + + // Response fields + Lock sync.Mutex Data []byte } @@ -108,25 +119,36 @@ func (req *CodeRequest) StoreResult(db ethdb.Database) { // BlockRequest is the ODR request type for retrieving block bodies type BlockRequest struct { - OdrRequest + // Request fields Hash common.Hash Number uint64 - Rlp []byte + + // Response fields + Lock sync.Mutex + Txs types.Transactions + Uncles []*types.Header } // StoreResult stores the retrieved data in local database func (req *BlockRequest) StoreResult(db ethdb.Database) { - rawdb.WriteBodyRLP(db, req.Hash, req.Number, req.Rlp) + data, err := rlp.EncodeToBytes(&types.Body{Transactions: req.Txs, Uncles: req.Uncles}) + if err != nil { + panic(err) // todo how to handle the error? + } + rawdb.WriteBodyRLP(db, req.Hash, req.Number, data) } // ReceiptsRequest is the ODR request type for retrieving block bodies type ReceiptsRequest struct { - OdrRequest + // Request fields Untrusted bool // Indicator whether the result retrieved is trusted or not Hash common.Hash Number uint64 Header *types.Header - Receipts types.Receipts + + // Response fields + Lock sync.Mutex + Receipts types.Receipts } // StoreResult stores the retrieved data in local database @@ -138,15 +160,18 @@ func (req *ReceiptsRequest) StoreResult(db ethdb.Database) { // ChtRequest is the ODR request type for state/storage trie entries type ChtRequest struct { - OdrRequest + // Request fields Untrusted bool // Indicator whether the result retrieved is trusted or not PeerId string // The specified peer id from which to retrieve data. Config *IndexerConfig ChtNum, BlockNum uint64 ChtRoot common.Hash - Header *types.Header - Td *big.Int - Proof *NodeSet + + // Response fields + Lock sync.Mutex + Header *types.Header + Td *big.Int + Proof *NodeSet } // StoreResult stores the retrieved data in local database @@ -162,25 +187,28 @@ func (req *ChtRequest) StoreResult(db ethdb.Database) { // BloomRequest is the ODR request type for retrieving bloom filters from a CHT structure type BloomRequest struct { - OdrRequest - Config *IndexerConfig - BloomTrieNum uint64 - BitIdx uint - SectionIndexList []uint64 - BloomTrieRoot common.Hash - BloomBits [][]byte - Proofs *NodeSet + // Request fields + Config *IndexerConfig + BitIndex uint + SectionList []uint64 + BloomTrieNum uint64 + BloomTrieRoot common.Hash + + // Response fields + Lock sync.Mutex + BloomBits [][]byte + Proofs *NodeSet } // StoreResult stores the retrieved data in local database func (req *BloomRequest) StoreResult(db ethdb.Database) { - for i, sectionIdx := range req.SectionIndexList { + for i, sectionIdx := range req.SectionList { sectionHead := rawdb.ReadCanonicalHash(db, (sectionIdx+1)*req.Config.BloomTrieSize-1) // if we don't have the canonical hash stored for this section head number, we'll still store it under // a key with a zero sectionHead. GetBloomBits will look there too if we still don't have the canonical // hash. In the unlikely case we've retrieved the section head hash since then, we'll just retrieve the // bit vector again from the network. - rawdb.WriteBloomBits(db, req.BitIdx, sectionIdx, sectionHead, req.BloomBits[i]) + rawdb.WriteBloomBits(db, req.BitIndex, sectionIdx, sectionHead, req.BloomBits[i]) } } @@ -193,8 +221,11 @@ type TxStatus struct { // TxStatusRequest is the ODR request type for retrieving transaction status type TxStatusRequest struct { - OdrRequest + // Request fields Hashes []common.Hash + + // Response fields + Lock sync.Mutex Status []TxStatus } diff --git a/light/odr_test.go b/light/odr_test.go index debd5544c3..455c4cd28f 100644 --- a/light/odr_test.go +++ b/light/odr_test.go @@ -74,7 +74,11 @@ func (odr *testOdr) Retrieve(ctx context.Context, req OdrRequest) error { case *BlockRequest: number := rawdb.ReadHeaderNumber(odr.sdb, req.Hash) if number != nil { - req.Rlp = rawdb.ReadBodyRLP(odr.sdb, req.Hash, *number) + blob := rawdb.ReadBodyRLP(odr.sdb, req.Hash, *number) + var body types.Body + rlp.DecodeBytes(blob, &body) + req.Txs, req.Uncles = body.Transactions, body.Uncles + return nil } case *ReceiptsRequest: number := rawdb.ReadHeaderNumber(odr.sdb, req.Hash) diff --git a/light/odr_util.go b/light/odr_util.go index 2c820d40c7..e47aa57606 100644 --- a/light/odr_util.go +++ b/light/odr_util.go @@ -99,9 +99,8 @@ func GetBodyRLP(ctx context.Context, odr OdrBackend, hash common.Hash, number ui r := &BlockRequest{Hash: hash, Number: number} if err := odr.Retrieve(ctx, r); err != nil { return nil, err - } else { - return r.Rlp, nil } + return rawdb.ReadBodyRLP(odr.Database(), hash, number), nil } // GetBody retrieves the block body (transactons, uncles) corresponding to the @@ -252,7 +251,7 @@ func GetBloomBits(ctx context.Context, odr OdrBackend, bitIdx uint, sectionIdxLi } r := &BloomRequest{BloomTrieRoot: GetBloomTrieRoot(db, bloomTrieCount-1, sectionHead), BloomTrieNum: bloomTrieCount - 1, - BitIdx: bitIdx, SectionIndexList: reqList, Config: odr.IndexerConfig()} + BitIndex: bitIdx, SectionList: reqList, Config: odr.IndexerConfig()} if err := odr.Retrieve(ctx, r); err != nil { return nil, err } else { diff --git a/light/postprocess.go b/light/postprocess.go index af3b257923..976ee4a579 100644 --- a/light/postprocess.go +++ b/light/postprocess.go @@ -134,6 +134,7 @@ type ChtIndexerBackend struct { section, sectionSize uint64 lastHash common.Hash trie *trie.Trie + tiler *chtTiler } // NewChtIndexer creates a Cht chain indexer @@ -146,7 +147,17 @@ func NewChtIndexer(db ethdb.Database, odr OdrBackend, size, confirms uint64) *co triedb: trie.NewDatabaseWithCache(trieTable, 1), // Use a tiny cache only to keep memory down sectionSize: size, } - return core.NewChainIndexer(db, rawdb.NewTable(db, "chtIndexV2-"), backend, size, confirms, time.Millisecond*100, "cht") + indexDb := rawdb.NewTable(db, "chtIndexV2-") + var count uint64 + data, _ := indexDb.Get([]byte("count")) + if len(data) == 8 { + count = binary.BigEndian.Uint64(data) + } + // Only enable it in the server side + if odr == nil { + backend.tiler = newCHTTiler(db, size, count) + } + return core.NewChainIndexer(db, indexDb, backend, size, confirms, time.Millisecond*100, "cht") } // fetchMissingNodes tries to retrieve the last entry of the latest trusted CHT from the @@ -158,7 +169,9 @@ func (c *ChtIndexerBackend) fetchMissingNodes(ctx context.Context, section uint6 err := c.odr.Retrieve(ctx, r) switch err { case nil: - r.Proof.Store(batch) + if r.Proof != nil { + r.Proof.Store(batch) + } return batch.Write() case ErrNoPeers: // if there are no peers to serve, retry later @@ -189,7 +202,6 @@ func (c *ChtIndexerBackend) Reset(ctx context.Context, section uint64, lastSecti c.trie, err = trie.New(root, c.triedb) } } - c.section = section return err } @@ -220,6 +232,11 @@ func (c *ChtIndexerBackend) Commit() error { log.Info("Storing CHT", "section", c.section, "head", fmt.Sprintf("%064x", c.lastHash), "root", fmt.Sprintf("%064x", root)) StoreChtRoot(c.diskdb, c.section, c.lastHash, root) + + // Push a new work to tiler. + if c.tiler != nil { + c.tiler.commit(c.section) + } return nil } @@ -284,7 +301,7 @@ func (b *BloomTrieIndexerBackend) fetchMissingNodes(ctx context.Context, section for i := 0; i < 20; i++ { go func() { for bitIndex := range indexCh { - r := &BloomRequest{BloomTrieRoot: root, BloomTrieNum: section - 1, BitIdx: bitIndex, SectionIndexList: []uint64{section - 1}, Config: b.odr.IndexerConfig()} + r := &BloomRequest{BloomTrieRoot: root, BloomTrieNum: section - 1, BitIndex: bitIndex, SectionList: []uint64{section - 1}, Config: b.odr.IndexerConfig()} for { if err := b.odr.Retrieve(ctx, r); err == ErrNoPeers { // if there are no peers to serve, retry later diff --git a/light/trie.go b/light/trie.go index e512bf6f95..46fea9f9e0 100644 --- a/light/trie.go +++ b/light/trie.go @@ -156,10 +156,11 @@ func (t *odrTrie) do(key []byte, fn func() error) error { if err == nil { err = fn() } - if _, ok := err.(*trie.MissingNodeError); !ok { + missError, ok := err.(*trie.MissingNodeError) + if !ok { return err } - r := &TrieRequest{Id: t.id, Key: key} + r := &TrieRequest{Id: t.id, Key: key, MissNodeHash: missError.NodeHash} if err := t.db.backend.Retrieve(t.db.ctx, r); err != nil { return err } diff --git a/trie/iterator.go b/trie/iterator.go index 40b7ba6190..25b3b2c016 100644 --- a/trie/iterator.go +++ b/trie/iterator.go @@ -580,7 +580,11 @@ func (it *unionIterator) Error() error { // IterateRefs decodes a trie node, iterates all its children and invokes a // callback for each hash node found. func IterateRefs(node []byte, onHashNode func([]byte, common.Hash) error) error { - return iterateRefs(mustDecodeNode(nil, node), nil, onHashNode) + n, err := decodeNode(nil, node) + if err != nil { + return err + } + return iterateRefs(n, nil, onHashNode) } // iterateRefs traverses the node hierarchy of a cached node and invokes the diff --git a/trie/proof.go b/trie/proof.go index 58ca69c680..fada39438f 100644 --- a/trie/proof.go +++ b/trie/proof.go @@ -21,6 +21,7 @@ import ( "fmt" "github.com/ethereum/go-ethereum/common" + "github.com/ethereum/go-ethereum/common/prque" "github.com/ethereum/go-ethereum/ethdb" "github.com/ethereum/go-ethereum/log" "github.com/ethereum/go-ethereum/rlp" @@ -150,3 +151,43 @@ func get(tn node, key []byte) ([]byte, node) { } } } + +// VerifyTrie checks whether the given sub trie is exactly matched with specified +// root hash. The main difference between VerifyTrie and VerifyProof is the target +// of the former is sub trie while the latter is trie path. +func VerifyTrie(rootHash common.Hash, proofDb ethdb.KeyValueReader, nodes int) error { + queue := prque.New(nil) + queue.Push(rootHash, 0) + + var ( + visit int // The number of checked trie nodes + count int // The number of trie nodes. + ) + for !queue.Empty() && visit < nodes { + item, _ := queue.Pop() + hash := item.(common.Hash) + blob, err := proofDb.Get(hash.Bytes()) + // We haven't traversed the whole db, but hash mismatch comes first. + if err != nil { + return err + } + node, err := decodeNode(hash.Bytes(), blob) + if err != nil { + return err + } + visit += 1 + err = iterateRefs(node, nil, func(i []byte, hash common.Hash) error { + // Short circuit if hash candidates already exceeds the total entries. + if count+2 > nodes { // plus on root node. + return nil + } + queue.Push(hash, int64(-1*count)) + count += 1 + return nil + }) + if err != nil { + return err + } + } + return nil +} diff --git a/trie/trie.go b/trie/trie.go index 78e2eff534..1aaa000f5d 100644 --- a/trie/trie.go +++ b/trie/trie.go @@ -40,6 +40,20 @@ var ( // between account and storage tries. type LeafCallback func(leaf []byte, parent common.Hash) error +// TraceConfig is a set of tracing config which caller can specify. +type TraceConfig struct { + RecordPath bool // Record the trie path of all read nodes. + RecordHash bool // Record the node hash of all read nodes. + RecordLeafBlob bool // Record RLP encoded value of read leaves. +} + +// TrieTrace represents a read step of trie. +type TrieTrace struct { + Path []byte // The path from root node to current visited node. + LeafBlob []byte // The RLP encoded value of leaf node. + Hash common.Hash // The hash of visited node, can be nil if it's not a standalone or dirty +} + // Trie is a Merkle Patricia Trie. // The zero value is an empty trie with no database. // Use New to create a trie that sits on top of a database. @@ -52,6 +66,11 @@ type Trie struct { // hashing operation. This number will not directly map to the number of // actually unhashed nodes unhashed int + + // The fields needed by tracing. + enableTrace bool + traceConfig *TraceConfig + traces []*TrieTrace } // newFlag returns the cache flag value for a newly created node. @@ -69,8 +88,27 @@ func New(root common.Hash, db *Database) (*Trie, error) { if db == nil { panic("trie.New called without a database") } + trie := &Trie{db: db} + if root != (common.Hash{}) && root != emptyRoot { + rootnode, err := trie.resolveHash(root[:], nil) + if err != nil { + return nil, err + } + trie.root = rootnode + } + return trie, nil +} + +// NewTraceTrie creates a traceable trie which can offer more tracing +// information during read opt. +func NewTraceTrie(root common.Hash, db *Database, config *TraceConfig) (*Trie, error) { + if db == nil { + panic("trie.NewTraceTrie called without a database") + } trie := &Trie{ - db: db, + enableTrace: true, + traceConfig: config, + db: db, } if root != (common.Hash{}) && root != emptyRoot { rootnode, err := trie.resolveHash(root[:], nil) @@ -103,6 +141,12 @@ func (t *Trie) Get(key []byte) []byte { // If a node was not found in the database, a MissingNodeError is returned. func (t *Trie) TryGet(key []byte) ([]byte, error) { key = keybytesToHex(key) + + // If read tracing is enable, clean all history + // before a new round reading. + if t.enableTrace { + t.traces = nil + } value, newroot, didResolve, err := t.tryGet(t.root, key, 0) if err == nil && didResolve { t.root = newroot @@ -115,12 +159,22 @@ func (t *Trie) tryGet(origNode node, key []byte, pos int) (value []byte, newnode case nil: return nil, nil, false, nil case valueNode: + if t.enableTrace { + t.trace(common.Hash{}, common.CopyBytes(key[:pos]), common.CopyBytes(n)) + } return n, n, false, nil case *shortNode: if len(key)-pos < len(n.Key) || !bytes.Equal(n.Key, key[pos:pos+len(n.Key)]) { // key not found in trie return nil, n, false, nil } + if t.enableTrace { + // Cached hash can be nil if it's dirty. + // Besides if it's an embedded node, the + // hash is also nil. + h, _ := n.cache() + t.trace(common.BytesToHash(h), common.CopyBytes(key[:pos]), nil) + } value, newnode, didResolve, err = t.tryGet(n.Val, key, pos+len(n.Key)) if err == nil && didResolve { n = n.copy() @@ -128,6 +182,13 @@ func (t *Trie) tryGet(origNode node, key []byte, pos int) (value []byte, newnode } return value, n, didResolve, err case *fullNode: + if t.enableTrace { + // Cached hash can be nil if it's dirty. + // Besides if it's an embedded node, the + // hash is also nil. + h, _ := n.cache() + t.trace(common.BytesToHash(h), common.CopyBytes(key[:pos]), nil) + } value, newnode, didResolve, err = t.tryGet(n.Children[key[pos]], key, pos+1) if err == nil && didResolve { n = n.copy() @@ -146,6 +207,21 @@ func (t *Trie) tryGet(origNode node, key []byte, pos int) (value []byte, newnode } } +// trace adds a new read step into trace record set. +func (t *Trie) trace(hash common.Hash, path []byte, leafBlob []byte) { + record := &TrieTrace{} + if t.traceConfig.RecordHash { + record.Hash = hash + } + if t.traceConfig.RecordPath { + record.Path = path + } + if t.traceConfig.RecordLeafBlob { + record.LeafBlob = leafBlob + } + t.traces = append(t.traces, record) +} + // Update associates key with value in the trie. Subsequent calls to // Get will return value. If value has length zero, any existing value // is deleted from the trie and calls to Get will return nil. @@ -473,3 +549,9 @@ func (t *Trie) hashRoot(db *Database) (node, node, error) { t.unhashed = 0 return hashed, cached, nil } + +// GetTraces returns the read traces recorded by last Get +// operation. Note, caller can't modify the returned value. +func (t *Trie) GetTraces() []*TrieTrace { + return t.traces +} diff --git a/trie/trie_test.go b/trie/trie_test.go index 172572dddc..ec80564764 100644 --- a/trie/trie_test.go +++ b/trie/trie_test.go @@ -347,7 +347,58 @@ func TestRandomCases(t *testing.T) { {op: 1, key: common.Hex2Bytes("fd"), value: common.Hex2Bytes("")}, // step 25 } runRandTest(rt) +} +func Example_TraceTrie() { + triedb := NewDatabase(memorydb.New()) + trie, _ := NewTraceTrie(common.Hash{}, triedb, &TraceConfig{RecordPath: true, RecordHash: true, RecordLeafBlob: true}) + vals := []struct{ k, v string }{ + {"do", "verb"}, + {"dog", "puppy"}, + {"doge", "coin"}, + } + for _, val := range vals { + updateString(trie, val.k, val.v) + } + root, err := trie.Commit(nil) + if err != nil { + fmt.Println("commit err", err) + } + _, err = trie.TryGet([]byte("dog")) + if err != nil { + fmt.Println("read err", err) + } + traces := trie.traces + for index, t := range traces { + fmt.Printf("step%d => path: %v, hash: %v, blob: %v\n", index+1, t.Path, t.Hash.Hex(), t.LeafBlob) + } + + // Reopen from database + trie, _ = NewTraceTrie(root, triedb, &TraceConfig{RecordPath: true, RecordHash: true, RecordLeafBlob: true}) + _, err = trie.TryGet([]byte("dog")) + if err != nil { + fmt.Println("read err", err) + } + traces = trie.traces + for index, t := range traces { + fmt.Printf("step%d => path: %v, hash: %v, blob: %v\n", index+1, t.Path, t.Hash.Hex(), t.LeafBlob) + } + // Output: + // step1 => path: [], hash: 0xef7b2fe20f5d2c30c46ad4d83c39811bcbf1721aef2e805c0e107947320888b6, blob: [] + // step2 => path: [6 4 6 15], hash: 0xd43b87fdcd4217013ccc92d04662e12d36e4cc25dc690077cd821a1956fc3e36, blob: [] + // step3 => path: [6 4 6 15 6], hash: 0x0000000000000000000000000000000000000000000000000000000000000000, blob: [] + // step4 => path: [6 4 6 15 6 7], hash: 0x0000000000000000000000000000000000000000000000000000000000000000, blob: [] + // step5 => path: [6 4 6 15 6 7 16], hash: 0x0000000000000000000000000000000000000000000000000000000000000000, blob: [112 117 112 112 121] + // step1 => path: [], hash: 0xef7b2fe20f5d2c30c46ad4d83c39811bcbf1721aef2e805c0e107947320888b6, blob: [] + // step2 => path: [6 4 6 15], hash: 0xd43b87fdcd4217013ccc92d04662e12d36e4cc25dc690077cd821a1956fc3e36, blob: [] + // step3 => path: [6 4 6 15 6], hash: 0x0000000000000000000000000000000000000000000000000000000000000000, blob: [] + // step4 => path: [6 4 6 15 6 7], hash: 0x0000000000000000000000000000000000000000000000000000000000000000, blob: [] + // step5 => path: [6 4 6 15 6 7 16], hash: 0x0000000000000000000000000000000000000000000000000000000000000000, blob: [112 117 112 112 121] +} + +type countingDB struct { + ethdb.KeyValueStore + gets map[string]int } // randTest performs random trie operations.