This commit is contained in:
Wenbiao Zheng 2018-06-03 01:40:22 +00:00 committed by GitHub
commit cb11f4e673
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
15 changed files with 179 additions and 47 deletions

View file

@ -102,7 +102,12 @@ func (b *LesApiBackend) GetTd(hash common.Hash) *big.Int {
return b.eth.blockchain.GetTdByHash(hash) return b.eth.blockchain.GetTdByHash(hash)
} }
func (b *LesApiBackend) GetEVM(ctx context.Context, msg core.Message, state *state.StateDB, header *types.Header, vmCfg vm.Config) (*vm.EVM, func() error, error) { func (b *LesApiBackend) GetEVM(
ctx context.Context,
msg core.Message,
state *state.StateDB,
header *types.Header,
vmCfg vm.Config) (*vm.EVM, func() error, error) {
state.SetBalance(msg.From(), math.MaxBig256) state.SetBalance(msg.From(), math.MaxBig256)
context := core.NewEVMContext(msg, header, b.eth.blockchain, nil) context := core.NewEVMContext(msg, header, b.eth.blockchain, nil)
return vm.NewEVM(context, state, b.eth.chainConfig, vmCfg), state.Error, nil return vm.NewEVM(context, state, b.eth.chainConfig, vmCfg), state.Error, nil

View file

@ -66,6 +66,7 @@ type LightEthereum struct {
chainDb ethdb.Database // Block chain database chainDb ethdb.Database // Block chain database
bloomRequests chan chan *bloombits.Retrieval // Channel receiving bloom data retrieval requests bloomRequests chan chan *bloombits.Retrieval // Channel receiving bloom data retrieval requests
bloomIndexer, chtIndexer, bloomTrieIndexer *core.ChainIndexer bloomIndexer, chtIndexer, bloomTrieIndexer *core.ChainIndexer
ApiBackend *LesApiBackend ApiBackend *LesApiBackend
@ -127,7 +128,10 @@ func New(ctx *node.ServiceContext, config *eth.Config) (*LightEthereum, error) {
} }
leth.txPool = light.NewTxPool(leth.chainConfig, leth.blockchain, leth.relay) leth.txPool = light.NewTxPool(leth.chainConfig, leth.blockchain, leth.relay)
if leth.protocolManager, err = NewProtocolManager(leth.chainConfig, true, ClientProtocolVersions, config.NetworkId, leth.eventMux, leth.engine, leth.peers, leth.blockchain, nil, chainDb, leth.odr, leth.relay, quitSync, &leth.wg); err != nil { if leth.protocolManager, err = NewProtocolManager(
leth.chainConfig, true, ClientProtocolVersions, config.NetworkId,
leth.eventMux, leth.engine, leth.peers, leth.blockchain,
nil, chainDb, leth.odr, leth.relay, quitSync, &leth.wg); err != nil {
return nil, err return nil, err
} }
leth.ApiBackend = &LesApiBackend{leth, nil} leth.ApiBackend = &LesApiBackend{leth, nil}

View file

@ -85,14 +85,14 @@ func newRequestDistributor(peers *peerSet, stopChn chan struct{}) *requestDistri
return d return d
} }
// registerPeer implements peerSetNotify // registerPeer adds a new peer
func (d *requestDistributor) registerPeer(p *peer) { func (d *requestDistributor) registerPeer(p *peer) {
d.peerLock.Lock() d.peerLock.Lock()
d.peers[p] = struct{}{} d.peers[p] = struct{}{}
d.peerLock.Unlock() d.peerLock.Unlock()
} }
// unregisterPeer implements peerSetNotify // unregisterPeer deletes an old peer
func (d *requestDistributor) unregisterPeer(p *peer) { func (d *requestDistributor) unregisterPeer(p *peer) {
d.peerLock.Lock() d.peerLock.Lock()
delete(d.peers, p) delete(d.peers, p)
@ -144,9 +144,12 @@ func (d *requestDistributor) loop() {
// queued request will wake up the loop // queued request will wake up the loop
break loop break loop
} }
d.loopNextSent = true // a "next" signal has been sent, do not send another one until this one has been received // a "next" signal has been sent,
// do not send another one until this one has been received
d.loopNextSent = true
if wait > distMaxWait { if wait > distMaxWait {
// waiting times may be reduced by incoming request replies, if it is too long, recalculate it periodically // waiting times may be reduced by incoming request replies,
// if it is too long, recalculate it periodically
wait = distMaxWait wait = distMaxWait
} }
go func() { go func() {
@ -200,7 +203,11 @@ func (d *requestDistributor) nextRequest() (distPeer, *distReq, time.Duration) {
if sel == nil { if sel == nil {
sel = newWeightedRandomSelect() sel = newWeightedRandomSelect()
} }
sel.update(selectPeerItem{peer: peer, req: req, weight: int64(bufRemain*1000000) + 1}) sel.update(selectPeerItem{
peer: peer,
req: req,
weight: int64(bufRemain*1000000) + 1,
})
} else { } else {
if bestReq == nil || wait < bestWait { if bestReq == nil || wait < bestWait {
bestPeer = peer bestPeer = peer

View file

@ -363,7 +363,8 @@ func (f *lightFetcher) peerHasBlock(p *peer, hash common.Hash, number uint64) bo
// //
// when syncing, just check if it is part of the known chain, there is nothing better we // when syncing, just check if it is part of the known chain, there is nothing better we
// can do since we do not know the most recent block hash yet // can do since we do not know the most recent block hash yet
return rawdb.ReadCanonicalHash(f.pm.chainDb, fp.root.number) == fp.root.hash && rawdb.ReadCanonicalHash(f.pm.chainDb, number) == hash return rawdb.ReadCanonicalHash(f.pm.chainDb, fp.root.number) == fp.root.hash &&
rawdb.ReadCanonicalHash(f.pm.chainDb, number) == hash
} }
// requestAmount calculates the amount of headers to be downloaded starting // requestAmount calculates the amount of headers to be downloaded starting
@ -476,13 +477,20 @@ func (f *lightFetcher) nextRequest() (*distReq, uint64) {
cost := p.GetRequestCost(GetBlockHeadersMsg, int(bestAmount)) cost := p.GetRequestCost(GetBlockHeadersMsg, int(bestAmount))
p.fcServer.QueueRequest(reqID, cost) p.fcServer.QueueRequest(reqID, cost)
f.reqMu.Lock() f.reqMu.Lock()
f.requested[reqID] = fetchRequest{hash: bestHash, amount: bestAmount, peer: p, sent: mclock.Now()} f.requested[reqID] = fetchRequest{
hash: bestHash,
amount: bestAmount,
peer: p,
sent: mclock.Now(),
}
f.reqMu.Unlock() f.reqMu.Unlock()
go func() { go func() {
time.Sleep(hardRequestTimeout) time.Sleep(hardRequestTimeout)
f.timeoutChn <- reqID f.timeoutChn <- reqID
}() }()
return func() { p.RequestHeadersByHash(reqID, cost, bestHash, int(bestAmount), 0, true) } return func() {
p.RequestHeadersByHash(reqID, cost, bestHash, int(bestAmount), 0, true)
}
}, },
} }
} }
@ -497,7 +505,8 @@ func (f *lightFetcher) deliverHeaders(peer *peer, reqID uint64, headers []*types
// processResponse processes header download request responses, returns true if successful // processResponse processes header download request responses, returns true if successful
func (f *lightFetcher) processResponse(req fetchRequest, resp fetchResponse) bool { func (f *lightFetcher) processResponse(req fetchRequest, resp fetchResponse) bool {
if uint64(len(resp.headers)) != req.amount || resp.headers[0].Hash() != req.hash { if uint64(len(resp.headers)) != req.amount || resp.headers[0].Hash() != req.hash {
req.peer.Log().Debug("Response content mismatch", "requested", len(resp.headers), "reqfrom", resp.headers[0], "delivered", req.amount, "delfrom", req.hash) req.peer.Log().Debug("Response content mismatch", "requested", len(resp.headers),
"reqfrom", resp.headers[0], "delivered", req.amount, "delfrom", req.hash)
return false return false
} }
headers := make([]*types.Header, req.amount) headers := make([]*types.Header, req.amount)

View file

@ -126,9 +126,15 @@ type ProtocolManager struct {
wg *sync.WaitGroup wg *sync.WaitGroup
} }
// NewProtocolManager returns a new ethereum sub protocol manager. The Ethereum sub protocol manages peers capable // NewProtocolManager returns a new ethereum sub protocol manager.
// with the ethereum network. // The Ethereum sub protocol manages peers capable with the ethereum network.
func NewProtocolManager(chainConfig *params.ChainConfig, lightSync bool, protocolVersions []uint, networkId uint64, mux *event.TypeMux, engine consensus.Engine, peers *peerSet, blockchain BlockChain, txpool txPool, chainDb ethdb.Database, odr *LesOdr, txrelay *LesTxRelay, quitSync chan struct{}, wg *sync.WaitGroup) (*ProtocolManager, error) { func NewProtocolManager(
chainConfig *params.ChainConfig, lightSync bool,
protocolVersions []uint, networkId uint64,
mux *event.TypeMux, engine consensus.Engine, peers *peerSet,
blockchain BlockChain, txpool txPool, chainDb ethdb.Database,
odr *LesOdr, txrelay *LesTxRelay,
quitSync chan struct{}, wg *sync.WaitGroup) (*ProtocolManager, error) {
// Create the protocol manager with the base fields // Create the protocol manager with the base fields
manager := &ProtocolManager{ manager := &ProtocolManager{
lightSync: lightSync, lightSync: lightSync,
@ -331,7 +337,19 @@ func (pm *ProtocolManager) handle(p *peer) error {
} }
} }
var reqList = []uint64{GetBlockHeadersMsg, GetBlockBodiesMsg, GetCodeMsg, GetReceiptsMsg, GetProofsV1Msg, SendTxMsg, SendTxV2Msg, GetTxStatusMsg, GetHeaderProofsMsg, GetProofsV2Msg, GetHelperTrieProofsMsg} var reqList = []uint64{
GetBlockHeadersMsg,
GetBlockBodiesMsg,
GetCodeMsg,
GetReceiptsMsg,
GetProofsV1Msg,
SendTxMsg,
SendTxV2Msg,
GetTxStatusMsg,
GetHeaderProofsMsg,
GetProofsV2Msg,
GetHelperTrieProofsMsg,
}
// handleMsg is invoked whenever an inbound message is received from a remote // handleMsg is invoked whenever an inbound message is received from a remote
// peer. The remote connection is torn down upon returning any error. // peer. The remote connection is torn down upon returning any error.

View file

@ -33,7 +33,13 @@ type LesOdr struct {
stop chan struct{} stop chan struct{}
} }
func NewLesOdr(db ethdb.Database, chtIndexer, bloomTrieIndexer, bloomIndexer *core.ChainIndexer, retriever *retrieveManager) *LesOdr { // NewLesOdr creates a struct instance of LesOdr
func NewLesOdr(
db ethdb.Database,
chtIndexer,
bloomTrieIndexer,
bloomIndexer *core.ChainIndexer,
retriever *retrieveManager) *LesOdr {
return &LesOdr{ return &LesOdr{
db: db, db: db,
chtIndexer: chtIndexer, chtIndexer: chtIndexer,
@ -108,7 +114,10 @@ func (odr *LesOdr) Retrieve(ctx context.Context, req light.OdrRequest) (err erro
}, },
} }
if err = odr.retriever.retrieve(ctx, reqID, rq, func(p distPeer, msg *Msg) error { return lreq.Validate(odr.db, msg) }, odr.stop); err == nil { if err = odr.retriever.retrieve(ctx, reqID, rq,
func(p distPeer, msg *Msg) error {
return lreq.Validate(odr.db, msg)
}, odr.stop); err == nil {
// retrieved from network, store in db // retrieved from network, store in db
req.StoreResult(odr.db) req.StoreResult(odr.db)
} else { } else {

View file

@ -365,7 +365,8 @@ func (r *ChtRequest) CanSend(peer *peer) bool {
peer.lock.RLock() peer.lock.RLock()
defer peer.lock.RUnlock() defer peer.lock.RUnlock()
return peer.headInfo.Number >= light.HelperTrieConfirmations && r.ChtNum <= (peer.headInfo.Number-light.HelperTrieConfirmations)/light.CHTFrequencyClient return peer.headInfo.Number >= light.HelperTrieConfirmations &&
r.ChtNum <= (peer.headInfo.Number-light.HelperTrieConfirmations)/light.CHTFrequencyClient
} }
// Request sends an ODR request to the LES network (implementation of LesOdrRequest) // Request sends an ODR request to the LES network (implementation of LesOdrRequest)
@ -484,7 +485,8 @@ func (r *BloomRequest) CanSend(peer *peer) bool {
if peer.version < lpv2 { if peer.version < lpv2 {
return false return false
} }
return peer.headInfo.Number >= light.HelperTrieConfirmations && r.BloomTrieNum <= (peer.headInfo.Number-light.HelperTrieConfirmations)/light.BloomTrieFrequency return peer.headInfo.Number >= light.HelperTrieConfirmations &&
r.BloomTrieNum <= (peer.headInfo.Number-light.HelperTrieConfirmations)/light.BloomTrieFrequency
} }
// Request sends an ODR request to the LES network (implementation of LesOdrRequest) // Request sends an ODR request to the LES network (implementation of LesOdrRequest)

View file

@ -131,7 +131,11 @@ func (p *peer) headBlockInfo() blockInfo {
p.lock.RLock() p.lock.RLock()
defer p.lock.RUnlock() defer p.lock.RUnlock()
return blockInfo{Hash: p.headInfo.Hash, Number: p.headInfo.Number, Td: p.headInfo.Td} return blockInfo{
Hash: p.headInfo.Hash,
Number: p.headInfo.Number,
Td: p.headInfo.Td,
}
} }
// Td retrieves the current total difficulty of a peer. // Td retrieves the current total difficulty of a peer.
@ -247,7 +251,11 @@ func (p *peer) RequestHeadersByHash(reqID, cost uint64, origin common.Hash, amou
// specified header query, based on the number of an origin block. // specified header query, based on the number of an origin block.
func (p *peer) RequestHeadersByNumber(reqID, cost, origin uint64, amount int, skip int, reverse bool) error { func (p *peer) RequestHeadersByNumber(reqID, cost, origin uint64, amount int, skip int, reverse bool) error {
p.Log().Debug("Fetching batch of headers", "count", amount, "fromnum", origin, "skip", skip, "reverse", reverse) p.Log().Debug("Fetching batch of headers", "count", amount, "fromnum", origin, "skip", skip, "reverse", reverse)
return sendRequest(p.rw, GetBlockHeadersMsg, reqID, cost, &getBlockHeadersData{Origin: hashOrNumber{Number: origin}, Amount: uint64(amount), Skip: uint64(skip), Reverse: reverse}) return sendRequest(p.rw, GetBlockHeadersMsg, reqID, cost, &getBlockHeadersData{
Origin: hashOrNumber{Number: origin},
Amount: uint64(amount),
Skip: uint64(skip),
Reverse: reverse})
} }
// RequestBodies fetches a batch of blocks' bodies corresponding to the hashes // RequestBodies fetches a batch of blocks' bodies corresponding to the hashes
@ -295,7 +303,11 @@ func (p *peer) RequestHelperTrieProofs(reqID, cost uint64, reqs []HelperTrieReq)
} }
blockNum := binary.BigEndian.Uint64(req.Key) blockNum := binary.BigEndian.Uint64(req.Key)
// convert HelperTrie request to old CHT request // convert HelperTrie request to old CHT request
reqsV1[i] = ChtReq{ChtNum: (req.TrieIdx + 1) * (light.CHTFrequencyClient / light.CHTFrequencyServer), BlockNum: blockNum, FromLevel: req.FromLevel} reqsV1[i] = ChtReq{
ChtNum: (req.TrieIdx + 1) * (light.CHTFrequencyClient / light.CHTFrequencyServer),
BlockNum: blockNum,
FromLevel: req.FromLevel,
}
} }
return sendRequest(p.rw, GetHeaderProofsMsg, reqID, cost, reqsV1) return sendRequest(p.rw, GetHeaderProofsMsg, reqID, cost, reqsV1)
case lpv2: case lpv2:

View file

@ -36,7 +36,10 @@ type weightedRandomSelect struct {
// newWeightedRandomSelect returns a new weightedRandomSelect structure // newWeightedRandomSelect returns a new weightedRandomSelect structure
func newWeightedRandomSelect() *weightedRandomSelect { func newWeightedRandomSelect() *weightedRandomSelect {
return &weightedRandomSelect{root: &wrsNode{maxItems: wrsBranches}, idx: make(map[wrsItem]int)} return &weightedRandomSelect{
root: &wrsNode{maxItems: wrsBranches},
idx: make(map[wrsItem]int),
}
} }
// update updates an item's weight, adds it if it was non-existent or removes it if // update updates an item's weight, adds it if it was non-existent or removes it if
@ -62,7 +65,12 @@ func (w *weightedRandomSelect) setWeight(item wrsItem, weight int64) {
if weight != 0 { if weight != 0 {
if w.root.itemCnt == w.root.maxItems { if w.root.itemCnt == w.root.maxItems {
// add a new level // add a new level
newRoot := &wrsNode{sumWeight: w.root.sumWeight, itemCnt: w.root.itemCnt, level: w.root.level + 1, maxItems: w.root.maxItems * wrsBranches} newRoot := &wrsNode{
sumWeight: w.root.sumWeight,
itemCnt: w.root.itemCnt,
level: w.root.level + 1,
maxItems: w.root.maxItems * wrsBranches,
}
newRoot.items[0] = w.root newRoot.items[0] = w.root
newRoot.weights[0] = w.root.sumWeight newRoot.weights[0] = w.root.sumWeight
w.root = newRoot w.root = newRoot

View file

@ -52,7 +52,21 @@ type LesServer struct {
func NewLesServer(eth *eth.Ethereum, config *eth.Config) (*LesServer, error) { func NewLesServer(eth *eth.Ethereum, config *eth.Config) (*LesServer, error) {
quitSync := make(chan struct{}) quitSync := make(chan struct{})
pm, err := NewProtocolManager(eth.BlockChain().Config(), false, ServerProtocolVersions, config.NetworkId, eth.EventMux(), eth.Engine(), newPeerSet(), eth.BlockChain(), eth.TxPool(), eth.ChainDb(), nil, nil, quitSync, new(sync.WaitGroup)) pm, err := NewProtocolManager(
eth.BlockChain().Config(),
false,
ServerProtocolVersions,
config.NetworkId,
eth.EventMux(),
eth.Engine(),
newPeerSet(),
eth.BlockChain(),
eth.TxPool(),
eth.ChainDb(),
nil,
nil,
quitSync,
new(sync.WaitGroup))
if err != nil { if err != nil {
return nil, err return nil, err
} }
@ -341,7 +355,12 @@ func (pm *ProtocolManager) blockLoop() {
log.Debug("Announcing block to peers", "number", number, "hash", hash, "td", td, "reorg", reorg) log.Debug("Announcing block to peers", "number", number, "hash", hash, "td", td, "reorg", reorg)
announce := announceData{Hash: hash, Number: number, Td: td, ReorgDepth: reorg} announce := announceData{
Hash: hash,
Number: number,
Td: td,
ReorgDepth: reorg,
}
var ( var (
signed bool signed bool
signedAnnounce announceData signedAnnounce announceData

View file

@ -85,10 +85,11 @@ const (
// initStatsWeight is used to initialize previously unknown peers with good // initStatsWeight is used to initialize previously unknown peers with good
// statistics to give a chance to prove themselves // statistics to give a chance to prove themselves
initStatsWeight = 1 initStatsWeight = 1
baseStatsWeight = 1000000000
) )
// serverPool implements a pool for storing and selecting newly discovered and already // serverPool implements a pool for storing and selecting newly discovered and already
// known light server nodes. It received discovered nodes, stores statistics about // known light server nodes. It receives discovered nodes, stores statistics about
// known nodes and takes care of always having enough good quality servers connected. // known nodes and takes care of always having enough good quality servers connected.
type serverPool struct { type serverPool struct {
db ethdb.Database db ethdb.Database
@ -560,7 +561,15 @@ type poolEntry struct {
} }
func (e *poolEntry) EncodeRLP(w io.Writer) error { func (e *poolEntry) EncodeRLP(w io.Writer) error {
return rlp.Encode(w, []interface{}{e.id, e.lastConnected.ip, e.lastConnected.port, e.lastConnected.fails, &e.connectStats, &e.delayStats, &e.responseStats, &e.timeoutStats}) return rlp.Encode(w, []interface{}{
e.id,
e.lastConnected.ip,
e.lastConnected.port,
e.lastConnected.fails,
&e.connectStats,
&e.delayStats,
&e.responseStats,
&e.timeoutStats})
} }
func (e *poolEntry) DecodeRLP(s *rlp.Stream) error { func (e *poolEntry) DecodeRLP(s *rlp.Stream) error {
@ -574,7 +583,12 @@ func (e *poolEntry) DecodeRLP(s *rlp.Stream) error {
if err := s.Decode(&entry); err != nil { if err := s.Decode(&entry); err != nil {
return err return err
} }
addr := &poolEntryAddress{ip: entry.IP, port: entry.Port, fails: entry.Fails, lastSeen: mclock.Now()} addr := &poolEntryAddress{
ip: entry.IP,
port: entry.Port,
fails: entry.Fails,
lastSeen: mclock.Now(),
}
e.id = entry.ID e.id = entry.ID
e.addr = make(map[string]*poolEntryAddress) e.addr = make(map[string]*poolEntryAddress)
e.addr[addr.strKey()] = addr e.addr[addr.strKey()] = addr
@ -600,9 +614,9 @@ func (e *discoveredEntry) Weight() int64 {
} }
t := time.Duration(mclock.Now() - e.lastDiscovered) t := time.Duration(mclock.Now() - e.lastDiscovered)
if t <= discoverExpireStart { if t <= discoverExpireStart {
return 1000000000 return baseStatsWeight
} }
return int64(1000000000 * math.Exp(-float64(t-discoverExpireStart)/float64(discoverExpireConst))) return int64(baseStatsWeight * math.Exp(-float64(t-discoverExpireStart)/float64(discoverExpireConst)))
} }
// knownEntry implements wrsItem // knownEntry implements wrsItem
@ -613,7 +627,12 @@ func (e *knownEntry) Weight() int64 {
if e.state != psNotConnected || !e.known || e.delayedRetry { if e.state != psNotConnected || !e.known || e.delayedRetry {
return 0 return 0
} }
return int64(1000000000 * e.connectStats.recentAvg() * math.Exp(-float64(e.lastConnected.fails)*failDropLn-e.responseStats.recentAvg()/float64(responseScoreTC)-e.delayStats.recentAvg()/float64(delayScoreTC)) * math.Pow(1-e.timeoutStats.recentAvg(), timeoutPow)) return int64(baseStatsWeight *
e.connectStats.recentAvg() *
math.Exp(-float64(e.lastConnected.fails)*failDropLn-
e.responseStats.recentAvg()/float64(responseScoreTC)-
e.delayStats.recentAvg()/float64(delayScoreTC)) *
math.Pow(1-e.timeoutStats.recentAvg(), timeoutPow))
} }
// poolEntryAddress is a separate object because currently it is necessary to remember // poolEntryAddress is a separate object because currently it is necessary to remember
@ -629,7 +648,9 @@ type poolEntryAddress struct {
func (a *poolEntryAddress) Weight() int64 { func (a *poolEntryAddress) Weight() int64 {
t := time.Duration(mclock.Now() - a.lastSeen) t := time.Duration(mclock.Now() - a.lastSeen)
return int64(1000000*math.Exp(-float64(t)/float64(discoverExpireConst)-float64(a.fails)*addrFailDropLn)) + 1 return 1 + int64(baseStatsWeight*
math.Exp(-float64(t)/float64(discoverExpireConst)-
float64(a.fails)*addrFailDropLn))
} }
func (a *poolEntryAddress) strKey() string { func (a *poolEntryAddress) strKey() string {
@ -688,7 +709,9 @@ func (s *poolStats) recentAvg() float64 {
} }
func (s *poolStats) EncodeRLP(w io.Writer) error { func (s *poolStats) EncodeRLP(w io.Writer) error {
return rlp.Encode(w, []interface{}{math.Float64bits(s.sum), math.Float64bits(s.weight)}) return rlp.Encode(w, []interface{}{
math.Float64bits(s.sum),
math.Float64bits(s.weight)})
} }
func (s *poolStats) DecodeRLP(st *rlp.Stream) error { func (s *poolStats) DecodeRLP(st *rlp.Stream) error {
@ -712,7 +735,11 @@ type poolEntryQueue struct {
// newPoolEntryQueue returns a new poolEntryQueue // newPoolEntryQueue returns a new poolEntryQueue
func newPoolEntryQueue(maxCnt int, removeFromPool func(*poolEntry)) poolEntryQueue { func newPoolEntryQueue(maxCnt int, removeFromPool func(*poolEntry)) poolEntryQueue {
return poolEntryQueue{queue: make(map[int]*poolEntry), maxCnt: maxCnt, removeFromPool: removeFromPool} return poolEntryQueue{
queue: make(map[int]*poolEntry),
maxCnt: maxCnt,
removeFromPool: removeFromPool,
}
} }
// fetchOldest returns and removes the least recently accessed entry // fetchOldest returns and removes the least recently accessed entry

View file

@ -104,7 +104,8 @@ func NewLightChain(odr OdrBackend, config *params.ChainConfig, engine consensus.
if err := bc.loadLastState(); err != nil { if err := bc.loadLastState(); err != nil {
return nil, err return nil, err
} }
// Check the current state of the block hashes and make sure that we do not have any of the bad blocks in our chain // Check the current state of the block hashes and
// make sure that we do not have any of the bad blocks in our chain
for hash := range core.BadHashes { for hash := range core.BadHashes {
if header := bc.GetHeaderByHash(hash); header != nil { if header := bc.GetHeaderByHash(hash); header != nil {
log.Error("Found bad hash, rewinding chain", "number", header.Number, "hash", header.ParentHash) log.Error("Found bad hash, rewinding chain", "number", header.Number, "hash", header.ParentHash)

View file

@ -49,7 +49,8 @@ func GetHeaderByNumber(ctx context.Context, odr OdrBackend, number uint64) (*typ
if odr.ChtIndexer() != nil { if odr.ChtIndexer() != nil {
chtCount, sectionHeadNum, sectionHead = odr.ChtIndexer().Sections() chtCount, sectionHeadNum, sectionHead = odr.ChtIndexer().Sections()
canonicalHash := rawdb.ReadCanonicalHash(db, sectionHeadNum) canonicalHash := rawdb.ReadCanonicalHash(db, sectionHeadNum)
// if the CHT was injected as a trusted checkpoint, we have no canonical hash yet so we accept zero hash too // if the CHT was injected as a trusted checkpoint,
// we have no canonical hash yet so we accept zero hash too
for chtCount > 0 && canonicalHash != sectionHead && canonicalHash != (common.Hash{}) { for chtCount > 0 && canonicalHash != sectionHead && canonicalHash != (common.Hash{}) {
chtCount-- chtCount--
if chtCount > 0 { if chtCount > 0 {
@ -94,8 +95,7 @@ func GetBodyRLP(ctx context.Context, odr OdrBackend, hash common.Hash, number ui
} }
} }
// GetBody retrieves the block body (transactons, uncles) corresponding to the // GetBody retrieves the block body (transactons, uncles) corresponding to the hash.
// hash.
func GetBody(ctx context.Context, odr OdrBackend, hash common.Hash, number uint64) (*types.Body, error) { func GetBody(ctx context.Context, odr OdrBackend, hash common.Hash, number uint64) (*types.Body, error) {
data, err := GetBodyRLP(ctx, odr, hash, number) data, err := GetBodyRLP(ctx, odr, hash, number)
if err != nil { if err != nil {
@ -173,7 +173,8 @@ func GetBlockLogs(ctx context.Context, odr OdrBackend, hash common.Hash, number
return logs, nil return logs, nil
} }
// GetBloomBits retrieves a batch of compressed bloomBits vectors belonging to the given bit index and section indexes // GetBloomBits retrieves a batch of compressed bloomBits vectors
// belonging to the given bit index and section indexes
func GetBloomBits(ctx context.Context, odr OdrBackend, bitIdx uint, sectionIdxList []uint64) ([][]byte, error) { func GetBloomBits(ctx context.Context, odr OdrBackend, bitIdx uint, sectionIdxList []uint64) ([][]byte, error) {
db := odr.Database() db := odr.Database()
result := make([][]byte, len(sectionIdxList)) result := make([][]byte, len(sectionIdxList))
@ -189,7 +190,8 @@ func GetBloomBits(ctx context.Context, odr OdrBackend, bitIdx uint, sectionIdxLi
if odr.BloomTrieIndexer() != nil { if odr.BloomTrieIndexer() != nil {
bloomTrieCount, sectionHeadNum, sectionHead = odr.BloomTrieIndexer().Sections() bloomTrieCount, sectionHeadNum, sectionHead = odr.BloomTrieIndexer().Sections()
canonicalHash := rawdb.ReadCanonicalHash(db, sectionHeadNum) canonicalHash := rawdb.ReadCanonicalHash(db, sectionHeadNum)
// if the BloomTrie was injected as a trusted checkpoint, we have no canonical hash yet so we accept zero hash too // if the BloomTrie was injected as a trusted checkpoint,
// we have no canonical hash yet so we accept zero hash too
for bloomTrieCount > 0 && canonicalHash != sectionHead && canonicalHash != (common.Hash{}) { for bloomTrieCount > 0 && canonicalHash != sectionHead && canonicalHash != (common.Hash{}) {
bloomTrieCount-- bloomTrieCount--
if bloomTrieCount > 0 { if bloomTrieCount > 0 {
@ -220,7 +222,12 @@ func GetBloomBits(ctx context.Context, odr OdrBackend, bitIdx uint, sectionIdxLi
return result, nil return result, nil
} }
r := &BloomRequest{BloomTrieRoot: GetBloomTrieRoot(db, bloomTrieCount-1, sectionHead), BloomTrieNum: bloomTrieCount - 1, BitIdx: bitIdx, SectionIdxList: reqList} r := &BloomRequest{
BloomTrieRoot: GetBloomTrieRoot(db, bloomTrieCount-1, sectionHead),
BloomTrieNum: bloomTrieCount - 1,
BitIdx: bitIdx,
SectionIdxList: reqList,
}
if err := odr.Retrieve(ctx, r); err != nil { if err := odr.Retrieve(ctx, r); err != nil {
return nil, err return nil, err
} else { } else {

View file

@ -126,7 +126,7 @@ type ChtIndexerBackend struct {
trie *trie.Trie trie *trie.Trie
} }
// NewBloomTrieIndexer creates a BloomTrie chain indexer // NewChtIndexer creates a cht chain indexer
func NewChtIndexer(db ethdb.Database, clientMode bool) *core.ChainIndexer { func NewChtIndexer(db ethdb.Database, clientMode bool) *core.ChainIndexer {
var sectionSize, confirmReq uint64 var sectionSize, confirmReq uint64
if clientMode { if clientMode {
@ -240,7 +240,8 @@ func NewBloomTrieIndexer(db ethdb.Database, clientMode bool) *core.ChainIndexer
} }
backend.bloomTrieRatio = BloomTrieFrequency / backend.parentSectionSize backend.bloomTrieRatio = BloomTrieFrequency / backend.parentSectionSize
backend.sectionHeads = make([]common.Hash, backend.bloomTrieRatio) backend.sectionHeads = make([]common.Hash, backend.bloomTrieRatio)
return core.NewChainIndexer(db, idb, backend, BloomTrieFrequency, confirmReq-ethBloomBitsConfirmations, time.Millisecond*100, "bloomtrie") return core.NewChainIndexer(db, idb, backend, BloomTrieFrequency,
confirmReq-ethBloomBitsConfirmations, time.Millisecond*100, "bloomtrie")
} }
// Reset implements core.ChainIndexerBackend // Reset implements core.ChainIndexerBackend
@ -300,7 +301,8 @@ func (b *BloomTrieIndexerBackend) Commit() error {
b.triedb.Commit(root, false) b.triedb.Commit(root, false)
sectionHead := b.sectionHeads[b.bloomTrieRatio-1] sectionHead := b.sectionHeads[b.bloomTrieRatio-1]
log.Info("Storing bloom trie", "section", b.section, "head", sectionHead, "root", root, "compression", float64(compSize)/float64(decompSize)) log.Info("Storing bloom trie", "section", b.section, "head", sectionHead,
"root", root, "compression", float64(compSize)/float64(decompSize))
StoreBloomTrieRoot(b.diskdb, b.section, sectionHead, root) StoreBloomTrieRoot(b.diskdb, b.section, sectionHead, root)
return nil return nil

View file

@ -416,7 +416,9 @@ func (self *TxPool) add(ctx context.Context, tx *types.Transaction) error {
} }
// Print a log message if low enough level is set // Print a log message if low enough level is set
log.Debug("Pooled new transaction", "hash", hash, "from", log.Lazy{Fn: func() common.Address { from, _ := types.Sender(self.signer, tx); return from }}, "to", tx.To()) log.Debug("Pooled new transaction", "hash", hash,
"from", log.Lazy{Fn: func() common.Address { from, _ := types.Sender(self.signer, tx); return from }},
"to", tx.To())
return nil return nil
} }