From 595bb1cbdbbcdc47ca942daf414997587a6ecce2 Mon Sep 17 00:00:00 2001 From: Zsolt Felfoldi Date: Tue, 24 Jan 2017 04:00:53 +0100 Subject: [PATCH] les, light, eth/filter: implement transformed bloom bitmap based quick log filtering --- core/blockchain.go | 4 +- core/bloombits/fetcher_test.go | 74 ++++++ core/bloombits/matcher.go | 465 +++++++++++++++++++++++++++++++++ core/bloombits/utils.go | 165 ++++++++++++ core/database_util.go | 53 ++++ core/headerchain.go | 8 +- core/types/bloom9.go | 13 + eth/api_backend.go | 13 + eth/filters/filter.go | 139 ++++++++-- les/api_backend.go | 5 + les/flowcontrol/control.go | 3 +- les/handler.go | 200 +++++++++++++- les/odr.go | 7 + les/odr_requests.go | 101 ++++++- les/peer.go | 17 +- les/protocol.go | 4 +- les/server.go | 75 ++++-- les/serverpool.go | 2 +- light/lightchain.go | 36 ++- light/odr.go | 20 +- light/odr_util.go | 48 +++- 21 files changed, 1376 insertions(+), 76 deletions(-) create mode 100644 core/bloombits/fetcher_test.go create mode 100644 core/bloombits/matcher.go create mode 100644 core/bloombits/utils.go diff --git a/core/blockchain.go b/core/blockchain.go index 281f28f361..60a278cc14 100644 --- a/core/blockchain.go +++ b/core/blockchain.go @@ -1283,7 +1283,7 @@ func (self *BlockChain) InsertHeaderChain(chain []*types.Header, checkFreq int) self.mu.Lock() defer self.mu.Unlock() - _, err := self.hc.WriteHeader(header) + _, err := self.hc.WriteHeader(header, nil) return err } @@ -1306,7 +1306,7 @@ func (self *BlockChain) writeHeader(header *types.Header) error { self.mu.Lock() defer self.mu.Unlock() - _, err := self.hc.WriteHeader(header) + _, err := self.hc.WriteHeader(header, nil) return err } diff --git a/core/bloombits/fetcher_test.go b/core/bloombits/fetcher_test.go new file mode 100644 index 0000000000..8b99ff814b --- /dev/null +++ b/core/bloombits/fetcher_test.go @@ -0,0 +1,74 @@ +// Copyright 2017 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 bloombits + +import ( + "bytes" + "encoding/binary" + "math/rand" + "testing" + "time" +) + +func testVector(b uint, s uint64) BitVector { + r := make(BitVector, 10) + binary.BigEndian.PutUint16(r[0:2], uint16(b)) + binary.BigEndian.PutUint64(r[2:10], s) + return r +} + +func TestFetcher(t *testing.T) { + f := &fetcher{ + reqMap: make(map[uint64]req), + distChn: make(chan distReq, channelCap), + } + in := make(chan uint64, channelCap) + stop := make(chan struct{}) + out := f.fetch(in, stop) + + for i := 0; i < 10; i++ { + go func() { + for { + req, ok := <-f.distChn + if !ok { + return + } + time.Sleep(time.Duration(rand.Intn(1000000))) + f.deliver([]uint64{req.sectionIdx}, []BitVector{testVector(req.bitIdx, req.sectionIdx)}) + } + }() + } + + go func() { + for i := uint64(0); i < 10000; i++ { + in <- i + } + }() + + for i := uint64(0); i < 10000; i++ { + bv := <-out + if !bytes.Equal(bv, testVector(0, i)) { + if len(bv) != 10 { + t.Errorf("Vector #%d length is %d, expected 10", i, len(bv)) + } else { + j := binary.BigEndian.Uint64(bv[2:10]) + t.Errorf("Expected vector #%d, fetched #%d", i, j) + } + } + } + + close(stop) +} diff --git a/core/bloombits/matcher.go b/core/bloombits/matcher.go new file mode 100644 index 0000000000..ea63be0255 --- /dev/null +++ b/core/bloombits/matcher.go @@ -0,0 +1,465 @@ +// Copyright 2017 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 bloombits + +import ( + "sync" + "time" + + "github.com/ethereum/go-ethereum/common" + "github.com/ethereum/go-ethereum/core/types" +) + +const ( + maxRequestLength = 16 + channelCap = 100 +) + +// when received by fetcher: data == nil, requested == false, fetched == chan struct{} +// when returned by NextRequest: data == nil, requested == true, fetched == chan struct{} +// when data is delivered: data == BitVector, requested == true, fetched == nil +type req struct { + data BitVector + requested bool + fetched chan struct{} +} + +type distReq struct { + bitIdx uint + sectionIdx uint64 +} + +type fetcher struct { + bitIdx uint + reqMap map[uint64]req + reqLock sync.RWMutex +} + +func (f *fetcher) fetch(sectionChn chan uint64, distChn chan distReq, stop chan struct{}) chan BitVector { + dataChn := make(chan BitVector, channelCap) + returnChn := make(chan uint64, channelCap) + + go func() { + defer close(returnChn) + + for { + select { + case <-stop: + return + case idx, ok := <-sectionChn: + if !ok { + return + } + + f.reqLock.Lock() + r := f.reqMap[idx] + if r.data == nil { + if r.fetched == nil { + r.fetched = make(chan struct{}) + } + if !r.requested { + distChn <- distReq{bitIdx: f.bitIdx, sectionIdx: idx} + } + f.reqMap[idx] = r + } + f.reqLock.Unlock() + returnChn <- idx + } + } + }() + + go func() { + defer close(dataChn) + + for { + select { + case <-stop: + return + case idx, ok := <-returnChn: + if !ok { + return + } + + f.reqLock.RLock() + r := f.reqMap[idx] + f.reqLock.RUnlock() + + if r.data == nil { + select { + case <-stop: + return + case <-r.fetched: + f.reqLock.RLock() + r = f.reqMap[idx] + f.reqLock.RUnlock() + } + } + dataChn <- r.data + } + } + }() + + return dataChn +} + +func (f *fetcher) requested(sectionIdxList []uint64) { + //fmt.Println("requested", f.bitIdx, sectionIdxList) + f.reqLock.Lock() + defer f.reqLock.Unlock() + + for _, idx := range sectionIdxList { + r := f.reqMap[idx] + r.requested = true + f.reqMap[idx] = r + } +} + +func (f *fetcher) deliver(sectionIdxList []uint64, data []BitVector) { + //fmt.Println("deliver", f.bitIdx, sectionIdxList, data != nil) + f.reqLock.Lock() + defer f.reqLock.Unlock() + + for i, idx := range sectionIdxList { + r := f.reqMap[idx] + if data != nil { + r.data = data[i] + close(r.fetched) + r.fetched = nil + } else { + r.requested = false + } + f.reqMap[idx] = r + } +} + +type Matcher struct { + addresses []types.BloomIndexList + topics [][]types.BloomIndexList + fetchers map[uint]*fetcher + + distChn chan distReq + getNextReqChn chan chan nextRequests +} + +func NewMatcher() *Matcher { + return &Matcher{fetchers: make(map[uint]*fetcher)} +} + +// SetAddresses matches only logs that are generated from addresses that are included +// in the given addresses. +func (m *Matcher) SetAddresses(addr []common.Address) { + m.addresses = make([]types.BloomIndexList, len(addr)) + for i, b := range addr { + m.addresses[i] = types.BloomIndexes(b.Bytes()) + } +} + +// SetTopics matches only logs that have topics matching the given topics. +func (m *Matcher) SetTopics(topics [][]common.Hash) { + m.topics = nil +loop: + for _, topicList := range topics { + t := make([]types.BloomIndexList, len(topicList)) + for i, b := range topicList { + if (b == common.Hash{}) { + continue loop + } + t[i] = types.BloomIndexes(b.Bytes()) + } + m.topics = append(m.topics, t) + } +} + +func (m *Matcher) match(sectionChn chan uint64, stop chan struct{}) (chan uint64, chan BitVector) { + subIdx := m.topics + if len(m.addresses) > 0 { + subIdx = append([][]types.BloomIndexList{m.addresses}, subIdx...) + } + //fmt.Println("idx", subIdx) + m.distChn = make(chan distReq, channelCap) + m.getNextReqChn = make(chan chan nextRequests) // should be a blocking channel + go m.distributeRequests(stop) + + s := sectionChn + var bv chan BitVector + for _, idx := range subIdx { + s, bv = m.subMatch(s, bv, idx, stop) + } + return s, bv +} + +func (m *Matcher) getOrNewFetcher(idx uint) *fetcher { + if f, ok := m.fetchers[idx]; ok { + return f + } + f := &fetcher{ + bitIdx: idx, + reqMap: make(map[uint64]req), + } + m.fetchers[idx] = f + return f +} + +// andVector == nil +func (m *Matcher) subMatch(sectionChn chan uint64, andVectorChn chan BitVector, idxs []types.BloomIndexList, stop chan struct{}) (chan uint64, chan BitVector) { + // set up fetchers + fetchIdx := make([][3]chan uint64, len(idxs)) + fetchData := make([][3]chan BitVector, len(idxs)) + for i, idx := range idxs { + for j, ii := range idx { + fetchIdx[i][j] = make(chan uint64, channelCap) + fetchData[i][j] = m.getOrNewFetcher(ii).fetch(fetchIdx[i][j], m.distChn, stop) + } + } + + processChn := make(chan uint64, channelCap) + resIdxChn := make(chan uint64, channelCap) + resDataChn := make(chan BitVector, channelCap) + + // goroutine for starting retrievals + go func() { + for { + select { + case <-stop: + return + case s, ok := <-sectionChn: + if !ok { + close(processChn) + for _, ff := range fetchIdx { + for _, f := range ff { + close(f) + } + } + return + } + + processChn <- s + for _, ff := range fetchIdx { + for _, f := range ff { + f <- s + } + } + } + } + }() + + // goroutine for processing retrieved data + go func() { + for { + select { + case <-stop: + return + case s, ok := <-processChn: + if !ok { + close(resIdxChn) + close(resDataChn) + return + } + + var orVector BitVector + for _, ff := range fetchData { + var andVector BitVector + for _, f := range ff { + data := <-f + if andVector == nil { + andVector = bvCopy(data) + } else { + bvAnd(andVector, data) + } + } + if orVector == nil { + orVector = andVector + } else { + bvOr(orVector, andVector) + } + } + + if orVector == nil { + orVector = bvZero() + } + if andVectorChn != nil { + bvAnd(orVector, <-andVectorChn) + } + if bvIsNonZero(orVector) { + resIdxChn <- s + resDataChn <- orVector + } + } + } + }() + + return resIdxChn, resDataChn +} + +func (m *Matcher) GetMatches(start, end uint64, stop chan struct{}) chan uint64 { + sectionChn := make(chan uint64, channelCap) + resultsChn := make(chan uint64, channelCap) + + s, bv := m.match(sectionChn, stop) + + startSection := start / SectionSize + endSection := end / SectionSize + + go func() { + defer close(sectionChn) + + for i := startSection; i <= endSection; i++ { + select { + case sectionChn <- i: + case <-stop: + return + } + } + }() + + go func() { + defer close(resultsChn) + + for { + select { + case idx, ok := <-s: + if !ok { + return + } + match := <-bv //nil check + sectionStart := idx * SectionSize + s := sectionStart + if start > s { + s = start + } + e := sectionStart + SectionSize - 1 + if end < e { + e = end + } + for i := s; i <= e; i++ { + b := match[(i-sectionStart)/8] + bit := 7 - i%8 + if b != 0 { + if b&(1< queue[i] { + i++ + } + reqs[r.bitIdx] = append(append(queue[:i], r.sectionIdx), queue[i:]...) + reqCnt++ + } + + storeReqs := func(r distReq) { + storeReq(r) + timeout := time.After(time.Microsecond) + for { + select { + case <-timeout: + return + case r := <-m.distChn: + storeReq(r) + } + } + } + + for { + if reqCnt == 0 { + select { + case r := <-m.distChn: + storeReqs(r) + case <-stop: + return + } + } else { + select { + case r := <-m.distChn: + storeReqs(r) + case <-stop: + return + case c := <-m.getNextReqChn: + var ( + found bool + bestBit uint + bestSection uint64 + ) + + for bitIdx, queue := range reqs { + if len(queue) > 0 && (!found || queue[0] < bestSection) { + found = true + bestBit = bitIdx + bestSection = queue[0] + } + } + if !found { + panic(nil) + } + + bestQueue := reqs[bestBit] + cnt := len(bestQueue) + if cnt > maxRequestLength { + cnt = maxRequestLength + } + res := nextRequests{bestBit, bestQueue[:cnt]} + reqs[bestBit] = bestQueue[cnt:] + reqCnt -= cnt + + c <- res + } + } + } +} + +func (m *Matcher) NextRequest(stop chan struct{}) (bitIdx uint, sectionIdxList []uint64) { + c := make(chan nextRequests) + select { + case m.getNextReqChn <- c: + r := <-c + //fmt.Println("request", r.bitIdx, r.sectionIdxList) + m.fetchers[r.bitIdx].requested(r.sectionIdxList) + return r.bitIdx, r.sectionIdxList + case <-stop: + return 0, nil + } +} + +// It is possible to deliver data even after GetMatches has been stopped. Once a vector has been +// requested, the next call to GetMatches will keep waiting for delivery. +// If retrieval has been cancelled, call Deliver with data == nil. In this case the next call to +// GetMatches will re-request it. +func (m *Matcher) Deliver(bitIdx uint, sectionIdxList []uint64, data []BitVector) { + m.fetchers[bitIdx].deliver(sectionIdxList, data) +} diff --git a/core/bloombits/utils.go b/core/bloombits/utils.go new file mode 100644 index 0000000000..fd5c03e32d --- /dev/null +++ b/core/bloombits/utils.go @@ -0,0 +1,165 @@ +// Copyright 2017 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 bloombits + +import ( + "github.com/ethereum/go-ethereum/core/types" +) + +const SectionSize = 4096 + +type ( + BitVector []byte + CompVector []byte +) + +func bvAnd(a, b BitVector) { + for i, bb := range b { + a[i] &= bb + } +} + +func bvOr(a, b BitVector) { + for i, bb := range b { + a[i] |= bb + } +} + +func bvZero() BitVector { + return make(BitVector, SectionSize/8) +} + +func bvCopy(a BitVector) BitVector { + c := make(BitVector, SectionSize/8) + copy(c, a) + return c +} + +func bvIsNonZero(a BitVector) bool { + for _, b := range a { + if b != 0 { + return true + } + } + return false +} + +func CompressBloomBits(bits BitVector) CompVector { + if len(bits) != SectionSize/8 { + panic(nil) + } + c := compressBits(bits) + if len(c) >= SectionSize/8 { + // make a copy so that output is always detached from input + return CompVector(bvCopy(bits)) + } + return CompVector(c) +} + +func compressBits(bits []byte) []byte { + l := len(bits) + b := make([]byte, l/8) + c := make([]byte, l) + cl := 0 + for i, v := range bits { + if v != 0 { + c[cl] = v + cl++ + b[i/8] |= 1 << byte(7-i%8) + } + } + if cl == 0 { + return nil + } + if l > 8 { + b = compressBits(b) + } + return append(b, c[0:cl]...) +} + +func DecompressBloomBits(bits CompVector) BitVector { + if len(bits) == SectionSize/8 { + // make a copy so that output is always detached from input + return bvCopy(BitVector(bits)) + } + dc, ofs := decompressBits(bits, SectionSize/8) + if ofs != len(bits) { + panic(nil) + } + return dc +} + +func decompressBits(bits []byte, targetLen int) ([]byte, int) { + lb := len(bits) + dc := make([]byte, targetLen) + if lb == 0 { + return dc, 0 + } + + l := targetLen / 8 + var ( + b []byte + ofs int + ) + if l == 1 { + b = bits[0:1] + ofs = 1 + } else { + b, ofs = decompressBits(bits, l) + } + for i, _ := range dc { + if b[i/8]&(1<= SectionSize { + panic("too many header blooms added") + } + + byteIdx := b.bitIdx / 8 + bitMask := byte(1) << byte(7-b.bitIdx%8) + for bloomBitIdx, _ := range b.blooms { + bloomByteIdx := BloomLength/8 - 1 - bloomBitIdx/8 + bloomBitMask := byte(1) << byte(bloomBitIdx%8) + if (bloom[bloomByteIdx] & bloomBitMask) != 0 { + b.blooms[bloomBitIdx][byteIdx] |= bitMask + } + } + b.bitIdx++ +} + +func (b *BloomBitsCreator) GetBitVector(idx uint) BitVector { + if b.bitIdx != SectionSize { + panic("not enough header blooms added") + } + + return BitVector(b.blooms[idx][:]) +} diff --git a/core/database_util.go b/core/database_util.go index 229f21b5b7..92e52e33bd 100644 --- a/core/database_util.go +++ b/core/database_util.go @@ -26,6 +26,7 @@ import ( "sync" "github.com/ethereum/go-ethereum/common" + "github.com/ethereum/go-ethereum/core/bloombits" "github.com/ethereum/go-ethereum/core/types" "github.com/ethereum/go-ethereum/ethdb" "github.com/ethereum/go-ethereum/logger" @@ -71,6 +72,9 @@ var ( preimageCounter = metrics.NewCounter("db/preimage/total") preimageHitCounter = metrics.NewCounter("db/preimage/hits") + + bloomBitsPrefix = []byte("bloomBits-") + bloomBitsAvailKey = []byte("bloomBitsAvailable") ) // encodeBlockNumber encodes a block number as big endian uint64 @@ -699,3 +703,52 @@ func FindCommonAncestor(db ethdb.Database, a, b *types.Header) *types.Header { } return a } + +func GetBloomBits(db ethdb.Database, bitIdx, sectionIdx uint64) (bloombits.CompVector, error) { + var encKey [10]byte + binary.BigEndian.PutUint16(encKey[0:2], uint16(bitIdx)) + binary.BigEndian.PutUint64(encKey[2:10], sectionIdx) + key := append(bloomBitsPrefix, encKey[:]...) + bloomBits, err := db.Get(key) + return bloombits.CompVector(bloomBits), err +} + +func StoreBloomBits(db ethdb.Database, bitIdx, sectionIdx uint64, bloomBits bloombits.CompVector) { + var encKey [10]byte + binary.BigEndian.PutUint16(encKey[0:2], uint16(bitIdx)) + binary.BigEndian.PutUint64(encKey[2:10], sectionIdx) + key := append(bloomBitsPrefix, encKey[:]...) + db.Put(key, bloomBits) +} + +func GetBloomBitsAvailable(db ethdb.Database) uint64 { + data, _ := db.Get(bloomBitsAvailKey) + if len(data) == 8 { + return binary.BigEndian.Uint64(data[:]) + } + return 0 +} + +func StoreBloomBitsAvailable(db ethdb.Database, cnt uint64) { + var data [8]byte + binary.BigEndian.PutUint64(data[:], cnt) + db.Put(bloomBitsAvailKey, data[:]) +} + +func MakeBloomBitsSection(db ethdb.Database, sectionIdx uint64) error { + bc := &bloombits.BloomBitsCreator{} + for i := sectionIdx * bloombits.SectionSize; i < (sectionIdx+1)*bloombits.SectionSize; i++ { + hash := GetCanonicalHash(db, i) + header := GetHeader(db, hash, i) + if header == nil { + glog.V(logger.Error).Infof("Error creating bloomBits section #%d: header #%d not found", sectionIdx, i) + return errors.New("Header not found") + } + bc.AddHeaderBloom(header.Bloom) + } + + for i := 0; i < bloombits.BloomLength; i++ { + StoreBloomBits(db, uint64(i), sectionIdx, bloombits.CompressBloomBits(bc.GetBitVector(uint(i)))) + } + return nil +} diff --git a/core/headerchain.go b/core/headerchain.go index 1dc189323d..17410cb1a0 100644 --- a/core/headerchain.go +++ b/core/headerchain.go @@ -130,6 +130,8 @@ func (hc *HeaderChain) GetBlockNumber(hash common.Hash) uint64 { return number } +type ReorgCallback func(*types.Header) + // WriteHeader writes a header into the local chain, given that its parent is // already known. If the total difficulty of the newly inserted header becomes // greater than the current known TD, the canonical chain is re-routed. @@ -139,7 +141,7 @@ func (hc *HeaderChain) GetBlockNumber(hash common.Hash) uint64 { // without the real blocks. Hence, writing headers directly should only be done // in two scenarios: pure-header mode of operation (light clients), or properly // separated header/block phases (non-archive clients). -func (hc *HeaderChain) WriteHeader(header *types.Header) (status WriteStatus, err error) { +func (hc *HeaderChain) WriteHeader(header *types.Header, reorgCallback ReorgCallback) (status WriteStatus, err error) { // Cache some values to prevent constant recalculation var ( hash = header.Hash() @@ -187,6 +189,10 @@ func (hc *HeaderChain) WriteHeader(header *types.Header) (status WriteStatus, er headHeader = hc.GetHeader(headHash, headNumber) } + if reorgCallback != nil && headNumber < hc.currentHeader.Number.Uint64() { + reorgCallback(headHeader) + } + // Extend the canonical chain with the new header if err := WriteCanonicalHash(hc.chainDb, hash, number); err != nil { glog.Fatalf("failed to insert header number: %v", err) diff --git a/core/types/bloom9.go b/core/types/bloom9.go index 32aa47a41b..40cda902e6 100644 --- a/core/types/bloom9.go +++ b/core/types/bloom9.go @@ -106,6 +106,19 @@ func LogsBloom(logs []*Log) *big.Int { return bin } +type BloomIndexList [3]uint + +func BloomIndexes(b []byte) BloomIndexList { + b = crypto.Keccak256(b[:]) + + var r [3]uint + for i, _ := range r { + r[i] = (uint(b[i+i+1]) + (uint(b[i+i]) << 8)) & 2047 + } + + return r +} + func bloom9(b []byte) *big.Int { b = crypto.Keccak256(b[:]) diff --git a/eth/api_backend.go b/eth/api_backend.go index 72ed76cc4a..61cb6a71a4 100644 --- a/eth/api_backend.go +++ b/eth/api_backend.go @@ -22,6 +22,7 @@ import ( "github.com/ethereum/go-ethereum/accounts" "github.com/ethereum/go-ethereum/common" "github.com/ethereum/go-ethereum/core" + "github.com/ethereum/go-ethereum/core/bloombits" "github.com/ethereum/go-ethereum/core/state" "github.com/ethereum/go-ethereum/core/types" "github.com/ethereum/go-ethereum/core/vm" @@ -199,6 +200,18 @@ func (b *EthApiBackend) AccountManager() *accounts.Manager { return b.eth.AccountManager() } +func (b *EthApiBackend) GetBloomBits(ctx context.Context, bitIdx uint64, sectionIdxList []uint64) ([]bloombits.CompVector, error) { + results := make([]bloombits.CompVector, len(sectionIdxList)) + var err error + for i, idx := range sectionIdxList { + results[i], err = core.GetBloomBits(b.eth.chainDb, bitIdx, idx) + if err != nil { + return nil, err + } + } + return results, nil +} + type EthApiState struct { state *state.StateDB } diff --git a/eth/filters/filter.go b/eth/filters/filter.go index 9a8e2fd70a..a3afecc0ac 100644 --- a/eth/filters/filter.go +++ b/eth/filters/filter.go @@ -24,6 +24,7 @@ import ( "github.com/ethereum/go-ethereum/common" "github.com/ethereum/go-ethereum/core" + "github.com/ethereum/go-ethereum/core/bloombits" "github.com/ethereum/go-ethereum/core/types" "github.com/ethereum/go-ethereum/ethdb" "github.com/ethereum/go-ethereum/event" @@ -36,12 +37,13 @@ type Backend interface { EventMux() *event.TypeMux HeaderByNumber(ctx context.Context, blockNr rpc.BlockNumber) (*types.Header, error) GetReceipts(ctx context.Context, blockHash common.Hash) (types.Receipts, error) + GetBloomBits(ctx context.Context, bitIdx uint64, sectionIdxList []uint64) ([]bloombits.CompVector, error) } // Filter can be used to retrieve and filter logs. type Filter struct { - backend Backend - useMipMap bool + backend Backend + useMipMap, useBloomBits bool created time.Time @@ -49,6 +51,8 @@ type Filter struct { begin, end int64 addresses []common.Address topics [][]common.Hash + + matcher *bloombits.Matcher } // New creates a new filter which uses a bloom filter on blocks to figure out whether @@ -57,9 +61,11 @@ type Filter struct { // to light clients. func New(backend Backend, useMipMap bool) *Filter { return &Filter{ - backend: backend, - useMipMap: useMipMap, - db: backend.ChainDb(), + backend: backend, + useMipMap: useMipMap, + useBloomBits: !useMipMap, + db: backend.ChainDb(), + matcher: bloombits.NewMatcher(), } } @@ -79,11 +85,17 @@ func (f *Filter) SetEndBlock(end int64) { // in the given addresses. func (f *Filter) SetAddresses(addr []common.Address) { f.addresses = addr + if f.useBloomBits { + f.matcher.SetAddresses(addr) + } } // SetTopics matches only logs that have topics matching the given topics. func (f *Filter) SetTopics(topics [][]common.Hash) { f.topics = topics + if f.useBloomBits { + f.matcher.SetTopics(topics) + } } // FindOnce searches the blockchain for matching log entries, returning @@ -167,7 +179,109 @@ func (f *Filter) mipFind(start, end uint64, depth int) (logs []*types.Log, block return nil, end } +func (f *Filter) serveMatcher(ctx context.Context, stop chan struct{}) chan error { + errChn := make(chan error) + for i := 0; i < 10; i++ { + go func(i int) { + for { + //fmt.Println(i, "NextRequest") + b, s := f.matcher.NextRequest(stop) + //fmt.Println(i, "NextRequest ret", b, s) + if s == nil { + return + } + data, err := f.backend.GetBloomBits(ctx, uint64(b), s) + //fmt.Println(i, "GetBloomBits", len(data), err) + if err != nil { + f.matcher.Deliver(b, s, nil) + errChn <- err + return + } + decomp := make([]bloombits.BitVector, len(data)) + for i, d := range data { + decomp[i] = bloombits.DecompressBloomBits(bloombits.CompVector(d)) + } + //fmt.Println(i, "Deliver") + f.matcher.Deliver(b, s, decomp) + //fmt.Println(i, "Deliver ret") + } + }(i) + } + + return errChn +} + func (f *Filter) getLogs(ctx context.Context, start, end uint64) (logs []*types.Log, blockNumber uint64, err error) { + + checkBlock := func(i uint64, header *types.Header) (logs []*types.Log, blockNumber uint64, err error) { + // Get the logs of the block + receipts, err := f.backend.GetReceipts(ctx, header.Hash()) + if err != nil { + return nil, end, err + } + var unfiltered []*types.Log + for _, receipt := range receipts { + unfiltered = append(unfiltered, ([]*types.Log)(receipt.Logs)...) + } + logs = filterLogs(unfiltered, nil, nil, f.addresses, f.topics) + if len(logs) > 0 { + return logs, i, nil + } + return nil, i, nil + } + + if f.useBloomBits { + haveBloomBitsBefore := core.GetBloomBitsAvailable(f.db) * bloombits.SectionSize + e := end + if haveBloomBitsBefore <= e { + e = haveBloomBitsBefore - 1 + } + + stop := make(chan struct{}) + defer close(stop) + //fmt.Println("GetMatches") + matches := f.matcher.GetMatches(start, e, stop) + //fmt.Println("GetMatches ret") + errChn := f.serveMatcher(ctx, stop) + + loop: + for { + select { + case i, ok := <-matches: + if !ok { + break loop + } + + blockNumber := rpc.BlockNumber(i) + header, err := f.backend.HeaderByNumber(ctx, blockNumber) + if header == nil || err != nil { + return logs, end, err + } + + l, b, e := checkBlock(i, header) + + //fmt.Println("match", i, f.bloomFilter(header.Bloom), len(l)) + /*for i := 0; i < 16; i++ { + fmt.Println(header.Bloom[i*16 : i*16+16]) + }*/ + + if l != nil || e != nil { + return l, b, e + } + case err := <-errChn: + return logs, end, err + case <-ctx.Done(): + return nil, end, ctx.Err() + } + } + + if end < haveBloomBitsBefore { + return logs, end, nil + } else { + start = haveBloomBitsBefore + } + } + for i := start; i <= end; i++ { blockNumber := rpc.BlockNumber(i) header, err := f.backend.HeaderByNumber(ctx, blockNumber) @@ -178,18 +292,9 @@ func (f *Filter) getLogs(ctx context.Context, start, end uint64) (logs []*types. // Use bloom filtering to see if this block is interesting given the // current parameters if f.bloomFilter(header.Bloom) { - // Get the logs of the block - receipts, err := f.backend.GetReceipts(ctx, header.Hash()) - if err != nil { - return nil, end, err - } - var unfiltered []*types.Log - for _, receipt := range receipts { - unfiltered = append(unfiltered, ([]*types.Log)(receipt.Logs)...) - } - logs = filterLogs(unfiltered, nil, nil, f.addresses, f.topics) - if len(logs) > 0 { - return logs, uint64(blockNumber), nil + l, b, e := checkBlock(i, header) + if l != nil || e != nil { + return l, b, e } } } diff --git a/les/api_backend.go b/les/api_backend.go index ed2a7cd13e..902b7d0811 100644 --- a/les/api_backend.go +++ b/les/api_backend.go @@ -22,6 +22,7 @@ import ( "github.com/ethereum/go-ethereum/accounts" "github.com/ethereum/go-ethereum/common" "github.com/ethereum/go-ethereum/core" + "github.com/ethereum/go-ethereum/core/bloombits" "github.com/ethereum/go-ethereum/core/types" "github.com/ethereum/go-ethereum/core/vm" "github.com/ethereum/go-ethereum/eth/downloader" @@ -153,3 +154,7 @@ func (b *LesApiBackend) EventMux() *event.TypeMux { func (b *LesApiBackend) AccountManager() *accounts.Manager { return b.eth.accountManager } + +func (b *LesApiBackend) GetBloomBits(ctx context.Context, bitIdx uint64, sectionIdxList []uint64) ([]bloombits.CompVector, error) { + return light.GetBloomBits(ctx, b.eth.odr, bitIdx, sectionIdxList) +} diff --git a/les/flowcontrol/control.go b/les/flowcontrol/control.go index e45537cf5e..74e4b0a305 100644 --- a/les/flowcontrol/control.go +++ b/les/flowcontrol/control.go @@ -130,6 +130,7 @@ func (peer *ServerNode) recalcBLE(time mclock.AbsTime) { const safetyMargin = time.Millisecond * 200 func (peer *ServerNode) canSend(maxCost uint64) time.Duration { + peer.recalcBLE(mclock.Now()) maxCost += uint64(safetyMargin) * peer.params.MinRecharge / uint64(fcTimeConst) if maxCost > peer.params.BufLimit { maxCost = peer.params.BufLimit @@ -204,13 +205,11 @@ func (peer *ServerNode) SendRequest(reqID, maxCost uint64) { peer.lock.Lock() } - peer.recalcBLE(mclock.Now()) wait := peer.canSend(maxCost) for wait > 0 { peer.lock.Unlock() time.Sleep(wait) peer.lock.Lock() - peer.recalcBLE(mclock.Now()) wait = peer.canSend(maxCost) } peer.assignedRequest = 0 diff --git a/les/handler.go b/les/handler.go index 42a45845d0..9a7deb2b96 100644 --- a/les/handler.go +++ b/les/handler.go @@ -18,6 +18,7 @@ package les import ( + "bytes" "encoding/binary" "errors" "fmt" @@ -28,12 +29,14 @@ import ( "github.com/ethereum/go-ethereum/common" "github.com/ethereum/go-ethereum/core" + "github.com/ethereum/go-ethereum/core/bloombits" "github.com/ethereum/go-ethereum/core/state" "github.com/ethereum/go-ethereum/core/types" "github.com/ethereum/go-ethereum/eth" "github.com/ethereum/go-ethereum/eth/downloader" "github.com/ethereum/go-ethereum/ethdb" "github.com/ethereum/go-ethereum/event" + "github.com/ethereum/go-ethereum/light" "github.com/ethereum/go-ethereum/logger" "github.com/ethereum/go-ethereum/logger/glog" "github.com/ethereum/go-ethereum/p2p" @@ -57,6 +60,7 @@ const ( MaxCodeFetch = 64 // Amount of contract codes to allow fetching per request MaxProofsFetch = 64 // Amount of merkle proofs to be fetched per retrieval request MaxHeaderProofsFetch = 64 // Amount of merkle proofs to be fetched per retrieval request + MaxBloomBitsFetch = 64 // Amount of merkle proofs to be fetched per retrieval request MaxTxSend = 64 // Amount of transactions to be send per request disableClientRemovePeer = false @@ -121,6 +125,11 @@ type ProtocolManager struct { syncing bool syncDone chan struct{} + bloomBitsUpdateChn chan uint64 + bloomBitsMu sync.Mutex + bloomBitsCalcValid bool + bloomBitsCalcIdx uint64 + // wait group is used for graceful shutdowns during downloading // and processing wg sync.WaitGroup @@ -131,19 +140,20 @@ type ProtocolManager struct { func NewProtocolManager(chainConfig *params.ChainConfig, lightSync bool, networkId int, mux *event.TypeMux, pow pow.PoW, blockchain BlockChain, txpool txPool, chainDb ethdb.Database, odr *LesOdr, txrelay *LesTxRelay) (*ProtocolManager, error) { // Create the protocol manager with the base fields manager := &ProtocolManager{ - lightSync: lightSync, - eventMux: mux, - blockchain: blockchain, - chainConfig: chainConfig, - chainDb: chainDb, - networkId: networkId, - txpool: txpool, - txrelay: txrelay, - odr: odr, - peers: newPeerSet(), - newPeerCh: make(chan *peer), - quitSync: make(chan struct{}), - noMorePeers: make(chan struct{}), + lightSync: lightSync, + eventMux: mux, + blockchain: blockchain, + chainConfig: chainConfig, + chainDb: chainDb, + networkId: networkId, + txpool: txpool, + txrelay: txrelay, + odr: odr, + peers: newPeerSet(), + newPeerCh: make(chan *peer), + quitSync: make(chan struct{}), + noMorePeers: make(chan struct{}), + bloomBitsUpdateChn: make(chan uint64, 100), } // Initiate a sub-protocol for every implemented version we can handle manager.SubProtocols = make([]p2p.Protocol, 0, len(ProtocolVersions)) @@ -203,6 +213,9 @@ func NewProtocolManager(chainConfig *params.ChainConfig, lightSync bool, network manager.downloader = downloader.New(downloader.LightSync, chainDb, manager.eventMux, blockchain.HasHeader, nil, blockchain.GetHeaderByHash, nil, blockchain.CurrentHeader, nil, nil, nil, blockchain.GetTdByHash, blockchain.InsertHeaderChain, nil, nil, blockchain.Rollback, removePeer) + + blockchain.(*light.LightChain).AddNewHeadCallback(manager.newHeadCallback) + go manager.bloomBitsUpdateLoop() } if odr != nil { @@ -220,7 +233,83 @@ func NewProtocolManager(chainConfig *params.ChainConfig, lightSync bool, network return manager, nil } +func (pm *ProtocolManager) bloomBitsUpdateLoop() { + tryUpdate := make(chan struct{}, 1) + updating := false + var targetSectionCnt uint64 + + for { + select { + case <-pm.quitSync: + return + case targetSectionCnt = <-pm.bloomBitsUpdateChn: + if !updating { + updating = true + tryUpdate <- struct{}{} + } + case <-tryUpdate: + pm.bloomBitsMu.Lock() + sectionIdx := core.GetBloomBitsAvailable(pm.chainDb) + if targetSectionCnt > sectionIdx { + pm.bloomBitsCalcValid = true + pm.bloomBitsCalcIdx = sectionIdx + + pm.bloomBitsMu.Unlock() + err := core.MakeBloomBitsSection(pm.chainDb, sectionIdx) + pm.bloomBitsMu.Lock() + + if err == nil && pm.bloomBitsCalcValid { + glog.V(logger.Info).Infof("Stored bloomBits section #%d", sectionIdx) + sectionIdx++ + core.StoreBloomBitsAvailable(pm.chainDb, sectionIdx) + } else { + // unsuccessful bloomBits calculation may happen because of a reorg + glog.V(logger.Info).Infof("Error calculating bloomBits section #%d: %v valid: %v", sectionIdx, err, pm.bloomBitsCalcValid) + } + pm.bloomBitsCalcValid = false + } + pm.bloomBitsMu.Unlock() + + if targetSectionCnt > sectionIdx { + go func() { + time.Sleep(time.Millisecond * 100) + tryUpdate <- struct{}{} + }() + } else { + updating = false + } + } + } +} + +const bloomBitsConfirmations = 200 + +func (pm *ProtocolManager) newHeadCallback(head *types.Header, rollback bool) { + pm.bloomBitsMu.Lock() + defer pm.bloomBitsMu.Unlock() + + headNum := head.Number.Uint64() + rbSectionCnt := headNum / bloombits.SectionSize + var newSectionCnt uint64 + if headNum >= bloomBitsConfirmations-1 { + newSectionCnt = (headNum + 1 - bloomBitsConfirmations) / bloombits.SectionSize + } + lastSectionCnt := core.GetBloomBitsAvailable(pm.chainDb) + + if rbSectionCnt <= pm.bloomBitsCalcIdx { + pm.bloomBitsCalcValid = false + } + if rbSectionCnt < lastSectionCnt { + core.StoreBloomBitsAvailable(pm.chainDb, rbSectionCnt) + pm.bloomBitsUpdateChn <- rbSectionCnt + } + if newSectionCnt > lastSectionCnt { + pm.bloomBitsUpdateChn <- newSectionCnt + } +} + func (pm *ProtocolManager) removePeer(id string) { + fmt.Println("removePeer") // Short circuit if the peer was already removed peer := pm.peers.Peer(id) if peer == nil { @@ -396,7 +485,7 @@ func (pm *ProtocolManager) handle(p *peer) error { } } -var reqList = []uint64{GetBlockHeadersMsg, GetBlockBodiesMsg, GetCodeMsg, GetReceiptsMsg, GetProofsMsg, SendTxMsg, GetHeaderProofsMsg} +var reqList = []uint64{GetBlockHeadersMsg, GetBlockBodiesMsg, GetCodeMsg, GetReceiptsMsg, GetProofsMsg, SendTxMsg, GetHeaderProofsMsg, GetBloomBitsMsg} // handleMsg is invoked whenever an inbound message is received from a remote // peer. The remote connection is torn down upon returning any error. @@ -863,6 +952,89 @@ func (pm *ProtocolManager) handleMsg(p *peer) error { Obj: resp.Data, } + case GetBloomBitsMsg: + glog.V(logger.Debug).Infof("<=== GetBloomBitsMsg from peer %v", p.id) + // Decode the retrieval message + var req struct { + ReqID uint64 + Reqs []BloomReq + } + if err := msg.Decode(&req); err != nil { + return errResp(ErrDecode, "msg %v: %v", msg, err) + } + // Gather state data until the fetch or network limits is reached + var ( + byteCnt int + proofs []BloomResp + ) + reqCnt := len(req.Reqs) + if reject(uint64(reqCnt), MaxBloomBitsFetch) { + return errResp(ErrRequestRejected, "") + } + var ( + lastRoot common.Hash + tr *trie.Trie + lastProof []rlp.RawValue + ) + for _, req := range req.Reqs { + if byteCnt >= softResponseLimit { + break + } + + if root := getChtRoot(pm.chainDb, req.ChtNum); root != (common.Hash{}) { + if root != lastRoot { + tr, _ = trie.New(root, pm.chainDb) + lastRoot = root + } + if tr != nil { + var encNumber [10]byte + binary.BigEndian.PutUint16(encNumber[0:2], uint16(req.BitIdx)) + binary.BigEndian.PutUint64(encNumber[2:10], req.SectionIdx) + proof := tr.Prove(append(bloomBitsTriePrefix, encNumber[:]...)) + if lastProof != nil { + fullProof := make([]rlp.RawValue, len(proof)) + copy(fullProof, proof) + loop: + for i, data := range proof { + if i < len(lastProof) && bytes.Equal(lastProof[i], data) { + proof[i] = []byte{0} + } else { + break loop + } + } + lastProof = fullProof + } else { + lastProof = proof + } + proofs = append(proofs, BloomResp{Proof: proof}) + byteCnt += len(proof) + } + } + } + bv, rcost := p.fcClient.RequestProcessed(costs.baseCost + uint64(reqCnt)*costs.reqCost) + pm.server.fcCostStats.update(msg.Code, uint64(reqCnt), rcost) + return p.SendBloomBits(req.ReqID, bv, proofs) + + case BloomBitsMsg: + if pm.odr == nil { + return errResp(ErrUnexpectedResponse, "") + } + + glog.V(logger.Debug).Infof("<=== BloomBitsMsg from peer %v", p.id) + var resp struct { + ReqID, BV uint64 + Data []BloomResp + } + if err := msg.Decode(&resp); err != nil { + return errResp(ErrDecode, "msg %v: %v", msg, err) + } + p.fcServer.GotReply(resp.ReqID, resp.BV) + deliverMsg = &Msg{ + MsgType: MsgBloomBits, + ReqID: resp.ReqID, + Obj: resp.Data, + } + case SendTxMsg: if pm.txpool == nil { return errResp(ErrUnexpectedResponse, "") diff --git a/les/odr.go b/les/odr.go index 88c7d85a59..7193539d07 100644 --- a/les/odr.go +++ b/les/odr.go @@ -88,6 +88,7 @@ const ( MsgReceipts MsgProofs MsgHeaderProofs + MsgBloomBits ) // Msg encodes a LES message that delivers reply data for a request @@ -189,12 +190,14 @@ func (self *LesOdr) networkRequest(ctx context.Context, lreq LesOdrRequest) erro for { var p *peer if self.serverPool != nil { + //fmt.Println("waiting for selection") p = self.serverPool.selectPeerWait(reqID, func(p *peer) (bool, time.Duration) { if _, ok := exclude[p]; ok || !lreq.CanSend(p) { return false, 0 } return true, p.fcServer.CanSend(lreq.GetCost(p)) }, ctx.Done()) + //fmt.Println("selected", p) } if p == nil { select { @@ -203,6 +206,8 @@ func (self *LesOdr) networkRequest(ctx context.Context, lreq LesOdrRequest) erro case <-req.answered: return nil case <-time.After(retryPeers): + // exclude = make(map[*peer]struct{}) ? + //fmt.Println("retryPeers") } } else { exclude[p] = struct{}{} @@ -213,7 +218,9 @@ func (self *LesOdr) networkRequest(ctx context.Context, lreq LesOdrRequest) erro req.lock.Unlock() reqWg.Add(1) cost := lreq.GetCost(p) + //fmt.Println("waiting to send") p.fcServer.SendRequest(reqID, cost) + //fmt.Println("sending") go self.requestPeer(req, p, delivered, timeout, reqWg) lreq.Request(reqID, p) diff --git a/les/odr_requests.go b/les/odr_requests.go index 2987eb297a..15b05cd11b 100644 --- a/les/odr_requests.go +++ b/les/odr_requests.go @@ -53,6 +53,8 @@ func LesRequest(req light.OdrRequest) LesOdrRequest { return (*CodeRequest)(r) case *light.ChtRequest: return (*ChtRequest)(r) + case *light.BloomRequest: + return (*BloomRequest)(r) default: return nil } @@ -190,12 +192,12 @@ func (self *TrieRequest) CanSend(peer *peer) bool { // Request sends an ODR request to the LES network (implementation of LesOdrRequest) func (self *TrieRequest) Request(reqID uint64, peer *peer) error { glog.V(logger.Debug).Infof("ODR: requesting trie root %08x key %08x from peer %v", self.Id.Root[:4], self.Key[:4], peer.id) - req := &ProofReq{ + req := ProofReq{ BHash: self.Id.BlockHash, AccKey: self.Id.AccKey, Key: self.Key, } - return peer.RequestProofs(reqID, self.GetCost(peer), []*ProofReq{req}) + return peer.RequestProofs(reqID, self.GetCost(peer), []ProofReq{req}) } // Valid processes an ODR request reply message from the LES network @@ -245,11 +247,11 @@ func (self *CodeRequest) CanSend(peer *peer) bool { // Request sends an ODR request to the LES network (implementation of LesOdrRequest) func (self *CodeRequest) Request(reqID uint64, peer *peer) error { glog.V(logger.Debug).Infof("ODR: requesting node data for hash %08x from peer %v", self.Hash[:4], peer.id) - req := &CodeReq{ + req := CodeReq{ BHash: self.Id.BlockHash, AccKey: self.Id.AccKey, } - return peer.RequestCode(reqID, self.GetCost(peer), []*CodeReq{req}) + return peer.RequestCode(reqID, self.GetCost(peer), []CodeReq{req}) } // Valid processes an ODR request reply message from the LES network @@ -305,11 +307,11 @@ func (self *ChtRequest) CanSend(peer *peer) bool { // Request sends an ODR request to the LES network (implementation of LesOdrRequest) func (self *ChtRequest) Request(reqID uint64, peer *peer) error { glog.V(logger.Debug).Infof("ODR: requesting CHT #%d block #%d from peer %v", self.ChtNum, self.BlockNum, peer.id) - req := &ChtReq{ + req := ChtReq{ ChtNum: self.ChtNum, BlockNum: self.BlockNum, } - return peer.RequestHeaderProofs(reqID, self.GetCost(peer), []*ChtReq{req}) + return peer.RequestHeaderProofs(reqID, self.GetCost(peer), []ChtReq{req}) } // Valid processes an ODR request reply message from the LES network @@ -351,3 +353,90 @@ func (self *ChtRequest) Valid(db ethdb.Database, msg *Msg) bool { glog.V(logger.Debug).Infof("ODR: validation successful") return true } + +type BloomReq struct { + ChtNum, BitIdx, SectionIdx, FromLevel uint64 +} + +type BloomResp struct { + Proof []rlp.RawValue +} + +// 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 +// peer's cost table (implementation of LesOdrRequest) +func (self *BloomRequest) GetCost(peer *peer) uint64 { + return peer.GetRequestCost(GetBloomBitsMsg, len(self.SectionIdxList)) +} + +// CanSend tells if a certain peer is suitable for serving the given request +func (self *BloomRequest) CanSend(peer *peer) bool { + peer.lock.RLock() + defer peer.lock.RUnlock() + + return self.ChtNum <= (peer.headInfo.Number-light.ChtConfirmations)/light.ChtFrequency +} + +// Request sends an ODR request to the LES network (implementation of LesOdrRequest) +func (self *BloomRequest) Request(reqID uint64, peer *peer) error { + glog.V(logger.Debug).Infof("ODR: requesting CHT #%d bloom bit #%d section #%d from peer %v", self.ChtNum, self.BitIdx, self.SectionIdxList[0], peer.id) + reqs := make([]BloomReq, len(self.SectionIdxList)) + for i, sectionIdx := range self.SectionIdxList { + reqs[i] = BloomReq{ + ChtNum: self.ChtNum, + BitIdx: self.BitIdx, + SectionIdx: sectionIdx, + } + } + return peer.RequestBloomBits(reqID, self.GetCost(peer), reqs) +} + +// Valid 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 (self *BloomRequest) Valid(db ethdb.Database, msg *Msg) bool { + glog.V(logger.Debug).Infof("ODR: validating CHT #%d bloom bit #%d section #%d", self.ChtNum, self.BitIdx, self.SectionIdxList[0]) + + if msg.MsgType != MsgBloomBits { + glog.V(logger.Debug).Infof("ODR: invalid message type") + return false + } + proofs := msg.Obj.([]BloomResp) + if len(proofs) != len(self.SectionIdxList) { + glog.V(logger.Debug).Infof("ODR: invalid number of entries: %d", len(proofs)) + return false + } + self.Proofs = make([][]rlp.RawValue, len(self.SectionIdxList)) + self.BloomBits = make([][]byte, len(self.SectionIdxList)) + + var encNumber [10]byte + binary.BigEndian.PutUint16(encNumber[0:2], uint16(self.BitIdx)) + var lastProof []rlp.RawValue + //fmt.Println("validating") + for i, proof := range proofs { + //fmt.Println("section", self.SectionIdxList[i], "proof len", len(proof.Proof)) + for i, data := range proof.Proof { + if len(data) == 1 && data[0] == 0 { + if i < len(lastProof) { + //fmt.Println("copying", i) + proof.Proof[i] = lastProof[i] + } else { + return false + } + } + } + lastProof = proof.Proof + binary.BigEndian.PutUint64(encNumber[2:10], self.SectionIdxList[i]) + value, err := trie.VerifyProof(self.ChtRoot, append(bloomBitsTriePrefix, encNumber[:]...), proof.Proof) + if err != nil { + glog.V(logger.Debug).Infof("ODR: CHT merkle proof verification error: %v", err) + return false + } + self.Proofs[i] = proof.Proof + self.BloomBits[i] = value + } + glog.V(logger.Debug).Infof("ODR: validation successful") + return true +} diff --git a/les/peer.go b/les/peer.go index d5008ded11..260052aa10 100644 --- a/les/peer.go +++ b/les/peer.go @@ -193,6 +193,11 @@ func (p *peer) SendHeaderProofs(reqID, bv uint64, proofs []ChtResp) error { return sendResponse(p.rw, HeaderProofsMsg, reqID, bv, proofs) } +// SendBloomBits sends a batch of bloom proofs, corresponding to the ones requested. +func (p *peer) SendBloomBits(reqID, bv uint64, proofs []BloomResp) error { + return sendResponse(p.rw, BloomBitsMsg, reqID, bv, proofs) +} + // RequestHeadersByHash fetches a batch of blocks' headers corresponding to the // specified header query, based on the hash of an origin block. func (p *peer) RequestHeadersByHash(reqID, cost uint64, origin common.Hash, amount int, skip int, reverse bool) error { @@ -216,7 +221,7 @@ func (p *peer) RequestBodies(reqID, cost uint64, hashes []common.Hash) error { // RequestCode fetches a batch of arbitrary data from a node's known state // data, corresponding to the specified hashes. -func (p *peer) RequestCode(reqID, cost uint64, reqs []*CodeReq) error { +func (p *peer) RequestCode(reqID, cost uint64, reqs []CodeReq) error { glog.V(logger.Debug).Infof("%v fetching %v state data", p, len(reqs)) return sendRequest(p.rw, GetCodeMsg, reqID, cost, reqs) } @@ -228,17 +233,23 @@ func (p *peer) RequestReceipts(reqID, cost uint64, hashes []common.Hash) error { } // RequestProofs fetches a batch of merkle proofs from a remote node. -func (p *peer) RequestProofs(reqID, cost uint64, reqs []*ProofReq) error { +func (p *peer) RequestProofs(reqID, cost uint64, reqs []ProofReq) error { glog.V(logger.Debug).Infof("%v fetching %v proofs", p, len(reqs)) return sendRequest(p.rw, GetProofsMsg, reqID, cost, reqs) } // RequestHeaderProofs fetches a batch of header merkle proofs from a remote node. -func (p *peer) RequestHeaderProofs(reqID, cost uint64, reqs []*ChtReq) error { +func (p *peer) RequestHeaderProofs(reqID, cost uint64, reqs []ChtReq) error { glog.V(logger.Debug).Infof("%v fetching %v header proofs", p, len(reqs)) return sendRequest(p.rw, GetHeaderProofsMsg, reqID, cost, reqs) } +// RequestBloomBits fetches a batch of bloom merkle proofs from a remote node. +func (p *peer) RequestBloomBits(reqID, cost uint64, reqs []BloomReq) error { + glog.V(logger.Debug).Infof("%v fetching %v bloom proofs", p, len(reqs)) + return sendRequest(p.rw, GetBloomBitsMsg, reqID, cost, reqs) +} + func (p *peer) SendTxs(cost uint64, txs types.Transactions) error { glog.V(logger.Debug).Infof("%v relaying %v txs", p, len(txs)) reqID := getNextReqID() diff --git a/les/protocol.go b/les/protocol.go index 46da2b8c8b..b2a448f16b 100644 --- a/les/protocol.go +++ b/les/protocol.go @@ -36,7 +36,7 @@ const ( var ProtocolVersions = []uint{lpv1} // Number of implemented message corresponding to different protocol versions. -var ProtocolLengths = []uint64{15} +var ProtocolLengths = []uint64{17} const ( NetworkId = 1 @@ -61,6 +61,8 @@ const ( SendTxMsg = 0x0c GetHeaderProofsMsg = 0x0d HeaderProofsMsg = 0x0e + GetBloomBitsMsg = 0x0f + BloomBitsMsg = 0x10 ) type errCode int diff --git a/les/server.go b/les/server.go index c4c6fcab52..4da3830c81 100644 --- a/les/server.go +++ b/les/server.go @@ -25,6 +25,7 @@ import ( "github.com/ethereum/go-ethereum/common" "github.com/ethereum/go-ethereum/core" + "github.com/ethereum/go-ethereum/core/bloombits" "github.com/ethereum/go-ethereum/core/types" "github.com/ethereum/go-ethereum/eth" "github.com/ethereum/go-ethereum/ethdb" @@ -311,7 +312,7 @@ func (pm *ProtocolManager) blockLoop() { more := makeCht(pm.chainDb) mu.Unlock() if more { - time.Sleep(time.Millisecond * 10) + time.Sleep(time.Millisecond * 100) newCht <- struct{}{} } }() @@ -325,8 +326,8 @@ func (pm *ProtocolManager) blockLoop() { } var ( - lastChtKey = []byte("LastChtNumber") // chtNum (uint64 big endian) - chtPrefix = []byte("cht") // chtPrefix + chtNum (uint64 big endian) -> trie root hash + lastChtKey = []byte("LastChtNumber6") // chtNum (uint64 big endian) + chtPrefix = []byte("cht") // chtPrefix + chtNum (uint64 big endian) -> trie root hash ) func getChtRoot(db ethdb.Database, num uint64) common.Hash { @@ -342,6 +343,8 @@ func storeChtRoot(db ethdb.Database, num uint64, root common.Hash) { db.Put(append(chtPrefix, encNumber[:]...), root[:]) } +var bloomBitsTriePrefix = []byte("bloom") + func makeCht(db ethdb.Database) bool { headHash := core.GetHeadBlockHash(db) headNum := core.GetBlockNumber(db, headHash) @@ -372,31 +375,63 @@ func makeCht(db ethdb.Database) bool { t, _ = trie.New(common.Hash{}, db) } - for num := lastChtNum * light.ChtFrequency; num < (lastChtNum+1)*light.ChtFrequency; num++ { - hash := core.GetCanonicalHash(db, num) - if hash == (common.Hash{}) { - panic("Canonical hash not found") + var compSize, decompSize uint64 +loop: + for newChtNum > lastChtNum { + bloomBitsCreator := &bloombits.BloomBitsCreator{} + + for num := lastChtNum * light.ChtFrequency; num < (lastChtNum+1)*light.ChtFrequency; num++ { + + hash := core.GetCanonicalHash(db, num) + if hash == (common.Hash{}) { + return false + } + td := core.GetTd(db, hash, num) + if td == nil { + return false + } + var encNumber [8]byte + binary.BigEndian.PutUint64(encNumber[:], num) + var node light.ChtNode + node.Hash = hash + node.Td = td + data, _ := rlp.EncodeToBytes(node) + t.Update(encNumber[:], data) + + header := core.GetHeader(db, hash, num) + if header == nil { + return false + } + + bloomBitsCreator.AddHeaderBloom(header.Bloom) } - td := core.GetTd(db, hash, num) - if td == nil { - panic("TD not found") + + for i := uint(0); i < bloombits.BloomLength; i++ { + var encKey [10]byte + binary.BigEndian.PutUint16(encKey[0:2], uint16(i)) + binary.BigEndian.PutUint64(encKey[2:10], lastChtNum) + key := append(bloomBitsTriePrefix, encKey[:]...) + data := bloombits.CompressBloomBits(bloomBitsCreator.GetBitVector(i)) + decompSize += bloombits.SectionSize / 8 + compSize += uint64(len(data)) + if len(data) > 0 { + t.Update(key, data) + } else { + t.Delete(key) + } + } + lastChtNum++ + + if lastChtNum%16 == 0 { + break loop } - var encNumber [8]byte - binary.BigEndian.PutUint64(encNumber[:], num) - var node light.ChtNode - node.Hash = hash - node.Td = td - data, _ := rlp.EncodeToBytes(node) - t.Update(encNumber[:], data) } root, err := t.Commit() if err != nil { lastChtNum = 0 } else { - lastChtNum++ - - glog.V(logger.Detail).Infof("cht: %d %064x", lastChtNum, root) + glog.V(logger.Info).Infof("Storing CHT #%d root hash: %064x compression ratio: %f", lastChtNum, root, float64(compSize)/float64(decompSize)) storeChtRoot(db, lastChtNum, root) var data [8]byte diff --git a/les/serverpool.go b/les/serverpool.go index 9735a718e5..e17d6d13cb 100644 --- a/les/serverpool.go +++ b/les/serverpool.go @@ -333,7 +333,7 @@ func (pool *serverPool) selectPeer(reqID uint64, canSend func(*peer) (bool, time func (pool *serverPool) selectPeerWait(reqID uint64, canSend func(*peer) (bool, time.Duration), abort <-chan struct{}) *peer { for { peer, wait, locked := pool.selectPeer(reqID, canSend) - if locked { + if locked || peer == nil { return peer } select { diff --git a/light/lightchain.go b/light/lightchain.go index 0d28ad2f45..87894f1bfd 100644 --- a/light/lightchain.go +++ b/light/lightchain.go @@ -65,6 +65,8 @@ type LightChain struct { procInterrupt int32 // interrupt signaler for block processing wg sync.WaitGroup + newHeadCallback newHeadCallback + pow pow.PoW validator core.HeaderValidator } @@ -102,14 +104,15 @@ func NewLightChain(odr OdrBackend, config *params.ChainConfig, pow pow.PoW, mux return nil, err } glog.V(logger.Info).Infoln("WARNING: Wrote default ethereum genesis block") + core.StoreBloomBitsAvailable(odr.Database(), 0) } if bc.genesisBlock.Hash() == (common.Hash{212, 229, 103, 64, 248, 118, 174, 248, 192, 16, 184, 106, 64, 213, 245, 103, 69, 161, 24, 208, 144, 106, 52, 230, 154, 236, 140, 13, 177, 203, 143, 163}) { // add trusted CHT if config.DAOForkSupport { WriteTrustedCht(bc.chainDb, TrustedCht{ - Number: 637, - Root: common.HexToHash("01e408d9b1942f05dba1a879f3eaafe34d219edaeb8223fecf1244cc023d3e23"), + Number: 752, + Root: common.HexToHash("4d2e4d9ce20626b62d6fe7598fffd8f63175c8cc2a60dc4211cf6d88f52a78e8"), }) } else { WriteTrustedCht(bc.chainDb, TrustedCht{ @@ -145,6 +148,15 @@ func NewLightChain(odr OdrBackend, config *params.ChainConfig, pow pow.PoW, mux return bc, nil } +type newHeadCallback func(*types.Header, bool) + +func (self *LightChain) AddNewHeadCallback(cb newHeadCallback) { + self.chainmu.Lock() + self.newHeadCallback = cb + cb(self.hc.CurrentHeader(), false) + self.chainmu.Unlock() +} + func (self *LightChain) getProcInterrupt() bool { return atomic.LoadInt32(&self.procInterrupt) == 1 } @@ -355,13 +367,20 @@ func (self *LightChain) Rollback(chain []common.Hash) { self.mu.Lock() defer self.mu.Unlock() + var rollbackHead *types.Header + for i := len(chain) - 1; i >= 0; i-- { hash := chain[i] if head := self.hc.CurrentHeader(); head.Hash() == hash { - self.hc.SetCurrentHeader(self.GetHeader(head.ParentHash, head.Number.Uint64()-1)) + rollbackHead = self.GetHeader(head.ParentHash, head.Number.Uint64()-1) + self.hc.SetCurrentHeader(rollbackHead) } } + + if rollbackHead != nil && self.newHeadCallback != nil { + self.newHeadCallback(rollbackHead, true) + } } // postChainEvents iterates over the events generated by a chain insertion and @@ -402,7 +421,11 @@ func (self *LightChain) InsertHeaderChain(chain []*types.Header, checkFreq int) self.mu.Lock() defer self.mu.Unlock() - status, err := self.hc.WriteHeader(header) + status, err := self.hc.WriteHeader(header, func(head *types.Header) { + if self.newHeadCallback != nil { + self.newHeadCallback(head, true) + } + }) switch status { case core.CanonStatTy: @@ -424,6 +447,11 @@ func (self *LightChain) InsertHeaderChain(chain []*types.Header, checkFreq int) return err } i, err := self.hc.InsertHeaderChain(chain, checkFreq, whFunc) + + if self.newHeadCallback != nil { + self.newHeadCallback(self.hc.CurrentHeader(), false) + } + go self.postChainEvents(events) return i, err } diff --git a/light/odr.go b/light/odr.go index 4f6ef6b9ed..159fc542ef 100644 --- a/light/odr.go +++ b/light/odr.go @@ -138,7 +138,7 @@ func (req *ReceiptsRequest) StoreResult(db ethdb.Database) { core.WriteBlockReceipts(db, req.Hash, req.Number, req.Receipts) } -// TrieRequest is the ODR request type for state/storage trie entries +// ChtRequest is the ODR request type for retrieving old headers from a CHT structure type ChtRequest struct { OdrRequest ChtNum, BlockNum uint64 @@ -155,5 +155,21 @@ func (req *ChtRequest) StoreResult(db ethdb.Database) { hash, num := req.Header.Hash(), req.Header.Number.Uint64() core.WriteTd(db, hash, num, req.Td) core.WriteCanonicalHash(db, hash, num) - //storeProof(db, req.Proof) +} + +// BloomRequest is the ODR request type for retrieving bloom filters from a CHT structure +type BloomRequest struct { + OdrRequest + ChtNum, BitIdx uint64 + SectionIdxList []uint64 + ChtRoot common.Hash + BloomBits [][]byte + Proofs [][]rlp.RawValue +} + +// StoreResult stores the retrieved data in local database +func (req *BloomRequest) StoreResult(db ethdb.Database) { + for i, sectionIdx := range req.SectionIdxList { + core.StoreBloomBits(db, req.BitIdx, sectionIdx, req.BloomBits[i]) + } } diff --git a/light/odr_util.go b/light/odr_util.go index 7617116212..40b820262c 100644 --- a/light/odr_util.go +++ b/light/odr_util.go @@ -23,6 +23,7 @@ import ( "github.com/ethereum/go-ethereum/common" "github.com/ethereum/go-ethereum/core" + "github.com/ethereum/go-ethereum/core/bloombits" "github.com/ethereum/go-ethereum/core/types" "github.com/ethereum/go-ethereum/crypto" "github.com/ethereum/go-ethereum/ethdb" @@ -37,10 +38,12 @@ var sha3_nil = crypto.Keccak256Hash(nil) var ( ErrNoTrustedCht = errors.New("No trusted canonical hash trie") ErrNoHeader = errors.New("Header not found") + trustedChtKey = []byte("TrustedCHT") +) - ChtFrequency = uint64(4096) - ChtConfirmations = uint64(2048) - trustedChtKey = []byte("TrustedCHT") +const ( + ChtFrequency = 4096 + ChtConfirmations = 2048 ) type ChtNode struct { @@ -65,6 +68,10 @@ func GetTrustedCht(db ethdb.Database) TrustedCht { func WriteTrustedCht(db ethdb.Database, cht TrustedCht) { data, _ := rlp.EncodeToBytes(cht) db.Put(trustedChtKey, data) + b := cht.Number * ChtFrequency / bloombits.SectionSize + if core.GetBloomBitsAvailable(db) < b { + core.StoreBloomBitsAvailable(db, b) + } } func DeleteTrustedCht(db ethdb.Database) { @@ -185,3 +192,38 @@ func GetBlockReceipts(ctx context.Context, odr OdrBackend, hash common.Hash, num return r.Receipts, nil } } + +func GetBloomBits(ctx context.Context, odr OdrBackend, bitIdx uint64, sectionIdxList []uint64) ([]bloombits.CompVector, error) { + result := make([]bloombits.CompVector, len(sectionIdxList)) + var ( + reqList []uint64 + reqIdx []int + ) + cht := GetTrustedCht(odr.Database()) + + for i, sectionIdx := range sectionIdxList { + bloomBits, err := core.GetBloomBits(odr.Database(), bitIdx, sectionIdx) + if err == nil { + result[i] = bloomBits + } else { + if sectionIdx >= cht.Number { + return nil, ErrNoTrustedCht + } + reqList = append(reqList, sectionIdx) + reqIdx = append(reqIdx, i) + } + } + if reqList == nil { + return result, nil + } + + r := &BloomRequest{ChtRoot: cht.Root, ChtNum: cht.Number, BitIdx: bitIdx, SectionIdxList: reqList} + if err := odr.Retrieve(ctx, r); err != nil { + return nil, err + } else { + for i, idx := range reqIdx { + result[idx] = r.BloomBits[i] + } + return result, nil + } +}