proxy: snapshot in-flight counts before sorting (least_conn selection O(n)) #124
@@ -760,13 +760,41 @@ func (e *Engine) baseURLAttemptOrder(remote models.Remote) []string {
|
|||||||
ordered[i] = urls[(start+i)%len(urls)]
|
ordered[i] = urls[(start+i)%len(urls)]
|
||||||
}
|
}
|
||||||
if remote.MirrorStrategy == models.MirrorStrategyLeastConn {
|
if remote.MirrorStrategy == models.MirrorStrategyLeastConn {
|
||||||
sort.SliceStable(ordered, func(a, b int) bool {
|
// Snapshot each mirror's in-flight count once, then sort the snapshot.
|
||||||
return e.inflightCounter(remote.Name, ordered[a]).Load() < e.inflightCounter(remote.Name, ordered[b]).Load()
|
// Reading the gauge inside the comparator would repeat an allocating
|
||||||
|
// sync.Map lookup on every comparison (O(n log n) lookups); this is O(n).
|
||||||
|
snap := make([]inflightSnapshot, len(ordered))
|
||||||
|
for i, url := range ordered {
|
||||||
|
snap[i] = inflightSnapshot{url: url, count: e.inflightCount(remote.Name, url)}
|
||||||
|
}
|
||||||
|
sort.SliceStable(snap, func(a, b int) bool {
|
||||||
|
return snap[a].count < snap[b].count
|
||||||
})
|
})
|
||||||
|
for i := range snap {
|
||||||
|
ordered[i] = snap[i].url
|
||||||
|
}
|
||||||
}
|
}
|
||||||
return ordered
|
return ordered
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// inflightSnapshot pairs a mirror URL with its sampled in-flight count so the
|
||||||
|
// least_conn sort compares plain ints instead of re-reading the gauge.
|
||||||
|
type inflightSnapshot struct {
|
||||||
|
url string
|
||||||
|
count int64
|
||||||
|
}
|
||||||
|
|
||||||
|
// inflightCount reads the in-flight request gauge for a given (remote, upstream
|
||||||
|
// URL) without creating it, returning 0 when the counter is absent. This keeps
|
||||||
|
// the selection read path allocation-free (plain Load, no LoadOrStore).
|
||||||
|
func (e *Engine) inflightCount(remoteName, url string) int64 {
|
||||||
|
v, ok := e.inflight.Load(remoteName + "\x00" + url)
|
||||||
|
if !ok {
|
||||||
|
return 0
|
||||||
|
}
|
||||||
|
return v.(*atomic.Int64).Load()
|
||||||
|
}
|
||||||
|
|
||||||
// inflightCounter returns the shared in-flight request gauge for a given
|
// inflightCounter returns the shared in-flight request gauge for a given
|
||||||
// (remote, upstream URL), creating it on first use.
|
// (remote, upstream URL), creating it on first use.
|
||||||
func (e *Engine) inflightCounter(remoteName, url string) *atomic.Int64 {
|
func (e *Engine) inflightCounter(remoteName, url string) *atomic.Int64 {
|
||||||
|
|||||||
@@ -40,6 +40,51 @@ func TestLeastConnPicksLeastLoaded(t *testing.T) {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// TestLeastConnStableTieBreak asserts that when every mirror carries equal
|
||||||
|
// in-flight load, least_conn falls back to the round-robin rotation: the
|
||||||
|
// snapshot sort is stable, so tied mirrors keep the RR-rotated order and the
|
||||||
|
// starting pick advances across the whole pool on successive calls.
|
||||||
|
func TestLeastConnStableTieBreak(t *testing.T) {
|
||||||
|
e := &Engine{}
|
||||||
|
r := models.Remote{
|
||||||
|
Name: "lc-tie",
|
||||||
|
BaseURL: "https://a.example",
|
||||||
|
Mirrorlist: []string{"https://b.example", "https://c.example"},
|
||||||
|
MirrorStrategy: models.MirrorStrategyLeastConn,
|
||||||
|
}
|
||||||
|
pool := r.UpstreamPool()
|
||||||
|
|
||||||
|
// Equal (zero) load on every mirror: order must equal the RR rotation.
|
||||||
|
starts := map[string]int{}
|
||||||
|
for i := 0; i < len(pool); i++ {
|
||||||
|
order := e.baseURLAttemptOrder(r)
|
||||||
|
if len(order) != len(pool) {
|
||||||
|
t.Fatalf("attempt %d: order len = %d, want %d", i, len(order), len(pool))
|
||||||
|
}
|
||||||
|
// A stable sort of an all-tied slice is a pure RR rotation: for the
|
||||||
|
// call whose cursor selects start s, order must be pool rotated by s.
|
||||||
|
start := indexOf(pool, order[0])
|
||||||
|
for j := range order {
|
||||||
|
if want := pool[(start+j)%len(pool)]; order[j] != want {
|
||||||
|
t.Fatalf("attempt %d: order[%d] = %q, want RR-rotated %q", i, j, order[j], want)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
starts[order[0]]++
|
||||||
|
}
|
||||||
|
if len(starts) != len(pool) {
|
||||||
|
t.Fatalf("tied least_conn did not rotate across the whole pool: %v", starts)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
func indexOf(s []string, v string) int {
|
||||||
|
for i := range s {
|
||||||
|
if s[i] == v {
|
||||||
|
return i
|
||||||
|
}
|
||||||
|
}
|
||||||
|
return -1
|
||||||
|
}
|
||||||
|
|
||||||
// TestRoundRobinDefaultUnchanged asserts an unset strategy still rotates the
|
// TestRoundRobinDefaultUnchanged asserts an unset strategy still rotates the
|
||||||
// starting mirror across the pool and ignores the in-flight gauge.
|
// starting mirror across the pool and ignores the in-flight gauge.
|
||||||
func TestRoundRobinDefaultUnchanged(t *testing.T) {
|
func TestRoundRobinDefaultUnchanged(t *testing.T) {
|
||||||
|
|||||||
Reference in New Issue
Block a user