diff options
author | Zsolt Felfoldi <zsfelfoldi@gmail.com> | 2017-08-19 03:52:20 +0800 |
---|---|---|
committer | Péter Szilágyi <peterke@gmail.com> | 2017-09-06 16:13:13 +0800 |
commit | 4ea4d2dc3473afd9d2eda6ef6b359accce1f0946 (patch) | |
tree | e651cfc2e3aa36083b333bf34dc3cccef2623f26 /core/bloombits | |
parent | 1e67378df879b1ce566f17dd95a3b126056254b5 (diff) | |
download | go-tangerine-4ea4d2dc3473afd9d2eda6ef6b359accce1f0946.tar go-tangerine-4ea4d2dc3473afd9d2eda6ef6b359accce1f0946.tar.gz go-tangerine-4ea4d2dc3473afd9d2eda6ef6b359accce1f0946.tar.bz2 go-tangerine-4ea4d2dc3473afd9d2eda6ef6b359accce1f0946.tar.lz go-tangerine-4ea4d2dc3473afd9d2eda6ef6b359accce1f0946.tar.xz go-tangerine-4ea4d2dc3473afd9d2eda6ef6b359accce1f0946.tar.zst go-tangerine-4ea4d2dc3473afd9d2eda6ef6b359accce1f0946.zip |
core, eth: add bloombit indexer, filter based on it
Diffstat (limited to 'core/bloombits')
-rw-r--r-- | core/bloombits/fetcher_test.go | 101 | ||||
-rw-r--r-- | core/bloombits/matcher.go | 579 | ||||
-rw-r--r-- | core/bloombits/matcher_test.go | 196 | ||||
-rw-r--r-- | core/bloombits/utils.go | 63 |
4 files changed, 939 insertions, 0 deletions
diff --git a/core/bloombits/fetcher_test.go b/core/bloombits/fetcher_test.go new file mode 100644 index 000000000..9c229cf8d --- /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 <http://www.gnu.org/licenses/>. +package bloombits + +import ( + "bytes" + "encoding/binary" + "math/rand" + "sync" + "sync/atomic" + "testing" + "time" +) + +const testFetcherReqCount = 5000 + +func fetcherTestVector(b uint, s uint64) []byte { + r := make([]byte, 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{ + requestMap: make(map[uint64]fetchRequest), + } + distCh := make(chan distRequest, channelCap) + stop := make(chan struct{}) + var reqCount uint32 + + for i := 0; i < 10; i++ { + go func() { + for { + req, ok := <-distCh + if !ok { + return + } + time.Sleep(time.Duration(rand.Intn(100000))) + atomic.AddUint32(&reqCount, 1) + f.deliver([]uint64{req.sectionIndex}, [][]byte{fetcherTestVector(req.bloomIndex, req.sectionIndex)}) + } + }() + } + + var wg, wg2 sync.WaitGroup + for cc := 0; cc < cnt; cc++ { + wg.Add(1) + in := make(chan uint64, channelCap) + out := f.fetch(in, distCh, stop, &wg2) + + time.Sleep(time.Millisecond * 10 * time.Duration(cc)) + go func() { + for i := uint64(0); i < testFetcherReqCount; i++ { + in <- i + } + }() + + go func() { + for i := uint64(0); i < testFetcherReqCount; 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 reqCount != testFetcherReqCount { + t.Errorf("Request count mismatch: expected %v, got %v", testFetcherReqCount, reqCount) + } +} diff --git a/core/bloombits/matcher.go b/core/bloombits/matcher.go new file mode 100644 index 000000000..5a7df6b1c --- /dev/null +++ b/core/bloombits/matcher.go @@ -0,0 +1,579 @@ +// 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" + + "github.com/ethereum/go-ethereum/common" + "github.com/ethereum/go-ethereum/common/bitutil" + "github.com/ethereum/go-ethereum/core/types" +) + +const channelCap = 100 + +// fetcher handles bit vector retrieval pipelines for a single bit index +type fetcher struct { + bloomIndex uint + requestMap map[uint64]fetchRequest + requestLock sync.RWMutex +} + +// fetchRequest represents the state of a bit vector requested from a fetcher. When a distRequest has been sent to the distributor but +// the data has not been delivered yet, queued is true. When delivered, it is stored in the data field and the delivered channel is closed. +type fetchRequest struct { + data []byte + queued bool + delivered chan struct{} +} + +// distRequest is sent by the fetcher to the distributor which groups and prioritizes these requests. +type distRequest struct { + bloomIndex uint + sectionIndex uint64 +} + +// fetch creates a retrieval pipeline, receiving section indexes from sectionCh 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 distCh. +func (f *fetcher) fetch(sectionCh chan uint64, distCh chan distRequest, stop chan struct{}, wg *sync.WaitGroup) chan []byte { + dataCh := make(chan []byte, channelCap) + returnCh := make(chan uint64, channelCap) + wg.Add(2) + + go func() { + defer wg.Done() + defer close(returnCh) + + for { + select { + case <-stop: + return + case idx, ok := <-sectionCh: + if !ok { + return + } + + req := false + f.requestLock.Lock() + r := f.requestMap[idx] + if r.data == nil { + req = !r.queued + r.queued = true + if r.delivered == nil { + r.delivered = make(chan struct{}) + } + f.requestMap[idx] = r + } + f.requestLock.Unlock() + if req { + distCh <- distRequest{bloomIndex: f.bloomIndex, sectionIndex: idx} // success is guaranteed, distibuteRequests shuts down after fetch + } + select { + case <-stop: + return + case returnCh <- idx: + } + } + } + }() + + go func() { + defer wg.Done() + defer close(dataCh) + + for { + select { + case <-stop: + return + case idx, ok := <-returnCh: + if !ok { + return + } + + f.requestLock.RLock() + r := f.requestMap[idx] + f.requestLock.RUnlock() + + if r.data == nil { + select { + case <-stop: + return + case <-r.delivered: + f.requestLock.RLock() + r = f.requestMap[idx] + f.requestLock.RUnlock() + } + } + select { + case <-stop: + return + case dataCh <- r.data: + } + } + } + }() + + return dataCh +} + +// deliver is called by the request distributor when a reply to a request has +// arrived +func (f *fetcher) deliver(sectionIdxList []uint64, data [][]byte) { + f.requestLock.Lock() + defer f.requestLock.Unlock() + + for i, sectionIdx := range sectionIdxList { + r := f.requestMap[sectionIdx] + if r.data != nil { + panic("BloomBits section data delivered twice") + } + r.data = data[i] + close(r.delivered) + f.requestMap[sectionIdx] = 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 + + distCh chan distRequest + reqs map[uint][]uint64 + freeQueues map[uint]struct{} + allocQueue []chan uint + running bool + stop chan struct{} + lock sync.Mutex + wg, distWg sync.WaitGroup +} + +// NewMatcher creates a new Matcher instance +func NewMatcher(sectionSize uint64, addresses []common.Address, topics [][]common.Hash) *Matcher { + m := &Matcher{ + fetchers: make(map[uint]*fetcher), + reqs: make(map[uint][]uint64), + freeQueues: make(map[uint]struct{}), + distCh: make(chan distRequest, channelCap), + sectionSize: sectionSize, + } + m.setAddresses(addresses) + m.setTopics(topics) + return m +} + +// setAddresses matches only logs that are generated from addresses that are included +// in the given addresses. +func (m *Matcher) setAddresses(addresses []common.Address) { + m.addresses = make([]types.BloomIndexList, len(addresses)) + for i, address := range addresses { + m.addresses[i] = types.BloomIndexes(address.Bytes()) + } + + for _, bloomIndexList := range m.addresses { + for _, bloomIndex := range bloomIndexList { + m.newFetcher(bloomIndex) + } + } +} + +// 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, topic := range topicList { + if (topic == common.Hash{}) { + continue loop + } + t[i] = types.BloomIndexes(topic.Bytes()) + } + m.topics = append(m.topics, t) + } + + for _, bloomIndexLists := range m.topics { + for _, bloomIndexList := range bloomIndexLists { + for _, bloomIndex := range bloomIndexList { + m.newFetcher(bloomIndex) + } + } + } +} + +// 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(processCh chan partialMatches) chan partialMatches { + indexLists := m.topics + if len(m.addresses) > 0 { + indexLists = append([][]types.BloomIndexList{m.addresses}, indexLists...) + } + m.distributeRequests() + + for _, subIndexList := range indexLists { + processCh = m.subMatch(processCh, subIndexList) + } + return processCh +} + +// partialMatches with a non-nil vector represents a section in which some sub-matchers have already +// found potential matches. Subsequent sub-matchers will binary AND their matches with this vector. +// If vector is nil, it represents a section to be processed by the first sub-matcher. +type partialMatches struct { + sectionIndex uint64 + vector []byte +} + +// 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{ + bloomIndex: idx, + requestMap: make(map[uint64]fetchRequest), + } + 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 (processCh) 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(processCh chan partialMatches, bloomIndexLists []types.BloomIndexList) chan partialMatches { + // set up fetchers + fetchIndexChannels := make([][3]chan uint64, len(bloomIndexLists)) + fetchDataChannels := make([][3]chan []byte, len(bloomIndexLists)) + for i, bloomIndexList := range bloomIndexLists { + for j, bloomIndex := range bloomIndexList { + fetchIndexChannels[i][j] = make(chan uint64, channelCap) + fetchDataChannels[i][j] = m.fetchers[bloomIndex].fetch(fetchIndexChannels[i][j], m.distCh, m.stop, &m.wg) + } + } + + fetchedCh := make(chan partialMatches, channelCap) // entries from processCh are forwarded here after fetches have been initiated + resultsCh := make(chan partialMatches, channelCap) + + m.wg.Add(2) + // goroutine for starting retrievals + go func() { + defer m.wg.Done() + + for { + select { + case <-m.stop: + return + case s, ok := <-processCh: + if !ok { + close(fetchedCh) + for _, fetchIndexChs := range fetchIndexChannels { + for _, fetchIndexCh := range fetchIndexChs { + close(fetchIndexCh) + } + } + return + } + + for _, fetchIndexChs := range fetchIndexChannels { + for _, fetchIndexCh := range fetchIndexChs { + select { + case <-m.stop: + return + case fetchIndexCh <- s.sectionIndex: + } + } + } + select { + case <-m.stop: + return + case fetchedCh <- s: + } + } + } + }() + + // goroutine for processing retrieved data + go func() { + defer m.wg.Done() + + for { + select { + case <-m.stop: + return + case s, ok := <-fetchedCh: + if !ok { + close(resultsCh) + return + } + + var orVector []byte + for _, fetchDataChs := range fetchDataChannels { + var andVector []byte + for _, fetchDataCh := range fetchDataChs { + var data []byte + select { + case <-m.stop: + return + case data = <-fetchDataCh: + } + if andVector == nil { + andVector = make([]byte, int(m.sectionSize/8)) + copy(andVector, data) + } else { + bitutil.ANDBytes(andVector, andVector, data) + } + } + if orVector == nil { + orVector = andVector + } else { + bitutil.ORBytes(orVector, orVector, andVector) + } + } + + if orVector == nil { + orVector = make([]byte, int(m.sectionSize/8)) + } + if s.vector != nil { + bitutil.ANDBytes(orVector, orVector, s.vector) + } + if bitutil.TestBytes(orVector) { + select { + case <-m.stop: + return + case resultsCh <- partialMatches{s.sectionIndex, orVector}: + } + } + } + } + }() + + return resultsCh +} + +// Start starts the matching process and returns a stream of bloom matches in +// a given range of blocks. +// It returns a results channel immediately and stops if Stop is called or there +// are no more matches in the range (in which case the results channel is closed). +// Start/Stop can be called multiple times for different ranges, in which case already +// delivered bit vectors are not requested again. +func (m *Matcher) Start(begin, end uint64) chan uint64 { + m.stop = make(chan struct{}) + processCh := make(chan partialMatches, channelCap) + resultsCh := make(chan uint64, channelCap) + + res := m.match(processCh) + + startSection := begin / m.sectionSize + endSection := end / m.sectionSize + + m.wg.Add(2) + go func() { + defer m.wg.Done() + defer close(processCh) + + for i := startSection; i <= endSection; i++ { + select { + case processCh <- partialMatches{i, nil}: + case <-m.stop: + return + } + } + }() + + go func() { + defer m.wg.Done() + defer close(resultsCh) + + for { + select { + case r, ok := <-res: + if !ok { + return + } + sectionStart := r.sectionIndex * m.sectionSize + s := sectionStart + if begin > s { + s = begin + } + e := sectionStart + m.sectionSize - 1 + if end < e { + e = end + } + for i := s; i <= e; i++ { + b := r.vector[(i-sectionStart)/8] + bit := 7 - i%8 + if b != 0 { + if b&(1<<bit) != 0 { + select { + case <-m.stop: + return + case resultsCh <- i: + } + } + } else { + i += bit + } + } + + case <-m.stop: + return + } + } + }() + + return resultsCh +} + +// Stop stops the matching process +func (m *Matcher) Stop() { + close(m.stop) + m.distWg.Wait() +} + +// 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() { + m.distWg.Add(1) + stopDist := make(chan struct{}) + go func() { + <-m.stop + m.wg.Wait() + close(stopDist) + }() + + m.running = true + + go func() { + for { + select { + case r := <-m.distCh: + m.lock.Lock() + queue := m.reqs[r.bloomIndex] + i := 0 + for i < len(queue) && r.sectionIndex > queue[i] { + i++ + } + queue = append(queue, 0) + copy(queue[i+1:], queue[i:len(queue)-1]) + queue[i] = r.sectionIndex + m.reqs[r.bloomIndex] = queue + if len(queue) == 1 { + m.freeQueue(r.bloomIndex) + } + m.lock.Unlock() + case <-stopDist: + m.lock.Lock() + for _, ch := range m.allocQueue { + close(ch) + } + m.allocQueue = nil + m.running = false + m.lock.Unlock() + m.distWg.Done() + return + } + } + }() +} + +// freeQueue marks a queue as free if there are no AllocSectionQueue functions +// waiting for allocation. If there is someone waiting, the queue is immediately +// allocated. +func (m *Matcher) freeQueue(bloomIndex uint) { + if len(m.allocQueue) > 0 { + m.allocQueue[0] <- bloomIndex + m.allocQueue = m.allocQueue[1:] + } else { + m.freeQueues[bloomIndex] = struct{}{} + } +} + +// AllocSectionQueue allocates a queue of requested section indexes belonging to the same +// bloom bit index for a client process that can either immediately fetch the contents +// of the queue or wait a little while for more section indexes to be requested. +func (m *Matcher) AllocSectionQueue() (uint, bool) { + m.lock.Lock() + if !m.running { + m.lock.Unlock() + return 0, false + } + + var allocCh chan uint + if len(m.freeQueues) > 0 { + var ( + found bool + bestSection uint64 + bestIndex uint + ) + for bloomIndex, _ := range m.freeQueues { + if !found || m.reqs[bloomIndex][0] < bestSection { + found = true + bestIndex = bloomIndex + bestSection = m.reqs[bloomIndex][0] + } + } + delete(m.freeQueues, bestIndex) + m.lock.Unlock() + return bestIndex, true + } else { + allocCh = make(chan uint) + m.allocQueue = append(m.allocQueue, allocCh) + } + m.lock.Unlock() + + bloomIndex, ok := <-allocCh + return bloomIndex, ok +} + +// SectionCount returns the length of the section index queue belonging to the given bloom bit index +func (m *Matcher) SectionCount(bloomIndex uint) int { + m.lock.Lock() + defer m.lock.Unlock() + + return len(m.reqs[bloomIndex]) +} + +// FetchSections fetches all or part of an already allocated queue and deallocates it +func (m *Matcher) FetchSections(bloomIndex uint, maxCount int) []uint64 { + m.lock.Lock() + defer m.lock.Unlock() + + queue := m.reqs[bloomIndex] + if maxCount < len(queue) { + // return only part of the existing queue, mark the rest as free + m.reqs[bloomIndex] = queue[maxCount:] + m.freeQueue(bloomIndex) + return queue[:maxCount] + } else { + // return the entire queue + delete(m.reqs, bloomIndex) + return queue + } +} + +// Deliver delivers a bit vector to the appropriate fetcher. +// It is possible to deliver data even after Stop has been called. Once a vector has been +// requested, the matcher will keep waiting for delivery. +func (m *Matcher) Deliver(bloomIndex uint, sectionIdxList []uint64, data [][]byte) { + m.fetchers[bloomIndex].deliver(sectionIdxList, data) +} diff --git a/core/bloombits/matcher_test.go b/core/bloombits/matcher_test.go new file mode 100644 index 000000000..bef1491b8 --- /dev/null +++ b/core/bloombits/matcher_test.go @@ -0,0 +1,196 @@ +// 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" + "time" + + "github.com/ethereum/go-ethereum/core/types" +) + +const testSectionSize = 4096 + +func matcherTestVector(b uint, s uint64) []byte { + r := make([]byte, 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, maxRequestLen int) { + // serve matcher with test vectors + for i := 0; i < 10; i++ { + go func() { + for { + select { + case <-stop: + return + default: + } + b, ok := m.AllocSectionQueue() + if !ok { + return + } + if m.SectionCount(b) < maxRequestLen { + time.Sleep(time.Microsecond * 100) + } + s := m.FetchSections(b, maxRequestLen) + res := make([][]byte, 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, expCount uint32) uint32 { + count1 := testMatcherWithReqCount(t, idxs, cnt, stopOnMatches, expCount, 1) + count16 := testMatcherWithReqCount(t, idxs, cnt, stopOnMatches, expCount, 16) + if count1 != count16 { + t.Errorf("Error matching idxs = %v count = %v stopOnMatches = %v: request count mismatch, %v with maxReqCount = 1 vs. %v with maxReqCount = 16", idxs, cnt, stopOnMatches, count1, count16) + } + return count1 +} + +func testMatcherWithReqCount(t *testing.T, idxs [][]types.BloomIndexList, cnt uint64, stopOnMatches bool, expCount uint32, maxReqCount int) uint32 { + m := NewMatcher(testSectionSize, nil, nil) + + for _, idxss := range idxs { + for _, idxs := range idxss { + for _, idx := range idxs { + m.newFetcher(idx) + } + } + } + + m.addresses = idxs[0] + m.topics = idxs[1:] + var reqCount uint32 + + stop := make(chan struct{}) + chn := m.Start(0, cnt-1) + testServeMatcher(m, stop, &reqCount, maxReqCount) + + 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 { + m.Stop() + close(stop) + stop = make(chan struct{}) + chn = m.Start(i+1, cnt-1) + testServeMatcher(m, stop, &reqCount, maxReqCount) + } + } + } + match, ok := <-chn + if ok { + t.Errorf("Error matching idxs = %v count = %v stopOnMatches = %v: expected closed channel, got #%v", idxs, cnt, stopOnMatches, match) + } + m.Stop() + close(stop) + + if expCount != 0 && expCount != reqCount { + t.Errorf("Error matching idxs = %v count = %v stopOnMatches = %v: request count mismatch, expected #%v, got #%v", idxs, cnt, stopOnMatches, expCount, reqCount) + } + + return reqCount +} + +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}}}, 100000, false, 75) + testMatcher(t, [][]types.BloomIndexList{{{32, 3125, 100}}, {{40, 50, 10}}}, 100000, false, 81) + testMatcher(t, [][]types.BloomIndexList{{{4, 8, 11}, {7, 8, 17}}, {{9, 9, 12}, {15, 20, 13}}, {{18, 15, 15}, {12, 10, 4}}}, 10000, false, 36) +} + +func TestMatcherStopOnMatches(t *testing.T) { + testMatcher(t, [][]types.BloomIndexList{{{10, 20, 30}}}, 100000, true, 75) + testMatcher(t, [][]types.BloomIndexList{{{4, 8, 11}, {7, 8, 17}}, {{9, 9, 12}, {15, 20, 13}}, {{18, 15, 15}, {12, 10, 4}}}, 10000, true, 36) +} + +func TestMatcherRandom(t *testing.T) { + for i := 0; i < 20; i++ { + testMatcher(t, testRandomIdxs([]int{1}, 50), 100000, false, 0) + testMatcher(t, testRandomIdxs([]int{3}, 50), 100000, false, 0) + testMatcher(t, testRandomIdxs([]int{2, 2, 2}, 20), 100000, false, 0) + testMatcher(t, testRandomIdxs([]int{5, 5, 5}, 50), 100000, false, 0) + idxs := testRandomIdxs([]int{2, 2, 2}, 20) + reqCount := testMatcher(t, idxs, 10000, false, 0) + testMatcher(t, idxs, 10000, true, reqCount) + } +} diff --git a/core/bloombits/utils.go b/core/bloombits/utils.go new file mode 100644 index 000000000..d0755cb65 --- /dev/null +++ b/core/bloombits/utils.go @@ -0,0 +1,63 @@ +// Copyright 2017 The go-ethereum Authors +// This file is part of the go-ethereum library. +// +// The go-ethereum library is free software: you can redistribute it and/or modify +// it under the terms of the GNU Lesser General Public License as published by +// the Free Software Foundation, either version 3 of the License, or +// (at your option) any later version. +// +// The go-ethereum library is distributed in the hope that it will be useful, +// but WITHOUT ANY WARRANTY; without even the implied warranty of +// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +// GNU Lesser General Public License for more details. +// +// You should have received a copy of the GNU Lesser General Public License +// along with the go-ethereum library. If not, see <http://www.gnu.org/licenses/>. +package bloombits + +import ( + "github.com/ethereum/go-ethereum/core/types" +) + +const 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, bitIndex 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.bitIndex >= b.sectionSize { + panic("too many header blooms added") + } + + byteIdx := b.bitIndex / 8 + bitMask := byte(1) << byte(7-b.bitIndex%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.bitIndex++ +} + +// GetBitVector returns the bit vector belonging to the given bit index after header blooms have been added +func (b *BloomBitsCreator) GetBitVector(idx uint) []byte { + if b.bitIndex != b.sectionSize { + panic("not enough header blooms added") + } + + return b.blooms[idx][:] +} |