mirror of
https://github.com/ethereum/go-ethereum.git
synced 2026-05-09 01:26:36 +00:00
0cba803fba
This changes the p2p protocol handlers to delay message decoding. It's the first part of a larger change that will delay decoding all the way through message processing. For responses, we delay the decoding until it is confirmed that the response matches an active request and does not exceed its limits. In order to make this work, all messages have been changed to use rlp.RawList instead of a slice of the decoded item type. For block bodies specifically, the decoding has been delayed all the way until after verification of the response hash. The role of p2p/tracker.Tracker changes significantly in this PR. The Tracker's original purpose was to maintain metrics about requests and responses in the peer-to-peer protocols. Each protocol maintained a single global Tracker instance. As of this change, the Tracker is now always active (regardless of metrics collection), and there is a separate instance of it for each peer. Whenever a response arrives, it is first verified that a request exists for it in the tracker. The tracker is also the place where limits are kept.
103 lines
4 KiB
Go
103 lines
4 KiB
Go
// Copyright 2021 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 downloader
|
|
|
|
import (
|
|
"time"
|
|
|
|
"github.com/ethereum/go-ethereum/common"
|
|
"github.com/ethereum/go-ethereum/eth/protocols/eth"
|
|
"github.com/ethereum/go-ethereum/log"
|
|
)
|
|
|
|
// bodyQueue implements typedQueue and is a type adapter between the generic
|
|
// concurrent fetcher and the downloader.
|
|
type bodyQueue Downloader
|
|
|
|
// waker returns a notification channel that gets pinged in case more body
|
|
// fetches have been queued up, so the fetcher might assign it to idle peers.
|
|
func (q *bodyQueue) waker() chan bool {
|
|
return q.queue.blockWakeCh
|
|
}
|
|
|
|
// pending returns the number of bodies that are currently queued for fetching
|
|
// by the concurrent downloader.
|
|
func (q *bodyQueue) pending() int {
|
|
return q.queue.PendingBodies()
|
|
}
|
|
|
|
// capacity is responsible for calculating how many bodies a particular peer is
|
|
// estimated to be able to retrieve within the allotted round trip time.
|
|
func (q *bodyQueue) capacity(peer *peerConnection, rtt time.Duration) int {
|
|
return peer.BodyCapacity(rtt)
|
|
}
|
|
|
|
// updateCapacity is responsible for updating how many bodies a particular peer
|
|
// is estimated to be able to retrieve in a unit time.
|
|
func (q *bodyQueue) updateCapacity(peer *peerConnection, items int, span time.Duration) {
|
|
peer.UpdateBodyRate(items, span)
|
|
}
|
|
|
|
// reserve is responsible for allocating a requested number of pending bodies
|
|
// from the download queue to the specified peer.
|
|
func (q *bodyQueue) reserve(peer *peerConnection, items int) (*fetchRequest, bool, bool) {
|
|
return q.queue.ReserveBodies(peer, items)
|
|
}
|
|
|
|
// unreserve is responsible for removing the current body retrieval allocation
|
|
// assigned to a specific peer and placing it back into the pool to allow
|
|
// reassigning to some other peer.
|
|
func (q *bodyQueue) unreserve(peer string) int {
|
|
fails := q.queue.ExpireBodies(peer)
|
|
if fails > 2 {
|
|
log.Trace("Body delivery timed out", "peer", peer)
|
|
} else {
|
|
log.Debug("Body delivery stalling", "peer", peer)
|
|
}
|
|
return fails
|
|
}
|
|
|
|
// request is responsible for converting a generic fetch request into a body
|
|
// one and sending it to the remote peer for fulfillment.
|
|
func (q *bodyQueue) request(peer *peerConnection, req *fetchRequest, resCh chan *eth.Response) (*eth.Request, error) {
|
|
peer.log.Trace("Requesting new batch of bodies", "count", len(req.Headers), "from", req.Headers[0].Number)
|
|
if q.bodyFetchHook != nil {
|
|
q.bodyFetchHook(req.Headers)
|
|
}
|
|
hashes := make([]common.Hash, 0, len(req.Headers))
|
|
for _, header := range req.Headers {
|
|
hashes = append(hashes, header.Hash())
|
|
}
|
|
return peer.peer.RequestBodies(hashes, resCh)
|
|
}
|
|
|
|
// deliver is responsible for taking a generic response packet from the concurrent
|
|
// fetcher, unpacking the body data and delivering it to the downloader's queue.
|
|
func (q *bodyQueue) deliver(peer *peerConnection, packet *eth.Response) (int, error) {
|
|
resp := packet.Res.(*eth.BlockBodiesResponse)
|
|
meta := packet.Meta.(eth.BlockBodyHashes)
|
|
accepted, err := q.queue.DeliverBodies(peer.id, meta, *resp)
|
|
switch {
|
|
case err == nil && len(*resp) == 0:
|
|
peer.log.Trace("Requested bodies delivered")
|
|
case err == nil:
|
|
peer.log.Trace("Delivered new batch of bodies", "count", len(*resp), "accepted", accepted)
|
|
default:
|
|
peer.log.Debug("Failed to deliver retrieved bodies", "err", err)
|
|
}
|
|
return accepted, err
|
|
}
|