mirror of
https://github.com/ethereum/go-ethereum.git
synced 2026-07-25 06:06:44 +00:00
core/filtermaps: get multiple filter map rows in a batch
This commit is contained in:
parent
dd8ea794a6
commit
31526bfad4
4 changed files with 94 additions and 62 deletions
|
|
@ -561,33 +561,60 @@ func (f *FilterMaps) getFilterMap(mapIndex uint32) (filterMap, error) {
|
||||||
}
|
}
|
||||||
fm := make(filterMap, f.mapHeight)
|
fm := make(filterMap, f.mapHeight)
|
||||||
for rowIndex := range fm {
|
for rowIndex := range fm {
|
||||||
var err error
|
rows, err := f.getFilterMapRows([]uint32{mapIndex}, uint32(rowIndex), false)
|
||||||
fm[rowIndex], err = f.getFilterMapRow(mapIndex, uint32(rowIndex), false)
|
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return nil, fmt.Errorf("failed to load filter map %d from database: %v", mapIndex, err)
|
return nil, fmt.Errorf("failed to load filter map %d from database: %v", mapIndex, err)
|
||||||
}
|
}
|
||||||
|
fm[rowIndex] = rows[0]
|
||||||
}
|
}
|
||||||
f.filterMapCache.Add(mapIndex, fm)
|
f.filterMapCache.Add(mapIndex, fm)
|
||||||
return fm, nil
|
return fm, nil
|
||||||
}
|
}
|
||||||
|
|
||||||
// getFilterMapRow fetches the given filter map row. If baseLayerOnly is true
|
// getFilterMapRows fetches a set of filter map rows at the corresponding map
|
||||||
// then only the first baseRowLength entries are returned.
|
// indices and a shared row index. If baseLayerOnly is true then only the first
|
||||||
func (f *FilterMaps) getFilterMapRow(mapIndex, rowIndex uint32, baseLayerOnly bool) (FilterRow, error) {
|
// baseRowLength entries are returned.
|
||||||
baseMapRowIndex := f.mapRowIndex(mapIndex&-f.baseRowGroupLength, rowIndex)
|
func (f *FilterMaps) getFilterMapRows(mapIndices []uint32, rowIndex uint32, baseLayerOnly bool) ([]FilterRow, error) {
|
||||||
|
rows := make([]FilterRow, len(mapIndices))
|
||||||
|
var ptr int
|
||||||
|
for len(mapIndices) > ptr {
|
||||||
|
baseMapIndex := mapIndices[ptr] & -f.baseRowGroupLength
|
||||||
|
groupLength := 1
|
||||||
|
for ptr+groupLength < len(mapIndices) && mapIndices[ptr+groupLength]&-f.baseRowGroupLength == baseMapIndex {
|
||||||
|
groupLength++
|
||||||
|
}
|
||||||
|
if err := f.getFilterMapRowsOfGroup(rows[ptr:ptr+groupLength], mapIndices[ptr:ptr+groupLength], rowIndex, baseLayerOnly); err != nil {
|
||||||
|
return nil, err
|
||||||
|
}
|
||||||
|
ptr += groupLength
|
||||||
|
}
|
||||||
|
return rows, nil
|
||||||
|
}
|
||||||
|
|
||||||
|
// getFilterMapRowsOfGroup fetches a set of filter map rows at map indices
|
||||||
|
// belonging to the same base row group.
|
||||||
|
func (f *FilterMaps) getFilterMapRowsOfGroup(target []FilterRow, mapIndices []uint32, rowIndex uint32, baseLayerOnly bool) error {
|
||||||
|
baseMapIndex := mapIndices[0] & -f.baseRowGroupLength
|
||||||
|
baseMapRowIndex := f.mapRowIndex(baseMapIndex, rowIndex)
|
||||||
baseRows, err := rawdb.ReadFilterMapBaseRows(f.db, baseMapRowIndex, f.baseRowGroupLength, f.logMapWidth)
|
baseRows, err := rawdb.ReadFilterMapBaseRows(f.db, baseMapRowIndex, f.baseRowGroupLength, f.logMapWidth)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return nil, fmt.Errorf("failed to retrieve filter map %d base rows %d: %v", mapIndex, rowIndex, err)
|
return fmt.Errorf("failed to retrieve filter map %d base rows %d: %v", baseMapIndex, rowIndex, err)
|
||||||
}
|
}
|
||||||
baseRow := baseRows[mapIndex&(f.baseRowGroupLength-1)]
|
for i, mapIndex := range mapIndices {
|
||||||
if baseLayerOnly {
|
if mapIndex&-f.baseRowGroupLength != baseMapIndex {
|
||||||
return baseRow, nil
|
panic("mapIndices are not in the same base row group")
|
||||||
|
}
|
||||||
|
row := baseRows[mapIndex&(f.baseRowGroupLength-1)]
|
||||||
|
if !baseLayerOnly {
|
||||||
|
extRow, err := rawdb.ReadFilterMapExtRow(f.db, f.mapRowIndex(mapIndex, rowIndex), f.logMapWidth)
|
||||||
|
if err != nil {
|
||||||
|
return fmt.Errorf("failed to retrieve filter map %d extended row %d: %v", mapIndex, rowIndex, err)
|
||||||
|
}
|
||||||
|
row = append(row, extRow...)
|
||||||
|
}
|
||||||
|
target[i] = row
|
||||||
}
|
}
|
||||||
extRow, err := rawdb.ReadFilterMapExtRow(f.db, f.mapRowIndex(mapIndex, rowIndex), f.logMapWidth)
|
return nil
|
||||||
if err != nil {
|
|
||||||
return nil, fmt.Errorf("failed to retrieve filter map %d extended row %d: %v", mapIndex, rowIndex, err)
|
|
||||||
}
|
|
||||||
return FilterRow(append(baseRow, extRow...)), nil
|
|
||||||
}
|
}
|
||||||
|
|
||||||
// storeFilterMapRows stores a set of filter map rows at the corresponding map
|
// storeFilterMapRows stores a set of filter map rows at the corresponding map
|
||||||
|
|
@ -617,7 +644,7 @@ func (f *FilterMaps) storeFilterMapRowsOfGroup(batch ethdb.Batch, mapIndices []u
|
||||||
var err error
|
var err error
|
||||||
baseRows, err = rawdb.ReadFilterMapBaseRows(f.db, baseMapRowIndex, f.baseRowGroupLength, f.logMapWidth)
|
baseRows, err = rawdb.ReadFilterMapBaseRows(f.db, baseMapRowIndex, f.baseRowGroupLength, f.logMapWidth)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return fmt.Errorf("failed to retrieve filter map %d base rows %d for modification: %v", mapIndices[0]&-f.baseRowGroupLength, rowIndex, err)
|
return fmt.Errorf("failed to retrieve filter map %d base rows %d for modification: %v", baseMapIndex, rowIndex, err)
|
||||||
}
|
}
|
||||||
} else {
|
} else {
|
||||||
baseRows = make([][]uint32, f.baseRowGroupLength)
|
baseRows = make([][]uint32, f.baseRowGroupLength)
|
||||||
|
|
|
||||||
|
|
@ -428,10 +428,12 @@ func (ts *testSetup) matcherViewHash() common.Hash {
|
||||||
binary.LittleEndian.PutUint32(enc[:4], r)
|
binary.LittleEndian.PutUint32(enc[:4], r)
|
||||||
for m := uint32(0); m <= headMap; m++ {
|
for m := uint32(0); m <= headMap; m++ {
|
||||||
binary.LittleEndian.PutUint32(enc[4:8], m)
|
binary.LittleEndian.PutUint32(enc[4:8], m)
|
||||||
row, _ := mb.GetFilterMapRow(ctx, m, r, false)
|
rows, _ := mb.GetFilterMapRows(ctx, []uint32{m}, r, false)
|
||||||
for _, v := range row {
|
for _, row := range rows {
|
||||||
binary.LittleEndian.PutUint32(enc[8:], v)
|
for _, v := range row {
|
||||||
hasher.Write(enc[:])
|
binary.LittleEndian.PutUint32(enc[8:], v)
|
||||||
|
hasher.Write(enc[:])
|
||||||
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
|
||||||
|
|
@ -20,7 +20,6 @@ import (
|
||||||
"context"
|
"context"
|
||||||
"errors"
|
"errors"
|
||||||
"fmt"
|
"fmt"
|
||||||
"math"
|
|
||||||
"sync"
|
"sync"
|
||||||
"sync/atomic"
|
"sync/atomic"
|
||||||
"time"
|
"time"
|
||||||
|
|
@ -45,7 +44,7 @@ var ErrMatchAll = errors.New("match all patterns not supported")
|
||||||
type MatcherBackend interface {
|
type MatcherBackend interface {
|
||||||
GetParams() *Params
|
GetParams() *Params
|
||||||
GetBlockLvPointer(ctx context.Context, blockNumber uint64) (uint64, error)
|
GetBlockLvPointer(ctx context.Context, blockNumber uint64) (uint64, error)
|
||||||
GetFilterMapRow(ctx context.Context, mapIndex, rowIndex uint32, baseLayerOnly bool) (FilterRow, error)
|
GetFilterMapRows(ctx context.Context, mapIndices []uint32, rowIndex uint32, baseLayerOnly bool) ([]FilterRow, error)
|
||||||
GetLogByLvIndex(ctx context.Context, lvIndex uint64) (*types.Log, error)
|
GetLogByLvIndex(ctx context.Context, lvIndex uint64) (*types.Log, error)
|
||||||
SyncLogIndex(ctx context.Context) (SyncRange, error)
|
SyncLogIndex(ctx context.Context) (SyncRange, error)
|
||||||
Close()
|
Close()
|
||||||
|
|
@ -365,50 +364,57 @@ func (m *singleMatcherInstance) getMatchesForLayer(ctx context.Context, layerInd
|
||||||
var st int
|
var st int
|
||||||
m.stats.setState(&st, stOther)
|
m.stats.setState(&st, stOther)
|
||||||
params := m.backend.GetParams()
|
params := m.backend.GetParams()
|
||||||
maskedMapIndex, rowIndex := uint32(math.MaxUint32), uint32(0)
|
var ptr int
|
||||||
for _, mapIndex := range m.mapIndices {
|
for len(m.mapIndices) > ptr {
|
||||||
filterRows, ok := m.filterRows[mapIndex]
|
// find next group of map indices mapped onto the same row
|
||||||
if !ok {
|
maskedMapIndex := params.maskedMapIndex(m.mapIndices[ptr], layerIndex)
|
||||||
continue
|
rowIndex := params.rowIndex(m.mapIndices[ptr], layerIndex, m.value)
|
||||||
}
|
groupLength := 1
|
||||||
if mm := params.maskedMapIndex(mapIndex, layerIndex); mm != maskedMapIndex {
|
for ptr+groupLength < len(m.mapIndices) && params.maskedMapIndex(m.mapIndices[ptr+groupLength], layerIndex) == maskedMapIndex {
|
||||||
// only recalculate rowIndex when necessary
|
groupLength++
|
||||||
maskedMapIndex = mm
|
|
||||||
rowIndex = params.rowIndex(mapIndex, layerIndex, m.value)
|
|
||||||
}
|
}
|
||||||
if layerIndex == 0 {
|
if layerIndex == 0 {
|
||||||
m.stats.setState(&st, stFetchFirst)
|
m.stats.setState(&st, stFetchFirst)
|
||||||
} else {
|
} else {
|
||||||
m.stats.setState(&st, stFetchMore)
|
m.stats.setState(&st, stFetchMore)
|
||||||
}
|
}
|
||||||
filterRow, err := m.backend.GetFilterMapRow(ctx, mapIndex, rowIndex, layerIndex == 0)
|
groupRows, err := m.backend.GetFilterMapRows(ctx, m.mapIndices[ptr:ptr+groupLength], rowIndex, layerIndex == 0)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
m.stats.setState(&st, stNone)
|
m.stats.setState(&st, stNone)
|
||||||
return nil, fmt.Errorf("failed to retrieve filter map %d row %d: %v", mapIndex, rowIndex, err)
|
return nil, fmt.Errorf("failed to retrieve filter map %d row %d: %v", m.mapIndices[ptr], rowIndex, err)
|
||||||
}
|
}
|
||||||
if layerIndex == 0 {
|
|
||||||
matchBaseRowAccessMeter.Mark(1)
|
|
||||||
matchBaseRowSizeMeter.Mark(int64(len(filterRow)))
|
|
||||||
} else {
|
|
||||||
matchExtRowAccessMeter.Mark(1)
|
|
||||||
matchExtRowSizeMeter.Mark(int64(len(filterRow)))
|
|
||||||
}
|
|
||||||
m.stats.addAmount(st, int64(len(filterRow)))
|
|
||||||
m.stats.setState(&st, stOther)
|
m.stats.setState(&st, stOther)
|
||||||
filterRows = append(filterRows, filterRow)
|
for i := range groupLength {
|
||||||
if uint32(len(filterRow)) < params.maxRowLength(layerIndex) {
|
mapIndex := m.mapIndices[ptr+i]
|
||||||
m.stats.setState(&st, stProcess)
|
filterRow := groupRows[i]
|
||||||
matches := params.potentialMatches(filterRows, mapIndex, m.value)
|
filterRows, ok := m.filterRows[mapIndex]
|
||||||
m.stats.addAmount(st, int64(len(matches)))
|
if !ok {
|
||||||
results = append(results, matcherResult{
|
panic("dropped map in mapIndices")
|
||||||
mapIndex: mapIndex,
|
}
|
||||||
matches: matches,
|
if layerIndex == 0 {
|
||||||
})
|
matchBaseRowAccessMeter.Mark(1)
|
||||||
m.stats.setState(&st, stOther)
|
matchBaseRowSizeMeter.Mark(int64(len(filterRow)))
|
||||||
delete(m.filterRows, mapIndex)
|
} else {
|
||||||
} else {
|
matchExtRowAccessMeter.Mark(1)
|
||||||
m.filterRows[mapIndex] = filterRows
|
matchExtRowSizeMeter.Mark(int64(len(filterRow)))
|
||||||
|
}
|
||||||
|
m.stats.addAmount(st, int64(len(filterRow)))
|
||||||
|
filterRows = append(filterRows, filterRow)
|
||||||
|
if uint32(len(filterRow)) < params.maxRowLength(layerIndex) {
|
||||||
|
m.stats.setState(&st, stProcess)
|
||||||
|
matches := params.potentialMatches(filterRows, mapIndex, m.value)
|
||||||
|
m.stats.addAmount(st, int64(len(matches)))
|
||||||
|
results = append(results, matcherResult{
|
||||||
|
mapIndex: mapIndex,
|
||||||
|
matches: matches,
|
||||||
|
})
|
||||||
|
m.stats.setState(&st, stOther)
|
||||||
|
delete(m.filterRows, mapIndex)
|
||||||
|
} else {
|
||||||
|
m.filterRows[mapIndex] = filterRows
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
ptr += groupLength
|
||||||
}
|
}
|
||||||
m.cleanMapIndices()
|
m.cleanMapIndices()
|
||||||
m.stats.setState(&st, stNone)
|
m.stats.setState(&st, stNone)
|
||||||
|
|
|
||||||
|
|
@ -67,18 +67,15 @@ func (fm *FilterMapsMatcherBackend) Close() {
|
||||||
delete(fm.f.matchers, fm)
|
delete(fm.f.matchers, fm)
|
||||||
}
|
}
|
||||||
|
|
||||||
// GetFilterMapRow returns the given row of the given map. If the row is empty
|
// GetFilterMapRows returns the given row of the given map. If the row is empty
|
||||||
// then a non-nil zero length row is returned. If baseLayerOnly is true then
|
// then a non-nil zero length row is returned. If baseLayerOnly is true then
|
||||||
// only the first baseRowLength entries of the row are guaranteed to be
|
// only the first baseRowLength entries of the row are guaranteed to be
|
||||||
// returned.
|
// returned.
|
||||||
// Note that the returned slices should not be modified, they should be copied
|
// Note that the returned slices should not be modified, they should be copied
|
||||||
// on write.
|
// on write.
|
||||||
// GetFilterMapRow implements MatcherBackend.
|
// GetFilterMapRows implements MatcherBackend.
|
||||||
func (fm *FilterMapsMatcherBackend) GetFilterMapRow(ctx context.Context, mapIndex, rowIndex uint32, baseLayerOnly bool) (FilterRow, error) {
|
func (fm *FilterMapsMatcherBackend) GetFilterMapRows(ctx context.Context, mapIndices []uint32, rowIndex uint32, baseLayerOnly bool) ([]FilterRow, error) {
|
||||||
fm.f.indexLock.RLock()
|
return fm.f.getFilterMapRows(mapIndices, rowIndex, baseLayerOnly)
|
||||||
defer fm.f.indexLock.RUnlock()
|
|
||||||
|
|
||||||
return fm.f.getFilterMapRow(mapIndex, rowIndex, baseLayerOnly)
|
|
||||||
}
|
}
|
||||||
|
|
||||||
// GetBlockLvPointer returns the starting log value index where the log values
|
// GetBlockLvPointer returns the starting log value index where the log values
|
||||||
|
|
|
||||||
Loading…
Reference in a new issue