123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434 |
- // 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 <http://www.gnu.org/licenses/>.
- package les
- import (
- "errors"
- "sync"
- "sync/atomic"
- "time"
- "github.com/ethereum/go-ethereum/common"
- "github.com/ethereum/go-ethereum/common/mclock"
- "github.com/ethereum/go-ethereum/core"
- "github.com/ethereum/go-ethereum/core/forkid"
- "github.com/ethereum/go-ethereum/core/rawdb"
- "github.com/ethereum/go-ethereum/core/state"
- "github.com/ethereum/go-ethereum/core/types"
- "github.com/ethereum/go-ethereum/ethdb"
- "github.com/ethereum/go-ethereum/les/flowcontrol"
- "github.com/ethereum/go-ethereum/light"
- "github.com/ethereum/go-ethereum/log"
- "github.com/ethereum/go-ethereum/metrics"
- "github.com/ethereum/go-ethereum/p2p"
- "github.com/ethereum/go-ethereum/rlp"
- "github.com/ethereum/go-ethereum/trie"
- )
- const (
- softResponseLimit = 2 * 1024 * 1024 // Target maximum size of returned blocks, headers or node data.
- estHeaderRlpSize = 500 // Approximate size of an RLP encoded block header
- MaxHeaderFetch = 192 // Amount of block headers to be fetched per retrieval request
- MaxBodyFetch = 32 // Amount of block bodies to be fetched per retrieval request
- MaxReceiptFetch = 128 // Amount of transaction receipts to allow fetching per request
- MaxCodeFetch = 64 // Amount of contract codes to allow fetching per request
- MaxProofsFetch = 64 // Amount of merkle proofs to be fetched per retrieval request
- MaxHelperTrieProofsFetch = 64 // Amount of helper tries to be fetched per retrieval request
- MaxTxSend = 64 // Amount of transactions to be send per request
- MaxTxStatus = 256 // Amount of transactions to queried per request
- )
- var (
- errTooManyInvalidRequest = errors.New("too many invalid requests made")
- )
- // serverHandler is responsible for serving light client and process
- // all incoming light requests.
- type serverHandler struct {
- forkFilter forkid.Filter
- blockchain *core.BlockChain
- chainDb ethdb.Database
- txpool *core.TxPool
- server *LesServer
- closeCh chan struct{} // Channel used to exit all background routines of handler.
- wg sync.WaitGroup // WaitGroup used to track all background routines of handler.
- synced func() bool // Callback function used to determine whether local node is synced.
- // Testing fields
- addTxsSync bool
- }
- func newServerHandler(server *LesServer, blockchain *core.BlockChain, chainDb ethdb.Database, txpool *core.TxPool, synced func() bool) *serverHandler {
- handler := &serverHandler{
- forkFilter: forkid.NewFilter(blockchain),
- server: server,
- blockchain: blockchain,
- chainDb: chainDb,
- txpool: txpool,
- closeCh: make(chan struct{}),
- synced: synced,
- }
- return handler
- }
- // start starts the server handler.
- func (h *serverHandler) start() {
- h.wg.Add(1)
- go h.broadcastLoop()
- }
- // stop stops the server handler.
- func (h *serverHandler) stop() {
- close(h.closeCh)
- h.wg.Wait()
- }
- // runPeer is the p2p protocol run function for the given version.
- func (h *serverHandler) runPeer(version uint, p *p2p.Peer, rw p2p.MsgReadWriter) error {
- peer := newClientPeer(int(version), h.server.config.NetworkId, p, newMeteredMsgWriter(rw, int(version)))
- defer peer.close()
- h.wg.Add(1)
- defer h.wg.Done()
- return h.handle(peer)
- }
- func (h *serverHandler) handle(p *clientPeer) error {
- p.Log().Debug("Light Ethereum peer connected", "name", p.Name())
- // Execute the LES handshake
- var (
- head = h.blockchain.CurrentHeader()
- hash = head.Hash()
- number = head.Number.Uint64()
- td = h.blockchain.GetTd(hash, number)
- forkID = forkid.NewID(h.blockchain.Config(), h.blockchain.Genesis().Hash(), h.blockchain.CurrentBlock().NumberU64())
- )
- if err := p.Handshake(td, hash, number, h.blockchain.Genesis().Hash(), forkID, h.forkFilter, h.server); err != nil {
- p.Log().Debug("Light Ethereum handshake failed", "err", err)
- return err
- }
- // Connected to another server, no messages expected, just wait for disconnection
- if p.server {
- if err := h.server.serverset.register(p); err != nil {
- return err
- }
- _, err := p.rw.ReadMsg()
- h.server.serverset.unregister(p)
- return err
- }
- // Setup flow control mechanism for the peer
- p.fcClient = flowcontrol.NewClientNode(h.server.fcManager, p.fcParams)
- defer p.fcClient.Disconnect()
- // Reject light clients if server is not synced. Put this checking here, so
- // that "non-synced" les-server peers are still allowed to keep the connection.
- if !h.synced() {
- p.Log().Debug("Light server not synced, rejecting peer")
- return p2p.DiscRequested
- }
- // Register the peer into the peerset and clientpool
- if err := h.server.peers.register(p); err != nil {
- return err
- }
- if p.balance = h.server.clientPool.Register(p); p.balance == nil {
- h.server.peers.unregister(p.ID())
- p.Log().Debug("Client pool already closed")
- return p2p.DiscRequested
- }
- p.connectedAt = mclock.Now()
- var wg sync.WaitGroup // Wait group used to track all in-flight task routines.
- defer func() {
- wg.Wait() // Ensure all background task routines have exited.
- h.server.clientPool.Unregister(p)
- h.server.peers.unregister(p.ID())
- p.balance = nil
- connectionTimer.Update(time.Duration(mclock.Now() - p.connectedAt))
- }()
- // Mark the peer as being served.
- atomic.StoreUint32(&p.serving, 1)
- defer atomic.StoreUint32(&p.serving, 0)
- // Spawn a main loop to handle all incoming messages.
- for {
- select {
- case err := <-p.errCh:
- p.Log().Debug("Failed to send light ethereum response", "err", err)
- return err
- default:
- }
- if err := h.handleMsg(p, &wg); err != nil {
- p.Log().Debug("Light Ethereum message handling failed", "err", err)
- return err
- }
- }
- }
- // beforeHandle will do a series of prechecks before handling message.
- func (h *serverHandler) beforeHandle(p *clientPeer, reqID, responseCount uint64, msg p2p.Msg, reqCnt uint64, maxCount uint64) (*servingTask, uint64) {
- // Ensure that the request sent by client peer is valid
- inSizeCost := h.server.costTracker.realCost(0, msg.Size, 0)
- if reqCnt == 0 || reqCnt > maxCount {
- p.fcClient.OneTimeCost(inSizeCost)
- return nil, 0
- }
- // Ensure that the client peer complies with the flow control
- // rules agreed by both sides.
- if p.isFrozen() {
- p.fcClient.OneTimeCost(inSizeCost)
- return nil, 0
- }
- maxCost := p.fcCosts.getMaxCost(msg.Code, reqCnt)
- accepted, bufShort, priority := p.fcClient.AcceptRequest(reqID, responseCount, maxCost)
- if !accepted {
- p.freeze()
- p.Log().Error("Request came too early", "remaining", common.PrettyDuration(time.Duration(bufShort*1000000/p.fcParams.MinRecharge)))
- p.fcClient.OneTimeCost(inSizeCost)
- return nil, 0
- }
- // Create a multi-stage task, estimate the time it takes for the task to
- // execute, and cache it in the request service queue.
- factor := h.server.costTracker.globalFactor()
- if factor < 0.001 {
- factor = 1
- p.Log().Error("Invalid global cost factor", "factor", factor)
- }
- maxTime := uint64(float64(maxCost) / factor)
- task := h.server.servingQueue.newTask(p, maxTime, priority)
- if !task.start() {
- p.fcClient.RequestProcessed(reqID, responseCount, maxCost, inSizeCost)
- return nil, 0
- }
- return task, maxCost
- }
- // Afterhandle will perform a series of operations after message handling,
- // such as updating flow control data, sending reply, etc.
- func (h *serverHandler) afterHandle(p *clientPeer, reqID, responseCount uint64, msg p2p.Msg, maxCost uint64, reqCnt uint64, task *servingTask, reply *reply) {
- if reply != nil {
- task.done()
- }
- p.responseLock.Lock()
- defer p.responseLock.Unlock()
- // Short circuit if the client is already frozen.
- if p.isFrozen() {
- realCost := h.server.costTracker.realCost(task.servingTime, msg.Size, 0)
- p.fcClient.RequestProcessed(reqID, responseCount, maxCost, realCost)
- return
- }
- // Positive correction buffer value with real cost.
- var replySize uint32
- if reply != nil {
- replySize = reply.size()
- }
- var realCost uint64
- if h.server.costTracker.testing {
- realCost = maxCost // Assign a fake cost for testing purpose
- } else {
- realCost = h.server.costTracker.realCost(task.servingTime, msg.Size, replySize)
- if realCost > maxCost {
- realCost = maxCost
- }
- }
- bv := p.fcClient.RequestProcessed(reqID, responseCount, maxCost, realCost)
- if reply != nil {
- // Feed cost tracker request serving statistic.
- h.server.costTracker.updateStats(msg.Code, reqCnt, task.servingTime, realCost)
- // Reduce priority "balance" for the specific peer.
- p.balance.RequestServed(realCost)
- p.queueSend(func() {
- if err := reply.send(bv); err != nil {
- select {
- case p.errCh <- err:
- default:
- }
- }
- })
- }
- }
- // handleMsg is invoked whenever an inbound message is received from a remote
- // peer. The remote connection is torn down upon returning any error.
- func (h *serverHandler) handleMsg(p *clientPeer, wg *sync.WaitGroup) error {
- // Read the next message from the remote peer, and ensure it's fully consumed
- msg, err := p.rw.ReadMsg()
- if err != nil {
- return err
- }
- p.Log().Trace("Light Ethereum message arrived", "code", msg.Code, "bytes", msg.Size)
- // Discard large message which exceeds the limitation.
- if msg.Size > ProtocolMaxMsgSize {
- clientErrorMeter.Mark(1)
- return errResp(ErrMsgTooLarge, "%v > %v", msg.Size, ProtocolMaxMsgSize)
- }
- defer msg.Discard()
- // Lookup the request handler table, ensure it's supported
- // message type by the protocol.
- req, ok := Les3[msg.Code]
- if !ok {
- p.Log().Trace("Received invalid message", "code", msg.Code)
- clientErrorMeter.Mark(1)
- return errResp(ErrInvalidMsgCode, "%v", msg.Code)
- }
- p.Log().Trace("Received " + req.Name)
- // Decode the p2p message, resolve the concrete handler for it.
- serve, reqID, reqCnt, err := req.Handle(msg)
- if err != nil {
- clientErrorMeter.Mark(1)
- return errResp(ErrDecode, "%v: %v", msg, err)
- }
- if metrics.EnabledExpensive {
- req.InPacketsMeter.Mark(1)
- req.InTrafficMeter.Mark(int64(msg.Size))
- }
- p.responseCount++
- responseCount := p.responseCount
- // First check this client message complies all rules before
- // handling it and return a processor if all checks are passed.
- task, maxCost := h.beforeHandle(p, reqID, responseCount, msg, reqCnt, req.MaxCount)
- if task == nil {
- return nil
- }
- wg.Add(1)
- go func() {
- defer wg.Done()
- reply := serve(h, p, task.waitOrStop)
- h.afterHandle(p, reqID, responseCount, msg, maxCost, reqCnt, task, reply)
- if metrics.EnabledExpensive {
- size := uint32(0)
- if reply != nil {
- size = reply.size()
- }
- req.OutPacketsMeter.Mark(1)
- req.OutTrafficMeter.Mark(int64(size))
- req.ServingTimeMeter.Update(time.Duration(task.servingTime))
- }
- }()
- // If the client has made too much invalid request(e.g. request a non-existent data),
- // reject them to prevent SPAM attack.
- if p.getInvalid() > maxRequestErrors {
- clientErrorMeter.Mark(1)
- return errTooManyInvalidRequest
- }
- return nil
- }
- // BlockChain implements serverBackend
- func (h *serverHandler) BlockChain() *core.BlockChain {
- return h.blockchain
- }
- // TxPool implements serverBackend
- func (h *serverHandler) TxPool() *core.TxPool {
- return h.txpool
- }
- // ArchiveMode implements serverBackend
- func (h *serverHandler) ArchiveMode() bool {
- return h.server.archiveMode
- }
- // AddTxsSync implements serverBackend
- func (h *serverHandler) AddTxsSync() bool {
- return h.addTxsSync
- }
- // getAccount retrieves an account from the state based on root.
- func getAccount(triedb *trie.Database, root, hash common.Hash) (state.Account, error) {
- trie, err := trie.New(root, triedb)
- if err != nil {
- return state.Account{}, err
- }
- blob, err := trie.TryGet(hash[:])
- if err != nil {
- return state.Account{}, err
- }
- var account state.Account
- if err = rlp.DecodeBytes(blob, &account); err != nil {
- return state.Account{}, err
- }
- return account, nil
- }
- // getHelperTrie returns the post-processed trie root for the given trie ID and section index
- func (h *serverHandler) GetHelperTrie(typ uint, index uint64) *trie.Trie {
- var (
- root common.Hash
- prefix string
- )
- switch typ {
- case htCanonical:
- sectionHead := rawdb.ReadCanonicalHash(h.chainDb, (index+1)*h.server.iConfig.ChtSize-1)
- root, prefix = light.GetChtRoot(h.chainDb, index, sectionHead), light.ChtTablePrefix
- case htBloomBits:
- sectionHead := rawdb.ReadCanonicalHash(h.chainDb, (index+1)*h.server.iConfig.BloomTrieSize-1)
- root, prefix = light.GetBloomTrieRoot(h.chainDb, index, sectionHead), light.BloomTrieTablePrefix
- }
- if root == (common.Hash{}) {
- return nil
- }
- trie, _ := trie.New(root, trie.NewDatabase(rawdb.NewTable(h.chainDb, prefix)))
- return trie
- }
- // broadcastLoop broadcasts new block information to all connected light
- // clients. According to the agreement between client and server, server should
- // only broadcast new announcement if the total difficulty is higher than the
- // last one. Besides server will add the signature if client requires.
- func (h *serverHandler) broadcastLoop() {
- defer h.wg.Done()
- headCh := make(chan core.ChainHeadEvent, 10)
- headSub := h.blockchain.SubscribeChainHeadEvent(headCh)
- defer headSub.Unsubscribe()
- var (
- lastHead *types.Header
- lastTd = common.Big0
- )
- for {
- select {
- case ev := <-headCh:
- header := ev.Block.Header()
- hash, number := header.Hash(), header.Number.Uint64()
- td := h.blockchain.GetTd(hash, number)
- if td == nil || td.Cmp(lastTd) <= 0 {
- continue
- }
- var reorg uint64
- if lastHead != nil {
- reorg = lastHead.Number.Uint64() - rawdb.FindCommonAncestor(h.chainDb, header, lastHead).Number.Uint64()
- }
- lastHead, lastTd = header, td
- log.Debug("Announcing block to peers", "number", number, "hash", hash, "td", td, "reorg", reorg)
- h.server.peers.broadcast(announceData{Hash: hash, Number: number, Td: td, ReorgDepth: reorg})
- case <-h.closeCh:
- return
- }
- }
- }
|