mirror of
https://github.com/ethereum/go-ethereum.git
synced 2026-07-27 15:16:43 +00:00
core/bloombits, eth/filter: transformed bloom bitmap based log search
This commit is contained in:
Zsolt Felfoldi
parent
f421827f71
commit
e860278a96
16 changed files with 1354 additions and 52 deletions
101
core/bloombits/fetcher_test.go
Normal file
101
core/bloombits/fetcher_test.go
Normal file
|
|
@ -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 <http://www.gnu.org/licenses/>.
|
||||
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)
|
||||
}
|
||||
}
|
||||
575
core/bloombits/matcher.go
Normal file
575
core/bloombits/matcher.go
Normal file
|
|
@ -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 <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
|
||||
)
|
||||
|
||||
// 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<<bit) != 0 {
|
||||
select {
|
||||
case <-stop:
|
||||
return
|
||||
case resultsChn <- i:
|
||||
}
|
||||
}
|
||||
} else {
|
||||
i += bit
|
||||
}
|
||||
}
|
||||
|
||||
case <-stop:
|
||||
return
|
||||
}
|
||||
}
|
||||
}()
|
||||
|
||||
return resultsChn
|
||||
}
|
||||
|
||||
type nextRequests struct {
|
||||
bitIdx uint
|
||||
sectionIdxList []uint64
|
||||
}
|
||||
|
||||
// distributeRequests receives requests from the fetchers and either queues them
|
||||
// or immediately forwards them to one of the waiting NextRequest functions.
|
||||
// Requests with a lower section idx are always prioritized.
|
||||
func (m *Matcher) distributeRequests(stop chan struct{}) {
|
||||
m.distWg.Add(1)
|
||||
stopDist := make(chan struct{})
|
||||
go func() {
|
||||
<-stop
|
||||
m.wg.Wait()
|
||||
close(stopDist)
|
||||
}()
|
||||
|
||||
go func() {
|
||||
defer m.distWg.Done()
|
||||
|
||||
reqCnt := 0
|
||||
for _, s := range m.reqs {
|
||||
reqCnt += len(s)
|
||||
}
|
||||
|
||||
storeReq := func(r distReq) {
|
||||
queue := m.reqs[r.bitIdx]
|
||||
i := 0
|
||||
for i < len(queue) && r.sectionIdx > 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)
|
||||
}
|
||||
180
core/bloombits/matcher_test.go
Normal file
180
core/bloombits/matcher_test.go
Normal file
|
|
@ -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 <http://www.gnu.org/licenses/>.
|
||||
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)
|
||||
}
|
||||
}
|
||||
186
core/bloombits/utils.go
Normal file
186
core/bloombits/utils.go
Normal file
|
|
@ -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 <http://www.gnu.org/licenses/>.
|
||||
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<<byte(7-i%8)) != 0 {
|
||||
if ofs == lb {
|
||||
panic(nil)
|
||||
}
|
||||
dc[i] = bits[ofs]
|
||||
ofs++
|
||||
}
|
||||
}
|
||||
return dc, ofs
|
||||
}
|
||||
|
||||
const BloomLength = 2048
|
||||
|
||||
// BloomBitsCreator takes SectionSize number of header bloom filters and calculates the bloomBits vectors of the section
|
||||
type BloomBitsCreator struct {
|
||||
blooms [BloomLength][]byte
|
||||
sectionSize, bitIdx uint64
|
||||
}
|
||||
|
||||
func NewBloomBitsCreator(sectionSize uint64) *BloomBitsCreator {
|
||||
b := &BloomBitsCreator{sectionSize: sectionSize}
|
||||
for i, _ := range b.blooms {
|
||||
b.blooms[i] = make([]byte, sectionSize/8)
|
||||
}
|
||||
return b
|
||||
}
|
||||
|
||||
// AddHeaderBloom takes a single bloom filter and sets the corresponding bit column in memory accordingly
|
||||
func (b *BloomBitsCreator) AddHeaderBloom(bloom types.Bloom) {
|
||||
if b.bitIdx >= 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][:])
|
||||
}
|
||||
|
|
@ -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[:])
|
||||
}
|
||||
|
|
|
|||
|
|
@ -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[:])
|
||||
|
||||
|
|
|
|||
|
|
@ -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
|
||||
}
|
||||
|
|
|
|||
|
|
@ -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",
|
||||
|
|
|
|||
|
|
@ -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)
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -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 {
|
||||
|
|
|
|||
|
|
@ -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
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -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")
|
||||
|
|
|
|||
|
|
@ -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)
|
||||
|
|
|
|||
|
|
@ -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")
|
||||
|
|
|
|||
|
|
@ -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
|
||||
}
|
||||
|
|
|
|||
|
|
@ -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",
|
||||
|
|
|
|||
Loading…
Reference in a new issue