diff --git a/cmd/utils/flags.go b/cmd/utils/flags.go index 7e19ebc0cb..812cc8d9aa 100644 --- a/cmd/utils/flags.go +++ b/cmd/utils/flags.go @@ -20,6 +20,7 @@ package utils import ( "crypto/ecdsa" "encoding/json" + "errors" "fmt" "io/ioutil" "math/big" @@ -1497,25 +1498,50 @@ 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) - } - return fullNode, err - }) + }); err != nil { + Fatalf("Failed to register the Light Client service: %v", err) + } } - if err != nil { - Fatalf("Failed to register the Ethereum service: %v", err) +} + +// 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(ðService) != 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) + } } } diff --git a/eth/backend.go b/eth/backend.go index fbf4dd7bbc..c3caa4603a 100644 --- a/eth/backend.go +++ b/eth/backend.go @@ -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 diff --git a/go-ethereum.code-workspace b/go-ethereum.code-workspace new file mode 100644 index 0000000000..876a1499c0 --- /dev/null +++ b/go-ethereum.code-workspace @@ -0,0 +1,8 @@ +{ + "folders": [ + { + "path": "." + } + ], + "settings": {} +} \ No newline at end of file diff --git a/les/api.go b/les/api.go index 3a8d49ca5e..fc28d066e9 100644 --- a/les/api.go +++ b/les/api.go @@ -31,8 +31,11 @@ import ( ) var ( - ErrMinCap = errors.New("capacity too small") - ErrTotalCap = errors.New("total capacity exceeded") + // 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 diff --git a/les/api_backend.go b/les/api_backend.go index 6de15e7bd2..f3ce43ae6f 100644 --- a/les/api_backend.go +++ b/les/api_backend.go @@ -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) } diff --git a/les/api_test.go b/les/api_test.go index cec9459625..a616d96145 100644 --- a/les/api_test.go +++ b/les/api_test.go @@ -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 } diff --git a/les/backend.go b/les/backend.go index ed0f45057f..ccdca5173f 100644 --- a/les/backend.go +++ b/les/backend.go @@ -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,16 +241,29 @@ 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) } -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]) } +// 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 } -func (s *LightEthereum) EventMux() *event.TypeMux { return s.eventMux } + +// EventMux is an event subscription notifier +func (s *LightEthereum) EventMux() *event.TypeMux { return s.eventMux } // Protocols implements node.Service, returning all the currently configured // network protocols to start. @@ -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)) diff --git a/les/benchmark.go b/les/benchmark.go index 925d1d89e8..d2beedc199 100644 --- a/les/benchmark.go +++ b/les/benchmark.go @@ -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) diff --git a/les/freeclient_test.go b/les/freeclient_test.go index 5a58a6c1c7..fb34662030 100644 --- a/les/freeclient_test.go +++ b/les/freeclient_test.go @@ -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") { diff --git a/les/handler.go b/les/handler.go index 59bfd81cd7..4a7b1c6cfd 100644 --- a/les/handler.go +++ b/les/handler.go @@ -50,14 +50,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 ) @@ -66,6 +74,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 @@ -87,6 +96,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 @@ -94,7 +104,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 @@ -131,7 +141,7 @@ func NewProtocolManager( chainConfig *params.ChainConfig, indexerConfig *light.IndexerConfig, client bool, - networkId uint64, + networkID uint64, mux *event.TypeMux, engine consensus.Engine, peers *peerSet, @@ -153,7 +163,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()) } diff --git a/les/helper_test.go b/les/helper_test.go index dbb0813444..a4a8610935 100644 --- a/les/helper_test.go +++ b/les/helper_test.go @@ -171,7 +171,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 } @@ -220,7 +220,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) @@ -257,8 +257,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) @@ -287,7 +287,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, testCost uint64) { 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) diff --git a/les/odr.go b/les/odr.go index 9176924cb8..db642d9e40 100644 --- a/les/odr.go +++ b/les/odr.go @@ -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 ) diff --git a/les/odr_requests.go b/les/odr_requests.go index 89c6091777..88b4eb863f 100644 --- a/les/odr_requests.go +++ b/les/odr_requests.go @@ -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 { diff --git a/les/peer.go b/les/peer.go index 6792d0611e..e2b51750ea 100644 --- a/les/peer.go +++ b/les/peer.go @@ -610,7 +610,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) diff --git a/les/peer_test.go b/les/peer_test.go index d5ce166945..e37bb6edb2 100644 --- a/les/peer_test.go +++ b/les/peer_test.go @@ -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) diff --git a/les/protocol.go b/les/protocol.go index ebf5814732..852f6f9e67 100644 --- a/les/protocol.go +++ b/les/protocol.go @@ -42,12 +42,19 @@ var ( AdvertiseProtocolVersions = []uint{lpv2} // clients are searching for the first advertised protocol in the list ) +<<<<<<< HEAD // Number of implemented message corresponding to different protocol versions. var ProtocolLengths = map[uint]uint64{lpv2: 22, lpv3: 24} +======= +// ProtocolLengths - Number of implemented message corresponding to different protocol versions. +var ProtocolLengths = map[uint]uint64{lpv2: 22} +>>>>>>> Further LES cleanup 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 @@ -95,20 +102,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 ) @@ -122,7 +144,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", diff --git a/les/server.go b/les/server.go index 836fa0d552..6810d77281 100644 --- a/les/server.go +++ b/les/server.go @@ -70,6 +70,7 @@ type LesServer struct { priorityClientPool *priorityClientPool } +// NewLesServer - ctor creating a LES server node service func NewLesServer(eth *eth.Ethereum, config *eth.Config) (*LesServer, error) { var csvLogger *csvlogger.Logger if logFileName != "" { @@ -153,9 +154,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{ { @@ -227,12 +232,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 { @@ -282,24 +288,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() { +func (s *LesServer) Stop() error { s.fcManager.Stop() 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() s.csvLogger.Stop() + return nil } // todo(rjl493456442) separate client and server implementation. diff --git a/les/ulc_test.go b/les/ulc_test.go index 38adeb95f1..59b15b9a75 100644 --- a/les/ulc_test.go +++ b/les/ulc_test.go @@ -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)