Skip to content

use context in indexer go routine instead of context.TODO #1491

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
merged 3 commits into from
Aug 28, 2025
Merged

use context in indexer go routine instead of context.TODO #1491

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
cff0f01
fixed bug of unprotected concurrent writes to a map in prefix indexer
nirrozenbaum Aug 28, 2025
13b05ef
revert locking changes in indexer
nirrozenbaum Aug 28, 2025
05d90aa
replace context.TODO with a context
nirrozenbaum Aug 28, 2025
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
14 changes: 7 additions & 7 deletions pkg/epp/scheduling/framework/plugins/multi/prefix/indexer.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -38,15 +38,15 @@ type indexer struct {
}

// newIndexer initializes an indexer with size limits and starts cache size reporting.
func newIndexer(maxLRUSize int) *indexer {
ix := &indexer{
func newIndexer(ctx context.Context, maxLRUSize int) *indexer {
indexer := &indexer{
hashToPods: make(map[BlockHash]podSet),
podToLRU: make(map[ServerID]*lru.Cache[BlockHash, struct{}]),
maxLRUSize: maxLRUSize,
}

go ix.ReportLRUSize(time.Second)
return ix
go indexer.reportLRUSize(ctx, time.Second)
return indexer
}

// Add adds a list of prefix hashes to the cache, tied to the server.
Expand Down Expand Up @@ -111,8 +111,8 @@ func (i *indexer) makeEvictionFn(pod ServerID) func(BlockHash, struct{}) {
}
}

// ReportLRUSize starts a goroutine that periodically reports the LRU cache size metric.
func (i *indexer) ReportLRUSize(interval time.Duration) {
// reportLRUSize starts a goroutine that periodically reports the LRU cache size metric.
func (i *indexer) reportLRUSize(ctx context.Context, interval time.Duration) {
ticker := time.NewTicker(interval)
defer ticker.Stop()
for range ticker.C {
Expand All @@ -137,7 +137,7 @@ func (i *indexer) ReportLRUSize(interval time.Duration) {
}

metrics.RecordPrefixCacheSize(int64(totalEntries))
log.FromContext(context.TODO()).V(logutil.TRACE).Info("Prefix cache state",
log.FromContext(ctx).V(logutil.TRACE).Info("Prefix cache state",
"total entries", totalEntries,
"# pods", numPods,
"avg entries per pod", avg,
Expand Down
3 changes: 2 additions & 1 deletion pkg/epp/scheduling/framework/plugins/multi/prefix/indexer_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -16,13 +16,14 @@ limitations under the License.
package prefix

import (
"context"
"testing"

"github.com/stretchr/testify/assert"
)

func TestIndexer_AddAndGet(t *testing.T) {
i := newIndexer(2)
i := newIndexer(context.Background(), 2)

hash1 := BlockHash(1)
server := ServerID{Namespace: "default", Name: "server1"}
Expand Down
4 changes: 2 additions & 2 deletions pkg/epp/scheduling/framework/plugins/multi/prefix/plugin.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -148,7 +148,7 @@ func New(ctx context.Context, config Config) *Plugin {
capacity := config.LRUCapacityPerServer
if capacity <= 0 {
capacity = DefaultLRUCapacityPerServer
log.FromContext(context.TODO()).V(logutil.DEFAULT).Info(
log.FromContext(ctx).V(logutil.DEFAULT).Info(
"LRUCapacityPerServer is not positive, using default value",
"defaultCapacity", DefaultLRUCapacityPerServer,
)
Expand All @@ -158,7 +158,7 @@ func New(ctx context.Context, config Config) *Plugin {
typedName: plugins.TypedName{Type: PrefixCachePluginType, Name: PrefixCachePluginType},
config: config,
pluginState: plugins.NewPluginState(ctx),
indexer: newIndexer(capacity),
indexer: newIndexer(ctx, capacity),
}
}

Expand Down