Skip to content

Refactor Snake into a queue 2/3 #925

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

Draft
bgwines wants to merge 5 commits into
base: bwines/snake-22-snake-cleanup
Choose a base branch
from
Draft

Refactor Snake into a queue 2/3 #925

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
5 commits
Select commit Hold shift + click to select a range
f65c5dc
Make Snake a queue
bgwines Aug 23, 2026
e952f3c
Mark Snake timer transitions as parent-locked
bgwines Aug 23, 2026
02d54b0
Restore request terminal state
bgwines Aug 23, 2026
847d7ff
Keep Snake timer ownership
bgwines Aug 23, 2026
4beb66b
Support matching Snake waiters
bgwines Aug 23, 2026
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
122 changes: 24 additions & 98 deletions go/vt/vttablet/tabletserver/loadshed/codelq.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -70,7 +70,7 @@ import (
| * timer: armed |
*---------------------------*
| ^ ^
timer | | | dequeue/release w/ sojourn < target
timer | | | dequeue w/ sojourn < target
fires, | | | or queue emptied (sets dropping=false)
NOT | | timer fires,
healthy | | healthy (dropping=false)
Expand All @@ -88,9 +88,9 @@ import (

Health condition (checked each easing timer fire):
healthy := dropping=false
dropping is unset by lockedOnGrant() when a granted request's queue-wait
sojourn < target, and by lockedPeek/lockedPopElem/lockedRemove/
lockedOnGrant when droppableLen reaches 0; reset each timer fire. Note:
dropping is unset by lockedDequeue() when a dequeued request's queue-wait
sojourn < target, and by lockedRemove/lockedDequeue when
droppableLen reaches 0; reset each timer fire. Note:
while easing, each re-arm transiently re-marks dropping=true; the next
fire re-evaluates health.

Expand Down Expand Up @@ -150,7 +150,6 @@ type (
// the mutex, which is defined in the files for the higher-level structure
CoDelQueue[T any] struct {
queue *list.List
firstWaiting *list.Element
dropping bool
dropNextNs int64
count int
Expand All @@ -164,23 +163,21 @@ type (
nowNs func() int64
scheduleDropTimer func(delayNs int64)
stopDropTimer func()
onPeekCleanup func(*Request[T])
}
)

func (e *DroppedRequestError) Error() string {
return "request dropped by CoDel queue"
}

func newCoDelQueue[T any](cfg CoDelConfig, nowNs func() int64, scheduleDropTimer func(delayNs int64), stopDropTimer func(), onPeekCleanup func(*Request[T])) *CoDelQueue[T] {
func newCoDelQueue[T any](cfg CoDelConfig, nowNs func() int64, scheduleDropTimer func(delayNs int64), stopDropTimer func()) *CoDelQueue[T] {
q := &CoDelQueue[T]{
queue: list.New(),
count: 1,
cfg: cfg,
nowNs: nowNs,
scheduleDropTimer: scheduleDropTimer,
stopDropTimer: stopDropTimer,
onPeekCleanup: onPeekCleanup,
}
q.droppable.init()
return q
Expand All @@ -200,10 +197,6 @@ func (q *CoDelQueue[T]) lockedEnqueue(req *Request[T]) {
req.codelqEnqueuedAtNs = now
req.codelqElem = q.queue.PushBack(req)

if q.firstWaiting == nil {
q.firstWaiting = req.codelqElem
}

if req.isDroppable() {
q.droppableLen++
q.droppable.insert(req)
Expand All @@ -219,71 +212,30 @@ func (q *CoDelQueue[T]) lockedEnqueue(req *Request[T]) {
}
}

// lockedFirstWaiting returns the first not-yet-granted request in the queue.
func (q *CoDelQueue[T]) lockedFirstWaiting() *Request[T] {
if q.firstWaiting == nil {
// lockedPeek returns the first waiting request in the queue.
func (q *CoDelQueue[T]) lockedPeek() *Request[T] {
first := q.queue.Front()
if first == nil {
return nil
}
return q.firstWaiting.Value.(*Request[T])
return first.Value.(*Request[T])
}

// lockedPeek returns the head request without removing it. As a side effect,
// cleans up done-and-not-granted requests at the head (requests whose result
// channel has an error). Empty queue transitions to healthy.
func (q *CoDelQueue[T]) lockedPeek() *Request[T] {
for q.queue.Len() > 0 {
front := q.queue.Front()
req := front.Value.(*Request[T])
if req.signaledValue == nil {
func (q *CoDelQueue[T]) lockedFind(match func(T) bool) *Request[T] {
for e := q.queue.Front(); e != nil; e = e.Next() {
req := e.Value.(*Request[T])
if match(req.value) {
return req
}
q.lockedAdvanceFirstWaiting(front)
q.queue.Remove(front)
req.codelqElem = nil
if req.isDroppable() {
q.droppableLen--
q.droppable.remove(req)
}
if q.onPeekCleanup != nil {
q.onPeekCleanup(req)
}
}
// Empty queue means the underlying resource is available.
q.dropping = false
return nil
return q.lockedPeek()
}

// lockedPopElem removes the given element from the queue, signals the request's
// result channel, and updates bookkeeping. Use with care: this bypasses the
// health-state transitions in peek/dequeue, so callers are responsible for
// updating dropping state if appropriate.
func (q *CoDelQueue[T]) lockedPopElem(elem *list.Element, err error) *Request[T] {

@bgwines bgwines Aug 24, 2026

Copy link
Copy Markdown
Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

this function existed publicly only for the cancel-after-grant race. Snake doesn't own granting anymore, so it isn't needed. There were one or two internal callers, dequeue and peek. Dequeue still gets all of the functionality it needs, and peek used it for cleanup of canceled requests, but cancelation is different now; entries are synchronously removed upon cancel, so we don't need the clean-up (see PR description)

req := elem.Value.(*Request[T])
q.lockedAdvanceFirstWaiting(elem)
q.queue.Remove(elem)
req.codelqElem = nil

if req.signaledValue == nil {
req.signal(err)
}

if req.isDroppable() {
q.droppableLen--
q.droppable.remove(req)
if q.droppableLen == 0 && q.dropping {
q.dropping = false
}
}

return req
}

// lockedRemove removes a specific request from the queue without signaling it.
// lockedRemove removes a specific request from the queue.
func (q *CoDelQueue[T]) lockedRemove(r *Request[T]) {
if r.codelqElem == nil {
return
}
q.lockedAdvanceFirstWaiting(r.codelqElem)
q.queue.Remove(r.codelqElem)
r.codelqElem = nil

Expand All @@ -296,44 +248,19 @@ func (q *CoDelQueue[T]) lockedRemove(r *Request[T]) {
}
}

func (q *CoDelQueue[T]) lockedOnGrant(r *Request[T]) {
// CoDel health check, measured at grant: if this request's queue-wait
func (q *CoDelQueue[T]) lockedDequeue(r *Request[T]) {
// CoDel health check, measured at dequeue: if this request's queue-wait
// (now - enqueue) was under target, the system is healthy — leave the
// dropping state. Separate from the droppableLen==0 clear below.
if q.nowNs()-r.codelqEnqueuedAtNs < q.cfg.TargetNs() {
q.dropping = false
}
if r.isDroppable() {
q.droppable.remove(r)
q.droppableLen--
if q.droppableLen == 0 {
q.dropping = false
}
}
q.lockedAdvanceFirstWaiting(r.codelqElem)
q.queue.Remove(r.codelqElem)
r.codelqElem = nil
Comment on lines -306 to -315

@bgwines bgwines Aug 24, 2026

Copy link
Copy Markdown
Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

this behavior wasn't removed; it just lives in lockedRemove()

}

// lockedAdvanceFirstWaiting advances the firstWaiting pointer past elem if
// elem is the current firstWaiting. elem is a queue entry that is no longer
// waiting — either because it was granted, removed, or dropped.
func (q *CoDelQueue[T]) lockedAdvanceFirstWaiting(elem *list.Element) {
if q.firstWaiting != elem {
return
}
for e := elem.Next(); e != nil; e = e.Next() {
if e.Value.(*Request[T]).signaledValue == nil {
q.firstWaiting = e
return
}
}
q.firstWaiting = nil
q.lockedRemove(r)
}

// lockedFindLowestPriorityDroppable finds the lowest-priority droppable
// element in the queue — the oldest one at the lowest priority present — or nil
// if none exists. O(1) via the droppable priority index (see droppableIndex[T]).
// if none exists. O(1) via the droppable priority index (see droppableIndex).
func (q *CoDelQueue[T]) lockedFindLowestPriorityDroppable() *list.Element {
req := q.droppable.min()
if req == nil {
Expand All @@ -343,7 +270,7 @@ func (q *CoDelQueue[T]) lockedFindLowestPriorityDroppable() *list.Element {
}

// lockedRunTimer runs the CoDel drop logic. It is invoked both by the backstop
// timer and synchronously from the release/dequeue path, so shedding is driven
// timer and synchronously from the dequeue path, so shedding is driven
// as slots free rather than waiting for the (possibly late) timer to fire.
func (q *CoDelQueue[T]) lockedRunTimer(dropFn func() bool) {
q.lockedRunTimerLimited(dropFn, -1)
Expand All @@ -370,8 +297,8 @@ func (q *CoDelQueue[T]) lockedRunTimerLimited(dropFn func() bool, maxDrops int)
// lockedAdvance runs the clock-driven core of the CoDel control law: it sheds
// every request that is due (now >= dropNextNs) while dropping, ramping count,
// then eases count back down while healthy. Every action is gated on the fresh
// `now`, so calling it is idempotent and safe outside the timer — the release
// (dequeue) path invokes it to shed stale requests in real time rather than
// `now`, so calling it is idempotent and safe outside the timer — the dequeue
// path invokes it to shed stale requests in real time rather than
// waiting on the possibly-late backstop timer. It does NOT arm/disarm the timer.
func (q *CoDelQueue[T]) lockedAdvance(now int64, dropFn func() bool) {
q.lockedAdvanceLimited(now, dropFn, -1)
Expand Down Expand Up @@ -406,8 +333,7 @@ func (q *CoDelQueue[T]) lockedAdvanceLimited(now int64, dropFn func() bool, maxD
// If a request arrived before the end of the interval then it is a
// candidate for dropping.
if q.droppableLen > 0 {
q.lockedPeek() // cleanup
if first := q.lockedFirstWaiting(); first != nil && first.codelqEnqueuedAtNs < q.dropNextNs {
if first := q.lockedPeek(); first != nil && first.codelqEnqueuedAtNs < q.dropNextNs {
q.dropping = true
}
}
Expand Down
Loading
Loading