// 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 <http://www.gnu.org/licenses/>.
// Package les implements the Light Ethereum Subprotocol.
package les
import (
"math/big"
"sync"
"time"
"github.com/ethereum/go-ethereum/common"
"github.com/ethereum/go-ethereum/core"
"github.com/ethereum/go-ethereum/core/types"
)
type lightFetcher struct {
pm *ProtocolManager
odr *LesOdr
chain BlockChain
headAnnouncedMu sync.Mutex
headAnnouncedBy map[common.Hash][]*peer
currentTd *big.Int
deliverChn chan fetchResponse
reqMu sync.RWMutex
requested map[uint64]fetchRequest
timeoutChn chan uint64
notifyChn chan bool // true if initiated from outside
syncing bool
syncDone chan struct{}
}
type fetchRequest struct {
hash common.Hash
amount uint64
peer *peer
}
type fetchResponse struct {
reqID uint64
headers []*types.Header
}
func newLightFetcher(pm *ProtocolManager) *lightFetcher {
f := &lightFetcher{
pm: pm,
chain: pm.blockchain,
odr: pm.odr,
headAnnouncedBy: make(map[common.Hash][]*peer),
deliverChn: make(chan fetchResponse, 100),
requested: make(map[uint64]fetchRequest),
timeoutChn: make(chan uint64),
notifyChn: make(chan bool, 100),
syncDone: make(chan struct{}),
currentTd: big.NewInt(0),
}
go f.syncLoop()
return f
}
func (f *lightFetcher) notify(p *peer, head *announceData) {
var headHash common.Hash
if head == nil {
// initial notify
headHash = p.Head()
} else {
if core.GetTd(f.pm.chainDb, head.Hash, head.Number) != nil {
head.haveHeaders = head.Number
}
//fmt.Println("notify", p.id, head.Number, head.ReorgDepth, head.haveHeaders)
if !p.addNotify(head) {
//fmt.Println("addNotify fail")
f.pm.removePeer(p.id)
}
headHash = head.Hash
}
f.headAnnouncedMu.Lock()
f.headAnnouncedBy[headHash] = append(f.headAnnouncedBy[headHash], p)
f.headAnnouncedMu.Unlock()
f.notifyChn <- true
}
func (f *lightFetcher) gotHeader(header *types.Header) {
f.headAnnouncedMu.Lock()
defer f.headAnnouncedMu.Unlock()
hash := header.Hash()
peerList := f.headAnnouncedBy[hash]
if peerList == nil {
return
}
number := header.GetNumberU64()
td := core.GetTd(f.pm.chainDb, hash, number)
for _, peer := range peerList {
peer.lock.Lock()
ok := peer.gotHeader(hash, number, td)
peer.lock.Unlock()
if !ok {
//fmt.Println("gotHeader fail")
f.pm.removePeer(peer.id)
}
}
delete(f.headAnnouncedBy, hash)
}
func (f *lightFetcher) nextRequest() (*peer, *announceData) {
var bestPeer *peer
bestTd := f.currentTd
for _, peer := range f.pm.peers.AllPeers() {
peer.lock.RLock()
if !peer.headInfo.requested && (peer.headInfo.Td.Cmp(bestTd) > 0 ||
(bestPeer != nil && peer.headInfo.Td.Cmp(bestTd) == 0 && peer.headInfo.haveHeaders > bestPeer.headInfo.haveHeaders)) {
bestPeer = peer
bestTd = peer.headInfo.Td
}
peer.lock.RUnlock()
}
if bestPeer == nil {
return nil, nil
}
bestPeer.lock.Lock()
res := bestPeer.headInfo
res.requested = true
bestPeer.lock.Unlock()
for _, peer := range f.pm.peers.AllPeers() {
if peer != bestPeer {
peer.lock.Lock()
if peer.headInfo.Hash == bestPeer.headInfo.Hash && peer.headInfo.haveHeaders == bestPeer.headInfo.haveHeaders {
peer.headInfo.requested = true
}
peer.lock.Unlock()
}
}
return bestPeer, res
}
func (f *lightFetcher) deliverHeaders(reqID uint64, headers []*types.Header) {
f.deliverChn <- fetchResponse{reqID: reqID, headers: headers}
}
func (f *lightFetcher) requestedID(reqID uint64) bool {
f.reqMu.RLock()
_, ok := f.requested[reqID]
f.reqMu.RUnlock()
return ok
}
func (f *lightFetcher) request(p *peer, block *announceData) {
//fmt.Println("request", p.id, block.Number, block.haveHeaders)
amount := block.Number - block.haveHeaders
if amount == 0 {
return
}
if amount > 100 {
f.syncing = true
go func() {
//fmt.Println("f.pm.synchronise(p)")
f.pm.synchronise(p)
//fmt.Println("sync done")
f.syncDone <- struct{}{}
}()
return
}
reqID := f.odr.getNextReqID()
f.reqMu.Lock()
f.requested[reqID] = fetchRequest{hash: block.Hash, amount: amount, peer: p}
f.reqMu.Unlock()
cost := p.GetRequestCost(GetBlockHeadersMsg, int(amount))
p.fcServer.SendRequest(reqID, cost)
go p.RequestHeadersByHash(reqID, cost, block.Hash, int(amount), 0, true)
go func() {
time.Sleep(hardRequestTimeout)
f.timeoutChn <- reqID
}()
}
func (f *lightFetcher) processResponse(req fetchRequest, resp fetchResponse) bool {
if uint64(len(resp.headers)) != req.amount || resp.headers[0].Hash() != req.hash {
return false
}
headers := make([]*types.Header, req.amount)
for i, header := range resp.headers {
headers[int(req.amount)-1-i] = header
}
if _, err := f.chain.InsertHeaderChain(headers, 1); err != nil {
return false
}
for _, header := range headers {
td := core.GetTd(f.pm.chainDb, header.Hash(), header.GetNumberU64())
if td == nil {
return false
}
if td.Cmp(f.currentTd) > 0 {
f.currentTd = td
}
f.gotHeader(header)
}
return true
}
func (f *lightFetcher) checkSyncedHeaders() {
//fmt.Println("checkSyncedHeaders()")
for _, peer := range f.pm.peers.AllPeers() {
peer.lock.Lock()
h := peer.firstHeadInfo
remove := false
loop:
for h != nil {
if td := core.GetTd(f.pm.chainDb, h.Hash, h.Number); td != nil {
//fmt.Println(" found", h.Number)
ok := peer.gotHeader(h.Hash, h.Number, td)
if !ok {
remove = true
break loop
}
if td.Cmp(f.currentTd) > 0 {
f.currentTd = td
}
}
h = h.next
}
peer.lock.Unlock()
if remove {
//fmt.Println("checkSync fail")
f.pm.removePeer(peer.id)
}
}
}
func (f *lightFetcher) syncLoop() {
f.pm.wg.Add(1)
defer f.pm.wg.Done()
srtoNotify := false
for {
select {
case <-f.pm.quitSync:
return
case ext := <-f.notifyChn:
//fmt.Println("<-f.notifyChn", f.syncing, ext, srtoNotify)
s := srtoNotify
srtoNotify = false
if !f.syncing && !(ext && s) {
if p, r := f.nextRequest(); r != nil {
srtoNotify = true
go func() {
time.Sleep(softRequestTimeout)
f.notifyChn <- false
}()
f.request(p, r)
}
}
case reqID := <-f.timeoutChn:
f.reqMu.Lock()
req, ok := f.requested[reqID]
if ok {
delete(f.requested, reqID)
}
f.reqMu.Unlock()
if ok {
//fmt.Println("hard timeout")
f.pm.removePeer(req.peer.id)
}
case resp := <-f.deliverChn:
//fmt.Println("<-f.deliverChn", f.syncing)
f.reqMu.Lock()
req, ok := f.requested[resp.reqID]
delete(f.requested, resp.reqID)
f.reqMu.Unlock()
if !ok || !(f.syncing || f.processResponse(req, resp)) {
//fmt.Println("processResponse fail")
f.pm.removePeer(req.peer.id)
}
case <-f.syncDone:
//fmt.Println("<-f.syncDone", f.syncing)
f.checkSyncedHeaders()
f.syncing = false
}
}
}
