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

les: move protocol definition to separate package

Browse source
This commit is contained in:
rjl493456442 2020-01-07 11:36:05 +08:00
parent 86d503ab15
commit 81aeba2798
20 changed files with 1021 additions and 867 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

3
ethstats/ethstats.go
View file

@ -38,6 +38,7 @@ import (
"github.com/ethereum/go-ethereum/eth"
"github.com/ethereum/go-ethereum/event"
"github.com/ethereum/go-ethereum/les"
lesproto "github.com/ethereum/go-ethereum/les/protocol"
"github.com/ethereum/go-ethereum/log"
"github.com/ethereum/go-ethereum/p2p"
"github.com/ethereum/go-ethereum/rpc"
@ -378,7 +379,7 @@ func (s *Service) login(conn *websocket.Conn) error {
protocol = fmt.Sprintf("eth/%d", eth.ProtocolVersions[0])
} else {
network = fmt.Sprintf("%d", infos.Protocols["les"].(*les.NodeInfo).Network)
protocol = fmt.Sprintf("les/%d", les.ClientProtocolVersions[0])
protocol = fmt.Sprintf("les/%d", lesproto.ClientProtocolVersions[0])
}
auth := &authMsg{
ID: s.node,

30
les/benchmark.go
View file

@ -30,6 +30,7 @@ import (
"github.com/ethereum/go-ethereum/core/types"
"github.com/ethereum/go-ethereum/crypto"
"github.com/ethereum/go-ethereum/les/flowcontrol"
"github.com/ethereum/go-ethereum/les/protocol"
"github.com/ethereum/go-ethereum/les/utilities"
"github.com/ethereum/go-ethereum/log"
"github.com/ethereum/go-ethereum/p2p"
@ -119,9 +120,9 @@ func (b *benchmarkProofsOrCode) request(peer *peer, index int) error {
key := make([]byte, 32)
rand.Read(key)
if b.code {
return peer.RequestCode(0, 0, []CodeReq{{BHash: b.headHash, AccKey: key}})
return peer.RequestCode(0, 0, []protocol.CodeRequest{{BlockHash: b.headHash, Account: key}})
} else {
return peer.RequestProofs(0, 0, []ProofReq{{BHash: b.headHash, Key: key}})
return peer.RequestProofs(0, 0, []protocol.TrieProofRequest{{BlockHash: b.headHash, Key: key}})
}
}
@ -146,7 +147,7 @@ func (b *benchmarkHelperTrie) init(h *serverHandler, count int) error {
}
func (b *benchmarkHelperTrie) request(peer *peer, index int) error {
reqs := make([]HelperTrieReq, b.reqCount)
reqs := make([]protocol.HelperTrieRequest, b.reqCount)
if b.bloom {
bitIdx := uint16(rand.Intn(2048))
@ -154,13 +155,22 @@ func (b *benchmarkHelperTrie) request(peer *peer, index int) error {
key := make([]byte, 10)
binary.BigEndian.PutUint16(key[:2], bitIdx)
binary.BigEndian.PutUint64(key[2:], uint64(rand.Int63n(int64(b.sectionCount))))
reqs[i] = HelperTrieReq{Type: htBloomBits, TrieIdx: b.sectionCount - 1, Key: key}
reqs[i] = protocol.HelperTrieRequest{
Type: protocol.HelperTrieBloomTrie,
TrieIndex: b.sectionCount - 1,
Key: key,
}
}
} else {
for i := range reqs {
key := make([]byte, 8)
binary.BigEndian.PutUint64(key[:], uint64(rand.Int63n(int64(b.headNum))))
reqs[i] = HelperTrieReq{Type: htCanonical, TrieIdx: b.sectionCount - 1, Key: key, AuxReq: auxHeader}
reqs[i] = protocol.HelperTrieRequest{
Type: protocol.HelperTrieCHT,
TrieIndex: b.sectionCount - 1,
Key: key,
AuxType: protocol.AuxHeader,
}
}
}
@ -284,13 +294,13 @@ func (h *serverHandler) measure(setup *benchmarkSetup, count int) error {
var id enode.ID
rand.Read(id[:])
clientPeer := newPeer(lpv2, NetworkId, false, p2p.NewPeer(id, "client", nil), clientMeteredPipe)
serverPeer := newPeer(lpv2, NetworkId, false, p2p.NewPeer(id, "server", nil), serverMeteredPipe)
clientPeer := newPeer(protocol.Lpv2, protocol.NetworkId, false, p2p.NewPeer(id, "client", nil), clientMeteredPipe)
serverPeer := newPeer(protocol.Lpv2, protocol.NetworkId, false, p2p.NewPeer(id, "server", nil), serverMeteredPipe)
serverPeer.sendQueue = utilities.NewExecQueue(count)
serverPeer.announceType = announceTypeNone
serverPeer.fcCosts = make(requestCostTable)
c := &requestCosts{}
for code := range requests {
serverPeer.fcCosts = make(protocol.RequestCostTable)
c := &protocol.RequestCost{}
for code := range protocol.LesRequests {
serverPeer.fcCosts[code] = c
}
serverPeer.fcParams = flowcontrol.ServerParams{BufLimit: 1, MinRecharge: 1}

9
les/client.go
View file

@ -37,6 +37,7 @@ import (
"github.com/ethereum/go-ethereum/event"
"github.com/ethereum/go-ethereum/internal/ethapi"
"github.com/ethereum/go-ethereum/les/checkpointoracle"
"github.com/ethereum/go-ethereum/les/protocol"
"github.com/ethereum/go-ethereum/light"
"github.com/ethereum/go-ethereum/log"
"github.com/ethereum/go-ethereum/node"
@ -217,14 +218,14 @@ func (s *LightEthereum) ResetWithGenesisBlock(gb *types.Block) {
func (s *LightEthereum) BlockChain() *light.LightChain { return s.blockchain }
func (s *LightEthereum) TxPool() *light.TxPool { return s.txPool }
func (s *LightEthereum) Engine() consensus.Engine { return s.engine }
func (s *LightEthereum) LesVersion() int { return int(ClientProtocolVersions[0]) }
func (s *LightEthereum) LesVersion() int { return int(protocol.ClientProtocolVersions[0]) }
func (s *LightEthereum) Downloader() *downloader.Downloader { return s.handler.downloader }
func (s *LightEthereum) EventMux() *event.TypeMux { return s.eventMux }
// Protocols implements node.Service, returning all the currently configured
// network protocols to start.
func (s *LightEthereum) Protocols() []p2p.Protocol {
return s.makeProtocols(ClientProtocolVersions, s.handler.runPeer, func(id enode.ID) interface{} {
return s.makeProtocols(protocol.ClientProtocolVersions, s.handler.runPeer, func(id enode.ID) interface{} {
if p := s.peers.Peer(peerIdToString(id)); p != nil {
return p.Info()
}
@ -244,8 +245,8 @@ func (s *LightEthereum) Start(srvr *p2p.Server) error {
s.netRPCService = ethapi.NewPublicNetAPI(srvr, s.config.NetworkId)
// clients are searching for the first advertised protocol in the list
protocolVersion := AdvertiseProtocolVersions[0]
s.serverPool.start(srvr, lesTopic(s.blockchain.Genesis().Hash(), protocolVersion))
protocolVersion := protocol.AdvertiseProtocolVersions[0]
s.serverPool.start(srvr, protocol.LesTopic(s.blockchain.Genesis().Hash(), protocolVersion))
return nil
}

75
les/client_handler.go
View file

@ -25,6 +25,7 @@ import (
"github.com/ethereum/go-ethereum/common/mclock"
"github.com/ethereum/go-ethereum/core/types"
"github.com/ethereum/go-ethereum/eth/downloader"
"github.com/ethereum/go-ethereum/les/protocol"
"github.com/ethereum/go-ethereum/light"
"github.com/ethereum/go-ethereum/log"
"github.com/ethereum/go-ethereum/p2p"
@ -151,8 +152,8 @@ func (h *clientHandler) handleMsg(p *peer) error {
}
p.Log().Trace("Light Ethereum message arrived", "code", msg.Code, "bytes", msg.Size)
if msg.Size > ProtocolMaxMsgSize {
return errResp(ErrMsgTooLarge, "%v > %v", msg.Size, ProtocolMaxMsgSize)
if msg.Size > protocol.ProtocolMaxMsgSize {
return protocol.ErrResp(protocol.ErrMsgTooLarge, "%v > %v", msg.Size, protocol.ProtocolMaxMsgSize)
}
defer msg.Discard()
@ -160,43 +161,43 @@ func (h *clientHandler) handleMsg(p *peer) error {
// Handle the message depending on its contents
switch msg.Code {
case AnnounceMsg:
case protocol.AnnounceMsg:
p.Log().Trace("Received announce message")
var req announceData
if err := msg.Decode(&req); err != nil {
return errResp(ErrDecode, "%v: %v", msg, err)
var anno protocol.Announcement
if err := msg.Decode(&anno); err != nil {
return protocol.ErrResp(protocol.ErrDecode, "%v: %v", msg, err)
}
if err := req.sanityCheck(); err != nil {
if err := anno.SanityCheck(); err != nil {
return err
}
update, size := req.Update.decode()
update, size := anno.Update.ToMap()
if p.rejectUpdate(size) {
return errResp(ErrRequestRejected, "")
return protocol.ErrResp(protocol.ErrRequestRejected, "")
}
p.updateFlowControl(update)
if req.Hash != (common.Hash{}) {
if anno.Hash != (common.Hash{}) {
if p.announceType == announceTypeNone {
return errResp(ErrUnexpectedResponse, "")
return protocol.ErrResp(protocol.ErrUnexpectedResponse, "")
}
if p.announceType == announceTypeSigned {
if err := req.checkSignature(p.ID(), update); err != nil {
if err := anno.CheckSignature(p.ID(), update); err != nil {
p.Log().Trace("Invalid announcement signature", "err", err)
return err
}
p.Log().Trace("Valid announcement signature")
}
p.Log().Trace("Announce message content", "number", req.Number, "hash", req.Hash, "td", req.Td, "reorg", req.ReorgDepth)
h.fetcher.announce(p, &req)
p.Log().Trace("Announce message content", "number", anno.Number, "hash", anno.Hash, "td", anno.Td, "reorg", anno.ReorgDepth)
h.fetcher.announce(p, &anno)
}
case BlockHeadersMsg:
case protocol.BlockHeadersMsg:
p.Log().Trace("Received block header response message")
var resp struct {
ReqID, BV uint64
Headers []*types.Header
}
if err := msg.Decode(&resp); err != nil {
return errResp(ErrDecode, "msg %v: %v", msg, err)
return protocol.ErrResp(protocol.ErrDecode, "msg %v: %v", msg, err)
}
p.fcServer.ReceivedReply(resp.ReqID, resp.BV)
if h.fetcher.requestedID(resp.ReqID) {
@ -206,14 +207,14 @@ func (h *clientHandler) handleMsg(p *peer) error {
log.Debug("Failed to deliver headers", "err", err)
}
}
case BlockBodiesMsg:
case protocol.BlockBodiesMsg:
p.Log().Trace("Received block bodies response")
var resp struct {
ReqID, BV uint64
Data []*types.Body
}
if err := msg.Decode(&resp); err != nil {
return errResp(ErrDecode, "msg %v: %v", msg, err)
return protocol.ErrResp(protocol.ErrDecode, "msg %v: %v", msg, err)
}
p.fcServer.ReceivedReply(resp.ReqID, resp.BV)
deliverMsg = &Msg{
@ -221,14 +222,14 @@ func (h *clientHandler) handleMsg(p *peer) error {
ReqID: resp.ReqID,
Obj: resp.Data,
}
case CodeMsg:
case protocol.CodeMsg:
p.Log().Trace("Received code response")
var resp struct {
ReqID, BV uint64
Data [][]byte
}
if err := msg.Decode(&resp); err != nil {
return errResp(ErrDecode, "msg %v: %v", msg, err)
return protocol.ErrResp(protocol.ErrDecode, "msg %v: %v", msg, err)
}
p.fcServer.ReceivedReply(resp.ReqID, resp.BV)
deliverMsg = &Msg{
@ -236,14 +237,14 @@ func (h *clientHandler) handleMsg(p *peer) error {
ReqID: resp.ReqID,
Obj: resp.Data,
}
case ReceiptsMsg:
case protocol.ReceiptsMsg:
p.Log().Trace("Received receipts response")
var resp struct {
ReqID, BV uint64
Receipts []types.Receipts
}
if err := msg.Decode(&resp); err != nil {
return errResp(ErrDecode, "msg %v: %v", msg, err)
return protocol.ErrResp(protocol.ErrDecode, "msg %v: %v", msg, err)
}
p.fcServer.ReceivedReply(resp.ReqID, resp.BV)
deliverMsg = &Msg{
@ -251,14 +252,14 @@ func (h *clientHandler) handleMsg(p *peer) error {
ReqID: resp.ReqID,
Obj: resp.Receipts,
}
case ProofsV2Msg:
case protocol.ProofsV2Msg:
p.Log().Trace("Received les/2 proofs response")
var resp struct {
ReqID, BV uint64
Data light.NodeList
}
if err := msg.Decode(&resp); err != nil {
return errResp(ErrDecode, "msg %v: %v", msg, err)
return protocol.ErrResp(protocol.ErrDecode, "msg %v: %v", msg, err)
}
p.fcServer.ReceivedReply(resp.ReqID, resp.BV)
deliverMsg = &Msg{
@ -266,14 +267,14 @@ func (h *clientHandler) handleMsg(p *peer) error {
ReqID: resp.ReqID,
Obj: resp.Data,
}
case HelperTrieProofsMsg:
case protocol.HelperTrieProofsMsg:
p.Log().Trace("Received helper trie proof response")
var resp struct {
ReqID, BV uint64
Data HelperTrieResps
Data protocol.HelperTrieResponse
}
if err := msg.Decode(&resp); err != nil {
return errResp(ErrDecode, "msg %v: %v", msg, err)
return protocol.ErrResp(protocol.ErrDecode, "msg %v: %v", msg, err)
}
p.fcServer.ReceivedReply(resp.ReqID, resp.BV)
deliverMsg = &Msg{
@ -281,14 +282,14 @@ func (h *clientHandler) handleMsg(p *peer) error {
ReqID: resp.ReqID,
Obj: resp.Data,
}
case TxStatusMsg:
case protocol.TxStatusMsg:
p.Log().Trace("Received tx status response")
var resp struct {
ReqID, BV uint64
Status []light.TxStatus
}
if err := msg.Decode(&resp); err != nil {
return errResp(ErrDecode, "msg %v: %v", msg, err)
return protocol.ErrResp(protocol.ErrDecode, "msg %v: %v", msg, err)
}
p.fcServer.ReceivedReply(resp.ReqID, resp.BV)
deliverMsg = &Msg{
@ -296,21 +297,21 @@ func (h *clientHandler) handleMsg(p *peer) error {
ReqID: resp.ReqID,
Obj: resp.Status,
}
case StopMsg:
case protocol.StopMsg:
p.freezeServer(true)
h.backend.retriever.frozen(p)
p.Log().Debug("Service stopped")
case ResumeMsg:
case protocol.ResumeMsg:
var bv uint64
if err := msg.Decode(&bv); err != nil {
return errResp(ErrDecode, "msg %v: %v", msg, err)
return protocol.ErrResp(protocol.ErrDecode, "msg %v: %v", msg, err)
}
p.fcServer.ResumeFreeze(bv)
p.freezeServer(false)
p.Log().Debug("Service resumed")
default:
p.Log().Trace("Received invalid message", "code", msg.Code)
return errResp(ErrInvalidMsgCode, "%v", msg.Code)
return protocol.ErrResp(protocol.ErrInvalidMsgCode, "%v", msg.Code)
}
// Deliver the received response to retriever.
if deliverMsg != nil {
@ -341,7 +342,7 @@ func (pc *peerConnection) RequestHeadersByHash(origin common.Hash, amount int, s
rq := &distReq{
getCost: func(dp distPeer) uint64 {
peer := dp.(*peer)
return peer.GetRequestCost(GetBlockHeadersMsg, amount)
return peer.GetRequestCost(protocol.GetBlockHeadersMsg, amount)
},
canSend: func(dp distPeer) bool {
return dp.(*peer) == pc.peer
@ -349,7 +350,7 @@ func (pc *peerConnection) RequestHeadersByHash(origin common.Hash, amount int, s
request: func(dp distPeer) func() {
reqID := genReqID()
peer := dp.(*peer)
cost := peer.GetRequestCost(GetBlockHeadersMsg, amount)
cost := peer.GetRequestCost(protocol.GetBlockHeadersMsg, amount)
peer.fcServer.QueuedRequest(reqID, cost)
return func() { peer.RequestHeadersByHash(reqID, cost, origin, amount, skip, reverse) }
},
@ -365,7 +366,7 @@ func (pc *peerConnection) RequestHeadersByNumber(origin uint64, amount int, skip
rq := &distReq{
getCost: func(dp distPeer) uint64 {
peer := dp.(*peer)
return peer.GetRequestCost(GetBlockHeadersMsg, amount)
return peer.GetRequestCost(protocol.GetBlockHeadersMsg, amount)
},
canSend: func(dp distPeer) bool {
return dp.(*peer) == pc.peer
@ -373,7 +374,7 @@ func (pc *peerConnection) RequestHeadersByNumber(origin uint64, amount int, skip
request: func(dp distPeer) func() {
reqID := genReqID()
peer := dp.(*peer)
cost := peer.GetRequestCost(GetBlockHeadersMsg, amount)
cost := peer.GetRequestCost(protocol.GetBlockHeadersMsg, amount)
peer.fcServer.QueuedRequest(reqID, cost)
return func() { peer.RequestHeadersByNumber(reqID, cost, origin, amount, skip, reverse) }
},

20
les/commons.go
View file

@ -17,7 +17,6 @@
package les
import (
"fmt"
"math/big"
"sync"
@ -28,28 +27,13 @@ import (
"github.com/ethereum/go-ethereum/eth"
"github.com/ethereum/go-ethereum/ethdb"
"github.com/ethereum/go-ethereum/les/checkpointoracle"
"github.com/ethereum/go-ethereum/les/protocol"
"github.com/ethereum/go-ethereum/light"
"github.com/ethereum/go-ethereum/p2p"
"github.com/ethereum/go-ethereum/p2p/discv5"
"github.com/ethereum/go-ethereum/p2p/enode"
"github.com/ethereum/go-ethereum/params"
)
func errResp(code errCode, format string, v ...interface{}) error {
return fmt.Errorf("%v - %v", code, fmt.Sprintf(format, v...))
}
func lesTopic(genesisHash common.Hash, protocolVersion uint) discv5.Topic {
var name string
switch protocolVersion {
case lpv2:
name = "LES2"
default:
panic(nil)
}
return discv5.Topic(name + "@" + common.Bytes2Hex(genesisHash.Bytes()[0:8]))
}
type chainReader interface {
CurrentHeader() *types.Header
}
@ -89,7 +73,7 @@ func (c *lesCommons) makeProtocols(versions []uint, runPeer func(version uint, p
protos[i] = p2p.Protocol{
Name: "les",
Version: version,
Length: ProtocolLengths[version],
Length: protocol.ProtocolLengths[version],
NodeInfo: c.nodeInfo,
Run: func(peer *p2p.Peer, rw p2p.MsgReadWriter) error {
return runPeer(version, peer, rw)

144
les/costtracker.go
View file

@ -27,6 +27,7 @@ import (
"github.com/ethereum/go-ethereum/eth"
"github.com/ethereum/go-ethereum/ethdb"
"github.com/ethereum/go-ethereum/les/flowcontrol"
"github.com/ethereum/go-ethereum/les/protocol"
"github.com/ethereum/go-ethereum/log"
"github.com/ethereum/go-ethereum/metrics"
)
@ -35,48 +36,48 @@ const makeCostStats = false // make request cost statistics during operation
var (
// average request cost estimates based on serving time
reqAvgTimeCost = requestCostTable{
GetBlockHeadersMsg: {150000, 30000},
GetBlockBodiesMsg: {0, 700000},
GetReceiptsMsg: {0, 1000000},
GetCodeMsg: {0, 450000},
GetProofsV2Msg: {0, 600000},
GetHelperTrieProofsMsg: {0, 1000000},
SendTxV2Msg: {0, 450000},
GetTxStatusMsg: {0, 250000},
reqAvgTimeCost = protocol.RequestCostTable{
protocol.GetBlockHeadersMsg: {150000, 30000},
protocol.GetBlockBodiesMsg: {0, 700000},
protocol.GetReceiptsMsg: {0, 1000000},
protocol.GetCodeMsg: {0, 450000},
protocol.GetProofsV2Msg: {0, 600000},
protocol.GetHelperTrieProofsMsg: {0, 1000000},
protocol.SendTxV2Msg: {0, 450000},
protocol.GetTxStatusMsg: {0, 250000},
}
// maximum incoming message size estimates
reqMaxInSize = requestCostTable{
GetBlockHeadersMsg: {40, 0},
GetBlockBodiesMsg: {0, 40},
GetReceiptsMsg: {0, 40},
GetCodeMsg: {0, 80},
GetProofsV2Msg: {0, 80},
GetHelperTrieProofsMsg: {0, 20},
SendTxV2Msg: {0, 16500},
GetTxStatusMsg: {0, 50},
reqMaxInSize = protocol.RequestCostTable{
protocol.GetBlockHeadersMsg: {40, 0},
protocol.GetBlockBodiesMsg: {0, 40},
protocol.GetReceiptsMsg: {0, 40},
protocol.GetCodeMsg: {0, 80},
protocol.GetProofsV2Msg: {0, 80},
protocol.GetHelperTrieProofsMsg: {0, 20},
protocol.SendTxV2Msg: {0, 16500},
protocol.GetTxStatusMsg: {0, 50},
}
// maximum outgoing message size estimates
reqMaxOutSize = requestCostTable{
GetBlockHeadersMsg: {0, 556},
GetBlockBodiesMsg: {0, 100000},
GetReceiptsMsg: {0, 200000},
GetCodeMsg: {0, 50000},
GetProofsV2Msg: {0, 4000},
GetHelperTrieProofsMsg: {0, 4000},
SendTxV2Msg: {0, 100},
GetTxStatusMsg: {0, 100},
reqMaxOutSize = protocol.RequestCostTable{
protocol.GetBlockHeadersMsg: {0, 556},
protocol.GetBlockBodiesMsg: {0, 100000},
protocol.GetReceiptsMsg: {0, 200000},
protocol.GetCodeMsg: {0, 50000},
protocol.GetProofsV2Msg: {0, 4000},
protocol.GetHelperTrieProofsMsg: {0, 4000},
protocol.SendTxV2Msg: {0, 100},
protocol.GetTxStatusMsg: {0, 100},
}
// request amounts that have to fit into the minimum buffer size minBufferMultiplier times
minBufferReqAmount = map[uint64]uint64{
GetBlockHeadersMsg: 192,
GetBlockBodiesMsg: 1,
GetReceiptsMsg: 1,
GetCodeMsg: 1,
GetProofsV2Msg: 1,
GetHelperTrieProofsMsg: 16,
SendTxV2Msg: 8,
GetTxStatusMsg: 64,
protocol.GetBlockHeadersMsg: 192,
protocol.GetBlockBodiesMsg: 1,
protocol.GetReceiptsMsg: 1,
protocol.GetCodeMsg: 1,
protocol.GetProofsV2Msg: 1,
protocol.GetHelperTrieProofsMsg: 16,
protocol.SendTxV2Msg: 8,
protocol.GetTxStatusMsg: 64,
}
minBufferMultiplier = 3
)
@ -132,7 +133,7 @@ type costTracker struct {
// TestHooks
testing bool // Disable real cost evaluation for testing purpose.
testCostList RequestCostList // Customized cost table for testing purpose.
testCostList protocol.RequestCostList // Customized cost table for testing purpose.
}
// newCostTracker creates a cost tracker and loads the cost factor statistics from the database.
@ -182,7 +183,7 @@ func (ct *costTracker) stop() {
// makeCostList returns upper cost estimates based on the hardcoded cost estimate
// tables and the optionally specified incoming/outgoing bandwidth limits
func (ct *costTracker) makeCostList(globalFactor float64) RequestCostList {
func (ct *costTracker) makeCostList(globalFactor float64) protocol.RequestCostList {
maxCost := func(avgTimeCost, inSize, outSize uint64) uint64 {
cost := avgTimeCost * maxCostFactor
inSizeCost := uint64(float64(inSize) * ct.inSizeFactor * globalFactor)
@ -195,10 +196,10 @@ func (ct *costTracker) makeCostList(globalFactor float64) RequestCostList {
}
return cost
}
var list RequestCostList
var list protocol.RequestCostList
for code, data := range reqAvgTimeCost {
baseCost := maxCost(data.baseCost, reqMaxInSize[code].baseCost, reqMaxOutSize[code].baseCost)
reqCost := maxCost(data.reqCost, reqMaxInSize[code].reqCost, reqMaxOutSize[code].reqCost)
baseCost := maxCost(data.BaseCost, reqMaxInSize[code].BaseCost, reqMaxOutSize[code].BaseCost)
reqCost := maxCost(data.ReqCost, reqMaxInSize[code].ReqCost, reqMaxOutSize[code].ReqCost)
if ct.minBufLimit != 0 {
// if minBufLimit is set then always enforce maximum request cost <= minBufLimit
maxCost := baseCost + reqCost*minBufferReqAmount[code]
@ -209,7 +210,7 @@ func (ct *costTracker) makeCostList(globalFactor float64) RequestCostList {
}
}
list = append(list, requestCostListItem{
list = append(list, protocol.RequestCostListItem{
MsgCode: code,
BaseCost: baseCost,
ReqCost: reqCost,
@ -278,21 +279,21 @@ func (ct *costTracker) gfLoop() {
// Record more metrics if we are debugging
if metrics.EnabledExpensive {
switch r.msgCode {
case GetBlockHeadersMsg:
case protocol.GetBlockHeadersMsg:
relativeCostHeaderHistogram.Update(relCost)
case GetBlockBodiesMsg:
case protocol.GetBlockBodiesMsg:
relativeCostBodyHistogram.Update(relCost)
case GetReceiptsMsg:
case protocol.GetReceiptsMsg:
relativeCostReceiptHistogram.Update(relCost)
case GetCodeMsg:
case protocol.GetCodeMsg:
relativeCostCodeHistogram.Update(relCost)
case GetProofsV2Msg:
case protocol.GetProofsV2Msg:
relativeCostProofHistogram.Update(relCost)
case GetHelperTrieProofsMsg:
case protocol.GetHelperTrieProofsMsg:
relativeCostHelperProofHistogram.Update(relCost)
case SendTxV2Msg:
case protocol.SendTxV2Msg:
relativeCostSendTxHistogram.Update(relCost)
case GetTxStatusMsg:
case protocol.GetTxStatusMsg:
relativeCostTxStatusHistogram.Update(relCost)
}
}
@ -302,7 +303,7 @@ func (ct *costTracker) gfLoop() {
// requests involve txpool query, which is usually unstable.
//
// TODO(rjl493456442) fixes this.
if r.msgCode == SendTxV2Msg || r.msgCode == GetTxStatusMsg {
if r.msgCode == protocol.SendTxV2Msg || r.msgCode == protocol.GetTxStatusMsg {
continue
}
requestServedMeter.Mark(int64(r.servingTime))
@ -410,7 +411,7 @@ func (ct *costTracker) subscribeTotalRecharge(ch chan uint64) uint64 {
// average estimate statistics
func (ct *costTracker) updateStats(code, amount, servingTime, realCost uint64) {
avg := reqAvgTimeCost[code]
avgTimeCost := avg.baseCost + amount*avg.reqCost
avgTimeCost := avg.BaseCost + amount*avg.ReqCost
select {
case ct.reqInfoCh <- reqInfo{float64(avgTimeCost), float64(servingTime), code}:
default:
@ -457,46 +458,9 @@ func (ct *costTracker) printStats() {
}
}
type (
// requestCostTable assigns a cost estimate function to each request type
// which is a linear function of the requested amount
// (cost = baseCost + reqCost * amount)
requestCostTable map[uint64]*requestCosts
requestCosts struct {
baseCost, reqCost uint64
}
// RequestCostList is a list representation of request costs which is used for
// database storage and communication through the network
RequestCostList []requestCostListItem
requestCostListItem struct {
MsgCode, BaseCost, ReqCost uint64
}
)
// getMaxCost calculates the estimated cost for a given request type and amount
func (table requestCostTable) getMaxCost(code, amount uint64) uint64 {
costs := table[code]
return costs.baseCost + amount*costs.reqCost
}
// decode converts a cost list to a cost table
func (list RequestCostList) decode(protocolLength uint64) requestCostTable {
table := make(requestCostTable)
for _, e := range list {
if e.MsgCode < protocolLength {
table[e.MsgCode] = &requestCosts{
baseCost: e.BaseCost,
reqCost: e.ReqCost,
}
}
}
return table
}
// testCostList returns a dummy request cost list used by tests
func testCostList(testCost uint64) RequestCostList {
cl := make(RequestCostList, len(reqAvgTimeCost))
func testCostList(testCost uint64) protocol.RequestCostList {
cl := make(protocol.RequestCostList, len(reqAvgTimeCost))
var max uint64
for code := range reqAvgTimeCost {
if code > max {

7
les/fetcher.go
View file

@ -26,6 +26,7 @@ import (
"github.com/ethereum/go-ethereum/consensus"
"github.com/ethereum/go-ethereum/core/rawdb"
"github.com/ethereum/go-ethereum/core/types"
"github.com/ethereum/go-ethereum/les/protocol"
"github.com/ethereum/go-ethereum/light"
"github.com/ethereum/go-ethereum/log"
)
@ -250,7 +251,7 @@ func (f *lightFetcher) unregisterPeer(p *peer) {
// announce processes a new announcement message received from a peer, adding new
// nodes to the peer's block tree and removing old nodes if necessary
func (f *lightFetcher) announce(p *peer, head *announceData) {
func (f *lightFetcher) announce(p *peer, head *protocol.Announcement) {
f.lock.Lock()
defer f.lock.Unlock()
p.Log().Debug("Received new announcement", "number", head.Number, "hash", head.Hash, "reorg", head.ReorgDepth)
@ -519,7 +520,7 @@ func (f *lightFetcher) newFetcherDistReq(bestHash common.Hash, reqID uint64, bes
return &distReq{
getCost: func(dp distPeer) uint64 {
p := dp.(*peer)
return p.GetRequestCost(GetBlockHeadersMsg, int(bestAmount))
return p.GetRequestCost(protocol.GetBlockHeadersMsg, int(bestAmount))
},
canSend: func(dp distPeer) bool {
p := dp.(*peer)
@ -548,7 +549,7 @@ func (f *lightFetcher) newFetcherDistReq(bestHash common.Hash, reqID uint64, bes
}
f.lock.Unlock()
cost := p.GetRequestCost(GetBlockHeadersMsg, int(bestAmount))
cost := p.GetRequestCost(protocol.GetBlockHeadersMsg, int(bestAmount))
p.fcServer.QueuedRequest(reqID, cost)
f.reqMu.Lock()
f.requested[reqID] = fetchRequest{hash: bestHash, amount: bestAmount, peer: p, sent: mclock.Now()}

193
les/handler_test.go
View file

@ -31,6 +31,7 @@ import (
"github.com/ethereum/go-ethereum/core/types"
"github.com/ethereum/go-ethereum/crypto"
"github.com/ethereum/go-ethereum/eth/downloader"
"github.com/ethereum/go-ethereum/les/protocol"
"github.com/ethereum/go-ethereum/light"
"github.com/ethereum/go-ethereum/p2p"
"github.com/ethereum/go-ethereum/params"
@ -50,8 +51,8 @@ func expectResponse(r p2p.MsgReader, msgcode, reqID, bv uint64, data interface{}
func TestGetBlockHeadersLes2(t *testing.T) { testGetBlockHeaders(t, 2) }
func TestGetBlockHeadersLes3(t *testing.T) { testGetBlockHeaders(t, 3) }
func testGetBlockHeaders(t *testing.T, protocol int) {
server, tearDown := newServerEnv(t, downloader.MaxHashFetch+15, protocol, nil, false, true, 0)
func testGetBlockHeaders(t *testing.T, p int) {
server, tearDown := newServerEnv(t, downloader.MaxHashFetch+15, p, nil, false, true, 0)
defer tearDown()
bc := server.handler.blockchain
@ -62,29 +63,29 @@ func testGetBlockHeaders(t *testing.T, protocol int) {
unknown[i] = byte(i)
}
// Create a batch of tests for various scenarios
limit := uint64(MaxHeaderFetch)
limit := uint64(protocol.MaxHeaderFetch)
tests := []struct {
query *getBlockHeadersData // The query to execute for header retrieval
query *protocol.GetBlockHeadersRequest // The query to execute for header retrieval
expect []common.Hash // The hashes of the block whose headers are expected
}{
// A single random block should be retrievable by hash and number too
{
&getBlockHeadersData{Origin: hashOrNumber{Hash: bc.GetBlockByNumber(limit / 2).Hash()}, Amount: 1},
&protocol.GetBlockHeadersRequest{Origin: protocol.HashOrNumber{Hash: bc.GetBlockByNumber(limit / 2).Hash()}, Amount: 1},
[]common.Hash{bc.GetBlockByNumber(limit / 2).Hash()},
}, {
&getBlockHeadersData{Origin: hashOrNumber{Number: limit / 2}, Amount: 1},
&protocol.GetBlockHeadersRequest{Origin: protocol.HashOrNumber{Number: limit / 2}, Amount: 1},
[]common.Hash{bc.GetBlockByNumber(limit / 2).Hash()},
},
// Multiple headers should be retrievable in both directions
{
&getBlockHeadersData{Origin: hashOrNumber{Number: limit / 2}, Amount: 3},
&protocol.GetBlockHeadersRequest{Origin: protocol.HashOrNumber{Number: limit / 2}, Amount: 3},
[]common.Hash{
bc.GetBlockByNumber(limit / 2).Hash(),
bc.GetBlockByNumber(limit/2 + 1).Hash(),
bc.GetBlockByNumber(limit/2 + 2).Hash(),
},
}, {
&getBlockHeadersData{Origin: hashOrNumber{Number: limit / 2}, Amount: 3, Reverse: true},
&protocol.GetBlockHeadersRequest{Origin: protocol.HashOrNumber{Number: limit / 2}, Amount: 3, Reverse: true},
[]common.Hash{
bc.GetBlockByNumber(limit / 2).Hash(),
bc.GetBlockByNumber(limit/2 - 1).Hash(),
@ -93,14 +94,14 @@ func testGetBlockHeaders(t *testing.T, protocol int) {
},
// Multiple headers with skip lists should be retrievable
{
&getBlockHeadersData{Origin: hashOrNumber{Number: limit / 2}, Skip: 3, Amount: 3},
&protocol.GetBlockHeadersRequest{Origin: protocol.HashOrNumber{Number: limit / 2}, Skip: 3, Amount: 3},
[]common.Hash{
bc.GetBlockByNumber(limit / 2).Hash(),
bc.GetBlockByNumber(limit/2 + 4).Hash(),
bc.GetBlockByNumber(limit/2 + 8).Hash(),
},
}, {
&getBlockHeadersData{Origin: hashOrNumber{Number: limit / 2}, Skip: 3, Amount: 3, Reverse: true},
&protocol.GetBlockHeadersRequest{Origin: protocol.HashOrNumber{Number: limit / 2}, Skip: 3, Amount: 3, Reverse: true},
[]common.Hash{
bc.GetBlockByNumber(limit / 2).Hash(),
bc.GetBlockByNumber(limit/2 - 4).Hash(),
@ -109,26 +110,26 @@ func testGetBlockHeaders(t *testing.T, protocol int) {
},
// The chain endpoints should be retrievable
{
&getBlockHeadersData{Origin: hashOrNumber{Number: 0}, Amount: 1},
&protocol.GetBlockHeadersRequest{Origin: protocol.HashOrNumber{Number: 0}, Amount: 1},
[]common.Hash{bc.GetBlockByNumber(0).Hash()},
}, {
&getBlockHeadersData{Origin: hashOrNumber{Number: bc.CurrentBlock().NumberU64()}, Amount: 1},
&protocol.GetBlockHeadersRequest{Origin: protocol.HashOrNumber{Number: bc.CurrentBlock().NumberU64()}, Amount: 1},
[]common.Hash{bc.CurrentBlock().Hash()},
},
// Ensure protocol limits are honored
// Ensure p limits are honored
//{
// &getBlockHeadersData{Origin: hashOrNumber{Number: bc.CurrentBlock().NumberU64() - 1}, Amount: limit + 10, Reverse: true},
// &protocol.GetBlockHeadersRequest{Origin: protocol.HashOrNumber{Number: bc.CurrentBlock().NumberU64() - 1}, Amount: limit + 10, Reverse: true},
// []common.Hash{},
//},
// Check that requesting more than available is handled gracefully
{
&getBlockHeadersData{Origin: hashOrNumber{Number: bc.CurrentBlock().NumberU64() - 4}, Skip: 3, Amount: 3},
&protocol.GetBlockHeadersRequest{Origin: protocol.HashOrNumber{Number: bc.CurrentBlock().NumberU64() - 4}, Skip: 3, Amount: 3},
[]common.Hash{
bc.GetBlockByNumber(bc.CurrentBlock().NumberU64() - 4).Hash(),
bc.GetBlockByNumber(bc.CurrentBlock().NumberU64()).Hash(),
},
}, {
&getBlockHeadersData{Origin: hashOrNumber{Number: 4}, Skip: 3, Amount: 3, Reverse: true},
&protocol.GetBlockHeadersRequest{Origin: protocol.HashOrNumber{Number: 4}, Skip: 3, Amount: 3, Reverse: true},
[]common.Hash{
bc.GetBlockByNumber(4).Hash(),
bc.GetBlockByNumber(0).Hash(),
@ -136,13 +137,13 @@ func testGetBlockHeaders(t *testing.T, protocol int) {
},
// Check that requesting more than available is handled gracefully, even if mid skip
{
&getBlockHeadersData{Origin: hashOrNumber{Number: bc.CurrentBlock().NumberU64() - 4}, Skip: 2, Amount: 3},
&protocol.GetBlockHeadersRequest{Origin: protocol.HashOrNumber{Number: bc.CurrentBlock().NumberU64() - 4}, Skip: 2, Amount: 3},
[]common.Hash{
bc.GetBlockByNumber(bc.CurrentBlock().NumberU64() - 4).Hash(),
bc.GetBlockByNumber(bc.CurrentBlock().NumberU64() - 1).Hash(),
},
}, {
&getBlockHeadersData{Origin: hashOrNumber{Number: 4}, Skip: 2, Amount: 3, Reverse: true},
&protocol.GetBlockHeadersRequest{Origin: protocol.HashOrNumber{Number: 4}, Skip: 2, Amount: 3, Reverse: true},
[]common.Hash{
bc.GetBlockByNumber(4).Hash(),
bc.GetBlockByNumber(1).Hash(),
@ -150,10 +151,10 @@ func testGetBlockHeaders(t *testing.T, protocol int) {
},
// Check that non existing headers aren't returned
{
&getBlockHeadersData{Origin: hashOrNumber{Hash: unknown}, Amount: 1},
&protocol.GetBlockHeadersRequest{Origin: protocol.HashOrNumber{Hash: unknown}, Amount: 1},
[]common.Hash{},
}, {
&getBlockHeadersData{Origin: hashOrNumber{Number: bc.CurrentBlock().NumberU64() + 1}, Amount: 1},
&protocol.GetBlockHeadersRequest{Origin: protocol.HashOrNumber{Number: bc.CurrentBlock().NumberU64() + 1}, Amount: 1},
[]common.Hash{},
},
}
@ -168,9 +169,9 @@ func testGetBlockHeaders(t *testing.T, protocol int) {
// Send the hash request and verify the response
reqID++
cost := server.peer.peer.GetRequestCost(GetBlockHeadersMsg, int(tt.query.Amount))
sendRequest(server.peer.app, GetBlockHeadersMsg, reqID, cost, tt.query)
if err := expectResponse(server.peer.app, BlockHeadersMsg, reqID, testBufLimit, headers); err != nil {
cost := server.peer.peer.GetRequestCost(protocol.GetBlockHeadersMsg, int(tt.query.Amount))
sendRequest(server.peer.app, protocol.GetBlockHeadersMsg, reqID, cost, tt.query)
if err := expectResponse(server.peer.app, protocol.BlockHeadersMsg, reqID, testBufLimit, headers); err != nil {
t.Errorf("test %d: headers mismatch: %v", i, err)
}
}
@ -180,14 +181,14 @@ func testGetBlockHeaders(t *testing.T, protocol int) {
func TestGetBlockBodiesLes2(t *testing.T) { testGetBlockBodies(t, 2) }
func TestGetBlockBodiesLes3(t *testing.T) { testGetBlockBodies(t, 3) }
func testGetBlockBodies(t *testing.T, protocol int) {
server, tearDown := newServerEnv(t, downloader.MaxBlockFetch+15, protocol, nil, false, true, 0)
func testGetBlockBodies(t *testing.T, p int) {
server, tearDown := newServerEnv(t, downloader.MaxBlockFetch+15, p, nil, false, true, 0)
defer tearDown()
bc := server.handler.blockchain
// Create a batch of tests for various scenarios
limit := MaxBodyFetch
limit := protocol.MaxBodyFetch
tests := []struct {
random int // Number of blocks to fetch randomly from the chain
explicit []common.Hash // Explicitly requested blocks
@ -246,9 +247,9 @@ func testGetBlockBodies(t *testing.T, protocol int) {
reqID++
// Send the hash request and verify the response
cost := server.peer.peer.GetRequestCost(GetBlockBodiesMsg, len(hashes))
sendRequest(server.peer.app, GetBlockBodiesMsg, reqID, cost, hashes)
if err := expectResponse(server.peer.app, BlockBodiesMsg, reqID, testBufLimit, bodies); err != nil {
cost := server.peer.peer.GetRequestCost(protocol.GetBlockBodiesMsg, len(hashes))
sendRequest(server.peer.app, protocol.GetBlockBodiesMsg, reqID, cost, hashes)
if err := expectResponse(server.peer.app, protocol.BlockBodiesMsg, reqID, testBufLimit, bodies); err != nil {
t.Errorf("test %d: bodies mismatch: %v", i, err)
}
}
@ -258,19 +259,19 @@ func testGetBlockBodies(t *testing.T, protocol int) {
func TestGetCodeLes2(t *testing.T) { testGetCode(t, 2) }
func TestGetCodeLes3(t *testing.T) { testGetCode(t, 3) }
func testGetCode(t *testing.T, protocol int) {
func testGetCode(t *testing.T, p int) {
// Assemble the test environment
server, tearDown := newServerEnv(t, 4, protocol, nil, false, true, 0)
server, tearDown := newServerEnv(t, 4, p, nil, false, true, 0)
defer tearDown()
bc := server.handler.blockchain
var codereqs []*CodeReq
var codereqs []protocol.CodeRequest
var codes [][]byte
for i := uint64(0); i <= bc.CurrentBlock().NumberU64(); i++ {
header := bc.GetHeaderByNumber(i)
req := &CodeReq{
BHash: header.Hash(),
AccKey: crypto.Keccak256(testContractAddr[:]),
req := protocol.CodeRequest{
BlockHash: header.Hash(),
Account: crypto.Keccak256(testContractAddr[:]),
}
codereqs = append(codereqs, req)
if i >= testContractDeployed {
@ -278,9 +279,9 @@ func testGetCode(t *testing.T, protocol int) {
}
}
cost := server.peer.peer.GetRequestCost(GetCodeMsg, len(codereqs))
sendRequest(server.peer.app, GetCodeMsg, 42, cost, codereqs)
if err := expectResponse(server.peer.app, CodeMsg, 42, testBufLimit, codes); err != nil {
cost := server.peer.peer.GetRequestCost(protocol.GetCodeMsg, len(codereqs))
sendRequest(server.peer.app, protocol.GetCodeMsg, 42, cost, codereqs)
if err := expectResponse(server.peer.app, protocol.CodeMsg, 42, testBufLimit, codes); err != nil {
t.Errorf("codes mismatch: %v", err)
}
}
@ -289,19 +290,19 @@ func testGetCode(t *testing.T, protocol int) {
func TestGetStaleCodeLes2(t *testing.T) { testGetStaleCode(t, 2) }
func TestGetStaleCodeLes3(t *testing.T) { testGetStaleCode(t, 3) }
func testGetStaleCode(t *testing.T, protocol int) {
server, tearDown := newServerEnv(t, core.TriesInMemory+4, protocol, nil, false, true, 0)
func testGetStaleCode(t *testing.T, p int) {
server, tearDown := newServerEnv(t, core.TriesInMemory+4, p, nil, false, true, 0)
defer tearDown()
bc := server.handler.blockchain
check := func(number uint64, expected [][]byte) {
req := &CodeReq{
BHash: bc.GetHeaderByNumber(number).Hash(),
AccKey: crypto.Keccak256(testContractAddr[:]),
req := protocol.CodeRequest{
BlockHash: bc.GetHeaderByNumber(number).Hash(),
Account: crypto.Keccak256(testContractAddr[:]),
}
cost := server.peer.peer.GetRequestCost(GetCodeMsg, 1)
sendRequest(server.peer.app, GetCodeMsg, 42, cost, []*CodeReq{req})
if err := expectResponse(server.peer.app, CodeMsg, 42, testBufLimit, expected); err != nil {
cost := server.peer.peer.GetRequestCost(protocol.GetCodeMsg, 1)
sendRequest(server.peer.app, protocol.GetCodeMsg, 42, cost, []protocol.CodeRequest{req})
if err := expectResponse(server.peer.app, protocol.CodeMsg, 42, testBufLimit, expected); err != nil {
t.Errorf("codes mismatch: %v", err)
}
}
@ -314,9 +315,9 @@ func testGetStaleCode(t *testing.T, protocol int) {
func TestGetReceiptLes2(t *testing.T) { testGetReceipt(t, 2) }
func TestGetReceiptLes3(t *testing.T) { testGetReceipt(t, 3) }
func testGetReceipt(t *testing.T, protocol int) {
func testGetReceipt(t *testing.T, p int) {
// Assemble the test environment
server, tearDown := newServerEnv(t, 4, protocol, nil, false, true, 0)
server, tearDown := newServerEnv(t, 4, p, nil, false, true, 0)
defer tearDown()
bc := server.handler.blockchain
@ -331,9 +332,9 @@ func testGetReceipt(t *testing.T, protocol int) {
receipts = append(receipts, rawdb.ReadRawReceipts(server.db, block.Hash(), block.NumberU64()))
}
// Send the hash request and verify the response
cost := server.peer.peer.GetRequestCost(GetReceiptsMsg, len(hashes))
sendRequest(server.peer.app, GetReceiptsMsg, 42, cost, hashes)
if err := expectResponse(server.peer.app, ReceiptsMsg, 42, testBufLimit, receipts); err != nil {
cost := server.peer.peer.GetRequestCost(protocol.GetReceiptsMsg, len(hashes))
sendRequest(server.peer.app, protocol.GetReceiptsMsg, 42, cost, hashes)
if err := expectResponse(server.peer.app, protocol.ReceiptsMsg, 42, testBufLimit, receipts); err != nil {
t.Errorf("receipts mismatch: %v", err)
}
}
@ -342,14 +343,14 @@ func testGetReceipt(t *testing.T, protocol int) {
func TestGetProofsLes2(t *testing.T) { testGetProofs(t, 2) }
func TestGetProofsLes3(t *testing.T) { testGetProofs(t, 3) }
func testGetProofs(t *testing.T, protocol int) {
func testGetProofs(t *testing.T, p int) {
// Assemble the test environment
server, tearDown := newServerEnv(t, 4, protocol, nil, false, true, 0)
server, tearDown := newServerEnv(t, 4, p, nil, false, true, 0)
defer tearDown()
bc := server.handler.blockchain
var proofreqs []ProofReq
var proofreqs []protocol.TrieProofRequest
proofsV2 := light.NewNodeSet()
accounts := []common.Address{bankAddr, userAddr1, userAddr2, signerAddr, {}}
@ -358,8 +359,8 @@ func testGetProofs(t *testing.T, protocol int) {
trie, _ := trie.New(header.Root, trie.NewDatabase(server.db))
for _, acc := range accounts {
req := ProofReq{
BHash: header.Hash(),
req := protocol.TrieProofRequest{
BlockHash: header.Hash(),
Key: crypto.Keccak256(acc[:]),
}
proofreqs = append(proofreqs, req)
@ -367,9 +368,9 @@ func testGetProofs(t *testing.T, protocol int) {
}
}
// Send the proof request and verify the response
cost := server.peer.peer.GetRequestCost(GetProofsV2Msg, len(proofreqs))
sendRequest(server.peer.app, GetProofsV2Msg, 42, cost, proofreqs)
if err := expectResponse(server.peer.app, ProofsV2Msg, 42, testBufLimit, proofsV2.NodeList()); err != nil {
cost := server.peer.peer.GetRequestCost(protocol.GetProofsV2Msg, len(proofreqs))
sendRequest(server.peer.app, protocol.GetProofsV2Msg, 42, cost, proofreqs)
if err := expectResponse(server.peer.app, protocol.ProofsV2Msg, 42, testBufLimit, proofsV2.NodeList()); err != nil {
t.Errorf("proofs mismatch: %v", err)
}
}
@ -378,8 +379,8 @@ func testGetProofs(t *testing.T, protocol int) {
func TestGetStaleProofLes2(t *testing.T) { testGetStaleProof(t, 2) }
func TestGetStaleProofLes3(t *testing.T) { testGetStaleProof(t, 3) }
func testGetStaleProof(t *testing.T, protocol int) {
server, tearDown := newServerEnv(t, core.TriesInMemory+4, protocol, nil, false, true, 0)
func testGetStaleProof(t *testing.T, p int) {
server, tearDown := newServerEnv(t, core.TriesInMemory+4, p, nil, false, true, 0)
defer tearDown()
bc := server.handler.blockchain
@ -388,12 +389,12 @@ func testGetStaleProof(t *testing.T, protocol int) {
header = bc.GetHeaderByNumber(number)
account = crypto.Keccak256(userAddr1.Bytes())
)
req := &ProofReq{
BHash: header.Hash(),
req := &protocol.TrieProofRequest{
BlockHash: header.Hash(),
Key: account,
}
cost := server.peer.peer.GetRequestCost(GetProofsV2Msg, 1)
sendRequest(server.peer.app, GetProofsV2Msg, 42, cost, []*ProofReq{req})
cost := server.peer.peer.GetRequestCost(protocol.GetProofsV2Msg, 1)
sendRequest(server.peer.app, protocol.GetProofsV2Msg, 42, cost, []*protocol.TrieProofRequest{req})
var expected []rlp.RawValue
if wantOK {
@ -402,7 +403,7 @@ func testGetStaleProof(t *testing.T, protocol int) {
t.Prove(account, 0, proofsV2)
expected = proofsV2.NodeList()
}
if err := expectResponse(server.peer.app, ProofsV2Msg, 42, testBufLimit, expected); err != nil {
if err := expectResponse(server.peer.app, protocol.ProofsV2Msg, 42, testBufLimit, expected); err != nil {
t.Errorf("codes mismatch: %v", err)
}
}
@ -415,7 +416,7 @@ func testGetStaleProof(t *testing.T, protocol int) {
func TestGetCHTProofsLes2(t *testing.T) { testGetCHTProofs(t, 2) }
func TestGetCHTProofsLes3(t *testing.T) { testGetCHTProofs(t, 3) }
func testGetCHTProofs(t *testing.T, protocol int) {
func testGetCHTProofs(t *testing.T, p int) {
config := light.TestServerIndexerConfig
waitIndexers := func(cIndexer, bIndexer, btIndexer *core.ChainIndexer) {
@ -427,7 +428,7 @@ func testGetCHTProofs(t *testing.T, protocol int) {
time.Sleep(10 * time.Millisecond)
}
}
server, tearDown := newServerEnv(t, int(config.ChtSize+config.ChtConfirms), protocol, waitIndexers, false, true, 0)
server, tearDown := newServerEnv(t, int(config.ChtSize+config.ChtConfirms), p, waitIndexers, false, true, 0)
defer tearDown()
bc := server.handler.blockchain
@ -439,23 +440,23 @@ func testGetCHTProofs(t *testing.T, protocol int) {
key := make([]byte, 8)
binary.BigEndian.PutUint64(key, config.ChtSize-1)
proofsV2 := HelperTrieResps{
proofsV2 := protocol.HelperTrieResponse{
AuxData: [][]byte{rlp},
}
root := light.GetChtRoot(server.db, 0, bc.GetHeaderByNumber(config.ChtSize-1).Hash())
trie, _ := trie.New(root, trie.NewDatabase(rawdb.NewTable(server.db, light.ChtTablePrefix)))
trie.Prove(key, 0, &proofsV2.Proofs)
// Assemble the requests for the different protocols
requestsV2 := []HelperTrieReq{{
Type: htCanonical,
TrieIdx: 0,
requestsV2 := []protocol.HelperTrieRequest{{
Type: protocol.HelperTrieCHT,
TrieIndex: 0,
Key: key,
AuxReq: auxHeader,
AuxType: protocol.AuxHeader,
}}
// Send the proof request and verify the response
cost := server.peer.peer.GetRequestCost(GetHelperTrieProofsMsg, len(requestsV2))
sendRequest(server.peer.app, GetHelperTrieProofsMsg, 42, cost, requestsV2)
if err := expectResponse(server.peer.app, HelperTrieProofsMsg, 42, testBufLimit, proofsV2); err != nil {
cost := server.peer.peer.GetRequestCost(protocol.GetHelperTrieProofsMsg, len(requestsV2))
sendRequest(server.peer.app, protocol.GetHelperTrieProofsMsg, 42, cost, requestsV2)
if err := expectResponse(server.peer.app, protocol.HelperTrieProofsMsg, 42, testBufLimit, proofsV2); err != nil {
t.Errorf("proofs mismatch: %v", err)
}
}
@ -464,7 +465,7 @@ func TestGetBloombitsProofsLes2(t *testing.T) { testGetBloombitsProofs(t, 2) }
func TestGetBloombitsProofsLes3(t *testing.T) { testGetBloombitsProofs(t, 3) }
// Tests that bloombits proofs can be correctly retrieved.
func testGetBloombitsProofs(t *testing.T, protocol int) {
func testGetBloombitsProofs(t *testing.T, p int) {
config := light.TestServerIndexerConfig
waitIndexers := func(cIndexer, bIndexer, btIndexer *core.ChainIndexer) {
@ -476,7 +477,7 @@ func testGetBloombitsProofs(t *testing.T, protocol int) {
time.Sleep(10 * time.Millisecond)
}
}
server, tearDown := newServerEnv(t, int(config.BloomTrieSize+config.BloomTrieConfirms), protocol, waitIndexers, false, true, 0)
server, tearDown := newServerEnv(t, int(config.BloomTrieSize+config.BloomTrieConfirms), p, waitIndexers, false, true, 0)
defer tearDown()
bc := server.handler.blockchain
@ -490,21 +491,21 @@ func testGetBloombitsProofs(t *testing.T, protocol int) {
// Only the first bloom section has data.
binary.BigEndian.PutUint64(key[2:], 0)
requests := []HelperTrieReq{{
Type: htBloomBits,
TrieIdx: 0,
requests := []protocol.HelperTrieRequest{{
Type: protocol.HelperTrieBloomTrie,
TrieIndex: 0,
Key: key,
}}
var proofs HelperTrieResps
var proofs protocol.HelperTrieResponse
root := light.GetBloomTrieRoot(server.db, 0, bc.GetHeaderByNumber(config.BloomTrieSize-1).Hash())
trie, _ := trie.New(root, trie.NewDatabase(rawdb.NewTable(server.db, light.BloomTrieTablePrefix)))
trie.Prove(key, 0, &proofs.Proofs)
// Send the proof request and verify the response
cost := server.peer.peer.GetRequestCost(GetHelperTrieProofsMsg, len(requests))
sendRequest(server.peer.app, GetHelperTrieProofsMsg, 42, cost, requests)
if err := expectResponse(server.peer.app, HelperTrieProofsMsg, 42, testBufLimit, proofs); err != nil {
cost := server.peer.peer.GetRequestCost(protocol.GetHelperTrieProofsMsg, len(requests))
sendRequest(server.peer.app, protocol.GetHelperTrieProofsMsg, 42, cost, requests)
if err := expectResponse(server.peer.app, protocol.HelperTrieProofsMsg, 42, testBufLimit, proofs); err != nil {
t.Errorf("bit %d: proofs mismatch: %v", bit, err)
}
}
@ -513,8 +514,8 @@ func testGetBloombitsProofs(t *testing.T, protocol int) {
func TestTransactionStatusLes2(t *testing.T) { testTransactionStatus(t, 2) }
func TestTransactionStatusLes3(t *testing.T) { testTransactionStatus(t, 3) }
func testTransactionStatus(t *testing.T, protocol int) {
server, tearDown := newServerEnv(t, 0, protocol, nil, false, true, 0)
func testTransactionStatus(t *testing.T, p int) {
server, tearDown := newServerEnv(t, 0, p, nil, false, true, 0)
defer tearDown()
server.handler.addTxsSync = true
@ -525,13 +526,13 @@ func testTransactionStatus(t *testing.T, protocol int) {
test := func(tx *types.Transaction, send bool, expStatus light.TxStatus) {
reqID++
if send {
cost := server.peer.peer.GetRequestCost(SendTxV2Msg, 1)
sendRequest(server.peer.app, SendTxV2Msg, reqID, cost, types.Transactions{tx})
cost := server.peer.peer.GetRequestCost(protocol.SendTxV2Msg, 1)
sendRequest(server.peer.app, protocol.SendTxV2Msg, reqID, cost, types.Transactions{tx})
} else {
cost := server.peer.peer.GetRequestCost(GetTxStatusMsg, 1)
sendRequest(server.peer.app, GetTxStatusMsg, reqID, cost, []common.Hash{tx.Hash()})
cost := server.peer.peer.GetRequestCost(protocol.GetTxStatusMsg, 1)
sendRequest(server.peer.app, protocol.GetTxStatusMsg, reqID, cost, []common.Hash{tx.Hash()})
}
if err := expectResponse(server.peer.app, TxStatusMsg, reqID, testBufLimit, []light.TxStatus{expStatus}); err != nil {
if err := expectResponse(server.peer.app, protocol.TxStatusMsg, reqID, testBufLimit, []light.TxStatus{expStatus}); err != nil {
t.Errorf("transaction status mismatch")
}
}
@ -620,14 +621,14 @@ func TestStopResumeLes3(t *testing.T) {
header := server.handler.blockchain.CurrentHeader()
req := func() {
reqID++
sendRequest(server.peer.app, GetBlockHeadersMsg, reqID, testCost, &getBlockHeadersData{Origin: hashOrNumber{Hash: header.Hash()}, Amount: 1})
sendRequest(server.peer.app, protocol.GetBlockHeadersMsg, reqID, testCost, &protocol.GetBlockHeadersRequest{Origin: protocol.HashOrNumber{Hash: header.Hash()}, Amount: 1})
}
for i := 1; i <= 5; i++ {
// send requests while we still have enough buffer and expect a response
for expBuf >= testCost {
req()
expBuf -= testCost
if err := expectResponse(server.peer.app, BlockHeadersMsg, reqID, expBuf, []*types.Header{header}); err != nil {
if err := expectResponse(server.peer.app, protocol.BlockHeadersMsg, reqID, expBuf, []*types.Header{header}); err != nil {
t.Errorf("expected response and failed: %v", err)
}
}
@ -637,7 +638,7 @@ func TestStopResumeLes3(t *testing.T) {
req()
c--
}
if err := p2p.ExpectMsg(server.peer.app, StopMsg, nil); err != nil {
if err := p2p.ExpectMsg(server.peer.app, protocol.StopMsg, nil); err != nil {
t.Errorf("expected StopMsg and failed: %v", err)
}
// wait until the buffer is recharged by half of the limit
@ -646,7 +647,7 @@ func TestStopResumeLes3(t *testing.T) {
// expect a ResumeMsg with the partially recharged buffer value
expBuf += testBufRecharge * wait
if err := p2p.ExpectMsg(server.peer.app, ResumeMsg, expBuf); err != nil {
if err := p2p.ExpectMsg(server.peer.app, protocol.ResumeMsg, expBuf); err != nil {
t.Errorf("expected ResumeMsg and failed: %v", err)
}
}

95
les/odr_requests.go
View file

@ -26,6 +26,7 @@ import (
"github.com/ethereum/go-ethereum/core/types"
"github.com/ethereum/go-ethereum/crypto"
"github.com/ethereum/go-ethereum/ethdb"
"github.com/ethereum/go-ethereum/les/protocol"
"github.com/ethereum/go-ethereum/light"
"github.com/ethereum/go-ethereum/log"
"github.com/ethereum/go-ethereum/rlp"
@ -79,7 +80,7 @@ type BlockRequest light.BlockRequest
// GetCost returns the cost of the given ODR request according to the serving
// peer's cost table (implementation of LesOdrRequest)
func (r *BlockRequest) GetCost(peer *peer) uint64 {
return peer.GetRequestCost(GetBlockBodiesMsg, 1)
return peer.GetRequestCost(protocol.GetBlockBodiesMsg, 1)
}
// CanSend tells if a certain peer is suitable for serving the given request
@ -135,7 +136,7 @@ type ReceiptsRequest light.ReceiptsRequest
// GetCost returns the cost of the given ODR request according to the serving
// peer's cost table (implementation of LesOdrRequest)
func (r *ReceiptsRequest) GetCost(peer *peer) uint64 {
return peer.GetRequestCost(GetReceiptsMsg, 1)
return peer.GetRequestCost(protocol.GetReceiptsMsg, 1)
}
// CanSend tells if a certain peer is suitable for serving the given request
@ -180,19 +181,13 @@ func (r *ReceiptsRequest) Validate(db ethdb.Database, msg *Msg) error {
return nil
}
type ProofReq struct {
BHash common.Hash
AccKey, Key []byte
FromLevel uint
}
// ODR request type for state/storage trie entries, see LesOdrRequest interface
type TrieRequest light.TrieRequest
// GetCost returns the cost of the given ODR request according to the serving
// peer's cost table (implementation of LesOdrRequest)
func (r *TrieRequest) GetCost(peer *peer) uint64 {
return peer.GetRequestCost(GetProofsV2Msg, 1)
return peer.GetRequestCost(protocol.GetProofsV2Msg, 1)
}
// CanSend tells if a certain peer is suitable for serving the given request
@ -203,12 +198,12 @@ func (r *TrieRequest) CanSend(peer *peer) bool {
// Request sends an ODR request to the LES network (implementation of LesOdrRequest)
func (r *TrieRequest) Request(reqID uint64, peer *peer) error {
peer.Log().Debug("Requesting trie proof", "root", r.Id.Root, "key", r.Key)
req := ProofReq{
BHash: r.Id.BlockHash,
AccKey: r.Id.AccKey,
req := protocol.TrieProofRequest{
BlockHash: r.Id.BlockHash,
Account: r.Id.AccKey,
Key: r.Key,
}
return peer.RequestProofs(reqID, r.GetCost(peer), []ProofReq{req})
return peer.RequestProofs(reqID, r.GetCost(peer), []protocol.TrieProofRequest{req})
}
// Valid processes an ODR request reply message from the LES network
@ -235,18 +230,13 @@ func (r *TrieRequest) Validate(db ethdb.Database, msg *Msg) error {
return nil
}
type CodeReq struct {
BHash common.Hash
AccKey []byte
}
// ODR request type for node data (used for retrieving contract code), see LesOdrRequest interface
type CodeRequest light.CodeRequest
// GetCost returns the cost of the given ODR request according to the serving
// peer's cost table (implementation of LesOdrRequest)
func (r *CodeRequest) GetCost(peer *peer) uint64 {
return peer.GetRequestCost(GetCodeMsg, 1)
return peer.GetRequestCost(protocol.GetCodeMsg, 1)
}
// CanSend tells if a certain peer is suitable for serving the given request
@ -257,11 +247,11 @@ func (r *CodeRequest) CanSend(peer *peer) bool {
// Request sends an ODR request to the LES network (implementation of LesOdrRequest)
func (r *CodeRequest) Request(reqID uint64, peer *peer) error {
peer.Log().Debug("Requesting code data", "hash", r.Hash)
req := CodeReq{
BHash: r.Id.BlockHash,
AccKey: r.Id.AccKey,
req := protocol.CodeRequest{
BlockHash: r.Id.BlockHash,
Account: r.Id.AccKey,
}
return peer.RequestCode(reqID, r.GetCost(peer), []CodeReq{req})
return peer.RequestCode(reqID, r.GetCost(peer), []protocol.CodeRequest{req})
}
// Valid processes an ODR request reply message from the LES network
@ -288,36 +278,13 @@ func (r *CodeRequest) Validate(db ethdb.Database, msg *Msg) error {
return nil
}
const (
// helper trie type constants
htCanonical = iota // Canonical hash trie
htBloomBits // BloomBits trie
// applicable for all helper trie requests
auxRoot = 1
// applicable for htCanonical
auxHeader = 2
)
type HelperTrieReq struct {
Type uint
TrieIdx uint64
Key []byte
FromLevel, AuxReq uint
}
type HelperTrieResps struct { // describes all responses, not just a single one
Proofs light.NodeList
AuxData [][]byte
}
// ODR request type for requesting headers by Canonical Hash Trie, see LesOdrRequest interface
type ChtRequest light.ChtRequest
// GetCost returns the cost of the given ODR request according to the serving
// peer's cost table (implementation of LesOdrRequest)
func (r *ChtRequest) GetCost(peer *peer) uint64 {
return peer.GetRequestCost(GetHelperTrieProofsMsg, 1)
return peer.GetRequestCost(protocol.GetHelperTrieProofsMsg, 1)
}
// CanSend tells if a certain peer is suitable for serving the given request
@ -337,13 +304,13 @@ func (r *ChtRequest) Request(reqID uint64, peer *peer) error {
peer.Log().Debug("Requesting CHT", "cht", r.ChtNum, "block", r.BlockNum)
var encNum [8]byte
binary.BigEndian.PutUint64(encNum[:], r.BlockNum)
req := HelperTrieReq{
Type: htCanonical,
TrieIdx: r.ChtNum,
req := protocol.HelperTrieRequest{
Type: protocol.HelperTrieCHT,
TrieIndex: r.ChtNum,
Key: encNum[:],
AuxReq: auxHeader,
AuxType: protocol.AuxHeader,
}
return peer.RequestHelperTrieProofs(reqID, r.GetCost(peer), []HelperTrieReq{req})
return peer.RequestHelperTrieProofs(reqID, r.GetCost(peer), []protocol.HelperTrieRequest{req})
}
// Valid processes an ODR request reply message from the LES network
@ -355,7 +322,7 @@ func (r *ChtRequest) Validate(db ethdb.Database, msg *Msg) error {
if msg.MsgType != MsgHelperTrieProofs {
return errInvalidMessageType
}
resp := msg.Obj.(HelperTrieResps)
resp := msg.Obj.(protocol.HelperTrieResponse)
if len(resp.AuxData) != 1 {
return errInvalidEntryCount
}
@ -404,17 +371,13 @@ func (r *ChtRequest) Validate(db ethdb.Database, msg *Msg) error {
return nil
}
type BloomReq struct {
BloomTrieNum, BitIdx, SectionIndex, FromLevel uint64
}
// ODR request type for requesting headers by Canonical Hash Trie, see LesOdrRequest interface
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 *peer) uint64 {
return peer.GetRequestCost(GetHelperTrieProofsMsg, len(r.SectionIndexList))
return peer.GetRequestCost(protocol.GetHelperTrieProofsMsg, len(r.SectionIndexList))
}
// CanSend tells if a certain peer is suitable for serving the given request
@ -422,7 +385,7 @@ func (r *BloomRequest) CanSend(peer *peer) bool {
peer.lock.RLock()
defer peer.lock.RUnlock()
if peer.version < lpv2 {
if peer.version < protocol.Lpv2 {
return false
}
return peer.headInfo.Number >= r.Config.BloomTrieConfirms && r.BloomTrieNum <= (peer.headInfo.Number-r.Config.BloomTrieConfirms)/r.Config.BloomTrieSize
@ -431,16 +394,16 @@ func (r *BloomRequest) CanSend(peer *peer) bool {
// Request sends an ODR request to the LES network (implementation of LesOdrRequest)
func (r *BloomRequest) Request(reqID uint64, peer *peer) error {
peer.Log().Debug("Requesting BloomBits", "bloomTrie", r.BloomTrieNum, "bitIdx", r.BitIdx, "sections", r.SectionIndexList)
reqs := make([]HelperTrieReq, len(r.SectionIndexList))
reqs := make([]protocol.HelperTrieRequest, len(r.SectionIndexList))
var encNumber [10]byte
binary.BigEndian.PutUint16(encNumber[:2], uint16(r.BitIdx))
for i, sectionIdx := range r.SectionIndexList {
binary.BigEndian.PutUint64(encNumber[2:], sectionIdx)
reqs[i] = HelperTrieReq{
Type: htBloomBits,
TrieIdx: r.BloomTrieNum,
reqs[i] = protocol.HelperTrieRequest{
Type: protocol.HelperTrieBloomTrie,
TrieIndex: r.BloomTrieNum,
Key: common.CopyBytes(encNumber[:]),
}
}
@ -457,7 +420,7 @@ func (r *BloomRequest) Validate(db ethdb.Database, msg *Msg) error {
if msg.MsgType != MsgHelperTrieProofs {
return errInvalidMessageType
}
resps := msg.Obj.(HelperTrieResps)
resps := msg.Obj.(protocol.HelperTrieResponse)
proofs := resps.Proofs
nodeSet := proofs.NodeSet()
reads := &readTraceDB{db: nodeSet}
@ -490,12 +453,12 @@ type TxStatusRequest light.TxStatusRequest
// GetCost returns the cost of the given ODR request according to the serving
// peer's cost table (implementation of LesOdrRequest)
func (r *TxStatusRequest) GetCost(peer *peer) uint64 {
return peer.GetRequestCost(GetTxStatusMsg, len(r.Hashes))
return peer.GetRequestCost(protocol.GetTxStatusMsg, len(r.Hashes))
}
// CanSend tells if a certain peer is suitable for serving the given request
func (r *TxStatusRequest) CanSend(peer *peer) bool {
return peer.version >= lpv2
return peer.version >= protocol.Lpv2
}
// Request sends an ODR request to the LES network (implementation of LesOdrRequest)

244
les/peer.go
View file

@ -32,6 +32,7 @@ import (
"github.com/ethereum/go-ethereum/core/types"
"github.com/ethereum/go-ethereum/eth"
"github.com/ethereum/go-ethereum/les/flowcontrol"
"github.com/ethereum/go-ethereum/les/protocol"
"github.com/ethereum/go-ethereum/les/utilities"
"github.com/ethereum/go-ethereum/light"
"github.com/ethereum/go-ethereum/p2p"
@ -89,7 +90,7 @@ type peer struct {
id string
headInfo *announceData
headInfo *protocol.Announcement
lock sync.RWMutex
sendQueue *utilities.ExecQueue
@ -112,7 +113,7 @@ type peer struct {
fcClient *flowcontrol.ClientNode // nil if the peer is server only
fcServer *flowcontrol.ServerNode // nil if the peer is client only
fcParams flowcontrol.ServerParams
fcCosts requestCostTable
fcCosts protocol.RequestCostTable
trusted, server bool
onlyAnnounce bool
@ -179,7 +180,7 @@ func (p *peer) rejectUpdate(size uint64) bool {
// region. The client is also notified about being frozen/unfrozen with a Stop/Resume
// message.
func (p *peer) freezeClient() {
if p.version < lpv3 {
if p.version < protocol.Lpv3 {
// if Stop/Resume is not supported then just drop the peer after setting
// its frozen status permanently
atomic.StoreUint32(&p.frozen, 1)
@ -258,11 +259,11 @@ func (p *peer) HeadAndTd() (hash common.Hash, td *big.Int) {
return hash, p.headInfo.Td
}
func (p *peer) headBlockInfo() blockInfo {
func (p *peer) headBlockInfo() protocol.HeadHeader {
p.lock.RLock()
defer p.lock.RUnlock()
return blockInfo{Hash: p.headInfo.Hash, Number: p.headInfo.Number, Td: p.headInfo.Td}
return protocol.HeadHeader{Hash: p.headInfo.Hash, Number: p.headInfo.Number, Td: p.headInfo.Td}
}
// Td retrieves the current total difficulty of a peer.
@ -286,10 +287,10 @@ func (p *peer) updateCapacity(cap uint64) {
p.fcParams = flowcontrol.ServerParams{MinRecharge: cap, BufLimit: cap * bufLimitRatio}
p.fcClient.UpdateParams(p.fcParams)
var kvList keyValueList
kvList = kvList.add("flowControl/MRR", cap)
kvList = kvList.add("flowControl/BL", cap*bufLimitRatio)
p.queueSend(func() { p.SendAnnounce(announceData{Update: kvList}) })
var kvList protocol.KeyValueList
kvList = kvList.Add("flowControl/MRR", cap)
kvList = kvList.Add("flowControl/BL", cap*bufLimitRatio)
p.queueSend(func() { p.SendAnnounce(protocol.Announcement{Update: kvList}) })
}
func (p *peer) responseID() uint64 {
@ -336,7 +337,7 @@ func (p *peer) GetRequestCost(msgcode uint64, amount int) uint64 {
if costs == nil {
return 0
}
cost := costs.baseCost + costs.reqCost*uint64(amount)
cost := costs.BaseCost + costs.ReqCost*uint64(amount)
if cost > p.fcParams.BufLimit {
cost = p.fcParams.BufLimit
}
@ -347,12 +348,12 @@ func (p *peer) GetTxRelayCost(amount, size int) uint64 {
p.lock.RLock()
defer p.lock.RUnlock()
costs := p.fcCosts[SendTxV2Msg]
costs := p.fcCosts[protocol.SendTxV2Msg]
if costs == nil {
return 0
}
cost := costs.baseCost + costs.reqCost*uint64(amount)
sizeCost := costs.baseCost + costs.reqCost*uint64(size)/txSizeCostLimit
cost := costs.BaseCost + costs.ReqCost*uint64(amount)
sizeCost := costs.BaseCost + costs.ReqCost*uint64(size)/txSizeCostLimit
if sizeCost > cost {
cost = sizeCost
}
@ -385,185 +386,144 @@ func (p *peer) HasBlock(hash common.Hash, number uint64, hasState bool) bool {
// SendAnnounce announces the availability of a number of blocks through
// a hash notification.
func (p *peer) SendAnnounce(request announceData) error {
return p2p.Send(p.rw, AnnounceMsg, request)
func (p *peer) SendAnnounce(request protocol.Announcement) error {
return p2p.Send(p.rw, protocol.AnnounceMsg, request)
}
// SendStop notifies the client about being in frozen state
func (p *peer) SendStop() error {
return p2p.Send(p.rw, StopMsg, struct{}{})
return p2p.Send(p.rw, protocol.StopMsg, struct{}{})
}
// SendResume notifies the client about getting out of frozen state
func (p *peer) SendResume(bv uint64) error {
return p2p.Send(p.rw, ResumeMsg, bv)
return p2p.Send(p.rw, protocol.ResumeMsg, bv)
}
// ReplyBlockHeaders creates a reply with a batch of block headers
func (p *peer) ReplyBlockHeaders(reqID uint64, headers []*types.Header) *reply {
data, _ := rlp.EncodeToBytes(headers)
return &reply{p.rw, BlockHeadersMsg, reqID, data}
return &reply{p.rw, protocol.BlockHeadersMsg, reqID, data}
}
// ReplyBlockBodiesRLP creates a reply with a batch of block contents from
// an already RLP encoded format.
func (p *peer) ReplyBlockBodiesRLP(reqID uint64, bodies []rlp.RawValue) *reply {
data, _ := rlp.EncodeToBytes(bodies)
return &reply{p.rw, BlockBodiesMsg, reqID, data}
return &reply{p.rw, protocol.BlockBodiesMsg, reqID, data}
}
// ReplyCode creates a reply with a batch of arbitrary internal data, corresponding to the
// hashes requested.
func (p *peer) ReplyCode(reqID uint64, codes [][]byte) *reply {
data, _ := rlp.EncodeToBytes(codes)
return &reply{p.rw, CodeMsg, reqID, data}
return &reply{p.rw, protocol.CodeMsg, reqID, data}
}
// ReplyReceiptsRLP creates a reply with a batch of transaction receipts, corresponding to the
// ones requested from an already RLP encoded format.
func (p *peer) ReplyReceiptsRLP(reqID uint64, receipts []rlp.RawValue) *reply {
data, _ := rlp.EncodeToBytes(receipts)
return &reply{p.rw, ReceiptsMsg, reqID, data}
return &reply{p.rw, protocol.ReceiptsMsg, reqID, data}
}
// ReplyProofsV2 creates a reply with a batch of merkle proofs, corresponding to the ones requested.
func (p *peer) ReplyProofsV2(reqID uint64, proofs light.NodeList) *reply {
data, _ := rlp.EncodeToBytes(proofs)
return &reply{p.rw, ProofsV2Msg, reqID, data}
return &reply{p.rw, protocol.ProofsV2Msg, reqID, data}
}
// ReplyHelperTrieProofs creates a reply with a batch of HelperTrie proofs, corresponding to the ones requested.
func (p *peer) ReplyHelperTrieProofs(reqID uint64, resp HelperTrieResps) *reply {
func (p *peer) ReplyHelperTrieProofs(reqID uint64, resp protocol.HelperTrieResponse) *reply {
data, _ := rlp.EncodeToBytes(resp)
return &reply{p.rw, HelperTrieProofsMsg, reqID, data}
return &reply{p.rw, protocol.HelperTrieProofsMsg, reqID, data}
}
// ReplyTxStatus creates a reply with a batch of transaction status records, corresponding to the ones requested.
func (p *peer) ReplyTxStatus(reqID uint64, stats []light.TxStatus) *reply {
data, _ := rlp.EncodeToBytes(stats)
return &reply{p.rw, TxStatusMsg, reqID, data}
return &reply{p.rw, protocol.TxStatusMsg, reqID, data}
}
// RequestHeadersByHash fetches a batch of blocks' headers corresponding to the
// specified header query, based on the hash of an origin block.
func (p *peer) RequestHeadersByHash(reqID, cost uint64, origin common.Hash, amount int, skip int, reverse bool) error {
p.Log().Debug("Fetching batch of headers", "count", amount, "fromhash", origin, "skip", skip, "reverse", reverse)
return sendRequest(p.rw, GetBlockHeadersMsg, reqID, cost, &getBlockHeadersData{Origin: hashOrNumber{Hash: origin}, Amount: uint64(amount), Skip: uint64(skip), Reverse: reverse})
return sendRequest(p.rw, protocol.GetBlockHeadersMsg, reqID, cost, &protocol.GetBlockHeadersRequest{Origin: protocol.HashOrNumber{Hash: origin}, Amount: uint64(amount), Skip: uint64(skip), Reverse: reverse})
}
// RequestHeadersByNumber fetches a batch of blocks' headers corresponding to the
// specified header query, based on the number of an origin block.
func (p *peer) RequestHeadersByNumber(reqID, cost, origin uint64, amount int, skip int, reverse bool) error {
p.Log().Debug("Fetching batch of headers", "count", amount, "fromnum", origin, "skip", skip, "reverse", reverse)
return sendRequest(p.rw, GetBlockHeadersMsg, reqID, cost, &getBlockHeadersData{Origin: hashOrNumber{Number: origin}, Amount: uint64(amount), Skip: uint64(skip), Reverse: reverse})
return sendRequest(p.rw, protocol.GetBlockHeadersMsg, reqID, cost, &protocol.GetBlockHeadersRequest{Origin: protocol.HashOrNumber{Number: origin}, Amount: uint64(amount), Skip: uint64(skip), Reverse: reverse})
}
// RequestBodies fetches a batch of blocks' bodies corresponding to the hashes
// specified.
func (p *peer) RequestBodies(reqID, cost uint64, hashes []common.Hash) error {
p.Log().Debug("Fetching batch of block bodies", "count", len(hashes))
return sendRequest(p.rw, GetBlockBodiesMsg, reqID, cost, hashes)
return sendRequest(p.rw, protocol.GetBlockBodiesMsg, reqID, cost, hashes)
}
// RequestCode fetches a batch of arbitrary data from a node's known state
// data, corresponding to the specified hashes.
func (p *peer) RequestCode(reqID, cost uint64, reqs []CodeReq) error {
func (p *peer) RequestCode(reqID, cost uint64, reqs []protocol.CodeRequest) error {
p.Log().Debug("Fetching batch of codes", "count", len(reqs))
return sendRequest(p.rw, GetCodeMsg, reqID, cost, reqs)
return sendRequest(p.rw, protocol.GetCodeMsg, reqID, cost, reqs)
}
// RequestReceipts fetches a batch of transaction receipts from a remote node.
func (p *peer) RequestReceipts(reqID, cost uint64, hashes []common.Hash) error {
p.Log().Debug("Fetching batch of receipts", "count", len(hashes))
return sendRequest(p.rw, GetReceiptsMsg, reqID, cost, hashes)
return sendRequest(p.rw, protocol.GetReceiptsMsg, reqID, cost, hashes)
}
// RequestProofs fetches a batch of merkle proofs from a remote node.
func (p *peer) RequestProofs(reqID, cost uint64, reqs []ProofReq) error {
func (p *peer) RequestProofs(reqID, cost uint64, reqs []protocol.TrieProofRequest) error {
p.Log().Debug("Fetching batch of proofs", "count", len(reqs))
return sendRequest(p.rw, GetProofsV2Msg, reqID, cost, reqs)
return sendRequest(p.rw, protocol.GetProofsV2Msg, reqID, cost, reqs)
}
// RequestHelperTrieProofs fetches a batch of HelperTrie merkle proofs from a remote node.
func (p *peer) RequestHelperTrieProofs(reqID, cost uint64, reqs []HelperTrieReq) error {
func (p *peer) RequestHelperTrieProofs(reqID, cost uint64, reqs []protocol.HelperTrieRequest) error {
p.Log().Debug("Fetching batch of HelperTrie proofs", "count", len(reqs))
return sendRequest(p.rw, GetHelperTrieProofsMsg, reqID, cost, reqs)
return sendRequest(p.rw, protocol.GetHelperTrieProofsMsg, reqID, cost, reqs)
}
// RequestTxStatus fetches a batch of transaction status records from a remote node.
func (p *peer) RequestTxStatus(reqID, cost uint64, txHashes []common.Hash) error {
p.Log().Debug("Requesting transaction status", "count", len(txHashes))
return sendRequest(p.rw, GetTxStatusMsg, reqID, cost, txHashes)
return sendRequest(p.rw, protocol.GetTxStatusMsg, reqID, cost, txHashes)
}
// SendTxStatus creates a reply with a batch of transactions to be added to the remote transaction pool.
func (p *peer) SendTxs(reqID, cost uint64, txs rlp.RawValue) error {
p.Log().Debug("Sending batch of transactions", "size", len(txs))
return sendRequest(p.rw, SendTxV2Msg, reqID, cost, txs)
return sendRequest(p.rw, protocol.SendTxV2Msg, reqID, cost, txs)
}
type keyValueEntry struct {
Key string
Value rlp.RawValue
}
type keyValueList []keyValueEntry
type keyValueMap map[string]rlp.RawValue
func (l keyValueList) add(key string, val interface{}) keyValueList {
var entry keyValueEntry
entry.Key = key
if val == nil {
val = uint64(0)
}
enc, err := rlp.EncodeToBytes(val)
if err == nil {
entry.Value = enc
}
return append(l, entry)
}
func (l keyValueList) decode() (keyValueMap, uint64) {
m := make(keyValueMap)
var size uint64
for _, entry := range l {
m[entry.Key] = entry.Value
size += uint64(len(entry.Key)) + uint64(len(entry.Value)) + 8
}
return m, size
}
func (m keyValueMap) get(key string, val interface{}) error {
enc, ok := m[key]
if !ok {
return errResp(ErrMissingKey, "%s", key)
}
if val == nil {
return nil
}
return rlp.DecodeBytes(enc, val)
}
func (p *peer) sendReceiveHandshake(sendList keyValueList) (keyValueList, error) {
func (p *peer) sendReceiveHandshake(sendList protocol.KeyValueList) (protocol.KeyValueList, error) {
// Send out own handshake in a new thread
errc := make(chan error, 1)
go func() {
errc <- p2p.Send(p.rw, StatusMsg, sendList)
errc <- p2p.Send(p.rw, protocol.StatusMsg, sendList)
}()
// In the mean time retrieve the remote status message
msg, err := p.rw.ReadMsg()
if err != nil {
return nil, err
}
if msg.Code != StatusMsg {
return nil, errResp(ErrNoStatusMsg, "first msg has code %x (!= %x)", msg.Code, StatusMsg)
if msg.Code != protocol.StatusMsg {
return nil, protocol.ErrResp(protocol.ErrNoStatusMsg, "first msg has code %x (!= %x)", msg.Code, protocol.StatusMsg)
}
if msg.Size > ProtocolMaxMsgSize {
return nil, errResp(ErrMsgTooLarge, "%v > %v", msg.Size, ProtocolMaxMsgSize)
if msg.Size > protocol.ProtocolMaxMsgSize {
return nil, protocol.ErrResp(protocol.ErrMsgTooLarge, "%v > %v", msg.Size, protocol.ProtocolMaxMsgSize)
}
// Decode the handshake
var recvList keyValueList
var recvList protocol.KeyValueList
if err := msg.Decode(&recvList); err != nil {
return nil, errResp(ErrDecode, "msg %v: %v", msg, err)
return nil, protocol.ErrResp(protocol.ErrDecode, "msg %v: %v", msg, err)
}
if err := <-errc; err != nil {
return nil, err
@ -577,21 +537,21 @@ func (p *peer) Handshake(td *big.Int, head common.Hash, headNum uint64, genesis
p.lock.Lock()
defer p.lock.Unlock()
var send keyValueList
var send protocol.KeyValueList
// Add some basic handshake fields
send = send.add("protocolVersion", uint64(p.version))
send = send.add("networkId", p.network)
send = send.add("headTd", td)
send = send.add("headHash", head)
send = send.add("headNum", headNum)
send = send.add("genesisHash", genesis)
send = send.Add("protocolVersion", uint64(p.version))
send = send.Add("networkId", p.network)
send = send.Add("headTd", td)
send = send.Add("headHash", head)
send = send.Add("headNum", headNum)
send = send.Add("genesisHash", genesis)
if server != nil {
// Add some information which services server can offer.
if !server.config.UltraLightOnlyAnnounce {
send = send.add("serveHeaders", nil)
send = send.add("serveChainSince", uint64(0))
send = send.add("serveStateSince", uint64(0))
send = send.Add("serveHeaders", nil)
send = send.Add("serveChainSince", uint64(0))
send = send.Add("serveStateSince", uint64(0))
// If local ethereum node is running in archive mode, advertise ourselves we have
// all version state data. Otherwise only recent state is available.
@ -599,20 +559,20 @@ func (p *peer) Handshake(td *big.Int, head common.Hash, headNum uint64, genesis
if server.archiveMode {
stateRecent = 0
}
send = send.add("serveRecentState", stateRecent)
send = send.add("txRelay", nil)
send = send.Add("serveRecentState", stateRecent)
send = send.Add("txRelay", nil)
}
send = send.add("flowControl/BL", server.defParams.BufLimit)
send = send.add("flowControl/MRR", server.defParams.MinRecharge)
send = send.Add("flowControl/BL", server.defParams.BufLimit)
send = send.Add("flowControl/MRR", server.defParams.MinRecharge)
var costList RequestCostList
var costList protocol.RequestCostList
if server.costTracker.testCostList != nil {
costList = server.costTracker.testCostList
} else {
costList = server.costTracker.makeCostList(server.costTracker.globalFactor())
}
send = send.add("flowControl/MRC", costList)
p.fcCosts = costList.decode(ProtocolLengths[uint(p.version)])
send = send.Add("flowControl/MRC", costList)
p.fcCosts = costList.ToTable(protocol.ProtocolLengths[uint(p.version)])
p.fcParams = server.defParams
// Add advertised checkpoint and register block height which
@ -620,8 +580,8 @@ func (p *peer) Handshake(td *big.Int, head common.Hash, headNum uint64, genesis
if server.oracle != nil && server.oracle.IsRunning() {
cp, height := server.oracle.StableCheckpoint()
if cp != nil {
send = send.add("checkpoint/value", cp)
send = send.add("checkpoint/registerHeight", height)
send = send.Add("checkpoint/value", cp)
send = send.Add("checkpoint/registerHeight", height)
}
}
} else {
@ -630,130 +590,136 @@ func (p *peer) Handshake(td *big.Int, head common.Hash, headNum uint64, genesis
if p.trusted {
p.announceType = announceTypeSigned
}
send = send.add("announceType", p.announceType)
send = send.Add("announceType", p.announceType)
}
recvList, err := p.sendReceiveHandshake(send)
if err != nil {
return err
}
recv, size := recvList.decode()
recv, size := recvList.ToMap()
if p.rejectUpdate(size) {
return errResp(ErrRequestRejected, "")
return protocol.ErrResp(protocol.ErrRequestRejected, "")
}
var rGenesis, rHash common.Hash
var rVersion, rNetwork, rNum uint64
var rTd *big.Int
if err := recv.get("protocolVersion", &rVersion); err != nil {
if err := recv.Get("protocolVersion", &rVersion); err != nil {
return err
}
if err := recv.get("networkId", &rNetwork); err != nil {
if err := recv.Get("networkId", &rNetwork); err != nil {
return err
}
if err := recv.get("headTd", &rTd); err != nil {
if err := recv.Get("headTd", &rTd); err != nil {
return err
}
if err := recv.get("headHash", &rHash); err != nil {
if err := recv.Get("headHash", &rHash); err != nil {
return err
}
if err := recv.get("headNum", &rNum); err != nil {
if err := recv.Get("headNum", &rNum); err != nil {
return err
}
if err := recv.get("genesisHash", &rGenesis); err != nil {
if err := recv.Get("genesisHash", &rGenesis); err != nil {
return err
}
if rGenesis != genesis {
return errResp(ErrGenesisBlockMismatch, "%x (!= %x)", rGenesis[:8], genesis[:8])
return protocol.ErrResp(protocol.ErrGenesisBlockMismatch, "%x (!= %x)", rGenesis[:8], genesis[:8])
}
if rNetwork != p.network {
return errResp(ErrNetworkIdMismatch, "%d (!= %d)", rNetwork, p.network)
return protocol.ErrResp(protocol.ErrNetworkIdMismatch, "%d (!= %d)", rNetwork, p.network)
}
if int(rVersion) != p.version {
return errResp(ErrProtocolVersionMismatch, "%d (!= %d)", rVersion, p.version)
return protocol.ErrResp(protocol.ErrProtocolVersionMismatch, "%d (!= %d)", rVersion, p.version)
}
if server != nil {
p.server = recv.get("flowControl/MRR", nil) == nil
p.server = recv.Get("flowControl/MRR", nil) == nil
if p.server {
p.announceType = announceTypeNone // connected to another server, send no messages
} else {
if recv.get("announceType", &p.announceType) != nil {
if recv.Get("announceType", &p.announceType) != nil {
// set default announceType on server side
p.announceType = announceTypeSimple
}
p.fcClient = flowcontrol.NewClientNode(server.fcManager, server.defParams)
}
} else {
if recv.get("serveChainSince", &p.chainSince) != nil {
if recv.Get("serveChainSince", &p.chainSince) != nil {
p.onlyAnnounce = true
}
if recv.get("serveRecentChain", &p.chainRecent) != nil {
if recv.Get("serveRecentChain", &p.chainRecent) != nil {
p.chainRecent = 0
}
if recv.get("serveStateSince", &p.stateSince) != nil {
if recv.Get("serveStateSince", &p.stateSince) != nil {
p.onlyAnnounce = true
}
if recv.get("serveRecentState", &p.stateRecent) != nil {
if recv.Get("serveRecentState", &p.stateRecent) != nil {
p.stateRecent = 0
}
if recv.get("txRelay", nil) != nil {
if recv.Get("txRelay", nil) != nil {
p.onlyAnnounce = true
}
if p.onlyAnnounce && !p.trusted {
return errResp(ErrUselessPeer, "peer cannot serve requests")
return protocol.ErrResp(protocol.ErrUselessPeer, "peer cannot serve requests")
}
var sParams flowcontrol.ServerParams
if err := recv.get("flowControl/BL", &sParams.BufLimit); err != nil {
if err := recv.Get("flowControl/BL", &sParams.BufLimit); err != nil {
return err
}
if err := recv.get("flowControl/MRR", &sParams.MinRecharge); err != nil {
if err := recv.Get("flowControl/MRR", &sParams.MinRecharge); err != nil {
return err
}
var MRC RequestCostList
if err := recv.get("flowControl/MRC", &MRC); err != nil {
var MRC protocol.RequestCostList
if err := recv.Get("flowControl/MRC", &MRC); err != nil {
return err
}
p.fcParams = sParams
p.fcServer = flowcontrol.NewServerNode(sParams, &mclock.System{})
p.fcCosts = MRC.decode(ProtocolLengths[uint(p.version)])
p.fcCosts = MRC.ToTable(protocol.ProtocolLengths[uint(p.version)])
recv.get("checkpoint/value", &p.checkpoint)
recv.get("checkpoint/registerHeight", &p.checkpointNumber)
recv.Get("checkpoint/value", &p.checkpoint)
recv.Get("checkpoint/registerHeight", &p.checkpointNumber)
if !p.onlyAnnounce {
for msgCode := range reqAvgTimeCost {
if p.fcCosts[msgCode] == nil {
return errResp(ErrUselessPeer, "peer does not support message %d", msgCode)
return protocol.ErrResp(protocol.ErrUselessPeer, "peer does not support message %d", msgCode)
}
}
}
p.server = true
}
p.headInfo = &announceData{Td: rTd, Hash: rHash, Number: rNum}
p.headInfo = &protocol.Announcement{
HeadHeader: protocol.HeadHeader{
Td: rTd,
Hash: rHash,
Number: rNum,
},
}
return nil
}
// updateFlowControl updates the flow control parameters belonging to the server
// node if the announced key/value set contains relevant fields
func (p *peer) updateFlowControl(update keyValueMap) {
func (p *peer) updateFlowControl(update protocol.KeyValueMap) {
if p.fcServer == nil {
return
}
// If any of the flow control params is nil, refuse to update.
var params flowcontrol.ServerParams
if update.get("flowControl/BL", &params.BufLimit) == nil && update.get("flowControl/MRR", &params.MinRecharge) == nil {
if update.Get("flowControl/BL", &params.BufLimit) == nil && update.Get("flowControl/MRR", &params.MinRecharge) == nil {
// todo can light client set a minimal acceptable flow control params?
p.fcParams = params
p.fcServer.UpdateParams(params)
}
var MRC RequestCostList
if update.get("flowControl/MRC", &MRC) == nil {
costUpdate := MRC.decode(ProtocolLengths[uint(p.version)])
var MRC protocol.RequestCostList
if update.Get("flowControl/MRC", &MRC) == nil {
costUpdate := MRC.ToTable(protocol.ProtocolLengths[uint(p.version)])
for code, cost := range costUpdate {
p.fcCosts[code] = cost
}

111
les/peer_test.go
View file

@ -27,12 +27,13 @@ import (
"github.com/ethereum/go-ethereum/crypto"
"github.com/ethereum/go-ethereum/eth"
"github.com/ethereum/go-ethereum/les/flowcontrol"
"github.com/ethereum/go-ethereum/les/protocol"
"github.com/ethereum/go-ethereum/p2p"
"github.com/ethereum/go-ethereum/p2p/enode"
"github.com/ethereum/go-ethereum/rlp"
)
const protocolVersion = lpv2
const protocolVersion = protocol.Lpv2
var (
hash = common.HexToHash("deadbeef")
@ -59,10 +60,10 @@ func TestPeerHandshakeSetAnnounceTypeToAnnounceTypeSignedForTrustedPeer(t *testi
version: protocolVersion,
trusted: true,
rw: &rwStub{
WriteHook: func(recvList keyValueList) {
recv, _ := recvList.decode()
WriteHook: func(recvList protocol.KeyValueList) {
recv, _ := recvList.ToMap()
var reqType uint64
err := recv.get("announceType", &reqType)
err := recv.Get("announceType", &reqType)
if err != nil {
t.Fatal(err)
}
@ -70,18 +71,18 @@ func TestPeerHandshakeSetAnnounceTypeToAnnounceTypeSignedForTrustedPeer(t *testi
t.Fatal("Expected announceTypeSigned")
}
},
ReadHook: func(l keyValueList) keyValueList {
l = l.add("serveHeaders", nil)
l = l.add("serveChainSince", uint64(0))
l = l.add("serveStateSince", uint64(0))
l = l.add("txRelay", nil)
l = l.add("flowControl/BL", uint64(0))
l = l.add("flowControl/MRR", uint64(0))
l = l.add("flowControl/MRC", testCostList(0))
ReadHook: func(l protocol.KeyValueList) protocol.KeyValueList {
l = l.Add("serveHeaders", nil)
l = l.Add("serveChainSince", uint64(0))
l = l.Add("serveStateSince", uint64(0))
l = l.Add("txRelay", nil)
l = l.Add("flowControl/BL", uint64(0))
l = l.Add("flowControl/MRR", uint64(0))
l = l.Add("flowControl/MRC", testCostList(0))
return l
},
},
network: NetworkId,
network: protocol.NetworkId,
}
err := p.Handshake(td, hash, headNum, genesis, nil)
if err != nil {
@ -98,11 +99,11 @@ func TestPeerHandshakeAnnounceTypeSignedForTrustedPeersPeerNotInTrusted(t *testi
Peer: p2p.NewPeer(id, "test peer", []p2p.Cap{}),
version: protocolVersion,
rw: &rwStub{
WriteHook: func(recvList keyValueList) {
WriteHook: func(recvList protocol.KeyValueList) {
// checking that ulc sends to peer allowedRequests=noRequests and announceType != announceTypeSigned
recv, _ := recvList.decode()
recv, _ := recvList.ToMap()
var reqType uint64
err := recv.get("announceType", &reqType)
err := recv.Get("announceType", &reqType)
if err != nil {
t.Fatal(err)
}
@ -110,18 +111,18 @@ func TestPeerHandshakeAnnounceTypeSignedForTrustedPeersPeerNotInTrusted(t *testi
t.Fatal("Expected not announceTypeSigned")
}
},
ReadHook: func(l keyValueList) keyValueList {
l = l.add("serveHeaders", nil)
l = l.add("serveChainSince", uint64(0))
l = l.add("serveStateSince", uint64(0))
l = l.add("txRelay", nil)
l = l.add("flowControl/BL", uint64(0))
l = l.add("flowControl/MRR", uint64(0))
l = l.add("flowControl/MRC", testCostList(0))
ReadHook: func(l protocol.KeyValueList) protocol.KeyValueList {
l = l.Add("serveHeaders", nil)
l = l.Add("serveChainSince", uint64(0))
l = l.Add("serveStateSince", uint64(0))
l = l.Add("txRelay", nil)
l = l.Add("flowControl/BL", uint64(0))
l = l.Add("flowControl/MRR", uint64(0))
l = l.Add("flowControl/MRC", testCostList(0))
return l
},
},
network: NetworkId,
network: protocol.NetworkId,
}
err := p.Handshake(td, hash, headNum, genesis, nil)
if err != nil {
@ -141,13 +142,13 @@ func TestPeerHandshakeDefaultAllRequests(t *testing.T) {
Peer: p2p.NewPeer(id, "test peer", []p2p.Cap{}),
version: protocolVersion,
rw: &rwStub{
ReadHook: func(l keyValueList) keyValueList {
l = l.add("announceType", uint64(announceTypeSigned))
l = l.add("allowedRequests", uint64(0))
ReadHook: func(l protocol.KeyValueList) protocol.KeyValueList {
l = l.Add("announceType", uint64(announceTypeSigned))
l = l.Add("allowedRequests", uint64(0))
return l
},
},
network: NetworkId,
network: protocol.NetworkId,
}
err := p.Handshake(td, hash, headNum, genesis, s)
@ -170,11 +171,11 @@ func TestPeerHandshakeServerSendOnlyAnnounceRequestsHeaders(t *testing.T) {
Peer: p2p.NewPeer(id, "test peer", []p2p.Cap{}),
version: protocolVersion,
rw: &rwStub{
ReadHook: func(l keyValueList) keyValueList {
l = l.add("announceType", uint64(announceTypeSigned))
ReadHook: func(l protocol.KeyValueList) protocol.KeyValueList {
l = l.Add("announceType", uint64(announceTypeSigned))
return l
},
WriteHook: func(l keyValueList) {
WriteHook: func(l protocol.KeyValueList) {
for _, v := range l {
if v.Key == "serveHeaders" ||
v.Key == "serveChainSince" ||
@ -185,7 +186,7 @@ func TestPeerHandshakeServerSendOnlyAnnounceRequestsHeaders(t *testing.T) {
}
},
},
network: NetworkId,
network: protocol.NetworkId,
}
err := p.Handshake(td, hash, headNum, genesis, s)
@ -200,17 +201,17 @@ func TestPeerHandshakeClientReceiveOnlyAnnounceRequestsHeaders(t *testing.T) {
Peer: p2p.NewPeer(id, "test peer", []p2p.Cap{}),
version: protocolVersion,
rw: &rwStub{
ReadHook: func(l keyValueList) keyValueList {
l = l.add("flowControl/BL", uint64(0))
l = l.add("flowControl/MRR", uint64(0))
l = l.add("flowControl/MRC", RequestCostList{})
ReadHook: func(l protocol.KeyValueList) protocol.KeyValueList {
l = l.Add("flowControl/BL", uint64(0))
l = l.Add("flowControl/MRR", uint64(0))
l = l.Add("flowControl/MRC", protocol.RequestCostList{})
l = l.add("announceType", uint64(announceTypeSigned))
l = l.Add("announceType", uint64(announceTypeSigned))
return l
},
},
network: NetworkId,
network: protocol.NetworkId,
trusted: true,
}
@ -231,15 +232,15 @@ func TestPeerHandshakeClientReturnErrorOnUselessPeer(t *testing.T) {
Peer: p2p.NewPeer(id, "test peer", []p2p.Cap{}),
version: protocolVersion,
rw: &rwStub{
ReadHook: func(l keyValueList) keyValueList {
l = l.add("flowControl/BL", uint64(0))
l = l.add("flowControl/MRR", uint64(0))
l = l.add("flowControl/MRC", RequestCostList{})
l = l.add("announceType", uint64(announceTypeSigned))
ReadHook: func(l protocol.KeyValueList) protocol.KeyValueList {
l = l.Add("flowControl/BL", uint64(0))
l = l.Add("flowControl/MRR", uint64(0))
l = l.Add("flowControl/MRC", protocol.RequestCostList{})
l = l.Add("announceType", uint64(announceTypeSigned))
return l
},
},
network: NetworkId,
network: protocol.NetworkId,
}
err := p.Handshake(td, hash, headNum, genesis, nil)
@ -264,18 +265,18 @@ func generateLesServer() *LesServer {
}
type rwStub struct {
ReadHook func(l keyValueList) keyValueList
WriteHook func(l keyValueList)
ReadHook func(l protocol.KeyValueList) protocol.KeyValueList
WriteHook func(l protocol.KeyValueList)
}
func (s *rwStub) ReadMsg() (p2p.Msg, error) {
payload := keyValueList{}
payload = payload.add("protocolVersion", uint64(protocolVersion))
payload = payload.add("networkId", uint64(NetworkId))
payload = payload.add("headTd", td)
payload = payload.add("headHash", hash)
payload = payload.add("headNum", headNum)
payload = payload.add("genesisHash", genesis)
payload := protocol.KeyValueList{}
payload = payload.Add("protocolVersion", uint64(protocolVersion))
payload = payload.Add("networkId", uint64(protocol.NetworkId))
payload = payload.Add("headTd", td)
payload = payload.Add("headHash", hash)
payload = payload.Add("headNum", headNum)
payload = payload.Add("genesisHash", genesis)
if s.ReadHook != nil {
payload = s.ReadHook(payload)
@ -291,7 +292,7 @@ func (s *rwStub) ReadMsg() (p2p.Msg, error) {
}
func (s *rwStub) WriteMsg(m p2p.Msg) error {
recvList := keyValueList{}
recvList := protocol.KeyValueList{}
if err := m.Decode(&recvList); err != nil {
return err
}

237
les/protocol.go
View file

@ -1,237 +0,0 @@
// Copyright 2016 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 (
"crypto/ecdsa"
"errors"
"fmt"
"io"
"math/big"
"github.com/ethereum/go-ethereum/common"
"github.com/ethereum/go-ethereum/crypto"
"github.com/ethereum/go-ethereum/p2p/enode"
"github.com/ethereum/go-ethereum/rlp"
)
// Constants to match up protocol versions and messages
const (
lpv2 = 2
lpv3 = 3
)
// Supported versions of the les protocol (first is primary)
var (
ClientProtocolVersions = []uint{lpv2, lpv3}
ServerProtocolVersions = []uint{lpv2, lpv3}
AdvertiseProtocolVersions = []uint{lpv2} // clients are searching for the first advertised protocol in the list
)
// Number of implemented message corresponding to different protocol versions.
var ProtocolLengths = map[uint]uint64{lpv2: 22, lpv3: 24}
const (
NetworkId = 1
ProtocolMaxMsgSize = 10 * 1024 * 1024 // Maximum cap on the size of a protocol message
)
// les protocol message codes
const (
// Protocol messages inherited from LPV1
StatusMsg = 0x00
AnnounceMsg = 0x01
GetBlockHeadersMsg = 0x02
BlockHeadersMsg = 0x03
GetBlockBodiesMsg = 0x04
BlockBodiesMsg = 0x05
GetReceiptsMsg = 0x06
ReceiptsMsg = 0x07
GetCodeMsg = 0x0a
CodeMsg = 0x0b
// Protocol messages introduced in LPV2
GetProofsV2Msg = 0x0f
ProofsV2Msg = 0x10
GetHelperTrieProofsMsg = 0x11
HelperTrieProofsMsg = 0x12
SendTxV2Msg = 0x13
GetTxStatusMsg = 0x14
TxStatusMsg = 0x15
// Protocol messages introduced in LPV3
StopMsg = 0x16
ResumeMsg = 0x17
)
type requestInfo struct {
name string
maxCount uint64
}
var requests = map[uint64]requestInfo{
GetBlockHeadersMsg: {"GetBlockHeaders", MaxHeaderFetch},
GetBlockBodiesMsg: {"GetBlockBodies", MaxBodyFetch},
GetReceiptsMsg: {"GetReceipts", MaxReceiptFetch},
GetCodeMsg: {"GetCode", MaxCodeFetch},
GetProofsV2Msg: {"GetProofsV2", MaxProofsFetch},
GetHelperTrieProofsMsg: {"GetHelperTrieProofs", MaxHelperTrieProofsFetch},
SendTxV2Msg: {"SendTxV2", MaxTxSend},
GetTxStatusMsg: {"GetTxStatus", MaxTxStatus},
}
type errCode int
const (
ErrMsgTooLarge = iota
ErrDecode
ErrInvalidMsgCode
ErrProtocolVersionMismatch
ErrNetworkIdMismatch
ErrGenesisBlockMismatch
ErrNoStatusMsg
ErrExtraStatusMsg
ErrSuspendedPeer
ErrUselessPeer
ErrRequestRejected
ErrUnexpectedResponse
ErrInvalidResponse
ErrTooManyTimeouts
ErrMissingKey
)
func (e errCode) String() string {
return errorToString[int(e)]
}
// XXX change once legacy code is out
var errorToString = map[int]string{
ErrMsgTooLarge: "Message too long",
ErrDecode: "Invalid message",
ErrInvalidMsgCode: "Invalid message code",
ErrProtocolVersionMismatch: "Protocol version mismatch",
ErrNetworkIdMismatch: "NetworkId mismatch",
ErrGenesisBlockMismatch: "Genesis block mismatch",
ErrNoStatusMsg: "No status message",
ErrExtraStatusMsg: "Extra status message",
ErrSuspendedPeer: "Suspended peer",
ErrRequestRejected: "Request rejected",
ErrUnexpectedResponse: "Unexpected response",
ErrInvalidResponse: "Invalid response",
ErrTooManyTimeouts: "Too many request timeouts",
ErrMissingKey: "Key missing from list",
}
type announceBlock struct {
Hash common.Hash // Hash of one particular block being announced
Number uint64 // Number of one particular block being announced
Td *big.Int // Total difficulty of one particular block being announced
}
// announceData is the network packet for the block announcements.
type announceData struct {
Hash common.Hash // Hash of one particular block being announced
Number uint64 // Number of one particular block being announced
Td *big.Int // Total difficulty of one particular block being announced
ReorgDepth uint64
Update keyValueList
}
// sanityCheck verifies that the values are reasonable, as a DoS protection
func (a *announceData) sanityCheck() error {
if tdlen := a.Td.BitLen(); tdlen > 100 {
return fmt.Errorf("too large block TD: bitlen %d", tdlen)
}
return nil
}
// sign adds a signature to the block announcement by the given privKey
func (a *announceData) sign(privKey *ecdsa.PrivateKey) {
rlp, _ := rlp.EncodeToBytes(announceBlock{a.Hash, a.Number, a.Td})
sig, _ := crypto.Sign(crypto.Keccak256(rlp), privKey)
a.Update = a.Update.add("sign", sig)
}
// checkSignature verifies if the block announcement has a valid signature by the given pubKey
func (a *announceData) checkSignature(id enode.ID, update keyValueMap) error {
var sig []byte
if err := update.get("sign", &sig); err != nil {
return err
}
rlp, _ := rlp.EncodeToBytes(announceBlock{a.Hash, a.Number, a.Td})
recPubkey, err := crypto.SigToPub(crypto.Keccak256(rlp), sig)
if err != nil {
return err
}
if id == enode.PubkeyToIDV4(recPubkey) {
return nil
}
return errors.New("wrong signature")
}
type blockInfo struct {
Hash common.Hash // Hash of one particular block being announced
Number uint64 // Number of one particular block being announced
Td *big.Int // Total difficulty of one particular block being announced
}
// getBlockHeadersData represents a block header query.
type getBlockHeadersData struct {
Origin hashOrNumber // Block from which to retrieve headers
Amount uint64 // Maximum number of headers to retrieve
Skip uint64 // Blocks to skip between consecutive headers
Reverse bool // Query direction (false = rising towards latest, true = falling towards genesis)
}
// hashOrNumber is a combined field for specifying an origin block.
type hashOrNumber struct {
Hash common.Hash // Block hash from which to retrieve headers (excludes Number)
Number uint64 // Block hash from which to retrieve headers (excludes Hash)
}
// EncodeRLP is a specialized encoder for hashOrNumber to encode only one of the
// two contained union fields.
func (hn *hashOrNumber) EncodeRLP(w io.Writer) error {
if hn.Hash == (common.Hash{}) {
return rlp.Encode(w, hn.Number)
}
if hn.Number != 0 {
return fmt.Errorf("both origin hash (%x) and number (%d) provided", hn.Hash, hn.Number)
}
return rlp.Encode(w, hn.Hash)
}
// DecodeRLP is a specialized decoder for hashOrNumber to decode the contents
// into either a block hash or a block number.
func (hn *hashOrNumber) DecodeRLP(s *rlp.Stream) error {
_, size, _ := s.Kind()
origin, err := s.Raw()
if err == nil {
switch {
case size == 32:
err = rlp.DecodeBytes(origin, &hn.Hash)
case size <= 8:
err = rlp.DecodeBytes(origin, &hn.Number)
default:
err = fmt.Errorf("invalid input size %d for origin", size)
}
}
return err
}
// CodeData is the network response packet for a node data retrieval.
type CodeData []struct {
Value []byte
}

86
les/protocol/keyvalueset.go Normal file
View file

@ -0,0 +1,86 @@
// Copyright 2020 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 protocol
import (
"errors"
"github.com/ethereum/go-ethereum/rlp"
)
// ErrNonexistentEntry is returned if the specified key is non-existent.
var ErrNonexistentEntry = errors.New("the entry is non-existent")
// KeyValueEntry is the entry contained in a List or Map
// which can be extended with no limitaion.
type KeyValueEntry struct {
Key string
Value rlp.RawValue
}
// KeyValueList is a set of entries in list format.
//
// Usually KeyValueList is used as the container for
// protocol handshake.
type KeyValueList []KeyValueEntry
// KeyValueMap is a set of entires in map format.
// All entires is identified with its key and saved
// in RLP-encoded format.
//
// Usually KeyValueMap is used as the container for
// protocol handshake.
type KeyValueMap map[string]rlp.RawValue
// Add adds a new entry with specified key and value into list.
func (l KeyValueList) Add(key string, val interface{}) KeyValueList {
var entry KeyValueEntry
entry.Key = key
if val == nil {
val = uint64(0) // Use empty uint64 as default value
}
enc, err := rlp.EncodeToBytes(val)
if err == nil {
entry.Value = enc
}
return append(l, entry)
}
// ToMap converts list format to map format. Also returns
// the total size of converted map.
func (l KeyValueList) ToMap() (KeyValueMap, uint64) {
m := make(KeyValueMap)
var size uint64
for _, entry := range l {
m[entry.Key] = entry.Value
size += uint64(len(entry.Key)) + uint64(len(entry.Value)) + 8
}
return m, size
}
// Get retrieves contained entry with specified key, decode the
// retrieved data in the provided container(interface).
func (m KeyValueMap) Get(key string, val interface{}) error {
enc, ok := m[key]
if !ok {
return ErrNonexistentEntry
}
if val == nil {
return nil
}
return rlp.DecodeBytes(enc, val)
}

47
les/protocol/keyvalueset_test.go Normal file
View file

@ -0,0 +1,47 @@
// Copyright 2020 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 protocol
import (
"testing"
"github.com/ethereum/go-ethereum/rlp"
)
func TestKeyValueSet(t *testing.T) {
var cases = []struct {
key string
value interface{}
}{
// {"key1", uint64(10)},
{"key2", false},
{"key3", nil},
}
var list KeyValueList
for _, c := range cases {
list = list.Add(c.key, c.value)
}
blob, err := rlp.EncodeToBytes(list)
if err != nil {
t.Fatalf("Failed to encode keyvalue list: %v", err)
}
var dec KeyValueList
err = rlp.DecodeBytes(blob, &dec)
if err != nil {
t.Fatalf("Failed to decode keyvalue list: %v", err)
}
}

362
les/protocol/protocol.go Normal file
View file

@ -0,0 +1,362 @@
// Copyright 2016 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 protocol defines all protocol related structures which will be
// used in both server side and client side.
package protocol
import (
"crypto/ecdsa"
"errors"
"fmt"
"io"
"math/big"
"github.com/ethereum/go-ethereum/common"
"github.com/ethereum/go-ethereum/crypto"
"github.com/ethereum/go-ethereum/light"
"github.com/ethereum/go-ethereum/p2p/discv5"
"github.com/ethereum/go-ethereum/p2p/enode"
"github.com/ethereum/go-ethereum/rlp"
)
// Constants to match up protocol versions and messages
const (
Lpv2 = 2
Lpv3 = 3
)
// Supported versions of the les protocol (first is primary)
var (
ClientProtocolVersions = []uint{Lpv2, Lpv3}
ServerProtocolVersions = []uint{Lpv2, Lpv3}
AdvertiseProtocolVersions = []uint{Lpv2} // clients are searching for the first advertised protocol in the list
)
// Number of implemented message corresponding to different protocol versions.
var ProtocolLengths = map[uint]uint64{Lpv2: 22, Lpv3: 24}
const (
NetworkId = 1 // Default ethereum mainnet network ID
ProtocolMaxMsgSize = 10 * 1024 * 1024 // Maximum cap on the size of a protocol message
)
// les protocol message codes
const (
// Protocol messages inherited from LPV1
StatusMsg = 0x00
AnnounceMsg = 0x01
GetBlockHeadersMsg = 0x02
BlockHeadersMsg = 0x03
GetBlockBodiesMsg = 0x04
BlockBodiesMsg = 0x05
GetReceiptsMsg = 0x06
ReceiptsMsg = 0x07
GetCodeMsg = 0x0a
CodeMsg = 0x0b
// Protocol messages introduced in LPV2
GetProofsV2Msg = 0x0f
ProofsV2Msg = 0x10
GetHelperTrieProofsMsg = 0x11
HelperTrieProofsMsg = 0x12
SendTxV2Msg = 0x13
GetTxStatusMsg = 0x14
TxStatusMsg = 0x15
// Protocol messages introduced in LPV3
StopMsg = 0x16
ResumeMsg = 0x17
)
// The maxmium amount of data requested per retrieval request.
const (
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
)
type RequestInfo struct {
Name string
MaxCount uint64
}
var LesRequests = map[uint64]RequestInfo{
GetBlockHeadersMsg: {"GetBlockHeaders", MaxHeaderFetch},
GetBlockBodiesMsg: {"GetBlockBodies", MaxBodyFetch},
GetReceiptsMsg: {"GetReceipts", MaxReceiptFetch},
GetCodeMsg: {"GetCode", MaxCodeFetch},
GetProofsV2Msg: {"GetProofsV2", MaxProofsFetch},
GetHelperTrieProofsMsg: {"GetHelperTrieProofs", MaxHelperTrieProofsFetch},
SendTxV2Msg: {"SendTxV2", MaxTxSend},
GetTxStatusMsg: {"GetTxStatus", MaxTxStatus},
}
type ErrCode int
const (
ErrMsgTooLarge = iota
ErrDecode
ErrInvalidMsgCode
ErrProtocolVersionMismatch
ErrNetworkIdMismatch
ErrGenesisBlockMismatch
ErrNoStatusMsg
ErrExtraStatusMsg
ErrSuspendedPeer
ErrUselessPeer
ErrRequestRejected
ErrUnexpectedResponse
ErrInvalidResponse
ErrTooManyTimeouts
ErrMissingKey
)
func (e ErrCode) String() string {
return errorToString[int(e)]
}
// XXX change once legacy code is out
var errorToString = map[int]string{
ErrMsgTooLarge: "Message too long",
ErrDecode: "Invalid message",
ErrInvalidMsgCode: "Invalid message code",
ErrProtocolVersionMismatch: "Protocol version mismatch",
ErrNetworkIdMismatch: "NetworkId mismatch",
ErrGenesisBlockMismatch: "Genesis block mismatch",
ErrNoStatusMsg: "No status message",
ErrExtraStatusMsg: "Extra status message",
ErrSuspendedPeer: "Suspended peer",
ErrRequestRejected: "Request rejected",
ErrUnexpectedResponse: "Unexpected response",
ErrInvalidResponse: "Invalid response",
ErrTooManyTimeouts: "Too many request timeouts",
ErrMissingKey: "Key missing from list",
}
// HeadHeader is the a part of announcement sent by the LES server to the
// LES client when a new block is generated in the network.
//
// HeadHeader can also be used to represent the head info of peer(both server
// and client).
type HeadHeader struct {
Hash common.Hash // Hash of one particular block being announced
Number uint64 // Number of one particular block being announced
Td *big.Int // Total difficulty of one particular block being announced
}
// Announcement is a network packet sent by the LES server to the LES client
// when a new block is generated in the network or the server has protocol
// parameters that need to be updated.
type Announcement struct {
HeadHeader // The data of new arrival header
ReorgDepth uint64 // The reorg depth of new arrival header
Update KeyValueList // Updated protocol parameters
}
// SanityCheck verifies that the values are reasonable, as a DoS protection
func (a *Announcement) SanityCheck() error {
if tdlen := a.Td.BitLen(); tdlen > 100 {
return fmt.Errorf("too large block TD: bitlen %d", tdlen)
}
return nil
}
// Sign adds a signature to the block announcement by the given privKey
func (a *Announcement) Sign(privKey *ecdsa.PrivateKey) {
rlp, _ := rlp.EncodeToBytes(HeadHeader{a.Hash, a.Number, a.Td})
sig, _ := crypto.Sign(crypto.Keccak256(rlp), privKey)
a.Update = a.Update.Add("sign", sig)
}
// CheckSignature verifies if the block announcement has a valid signature
// by the given pubKey.
func (a *Announcement) CheckSignature(id enode.ID, update KeyValueMap) error {
var sig []byte
if err := update.Get("sign", &sig); err != nil {
return err
}
rlp, _ := rlp.EncodeToBytes(HeadHeader{a.Hash, a.Number, a.Td})
recPubkey, err := crypto.SigToPub(crypto.Keccak256(rlp), sig)
if err != nil {
return err
}
if id == enode.PubkeyToIDV4(recPubkey) {
return nil
}
return errors.New("wrong signature")
}
// GetBlockHeadersRequest represents a block header query sent by les client.
type GetBlockHeadersRequest struct {
Origin HashOrNumber // Block from which to retrieve headers
Amount uint64 // Maximum number of headers to retrieve
Skip uint64 // Blocks to skip between consecutive headers
Reverse bool // Query direction (false = rising towards latest, true = falling towards genesis)
}
// HashOrNumber is a combined field for specifying an origin block.
type HashOrNumber struct {
Hash common.Hash // Block hash from which to retrieve headers (excludes Number)
Number uint64 // Block hash from which to retrieve headers (excludes Hash)
}
// EncodeRLP is a specialized encoder for HashOrNumber to encode only one of the
// two contained union fields.
func (hn *HashOrNumber) EncodeRLP(w io.Writer) error {
if hn.Hash == (common.Hash{}) {
return rlp.Encode(w, hn.Number)
}
if hn.Number != 0 {
return fmt.Errorf("both origin hash (%x) and number (%d) provided", hn.Hash, hn.Number)
}
return rlp.Encode(w, hn.Hash)
}
// DecodeRLP is a specialized decoder for HashOrNumber to decode the contents
// into either a block hash or a block number.
func (hn *HashOrNumber) DecodeRLP(s *rlp.Stream) error {
_, size, _ := s.Kind()
origin, err := s.Raw()
if err == nil {
switch {
case size == 32:
err = rlp.DecodeBytes(origin, &hn.Hash)
case size <= 8:
err = rlp.DecodeBytes(origin, &hn.Number)
default:
err = fmt.Errorf("invalid input size %d for origin", size)
}
}
return err
}
// TrieProofRequest represents a state/storage trie proof query
// sent by les client.
type TrieProofRequest struct {
BlockHash common.Hash // The corresponding block hash of state
Account []byte // The address of target account, nil if it's a global state trie proof request
Key []byte // The key of target storage slot or account
FromLevel uint // The node level beyond which all trie nodes are contained in the proof
}
// CodeRequest represents a contract code query sent by les client.
type CodeRequest struct {
BlockHash common.Hash // The corresponding block hash of state
Account []byte // The address of target account
}
const (
// HelperTrieCHT is the indicator of canonical hash trie, check
// https://github.com/ethereum/devp2p/blob/master/caps/les.md#canonical-hash-trie
// for more details.
HelperTrieCHT = iota
// HelperTrieBloomTrie is the indicator of bloom trie, check
// https://github.com/ethereum/devp2p/blob/master/caps/les.md#bloombits-trie
// for more details
HelperTrieBloomTrie
// The auxiliary data type of helperTrie request which is available for
// all helperTrie request.
AuxRoot = 1
// The auxiliary data type of CHT request - corresponding block header
// which is only avaiable for CHT request.
AuxHeader = 2
)
// HelperTrieRequest represents a helper trie query sent by les client.
// HelperTrie includes: CHT and bloom trie. It's a shared structure between
// these two kinds of request.
//
// Except the helperTrie proof of requested entry will be returned, caller
// can specify more additional auxiliary data to be returned via `AuxType`.
type HelperTrieRequest struct {
Type uint // Indicator of request type, 0 represents CHT, 1 represents Bloom trie
TrieIndex uint64 // The index(section index) of requested trie
Key []byte // The list of entry keys, caller can request a batch of entries in a single request.
FromLevel uint // The node level beyond which all trie nodes are contained in the proof
AuxType uint // The type of auxiliary data requested
}
// HelperTrieResponse represents the response of corresponding helperTrie
// request. A single response contains a batch of requested proofs and
// corresponding auxiliary data.
type HelperTrieResponse struct {
Proofs light.NodeList // The container for storing all requested proofs
AuxData [][]byte // The batch of requested auxiliary data
}
// ErrResp returns an protocol error with given error code and additional
// error message.
func ErrResp(code ErrCode, format string, v ...interface{}) error {
return fmt.Errorf("%v - %v", code, fmt.Sprintf(format, v...))
}
// LesTopic constructs the discovery v5 topic for LES protocol.
func LesTopic(genesisHash common.Hash, protocolVersion uint) discv5.Topic {
var name string
switch protocolVersion {
case Lpv2:
name = "LES2"
default:
panic(nil)
}
return discv5.Topic(name + "@" + common.Bytes2Hex(genesisHash.Bytes()[0:8]))
}
type (
// RequestCost represents a cost policy of a specified request type.
RequestCost struct {
BaseCost, ReqCost uint64
}
// RequestCostTable assigns a cost estimate function to each request type
// which is a linear function of the requested amount
// (cost = BaseCost + ReqCost * amount)
RequestCostTable map[uint64]*RequestCost
// RequestCostList is a list representation of request costs which is used for
// database storage and communication through the network
RequestCostList []RequestCostListItem
RequestCostListItem struct {
MsgCode, BaseCost, ReqCost uint64
}
)
// GetMaxCost calculates the estimated cost for a given request type and amount
func (table RequestCostTable) GetMaxCost(code, amount uint64) uint64 {
costs := table[code]
return costs.BaseCost + amount*costs.ReqCost
}
// ToTable converts a cost list to a cost table
func (list RequestCostList) ToTable(protocolLength uint64) RequestCostTable {
table := make(RequestCostTable)
for _, e := range list {
if e.MsgCode < protocolLength {
table[e.MsgCode] = &RequestCost{
BaseCost: e.BaseCost,
ReqCost: e.ReqCost,
}
}
}
return table
}

7
les/retrieve.go
View file

@ -25,6 +25,7 @@ import (
"time"
"github.com/ethereum/go-ethereum/common/mclock"
"github.com/ethereum/go-ethereum/les/protocol"
"github.com/ethereum/go-ethereum/light"
)
@ -171,7 +172,7 @@ func (rm *retrieveManager) deliver(peer distPeer, msg *Msg) error {
if ok {
return req.deliver(peer, msg)
}
return errResp(ErrUnexpectedResponse, "reqID = %v", msg.ReqID)
return protocol.ErrResp(protocol.ErrUnexpectedResponse, "reqID = %v", msg.ReqID)
}
// frozen is called by the LES protocol manager when a server has suspended its service and we
@ -389,7 +390,7 @@ func (r *sentReq) deliver(peer distPeer, msg *Msg) error {
s, ok := r.sentTo[peer]
if !ok || s.delivered {
return errResp(ErrUnexpectedResponse, "reqID = %v", msg.ReqID)
return protocol.ErrResp(protocol.ErrUnexpectedResponse, "reqID = %v", msg.ReqID)
}
if s.frozen {
return nil
@ -402,7 +403,7 @@ func (r *sentReq) deliver(peer distPeer, msg *Msg) error {
s.event <- rpDeliveredInvalid
}
if !valid {
return errResp(ErrInvalidResponse, "reqID = %v", msg.ReqID)
return protocol.ErrResp(protocol.ErrInvalidResponse, "reqID = %v", msg.ReqID)
}
return nil
}

9
les/server.go
View file

@ -26,6 +26,7 @@ import (
"github.com/ethereum/go-ethereum/eth"
"github.com/ethereum/go-ethereum/les/checkpointoracle"
"github.com/ethereum/go-ethereum/les/flowcontrol"
"github.com/ethereum/go-ethereum/les/protocol"
"github.com/ethereum/go-ethereum/light"
"github.com/ethereum/go-ethereum/log"
"github.com/ethereum/go-ethereum/p2p"
@ -58,9 +59,9 @@ type LesServer struct {
func NewLesServer(e *eth.Ethereum, config *eth.Config) (*LesServer, error) {
// Collect les protocol version information supported by local node.
lesTopics := make([]discv5.Topic, len(AdvertiseProtocolVersions))
for i, pv := range AdvertiseProtocolVersions {
lesTopics[i] = lesTopic(e.BlockChain().Genesis().Hash(), pv)
lesTopics := make([]discv5.Topic, len(protocol.AdvertiseProtocolVersions))
for i, pv := range protocol.AdvertiseProtocolVersions {
lesTopics[i] = protocol.LesTopic(e.BlockChain().Genesis().Hash(), pv)
}
// Calculate the number of threads used to service the light client
// requests based on the user-specified value.
@ -151,7 +152,7 @@ func (s *LesServer) APIs() []rpc.API {
}
func (s *LesServer) Protocols() []p2p.Protocol {
ps := s.makeProtocols(ServerProtocolVersions, s.handler.runPeer, func(id enode.ID) interface{} {
ps := s.makeProtocols(protocol.ServerProtocolVersions, s.handler.runPeer, func(id enode.ID) interface{} {
if p := s.peers.Peer(peerIdToString(id)); p != nil {
return p.Info()
}

135
les/server_handler.go
View file

@ -31,6 +31,7 @@ import (
"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/protocol"
"github.com/ethereum/go-ethereum/light"
"github.com/ethereum/go-ethereum/log"
"github.com/ethereum/go-ethereum/metrics"
@ -43,15 +44,6 @@ 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
ethVersion = 63 // equivalent eth version for the downloader
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 (
@ -182,9 +174,9 @@ func (h *serverHandler) handleMsg(p *peer, wg *sync.WaitGroup) error {
p.Log().Trace("Light Ethereum message arrived", "code", msg.Code, "bytes", msg.Size)
// Discard large message which exceeds the limitation.
if msg.Size > ProtocolMaxMsgSize {
if msg.Size > protocol.ProtocolMaxMsgSize {
clientErrorMeter.Mark(1)
return errResp(ErrMsgTooLarge, "%v > %v", msg.Size, ProtocolMaxMsgSize)
return protocol.ErrResp(protocol.ErrMsgTooLarge, "%v > %v", msg.Size, protocol.ProtocolMaxMsgSize)
}
defer msg.Discard()
@ -204,7 +196,7 @@ func (h *serverHandler) handleMsg(p *peer, wg *sync.WaitGroup) error {
return false
}
// Prepaid max cost units before request been serving.
maxCost = p.fcCosts.getMaxCost(msg.Code, reqCnt)
maxCost = p.fcCosts.GetMaxCost(msg.Code, reqCnt)
accepted, bufShort, priority := p.fcClient.AcceptRequest(reqID, responseCount, maxCost)
if !accepted {
p.freezeClient()
@ -268,7 +260,7 @@ func (h *serverHandler) handleMsg(p *peer, wg *sync.WaitGroup) error {
}
}
switch msg.Code {
case GetBlockHeadersMsg:
case protocol.GetBlockHeadersMsg:
p.Log().Trace("Received block header request")
if metrics.EnabledExpensive {
miscInHeaderPacketsMeter.Mark(1)
@ -277,14 +269,14 @@ func (h *serverHandler) handleMsg(p *peer, wg *sync.WaitGroup) error {
}
var req struct {
ReqID uint64
Query getBlockHeadersData
Query protocol.GetBlockHeadersRequest
}
if err := msg.Decode(&req); err != nil {
clientErrorMeter.Mark(1)
return errResp(ErrDecode, "%v: %v", msg, err)
return protocol.ErrResp(protocol.ErrDecode, "%v: %v", msg, err)
}
query := req.Query
if accept(req.ReqID, query.Amount, MaxHeaderFetch) {
if accept(req.ReqID, query.Amount, protocol.MaxHeaderFetch) {
wg.Add(1)
go func() {
defer wg.Done()
@ -381,7 +373,7 @@ func (h *serverHandler) handleMsg(p *peer, wg *sync.WaitGroup) error {
}()
}
case GetBlockBodiesMsg:
case protocol.GetBlockBodiesMsg:
p.Log().Trace("Received block bodies request")
if metrics.EnabledExpensive {
miscInBodyPacketsMeter.Mark(1)
@ -394,14 +386,14 @@ func (h *serverHandler) handleMsg(p *peer, wg *sync.WaitGroup) error {
}
if err := msg.Decode(&req); err != nil {
clientErrorMeter.Mark(1)
return errResp(ErrDecode, "msg %v: %v", msg, err)
return protocol.ErrResp(protocol.ErrDecode, "msg %v: %v", msg, err)
}
var (
bytes int
bodies []rlp.RawValue
)
reqCnt := len(req.Hashes)
if accept(req.ReqID, uint64(reqCnt), MaxBodyFetch) {
if accept(req.ReqID, uint64(reqCnt), protocol.MaxBodyFetch) {
wg.Add(1)
go func() {
defer wg.Done()
@ -430,7 +422,7 @@ func (h *serverHandler) handleMsg(p *peer, wg *sync.WaitGroup) error {
}()
}
case GetCodeMsg:
case protocol.GetCodeMsg:
p.Log().Trace("Received code request")
if metrics.EnabledExpensive {
miscInCodePacketsMeter.Mark(1)
@ -439,18 +431,18 @@ func (h *serverHandler) handleMsg(p *peer, wg *sync.WaitGroup) error {
}
var req struct {
ReqID uint64
Reqs []CodeReq
Reqs []protocol.CodeRequest
}
if err := msg.Decode(&req); err != nil {
clientErrorMeter.Mark(1)
return errResp(ErrDecode, "msg %v: %v", msg, err)
return protocol.ErrResp(protocol.ErrDecode, "msg %v: %v", msg, err)
}
var (
bytes int
data [][]byte
)
reqCnt := len(req.Reqs)
if accept(req.ReqID, uint64(reqCnt), MaxCodeFetch) {
if accept(req.ReqID, uint64(reqCnt), protocol.MaxCodeFetch) {
wg.Add(1)
go func() {
defer wg.Done()
@ -460,9 +452,9 @@ func (h *serverHandler) handleMsg(p *peer, wg *sync.WaitGroup) error {
return
}
// Look up the root hash belonging to the request
header := h.blockchain.GetHeaderByHash(request.BHash)
header := h.blockchain.GetHeaderByHash(request.BlockHash)
if header == nil {
p.Log().Warn("Failed to retrieve associate header for code", "hash", request.BHash)
p.Log().Warn("Failed to retrieve associate header for code", "hash", request.BlockHash)
atomic.AddUint32(&p.invalidCount, 1)
continue
}
@ -476,15 +468,15 @@ func (h *serverHandler) handleMsg(p *peer, wg *sync.WaitGroup) error {
}
triedb := h.blockchain.StateCache().TrieDB()
account, err := h.getAccount(triedb, header.Root, common.BytesToHash(request.AccKey))
account, err := h.getAccount(triedb, header.Root, common.BytesToHash(request.Account))
if err != nil {
p.Log().Warn("Failed to retrieve account for code", "block", header.Number, "hash", header.Hash(), "account", common.BytesToHash(request.AccKey), "err", err)
p.Log().Warn("Failed to retrieve account for code", "block", header.Number, "hash", header.Hash(), "account", common.BytesToHash(request.Account), "err", err)
atomic.AddUint32(&p.invalidCount, 1)
continue
}
code, err := triedb.Node(common.BytesToHash(account.CodeHash))
if err != nil {
p.Log().Warn("Failed to retrieve account code", "block", header.Number, "hash", header.Hash(), "account", common.BytesToHash(request.AccKey), "codehash", common.BytesToHash(account.CodeHash), "err", err)
p.Log().Warn("Failed to retrieve account code", "block", header.Number, "hash", header.Hash(), "account", common.BytesToHash(request.Account), "codehash", common.BytesToHash(account.CodeHash), "err", err)
continue
}
// Accumulate the code and abort if enough data was retrieved
@ -502,7 +494,7 @@ func (h *serverHandler) handleMsg(p *peer, wg *sync.WaitGroup) error {
}()
}
case GetReceiptsMsg:
case protocol.GetReceiptsMsg:
p.Log().Trace("Received receipts request")
if metrics.EnabledExpensive {
miscInReceiptPacketsMeter.Mark(1)
@ -515,14 +507,14 @@ func (h *serverHandler) handleMsg(p *peer, wg *sync.WaitGroup) error {
}
if err := msg.Decode(&req); err != nil {
clientErrorMeter.Mark(1)
return errResp(ErrDecode, "msg %v: %v", msg, err)
return protocol.ErrResp(protocol.ErrDecode, "msg %v: %v", msg, err)
}
var (
bytes int
receipts []rlp.RawValue
)
reqCnt := len(req.Hashes)
if accept(req.ReqID, uint64(reqCnt), MaxReceiptFetch) {
if accept(req.ReqID, uint64(reqCnt), protocol.MaxReceiptFetch) {
wg.Add(1)
go func() {
defer wg.Done()
@ -559,7 +551,7 @@ func (h *serverHandler) handleMsg(p *peer, wg *sync.WaitGroup) error {
}()
}
case GetProofsV2Msg:
case protocol.GetProofsV2Msg:
p.Log().Trace("Received les/2 proofs request")
if metrics.EnabledExpensive {
miscInTrieProofPacketsMeter.Mark(1)
@ -568,11 +560,11 @@ func (h *serverHandler) handleMsg(p *peer, wg *sync.WaitGroup) error {
}
var req struct {
ReqID uint64
Reqs []ProofReq
Reqs []protocol.TrieProofRequest
}
if err := msg.Decode(&req); err != nil {
clientErrorMeter.Mark(1)
return errResp(ErrDecode, "msg %v: %v", msg, err)
return protocol.ErrResp(protocol.ErrDecode, "msg %v: %v", msg, err)
}
// Gather state data until the fetch or network limits is reached
var (
@ -580,7 +572,7 @@ func (h *serverHandler) handleMsg(p *peer, wg *sync.WaitGroup) error {
root common.Hash
)
reqCnt := len(req.Reqs)
if accept(req.ReqID, uint64(reqCnt), MaxProofsFetch) {
if accept(req.ReqID, uint64(reqCnt), protocol.MaxProofsFetch) {
wg.Add(1)
go func() {
defer wg.Done()
@ -596,11 +588,11 @@ func (h *serverHandler) handleMsg(p *peer, wg *sync.WaitGroup) error {
header *types.Header
trie state.Trie
)
if request.BHash != lastBHash {
root, lastBHash = common.Hash{}, request.BHash
if request.BlockHash != lastBHash {
root, lastBHash = common.Hash{}, request.BlockHash
if header = h.blockchain.GetHeaderByHash(request.BHash); header == nil {
p.Log().Warn("Failed to retrieve header for proof", "hash", request.BHash)
if header = h.blockchain.GetHeaderByHash(request.BlockHash); header == nil {
p.Log().Warn("Failed to retrieve header for proof", "hash", request.BlockHash)
atomic.AddUint32(&p.invalidCount, 1)
continue
}
@ -622,7 +614,7 @@ func (h *serverHandler) handleMsg(p *peer, wg *sync.WaitGroup) error {
// Open the account or storage trie for the request
statedb := h.blockchain.StateCache()
switch len(request.AccKey) {
switch len(request.Account) {
case 0:
// No account key specified, open an account trie
trie, err = statedb.OpenTrie(root)
@ -632,15 +624,15 @@ func (h *serverHandler) handleMsg(p *peer, wg *sync.WaitGroup) error {
}
default:
// Account key specified, open a storage trie
account, err := h.getAccount(statedb.TrieDB(), root, common.BytesToHash(request.AccKey))
account, err := h.getAccount(statedb.TrieDB(), root, common.BytesToHash(request.Account))
if err != nil {
p.Log().Warn("Failed to retrieve account for proof", "block", header.Number, "hash", header.Hash(), "account", common.BytesToHash(request.AccKey), "err", err)
p.Log().Warn("Failed to retrieve account for proof", "block", header.Number, "hash", header.Hash(), "account", common.BytesToHash(request.Account), "err", err)
atomic.AddUint32(&p.invalidCount, 1)
continue
}
trie, err = statedb.OpenStorageTrie(common.BytesToHash(request.AccKey), account.Root)
trie, err = statedb.OpenStorageTrie(common.BytesToHash(request.Account), account.Root)
if trie == nil || err != nil {
p.Log().Warn("Failed to open storage trie for proof", "block", header.Number, "hash", header.Hash(), "account", common.BytesToHash(request.AccKey), "root", account.Root, "err", err)
p.Log().Warn("Failed to open storage trie for proof", "block", header.Number, "hash", header.Hash(), "account", common.BytesToHash(request.Account), "root", account.Root, "err", err)
continue
}
}
@ -662,7 +654,7 @@ func (h *serverHandler) handleMsg(p *peer, wg *sync.WaitGroup) error {
}()
}
case GetHelperTrieProofsMsg:
case protocol.GetHelperTrieProofsMsg:
p.Log().Trace("Received helper trie proof request")
if metrics.EnabledExpensive {
miscInHelperTriePacketsMeter.Mark(1)
@ -671,11 +663,11 @@ func (h *serverHandler) handleMsg(p *peer, wg *sync.WaitGroup) error {
}
var req struct {
ReqID uint64
Reqs []HelperTrieReq
Reqs []protocol.HelperTrieRequest
}
if err := msg.Decode(&req); err != nil {
clientErrorMeter.Mark(1)
return errResp(ErrDecode, "msg %v: %v", msg, err)
return protocol.ErrResp(protocol.ErrDecode, "msg %v: %v", msg, err)
}
// Gather state data until the fetch or network limits is reached
var (
@ -683,7 +675,7 @@ func (h *serverHandler) handleMsg(p *peer, wg *sync.WaitGroup) error {
auxData [][]byte
)
reqCnt := len(req.Reqs)
if accept(req.ReqID, uint64(reqCnt), MaxHelperTrieProofsFetch) {
if accept(req.ReqID, uint64(reqCnt), protocol.MaxHelperTrieProofsFetch) {
wg.Add(1)
go func() {
defer wg.Done()
@ -699,15 +691,15 @@ func (h *serverHandler) handleMsg(p *peer, wg *sync.WaitGroup) error {
sendResponse(req.ReqID, 0, nil, task.servingTime)
return
}
if auxTrie == nil || request.Type != lastType || request.TrieIdx != lastIdx {
auxTrie, lastType, lastIdx = nil, request.Type, request.TrieIdx
if auxTrie == nil || request.Type != lastType || request.TrieIndex != lastIdx {
auxTrie, lastType, lastIdx = nil, request.Type, request.TrieIndex
var prefix string
if root, prefix = h.getHelperTrie(request.Type, request.TrieIdx); root != (common.Hash{}) {
if root, prefix = h.getHelperTrie(request.Type, request.TrieIndex); root != (common.Hash{}) {
auxTrie, _ = trie.New(root, trie.NewDatabase(rawdb.NewTable(h.chainDb, prefix)))
}
}
if request.AuxReq == auxRoot {
if request.AuxType == protocol.AuxRoot {
var data []byte
if root != (common.Hash{}) {
data = root[:]
@ -718,7 +710,7 @@ func (h *serverHandler) handleMsg(p *peer, wg *sync.WaitGroup) error {
if auxTrie != nil {
auxTrie.Prove(request.Key, request.FromLevel, nodes)
}
if request.AuxReq != 0 {
if request.AuxType != 0 {
data := h.getAuxiliaryHeaders(request)
auxData = append(auxData, data)
auxBytes += len(data)
@ -728,7 +720,7 @@ func (h *serverHandler) handleMsg(p *peer, wg *sync.WaitGroup) error {
break
}
}
reply := p.ReplyHelperTrieProofs(req.ReqID, HelperTrieResps{Proofs: nodes.NodeList(), AuxData: auxData})
reply := p.ReplyHelperTrieProofs(req.ReqID, protocol.HelperTrieResponse{Proofs: nodes.NodeList(), AuxData: auxData})
sendResponse(req.ReqID, uint64(reqCnt), reply, task.done())
if metrics.EnabledExpensive {
miscOutHelperTriePacketsMeter.Mark(1)
@ -737,7 +729,7 @@ func (h *serverHandler) handleMsg(p *peer, wg *sync.WaitGroup) error {
}()
}
case SendTxV2Msg:
case protocol.SendTxV2Msg:
p.Log().Trace("Received new transactions")
if metrics.EnabledExpensive {
miscInTxsPacketsMeter.Mark(1)
@ -750,10 +742,10 @@ func (h *serverHandler) handleMsg(p *peer, wg *sync.WaitGroup) error {
}
if err := msg.Decode(&req); err != nil {
clientErrorMeter.Mark(1)
return errResp(ErrDecode, "msg %v: %v", msg, err)
return protocol.ErrResp(protocol.ErrDecode, "msg %v: %v", msg, err)
}
reqCnt := len(req.Txs)
if accept(req.ReqID, uint64(reqCnt), MaxTxSend) {
if accept(req.ReqID, uint64(reqCnt), protocol.MaxTxSend) {
wg.Add(1)
go func() {
defer wg.Done()
@ -786,7 +778,7 @@ func (h *serverHandler) handleMsg(p *peer, wg *sync.WaitGroup) error {
}()
}
case GetTxStatusMsg:
case protocol.GetTxStatusMsg:
p.Log().Trace("Received transaction status query request")
if metrics.EnabledExpensive {
miscInTxStatusPacketsMeter.Mark(1)
@ -799,10 +791,10 @@ func (h *serverHandler) handleMsg(p *peer, wg *sync.WaitGroup) error {
}
if err := msg.Decode(&req); err != nil {
clientErrorMeter.Mark(1)
return errResp(ErrDecode, "msg %v: %v", msg, err)
return protocol.ErrResp(protocol.ErrDecode, "msg %v: %v", msg, err)
}
reqCnt := len(req.Hashes)
if accept(req.ReqID, uint64(reqCnt), MaxTxStatus) {
if accept(req.ReqID, uint64(reqCnt), protocol.MaxTxStatus) {
wg.Add(1)
go func() {
defer wg.Done()
@ -826,7 +818,7 @@ func (h *serverHandler) handleMsg(p *peer, wg *sync.WaitGroup) error {
default:
p.Log().Trace("Received invalid message", "code", msg.Code)
clientErrorMeter.Mark(1)
return errResp(ErrInvalidMsgCode, "%v", msg.Code)
return protocol.ErrResp(protocol.ErrInvalidMsgCode, "%v", msg.Code)
}
// If the client has made too much invalid request(e.g. request a non-exist data),
// reject them to prevent SPAM attack.
@ -857,10 +849,10 @@ func (h *serverHandler) getAccount(triedb *trie.Database, root, hash common.Hash
// getHelperTrie returns the post-processed trie root for the given trie ID and section index
func (h *serverHandler) getHelperTrie(typ uint, index uint64) (common.Hash, string) {
switch typ {
case htCanonical:
case protocol.HelperTrieCHT:
sectionHead := rawdb.ReadCanonicalHash(h.chainDb, (index+1)*h.server.iConfig.ChtSize-1)
return light.GetChtRoot(h.chainDb, index, sectionHead), light.ChtTablePrefix
case htBloomBits:
case protocol.HelperTrieBloomTrie:
sectionHead := rawdb.ReadCanonicalHash(h.chainDb, (index+1)*h.server.iConfig.BloomTrieSize-1)
return light.GetBloomTrieRoot(h.chainDb, index, sectionHead), light.BloomTrieTablePrefix
}
@ -868,8 +860,8 @@ func (h *serverHandler) getHelperTrie(typ uint, index uint64) (common.Hash, stri
}
// getAuxiliaryHeaders returns requested auxiliary headers for the CHT request.
func (h *serverHandler) getAuxiliaryHeaders(req HelperTrieReq) []byte {
if req.Type == htCanonical && req.AuxReq == auxHeader && len(req.Key) == 8 {
func (h *serverHandler) getAuxiliaryHeaders(req protocol.HelperTrieRequest) []byte {
if req.Type == protocol.HelperTrieCHT && req.AuxType == protocol.AuxHeader && len(req.Key) == 8 {
blockNum := binary.BigEndian.Uint64(req.Key)
hash := rawdb.ReadCanonicalHash(h.chainDb, blockNum)
return rawdb.ReadHeaderRLP(h.chainDb, hash, blockNum)
@ -931,9 +923,16 @@ func (h *serverHandler) broadcastHeaders() {
log.Debug("Announcing block to peers", "number", number, "hash", hash, "td", td, "reorg", reorg)
var (
signed bool
signedAnnounce announceData
signedAnnounce protocol.Announcement
)
announce := announceData{Hash: hash, Number: number, Td: td, ReorgDepth: reorg}
announce := protocol.Announcement{
HeadHeader: protocol.HeadHeader{
Hash: hash,
Number: number,
Td: td,
},
ReorgDepth: reorg,
}
for _, p := range peers {
p := p
switch p.announceType {
@ -942,7 +941,7 @@ func (h *serverHandler) broadcastHeaders() {
case announceTypeSigned:
if !signed {
signedAnnounce = announce
signedAnnounce.sign(h.server.privateKey)
signedAnnounce.Sign(h.server.privateKey)
signed = true
}
p.queueSend(func() { p.SendAnnounce(signedAnnounce) })

49
les/test_helper.go
View file

@ -41,6 +41,7 @@ import (
"github.com/ethereum/go-ethereum/event"
"github.com/ethereum/go-ethereum/les/checkpointoracle"
"github.com/ethereum/go-ethereum/les/flowcontrol"
"github.com/ethereum/go-ethereum/les/protocol"
"github.com/ethereum/go-ethereum/light"
"github.com/ethereum/go-ethereum/p2p"
"github.com/ethereum/go-ethereum/p2p/enode"
@ -200,7 +201,7 @@ func newTestClientHandler(backend *backends.SimulatedBackend, odr *LesOdr, index
client := &LightEthereum{
lesCommons: lesCommons{
genesis: genesis.Hash(),
config: &eth.Config{LightPeers: 100, NetworkId: NetworkId},
config: &eth.Config{LightPeers: 100, NetworkId: protocol.NetworkId},
chainConfig: params.AllEthashProtocolChanges,
iConfig: light.TestClientIndexerConfig,
chainDb: db,
@ -263,7 +264,7 @@ func newTestServerHandler(blocks int, indexers []*core.ChainIndexer, db ethdb.Da
server := &LesServer{
lesCommons: lesCommons{
genesis: genesis.Hash(),
config: &eth.Config{LightPeers: 100, NetworkId: NetworkId},
config: &eth.Config{LightPeers: 100, NetworkId: protocol.NetworkId},
chainConfig: params.AllEthashProtocolChanges,
iConfig: light.TestServerIndexerConfig,
chainDb: db,
@ -308,7 +309,7 @@ func newTestPeer(t *testing.T, name string, version int, handler *serverHandler,
// Generate a random id and create the peer
var id enode.ID
rand.Read(id[:])
peer := newPeer(version, NetworkId, false, p2p.NewPeer(id, name, nil), net)
peer := newPeer(version, protocol.NetworkId, false, p2p.NewPeer(id, name, nil), net)
// Start the peer on a new thread
errCh := make(chan error, 1)
@ -354,8 +355,8 @@ func newTestPeerPair(name string, version int, server *serverHandler, client *cl
var id enode.ID
rand.Read(id[:])
peer1 := newPeer(version, NetworkId, false, p2p.NewPeer(id, name, nil), net)
peer2 := newPeer(version, NetworkId, false, p2p.NewPeer(id, name, nil), app)
peer1 := newPeer(version, protocol.NetworkId, false, p2p.NewPeer(id, name, nil), net)
peer2 := newPeer(version, protocol.NetworkId, false, p2p.NewPeer(id, name, nil), app)
// Start the peer on a new thread
errc1 := make(chan error, 1)
@ -379,29 +380,29 @@ func newTestPeerPair(name string, version int, server *serverHandler, client *cl
// handshake simulates a trivial handshake that expects the same state from the
// remote side as we are simulating locally.
func (p *testPeer) handshake(t *testing.T, td *big.Int, head common.Hash, headNum uint64, genesis common.Hash, costList RequestCostList) {
var expList keyValueList
expList = expList.add("protocolVersion", uint64(p.peer.version))
expList = expList.add("networkId", uint64(NetworkId))
expList = expList.add("headTd", td)
expList = expList.add("headHash", head)
expList = expList.add("headNum", headNum)
expList = expList.add("genesisHash", genesis)
sendList := make(keyValueList, len(expList))
func (p *testPeer) handshake(t *testing.T, td *big.Int, head common.Hash, headNum uint64, genesis common.Hash, costList protocol.RequestCostList) {
var expList protocol.KeyValueList
expList = expList.Add("protocolVersion", uint64(p.peer.version))
expList = expList.Add("networkId", uint64(protocol.NetworkId))
expList = expList.Add("headTd", td)
expList = expList.Add("headHash", head)
expList = expList.Add("headNum", headNum)
expList = expList.Add("genesisHash", genesis)
sendList := make(protocol.KeyValueList, len(expList))
copy(sendList, expList)
expList = expList.add("serveHeaders", nil)
expList = expList.add("serveChainSince", uint64(0))
expList = expList.add("serveStateSince", uint64(0))
expList = expList.add("serveRecentState", uint64(core.TriesInMemory-4))
expList = expList.add("txRelay", nil)
expList = expList.add("flowControl/BL", testBufLimit)
expList = expList.add("flowControl/MRR", testBufRecharge)
expList = expList.add("flowControl/MRC", costList)
expList = expList.Add("serveHeaders", nil)
expList = expList.Add("serveChainSince", uint64(0))
expList = expList.Add("serveStateSince", uint64(0))
expList = expList.Add("serveRecentState", uint64(core.TriesInMemory-4))
expList = expList.Add("txRelay", nil)
expList = expList.Add("flowControl/BL", testBufLimit)
expList = expList.Add("flowControl/MRR", testBufRecharge)
expList = expList.Add("flowControl/MRC", costList)
if err := p2p.ExpectMsg(p.app, StatusMsg, expList); err != nil {
if err := p2p.ExpectMsg(p.app, protocol.StatusMsg, expList); err != nil {
t.Fatalf("status recv: %v", err)
}
if err := p2p.Send(p.app, StatusMsg, sendList); err != nil {
if err := p2p.Send(p.app, protocol.StatusMsg, sendList); err != nil {
t.Fatalf("status send: %v", err)
}
p.peer.fcParams = flowcontrol.ServerParams{

7
les/ulc_test.go
View file

@ -24,6 +24,7 @@ import (
"time"
"github.com/ethereum/go-ethereum/crypto"
"github.com/ethereum/go-ethereum/les/protocol"
"github.com/ethereum/go-ethereum/p2p"
"github.com/ethereum/go-ethereum/p2p/enode"
)
@ -86,15 +87,15 @@ func testULCAnnounceThreshold(t *testing.T, protocol int) {
}
}
func connect(server *serverHandler, serverId enode.ID, client *clientHandler, protocol int) (*peer, *peer, error) {
func connect(server *serverHandler, serverId enode.ID, client *clientHandler, p int) (*peer, *peer, error) {
// Create a message pipe to communicate through
app, net := p2p.MsgPipe()
var id enode.ID
rand.Read(id[:])
peer1 := newPeer(protocol, NetworkId, true, p2p.NewPeer(serverId, "", nil), net) // Mark server as trusted
peer2 := newPeer(protocol, NetworkId, false, p2p.NewPeer(id, "", nil), app)
peer1 := newPeer(p, protocol.NetworkId, true, p2p.NewPeer(serverId, "", nil), net) // Mark server as trusted
peer2 := newPeer(p, protocol.NetworkId, false, p2p.NewPeer(id, "", nil), app)
// Start the peerLight on a new thread
errc1 := make(chan error, 1)