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

Further LES cleanup

Browse source
This commit is contained in:
Frank Szendzielarz 2019-05-30 10:41:03 +02:00
parent a7eaf583eb
commit 3712581a00
18 changed files with 319 additions and 184 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

54
cmd/utils/flags.go
View file

@ -20,6 +20,7 @@ package utils
import (
"crypto/ecdsa"
"encoding/json"
"errors"
"fmt"
"io/ioutil"
"math/big"
@ -1470,26 +1471,51 @@ func SetDashboardConfig(ctx *cli.Context, cfg *dashboard.Config) {
cfg.Refresh = ctx.GlobalDuration(DashboardRefreshFlag.Name)
}
// RegisterEthService adds an Ethereum client to the stack.
func RegisterEthService(stack *node.Node, cfg *eth.Config) {
var err error
// RegisterLesClientService adds a Light Client service to the stack if the config permits
func RegisterLesClientService(stack *node.Node, cfg *eth.Config) {
//if the light client is requested
if cfg.SyncMode == downloader.LightSync {
err = stack.Register(func(ctx *node.ServiceContext) (node.Service, error) {
//register a light client
if err := stack.Register(func(ctx *node.ServiceContext) (node.Service, error) {
return les.New(ctx, cfg)
})
} else {
err = stack.Register(func(ctx *node.ServiceContext) (node.Service, error) {
fullNode, err := eth.New(ctx, cfg)
if fullNode != nil && cfg.LightServ > 0 {
ls, _ := les.NewLesServer(fullNode, cfg)
fullNode.AddLesServer(ls)
}); err != nil {
Fatalf("Failed to register the Light Client service: %v", err)
}
return fullNode, err
})
}
if err != nil {
}
// RegisterLesServerService adds a Light Server service to the stack if the config permits
func RegisterLesServerService(stack *node.Node, cfg *eth.Config) {
//if the light client is not requested and the light server is requested
if cfg.SyncMode != downloader.LightSync && cfg.LightServ > 0 {
//register a light server service
if err := stack.Register(func(ctx *node.ServiceContext) (node.Service, error) {
//check if the main Eth protocol service is running
var ethService *eth.Ethereum
if ctx.Service(&ethService) != nil {
return nil, errors.New("Failed to start the Les Server service because Ethereum service is not running")
}
return les.NewLesServer(ethService, cfg)
}); err != nil {
Fatalf("Failed to register the Light Server service: %v", err)
}
}
}
// RegisterEthService adds an Ethereum client to the stack if the config permits
func RegisterEthService(stack *node.Node, cfg *eth.Config) {
//if the light client is not requested
if cfg.SyncMode != downloader.LightSync {
//register a service for the Eth protocol
if err := stack.Register(func(ctx *node.ServiceContext) (node.Service, error) {
return eth.New(ctx, cfg)
}); err != nil {
Fatalf("Failed to register the Ethereum service: %v", err)
}
}
}
// RegisterDashboardService adds a dashboard to the stack.

37
eth/backend.go
View file

@ -51,14 +51,6 @@ import (
"github.com/ethereum/go-ethereum/rpc"
)
type LesServer interface {
Start(srvr *p2p.Server)
Stop()
APIs() []rpc.API
Protocols() []p2p.Protocol
SetBloomBitsIndexer(bbIndexer *core.ChainIndexer)
}
// Ethereum implements the Ethereum full node service.
type Ethereum struct {
config *Config
@ -70,7 +62,6 @@ type Ethereum struct {
txPool *core.TxPool
blockchain *core.BlockChain
protocolManager *ProtocolManager
lesServer LesServer
// DB interfaces
chainDb ethdb.Database // Block chain database
@ -94,11 +85,6 @@ type Ethereum struct {
lock sync.RWMutex // Protects the variadic fields (e.g. gas price and etherbase)
}
func (s *Ethereum) AddLesServer(ls LesServer) {
s.lesServer = ls
ls.SetBloomBitsIndexer(s.bloomIndexer)
}
// New creates a new Ethereum object (including the
// initialisation of the common Ethereum object)
func New(ctx *node.ServiceContext, config *Config) (*Ethereum, error) {
@ -256,15 +242,13 @@ func CreateConsensusEngine(ctx *node.ServiceContext, chainConfig *params.ChainCo
}
}
// APIs return the collection of RPC services the ethereum package offers.
// NOTE, some of these services probably need to be moved to somewhere else.
func (s *Ethereum) APIs() []rpc.API {
apis := ethapi.GetAPIs(s.APIBackend)
// Append any APIs exposed explicitly by the les server
if s.lesServer != nil {
apis = append(apis, s.lesServer.APIs()...)
}
// Append any APIs exposed explicitly by the consensus engine
apis = append(apis, s.engine.APIs(s.BlockChain())...)
@ -316,6 +300,11 @@ func (s *Ethereum) APIs() []rpc.API {
},
}...)
}
// GetBloomIndexer returns the bloombits indexer, primarily
// so that the Les Server service can re-use it
func (s *Ethereum) GetBloomIndexer() *core.ChainIndexer {
return s.bloomIndexer
}
func (s *Ethereum) ResetWithGenesisBlock(gb *types.Block) {
s.blockchain.ResetWithGenesisBlock(gb)
@ -485,10 +474,7 @@ func (s *Ethereum) Synced() bool { return atomic.LoadUint3
// Protocols implements node.Service, returning all the currently configured
// network protocols to start.
func (s *Ethereum) Protocols() []p2p.Protocol {
if s.lesServer == nil {
return s.protocolManager.SubProtocols
}
return append(s.protocolManager.SubProtocols, s.lesServer.Protocols()...)
}
// Start implements node.Service, starting all internal goroutines needed by the
@ -508,11 +494,10 @@ func (s *Ethereum) Start(srvr *p2p.Server) error {
}
maxPeers -= s.config.LightPeers
}
// Start the networking layer and the light server if requested
s.protocolManager.Start(maxPeers)
if s.lesServer != nil {
s.lesServer.Start(srvr)
}
return nil
}
@ -523,13 +508,9 @@ func (s *Ethereum) Stop() error {
s.blockchain.Stop()
s.engine.Close()
s.protocolManager.Stop()
if s.lesServer != nil {
s.lesServer.Stop()
}
s.txPool.Stop()
s.miner.Stop()
s.eventMux.Stop()
s.chainDb.Close()
close(s.shutdownChan)
return nil

8
go-ethereum.code-workspace Normal file
View file

@ -0,0 +1,8 @@
{
"folders": [
{
"path": "."
}
],
"settings": {}
}

3
les/api.go
View file

@ -29,8 +29,11 @@ import (
)
var (
// ErrMinCap capacity too small
ErrMinCap = errors.New("capacity too small")
// ErrTotalCap total capacity exceeded
ErrTotalCap = errors.New("total capacity exceeded")
// ErrUnknownBenchmarkType unknown benchmark type
ErrUnknownBenchmarkType = errors.New("unknown benchmark type")
dropCapacityDelay = time.Second // delay applied to decreasing capacity changes

117
les/api_backend.go
View file

@ -39,37 +39,44 @@ import (
"github.com/ethereum/go-ethereum/rpc"
)
type LesApiBackend struct {
// ClientBackend is a backend implementation for the light Client only
type ClientBackend struct {
extRPCEnabled bool
eth *LightEthereum
gpo *gasprice.Oracle
}
func (b *LesApiBackend) ChainConfig() *params.ChainConfig {
// ChainConfig returns the LightEthereum service config
func (b *ClientBackend) ChainConfig() *params.ChainConfig {
return b.eth.chainConfig
}
func (b *LesApiBackend) CurrentBlock() *types.Block {
// CurrentBlock returns the LightEthereum service chain current header block
func (b *ClientBackend) CurrentBlock() *types.Block {
return types.NewBlockWithHeader(b.eth.BlockChain().CurrentHeader())
}
func (b *LesApiBackend) SetHead(number uint64) {
// SetHead cancels the downloader and sets the head to a certain number
func (b *ClientBackend) SetHead(number uint64) {
b.eth.protocolManager.downloader.Cancel()
b.eth.blockchain.SetHead(number)
}
func (b *LesApiBackend) HeaderByNumber(ctx context.Context, blockNr rpc.BlockNumber) (*types.Header, error) {
// HeaderByNumber returns either the chain's current header or an on-demand-requested header, given a block number
func (b *ClientBackend) HeaderByNumber(ctx context.Context, blockNr rpc.BlockNumber) (*types.Header, error) {
if blockNr == rpc.LatestBlockNumber || blockNr == rpc.PendingBlockNumber {
return b.eth.blockchain.CurrentHeader(), nil
}
return b.eth.blockchain.GetHeaderByNumberOdr(ctx, uint64(blockNr))
}
func (b *LesApiBackend) HeaderByHash(ctx context.Context, hash common.Hash) (*types.Header, error) {
// HeaderByHash returns a block header for the given block hash
func (b *ClientBackend) HeaderByHash(ctx context.Context, hash common.Hash) (*types.Header, error) {
return b.eth.blockchain.GetHeaderByHash(hash), nil
}
func (b *LesApiBackend) BlockByNumber(ctx context.Context, blockNr rpc.BlockNumber) (*types.Block, error) {
// BlockByNumber identifies a header by number and then returns the block with the corresponding block hash
func (b *ClientBackend) BlockByNumber(ctx context.Context, blockNr rpc.BlockNumber) (*types.Block, error) {
header, err := b.HeaderByNumber(ctx, blockNr)
if header == nil || err != nil {
return nil, err
@ -77,7 +84,8 @@ func (b *LesApiBackend) BlockByNumber(ctx context.Context, blockNr rpc.BlockNumb
return b.GetBlock(ctx, header.Hash())
}
func (b *LesApiBackend) StateAndHeaderByNumber(ctx context.Context, blockNr rpc.BlockNumber) (*state.StateDB, *types.Header, error) {
// StateAndHeaderByNumber identifies a block header by the supplied number, then returns that header's associated state and the header itself
func (b *ClientBackend) StateAndHeaderByNumber(ctx context.Context, blockNr rpc.BlockNumber) (*state.StateDB, *types.Header, error) {
header, err := b.HeaderByNumber(ctx, blockNr)
if err != nil {
return nil, nil, err
@ -88,123 +96,157 @@ func (b *LesApiBackend) StateAndHeaderByNumber(ctx context.Context, blockNr rpc.
return light.NewState(ctx, header, b.eth.odr), header, nil
}
func (b *LesApiBackend) GetBlock(ctx context.Context, blockHash common.Hash) (*types.Block, error) {
// GetBlock returns a block identified by block hash
func (b *ClientBackend) GetBlock(ctx context.Context, blockHash common.Hash) (*types.Block, error) {
return b.eth.blockchain.GetBlockByHash(ctx, blockHash)
}
func (b *LesApiBackend) GetReceipts(ctx context.Context, hash common.Hash) (types.Receipts, error) {
// GetReceipts returns a set of receipts for a block, identified by the supplied hash
func (b *ClientBackend) GetReceipts(ctx context.Context, hash common.Hash) (types.Receipts, error) {
if number := rawdb.ReadHeaderNumber(b.eth.chainDb, hash); number != nil {
return light.GetBlockReceipts(ctx, b.eth.odr, hash, *number)
}
return nil, nil
}
func (b *LesApiBackend) GetLogs(ctx context.Context, hash common.Hash) ([][]*types.Log, error) {
// GetLogs identifies the block header number given the supplied hash and then returns the block logs
func (b *ClientBackend) GetLogs(ctx context.Context, hash common.Hash) ([][]*types.Log, error) {
if number := rawdb.ReadHeaderNumber(b.eth.chainDb, hash); number != nil {
return light.GetBlockLogs(ctx, b.eth.odr, hash, *number)
}
return nil, nil
}
func (b *LesApiBackend) GetTd(hash common.Hash) *big.Int {
// GetTd retrieves a block's total difficulty in the canonical chain by hash
func (b *ClientBackend) GetTd(hash common.Hash) *big.Int {
return b.eth.blockchain.GetTdByHash(hash)
}
func (b *LesApiBackend) GetEVM(ctx context.Context, msg core.Message, state *state.StateDB, header *types.Header) (*vm.EVM, func() error, error) {
// GetEVM creates an EVM initialised with an account and balance from the supplied message
func (b *ClientBackend) GetEVM(ctx context.Context, msg core.Message, state *state.StateDB, header *types.Header) (*vm.EVM, func() error, error) {
state.SetBalance(msg.From(), math.MaxBig256)
context := core.NewEVMContext(msg, header, b.eth.blockchain, nil)
return vm.NewEVM(context, state, b.eth.chainConfig, vm.Config{}), state.Error, nil
}
func (b *LesApiBackend) SendTx(ctx context.Context, signedTx *types.Transaction) error {
// SendTx add the transaction to the local transaction pool
func (b *ClientBackend) SendTx(ctx context.Context, signedTx *types.Transaction) error {
return b.eth.txPool.Add(ctx, signedTx)
}
func (b *LesApiBackend) RemoveTx(txHash common.Hash) {
// RemoveTx removes a transaction from the local transaction pool
func (b *ClientBackend) RemoveTx(txHash common.Hash) {
b.eth.txPool.RemoveTx(txHash)
}
func (b *LesApiBackend) GetPoolTransactions() (types.Transactions, error) {
// GetPoolTransactions returns all currently processable transactions.
func (b *ClientBackend) GetPoolTransactions() (types.Transactions, error) {
return b.eth.txPool.GetTransactions()
}
func (b *LesApiBackend) GetPoolTransaction(txHash common.Hash) *types.Transaction {
// GetPoolTransaction returns a transaction if it is contained in the pool and nil otherwise.
func (b *ClientBackend) GetPoolTransaction(txHash common.Hash) *types.Transaction {
return b.eth.txPool.GetTransaction(txHash)
}
func (b *LesApiBackend) GetTransaction(ctx context.Context, txHash common.Hash) (*types.Transaction, common.Hash, uint64, uint64, error) {
// GetTransaction retrieves a canonical transaction by hash and also returns its position in the chain
func (b *ClientBackend) GetTransaction(ctx context.Context, txHash common.Hash) (*types.Transaction, common.Hash, uint64, uint64, error) {
return light.GetTransaction(ctx, b.eth.odr, txHash)
}
func (b *LesApiBackend) GetPoolNonce(ctx context.Context, addr common.Address) (uint64, error) {
// GetPoolNonce returns the "pending" nonce of a given address. It always queries
// the nonce belonging to the latest header too in order to detect if another
// client using the same key sent a transaction.
func (b *ClientBackend) GetPoolNonce(ctx context.Context, addr common.Address) (uint64, error) {
return b.eth.txPool.GetNonce(ctx, addr)
}
func (b *LesApiBackend) Stats() (pending int, queued int) {
// Stats returns the number of currently pending (locally created) transactions
func (b *ClientBackend) Stats() (pending int, queued int) {
return b.eth.txPool.Stats(), 0
}
func (b *LesApiBackend) TxPoolContent() (map[common.Address]types.Transactions, map[common.Address]types.Transactions) {
// TxPoolContent retrieves the data content of the transaction pool, returning all the
// pending as well as queued transactions, grouped by account and nonce.
func (b *ClientBackend) TxPoolContent() (map[common.Address]types.Transactions, map[common.Address]types.Transactions) {
return b.eth.txPool.Content()
}
func (b *LesApiBackend) SubscribeNewTxsEvent(ch chan<- core.NewTxsEvent) event.Subscription {
// SubscribeNewTxsEvent registers a subscription of core.NewTxsEvent and
// starts sending event to the given channel.
func (b *ClientBackend) SubscribeNewTxsEvent(ch chan<- core.NewTxsEvent) event.Subscription {
return b.eth.txPool.SubscribeNewTxsEvent(ch)
}
func (b *LesApiBackend) SubscribeChainEvent(ch chan<- core.ChainEvent) event.Subscription {
// SubscribeChainEvent registers a subscription of ChainEvent.
func (b *ClientBackend) SubscribeChainEvent(ch chan<- core.ChainEvent) event.Subscription {
return b.eth.blockchain.SubscribeChainEvent(ch)
}
func (b *LesApiBackend) SubscribeChainHeadEvent(ch chan<- core.ChainHeadEvent) event.Subscription {
// SubscribeChainHeadEvent registers a subscription of ChainHeadEvent.
func (b *ClientBackend) SubscribeChainHeadEvent(ch chan<- core.ChainHeadEvent) event.Subscription {
return b.eth.blockchain.SubscribeChainHeadEvent(ch)
}
func (b *LesApiBackend) SubscribeChainSideEvent(ch chan<- core.ChainSideEvent) event.Subscription {
// SubscribeChainSideEvent registers a subscription of ChainSideEvent.
func (b *ClientBackend) SubscribeChainSideEvent(ch chan<- core.ChainSideEvent) event.Subscription {
return b.eth.blockchain.SubscribeChainSideEvent(ch)
}
func (b *LesApiBackend) SubscribeLogsEvent(ch chan<- []*types.Log) event.Subscription {
// SubscribeLogsEvent implements the interface of filters.Backend
// LightChain does not send logs events, so return an empty subscription.
func (b *ClientBackend) SubscribeLogsEvent(ch chan<- []*types.Log) event.Subscription {
return b.eth.blockchain.SubscribeLogsEvent(ch)
}
func (b *LesApiBackend) SubscribeRemovedLogsEvent(ch chan<- core.RemovedLogsEvent) event.Subscription {
// SubscribeRemovedLogsEvent implements the interface of filters.Backend
// LightChain does not send core.RemovedLogsEvent, so return an empty subscription.
func (b *ClientBackend) SubscribeRemovedLogsEvent(ch chan<- core.RemovedLogsEvent) event.Subscription {
return b.eth.blockchain.SubscribeRemovedLogsEvent(ch)
}
func (b *LesApiBackend) Downloader() *downloader.Downloader {
// Downloader returns the LightEthereum downloader
func (b *ClientBackend) Downloader() *downloader.Downloader {
return b.eth.Downloader()
}
func (b *LesApiBackend) ProtocolVersion() int {
// ProtocolVersion return the Les version number offset by 10000
func (b *ClientBackend) ProtocolVersion() int {
return b.eth.LesVersion() + 10000
}
func (b *LesApiBackend) SuggestPrice(ctx context.Context) (*big.Int, error) {
// SuggestPrice returns the gas price oracle price suggestion
func (b *ClientBackend) SuggestPrice(ctx context.Context) (*big.Int, error) {
return b.gpo.SuggestPrice(ctx)
}
func (b *LesApiBackend) ChainDb() ethdb.Database {
// ChainDb returns the ClientBackend db representation
func (b *ClientBackend) ChainDb() ethdb.Database {
return b.eth.chainDb
}
func (b *LesApiBackend) EventMux() *event.TypeMux {
// EventMux ...
func (b *ClientBackend) EventMux() *event.TypeMux {
return b.eth.eventMux
}
func (b *LesApiBackend) AccountManager() *accounts.Manager {
// AccountManager returns an overarching account manager needed for signing transactions
func (b *ClientBackend) AccountManager() *accounts.Manager {
return b.eth.accountManager
}
func (b *LesApiBackend) ExtRPCEnabled() bool {
// ExtRPCEnabled indicates if the external RPC API is enabled
func (b *ClientBackend) ExtRPCEnabled() bool {
return b.extRPCEnabled
}
func (b *LesApiBackend) RPCGasCap() *big.Int {
// RPCGasCap - return the global gas cap for eth_call over rpc (DoS protection)
func (b *ClientBackend) RPCGasCap() *big.Int {
return b.eth.config.RPCGasCap
}
func (b *LesApiBackend) BloomStatus() (uint64, uint64) {
// BloomStatus ..
func (b *ClientBackend) BloomStatus() (uint64, uint64) {
if b.eth.bloomIndexer == nil {
return 0, 0
}
@ -212,7 +254,8 @@ func (b *LesApiBackend) BloomStatus() (uint64, uint64) {
return params.BloomBitsBlocksClient, sections
}
func (b *LesApiBackend) ServiceFilter(ctx context.Context, session *bloombits.MatcherSession) {
// ServiceFilter ...
func (b *ClientBackend) ServiceFilter(ctx context.Context, session *bloombits.MatcherSession) {
for i := 0; i < bloomFilterThreads; i++ {
go session.Multiplex(bloomRetrievalBatch, bloomRetrievalWait, b.eth.bloomRequests)
}

40
les/api_test.go
View file

@ -90,15 +90,15 @@ func testCapacityAPI(t *testing.T, clientCount int) {
}
server := servers[0]
clientRpcClients := make([]*rpc.Client, len(clients))
clientRPCClients := make([]*rpc.Client, len(clients))
serverRpcClient, err := server.Client()
serverRPCClient, err := server.Client()
if err != nil {
t.Fatalf("Failed to obtain rpc client: %v", err)
}
headNum, headHash := getHead(ctx, t, serverRpcClient)
totalCap := getTotalCap(ctx, t, serverRpcClient)
minCap := getMinCap(ctx, t, serverRpcClient)
headNum, headHash := getHead(ctx, t, serverRPCClient)
totalCap := getTotalCap(ctx, t, serverRPCClient)
minCap := getMinCap(ctx, t, serverRPCClient)
testCap := totalCap * 3 / 4
fmt.Printf("Server testCap: %d minCap: %d head number: %d head hash: %064x\n", testCap, minCap, headNum, headHash)
reqMinCap := uint64(float64(testCap) * minRelCap / (minRelCap + float64(len(clients)-1)))
@ -107,18 +107,18 @@ func testCapacityAPI(t *testing.T, clientCount int) {
}
freeIdx := rand.Intn(len(clients))
freeCap := getFreeCap(ctx, t, serverRpcClient)
freeCap := getFreeCap(ctx, t, serverRPCClient)
for i, client := range clients {
var err error
clientRpcClients[i], err = client.Client()
clientRPCClients[i], err = client.Client()
if err != nil {
t.Fatalf("Failed to obtain rpc client: %v", err)
}
fmt.Println("connecting client", i)
if i != freeIdx {
setCapacity(ctx, t, serverRpcClient, client.ID(), testCap/uint64(len(clients)))
setCapacity(ctx, t, serverRPCClient, client.ID(), testCap/uint64(len(clients)))
}
net.Connect(client.ID(), server.ID())
@ -128,7 +128,7 @@ func testCapacityAPI(t *testing.T, clientCount int) {
t.Fatalf("Timeout")
default:
}
num, hash := getHead(ctx, t, clientRpcClients[i])
num, hash := getHead(ctx, t, clientRPCClients[i])
if num == headNum && hash == headHash {
fmt.Println("client", i, "synced")
break
@ -140,9 +140,9 @@ func testCapacityAPI(t *testing.T, clientCount int) {
var wg sync.WaitGroup
stop := make(chan struct{})
reqCount := make([]uint64, len(clientRpcClients))
reqCount := make([]uint64, len(clientRPCClients))
for i, c := range clientRpcClients {
for i, c := range clientRPCClients {
wg.Add(1)
i, c := i, c
go func() {
@ -194,7 +194,7 @@ func testCapacityAPI(t *testing.T, clientCount int) {
weights := make([]float64, len(clients))
for c := 0; c < 5; c++ {
setCapacity(ctx, t, serverRpcClient, clients[freeIdx].ID(), freeCap)
setCapacity(ctx, t, serverRPCClient, clients[freeIdx].ID(), freeCap)
freeIdx = rand.Intn(len(clients))
var sum float64
for i := range clients {
@ -208,15 +208,15 @@ func testCapacityAPI(t *testing.T, clientCount int) {
for i, client := range clients {
weights[i] *= float64(testCap-freeCap-100) / sum
capacity := uint64(weights[i])
if i != freeIdx && capacity < getCapacity(ctx, t, serverRpcClient, client.ID()) {
setCapacity(ctx, t, serverRpcClient, client.ID(), capacity)
if i != freeIdx && capacity < getCapacity(ctx, t, serverRPCClient, client.ID()) {
setCapacity(ctx, t, serverRPCClient, client.ID(), capacity)
}
}
setCapacity(ctx, t, serverRpcClient, clients[freeIdx].ID(), 0)
setCapacity(ctx, t, serverRPCClient, clients[freeIdx].ID(), 0)
for i, client := range clients {
capacity := uint64(weights[i])
if i != freeIdx && capacity > getCapacity(ctx, t, serverRpcClient, client.ID()) {
setCapacity(ctx, t, serverRpcClient, client.ID(), capacity)
if i != freeIdx && capacity > getCapacity(ctx, t, serverRPCClient, client.ID()) {
setCapacity(ctx, t, serverRPCClient, client.ID(), capacity)
}
}
weights[freeIdx] = float64(freeCap)
@ -239,7 +239,7 @@ func testCapacityAPI(t *testing.T, clientCount int) {
default:
}
totalCap = getTotalCap(ctx, t, serverRpcClient)
totalCap = getTotalCap(ctx, t, serverRPCClient)
if totalCap < testCap {
fmt.Println("Total capacity underrun")
close(stop)
@ -520,6 +520,6 @@ func newLesServerService(ctx *adapters.ServiceContext) (node.Service, error) {
if err != nil {
return nil, err
}
ethereum.AddLesServer(server)
return ethereum, nil
return server, nil
}

32
les/backend.go
View file

@ -46,6 +46,7 @@ import (
rpc "github.com/ethereum/go-ethereum/rpc"
)
// LightEthereum is a light client service
type LightEthereum struct {
lesCommons
@ -66,18 +67,19 @@ type LightEthereum struct {
bloomRequests chan chan *bloombits.Retrieval // Channel receiving bloom data retrieval requests
bloomIndexer *core.ChainIndexer
ApiBackend *LesApiBackend
APIBackend *ClientBackend
eventMux *event.TypeMux
engine consensus.Engine
accountManager *accounts.Manager
networkId uint64
networkID uint64
netRPCService *ethapi.PublicNetAPI
wg sync.WaitGroup
}
// New constructs a LightEthereum client
func New(ctx *node.ServiceContext, config *eth.Config) (*LightEthereum, error) {
chainDb, err := ctx.OpenDatabase("lightchaindata", config.DatabaseCache, config.DatabaseHandles, "eth/db/chaindata/")
if err != nil {
@ -105,7 +107,7 @@ func New(ctx *node.ServiceContext, config *eth.Config) (*LightEthereum, error) {
accountManager: ctx.AccountManager,
engine: eth.CreateConsensusEngine(ctx, chainConfig, &config.Ethash, nil, false, chainDb),
shutdownChan: make(chan bool),
networkId: config.NetworkId,
networkID: config.NetworkId,
bloomRequests: make(chan chan *bloombits.Retrieval),
bloomIndexer: eth.NewBloomIndexer(chainDb, params.BloomBitsBlocksClient, params.HelperTrieConfirmations),
}
@ -167,13 +169,13 @@ func New(ctx *node.ServiceContext, config *eth.Config) (*LightEthereum, error) {
log.Warn("Ultra light client is enabled", "trustedNodes", len(leth.protocolManager.ulc.trustedKeys), "minTrustedFraction", leth.protocolManager.ulc.minTrustedFraction)
leth.blockchain.DisableCheckFreq()
}
leth.ApiBackend = &LesApiBackend{ctx.ExtRPCEnabled(), leth, nil}
leth.APIBackend = &ClientBackend{ctx.ExtRPCEnabled(), leth, nil}
gpoParams := config.GPO
if gpoParams.Default == nil {
gpoParams.Default = config.Miner.GasPrice
}
leth.ApiBackend.gpo = gasprice.NewOracle(leth.ApiBackend, gpoParams)
leth.APIBackend.gpo = gasprice.NewOracle(leth.APIBackend, gpoParams)
return leth, nil
}
@ -188,6 +190,7 @@ func lesTopic(genesisHash common.Hash, protocolVersion uint) discv5.Topic {
return discv5.Topic(name + "@" + common.Bytes2Hex(genesisHash.Bytes()[0:8]))
}
// LightDummyAPI is a fake api that serves default values in the role of a light client
type LightDummyAPI struct{}
// Etherbase is the address that mining rewards will be send to
@ -213,7 +216,7 @@ func (s *LightDummyAPI) Mining() bool {
// APIs returns the collection of RPC services the ethereum package offers.
// NOTE, some of these services probably need to be moved to somewhere else.
func (s *LightEthereum) APIs() []rpc.API {
return append(ethapi.GetAPIs(s.ApiBackend), []rpc.API{
return append(ethapi.GetAPIs(s.APIBackend), []rpc.API{
{
Namespace: "eth",
Version: "1.0",
@ -227,7 +230,7 @@ func (s *LightEthereum) APIs() []rpc.API {
}, {
Namespace: "eth",
Version: "1.0",
Service: filters.NewPublicFilterAPI(s.ApiBackend, true),
Service: filters.NewPublicFilterAPI(s.APIBackend, true),
Public: true,
}, {
Namespace: "net",
@ -238,15 +241,28 @@ func (s *LightEthereum) APIs() []rpc.API {
}...)
}
// ResetWithGenesisBlock purges the entire blockchain, restoring it to the
// specified genesis state.
func (s *LightEthereum) ResetWithGenesisBlock(gb *types.Block) {
s.blockchain.ResetWithGenesisBlock(gb)
}
// BlockChain returns the light client's blockchain
func (s *LightEthereum) BlockChain() *light.LightChain { return s.blockchain }
// TxPool the pool of locally created transactions
func (s *LightEthereum) TxPool() *light.TxPool { return s.txPool }
// Engine is the employed consensus engine
func (s *LightEthereum) Engine() consensus.Engine { return s.engine }
// LesVersion is the current version of Les used by this light client
func (s *LightEthereum) LesVersion() int { return int(ClientProtocolVersions[0]) }
// Downloader is the employed downloader
func (s *LightEthereum) Downloader() *downloader.Downloader { return s.protocolManager.downloader }
// EventMux is an event subscription notifier
func (s *LightEthereum) EventMux() *event.TypeMux { return s.eventMux }
// Protocols implements node.Service, returning all the currently configured
@ -260,7 +276,7 @@ func (s *LightEthereum) Protocols() []p2p.Protocol {
func (s *LightEthereum) Start(srvr *p2p.Server) error {
log.Warn("Light client mode is an experimental feature")
s.startBloomHandlers(params.BloomBitsBlocksClient)
s.netRPCService = ethapi.NewPublicNetAPI(srvr, s.networkId)
s.netRPCService = ethapi.NewPublicNetAPI(srvr, s.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))

16
les/benchmark.go
View file

@ -74,9 +74,9 @@ func (b *benchmarkBlockHeaders) init(pm *ProtocolManager, count int) error {
func (b *benchmarkBlockHeaders) request(peer *peer, index int) error {
if b.byHash {
return peer.RequestHeadersByHash(0, 0, b.hashes[index], b.amount, b.skip, b.reverse)
} else {
return peer.RequestHeadersByNumber(0, 0, uint64(b.offset+rand.Int63n(b.randMax)), b.amount, b.skip, b.reverse)
}
return peer.RequestHeadersByNumber(0, 0, uint64(b.offset+rand.Int63n(b.randMax)), b.amount, b.skip, b.reverse)
}
// benchmarkBodiesOrReceipts implements requestBenchmark
@ -97,9 +97,9 @@ func (b *benchmarkBodiesOrReceipts) init(pm *ProtocolManager, count int) error {
func (b *benchmarkBodiesOrReceipts) request(peer *peer, index int) error {
if b.receipts {
return peer.RequestReceipts(0, 0, []common.Hash{b.hashes[index]})
} else {
return peer.RequestBodies(0, 0, []common.Hash{b.hashes[index]})
}
return peer.RequestBodies(0, 0, []common.Hash{b.hashes[index]})
}
// benchmarkProofsOrCode implements requestBenchmark
@ -118,9 +118,9 @@ func (b *benchmarkProofsOrCode) request(peer *peer, index int) error {
rand.Read(key)
if b.code {
return peer.RequestCode(0, 0, []CodeReq{{BHash: b.headHash, AccKey: key}})
} else {
return peer.RequestProofs(0, 0, []ProofReq{{BHash: b.headHash, Key: key}})
}
return peer.RequestProofs(0, 0, []ProofReq{{BHash: b.headHash, Key: key}})
}
// benchmarkHelperTrie implements requestBenchmark
@ -281,8 +281,8 @@ func (pm *ProtocolManager) measure(setup *benchmarkSetup, count int) error {
serverMeteredPipe := &meteredPipe{rw: serverPipe}
var id enode.ID
rand.Read(id[:])
clientPeer := pm.newPeer(lpv2, NetworkId, p2p.NewPeer(id, "client", nil), clientMeteredPipe)
serverPeer := pm.newPeer(lpv2, NetworkId, p2p.NewPeer(id, "server", nil), serverMeteredPipe)
clientPeer := pm.newPeer(lpv2, NetworkID, p2p.NewPeer(id, "client", nil), clientMeteredPipe)
serverPeer := pm.newPeer(lpv2, NetworkID, p2p.NewPeer(id, "server", nil), serverMeteredPipe)
serverPeer.sendQueue = newExecQueue(count)
serverPeer.announceType = announceTypeNone
serverPeer.fcCosts = make(requestCostTable)

10
les/freeclient_test.go
View file

@ -51,7 +51,7 @@ func testFreeClientPool(t *testing.T, connLimit, clientCount int) {
peerAddress = func(i int) string {
return fmt.Sprintf("addr #%d", i)
}
peerId = func(i int) string {
peerID = func(i int) string {
return fmt.Sprintf("id #%d", i)
}
disconnFn = func(id string) {
@ -67,14 +67,14 @@ func testFreeClientPool(t *testing.T, connLimit, clientCount int) {
// pool should accept new peers up to its connected limit
for i := 0; i < connLimit; i++ {
if pool.connect(peerAddress(i), peerId(i)) {
if pool.connect(peerAddress(i), peerID(i)) {
connected[i] = true
} else {
t.Fatalf("Test peer #%d rejected", i)
}
}
// since all accepted peers are new and should not be kicked out, the next one should be rejected
if pool.connect(peerAddress(connLimit), peerId(connLimit)) {
if pool.connect(peerAddress(connLimit), peerID(connLimit)) {
connected[connLimit] = true
t.Fatalf("Peer accepted over connected limit")
}
@ -89,7 +89,7 @@ func testFreeClientPool(t *testing.T, connLimit, clientCount int) {
connected[i] = false
connTicks[i] += tickCounter
} else {
if pool.connect(peerAddress(i), peerId(i)) {
if pool.connect(peerAddress(i), peerID(i)) {
connected[i] = true
connTicks[i] -= tickCounter
}
@ -135,7 +135,7 @@ func testFreeClientPool(t *testing.T, connLimit, clientCount int) {
// try connecting all known peers (connLimit should be filled up)
for i := 0; i < clientCount; i++ {
pool.connect(peerAddress(i), peerId(i))
pool.connect(peerAddress(i), peerID(i))
}
// expect pool to remember known nodes and kick out one of them to accept a new one
if !pool.connect("newAddr2", "newId2") {

36
les/handler.go
View file

@ -49,14 +49,22 @@ const (
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 merkle proofs 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
// MaxHeaderFetch - Amount of block headers to be fetched per retrieval request
MaxHeaderFetch = 192
// MaxBodyFetch - Amount of block bodies to be fetched per retrieval request
MaxBodyFetch = 32
// MaxReceiptFetch - Amount of transaction receipts to allow fetching per request
MaxReceiptFetch = 128
// MaxCodeFetch - Amount of contract codes to allow fetching per request
MaxCodeFetch = 64
// MaxProofsFetch - Amount of merkle proofs to be fetched per retrieval request
MaxProofsFetch = 64
// MaxHelperTrieProofsFetch - Amount of merkle proofs to be fetched per retrieval request
MaxHelperTrieProofsFetch = 64
// MaxTxSend - Amount of transactions to be send per request
MaxTxSend = 64
// MaxTxStatus - Amount of transactions to queried per request
MaxTxStatus = 256
disableClientRemovePeer = false
)
@ -65,6 +73,7 @@ func errResp(code errCode, format string, v ...interface{}) error {
return fmt.Errorf("%v - %v", code, fmt.Sprintf(format, v...))
}
// BlockChain - a blockchain
type BlockChain interface {
Config() *params.ChainConfig
HasHeader(hash common.Hash, number uint64) bool
@ -86,6 +95,7 @@ type txPool interface {
Status(hashes []common.Hash) []core.TxStatus
}
// ProtocolManager handles protocol messages and other protocol-specific tasks
type ProtocolManager struct {
// Configs
chainConfig *params.ChainConfig
@ -93,7 +103,7 @@ type ProtocolManager struct {
client bool // The indicator whether the node is light client
maxPeers int // The maximum number peers allowed to connect.
networkId uint64 // The identity of network.
networkID uint64 // The identity of network.
txpool txPool
txrelay *LesTxRelay
@ -129,7 +139,7 @@ func NewProtocolManager(
chainConfig *params.ChainConfig,
indexerConfig *light.IndexerConfig,
client bool,
networkId uint64,
networkID uint64,
mux *event.TypeMux,
engine consensus.Engine,
peers *peerSet,
@ -151,7 +161,7 @@ func NewProtocolManager(
iConfig: indexerConfig,
chainDb: chainDb,
odr: odr,
networkId: networkId,
networkID: networkID,
txpool: txpool,
txrelay: txrelay,
serverPool: serverPool,
@ -194,6 +204,7 @@ func (pm *ProtocolManager) removePeer(id string) {
pm.peers.Unregister(id)
}
// Start starts the protocol manager and is called during node startup only
func (pm *ProtocolManager) Start(maxPeers int) {
pm.maxPeers = maxPeers
if pm.client {
@ -206,6 +217,7 @@ func (pm *ProtocolManager) Start(maxPeers int) {
}
}
// Stop stops the protocol manager and is called during node shutdown only
func (pm *ProtocolManager) Stop() {
// Showing a log message. During download / process this could actually
// take between 5 to 10 seconds and therefor feedback is required.
@ -236,7 +248,7 @@ func (pm *ProtocolManager) Stop() {
// runPeer is the p2p protocol run function for the given version.
func (pm *ProtocolManager) runPeer(version uint, p *p2p.Peer, rw p2p.MsgReadWriter) error {
var entry *poolEntry
peer := pm.newPeer(int(version), pm.networkId, p, rw)
peer := pm.newPeer(int(version), pm.networkID, p, rw)
if pm.serverPool != nil {
entry = pm.serverPool.connect(peer, peer.Node())
}

10
les/helper_test.go
View file

@ -170,7 +170,7 @@ func newTestProtocolManager(lightSync bool, blocks int, generator func(int, *cor
if lightSync {
indexConfig = light.TestClientIndexerConfig
}
pm, err := NewProtocolManager(gspec.Config, indexConfig, lightSync, NetworkId, evmux, engine, peers, chain, pool, db, odr, nil, nil, make(chan struct{}), new(sync.WaitGroup), ulcConfig, func() bool { return true })
pm, err := NewProtocolManager(gspec.Config, indexConfig, lightSync, NetworkID, evmux, engine, peers, chain, pool, db, odr, nil, nil, make(chan struct{}), new(sync.WaitGroup), ulcConfig, func() bool { return true })
if err != nil {
return nil, err
}
@ -218,7 +218,7 @@ func newTestPeer(t *testing.T, name string, version int, pm *ProtocolManager, sh
var id enode.ID
rand.Read(id[:])
peer := pm.newPeer(version, NetworkId, p2p.NewPeer(id, name, nil), net)
peer := pm.newPeer(version, NetworkID, p2p.NewPeer(id, name, nil), net)
// Start the peer on a new thread
errc := make(chan error, 1)
@ -255,8 +255,8 @@ func newTestPeerPair(name string, version int, pm, pm2 *ProtocolManager) (*peer,
var id enode.ID
rand.Read(id[:])
peer := pm.newPeer(version, NetworkId, p2p.NewPeer(id, name, nil), net)
peer2 := pm2.newPeer(version, NetworkId, p2p.NewPeer(id, name, nil), app)
peer := pm.newPeer(version, NetworkID, p2p.NewPeer(id, name, nil), net)
peer2 := pm2.newPeer(version, NetworkID, p2p.NewPeer(id, name, nil), app)
// Start the peer on a new thread
errc := make(chan error, 1)
@ -285,7 +285,7 @@ func newTestPeerPair(name string, version int, pm, pm2 *ProtocolManager) (*peer,
func (p *testPeer) handshake(t *testing.T, td *big.Int, head common.Hash, headNum uint64, genesis common.Hash) {
var expList keyValueList
expList = expList.add("protocolVersion", uint64(p.version))
expList = expList.add("networkId", uint64(NetworkId))
expList = expList.add("networkId", uint64(NetworkID))
expList = expList.add("headTd", td)
expList = expList.add("headHash", head)
expList = expList.add("headNum", headNum)

7
les/odr.go
View file

@ -34,6 +34,7 @@ type LesOdr struct {
stop chan struct{}
}
// NewLesOdr - construct a new LES on demand request
func NewLesOdr(db ethdb.Database, config *light.IndexerConfig, retriever *retrieveManager) *LesOdr {
return &LesOdr{
db: db,
@ -81,11 +82,17 @@ func (odr *LesOdr) IndexerConfig() *light.IndexerConfig {
}
const (
// MsgBlockBodies - block bodies request
MsgBlockBodies = iota
// MsgCode - code request
MsgCode
// MsgReceipts - receipts request
MsgReceipts
// MsgProofsV2 - proofs request
MsgProofsV2
// MsgHelperTrieProofs - helper trie proofs request
MsgHelperTrieProofs
// MsgTxStatus - transaction status request
MsgTxStatus
)

29
les/odr_requests.go
View file

@ -45,6 +45,7 @@ var (
errUselessNodes = errors.New("useless nodes in merkle proof nodeset")
)
// LesOdrRequest - an on-demand request
type LesOdrRequest interface {
GetCost(*peer) uint64
CanSend(*peer) bool
@ -52,6 +53,7 @@ type LesOdrRequest interface {
Validate(ethdb.Database, *Msg) error
}
// LesRequest - generate a LES on demand request from a light one
func LesRequest(req light.OdrRequest) LesOdrRequest {
switch r := req.(type) {
case *light.BlockRequest:
@ -93,7 +95,7 @@ func (r *BlockRequest) Request(reqID uint64, peer *peer) error {
return peer.RequestBodies(reqID, r.GetCost(peer), []common.Hash{r.Hash})
}
// Valid processes an ODR request reply message from the LES network
// Validate processes an ODR request reply message from the LES network
// returns true and stores results in memory if the message was a valid reply
// to the request (implementation of LesOdrRequest)
func (r *BlockRequest) Validate(db ethdb.Database, msg *Msg) error {
@ -149,7 +151,7 @@ func (r *ReceiptsRequest) Request(reqID uint64, peer *peer) error {
return peer.RequestReceipts(reqID, r.GetCost(peer), []common.Hash{r.Hash})
}
// Valid processes an ODR request reply message from the LES network
// Validate processes an ODR request reply message from the LES network
// returns true and stores results in memory if the message was a valid reply
// to the request (implementation of LesOdrRequest)
func (r *ReceiptsRequest) Validate(db ethdb.Database, msg *Msg) error {
@ -178,13 +180,14 @@ func (r *ReceiptsRequest) Validate(db ethdb.Database, msg *Msg) error {
return nil
}
// ProofReq - a request for a proof
type ProofReq struct {
BHash common.Hash
AccKey, Key []byte
FromLevel uint
}
// ODR request type for state/storage trie entries, see LesOdrRequest interface
// TrieRequest - 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
@ -209,7 +212,7 @@ func (r *TrieRequest) Request(reqID uint64, peer *peer) error {
return peer.RequestProofs(reqID, r.GetCost(peer), []ProofReq{req})
}
// Valid processes an ODR request reply message from the LES network
// Validate processes an ODR request reply message from the LES network
// returns true and stores results in memory if the message was a valid reply
// to the request (implementation of LesOdrRequest)
func (r *TrieRequest) Validate(db ethdb.Database, msg *Msg) error {
@ -233,12 +236,13 @@ func (r *TrieRequest) Validate(db ethdb.Database, msg *Msg) error {
return nil
}
// CodeReq - a request for contract code
type CodeReq struct {
BHash common.Hash
AccKey []byte
}
// ODR request type for node data (used for retrieving contract code), see LesOdrRequest interface
// CodeRequest - 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
@ -262,7 +266,7 @@ func (r *CodeRequest) Request(reqID uint64, peer *peer) error {
return peer.RequestCode(reqID, r.GetCost(peer), []CodeReq{req})
}
// Valid processes an ODR request reply message from the LES network
// Validate processes an ODR request reply message from the LES network
// returns true and stores results in memory if the message was a valid reply
// to the request (implementation of LesOdrRequest)
func (r *CodeRequest) Validate(db ethdb.Database, msg *Msg) error {
@ -297,6 +301,7 @@ const (
auxHeader = 2
)
// HelperTrieReq - a request for part of a trie
type HelperTrieReq struct {
Type uint
TrieIdx uint64
@ -304,12 +309,13 @@ type HelperTrieReq struct {
FromLevel, AuxReq uint
}
// HelperTrieResps - response to HelperTrieReq, providing merkle proofs
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
// ChtRequest - 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
@ -340,7 +346,7 @@ func (r *ChtRequest) Request(reqID uint64, peer *peer) error {
return peer.RequestHelperTrieProofs(reqID, r.GetCost(peer), []HelperTrieReq{req})
}
// Valid processes an ODR request reply message from the LES network
// Validate processes an ODR request reply message from the LES network
// returns true and stores results in memory if the message was a valid reply
// to the request (implementation of LesOdrRequest)
func (r *ChtRequest) Validate(db ethdb.Database, msg *Msg) error {
@ -393,11 +399,12 @@ func (r *ChtRequest) Validate(db ethdb.Database, msg *Msg) error {
return nil
}
// BloomReq -
type BloomReq struct {
BloomTrieNum, BitIdx, SectionIndex, FromLevel uint64
}
// ODR request type for requesting headers by Canonical Hash Trie, see LesOdrRequest interface
// BloomRequest - 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
@ -436,7 +443,7 @@ func (r *BloomRequest) Request(reqID uint64, peer *peer) error {
return peer.RequestHelperTrieProofs(reqID, r.GetCost(peer), reqs)
}
// Valid processes an ODR request reply message from the LES network
// Validate processes an ODR request reply message from the LES network
// returns true and stores results in memory if the message was a valid reply
// to the request (implementation of LesOdrRequest)
func (r *BloomRequest) Validate(db ethdb.Database, msg *Msg) error {
@ -493,7 +500,7 @@ func (r *TxStatusRequest) Request(reqID uint64, peer *peer) error {
return peer.RequestTxStatus(reqID, r.GetCost(peer), r.Hashes)
}
// Valid processes an ODR request reply message from the LES network
// Validate processes an ODR request reply message from the LES network
// returns true and stores results in memory if the message was a valid reply
// to the request (implementation of LesOdrRequest)
func (r *TxStatusRequest) Validate(db ethdb.Database, msg *Msg) error {

2
les/peer.go
View file

@ -526,7 +526,7 @@ func (p *peer) Handshake(td *big.Int, head common.Hash, headNum uint64, genesis
return errResp(ErrGenesisBlockMismatch, "%x (!= %x)", rGenesis[:8], genesis[:8])
}
if rNetwork != p.network {
return errResp(ErrNetworkIdMismatch, "%d (!= %d)", rNetwork, p.network)
return errResp(ErrNetworkIDMismatch, "%d (!= %d)", rNetwork, p.network)
}
if int(rVersion) != p.version {
return errResp(ErrProtocolVersionMismatch, "%d (!= %d)", rVersion, p.version)

32
les/peer_test.go
View file

@ -12,8 +12,8 @@ import (
)
const (
test_networkid = 10
protocol_version = lpv2
testNetworkID = 10
protocolVersion = lpv2
)
var (
@ -30,7 +30,7 @@ func TestPeerHandshakeSetAnnounceTypeToAnnounceTypeSignedForTrustedPeer(t *testi
//peer to connect(on ulc side)
p := peer{
Peer: p2p.NewPeer(id, "test peer", []p2p.Cap{}),
version: protocol_version,
version: protocolVersion,
isTrusted: true,
rw: &rwStub{
WriteHook: func(recvList keyValueList) {
@ -59,7 +59,7 @@ func TestPeerHandshakeSetAnnounceTypeToAnnounceTypeSignedForTrustedPeer(t *testi
return l
},
},
network: test_networkid,
network: testNetworkID,
}
err := p.Handshake(td, hash, headNum, genesis, nil)
@ -76,7 +76,7 @@ func TestPeerHandshakeAnnounceTypeSignedForTrustedPeersPeerNotInTrusted(t *testi
id := newNodeID(t).ID()
p := peer{
Peer: p2p.NewPeer(id, "test peer", []p2p.Cap{}),
version: protocol_version,
version: protocolVersion,
rw: &rwStub{
WriteHook: func(recvList keyValueList) {
//checking that ulc sends to peer allowedRequests=noRequests and announceType != announceTypeSigned
@ -104,7 +104,7 @@ func TestPeerHandshakeAnnounceTypeSignedForTrustedPeersPeerNotInTrusted(t *testi
return l
},
},
network: test_networkid,
network: testNetworkID,
}
err := p.Handshake(td, hash, headNum, genesis, nil)
@ -123,7 +123,7 @@ func TestPeerHandshakeDefaultAllRequests(t *testing.T) {
p := peer{
Peer: p2p.NewPeer(id, "test peer", []p2p.Cap{}),
version: protocol_version,
version: protocolVersion,
rw: &rwStub{
ReadHook: func(l keyValueList) keyValueList {
l = l.add("announceType", uint64(announceTypeSigned))
@ -132,7 +132,7 @@ func TestPeerHandshakeDefaultAllRequests(t *testing.T) {
return l
},
},
network: test_networkid,
network: testNetworkID,
}
err := p.Handshake(td, hash, headNum, genesis, s)
@ -153,7 +153,7 @@ func TestPeerHandshakeServerSendOnlyAnnounceRequestsHeaders(t *testing.T) {
p := peer{
Peer: p2p.NewPeer(id, "test peer", []p2p.Cap{}),
version: protocol_version,
version: protocolVersion,
rw: &rwStub{
ReadHook: func(l keyValueList) keyValueList {
l = l.add("announceType", uint64(announceTypeSigned))
@ -171,7 +171,7 @@ func TestPeerHandshakeServerSendOnlyAnnounceRequestsHeaders(t *testing.T) {
}
},
},
network: test_networkid,
network: testNetworkID,
}
err := p.Handshake(td, hash, headNum, genesis, s)
@ -184,7 +184,7 @@ func TestPeerHandshakeClientReceiveOnlyAnnounceRequestsHeaders(t *testing.T) {
p := peer{
Peer: p2p.NewPeer(id, "test peer", []p2p.Cap{}),
version: protocol_version,
version: protocolVersion,
rw: &rwStub{
ReadHook: func(l keyValueList) keyValueList {
l = l.add("flowControl/BL", uint64(0))
@ -196,7 +196,7 @@ func TestPeerHandshakeClientReceiveOnlyAnnounceRequestsHeaders(t *testing.T) {
return l
},
},
network: test_networkid,
network: testNetworkID,
isTrusted: true,
}
@ -215,7 +215,7 @@ func TestPeerHandshakeClientReturnErrorOnUselessPeer(t *testing.T) {
p := peer{
Peer: p2p.NewPeer(id, "test peer", []p2p.Cap{}),
version: protocol_version,
version: protocolVersion,
rw: &rwStub{
ReadHook: func(l keyValueList) keyValueList {
l = l.add("flowControl/BL", uint64(0))
@ -227,7 +227,7 @@ func TestPeerHandshakeClientReturnErrorOnUselessPeer(t *testing.T) {
return l
},
},
network: test_networkid,
network: testNetworkID,
}
err := p.Handshake(td, hash, headNum, genesis, nil)
@ -254,8 +254,8 @@ type rwStub struct {
func (s *rwStub) ReadMsg() (p2p.Msg, error) {
payload := keyValueList{}
payload = payload.add("protocolVersion", uint64(protocol_version))
payload = payload.add("networkId", uint64(test_networkid))
payload = payload.add("protocolVersion", uint64(protocolVersion))
payload = payload.add("networkId", uint64(testNetworkID))
payload = payload.add("headTd", td)
payload = payload.add("headHash", hash)
payload = payload.add("headNum", headNum)

27
les/protocol.go
View file

@ -41,12 +41,14 @@ var (
AdvertiseProtocolVersions = []uint{lpv2} // clients are searching for the first advertised protocol in the list
)
// Number of implemented message corresponding to different protocol versions.
// ProtocolLengths - Number of implemented message corresponding to different protocol versions.
var ProtocolLengths = map[uint]uint64{lpv2: 22}
const (
NetworkId = 1
ProtocolMaxMsgSize = 10 * 1024 * 1024 // Maximum cap on the size of a protocol message
// NetworkID - the network id
NetworkID = 1
// ProtocolMaxMsgSize - Maximum cap on the size of a protocol message
ProtocolMaxMsgSize = 10 * 1024 * 1024
)
// les protocol message codes
@ -91,20 +93,35 @@ var requests = map[uint64]requestInfo{
type errCode int
const (
// ErrMsgTooLarge - message too large
ErrMsgTooLarge = iota
// ErrDecode - decode error
ErrDecode
// ErrInvalidMsgCode - invalid message error
ErrInvalidMsgCode
// ErrProtocolVersionMismatch - mismatched protocol version
ErrProtocolVersionMismatch
ErrNetworkIdMismatch
// ErrNetworkIDMismatch - wrong network id
ErrNetworkIDMismatch
// ErrGenesisBlockMismatch - incompatible genesis blocks
ErrGenesisBlockMismatch
// ErrNoStatusMsg -
ErrNoStatusMsg
// ErrExtraStatusMsg -
ErrExtraStatusMsg
// ErrSuspendedPeer -
ErrSuspendedPeer
// ErrUselessPeer -
ErrUselessPeer
// ErrRequestRejected -
ErrRequestRejected
// ErrUnexpectedResponse -
ErrUnexpectedResponse
// ErrInvalidResponse -
ErrInvalidResponse
// ErrTooManyTimeouts -
ErrTooManyTimeouts
// ErrMissingKey -
ErrMissingKey
)
@ -118,7 +135,7 @@ var errorToString = map[int]string{
ErrDecode: "Invalid message",
ErrInvalidMsgCode: "Invalid message code",
ErrProtocolVersionMismatch: "Protocol version mismatch",
ErrNetworkIdMismatch: "NetworkId mismatch",
ErrNetworkIDMismatch: "NetworkId mismatch",
ErrGenesisBlockMismatch: "Genesis block mismatch",
ErrNoStatusMsg: "No status message",
ErrExtraStatusMsg: "Extra status message",

21
les/server.go
View file

@ -37,6 +37,7 @@ import (
const bufLimitRatio = 6000 // fixed bufLimit/MRR ratio
// LesServer - a node service running a LES server
type LesServer struct {
lesCommons
@ -56,6 +57,7 @@ type LesServer struct {
priorityClientPool *priorityClientPool
}
// NewLesServer - ctor creating a LES server node service
func NewLesServer(eth *eth.Ethereum, config *eth.Config) (*LesServer, error) {
quitSync := make(chan struct{})
pm, err := NewProtocolManager(
@ -124,9 +126,13 @@ func NewLesServer(eth *eth.Ethereum, config *eth.Config) (*LesServer, error) {
}
srv.chtIndexer.Start(eth.BlockChain())
eth.GetBloomIndexer().AddChildIndexer(srv.bloomTrieIndexer)
return srv, nil
}
// APIs - the RPC APIs offered by this server
func (s *LesServer) APIs() []rpc.API {
return []rpc.API{
{
@ -180,12 +186,13 @@ func (s *LesServer) startEventLoop() {
}()
}
// Protocols - rlpx capabilities offered by this LES server
func (s *LesServer) Protocols() []p2p.Protocol {
return s.makeProtocols(ServerProtocolVersions)
}
// Start starts the LES server
func (s *LesServer) Start(srvr *p2p.Server) {
func (s *LesServer) Start(srvr *p2p.Server) error {
s.maxPeers = s.config.LightPeers
totalRecharge := s.costTracker.totalRecharge()
if s.maxPeers > 0 {
@ -225,22 +232,30 @@ func (s *LesServer) Start(srvr *p2p.Server) {
}
s.privateKey = srvr.PrivateKey
s.protocolManager.blockLoop()
return nil
}
// SetBloomBitsIndexer - adds the bloombits indexer as a child indexer to the bloomtrie indexer
func (s *LesServer) SetBloomBitsIndexer(bloomIndexer *core.ChainIndexer) {
bloomIndexer.AddChildIndexer(s.bloomTrieIndexer)
}
// Stop stops the LES service
func (s *LesServer) Stop() {
s.chtIndexer.Close()
func (s *LesServer) Stop() error {
err := s.chtIndexer.Close()
// bloom trie indexer is closed by parent bloombits indexer
go func() {
<-s.protocolManager.noMorePeers
}()
s.freeClientPool.stop()
s.costTracker.stop()
s.protocolManager.Stop()
return err
}
// todo(rjl493456442) separate client and server implementation.

4
les/ulc_test.go
View file

@ -176,8 +176,8 @@ func connectPeers(full, light pairPeer, version int) (*peer, *peer, error) {
// Create a message pipe to communicate through
app, net := p2p.MsgPipe()
peerLight := full.PM.newPeer(version, NetworkId, p2p.NewPeer(light.Node.ID(), light.Name, nil), net)
peerFull := light.PM.newPeer(version, NetworkId, p2p.NewPeer(full.Node.ID(), full.Name, nil), app)
peerLight := full.PM.newPeer(version, NetworkID, p2p.NewPeer(light.Node.ID(), light.Name, nil), net)
peerFull := light.PM.newPeer(version, NetworkID, p2p.NewPeer(full.Node.ID(), full.Name, nil), app)
// Start the peerLight on a new thread
errc1 := make(chan error, 1)