les, light, eth/filter: implement transformed bloom bitmap based quick log filtering

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
 }
 

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 <http://www.gnu.org/licenses/>.
+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)
+}

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 <http://www.gnu.org/licenses/>.
+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<<bit) != 0 {
+							resultsChn <- i
+						}
+					} else {
+						i += bit
+					}
+				}
+
+			case <-stop:
+				return
+			}
+		}
+	}()
+
+	return resultsChn
+}
+
+type nextRequests struct {
+	bitIdx         uint
+	sectionIdxList []uint64
+}
+
+func (m *Matcher) distributeRequests(stop chan struct{}) {
+	reqCnt := 0
+	reqs := make(map[uint][]uint64)
+	storeReq := func(r distReq) {
+		queue := reqs[r.bitIdx]
+		i := 0
+		for i < len(queue) && r.sectionIdx > 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)
+}

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 <http://www.gnu.org/licenses/>.
+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<<byte(7-i%8)) != 0 {
+			if ofs == lb {
+				panic(nil)
+			}
+			dc[i] = bits[ofs]
+			ofs++
+		}
+	}
+	return dc, ofs
+}
+
+const BloomLength = 2048
+
+type BloomBitsCreator struct {
+	blooms [BloomLength][SectionSize / 8]byte
+	bitIdx uint
+}
+
+func (b *BloomBitsCreator) AddHeaderBloom(bloom types.Bloom) {
+	if b.bitIdx >= 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][:])
+}

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

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)

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[:])
 

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
 }

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
+	backend                 Backend
-	useMipMap bool
+	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,
+		backend:      backend,
-		useMipMap: useMipMap,
+		useMipMap:    useMipMap,
-		db:        backend.ChainDb(),
+		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
+			l, b, e := checkBlock(i, header)
-			receipts, err := f.backend.GetReceipts(ctx, header.Hash())
+			if l != nil || e != nil {
-			if err != nil {
+				return l, b, e
-				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
 			}
 		}
 	}

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

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

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,
+		lightSync:          lightSync,
-		eventMux:    mux,
+		eventMux:           mux,
-		blockchain:  blockchain,
+		blockchain:         blockchain,
-		chainConfig: chainConfig,
+		chainConfig:        chainConfig,
-		chainDb:     chainDb,
+		chainDb:            chainDb,
-		networkId:   networkId,
+		networkId:          networkId,
-		txpool:      txpool,
+		txpool:             txpool,
-		txrelay:     txrelay,
+		txrelay:            txrelay,
-		odr:         odr,
+		odr:                odr,
-		peers:       newPeerSet(),
+		peers:              newPeerSet(),
-		newPeerCh:   make(chan *peer),
+		newPeerCh:          make(chan *peer),
-		quitSync:    make(chan struct{}),
+		quitSync:           make(chan struct{}),
-		noMorePeers: 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, "")

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)
 

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

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()

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

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)
+	lastChtKey = []byte("LastChtNumber6") // chtNum (uint64 big endian)
-	chtPrefix  = []byte("cht")           // chtPrefix + chtNum (uint64 big endian) -> trie root hash
+	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++ {
+	var compSize, decompSize uint64
-		hash := core.GetCanonicalHash(db, num)
+loop:
-		if hash == (common.Hash{}) {
+	for newChtNum > lastChtNum {
-			panic("Canonical hash not found")
+		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 {
+		for i := uint(0); i < bloombits.BloomLength; i++ {
-			panic("TD not found")
+			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.Info).Infof("Storing CHT #%d  root hash: %064x  compression ratio: %f", lastChtNum, root, float64(compSize)/float64(decompSize))
-
-		glog.V(logger.Detail).Infof("cht: %d %064x", lastChtNum, root)
 
 		storeChtRoot(db, lastChtNum, root)
 		var data [8]byte

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 {

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,
+				Number: 752,
-				Root:   common.HexToHash("01e408d9b1942f05dba1a879f3eaafe34d219edaeb8223fecf1244cc023d3e23"),
+				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
 }

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

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)
+const (
-	ChtConfirmations = uint64(2048)
+	ChtFrequency     = 4096
-	trustedChtKey    = []byte("TrustedCHT")
+	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
+	}
+}