core/filtermaps: two dimensional log filter

Author: zsfelfoldi

core/filtermaps/filtermaps.go

@@ -0,0 +1,304 @@
+package filtermaps
+
+import (
+	"context"
+	"sync"
+
+	"github.com/ethereum/go-ethereum/common"
+	"github.com/ethereum/go-ethereum/core/types"
+)
+
+type Backend interface {
+	GetBlockLvPointer(ctx context.Context, blockNumber uint64) (uint64, error) // returns head lv pointer if blockNumber > head number
+	GetFilterMapRow(ctx context.Context, mapIndex, rowIndex uint32) (FilterRow, error)
+	GetLogByLvIndex(ctx context.Context, lvIndex uint64) (*types.Log, error)
+}
+
+func GetPotentialMatches(ctx context.Context, backend Backend, firstBlock, lastBlock uint64, addresses []common.Address, topics [][]common.Hash) ([]*types.Log, error) {
+	firstIndex, err := backend.GetBlockLvPointer(ctx, firstBlock)
+	if err != nil {
+		return nil, err
+	}
+	lastIndex, err := backend.GetBlockLvPointer(ctx, lastBlock+1)
+	if err != nil {
+		return nil, err
+	}
+	firstMap, lastMap := uint32(firstIndex>>logMapsPerEpoch), uint32(lastIndex>>logMapsPerEpoch)
+
+	matcher := make(matchSequence, len(topics)+1)
+	matchAddress := make(matchAny, len(addresses))
+	for i, address := range addresses {
+		matchAddress[i] = &singleMatcher{backend: backend, value: addressValue(address)}
+	}
+	matcher[0] = newCachedMatcher(matchAddress, firstMap)
+	for i, topicList := range topics {
+		matchTopic := make(matchAny, len(topicList))
+		for j, topic := range topicList {
+			matchTopic[j] = &singleMatcher{backend: backend, value: topicValue(topic)}
+		}
+		matcher[i+1] = newCachedMatcher(matchTopic, firstMap)
+	}
+
+	var logs []*types.Log
+	for mapIndex := firstMap; mapIndex <= lastMap; mapIndex++ {
+		//TODO parallelize, check total data size
+		matches, err := matcher.getMatches(ctx, mapIndex)
+		if err != nil {
+			return logs, err
+		}
+		matcher.clearUntil(mapIndex)
+		mlogs, err := getLogsFromMatches(ctx, backend, firstIndex, lastIndex, matches)
+		if err != nil {
+			return logs, err
+		}
+		logs = append(logs, mlogs...)
+	}
+	return logs, nil
+}
+
+type matcher interface {
+	getMatches(ctx context.Context, mapIndex uint32) (potentialMatches, error)
+	clearUntil(mapIndex uint32)
+}
+
+func getLogsFromMatches(ctx context.Context, backend Backend, firstIndex, lastIndex uint64, matches potentialMatches) ([]*types.Log, error) {
+	var logs []*types.Log
+	for _, match := range matches {
+		if match < firstIndex || match > lastIndex {
+			continue
+		}
+		log, err := backend.GetLogByLvIndex(ctx, match)
+		if err != nil {
+			return logs, err
+		}
+		logs = append(logs, log)
+	}
+	return logs, nil
+}
+
+// singleMatcher implements matcher
+type singleMatcher struct {
+	backend Backend
+	value   common.Hash
+}
+
+func (s *singleMatcher) getMatches(ctx context.Context, mapIndex uint32) (potentialMatches, error) {
+	filterRow, err := s.backend.GetFilterMapRow(ctx, mapIndex, rowIndex(mapIndex>>logMapsPerEpoch, s.value))
+	if err != nil {
+		return nil, err
+	}
+	return filterRow.potentialMatches(mapIndex, s.value), nil
+}
+
+func (s *singleMatcher) clearUntil(mapIndex uint32) {}
+
+// matchAny implements matcher
+type matchAny []matcher
+
+func (m matchAny) getMatches(ctx context.Context, mapIndex uint32) (potentialMatches, error) {
+	if len(m) == 0 {
+		return nil, nil
+	}
+	if len(m) == 1 {
+		return m[0].getMatches(ctx, mapIndex)
+	}
+	matches := make([]potentialMatches, len(m))
+	for i, matcher := range m {
+		var err error
+		if matches[i], err = matcher.getMatches(ctx, mapIndex); err != nil {
+			return nil, err
+		}
+	}
+	return mergeResults(matches), nil
+}
+
+func (m matchAny) clearUntil(mapIndex uint32) {
+	for _, matcher := range m {
+		matcher.clearUntil(mapIndex)
+	}
+}
+
+func mergeResults(results []potentialMatches) potentialMatches {
+	if len(results) == 0 {
+		return nil
+	}
+	var sumLen int
+	for _, res := range results {
+		if res == nil {
+			// nil is a wild card; all indices in map range are potential matches
+			return nil
+		}
+		sumLen += len(res)
+	}
+	merged := make(potentialMatches, 0, sumLen)
+	for {
+		best := -1
+		for i, res := range results {
+			if len(res) == 0 {
+				continue
+			}
+			if best < 0 || res[0] < results[best][0] {
+				best = i
+			}
+		}
+		if best < 0 {
+			return merged
+		}
+		if len(merged) == 0 || results[best][0] > merged[len(merged)-1] {
+			merged = append(merged, results[best][0])
+		}
+		results[best] = results[best][1:]
+	}
+}
+
+// matchSequence implements matcher
+type matchSequence []matcher
+
+func (m matchSequence) getMatches(ctx context.Context, mapIndex uint32) (potentialMatches, error) {
+	if len(m) == 0 {
+		return nil, nil
+	}
+	if len(m) == 1 {
+		return m[0].getMatches(ctx, mapIndex)
+	}
+	base, next, offset := m[:len(m)-1], m[len(m)-1], uint64(len(m)-1)
+	baseRes, err := base.getMatches(ctx, mapIndex)
+	if err != nil {
+		return nil, err
+	}
+	if baseRes != nil && len(baseRes) == 0 {
+		return baseRes, nil
+	}
+	nextRes, err := next.getMatches(ctx, mapIndex)
+	if err != nil {
+		return nil, err
+	}
+	nextNextRes, err := next.getMatches(ctx, mapIndex+1)
+	if err != nil {
+		return nil, err
+	}
+	return matchResults(mapIndex, offset, baseRes, nextRes, nextNextRes), nil
+}
+
+func (m matchSequence) clearUntil(mapIndex uint32) {
+	for _, matcher := range m {
+		matcher.clearUntil(mapIndex)
+	}
+}
+
+func matchResults(mapIndex uint32, offset uint64, baseRes, nextRes, nextNextRes potentialMatches) potentialMatches {
+	if nextRes == nil || (baseRes != nil && len(baseRes) == 0) {
+		return baseRes
+	}
+	if len(nextRes) > 0 {
+		start := 0
+		for start < len(nextRes) && nextRes[start] < uint64(mapIndex)<<logValuesPerMap+offset {
+			start++
+		}
+		nextRes = nextRes[start:]
+	}
+	if len(nextNextRes) > 0 {
+		stop := 0
+		for stop < len(nextNextRes) && nextNextRes[stop] < uint64(mapIndex+1)<<logValuesPerMap+offset {
+			stop++
+		}
+		nextNextRes = nextNextRes[:stop]
+	}
+	maxLen := len(nextRes) + len(nextNextRes)
+	if maxLen == 0 {
+		return nextRes
+	}
+	if len(baseRes) < maxLen {
+		maxLen = len(baseRes)
+	}
+	matchedRes := make(potentialMatches, 0, maxLen)
+	for _, res := range []potentialMatches{nextRes, nextNextRes} {
+		if baseRes != nil {
+			for len(res) > 0 && len(baseRes) > 0 {
+				if res[0] > baseRes[0]+offset {
+					baseRes = baseRes[1:]
+				} else if res[0] < baseRes[0]+offset {
+					res = res[1:]
+				} else {
+					matchedRes = append(matchedRes, baseRes[0])
+					res = res[1:]
+					baseRes = baseRes[1:]
+				}
+			}
+		} else {
+			for len(res) > 0 {
+				matchedRes = append(matchedRes, res[0]-offset)
+				res = res[1:]
+			}
+		}
+	}
+	return matchedRes
+}
+
+type cachedMatcher struct {
+	lock          sync.Mutex
+	matcher       matcher
+	cache         map[uint32]cachedMatches
+	clearedBefore uint32
+}
+
+func newCachedMatcher(m matcher, start uint32) *cachedMatcher {
+	return &cachedMatcher{
+		matcher:       m,
+		cache:         make(map[uint32]cachedMatches),
+		clearedBefore: start,
+	}
+}
+
+type cachedMatches struct {
+	matches potentialMatches
+	reqCh   chan struct{}
+}
+
+func (c *cachedMatcher) getMatches(ctx context.Context, mapIndex uint32) (potentialMatches, error) {
+	c.lock.Lock()
+	if mapIndex < c.clearedBefore {
+		panic("invalid cachedMatcher access")
+	}
+	cm, existed := c.cache[mapIndex]
+	if !existed {
+		cm = cachedMatches{reqCh: make(chan struct{})}
+		c.cache[mapIndex] = cm
+	}
+	c.lock.Unlock()
+	if existed {
+		if cm.matches != nil {
+			return cm.matches, nil
+		}
+		select {
+		case <-cm.reqCh:
+		case <-ctx.Done():
+			return nil, ctx.Err()
+		}
+		c.lock.Lock()
+		cm = c.cache[mapIndex]
+		c.lock.Unlock()
+		if cm.matches == nil {
+			panic("cached matches missing")
+		}
+		return cm.matches, nil
+	}
+	matches, err := c.matcher.getMatches(ctx, mapIndex)
+	if err != nil {
+		return nil, err
+	}
+	c.lock.Lock()
+	c.cache[mapIndex] = cachedMatches{matches: matches}
+	c.lock.Unlock()
+	close(cm.reqCh)
+	return matches, nil
+}
+
+func (c *cachedMatcher) clearUntil(mapIndex uint32) {
+	c.lock.Lock()
+	for c.clearedBefore <= mapIndex {
+		delete(c.cache, c.clearedBefore)
+		c.clearedBefore++
+	}
+	c.lock.Unlock()
+}

core/filtermaps/indexer.go

@@ -18,6 +18,8 @@ package rawdb
 
 import (
 	"bytes"
+	"encoding/binary"
+	"errors"
 	"math/big"
 
 	"github.com/ethereum/go-ethereum/common"
@@ -179,3 +181,79 @@ func DeleteBloombits(db ethdb.Database, bit uint, from uint64, to uint64) {
 		log.Crit("Failed to delete bloom bits", "err", it.Error())
 	}
 }
+
+func ReadFilterMapRow(db ethdb.KeyValueReader, mapRowIndex uint64) ([]byte, error) {
+	return db.Get(filterMapRowKey(mapRowIndex))
+}
+
+func WriteFilterMapRow(db ethdb.KeyValueWriter, mapRowIndex uint64, rowEnc []byte) {
+	if err := db.Put(filterMapRowKey(mapRowIndex), rowEnc); err != nil {
+		log.Crit("Failed to store filter map row", "err", err)
+	}
+}
+
+func DeleteFilterMapRow(db ethdb.KeyValueWriter, mapRowIndex uint64) {
+	if err := db.Delete(filterMapRowKey(mapRowIndex)); err != nil {
+		log.Crit("Failed to delete filter map row", "err", err)
+	}
+}
+
+func ReadFilterMapBlockPtr(db ethdb.KeyValueReader, mapIndex uint32) (uint64, error) {
+	encPtr, err := db.Get(filterMapBlockPtrKey(mapIndex))
+	if err != nil {
+		return 0, err
+	}
+	if len(encPtr) != 8 {
+		return 0, errors.New("Invalid block number encoding")
+	}
+	return binary.BigEndian.Uint64(encPtr), nil
+}
+
+func WriteFilterMapBlockPtr(db ethdb.KeyValueWriter, mapIndex uint32, blockNumber uint64) {
+	var encPtr [8]byte
+	binary.BigEndian.PutUint64(encPtr[:], blockNumber)
+	if err := db.Put(filterMapBlockPtrKey(mapIndex), encPtr[:]); err != nil {
+		log.Crit("Failed to store filter map block pointer", "err", err)
+	}
+}
+
+func DeleteFilterMapBlockPtr(db ethdb.KeyValueWriter, mapIndex uint32) {
+	if err := db.Delete(filterMapBlockPtrKey(mapIndex)); err != nil {
+		log.Crit("Failed to delete filter map block pointer", "err", err)
+	}
+}
+
+func ReadBlockLvPointer(db ethdb.KeyValueReader, blockNumber uint64) (uint64, error) {
+	encPtr, err := db.Get(blockLVKey(blockNumber))
+	if err != nil {
+		return 0, err
+	}
+	if len(encPtr) != 8 {
+		return 0, errors.New("Invalid log value pointer encoding")
+	}
+	return binary.BigEndian.Uint64(encPtr), nil
+}
+
+func WriteBlockLvPointer(db ethdb.KeyValueWriter, blockNumber, lvPointer uint64) {
+	var encPtr [8]byte
+	binary.BigEndian.PutUint64(encPtr[:], lvPointer)
+	if err := db.Put(blockLVKey(blockNumber), encPtr[:]); err != nil {
+		log.Crit("Failed to store block log value pointer", "err", err)
+	}
+}
+
+func DeleteBlockLvPointer(db ethdb.KeyValueWriter, blockNumber uint64) {
+	if err := db.Delete(blockLVKey(blockNumber)); err != nil {
+		log.Crit("Failed to delete block log value pointer", "err", err)
+	}
+}
+
+func ReadFilterMapsRange(db ethdb.KeyValueReader) ([]byte, error) {
+	return db.Get(filterMapsRangeKey)
+}
+
+func WriteFilterMapsRange(db ethdb.KeyValueWriter, encRange []byte) {
+	if err := db.Put(filterMapsRangeKey, encRange); err != nil {
+		log.Crit("Failed to store filter maps range", "err", err)
+	}
+}