diff --git a/les/fetcher.go b/les/fetcher.go new file mode 100644 index 0000000000..f7595cc789 --- /dev/null +++ b/les/fetcher.go @@ -0,0 +1,207 @@ +// Copyright 2015 The go-ethereum Authors +// This file is part of the go-ethereum library. +// +// The go-ethereum library is free software: you can redistribute it and/or modify +// it under the terms of the GNU Lesser General Public License as published by +// the Free Software Foundation, either version 3 of the License, or +// (at your option) any later version. +// +// The go-ethereum library is distributed in the hope that it will be useful, +// but WITHOUT ANY WARRANTY; without even the implied warranty of +// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +// GNU Lesser General Public License for more details. +// +// You should have received a copy of the GNU Lesser General Public License +// along with the go-ethereum library. If not, see . + +// Package les implements the Light Ethereum Subprotocol. +package les + +import ( +"fmt" + "sync" + "time" + + "github.com/ethereum/go-ethereum/core" + "github.com/ethereum/go-ethereum/core/types" +) + +type lightFetcher struct{ + pm *ProtocolManager + odr *LesOdr + chain BlockChain + reqMu sync.RWMutex + requested map[uint64]chan *types.Header + syncPoolMu sync.Mutex + syncPool map[*peer]struct{} + syncPoolNotify chan struct{} +} + +func newLightFetcher(pm *ProtocolManager) *lightFetcher { + f := &lightFetcher{ + pm: pm, + chain: pm.blockchain, + odr: pm.odr, + requested: make(map[uint64]chan *types.Header), + syncPool: make(map[*peer]struct{}), + syncPoolNotify: make(chan struct{}), + } + go f.syncLoop() + return f +} + +func (f *lightFetcher) requestedID(reqID uint64) bool { + f.reqMu.RLock() + _, ok := f.requested[reqID] + f.reqMu.RUnlock() + return ok +} + +func (f *lightFetcher) deliverHeaders(reqID uint64, headers []*types.Header) { + f.reqMu.Lock() + chn := f.requested[reqID] + if len(headers) == 1 { + chn <- headers[0] + } else { + chn <- nil + } + close(chn) + delete(f.requested, reqID) + f.reqMu.Unlock() +} + +func (f *lightFetcher) notify(p *peer, block blockInfo) { + p.lock.Lock() + if block.Td.Cmp(p.headInfo.Td) <= 0 { + p.lock.Unlock() + return + } + p.headInfo = block + p.lock.Unlock() + + head := f.pm.blockchain.CurrentHeader() + currentTd := core.GetTd(f.pm.chainDb, head.Hash(), head.Number.Uint64()) + if block.Td.Cmp(currentTd) > 0 { + f.syncPoolMu.Lock() + f.syncPool[p] = struct{}{} + f.syncPoolMu.Unlock() + f.syncPoolNotify <- struct{}{} + } +} + +func (f *lightFetcher) fetchBestFromPool() *peer { + head := f.pm.blockchain.CurrentHeader() + currentTd := core.GetTd(f.pm.chainDb, head.Hash(), head.Number.Uint64()) + + f.syncPoolMu.Lock() + var best *peer + for p, _ := range f.syncPool { + td := p.Td() + if td.Cmp(currentTd) <= 0 { + delete(f.syncPool, p) + } else { + if best == nil || td.Cmp(best.Td()) > 0 { + best = p + } + } + } + if best != nil { + delete(f.syncPool, best) + } + f.syncPoolMu.Unlock() + return best +} + +func (f *lightFetcher) syncLoop() { + for { + select { + case <-f.syncPoolNotify: + chn := f.pm.getSyncLock(false) + if chn != nil { + go func() { + <-chn + f.syncPoolNotify <- struct{}{} + }() + } else { + if p := f.fetchBestFromPool(); p != nil { + go f.syncWithPeer(p) + go func() { + time.Sleep(softRequestTimeout) + f.syncPoolNotify <- struct{}{} + }() + } + } + case <-f.pm.quitSync: + return + } + } +} + +func (f *lightFetcher) syncWithPeer(p *peer) bool { + headNum := f.chain.CurrentHeader().Number.Uint64() + peerHead := p.headBlockInfo() + + if !f.pm.needToSync(peerHead) { + return true + } + + if peerHead.Number <= headNum+1 { + var header *types.Header + reqID, chn := f.request(p, peerHead) + select { + case header = <-chn: + if header == nil || header.Hash() != peerHead.Hash || + header.Number.Uint64() != peerHead.Number { + // missing or wrong header returned +fmt.Println("removePeer 1") + f.pm.removePeer(p.id) + return false + } + + case <-time.After(hardRequestTimeout): +fmt.Println("removePeer 2") + f.pm.removePeer(p.id) + f.reqMu.Lock() + close(f.requested[reqID]) + delete(f.requested, reqID) + f.reqMu.Unlock() + return false + } + + // got the header, try to insert + f.chain.InsertHeaderChain([]*types.Header{header}, 1) + + defer func() { + // check header td at the end of syncing, drop peer if it was fake + headerTd := core.GetTd(f.pm.chainDb, header.Hash(), header.Number.Uint64()) + if headerTd != nil && headerTd.Cmp(peerHead.Td) != 0 { +fmt.Println("removePeer 3") + f.pm.removePeer(p.id) + } + }() + if !f.pm.needToSync(peerHead) { + return true + } + } + + f.pm.waitSyncLock() + if !f.pm.needToSync(peerHead) { + // synced up by the one we've been waiting to end + f.pm.releaseSyncLock() + return true + } + f.pm.syncWithLockAcquired(p) + return !f.pm.needToSync(peerHead) +} + +func (f *lightFetcher) request(p *peer, block blockInfo) (uint64, chan *types.Header) { + reqID := f.odr.getNextReqID() + f.reqMu.Lock() + chn := make(chan *types.Header, 1) + f.requested[reqID] = chn + f.reqMu.Unlock() + cost := p.GetRequestCost(GetBlockHeadersMsg, 1) + p.fcServer.SendRequest(reqID, cost) + p.RequestHeadersByHash(reqID, cost, block.Hash, 1, 0, false) + return reqID, chn +}