diff --git a/server/queue/fifo.go b/server/queue/fifo.go index 3c658b645db..8d9bcee0290 100644 --- a/server/queue/fifo.go +++ b/server/queue/fifo.go @@ -73,11 +73,12 @@ func NewMemoryQueue(ctx context.Context) Queue { return q } -// PushAtOnce pushes multiple tasks to the tail of this queue. +// PushAtOnce pushes multiple tasks to this queue, each into its creation-order +// position among the tasks already pending. func (q *fifo) PushAtOnce(_ context.Context, tasks []*model.Task) error { q.Lock() for _, task := range tasks { - q.pending.PushBack(task) + q.pushPending(task) } q.Unlock() return nil @@ -287,10 +288,26 @@ func (q *fifo) process() { } func (q *fifo) filterWaiting() { - // resubmits all waiting tasks to pending, deps may have cleared + // Resubmit all waiting tasks to pending; deps may have cleared. Both lists + // are ordered by taskOrderLess — waitingOnDeps is rebuilt below by scanning + // pending in order — so this is a merge of two sorted lists, walked once, + // rather than a search per task. A drained task from an older pipeline + // therefore costs no more than the newer tail it sorts ahead of. + at := q.pending.Front() for element := q.waitingOnDeps.Front(); element != nil; element = element.Next() { task, _ := element.Value.(*model.Task) - q.pending.PushBack(task) + for at != nil { + other, _ := at.Value.(*model.Task) + if taskOrderLess(task, other) { + break + } + at = at.Next() + } + if at == nil { + q.pending.PushBack(task) + continue + } + q.pending.InsertBefore(task, at) } // rebuild waitingDeps @@ -315,7 +332,7 @@ func (q *fifo) assignToWorker() (*list.Element, *worker) { var bestWorker *worker var bestScore int - for _, element := range q.pendingByCreation() { + for element := q.pending.Front(); element != nil; element = element.Next() { task, _ := element.Value.(*model.Task) log.Debug().Msgf("queue: trying to assign task: %v with deps %v", task.ID, task.Dependencies) @@ -342,35 +359,73 @@ func (q *fifo) assignToWorker() (*list.Element, *worker) { return nil, nil } -// pendingByCreation returns the pending tasks' list elements ordered by -// taskOrderLess (creation time, then workflow name). Dispatch walks this order -// so a workflow whose dependencies have cleared keeps its creation-order -// priority instead of being overtaken by tasks from pipelines created later, -// regardless of the order tasks were appended to the pending list. The sort is -// stable, so elements the comparator treats as equal keep their relative order. +// pushPending inserts a task into the pending list, keeping the list ordered by +// taskOrderLess (creation time, then workflow name). The sort key is immutable +// once a task is queued, so maintaining the order at insertion time is +// equivalent to — and replaces — re-sorting the whole pending set on every +// dispatch. Dispatch therefore walks the list directly, and a workflow whose +// dependencies have cleared keeps its creation-order priority instead of being +// overtaken by tasks from pipelines created later. +// +// A task is inserted after the tasks it compares equal to, matching the +// append-then-stable-sort order this replaces. Expects the queue to be locked +// by the caller. // -// This orders the post-filterWaiting pending set (assignToWorker runs right -// after filterWaiting on each process tick), where pending never simultaneously -// holds a task and a task that depends on it, so reordering cannot dispatch a -// dependent before its dependency. Expects the queue to be locked by the caller. -func (q *fifo) pendingByCreation() []*list.Element { - elements := make([]*list.Element, 0, q.pending.Len()) +// The scan runs back-to-front, which is one step for the common case of a new +// pipeline's batch appended to the tail. A task belonging ahead of the whole +// list is that direction's worst case, so it is taken by a front check first. +// The check is on a strict less-than, so it never reorders equals. +func (q *fifo) pushPending(task *model.Task) { + if front := q.pending.Front(); front == nil { + q.pending.PushFront(task) + return + } else if other, _ := front.Value.(*model.Task); taskOrderLess(task, other) { + q.pending.PushFront(task) + return + } + + for element := q.pending.Back(); element != nil; element = element.Prev() { + other, _ := element.Value.(*model.Task) + if !taskOrderLess(task, other) { + q.pending.InsertAfter(task, element) + return + } + } + // Unreachable: the front-check above returns when the task sorts before the + // head, so the back-to-front scan always finds an insertion point. Kept as a + // defensive fallback. + q.pending.PushFront(task) +} + +// pushPendingFront inserts a task ahead of the tasks it compares equal to, +// otherwise keeping the taskOrderLess order pushPending maintains. A resubmitted +// expired task has already been dispatched once, so it keeps the head position +// among its equals that a plain list push-front used to give it. +// Expects the queue to be locked by the caller. +// +// The scan runs front-to-back, the good end for a resubmitted task from an +// older pipeline. A task sorting past the whole list is taken by a back check +// first, on a strict less-than so equals are never reordered. +func (q *fifo) pushPendingFront(task *model.Task) { + if back := q.pending.Back(); back == nil { + q.pending.PushBack(task) + return + } else if other, _ := back.Value.(*model.Task); taskOrderLess(other, task) { + q.pending.PushBack(task) + return + } + for element := q.pending.Front(); element != nil; element = element.Next() { - elements = append(elements, element) - } - slices.SortStableFunc(elements, func(a, b *list.Element) int { - taskA, _ := a.Value.(*model.Task) - taskB, _ := b.Value.(*model.Task) - switch { - case taskOrderLess(taskA, taskB): - return -1 - case taskOrderLess(taskB, taskA): - return 1 - default: - return 0 + other, _ := element.Value.(*model.Task) + if !taskOrderLess(other, task) { + q.pending.InsertBefore(task, element) + return } - }) - return elements + } + // Unreachable: the back-check above returns when the task sorts after the + // tail, so the front-to-back scan always finds an insertion point. Kept as a + // defensive fallback. + q.pending.PushBack(task) } // canRunConcurrent reports whether the given task may currently start without @@ -454,7 +509,7 @@ func (q *fifo) resubmitExpiredPipelines() { if time.Now().After(taskState.deadline) { log.Info().Msgf("queue: resubmitting expired task %s", taskID) taskState.error = ErrTaskExpired - q.pending.PushFront(taskState.item) + q.pushPendingFront(taskState.item) delete(q.running, taskID) close(taskState.done) } diff --git a/server/queue/fifo_test.go b/server/queue/fifo_test.go index 0840a8bac95..dec5d0d8978 100644 --- a/server/queue/fifo_test.go +++ b/server/queue/fifo_test.go @@ -15,6 +15,7 @@ package queue import ( + "container/list" "context" "errors" "fmt" @@ -1620,3 +1621,162 @@ func TestFifoFairDispatch(t *testing.T) { assert.Len(t, info.Running, 0) }) } + +func TestFifoMultiDispatchTickOrder(t *testing.T) { + ctx, cancel, q := setupTestQueue(t) + defer cancel(nil) + + // A single process tick dispatches in a loop, so several tasks leave the + // pending list in one pass. Only the earliest-Created tasks may go, however + // many workers are waiting: with fewer workers than tasks, the tail of the + // creation order must stay pending. Tasks are pushed newest-first, and in + // separate batches, so a pending list that is not kept in creation order + // dispatches the wrong ones. + pushes := [][]*model.Task{ + {{ID: "50", Created: 500}, {ID: "40", Created: 400}}, + {{ID: "30", Created: 300}}, + {{ID: "20", Created: 200}, {ID: "10", Created: 100}}, + } + for _, batch := range pushes { + assert.NoError(t, q.PushAtOnce(ctx, batch)) + } + + // two workers, five pending tasks: exactly the two earliest dispatch. + results := make(chan *model.Task, 2) + for agentID := int64(1); agentID <= 2; agentID++ { + go func() { + task, _ := q.Poll(ctx, agentID, filterFnTrue) + results <- task + }() + } + + dispatched := make([]string, 0, 2) + for range 2 { + select { + case task := <-results: + assert.NotNil(t, task) + dispatched = append(dispatched, task.ID) + case <-time.After(time.Second): + t.Fatal("timeout waiting for dispatched tasks") + } + } + assert.ElementsMatch(t, []string{"10", "20"}, dispatched, + "the two earliest-Created tasks must be the ones dispatched in the tick") + + waitForProcess() + info := q.Info(ctx) + pending := make([]string, 0, len(info.Pending)) + for _, task := range info.Pending { + pending = append(pending, task.ID) + } + assert.Equal(t, []string{"30", "40", "50"}, pending, + "the remaining tasks stay pending in creation order") + + for _, id := range dispatched { + assert.NoError(t, q.Done(ctx, id, model.StatusSuccess)) + } +} + +func TestFifoPendingInsertOrder(t *testing.T) { + // The two insert helpers differ only in where they place a task among the + // ones it compares equal to: a normal push goes after them (the order the + // append-then-stable-sort it replaces produced), while a resubmitted expired + // task goes ahead of them, keeping the retry priority a list push-front used + // to give it. Asserted directly on the list: through the queue the two are + // indistinguishable whenever the comparator already separates the tasks. + pendingIDs := func(q *fifo) []string { + ids := make([]string, 0, q.pending.Len()) + for element := q.pending.Front(); element != nil; element = element.Next() { + task, _ := element.Value.(*model.Task) + ids = append(ids, task.ID) + } + return ids + } + + q := &fifo{pending: list.New()} + q.pushPending(&model.Task{ID: "b1", Created: 200}) + q.pushPending(&model.Task{ID: "a", Created: 100}) + q.pushPending(&model.Task{ID: "c", Created: 300}) + q.pushPending(&model.Task{ID: "b2", Created: 200}) + assert.Equal(t, []string{"a", "b1", "b2", "c"}, pendingIDs(q), + "a push is ordered by Created and lands after the tasks it ties with") + + q.pushPendingFront(&model.Task{ID: "retry", Created: 200}) + assert.Equal(t, []string{"a", "retry", "b1", "b2", "c"}, pendingIDs(q), + "a resubmitted task is ordered by Created but lands ahead of its ties") + + // pushPendingFront's empty-list branch: its first call above went into a + // populated list, so cover the fresh-list case too (symmetric with the + // pushPending empty-list push at the top). + empty := &fifo{pending: list.New()} + empty.pushPendingFront(&model.Task{ID: "only", Created: 100}) + assert.Equal(t, []string{"only"}, pendingIDs(empty), + "pushPendingFront into an empty list seeds the list") + + // Both helpers must also handle the ends of the list, not just the middle. + q.pushPendingFront(&model.Task{ID: "first", Created: 50}) + q.pushPending(&model.Task{ID: "last", Created: 400}) + assert.Equal(t, []string{"first", "a", "retry", "b1", "b2", "c", "last"}, pendingIDs(q)) +} + +func TestFifoFilterWaitingDrainOrder(t *testing.T) { + // filterWaiting drains dependency-cleared tasks back into pending on every + // tick. They must land in creation order among the tasks already there — + // a task from an older pipeline goes ahead of the newer tail, which is the + // whole point of fair dispatch. Asserted on the list rather than through a + // dispatch, so the invariant has a guard of its own. + // A single drained task that sorts to the very front stops the cursor on the + // first comparison — the simplest merge case. + q := &fifo{pending: list.New(), waitingOnDeps: list.New(), running: map[string]*entry{}} + q.pending.PushBack(&model.Task{ID: "new1", Created: 300}) + q.pending.PushBack(&model.Task{ID: "new2", Created: 400}) + q.waitingOnDeps.PushBack(&model.Task{ID: "old", Created: 100}) + + q.filterWaiting() + + pendingIDs := func(q *fifo) []string { + ids := make([]string, 0, q.pending.Len()) + for element := q.pending.Front(); element != nil; element = element.Next() { + task, _ := element.Value.(*model.Task) + ids = append(ids, task.ID) + } + return ids + } + assert.Equal(t, []string{"old", "new1", "new2"}, pendingIDs(q), + "a drained task keeps its creation-order priority over newer pending tasks") + + // Multiple drained tasks interleaving into the middle and tail of a + // multi-element pending list: exercises the forward cursor advancing across + // iterations and the at==nil tail-PushBack branch. waitingOnDeps must be + // ascending (the invariant filterWaiting's rebuild maintains). + q2 := &fifo{pending: list.New(), waitingOnDeps: list.New(), running: map[string]*entry{}} + for _, id := range []struct { + name string + created int64 + }{{"p100", 100}, {"p300", 300}, {"p500", 500}} { + q2.pending.PushBack(&model.Task{ID: id.name, Created: id.created}) + } + for _, id := range []struct { + name string + created int64 + }{{"w200", 200}, {"w400", 400}, {"w600", 600}} { + q2.waitingOnDeps.PushBack(&model.Task{ID: id.name, Created: id.created}) + } + + q2.filterWaiting() + + assert.Equal(t, []string{"p100", "w200", "p300", "w400", "p500", "w600"}, pendingIDs(q2), + "drained tasks merge into creation order, advancing the cursor through the middle and appending at the tail (w600)") + + // A drained task whose Created ties a pending task lands AFTER the equal-key + // pending task, matching the append-then-stable-sort semantics the merge + // replaces (taskOrderLess is a strict less-than on Created, then Name). + q3 := &fifo{pending: list.New(), waitingOnDeps: list.New(), running: map[string]*entry{}} + q3.pending.PushBack(&model.Task{ID: "pending-200", Created: 200, Name: "a"}) + q3.waitingOnDeps.PushBack(&model.Task{ID: "drained-200", Created: 200, Name: "z"}) + + q3.filterWaiting() + + assert.Equal(t, []string{"pending-200", "drained-200"}, pendingIDs(q3), + "a drained task tying a pending task's Created lands after it") +} diff --git a/server/queue/persistent.go b/server/queue/persistent.go index 9448c8ce7bf..be3731ea340 100644 --- a/server/queue/persistent.go +++ b/server/queue/persistent.go @@ -58,7 +58,8 @@ func isTerminalWorkflowState(state model.StatusValue) bool { } } -// PushAtOnce pushes multiple tasks to the tail of this queue. +// PushAtOnce persists multiple tasks and pushes them to this queue, each into +// its creation-order position among the tasks already pending. func (q *persistentQueue) PushAtOnce(c context.Context, tasks []*model.Task) error { // TODO: invent store.NewSession who return context including a session and make TaskInsert & TaskDelete use it for _, task := range tasks {