From 9857f126e168e68f4362ccc373f67fcf64847392 Mon Sep 17 00:00:00 2001 From: manjari Date: Fri, 16 Oct 2026 19:20:00 +0000 Subject: [PATCH 1/2] feat(conflict): Add Tango signature cache --- .../extension/conflict/tango/BUILD.bazel | 11 +- submitqueue/extension/conflict/tango/cache.go | 105 +++++++++++ .../extension/conflict/tango/cache_test.go | 166 ++++++++++++++++++ 3 files changed, 280 insertions(+), 2 deletions(-) create mode 100644 submitqueue/extension/conflict/tango/cache.go create mode 100644 submitqueue/extension/conflict/tango/cache_test.go diff --git a/submitqueue/extension/conflict/tango/BUILD.bazel b/submitqueue/extension/conflict/tango/BUILD.bazel index e9064c34..a48a755c 100644 --- a/submitqueue/extension/conflict/tango/BUILD.bazel +++ b/submitqueue/extension/conflict/tango/BUILD.bazel @@ -3,6 +3,7 @@ load("@rules_go//go:def.bzl", "go_library", "go_test") go_library( name = "go_default_library", srcs = [ + "cache.go", "mock_store.go", "signature.go", "store.go", @@ -16,7 +17,13 @@ go_library( go_test( name = "go_default_test", - srcs = ["signature_test.go"], + srcs = [ + "cache_test.go", + "signature_test.go", + ], embed = [":go_default_library"], - deps = ["@com_github_stretchr_testify//assert:go_default_library"], + deps = [ + "@com_github_stretchr_testify//assert:go_default_library", + "@com_github_stretchr_testify//require:go_default_library", + ], ) diff --git a/submitqueue/extension/conflict/tango/cache.go b/submitqueue/extension/conflict/tango/cache.go new file mode 100644 index 00000000..33c3e139 --- /dev/null +++ b/submitqueue/extension/conflict/tango/cache.go @@ -0,0 +1,105 @@ +// Copyright (c) 2026 Uber Technologies, Inc. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package tango + +import ( + "container/list" + "sync" +) + +// The budget counts memory: a SignatureTarget is 16 bytes (uint64 plus int32, +// padded), and the fixed overhead covers the entry, key and map bookkeeping. +const ( + signatureTargetBytes = 16 + signatureEntryOverhead = 256 +) + +type signatureCacheKey struct { + queue string + batchID string +} + +type signatureCacheEntry struct { + key signatureCacheKey + signature TargetSignature + bytes int64 +} + +// signatureCache is a byte-bounded LRU of target signatures keyed by queue and +// batch ID, safe for concurrent use. Signatures are immutable, so a cached +// value is shared with callers: neither side may modify Targets. +type signatureCache struct { + mu sync.Mutex + maxBytes int64 + usedBytes int64 + byRecency *list.List + entries map[signatureCacheKey]*list.Element +} + +func newSignatureCache(maxBytes int64) *signatureCache { + return &signatureCache{ + maxBytes: maxBytes, + byRecency: list.New(), + entries: make(map[signatureCacheKey]*list.Element), + } +} + +func signatureCacheBytes(signature TargetSignature) int64 { + return signatureEntryOverhead + + int64(len(signature.Queue)+len(signature.BatchID)+len(signature.BaseRevision)) + + int64(len(signature.Targets))*signatureTargetBytes +} + +// get returns the cached signature and marks it most recently used. +func (c *signatureCache) get(queue, batchID string) (TargetSignature, bool) { + c.mu.Lock() + defer c.mu.Unlock() + + element, ok := c.entries[signatureCacheKey{queue: queue, batchID: batchID}] + if !ok { + return TargetSignature{}, false + } + c.byRecency.MoveToFront(element) + return element.Value.(*signatureCacheEntry).signature, true +} + +// put caches signature, evicting least recently used entries to stay within the +// budget. A signature larger than the whole budget is not cached. +func (c *signatureCache) put(signature TargetSignature) { + size := signatureCacheBytes(signature) + if size > c.maxBytes { + return + } + key := signatureCacheKey{queue: signature.Queue, batchID: signature.BatchID} + + c.mu.Lock() + defer c.mu.Unlock() + + if element, ok := c.entries[key]; ok { + c.removeLocked(element) + } + c.entries[key] = c.byRecency.PushFront(&signatureCacheEntry{key: key, signature: signature, bytes: size}) + c.usedBytes += size + + for c.usedBytes > c.maxBytes { + c.removeLocked(c.byRecency.Back()) + } +} + +func (c *signatureCache) removeLocked(element *list.Element) { + entry := c.byRecency.Remove(element).(*signatureCacheEntry) + delete(c.entries, entry.key) + c.usedBytes -= entry.bytes +} diff --git a/submitqueue/extension/conflict/tango/cache_test.go b/submitqueue/extension/conflict/tango/cache_test.go new file mode 100644 index 00000000..a0765801 --- /dev/null +++ b/submitqueue/extension/conflict/tango/cache_test.go @@ -0,0 +1,166 @@ +// Copyright (c) 2026 Uber Technologies, Inc. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package tango + +import ( + "strconv" + "sync" + "testing" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +func cacheTestSignature(queue, batchID string, targets int) TargetSignature { + signature := TargetSignature{Queue: queue, BatchID: batchID, BaseRevision: "rev"} + for i := 0; i < targets; i++ { + signature.Targets = append(signature.Targets, SignatureTarget{ID: uint64(i), Distance: int32(i % 3)}) + } + return signature +} + +func TestSignatureCache_GetPut(t *testing.T) { + stored := cacheTestSignature("q", "1", 4) + + tests := []struct { + name string + put []TargetSignature + queue string + batchID string + wantFound bool + }{ + {name: "miss on empty cache", queue: "q", batchID: "1"}, + {name: "hit after put", put: []TargetSignature{stored}, queue: "q", batchID: "1", wantFound: true}, + {name: "same batch ID in another queue is a miss", put: []TargetSignature{stored}, queue: "other", batchID: "1"}, + {name: "other batch ID is a miss", put: []TargetSignature{stored}, queue: "q", batchID: "2"}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + cache := newSignatureCache(1 << 20) + for _, signature := range tt.put { + cache.put(signature) + } + + got, found := cache.get(tt.queue, tt.batchID) + + assert.Equal(t, tt.wantFound, found) + if tt.wantFound { + assert.Equal(t, stored, got) + } + }) + } +} + +func TestSignatureCache_Eviction(t *testing.T) { + a := cacheTestSignature("q", "a", 10) + b := cacheTestSignature("q", "b", 10) + c := cacheTestSignature("q", "c", 10) + budgetForTwo := signatureCacheBytes(a) + signatureCacheBytes(b) + + tests := []struct { + name string + run func(cache *signatureCache) + wantFound map[string]bool + }{ + { + name: "least recently put is evicted first", + run: func(cache *signatureCache) { + cache.put(a) + cache.put(b) + cache.put(c) + }, + wantFound: map[string]bool{"a": false, "b": true, "c": true}, + }, + { + name: "a read refreshes recency", + run: func(cache *signatureCache) { + cache.put(a) + cache.put(b) + cache.get("q", "a") + cache.put(c) + }, + wantFound: map[string]bool{"a": true, "b": false, "c": true}, + }, + { + name: "putting an existing key does not evict a neighbour", + run: func(cache *signatureCache) { + cache.put(a) + cache.put(b) + for i := 0; i < 100; i++ { + cache.put(a) + } + }, + wantFound: map[string]bool{"a": true, "b": true, "c": false}, + }, + { + name: "a signature larger than the budget is not cached", + run: func(cache *signatureCache) { + cache.put(a) + cache.put(cacheTestSignature("q", "c", 1000)) + }, + wantFound: map[string]bool{"a": true, "b": false, "c": false}, + }, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + cache := newSignatureCache(budgetForTwo) + + tt.run(cache) + + for batchID, want := range tt.wantFound { + _, found := cache.get("q", batchID) + assert.Equal(t, want, found, "batch %s", batchID) + } + }) + } +} + +func TestSignatureCache_ZeroBudgetCachesNothing(t *testing.T) { + cache := newSignatureCache(0) + + cache.put(cacheTestSignature("q", "1", 0)) + + _, found := cache.get("q", "1") + assert.False(t, found) +} + +func TestSignatureCache_ConcurrentUse(t *testing.T) { + const workers = 8 + const perWorker = 200 + sample := cacheTestSignature("q", "0", 5) + cache := newSignatureCache(signatureCacheBytes(sample) * 4) + + var wg sync.WaitGroup + for w := 0; w < workers; w++ { + wg.Add(1) + go func(w int) { + defer wg.Done() + for i := 0; i < perWorker; i++ { + batchID := strconv.Itoa((w + i) % 10) + cache.put(cacheTestSignature("q", batchID, 5)) + cache.get("q", batchID) + } + }(w) + } + wg.Wait() + + retained := 0 + for i := 0; i < 10; i++ { + if _, found := cache.get("q", strconv.Itoa(i)); found { + retained++ + } + } + require.LessOrEqual(t, retained, 4) +} From 1fafd1f23e87667b84d36fac44cf43abafc8d234 Mon Sep 17 00:00:00 2001 From: manjari Date: Fri, 16 Oct 2026 21:56:27 +0000 Subject: [PATCH 2/2] fix(conflict): Charge the signature cache by target capacity --- submitqueue/extension/conflict/tango/cache.go | 5 +++-- .../extension/conflict/tango/cache_test.go | 16 ++++++++++++++-- 2 files changed, 17 insertions(+), 4 deletions(-) diff --git a/submitqueue/extension/conflict/tango/cache.go b/submitqueue/extension/conflict/tango/cache.go index 33c3e139..aca0711a 100644 --- a/submitqueue/extension/conflict/tango/cache.go +++ b/submitqueue/extension/conflict/tango/cache.go @@ -20,7 +20,8 @@ import ( ) // The budget counts memory: a SignatureTarget is 16 bytes (uint64 plus int32, -// padded), and the fixed overhead covers the entry, key and map bookkeeping. +// padded) and the backing array is charged by capacity, and the fixed overhead +// covers the entry, key and map bookkeeping. const ( signatureTargetBytes = 16 signatureEntryOverhead = 256 @@ -59,7 +60,7 @@ func newSignatureCache(maxBytes int64) *signatureCache { func signatureCacheBytes(signature TargetSignature) int64 { return signatureEntryOverhead + int64(len(signature.Queue)+len(signature.BatchID)+len(signature.BaseRevision)) + - int64(len(signature.Targets))*signatureTargetBytes + int64(cap(signature.Targets))*signatureTargetBytes } // get returns the cached signature and marks it most recently used. diff --git a/submitqueue/extension/conflict/tango/cache_test.go b/submitqueue/extension/conflict/tango/cache_test.go index a0765801..2fb171af 100644 --- a/submitqueue/extension/conflict/tango/cache_test.go +++ b/submitqueue/extension/conflict/tango/cache_test.go @@ -102,7 +102,7 @@ func TestSignatureCache_Eviction(t *testing.T) { cache.put(a) } }, - wantFound: map[string]bool{"a": true, "b": true, "c": false}, + wantFound: map[string]bool{"a": true, "b": true}, }, { name: "a signature larger than the budget is not cached", @@ -110,7 +110,7 @@ func TestSignatureCache_Eviction(t *testing.T) { cache.put(a) cache.put(cacheTestSignature("q", "c", 1000)) }, - wantFound: map[string]bool{"a": true, "b": false, "c": false}, + wantFound: map[string]bool{"a": true, "c": false}, }, } for _, tt := range tests { @@ -127,6 +127,18 @@ func TestSignatureCache_Eviction(t *testing.T) { } } +func TestSignatureCache_ChargesSpareCapacity(t *testing.T) { + tight := cacheTestSignature("q", "tight", 10) + roomy := cacheTestSignature("q", "roomy", 10) + roomy.Targets = append(make([]SignatureTarget, 0, 1000), roomy.Targets...) + cache := newSignatureCache(signatureCacheBytes(tight) * 2) + + cache.put(roomy) + + _, found := cache.get("q", "roomy") + assert.False(t, found) +} + func TestSignatureCache_ZeroBudgetCachesNothing(t *testing.T) { cache := newSignatureCache(0)