Skip to content

Commit c7a67dd

Browse files
manjari25manjari25
authored
feat(conflict): Add Tango signature cache (#882)
## Why? The Tango-backed conflict analyzer compares a new batch against the signature of every in-flight batch. Reading all of those from the signature store on every analysis would be slow, especially after a restart. This PR adds the in-memory cache that sits in front of the store. ## What? Everything is added under `submitqueue/extension/conflict/tango`: - Adds a byte-bounded LRU cache of `TargetSignature`s, keyed by queue and batch ID. One cache serves every queue, with a single memory budget, so a busy queue can use space an idle one left behind. - A lookup refreshes an entry's recency. Adding a signature evicts the least recently used entries until the cache is back within budget. A signature larger than the whole budget is not cached, and a zero budget caches nothing. - Entries are sized from their targets (16 bytes each) plus a fixed overhead. The figure is an estimate that keeps the budget tied to memory use. - Signatures are immutable, so a cached value is shared with callers instead of copied. Signatures can have millions of targets, and copying them on every read would defeat the cache. Nobody may modify `Targets`. - A miss costs one store read, never a call to Tango, so evicting an in-flight batch is safe. That is why there is no explicit eviction when a batch leaves the in-flight set. - The cache is unexported and nothing uses it yet. The analyzer that uses it is a later PR in this stack. - Tests cover hits and misses, queue, a read refreshing recency, repeated puts of one key not evicting a neighbour, an oversized signature, a zero budget, and a concurrent get and put test under `-race`. ## Test Plan `./tool/bazel test //submitqueue/extension/conflict/tango/...` `go test -race ./submitqueue/extension/conflict/tango/` `make check-gazelle` `make check-mocks` `make lint` CI ## Issue
1 parent aa574f4 commit c7a67dd

3 files changed

Lines changed: 293 additions & 2 deletions

File tree

‎submitqueue/extension/conflict/tango/BUILD.bazel‎

Lines changed: 9 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -3,6 +3,7 @@ load("@rules_go//go:def.bzl", "go_library", "go_test")
33
go_library(
44
name = "go_default_library",
55
srcs = [
6+
"cache.go",
67
"mock_store.go",
78
"signature.go",
89
"store.go",
@@ -16,7 +17,13 @@ go_library(
1617

1718
go_test(
1819
name = "go_default_test",
19-
srcs = ["signature_test.go"],
20+
srcs = [
21+
"cache_test.go",
22+
"signature_test.go",
23+
],
2024
embed = [":go_default_library"],
21-
deps = ["@com_github_stretchr_testify//assert:go_default_library"],
25+
deps = [
26+
"@com_github_stretchr_testify//assert:go_default_library",
27+
"@com_github_stretchr_testify//require:go_default_library",
28+
],
2229
)

‎submitqueue/extension/conflict/tango/cache.go‎

Lines changed: 106 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,106 @@
1+
// Copyright (c) 2026 Uber Technologies, Inc.
2+
//
3+
// Licensed under the Apache License, Version 2.0 (the "License");
4+
// you may not use this file except in compliance with the License.
5+
// You may obtain a copy of the License at
6+
//
7+
// http://www.apache.org/licenses/LICENSE-2.0
8+
//
9+
// Unless required by applicable law or agreed to in writing, software
10+
// distributed under the License is distributed on an "AS IS" BASIS,
11+
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12+
// See the License for the specific language governing permissions and
13+
// limitations under the License.
14+
15+
package tango
16+
17+
import (
18+
"container/list"
19+
"sync"
20+
)
21+
22+
// The budget counts memory: a SignatureTarget is 16 bytes (uint64 plus int32,
23+
// padded) and the backing array is charged by capacity, and the fixed overhead
24+
// covers the entry, key and map bookkeeping.
25+
const (
26+
signatureTargetBytes = 16
27+
signatureEntryOverhead = 256
28+
)
29+
30+
type signatureCacheKey struct {
31+
queue string
32+
batchID string
33+
}
34+
35+
type signatureCacheEntry struct {
36+
key signatureCacheKey
37+
signature TargetSignature
38+
bytes int64
39+
}
40+
41+
// signatureCache is a byte-bounded LRU of target signatures keyed by queue and
42+
// batch ID, safe for concurrent use. Signatures are immutable, so a cached
43+
// value is shared with callers: neither side may modify Targets.
44+
type signatureCache struct {
45+
mu sync.Mutex
46+
maxBytes int64
47+
usedBytes int64
48+
byRecency *list.List
49+
entries map[signatureCacheKey]*list.Element
50+
}
51+
52+
func newSignatureCache(maxBytes int64) *signatureCache {
53+
return &signatureCache{
54+
maxBytes: maxBytes,
55+
byRecency: list.New(),
56+
entries: make(map[signatureCacheKey]*list.Element),
57+
}
58+
}
59+
60+
func signatureCacheBytes(signature TargetSignature) int64 {
61+
return signatureEntryOverhead +
62+
int64(len(signature.Queue)+len(signature.BatchID)+len(signature.BaseRevision)) +
63+
int64(cap(signature.Targets))*signatureTargetBytes
64+
}
65+
66+
// get returns the cached signature and marks it most recently used.
67+
func (c *signatureCache) get(queue, batchID string) (TargetSignature, bool) {
68+
c.mu.Lock()
69+
defer c.mu.Unlock()
70+
71+
element, ok := c.entries[signatureCacheKey{queue: queue, batchID: batchID}]
72+
if !ok {
73+
return TargetSignature{}, false
74+
}
75+
c.byRecency.MoveToFront(element)
76+
return element.Value.(*signatureCacheEntry).signature, true
77+
}
78+
79+
// put caches signature, evicting least recently used entries to stay within the
80+
// budget. A signature larger than the whole budget is not cached.
81+
func (c *signatureCache) put(signature TargetSignature) {
82+
size := signatureCacheBytes(signature)
83+
if size > c.maxBytes {
84+
return
85+
}
86+
key := signatureCacheKey{queue: signature.Queue, batchID: signature.BatchID}
87+
88+
c.mu.Lock()
89+
defer c.mu.Unlock()
90+
91+
if element, ok := c.entries[key]; ok {
92+
c.removeLocked(element)
93+
}
94+
c.entries[key] = c.byRecency.PushFront(&signatureCacheEntry{key: key, signature: signature, bytes: size})
95+
c.usedBytes += size
96+
97+
for c.usedBytes > c.maxBytes {
98+
c.removeLocked(c.byRecency.Back())
99+
}
100+
}
101+
102+
func (c *signatureCache) removeLocked(element *list.Element) {
103+
entry := c.byRecency.Remove(element).(*signatureCacheEntry)
104+
delete(c.entries, entry.key)
105+
c.usedBytes -= entry.bytes
106+
}

‎submitqueue/extension/conflict/tango/cache_test.go‎

Lines changed: 178 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,178 @@
1+
// Copyright (c) 2026 Uber Technologies, Inc.
2+
//
3+
// Licensed under the Apache License, Version 2.0 (the "License");
4+
// you may not use this file except in compliance with the License.
5+
// You may obtain a copy of the License at
6+
//
7+
// http://www.apache.org/licenses/LICENSE-2.0
8+
//
9+
// Unless required by applicable law or agreed to in writing, software
10+
// distributed under the License is distributed on an "AS IS" BASIS,
11+
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12+
// See the License for the specific language governing permissions and
13+
// limitations under the License.
14+
15+
package tango
16+
17+
import (
18+
"strconv"
19+
"sync"
20+
"testing"
21+
22+
"github.com/stretchr/testify/assert"
23+
"github.com/stretchr/testify/require"
24+
)
25+
26+
func cacheTestSignature(queue, batchID string, targets int) TargetSignature {
27+
signature := TargetSignature{Queue: queue, BatchID: batchID, BaseRevision: "rev"}
28+
for i := 0; i < targets; i++ {
29+
signature.Targets = append(signature.Targets, SignatureTarget{ID: uint64(i), Distance: int32(i % 3)})
30+
}
31+
return signature
32+
}
33+
34+
func TestSignatureCache_GetPut(t *testing.T) {
35+
stored := cacheTestSignature("q", "1", 4)
36+
37+
tests := []struct {
38+
name string
39+
put []TargetSignature
40+
queue string
41+
batchID string
42+
wantFound bool
43+
}{
44+
{name: "miss on empty cache", queue: "q", batchID: "1"},
45+
{name: "hit after put", put: []TargetSignature{stored}, queue: "q", batchID: "1", wantFound: true},
46+
{name: "same batch ID in another queue is a miss", put: []TargetSignature{stored}, queue: "other", batchID: "1"},
47+
{name: "other batch ID is a miss", put: []TargetSignature{stored}, queue: "q", batchID: "2"},
48+
}
49+
for _, tt := range tests {
50+
t.Run(tt.name, func(t *testing.T) {
51+
cache := newSignatureCache(1 << 20)
52+
for _, signature := range tt.put {
53+
cache.put(signature)
54+
}
55+
56+
got, found := cache.get(tt.queue, tt.batchID)
57+
58+
assert.Equal(t, tt.wantFound, found)
59+
if tt.wantFound {
60+
assert.Equal(t, stored, got)
61+
}
62+
})
63+
}
64+
}
65+
66+
func TestSignatureCache_Eviction(t *testing.T) {
67+
a := cacheTestSignature("q", "a", 10)
68+
b := cacheTestSignature("q", "b", 10)
69+
c := cacheTestSignature("q", "c", 10)
70+
budgetForTwo := signatureCacheBytes(a) + signatureCacheBytes(b)
71+
72+
tests := []struct {
73+
name string
74+
run func(cache *signatureCache)
75+
wantFound map[string]bool
76+
}{
77+
{
78+
name: "least recently put is evicted first",
79+
run: func(cache *signatureCache) {
80+
cache.put(a)
81+
cache.put(b)
82+
cache.put(c)
83+
},
84+
wantFound: map[string]bool{"a": false, "b": true, "c": true},
85+
},
86+
{
87+
name: "a read refreshes recency",
88+
run: func(cache *signatureCache) {
89+
cache.put(a)
90+
cache.put(b)
91+
cache.get("q", "a")
92+
cache.put(c)
93+
},
94+
wantFound: map[string]bool{"a": true, "b": false, "c": true},
95+
},
96+
{
97+
name: "putting an existing key does not evict a neighbour",
98+
run: func(cache *signatureCache) {
99+
cache.put(a)
100+
cache.put(b)
101+
for i := 0; i < 100; i++ {
102+
cache.put(a)
103+
}
104+
},
105+
wantFound: map[string]bool{"a": true, "b": true},
106+
},
107+
{
108+
name: "a signature larger than the budget is not cached",
109+
run: func(cache *signatureCache) {
110+
cache.put(a)
111+
cache.put(cacheTestSignature("q", "c", 1000))
112+
},
113+
wantFound: map[string]bool{"a": true, "c": false},
114+
},
115+
}
116+
for _, tt := range tests {
117+
t.Run(tt.name, func(t *testing.T) {
118+
cache := newSignatureCache(budgetForTwo)
119+
120+
tt.run(cache)
121+
122+
for batchID, want := range tt.wantFound {
123+
_, found := cache.get("q", batchID)
124+
assert.Equal(t, want, found, "batch %s", batchID)
125+
}
126+
})
127+
}
128+
}
129+
130+
func TestSignatureCache_ChargesSpareCapacity(t *testing.T) {
131+
tight := cacheTestSignature("q", "tight", 10)
132+
roomy := cacheTestSignature("q", "roomy", 10)
133+
roomy.Targets = append(make([]SignatureTarget, 0, 1000), roomy.Targets...)
134+
cache := newSignatureCache(signatureCacheBytes(tight) * 2)
135+
136+
cache.put(roomy)
137+
138+
_, found := cache.get("q", "roomy")
139+
assert.False(t, found)
140+
}
141+
142+
func TestSignatureCache_ZeroBudgetCachesNothing(t *testing.T) {
143+
cache := newSignatureCache(0)
144+
145+
cache.put(cacheTestSignature("q", "1", 0))
146+
147+
_, found := cache.get("q", "1")
148+
assert.False(t, found)
149+
}
150+
151+
func TestSignatureCache_ConcurrentUse(t *testing.T) {
152+
const workers = 8
153+
const perWorker = 200
154+
sample := cacheTestSignature("q", "0", 5)
155+
cache := newSignatureCache(signatureCacheBytes(sample) * 4)
156+
157+
var wg sync.WaitGroup
158+
for w := 0; w < workers; w++ {
159+
wg.Add(1)
160+
go func(w int) {
161+
defer wg.Done()
162+
for i := 0; i < perWorker; i++ {
163+
batchID := strconv.Itoa((w + i) % 10)
164+
cache.put(cacheTestSignature("q", batchID, 5))
165+
cache.get("q", batchID)
166+
}
167+
}(w)
168+
}
169+
wg.Wait()
170+
171+
retained := 0
172+
for i := 0; i < 10; i++ {
173+
if _, found := cache.get("q", strconv.Itoa(i)); found {
174+
retained++
175+
}
176+
}
177+
require.LessOrEqual(t, retained, 4)
178+
}

0 commit comments

Comments
 (0)