path: root/core/bloombits/matcher.go

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