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
+ }
+}