From 31526bfad4532e4917b504aec21d43ad783a03f0 Mon Sep 17 00:00:00 2001 From: Zsolt Felfoldi Date: Thu, 15 May 2025 01:52:11 +0200 Subject: [PATCH] core/filtermaps: get multiple filter map rows in a batch --- core/filtermaps/filtermaps.go | 59 ++++++++++++++++------- core/filtermaps/indexer_test.go | 10 ++-- core/filtermaps/matcher.go | 76 ++++++++++++++++-------------- core/filtermaps/matcher_backend.go | 11 ++--- 4 files changed, 94 insertions(+), 62 deletions(-) diff --git a/core/filtermaps/filtermaps.go b/core/filtermaps/filtermaps.go index 6948c8b0e1..cf6ce00ced 100644 --- a/core/filtermaps/filtermaps.go +++ b/core/filtermaps/filtermaps.go @@ -561,33 +561,60 @@ func (f *FilterMaps) getFilterMap(mapIndex uint32) (filterMap, error) { } fm := make(filterMap, f.mapHeight) for rowIndex := range fm { - var err error - fm[rowIndex], err = f.getFilterMapRow(mapIndex, uint32(rowIndex), false) + rows, err := f.getFilterMapRows([]uint32{mapIndex}, uint32(rowIndex), false) if err != nil { return nil, fmt.Errorf("failed to load filter map %d from database: %v", mapIndex, err) } + fm[rowIndex] = rows[0] } f.filterMapCache.Add(mapIndex, fm) return fm, nil } -// getFilterMapRow fetches the given filter map row. If baseLayerOnly is true -// then only the first baseRowLength entries are returned. -func (f *FilterMaps) getFilterMapRow(mapIndex, rowIndex uint32, baseLayerOnly bool) (FilterRow, error) { - baseMapRowIndex := f.mapRowIndex(mapIndex&-f.baseRowGroupLength, rowIndex) +// getFilterMapRows fetches a set of filter map rows at the corresponding map +// indices and a shared row index. If baseLayerOnly is true then only the first +// baseRowLength entries are returned. +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) 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)] - if baseLayerOnly { - return baseRow, nil + for i, mapIndex := range mapIndices { + if mapIndex&-f.baseRowGroupLength != baseMapIndex { + 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) - 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 + return nil } // 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 baseRows, err = rawdb.ReadFilterMapBaseRows(f.db, baseMapRowIndex, f.baseRowGroupLength, f.logMapWidth) 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 { baseRows = make([][]uint32, f.baseRowGroupLength) diff --git a/core/filtermaps/indexer_test.go b/core/filtermaps/indexer_test.go index 5ed397a90e..4144f4cccc 100644 --- a/core/filtermaps/indexer_test.go +++ b/core/filtermaps/indexer_test.go @@ -428,10 +428,12 @@ func (ts *testSetup) matcherViewHash() common.Hash { binary.LittleEndian.PutUint32(enc[:4], r) for m := uint32(0); m <= headMap; m++ { binary.LittleEndian.PutUint32(enc[4:8], m) - row, _ := mb.GetFilterMapRow(ctx, m, r, false) - for _, v := range row { - binary.LittleEndian.PutUint32(enc[8:], v) - hasher.Write(enc[:]) + rows, _ := mb.GetFilterMapRows(ctx, []uint32{m}, r, false) + for _, row := range rows { + for _, v := range row { + binary.LittleEndian.PutUint32(enc[8:], v) + hasher.Write(enc[:]) + } } } } diff --git a/core/filtermaps/matcher.go b/core/filtermaps/matcher.go index a5eeaaa5f0..82ea2ceee6 100644 --- a/core/filtermaps/matcher.go +++ b/core/filtermaps/matcher.go @@ -20,7 +20,6 @@ import ( "context" "errors" "fmt" - "math" "sync" "sync/atomic" "time" @@ -45,7 +44,7 @@ var ErrMatchAll = errors.New("match all patterns not supported") type MatcherBackend interface { GetParams() *Params 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) SyncLogIndex(ctx context.Context) (SyncRange, error) Close() @@ -365,50 +364,57 @@ func (m *singleMatcherInstance) getMatchesForLayer(ctx context.Context, layerInd var st int m.stats.setState(&st, stOther) params := m.backend.GetParams() - maskedMapIndex, rowIndex := uint32(math.MaxUint32), uint32(0) - for _, mapIndex := range m.mapIndices { - filterRows, ok := m.filterRows[mapIndex] - if !ok { - continue - } - if mm := params.maskedMapIndex(mapIndex, layerIndex); mm != maskedMapIndex { - // only recalculate rowIndex when necessary - maskedMapIndex = mm - rowIndex = params.rowIndex(mapIndex, layerIndex, m.value) + var ptr int + for len(m.mapIndices) > ptr { + // find next group of map indices mapped onto the same row + maskedMapIndex := params.maskedMapIndex(m.mapIndices[ptr], layerIndex) + rowIndex := params.rowIndex(m.mapIndices[ptr], layerIndex, m.value) + groupLength := 1 + for ptr+groupLength < len(m.mapIndices) && params.maskedMapIndex(m.mapIndices[ptr+groupLength], layerIndex) == maskedMapIndex { + groupLength++ } if layerIndex == 0 { m.stats.setState(&st, stFetchFirst) } else { 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 { 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) - 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 + for i := range groupLength { + mapIndex := m.mapIndices[ptr+i] + filterRow := groupRows[i] + filterRows, ok := m.filterRows[mapIndex] + if !ok { + panic("dropped map in mapIndices") + } + 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))) + 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.stats.setState(&st, stNone) diff --git a/core/filtermaps/matcher_backend.go b/core/filtermaps/matcher_backend.go index e19a63e18b..abd0fd9283 100644 --- a/core/filtermaps/matcher_backend.go +++ b/core/filtermaps/matcher_backend.go @@ -67,18 +67,15 @@ func (fm *FilterMapsMatcherBackend) Close() { 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 // only the first baseRowLength entries of the row are guaranteed to be // returned. // Note that the returned slices should not be modified, they should be copied // on write. -// GetFilterMapRow implements MatcherBackend. -func (fm *FilterMapsMatcherBackend) GetFilterMapRow(ctx context.Context, mapIndex, rowIndex uint32, baseLayerOnly bool) (FilterRow, error) { - fm.f.indexLock.RLock() - defer fm.f.indexLock.RUnlock() - - return fm.f.getFilterMapRow(mapIndex, rowIndex, baseLayerOnly) +// GetFilterMapRows implements MatcherBackend. +func (fm *FilterMapsMatcherBackend) GetFilterMapRows(ctx context.Context, mapIndices []uint32, rowIndex uint32, baseLayerOnly bool) ([]FilterRow, error) { + return fm.f.getFilterMapRows(mapIndices, rowIndex, baseLayerOnly) } // GetBlockLvPointer returns the starting log value index where the log values