diff --git a/core/bloombits/fetcher_test.go b/core/bloombits/fetcher_test.go
new file mode 100644
index 0000000000..3f436dfaa3
--- /dev/null
+++ b/core/bloombits/fetcher_test.go
@@ -0,0 +1,101 @@
+// 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"
+ "sync"
+ "sync/atomic"
+ "testing"
+ "time"
+)
+
+const testFetcherReqCnt = 50000
+
+func fetcherTestVector(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) {
+ testFetcher(t, 1)
+}
+
+func TestFetcherMultipleReaders(t *testing.T) {
+ testFetcher(t, 10)
+}
+
+func testFetcher(t *testing.T, cnt int) {
+ f := &fetcher{
+ reqMap: make(map[uint64]req),
+ }
+ distChn := make(chan distReq, channelCap)
+ stop := make(chan struct{})
+ var reqCnt uint32
+
+ for i := 0; i < 10; i++ {
+ go func() {
+ for {
+ req, ok := <-distChn
+ if !ok {
+ return
+ }
+ time.Sleep(time.Duration(rand.Intn(1000000)))
+ atomic.AddUint32(&reqCnt, 1)
+ f.deliver([]uint64{req.sectionIdx}, []BitVector{fetcherTestVector(req.bitIdx, req.sectionIdx)})
+ }
+ }()
+ }
+
+ var wg, wg2 sync.WaitGroup
+ for cc := 0; cc < cnt; cc++ {
+ wg.Add(1)
+ in := make(chan uint64, channelCap)
+ out := f.fetch(in, distChn, stop, &wg2)
+
+ time.Sleep(time.Millisecond * 100 * time.Duration(cc))
+ go func() {
+ for i := uint64(0); i < testFetcherReqCnt; i++ {
+ in <- i
+ }
+ }()
+
+ go func() {
+ for i := uint64(0); i < testFetcherReqCnt; i++ {
+ bv := <-out
+ if !bytes.Equal(bv, fetcherTestVector(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)
+ }
+ }
+ }
+ wg.Done()
+ }()
+ }
+
+ wg.Wait()
+ close(stop)
+ if reqCnt != testFetcherReqCnt {
+ t.Errorf("Request count mismatch: expected %v, got %v", testFetcherReqCnt, reqCnt)
+ }
+}
diff --git a/core/bloombits/matcher.go b/core/bloombits/matcher.go
new file mode 100644
index 0000000000..abd5bd734b
--- /dev/null
+++ b/core/bloombits/matcher.go
@@ -0,0 +1,575 @@
+// 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
+)
+
+// fetcher handles bit vector retrieval pipelines for a single bit index
+type fetcher struct {
+ bitIdx, ii uint
+ reqMap map[uint64]req
+ reqLock sync.RWMutex
+}
+
+type req struct {
+ data BitVector
+ queued bool
+ fetched chan struct{}
+}
+
+type distReq struct {
+ bitIdx uint
+ sectionIdx uint64
+}
+
+// fetch creates a retrieval pipeline, receiving section indexes from sectionChn and returning the results
+// in the same order through the returned channel. Multiple fetch instances of the same fetcher are allowed
+// to run in parallel, in case the same bit index appears multiple times in the filter structure. Each section
+// is requested only once, requests are sent to the request distributor (part of Matcher) through distChn.
+func (f *fetcher) fetch(sectionChn chan uint64, distChn chan distReq, stop chan struct{}, wg *sync.WaitGroup) chan BitVector {
+ dataChn := make(chan BitVector, channelCap)
+ returnChn := make(chan uint64, channelCap)
+ wg.Add(2)
+
+ go func() {
+ defer func() {
+ close(returnChn)
+ wg.Done()
+ }()
+
+ for {
+ select {
+ case <-stop:
+ return
+ case idx, ok := <-sectionChn:
+ if !ok {
+ return
+ }
+
+ req := false
+ f.reqLock.Lock()
+ r := f.reqMap[idx]
+ if r.data == nil {
+ req = !r.queued
+ r.queued = true
+ if r.fetched == nil {
+ r.fetched = make(chan struct{})
+ }
+ f.reqMap[idx] = r
+ }
+ f.reqLock.Unlock()
+ if req {
+ distChn <- distReq{bitIdx: f.bitIdx, sectionIdx: idx} // success is guaranteed, distibuteRequests shuts down after fetch
+ }
+ select {
+ case <-stop:
+ return
+ case returnChn <- idx:
+ }
+ }
+ }
+ }()
+
+ go func() {
+ defer func() {
+ close(dataChn)
+ wg.Done()
+ }()
+
+ 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()
+ }
+ }
+ select {
+ case <-stop:
+ return
+ case dataChn <- r.data:
+ }
+ }
+ }
+ }()
+
+ return dataChn
+}
+
+// deliver is called by the request distributor when a reply to a request has
+// arrived
+func (f *fetcher) deliver(sectionIdxList []uint64, data []BitVector) {
+ f.reqLock.Lock()
+ defer f.reqLock.Unlock()
+
+ for i, idx := range sectionIdxList {
+ r := f.reqMap[idx]
+ if r.data != nil {
+ panic("BloomBits section data delivered twice")
+ }
+ r.data = data[i]
+ close(r.fetched)
+ f.reqMap[idx] = r
+ }
+}
+
+// Matcher is a pipelined structure of fetchers and logic matchers which perform
+// binary and/or operations on the bitstreams, finally creating a stream of potential matches.
+type Matcher struct {
+ addresses []types.BloomIndexList
+ topics [][]types.BloomIndexList
+ fetchers map[uint]*fetcher
+ sectionSize uint64
+
+ distChn chan distReq
+ reqs map[uint][]uint64
+ getNextReqChn chan chan nextRequests
+ wg, distWg sync.WaitGroup
+}
+
+// NewMatcher creates a new Matcher instance
+func NewMatcher(sectionSize uint64) *Matcher {
+ return &Matcher{fetchers: make(map[uint]*fetcher), reqs: make(map[uint][]uint64), distChn: make(chan distReq, channelCap), sectionSize: sectionSize}
+}
+
+// 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())
+ }
+
+ for _, idxs := range m.addresses {
+ for _, idx := range idxs {
+ m.newFetcher(idx)
+ }
+ }
+}
+
+// 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)
+ }
+
+ for _, idxss := range m.topics {
+ for _, idxs := range idxss {
+ for _, idx := range idxs {
+ m.newFetcher(idx)
+ }
+ }
+ }
+}
+
+// match creates a daisy-chain of sub-matchers, one for the address set and one for each topic set, each
+// sub-matcher receiving a section only if the previous ones have all found a potential match in one of
+// the blocks of the section, then binary and-ing its own matches and forwaring the result to the next one
+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.getNextReqChn = make(chan chan nextRequests) // should be a blocking channel
+ m.distributeRequests(stop)
+
+ s := sectionChn
+ var bv chan BitVector
+ for _, idx := range subIdx {
+ s, bv = m.subMatch(s, bv, idx, stop)
+ }
+ return s, bv
+}
+
+// newFetcher adds a fetcher for the given bit index if it has not existed before
+func (m *Matcher) newFetcher(idx uint) {
+ if _, ok := m.fetchers[idx]; ok {
+ return
+ }
+ f := &fetcher{
+ bitIdx: idx,
+ reqMap: make(map[uint64]req),
+ }
+ m.fetchers[idx] = f
+}
+
+// subMatch creates a sub-matcher that filters for a set of addresses or topics, binary or-s those matches, then
+// binary and-s the result to the daisy-chain input (sectionChn/andVectorChn) and forwards it to the daisy-chain output.
+// The matches of each address/topic are calculated by fetching the given sections of the three bloom bit indexes belonging to
+// that address/topic, and binary and-ing those vectors together.
+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.fetchers[ii].fetch(fetchIdx[i][j], m.distChn, stop, &m.wg)
+ }
+ }
+
+ processChn := make(chan uint64, channelCap)
+ resIdxChn := make(chan uint64, channelCap)
+ resDataChn := make(chan BitVector, channelCap)
+
+ m.wg.Add(2)
+ // goroutine for starting retrievals
+ go func() {
+ defer m.wg.Done()
+
+ for {
+ select {
+ case <-stop:
+ return
+ case s, ok := <-sectionChn:
+ if !ok {
+ close(processChn)
+ for _, ff := range fetchIdx {
+ for _, f := range ff {
+ close(f)
+ }
+ }
+ return
+ }
+
+ select {
+ case <-stop:
+ return
+ case processChn <- s:
+ }
+ for _, ff := range fetchIdx {
+ for _, f := range ff {
+ select {
+ case <-stop:
+ return
+ case f <- s:
+ }
+ }
+ }
+ }
+ }
+ }()
+
+ // goroutine for processing retrieved data
+ go func() {
+ defer m.wg.Done()
+
+ 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 {
+ var data BitVector
+ select {
+ case <-stop:
+ return
+ case data = <-f:
+ }
+ if andVector == nil {
+ andVector = bvCopy(data, int(m.sectionSize))
+ } else {
+ bvAnd(andVector, data)
+ }
+ }
+ if orVector == nil {
+ orVector = andVector
+ } else {
+ bvOr(orVector, andVector)
+ }
+ }
+
+ if orVector == nil {
+ orVector = bvZero(int(m.sectionSize))
+ }
+ if andVectorChn != nil {
+ select {
+ case <-stop:
+ return
+ case andVector := <-andVectorChn:
+ bvAnd(orVector, andVector)
+ }
+ }
+ if bvIsNonZero(orVector) {
+ select {
+ case <-stop:
+ return
+ case resIdxChn <- s:
+ }
+ select {
+ case <-stop:
+ return
+ case resDataChn <- orVector:
+ }
+ }
+ }
+ }
+ }()
+
+ return resIdxChn, resDataChn
+}
+
+// GetMatches returns a stream of bloom matches in a given range of blocks.
+// It returns a results channel immediately and stops if the stop channel is closed or
+// there are no more matches in the range (in which case the results channel is closed).
+// GetMatches can be called multiple times for different ranges, in which case already
+// delivered bit vectors are not requested again.
+func (m *Matcher) GetMatches(start, end uint64, stop chan struct{}) chan uint64 {
+ m.distWg.Wait()
+
+ sectionChn := make(chan uint64, channelCap)
+ resultsChn := make(chan uint64, channelCap)
+
+ s, bv := m.match(sectionChn, stop)
+
+ startSection := start / m.sectionSize
+ endSection := end / m.sectionSize
+
+ m.wg.Add(2)
+ go func() {
+ defer func() {
+ close(sectionChn)
+ m.wg.Done()
+ }()
+
+ for i := startSection; i <= endSection; i++ {
+ select {
+ case sectionChn <- i:
+ case <-stop:
+ return
+ }
+ }
+ }()
+
+ go func() {
+ defer func() {
+ close(resultsChn)
+ m.wg.Done()
+ }()
+
+ for {
+ select {
+ case idx, ok := <-s:
+ if !ok {
+ return
+ }
+ var match BitVector
+ select {
+ case <-stop:
+ return
+ case match = <-bv:
+ }
+ sectionStart := idx * m.sectionSize
+ s := sectionStart
+ if start > s {
+ s = start
+ }
+ e := sectionStart + m.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++
+ }
+ queue = append(queue, 0)
+ copy(queue[i+1:], queue[i:len(queue)-1])
+ queue[i] = r.sectionIdx
+
+ m.reqs[r.bitIdx] = queue
+ reqCnt++
+ }
+
+ storeReqs := func(r distReq) {
+ storeReq(r)
+ timeout := time.After(time.Microsecond)
+ for {
+ select {
+ case <-timeout:
+ return
+ case r := <-m.distChn:
+ storeReq(r)
+ case <-stopDist:
+ return
+ }
+ }
+ }
+
+ for {
+ if reqCnt == 0 {
+ select {
+ case r := <-m.distChn:
+ storeReqs(r)
+ case <-stopDist:
+ return
+ }
+ } else {
+ select {
+ case r := <-m.distChn:
+ storeReqs(r)
+ case <-stopDist:
+ return
+ case c := <-m.getNextReqChn:
+ var (
+ found bool
+ bestBit uint
+ bestSection uint64
+ )
+
+ for bitIdx, queue := range m.reqs {
+ if len(queue) > 0 && (!found || queue[0] < bestSection) {
+ found = true
+ bestBit = bitIdx
+ bestSection = queue[0]
+ }
+ }
+ if !found {
+ panic(nil)
+ }
+
+ bestQueue := m.reqs[bestBit]
+ cnt := len(bestQueue)
+ if cnt > maxRequestLength {
+ cnt = maxRequestLength
+ }
+ res := nextRequests{bestBit, bestQueue[:cnt]}
+ m.reqs[bestBit] = bestQueue[cnt:]
+ reqCnt -= cnt
+
+ c <- res
+ }
+ }
+ }
+ }()
+}
+
+// NextRequest asks for a request from the request distributor, returns immediately if there
+// was a queued one, otherwise waits until the distributor forwards one of the requests sent
+// by one of the fetchers.
+func (m *Matcher) NextRequest(stop chan struct{}) (bitIdx uint, sectionIdxList []uint64) {
+ c := make(chan nextRequests)
+ select {
+ case m.getNextReqChn <- c:
+ r := <-c
+ return r.bitIdx, r.sectionIdxList
+ case <-stop:
+ return 0, nil
+ }
+}
+
+// Deliver delivers a bit vector to the appropriate fetcher.
+// 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.
+func (m *Matcher) Deliver(bitIdx uint, sectionIdxList []uint64, data []BitVector) {
+ m.fetchers[bitIdx].deliver(sectionIdxList, data)
+}
diff --git a/core/bloombits/matcher_test.go b/core/bloombits/matcher_test.go
new file mode 100644
index 0000000000..21d617b840
--- /dev/null
+++ b/core/bloombits/matcher_test.go
@@ -0,0 +1,180 @@
+// 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 (
+ "math/rand"
+ "sync/atomic"
+ "testing"
+
+ "github.com/ethereum/go-ethereum/core/types"
+)
+
+const testSectionSize = 4096
+
+func matcherTestVector(b uint, s uint64) BitVector {
+ r := make(BitVector, testSectionSize/8)
+ for i, _ := range r {
+ var bb byte
+ for bit := 0; bit < 8; bit++ {
+ blockIdx := s*testSectionSize + uint64(i*8+bit)
+ bb += bb
+ if (blockIdx % uint64(b)) == 0 {
+ bb++
+ }
+ }
+ r[i] = bb
+ }
+ return r
+}
+
+func expMatch1(idxs types.BloomIndexList, i uint64) bool {
+ for _, ii := range idxs {
+ if (i % uint64(ii)) != 0 {
+ return false
+ }
+ }
+ return true
+}
+
+func expMatch2(idxs []types.BloomIndexList, i uint64) bool {
+ for _, ii := range idxs {
+ if expMatch1(ii, i) {
+ return true
+ }
+ }
+ return false
+}
+
+func expMatch3(idxs [][]types.BloomIndexList, i uint64) bool {
+ for _, ii := range idxs {
+ if !expMatch2(ii, i) {
+ return false
+ }
+ }
+ return true
+}
+
+func testServeMatcher(m *Matcher, stop chan struct{}, cnt *uint32) {
+ // serve matcher with test vectors
+ for i := 0; i < 10; i++ {
+ go func() {
+ for {
+ select {
+ case <-stop:
+ return
+ default:
+ }
+ b, s := m.NextRequest(stop)
+ if s == nil {
+ return
+ }
+ res := make([]BitVector, len(s))
+ for i, ss := range s {
+ res[i] = matcherTestVector(b, ss)
+ atomic.AddUint32(cnt, 1)
+ }
+ m.Deliver(b, s, res)
+ }
+ }()
+ }
+}
+
+func testMatcher(t *testing.T, idxs [][]types.BloomIndexList, cnt uint64, stopOnMatches bool, expCnt uint32) uint32 {
+ m := NewMatcher(testSectionSize)
+
+ for _, idxss := range idxs {
+ for _, idxs := range idxss {
+ for _, idx := range idxs {
+ m.newFetcher(idx)
+ }
+ }
+ }
+
+ m.addresses = idxs[0]
+ m.topics = idxs[1:]
+ var reqCnt uint32
+
+ stop := make(chan struct{})
+ chn := m.GetMatches(0, cnt-1, stop)
+ testServeMatcher(m, stop, &reqCnt)
+
+ for i := uint64(0); i < cnt; i++ {
+ if expMatch3(idxs, i) {
+ match, ok := <-chn
+ if !ok {
+ t.Errorf("Error matching idxs = %v count = %v stopOnMatches = %v: expected #%v, results channel closed", idxs, cnt, stopOnMatches, i)
+ return 0
+ }
+ if match != i {
+ t.Errorf("Error matching idxs = %v count = %v stopOnMatches = %v: expected #%v, got #%v", idxs, cnt, stopOnMatches, i, match)
+ }
+ if stopOnMatches {
+ close(stop)
+ stop = make(chan struct{})
+ chn = m.GetMatches(i+1, cnt-1, stop)
+ testServeMatcher(m, stop, &reqCnt)
+ }
+ }
+ }
+ match, ok := <-chn
+ if ok {
+ t.Errorf("Error matching idxs = %v count = %v stopOnMatches = %v: expected closed channel, got #%v", idxs, cnt, stopOnMatches, match)
+ }
+ close(stop)
+
+ if expCnt != 0 && expCnt != reqCnt {
+ t.Errorf("Error matching idxs = %v count = %v stopOnMatches = %v: request count mismatch, expected #%v, got #%v", idxs, cnt, stopOnMatches, expCnt, reqCnt)
+ }
+
+ return reqCnt
+}
+
+func testRandomIdxs(l []int, max int) [][]types.BloomIndexList {
+ res := make([][]types.BloomIndexList, len(l))
+ for i, ll := range l {
+ res[i] = make([]types.BloomIndexList, ll)
+ for j, _ := range res[i] {
+ for k, _ := range res[i][j] {
+ res[i][j][k] = uint(rand.Intn(max-1) + 2)
+ }
+ }
+ }
+ return res
+}
+
+func TestMatcher(t *testing.T) {
+ testMatcher(t, [][]types.BloomIndexList{{{10, 20, 30}}}, 1000000, false, 735)
+ testMatcher(t, [][]types.BloomIndexList{{{32, 3125, 100}}, {{40, 50, 10}}}, 1000000, false, 768)
+ testMatcher(t, [][]types.BloomIndexList{{{4, 8, 11}, {7, 8, 17}}, {{9, 9, 12}, {15, 20, 13}}, {{18, 15, 15}, {12, 10, 4}}}, 100000, false, 300)
+}
+
+func TestMatcherStopOnMatches(t *testing.T) {
+ testMatcher(t, [][]types.BloomIndexList{{{10, 20, 30}}}, 1000000, true, 735)
+ testMatcher(t, [][]types.BloomIndexList{{{4, 8, 11}, {7, 8, 17}}, {{9, 9, 12}, {15, 20, 13}}, {{18, 15, 15}, {12, 10, 4}}}, 100000, true, 300)
+}
+
+func TestMatcherRandom(t *testing.T) {
+ for i := 0; i < 20; i++ {
+ testMatcher(t, testRandomIdxs([]int{1}, 50), 1000000, false, 0)
+ testMatcher(t, testRandomIdxs([]int{3}, 50), 1000000, false, 0)
+ testMatcher(t, testRandomIdxs([]int{2, 2, 2}, 20), 1000000, false, 0)
+ testMatcher(t, testRandomIdxs([]int{5, 5, 5}, 50), 1000000, false, 0)
+ idxs := testRandomIdxs([]int{2, 2, 2}, 20)
+ reqCnt := testMatcher(t, idxs, 100000, false, 0)
+ testMatcher(t, idxs, 100000, true, reqCnt)
+ }
+}
diff --git a/core/bloombits/utils.go b/core/bloombits/utils.go
new file mode 100644
index 0000000000..166a7d6039
--- /dev/null
+++ b/core/bloombits/utils.go
@@ -0,0 +1,186 @@
+// 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"
+)
+
+type (
+ BitVector []byte
+ CompVector []byte
+)
+
+// bvAnd binary ANDs b to a
+func bvAnd(a, b BitVector) {
+ for i, bb := range b {
+ a[i] &= bb
+ }
+}
+
+// bvOr binary ORs b to a
+func bvOr(a, b BitVector) {
+ for i, bb := range b {
+ a[i] |= bb
+ }
+}
+
+// bvZero returns an all-zero bit vector
+func bvZero(sectionSize int) BitVector {
+ return make(BitVector, sectionSize/8)
+}
+
+// bvCopy creates a copy of the given bit vector
+// If the source vector is nil, returns an all-zero bit vector
+func bvCopy(a BitVector, sectionSize int) BitVector {
+ c := make(BitVector, sectionSize/8)
+ copy(c, a)
+ return c
+}
+
+// bvIsNonZero returns true if the bit vector has at least one "1" bit
+func bvIsNonZero(a BitVector) bool {
+ for _, b := range a {
+ if b != 0 {
+ return true
+ }
+ }
+ return false
+}
+
+// CompressBloomBits compresses a bit vector for storage/network transfer purposes
+func CompressBloomBits(bits BitVector, sectionSize int) 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, sectionSize))
+ }
+ return CompVector(c)
+}
+
+func compressBits(bits []byte) []byte {
+ l := len(bits)
+ ll := l / 8
+ if ll == 0 {
+ ll = 1
+ }
+ b := make([]byte, ll)
+ 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 ll > 1 {
+ b = compressBits(b)
+ }
+ return append(b, c[0:cl]...)
+}
+
+// DeompressBloomBits decompresses a bit vector
+func DecompressBloomBits(bits CompVector, sectionSize int) BitVector {
+ if len(bits) == sectionSize/8 {
+ // make a copy so that output is always detached from input
+ return bvCopy(BitVector(bits), sectionSize)
+ }
+ 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<= b.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++
+}
+
+// GetBitVector returns the bit vector belonging to the given bit index after header blooms have been added
+func (b *BloomBitsCreator) GetBitVector(idx uint) BitVector {
+ if b.bitIdx != b.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 bcd99be5fb..affe6e4130 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/log"
@@ -70,6 +71,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
@@ -675,3 +679,38 @@ func FindCommonAncestor(db ethdb.Database, a, b *types.Header) *types.Header {
}
return a
}
+
+// GetBloomBits reads the compressed bloomBits vector belonging to the given section and bit index from the db
+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
+}
+
+// StoreBloomBits writes the compressed bloomBits vector belonging to the given section and bit index to the db
+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)
+}
+
+// GetBloomBitsAvailable reads the number of available bloomBits sections from the db
+func GetBloomBitsAvailable(db ethdb.Database) uint64 {
+ data, _ := db.Get(bloomBitsAvailKey)
+ if len(data) == 8 {
+ return binary.BigEndian.Uint64(data[:])
+ }
+ return 0
+}
+
+// StoreBloomBitsAvailable writes the number of available bloomBits sections to the db
+func StoreBloomBitsAvailable(db ethdb.Database, cnt uint64) {
+ var data [8]byte
+ binary.BigEndian.PutUint64(data[:], cnt)
+ db.Put(bloomBitsAvailKey, data[:])
+}
diff --git a/core/types/bloom9.go b/core/types/bloom9.go
index 60aacc3016..bdc6e60e79 100644
--- a/core/types/bloom9.go
+++ b/core/types/bloom9.go
@@ -106,6 +106,20 @@ func LogsBloom(logs []*Log) *big.Int {
return bin
}
+type BloomIndexList [3]uint
+
+// BloomIndexes returns the bloom filter bit indexes belonging to the given key
+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 fe108d2720..19c8ce271d 100644
--- a/eth/api_backend.go
+++ b/eth/api_backend.go
@@ -24,6 +24,7 @@ import (
"github.com/ethereum/go-ethereum/common"
"github.com/ethereum/go-ethereum/common/math"
"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"
@@ -201,6 +202,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/backend.go b/eth/backend.go
index f864b1d88b..7dd9eeddd3 100644
--- a/eth/backend.go
+++ b/eth/backend.go
@@ -239,6 +239,10 @@ func (s *Ethereum) APIs() []rpc.API {
// Append any APIs exposed explicitly by the consensus engine
apis = append(apis, s.engine.APIs(s.BlockChain())...)
+ var bbSection uint64
+ if useBloomBits {
+ bbSection = bloomBitsSection
+ }
// Append all the local APIs and return
return append(apis, []rpc.API{
{
@@ -264,7 +268,7 @@ func (s *Ethereum) APIs() []rpc.API {
}, {
Namespace: "eth",
Version: "1.0",
- Service: filters.NewPublicFilterAPI(s.ApiBackend, false),
+ Service: filters.NewPublicFilterAPI(s.ApiBackend, false, bbSection),
Public: true,
}, {
Namespace: "admin",
diff --git a/eth/db_upgrade.go b/eth/db_upgrade.go
index 82cdd7e55e..2d63096df4 100644
--- a/eth/db_upgrade.go
+++ b/eth/db_upgrade.go
@@ -26,6 +26,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/log"
@@ -294,3 +295,64 @@ func addMipmapBloomBins(db ethdb.Database) (err error) {
log.Info("Bloom-bin upgrade completed", "elapsed", common.PrettyDuration(time.Since(tstart)))
return nil
}
+
+// BloomBitsProcessorBackend implements ChainSectionProcessorBackend
+type BloomBitsProcessorBackend struct {
+ db ethdb.Database
+ lastCount, lastProg uint64
+}
+
+// number of confirmation blocks before a section is considered probably final and its bloom bits are calculated
+const bloomBitsConfirmations = 512
+
+// NewBloomBitsProcessor returns a chain processor that generates bloom bits data for the canonical chain
+func NewBloomBitsProcessor(db ethdb.Database, stop chan struct{}) *core.ChainSectionProcessor {
+ backend := &BloomBitsProcessorBackend{db: db, lastCount: core.GetBloomBitsAvailable(db)}
+ return core.NewChainSectionProcessor(backend, bloomBitsSection, bloomBitsConfirmations, time.Millisecond*100, stop)
+}
+
+// Process calculates bloom bits for a given section
+func (b *BloomBitsProcessorBackend) Process(sectionIdx uint64) bool {
+ bc := bloombits.NewBloomBitsCreator(bloomBitsSection)
+ var header *types.Header
+ for i := sectionIdx * bloomBitsSection; i < (sectionIdx+1)*bloomBitsSection; i++ {
+ hash := core.GetCanonicalHash(b.db, i)
+ header = core.GetHeader(b.db, hash, i)
+ if header == nil {
+ log.Error("Error creating bloomBits data", "section", sectionIdx, "missing header", i)
+ return false
+ }
+ bc.AddHeaderBloom(header.Bloom)
+ }
+
+ for i := 0; i < bloombits.BloomLength; i++ {
+ core.StoreBloomBits(b.db, uint64(i), sectionIdx, bloombits.CompressBloomBits(bc.GetBitVector(uint(i)), bloomBitsSection))
+ }
+
+ return true
+}
+
+// SetStored sets the number of stored sections (last valid section is count-1)
+func (b *BloomBitsProcessorBackend) SetStored(count uint64) {
+ if count > b.lastCount {
+ log.Debug("Stored bloomBits data", "count", count)
+ } else {
+ log.Debug("Rolled back bloomBits data", "count", count)
+ }
+ b.lastCount = count
+ core.StoreBloomBitsAvailable(b.db, count)
+}
+
+// GetStored retrieves the number of stored sections (last valid section is count-1)
+func (b *BloomBitsProcessorBackend) GetStored() uint64 {
+ return core.GetBloomBitsAvailable(b.db)
+}
+
+// UpdateMsg prints an update message after each 5% of progress
+func (b *BloomBitsProcessorBackend) UpdateMsg(done, all uint64) {
+ prog := done * 20 / all
+ if done == 0 || prog > b.lastProg {
+ b.lastProg = prog
+ log.Info("Updating quick bloom filter database", "%", prog*5)
+ }
+}
diff --git a/eth/filters/api.go b/eth/filters/api.go
index 61647a5d07..7874ebac69 100644
--- a/eth/filters/api.go
+++ b/eth/filters/api.go
@@ -51,25 +51,27 @@ type filter struct {
// PublicFilterAPI offers support to create and manage filters. This will allow external clients to retrieve various
// information related to the Ethereum protocol such als blocks, transactions and logs.
type PublicFilterAPI struct {
- backend Backend
- useMipMap bool
- mux *event.TypeMux
- quit chan struct{}
- chainDb ethdb.Database
- events *EventSystem
- filtersMu sync.Mutex
- filters map[rpc.ID]*filter
+ backend Backend
+ useMipMap bool
+ bloomBitsSection uint64
+ mux *event.TypeMux
+ quit chan struct{}
+ chainDb ethdb.Database
+ events *EventSystem
+ filtersMu sync.Mutex
+ filters map[rpc.ID]*filter
}
// NewPublicFilterAPI returns a new PublicFilterAPI instance.
-func NewPublicFilterAPI(backend Backend, lightMode bool) *PublicFilterAPI {
+func NewPublicFilterAPI(backend Backend, lightMode bool, bloomBitsSection uint64) *PublicFilterAPI {
api := &PublicFilterAPI{
- backend: backend,
- useMipMap: !lightMode,
- mux: backend.EventMux(),
- chainDb: backend.ChainDb(),
- events: NewEventSystem(backend.EventMux(), backend, lightMode),
- filters: make(map[rpc.ID]*filter),
+ backend: backend,
+ useMipMap: bloomBitsSection == 0,
+ bloomBitsSection: bloomBitsSection,
+ mux: backend.EventMux(),
+ chainDb: backend.ChainDb(),
+ events: NewEventSystem(backend.EventMux(), backend, lightMode),
+ filters: make(map[rpc.ID]*filter),
}
go api.timeoutLoop()
@@ -333,7 +335,7 @@ func (api *PublicFilterAPI) GetLogs(ctx context.Context, crit FilterCriteria) ([
crit.ToBlock = big.NewInt(rpc.LatestBlockNumber.Int64())
}
- filter := New(api.backend, api.useMipMap)
+ filter := New(api.backend, api.useMipMap, api.bloomBitsSection)
filter.SetBeginBlock(crit.FromBlock.Int64())
filter.SetEndBlock(crit.ToBlock.Int64())
filter.SetAddresses(crit.Addresses)
@@ -373,7 +375,7 @@ func (api *PublicFilterAPI) GetFilterLogs(ctx context.Context, id rpc.ID) ([]*ty
return nil, fmt.Errorf("filter not found")
}
- filter := New(api.backend, api.useMipMap)
+ filter := New(api.backend, api.useMipMap, api.bloomBitsSection)
if f.crit.FromBlock != nil {
filter.SetBeginBlock(f.crit.FromBlock.Int64())
} else {
diff --git a/eth/filters/filter.go b/eth/filters/filter.go
index 0a0b81224e..c17619abeb 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"
@@ -35,12 +36,14 @@ 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
+ bloomBitsSection uint64
created time.Time
@@ -48,17 +51,22 @@ 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
// a particular block is interesting or not.
// MipMaps allow past blocks to be searched much more efficiently, but are not available
// to light clients.
-func New(backend Backend, useMipMap bool) *Filter {
+func New(backend Backend, useMipMap bool, bloomBitsSection uint64) *Filter {
return &Filter{
- backend: backend,
- useMipMap: useMipMap,
- db: backend.ChainDb(),
+ backend: backend,
+ useMipMap: useMipMap,
+ useBloomBits: bloomBitsSection != 0,
+ bloomBitsSection: bloomBitsSection,
+ db: backend.ChainDb(),
+ matcher: bloombits.NewMatcher(bloomBitsSection),
}
}
@@ -78,11 +86,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
@@ -166,7 +180,100 @@ func (f *Filter) mipFind(start, end uint64, depth int) (logs []*types.Log, block
return nil, end
}
+// serveMatcher serves the bloomBits matcher by fetching the requested vectors
+// through the filter backend
+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 {
+ b, s := f.matcher.NextRequest(stop)
+ if s == nil {
+ return
+ }
+ data, err := f.backend.GetBloomBits(ctx, uint64(b), s)
+ if err != nil {
+ errChn <- err
+ return
+ }
+ decomp := make([]bloombits.BitVector, len(data))
+ for i, d := range data {
+ decomp[i] = bloombits.DecompressBloomBits(bloombits.CompVector(d), int(f.bloomBitsSection))
+ }
+ f.matcher.Deliver(b, s, decomp)
+ }
+ }(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) * f.bloomBitsSection
+ 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)
+ 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)
+
+ 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
+ }
+ }
+
+ // search the rest with regular block-by-block bloom filtering
for i := start; i <= end; i++ {
blockNumber := rpc.BlockNumber(i)
header, err := f.backend.HeaderByNumber(ctx, blockNumber)
@@ -177,18 +284,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/eth/filters/filter_system_test.go b/eth/filters/filter_system_test.go
index 822580b56c..d614b119d4 100644
--- a/eth/filters/filter_system_test.go
+++ b/eth/filters/filter_system_test.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/ethdb"
"github.com/ethereum/go-ethereum/event"
@@ -63,6 +64,18 @@ func (b *testBackend) GetReceipts(ctx context.Context, blockHash common.Hash) (t
return core.GetBlockReceipts(b.db, blockHash, num), nil
}
+func (b *testBackend) 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.db, bitIdx, idx)
+ if err != nil {
+ return nil, err
+ }
+ }
+ return results, nil
+}
+
// TestBlockSubscription tests if a block subscription returns block hashes for posted chain events.
// It creates multiple subscriptions:
// - one at the start and should receive all posted chain events and a second (blockHashes)
@@ -75,7 +88,7 @@ func TestBlockSubscription(t *testing.T) {
mux = new(event.TypeMux)
db, _ = ethdb.NewMemDatabase()
backend = &testBackend{mux, db}
- api = NewPublicFilterAPI(backend, false)
+ api = NewPublicFilterAPI(backend, false, 0)
genesis = new(core.Genesis).MustCommit(db)
chain, _ = core.GenerateChain(params.TestChainConfig, genesis, db, 10, func(i int, gen *core.BlockGen) {})
chainEvents = []core.ChainEvent{}
@@ -128,7 +141,7 @@ func TestPendingTxFilter(t *testing.T) {
mux = new(event.TypeMux)
db, _ = ethdb.NewMemDatabase()
backend = &testBackend{mux, db}
- api = NewPublicFilterAPI(backend, false)
+ api = NewPublicFilterAPI(backend, false, 0)
transactions = []*types.Transaction{
types.NewTransaction(0, common.HexToAddress("0xb794f5ea0ba39494ce83a213fffba74279579268"), new(big.Int), new(big.Int), new(big.Int), nil),
@@ -178,7 +191,7 @@ func TestLogFilterCreation(t *testing.T) {
mux = new(event.TypeMux)
db, _ = ethdb.NewMemDatabase()
backend = &testBackend{mux, db}
- api = NewPublicFilterAPI(backend, false)
+ api = NewPublicFilterAPI(backend, false, 0)
testCases = []struct {
crit FilterCriteria
@@ -223,7 +236,7 @@ func TestInvalidLogFilterCreation(t *testing.T) {
mux = new(event.TypeMux)
db, _ = ethdb.NewMemDatabase()
backend = &testBackend{mux, db}
- api = NewPublicFilterAPI(backend, false)
+ api = NewPublicFilterAPI(backend, false, 0)
)
// different situations where log filter creation should fail.
@@ -249,7 +262,7 @@ func TestLogFilter(t *testing.T) {
mux = new(event.TypeMux)
db, _ = ethdb.NewMemDatabase()
backend = &testBackend{mux, db}
- api = NewPublicFilterAPI(backend, false)
+ api = NewPublicFilterAPI(backend, false, 0)
firstAddr = common.HexToAddress("0x1111111111111111111111111111111111111111")
secondAddr = common.HexToAddress("0x2222222222222222222222222222222222222222")
@@ -357,7 +370,7 @@ func TestPendingLogsSubscription(t *testing.T) {
mux = new(event.TypeMux)
db, _ = ethdb.NewMemDatabase()
backend = &testBackend{mux, db}
- api = NewPublicFilterAPI(backend, false)
+ api = NewPublicFilterAPI(backend, false, 0)
firstAddr = common.HexToAddress("0x1111111111111111111111111111111111111111")
secondAddr = common.HexToAddress("0x2222222222222222222222222222222222222222")
diff --git a/eth/filters/filter_test.go b/eth/filters/filter_test.go
index cd5e7cafd5..ed32c71963 100644
--- a/eth/filters/filter_test.go
+++ b/eth/filters/filter_test.go
@@ -104,7 +104,7 @@ func BenchmarkMipmaps(b *testing.B) {
}
b.ResetTimer()
- filter := New(backend, true)
+ filter := New(backend, true, 0)
filter.SetAddresses([]common.Address{addr1, addr2, addr3, addr4})
filter.SetBeginBlock(0)
filter.SetEndBlock(-1)
@@ -207,7 +207,7 @@ func TestFilters(t *testing.T) {
}
}
- filter := New(backend, true)
+ filter := New(backend, true, 0)
filter.SetAddresses([]common.Address{addr})
filter.SetTopics([][]common.Hash{{hash1, hash2, hash3, hash4}})
filter.SetBeginBlock(0)
@@ -218,7 +218,7 @@ func TestFilters(t *testing.T) {
t.Error("expected 4 log, got", len(logs))
}
- filter = New(backend, true)
+ filter = New(backend, true, 0)
filter.SetAddresses([]common.Address{addr})
filter.SetTopics([][]common.Hash{{hash3}})
filter.SetBeginBlock(900)
@@ -231,7 +231,7 @@ func TestFilters(t *testing.T) {
t.Errorf("expected log[0].Topics[0] to be %x, got %x", hash3, logs[0].Topics[0])
}
- filter = New(backend, true)
+ filter = New(backend, true, 0)
filter.SetAddresses([]common.Address{addr})
filter.SetTopics([][]common.Hash{{hash3}})
filter.SetBeginBlock(990)
@@ -244,7 +244,7 @@ func TestFilters(t *testing.T) {
t.Errorf("expected log[0].Topics[0] to be %x, got %x", hash3, logs[0].Topics[0])
}
- filter = New(backend, true)
+ filter = New(backend, true, 0)
filter.SetTopics([][]common.Hash{{hash1, hash2}})
filter.SetBeginBlock(1)
filter.SetEndBlock(10)
@@ -255,7 +255,7 @@ func TestFilters(t *testing.T) {
}
failHash := common.BytesToHash([]byte("fail"))
- filter = New(backend, true)
+ filter = New(backend, true, 0)
filter.SetTopics([][]common.Hash{{failHash}})
filter.SetBeginBlock(0)
filter.SetEndBlock(-1)
@@ -266,7 +266,7 @@ func TestFilters(t *testing.T) {
}
failAddr := common.BytesToAddress([]byte("failmenow"))
- filter = New(backend, true)
+ filter = New(backend, true, 0)
filter.SetAddresses([]common.Address{failAddr})
filter.SetBeginBlock(0)
filter.SetEndBlock(-1)
@@ -276,7 +276,7 @@ func TestFilters(t *testing.T) {
t.Error("expected 0 log, got", len(logs))
}
- filter = New(backend, true)
+ filter = New(backend, true, 0)
filter.SetTopics([][]common.Hash{{failHash}, {hash1}})
filter.SetBeginBlock(0)
filter.SetEndBlock(-1)
diff --git a/eth/handler.go b/eth/handler.go
index 16e3712273..d8684c3bf0 100644
--- a/eth/handler.go
+++ b/eth/handler.go
@@ -45,6 +45,9 @@ import (
const (
softResponseLimit = 2 * 1024 * 1024 // Target maximum size of returned blocks, headers or node data.
estHeaderRlpSize = 500 // Approximate size of an RLP encoded block header
+
+ useBloomBits = false
+ bloomBitsSection = 4096
)
var (
@@ -87,7 +90,8 @@ type ProtocolManager struct {
quitSync chan struct{}
noMorePeers chan struct{}
- lesServer LesServer
+ lesServer LesServer
+ bloomBitsProcessor *core.ChainSectionProcessor
// wait group is used for graceful shutdowns during downloading
// and processing
@@ -112,6 +116,12 @@ func NewProtocolManager(config *params.ChainConfig, mode downloader.SyncMode, ne
txsyncCh: make(chan *txsync),
quitSync: make(chan struct{}),
}
+
+ if useBloomBits {
+ manager.bloomBitsProcessor = NewBloomBitsProcessor(manager.chaindb, manager.quitSync)
+ blockchain.AddChainProcessor(manager.bloomBitsProcessor)
+ }
+
// Figure out whether to allow fast sync or not
if mode == downloader.FastSync && blockchain.CurrentBlock().NumberU64() > 0 {
log.Warn("Blockchain not empty, fast sync disabled")
diff --git a/les/api_backend.go b/les/api_backend.go
index 7d69046deb..9bfc286feb 100644
--- a/les/api_backend.go
+++ b/les/api_backend.go
@@ -24,6 +24,7 @@ import (
"github.com/ethereum/go-ethereum/common"
"github.com/ethereum/go-ethereum/common/math"
"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"
@@ -155,3 +156,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 nil, nil // implemented in a subsequent PR
+}
diff --git a/les/backend.go b/les/backend.go
index 646c81a7b1..75f999aaa0 100644
--- a/les/backend.go
+++ b/les/backend.go
@@ -155,7 +155,7 @@ func (s *LightEthereum) APIs() []rpc.API {
}, {
Namespace: "eth",
Version: "1.0",
- Service: filters.NewPublicFilterAPI(s.ApiBackend, true),
+ Service: filters.NewPublicFilterAPI(s.ApiBackend, true, 0),
Public: true,
}, {
Namespace: "net",