Skip to content

Commit 361eead

Browse files
manjari25manjari25
committed
feat(conflict): Add Tango signature cache
1 parent a25bff3 commit 361eead

3 files changed

Lines changed: 280 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: 105 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,105 @@
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 fixed overhead covers the entry, key and map bookkeeping.
24+
const (
25+
signatureTargetBytes = 16
26+
signatureEntryOverhead = 256
27+
)
28+
29+
type signatureCacheKey struct {
30+
queue string
31+
batchID string
32+
}
33+
34+
type signatureCacheEntry struct {
35+
key signatureCacheKey
36+
signature TargetSignature
37+
bytes int64
38+
}
39+
40+
// signatureCache is a byte-bounded LRU of target signatures keyed by queue and
41+
// batch ID, safe for concurrent use. Signatures are immutable, so a cached
42+
// value is shared with callers: neither side may modify Targets.
43+
type signatureCache struct {
44+
mu sync.Mutex
45+
maxBytes int64
46+
usedBytes int64
47+
byRecency *list.List
48+
entries map[signatureCacheKey]*list.Element
49+
}
50+
51+
func newSignatureCache(maxBytes int64) *signatureCache {
52+
return &signatureCache{
53+
maxBytes: maxBytes,
54+
byRecency: list.New(),
55+
entries: make(map[signatureCacheKey]*list.Element),
56+
}
57+
}
58+
59+
func signatureCacheBytes(signature TargetSignature) int64 {
60+
return signatureEntryOverhead +
61+
int64(len(signature.Queue)+len(signature.BatchID)+len(signature.BaseRevision)) +
62+
int64(len(signature.Targets))*signatureTargetBytes
63+
}
64+
65+
// get returns the cached signature and marks it most recently used.
66+
func (c *signatureCache) get(queue, batchID string) (TargetSignature, bool) {
67+
c.mu.Lock()
68+
defer c.mu.Unlock()
69+
70+
element, ok := c.entries[signatureCacheKey{queue: queue, batchID: batchID}]
71+
if !ok {
72+
return TargetSignature{}, false
73+
}
74+
c.byRecency.MoveToFront(element)
75+
return element.Value.(*signatureCacheEntry).signature, true
76+
}
77+
78+
// put caches signature, evicting least recently used entries to stay within the
79+
// budget. A signature larger than the whole budget is not cached.
80+
func (c *signatureCache) put(signature TargetSignature) {
81+
size := signatureCacheBytes(signature)
82+
if size > c.maxBytes {
83+
return
84+
}
85+
key := signatureCacheKey{queue: signature.Queue, batchID: signature.BatchID}
86+
87+
c.mu.Lock()
88+
defer c.mu.Unlock()
89+
90+
if element, ok := c.entries[key]; ok {
91+
c.removeLocked(element)
92+
}
93+
c.entries[key] = c.byRecency.PushFront(&signatureCacheEntry{key: key, signature: signature, bytes: size})
94+
c.usedBytes += size
95+
96+
for c.usedBytes > c.maxBytes {
97+
c.removeLocked(c.byRecency.Back())
98+
}
99+
}
100+
101+
func (c *signatureCache) removeLocked(element *list.Element) {
102+
entry := c.byRecency.Remove(element).(*signatureCacheEntry)
103+
delete(c.entries, entry.key)
104+
c.usedBytes -= entry.bytes
105+
}

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

Lines changed: 166 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,166 @@
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, "c": false},
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, "b": false, "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_ZeroBudgetCachesNothing(t *testing.T) {
131+
cache := newSignatureCache(0)
132+
133+
cache.put(cacheTestSignature("q", "1", 0))
134+
135+
_, found := cache.get("q", "1")
136+
assert.False(t, found)
137+
}
138+
139+
func TestSignatureCache_ConcurrentUse(t *testing.T) {
140+
const workers = 8
141+
const perWorker = 200
142+
sample := cacheTestSignature("q", "0", 5)
143+
cache := newSignatureCache(signatureCacheBytes(sample) * 4)
144+
145+
var wg sync.WaitGroup
146+
for w := 0; w < workers; w++ {
147+
wg.Add(1)
148+
go func(w int) {
149+
defer wg.Done()
150+
for i := 0; i < perWorker; i++ {
151+
batchID := strconv.Itoa((w + i) % 10)
152+
cache.put(cacheTestSignature("q", batchID, 5))
153+
cache.get("q", batchID)
154+
}
155+
}(w)
156+
}
157+
wg.Wait()
158+
159+
retained := 0
160+
for i := 0; i < 10; i++ {
161+
if _, found := cache.get("q", strconv.Itoa(i)); found {
162+
retained++
163+
}
164+
}
165+
require.LessOrEqual(t, retained, 4)
166+
}

0 commit comments

Comments
 (0)