2022-10-18 01:24:12 +00:00
|
|
|
// Copyright 2022 Woodpecker Authors
|
|
|
|
//
|
|
|
|
// Licensed under the Apache License, Version 2.0 (the "License");
|
|
|
|
// you may not use this file except in compliance with the License.
|
|
|
|
// You may obtain a copy of the License at
|
|
|
|
//
|
|
|
|
// http://www.apache.org/licenses/LICENSE-2.0
|
|
|
|
//
|
|
|
|
// Unless required by applicable law or agreed to in writing, software
|
|
|
|
// distributed under the License is distributed on an "AS IS" BASIS,
|
|
|
|
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
|
|
|
// See the License for the specific language governing permissions and
|
|
|
|
// limitations under the License.
|
|
|
|
|
2017-03-05 07:56:08 +00:00
|
|
|
package queue
|
|
|
|
|
|
|
|
import (
|
|
|
|
"container/list"
|
|
|
|
"context"
|
2024-02-16 09:04:13 +00:00
|
|
|
"fmt"
|
2017-03-05 07:56:08 +00:00
|
|
|
"sync"
|
|
|
|
"time"
|
2019-06-13 15:38:19 +00:00
|
|
|
|
2021-10-12 07:25:13 +00:00
|
|
|
"github.com/rs/zerolog/log"
|
2024-01-14 17:22:06 +00:00
|
|
|
|
|
|
|
"go.woodpecker-ci.org/woodpecker/v2/server/model"
|
2024-09-20 07:40:48 +00:00
|
|
|
"go.woodpecker-ci.org/woodpecker/v2/shared/constant"
|
2017-03-05 07:56:08 +00:00
|
|
|
)
|
|
|
|
|
|
|
|
type entry struct {
|
2023-03-20 03:50:56 +00:00
|
|
|
item *model.Task
|
2017-03-05 07:56:08 +00:00
|
|
|
done chan bool
|
|
|
|
error error
|
|
|
|
deadline time.Time
|
|
|
|
}
|
|
|
|
|
|
|
|
type worker struct {
|
2023-03-21 13:10:43 +00:00
|
|
|
agentID int64
|
2022-05-30 23:12:18 +00:00
|
|
|
filter FilterFn
|
2023-03-20 03:50:56 +00:00
|
|
|
channel chan *model.Task
|
2024-02-16 09:04:13 +00:00
|
|
|
stop context.CancelCauseFunc
|
2017-03-05 07:56:08 +00:00
|
|
|
}
|
|
|
|
|
|
|
|
type fifo struct {
|
|
|
|
sync.Mutex
|
|
|
|
|
2024-09-20 07:40:48 +00:00
|
|
|
ctx context.Context
|
2019-07-09 14:23:56 +00:00
|
|
|
workers map[*worker]struct{}
|
|
|
|
running map[string]*entry
|
|
|
|
pending *list.List
|
|
|
|
waitingOnDeps *list.List
|
|
|
|
extension time.Duration
|
|
|
|
paused bool
|
2017-03-05 07:56:08 +00:00
|
|
|
}
|
|
|
|
|
2024-09-20 07:40:48 +00:00
|
|
|
// processTimeInterval is the time till the queue rearranges things,
|
|
|
|
// as the agent pull in 10 milliseconds we should also give them work asap.
|
|
|
|
const processTimeInterval = 100 * time.Millisecond
|
|
|
|
|
2017-03-05 07:56:08 +00:00
|
|
|
// New returns a new fifo queue.
|
2024-09-20 07:40:48 +00:00
|
|
|
func New(ctx context.Context) Queue {
|
|
|
|
q := &fifo{
|
|
|
|
ctx: ctx,
|
2019-07-09 14:23:56 +00:00
|
|
|
workers: map[*worker]struct{}{},
|
|
|
|
running: map[string]*entry{},
|
|
|
|
pending: list.New(),
|
|
|
|
waitingOnDeps: list.New(),
|
2024-09-20 07:40:48 +00:00
|
|
|
extension: constant.TaskTimeout,
|
2019-07-09 14:23:56 +00:00
|
|
|
paused: false,
|
2017-03-05 07:56:08 +00:00
|
|
|
}
|
2024-09-20 07:40:48 +00:00
|
|
|
go q.process()
|
|
|
|
return q
|
2017-03-05 07:56:08 +00:00
|
|
|
}
|
|
|
|
|
2024-02-16 09:04:13 +00:00
|
|
|
// Push pushes a task to the tail of this queue.
|
2023-03-20 03:50:56 +00:00
|
|
|
func (q *fifo) Push(_ context.Context, task *model.Task) error {
|
2017-03-05 07:56:08 +00:00
|
|
|
q.Lock()
|
|
|
|
q.pending.PushBack(task)
|
|
|
|
q.Unlock()
|
|
|
|
return nil
|
|
|
|
}
|
|
|
|
|
2024-02-16 09:04:13 +00:00
|
|
|
// PushAtOnce pushes multiple tasks to the tail of this queue.
|
2023-03-20 03:50:56 +00:00
|
|
|
func (q *fifo) PushAtOnce(_ context.Context, tasks []*model.Task) error {
|
2019-06-13 15:38:19 +00:00
|
|
|
q.Lock()
|
|
|
|
for _, task := range tasks {
|
|
|
|
q.pending.PushBack(task)
|
|
|
|
}
|
|
|
|
q.Unlock()
|
|
|
|
return nil
|
|
|
|
}
|
|
|
|
|
2024-02-16 09:04:13 +00:00
|
|
|
// Poll retrieves and removes a task head of this queue.
|
2023-03-21 13:10:43 +00:00
|
|
|
func (q *fifo) Poll(c context.Context, agentID int64, f FilterFn) (*model.Task, error) {
|
2017-03-05 07:56:08 +00:00
|
|
|
q.Lock()
|
2024-02-16 09:04:13 +00:00
|
|
|
ctx, stop := context.WithCancelCause(c)
|
|
|
|
|
2017-03-05 07:56:08 +00:00
|
|
|
w := &worker{
|
2023-03-21 13:10:43 +00:00
|
|
|
agentID: agentID,
|
2023-03-20 03:50:56 +00:00
|
|
|
channel: make(chan *model.Task, 1),
|
2017-03-05 07:56:08 +00:00
|
|
|
filter: f,
|
2024-02-16 09:04:13 +00:00
|
|
|
stop: stop,
|
2017-03-05 07:56:08 +00:00
|
|
|
}
|
|
|
|
q.workers[w] = struct{}{}
|
|
|
|
q.Unlock()
|
|
|
|
|
|
|
|
for {
|
|
|
|
select {
|
2024-02-16 09:04:13 +00:00
|
|
|
case <-ctx.Done():
|
2017-03-05 07:56:08 +00:00
|
|
|
q.Lock()
|
|
|
|
delete(q.workers, w)
|
|
|
|
q.Unlock()
|
2024-02-16 09:04:13 +00:00
|
|
|
return nil, ctx.Err()
|
2017-03-05 07:56:08 +00:00
|
|
|
case t := <-w.channel:
|
|
|
|
return t, nil
|
|
|
|
}
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
2024-02-16 09:04:13 +00:00
|
|
|
// Done signals the task is complete.
|
2023-03-18 19:35:27 +00:00
|
|
|
func (q *fifo) Done(_ context.Context, id string, exitStatus model.StatusValue) error {
|
2023-03-20 03:50:56 +00:00
|
|
|
return q.finished([]string{id}, exitStatus, nil)
|
2017-03-05 07:56:08 +00:00
|
|
|
}
|
|
|
|
|
2024-02-16 09:04:13 +00:00
|
|
|
// Error signals the task is done with an error.
|
2023-03-18 19:35:27 +00:00
|
|
|
func (q *fifo) Error(_ context.Context, id string, err error) error {
|
2023-03-20 03:50:56 +00:00
|
|
|
return q.finished([]string{id}, model.StatusFailure, err)
|
2019-09-16 13:18:15 +00:00
|
|
|
}
|
|
|
|
|
2024-02-16 09:04:13 +00:00
|
|
|
// ErrorAtOnce signals multiple done are complete with an error.
|
|
|
|
func (q *fifo) ErrorAtOnce(_ context.Context, ids []string, err error) error {
|
|
|
|
return q.finished(ids, model.StatusFailure, err)
|
2019-07-22 10:43:59 +00:00
|
|
|
}
|
|
|
|
|
2023-03-20 03:50:56 +00:00
|
|
|
func (q *fifo) finished(ids []string, exitStatus model.StatusValue, err error) error {
|
2017-03-05 07:56:08 +00:00
|
|
|
q.Lock()
|
2019-09-16 13:18:15 +00:00
|
|
|
|
|
|
|
for _, id := range ids {
|
|
|
|
taskEntry, ok := q.running[id]
|
|
|
|
if ok {
|
|
|
|
taskEntry.error = err
|
|
|
|
close(taskEntry.done)
|
|
|
|
delete(q.running, id)
|
|
|
|
} else {
|
|
|
|
q.removeFromPending(id)
|
|
|
|
}
|
|
|
|
q.updateDepStatusInQueue(id, exitStatus)
|
2017-03-05 07:56:08 +00:00
|
|
|
}
|
2019-09-16 13:18:15 +00:00
|
|
|
|
2017-03-05 07:56:08 +00:00
|
|
|
q.Unlock()
|
|
|
|
return nil
|
|
|
|
}
|
|
|
|
|
2017-04-06 16:04:25 +00:00
|
|
|
// Evict removes a pending task from the queue.
|
|
|
|
func (q *fifo) Evict(c context.Context, id string) error {
|
2019-09-16 13:18:15 +00:00
|
|
|
return q.EvictAtOnce(c, []string{id})
|
|
|
|
}
|
|
|
|
|
2024-02-16 09:04:13 +00:00
|
|
|
// EvictAtOnce removes multiple pending tasks from the queue.
|
2023-03-18 19:35:27 +00:00
|
|
|
func (q *fifo) EvictAtOnce(_ context.Context, ids []string) error {
|
2017-04-06 16:04:25 +00:00
|
|
|
q.Lock()
|
|
|
|
defer q.Unlock()
|
|
|
|
|
2019-09-16 13:18:15 +00:00
|
|
|
for _, id := range ids {
|
|
|
|
var next *list.Element
|
|
|
|
for e := q.pending.Front(); e != nil; e = next {
|
|
|
|
next = e.Next()
|
2023-03-20 03:50:56 +00:00
|
|
|
task, ok := e.Value.(*model.Task)
|
2019-09-16 13:18:15 +00:00
|
|
|
if ok && task.ID == id {
|
|
|
|
q.pending.Remove(e)
|
|
|
|
return nil
|
|
|
|
}
|
2017-04-06 16:04:25 +00:00
|
|
|
}
|
|
|
|
}
|
|
|
|
return ErrNotFound
|
|
|
|
}
|
|
|
|
|
2017-03-05 07:56:08 +00:00
|
|
|
// Wait waits until the item is done executing.
|
|
|
|
func (q *fifo) Wait(c context.Context, id string) error {
|
|
|
|
q.Lock()
|
|
|
|
state := q.running[id]
|
|
|
|
q.Unlock()
|
|
|
|
if state != nil {
|
|
|
|
select {
|
|
|
|
case <-c.Done():
|
|
|
|
case <-state.done:
|
|
|
|
return state.error
|
|
|
|
}
|
|
|
|
}
|
|
|
|
return nil
|
|
|
|
}
|
|
|
|
|
|
|
|
// Extend extends the task execution deadline.
|
2023-03-18 19:35:27 +00:00
|
|
|
func (q *fifo) Extend(_ context.Context, id string) error {
|
2017-03-05 07:56:08 +00:00
|
|
|
q.Lock()
|
|
|
|
defer q.Unlock()
|
|
|
|
|
|
|
|
state, ok := q.running[id]
|
|
|
|
if ok {
|
|
|
|
state.deadline = time.Now().Add(q.extension)
|
|
|
|
return nil
|
|
|
|
}
|
|
|
|
return ErrNotFound
|
|
|
|
}
|
|
|
|
|
|
|
|
// Info returns internal queue information.
|
2023-03-18 19:35:27 +00:00
|
|
|
func (q *fifo) Info(_ context.Context) InfoT {
|
2017-03-05 07:56:08 +00:00
|
|
|
q.Lock()
|
|
|
|
stats := InfoT{}
|
|
|
|
stats.Stats.Workers = len(q.workers)
|
|
|
|
stats.Stats.Pending = q.pending.Len()
|
2019-07-09 14:23:56 +00:00
|
|
|
stats.Stats.WaitingOnDeps = q.waitingOnDeps.Len()
|
2017-03-05 07:56:08 +00:00
|
|
|
stats.Stats.Running = len(q.running)
|
|
|
|
|
|
|
|
for e := q.pending.Front(); e != nil; e = e.Next() {
|
2024-01-12 01:01:02 +00:00
|
|
|
task, _ := e.Value.(*model.Task)
|
|
|
|
stats.Pending = append(stats.Pending, task)
|
2017-03-05 07:56:08 +00:00
|
|
|
}
|
2019-07-09 14:23:56 +00:00
|
|
|
for e := q.waitingOnDeps.Front(); e != nil; e = e.Next() {
|
2024-01-12 01:01:02 +00:00
|
|
|
task, _ := e.Value.(*model.Task)
|
|
|
|
stats.WaitingOnDeps = append(stats.WaitingOnDeps, task)
|
2019-07-09 14:23:56 +00:00
|
|
|
}
|
2017-03-05 07:56:08 +00:00
|
|
|
for _, entry := range q.running {
|
|
|
|
stats.Running = append(stats.Running, entry.item)
|
|
|
|
}
|
2019-06-28 06:29:57 +00:00
|
|
|
stats.Paused = q.paused
|
2017-03-05 07:56:08 +00:00
|
|
|
|
|
|
|
q.Unlock()
|
|
|
|
return stats
|
|
|
|
}
|
|
|
|
|
2024-05-13 20:58:21 +00:00
|
|
|
// Pause stops the queue from handing out new work items in Poll.
|
2019-06-28 06:29:57 +00:00
|
|
|
func (q *fifo) Pause() {
|
|
|
|
q.Lock()
|
|
|
|
q.paused = true
|
|
|
|
q.Unlock()
|
|
|
|
}
|
|
|
|
|
2024-02-16 09:04:13 +00:00
|
|
|
// Resume starts the queue again.
|
2019-06-28 06:29:57 +00:00
|
|
|
func (q *fifo) Resume() {
|
|
|
|
q.Lock()
|
|
|
|
q.paused = false
|
|
|
|
q.Unlock()
|
|
|
|
}
|
|
|
|
|
2024-02-16 09:04:13 +00:00
|
|
|
// KickAgentWorkers kicks all workers for a given agent.
|
|
|
|
func (q *fifo) KickAgentWorkers(agentID int64) {
|
|
|
|
q.Lock()
|
|
|
|
defer q.Unlock()
|
|
|
|
|
|
|
|
for w := range q.workers {
|
|
|
|
if w.agentID == agentID {
|
|
|
|
w.stop(fmt.Errorf("worker was kicked"))
|
|
|
|
delete(q.workers, w)
|
|
|
|
}
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
2017-03-05 07:56:08 +00:00
|
|
|
// helper function that loops through the queue and attempts to
|
2024-09-20 07:40:48 +00:00
|
|
|
// match the item to a single subscriber until context got cancel.
|
2017-03-05 07:56:08 +00:00
|
|
|
func (q *fifo) process() {
|
2024-09-20 07:40:48 +00:00
|
|
|
for {
|
|
|
|
select {
|
|
|
|
case <-time.After(processTimeInterval):
|
|
|
|
case <-q.ctx.Done():
|
|
|
|
return
|
|
|
|
}
|
2019-10-08 10:09:31 +00:00
|
|
|
|
2024-09-20 07:40:48 +00:00
|
|
|
q.Lock()
|
|
|
|
if q.paused {
|
|
|
|
q.Unlock()
|
|
|
|
continue
|
|
|
|
}
|
2019-06-28 06:29:57 +00:00
|
|
|
|
2024-09-20 07:40:48 +00:00
|
|
|
q.resubmitExpiredPipelines()
|
|
|
|
q.filterWaiting()
|
|
|
|
for pending, worker := q.assignToWorker(); pending != nil && worker != nil; pending, worker = q.assignToWorker() {
|
|
|
|
task, _ := pending.Value.(*model.Task)
|
|
|
|
task.AgentID = worker.agentID
|
|
|
|
delete(q.workers, worker)
|
|
|
|
q.pending.Remove(pending)
|
|
|
|
q.running[task.ID] = &entry{
|
|
|
|
item: task,
|
|
|
|
done: make(chan bool),
|
|
|
|
deadline: time.Now().Add(q.extension),
|
|
|
|
}
|
|
|
|
worker.channel <- task
|
2019-06-24 20:35:50 +00:00
|
|
|
}
|
2024-09-20 07:40:48 +00:00
|
|
|
q.Unlock()
|
2019-06-24 20:35:50 +00:00
|
|
|
}
|
|
|
|
}
|
|
|
|
|
2019-07-09 14:23:56 +00:00
|
|
|
func (q *fifo) filterWaiting() {
|
|
|
|
// resubmits all waiting tasks to pending, deps may have cleared
|
|
|
|
var nextWaiting *list.Element
|
|
|
|
for e := q.waitingOnDeps.Front(); e != nil; e = nextWaiting {
|
|
|
|
nextWaiting = e.Next()
|
2024-01-12 01:01:02 +00:00
|
|
|
task, _ := e.Value.(*model.Task)
|
2019-07-09 14:23:56 +00:00
|
|
|
q.pending.PushBack(task)
|
|
|
|
}
|
|
|
|
|
|
|
|
// rebuild waitingDeps
|
|
|
|
q.waitingOnDeps = list.New()
|
2021-09-24 14:29:26 +00:00
|
|
|
var filtered []*list.Element
|
2019-07-09 14:23:56 +00:00
|
|
|
var nextPending *list.Element
|
|
|
|
for e := q.pending.Front(); e != nil; e = nextPending {
|
|
|
|
nextPending = e.Next()
|
2024-01-12 01:01:02 +00:00
|
|
|
task, _ := e.Value.(*model.Task)
|
2019-07-09 14:23:56 +00:00
|
|
|
if q.depsInQueue(task) {
|
2021-10-12 07:25:13 +00:00
|
|
|
log.Debug().Msgf("queue: waiting due to unmet dependencies %v", task.ID)
|
2019-07-09 14:23:56 +00:00
|
|
|
q.waitingOnDeps.PushBack(task)
|
|
|
|
filtered = append(filtered, e)
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
|
|
|
// filter waiting tasks
|
|
|
|
for _, f := range filtered {
|
|
|
|
q.pending.Remove(f)
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
2019-06-24 20:35:50 +00:00
|
|
|
func (q *fifo) assignToWorker() (*list.Element, *worker) {
|
2017-03-05 07:56:08 +00:00
|
|
|
var next *list.Element
|
|
|
|
for e := q.pending.Front(); e != nil; e = next {
|
|
|
|
next = e.Next()
|
2024-01-12 01:01:02 +00:00
|
|
|
task, _ := e.Value.(*model.Task)
|
2021-10-12 07:25:13 +00:00
|
|
|
log.Debug().Msgf("queue: trying to assign task: %v with deps %v", task.ID, task.Dependencies)
|
2019-06-24 20:35:50 +00:00
|
|
|
|
2017-03-05 07:56:08 +00:00
|
|
|
for w := range q.workers {
|
2019-06-13 15:38:19 +00:00
|
|
|
if w.filter(task) {
|
2021-10-12 07:25:13 +00:00
|
|
|
log.Debug().Msgf("queue: assigned task: %v with deps %v", task.ID, task.Dependencies)
|
2019-06-24 20:35:50 +00:00
|
|
|
return e, w
|
2017-03-05 07:56:08 +00:00
|
|
|
}
|
|
|
|
}
|
|
|
|
}
|
2019-06-24 20:35:50 +00:00
|
|
|
|
|
|
|
return nil, nil
|
2017-03-05 07:56:08 +00:00
|
|
|
}
|
2019-06-13 15:38:19 +00:00
|
|
|
|
2022-10-18 01:24:12 +00:00
|
|
|
func (q *fifo) resubmitExpiredPipelines() {
|
2019-06-21 09:55:43 +00:00
|
|
|
for id, state := range q.running {
|
|
|
|
if time.Now().After(state.deadline) {
|
|
|
|
q.pending.PushFront(state.item)
|
|
|
|
delete(q.running, id)
|
|
|
|
close(state.done)
|
|
|
|
}
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
2023-03-20 03:50:56 +00:00
|
|
|
func (q *fifo) depsInQueue(task *model.Task) bool {
|
2019-06-13 15:38:19 +00:00
|
|
|
var next *list.Element
|
|
|
|
for e := q.pending.Front(); e != nil; e = next {
|
|
|
|
next = e.Next()
|
2023-03-20 03:50:56 +00:00
|
|
|
possibleDep, ok := e.Value.(*model.Task)
|
2021-10-12 07:25:13 +00:00
|
|
|
log.Debug().Msgf("queue: pending right now: %v", possibleDep.ID)
|
2019-06-13 15:38:19 +00:00
|
|
|
for _, dep := range task.Dependencies {
|
|
|
|
if ok && possibleDep.ID == dep {
|
|
|
|
return true
|
|
|
|
}
|
|
|
|
}
|
|
|
|
}
|
|
|
|
for possibleDepID := range q.running {
|
2021-10-12 07:25:13 +00:00
|
|
|
log.Debug().Msgf("queue: running right now: %v", possibleDepID)
|
2019-06-13 15:38:19 +00:00
|
|
|
for _, dep := range task.Dependencies {
|
|
|
|
if possibleDepID == dep {
|
|
|
|
return true
|
|
|
|
}
|
|
|
|
}
|
|
|
|
}
|
|
|
|
return false
|
|
|
|
}
|
2019-06-16 13:26:45 +00:00
|
|
|
|
2023-03-20 03:50:56 +00:00
|
|
|
func (q *fifo) updateDepStatusInQueue(taskID string, status model.StatusValue) {
|
2019-06-16 13:26:45 +00:00
|
|
|
var next *list.Element
|
|
|
|
for e := q.pending.Front(); e != nil; e = next {
|
|
|
|
next = e.Next()
|
2023-03-20 03:50:56 +00:00
|
|
|
pending, ok := e.Value.(*model.Task)
|
2019-06-16 13:26:45 +00:00
|
|
|
for _, dep := range pending.Dependencies {
|
|
|
|
if ok && taskID == dep {
|
2019-07-22 10:43:59 +00:00
|
|
|
pending.DepStatus[dep] = status
|
2019-06-16 13:26:45 +00:00
|
|
|
}
|
|
|
|
}
|
|
|
|
}
|
2019-07-09 14:23:56 +00:00
|
|
|
|
2019-06-16 13:26:45 +00:00
|
|
|
for _, running := range q.running {
|
|
|
|
for _, dep := range running.item.Dependencies {
|
|
|
|
if taskID == dep {
|
2019-07-22 10:43:59 +00:00
|
|
|
running.item.DepStatus[dep] = status
|
2019-06-16 13:26:45 +00:00
|
|
|
}
|
|
|
|
}
|
|
|
|
}
|
2019-07-09 14:23:56 +00:00
|
|
|
|
2021-05-25 11:05:31 +00:00
|
|
|
for e := q.waitingOnDeps.Front(); e != nil; e = next {
|
2019-07-09 14:23:56 +00:00
|
|
|
next = e.Next()
|
2023-03-20 03:50:56 +00:00
|
|
|
waiting, ok := e.Value.(*model.Task)
|
2019-07-09 14:23:56 +00:00
|
|
|
for _, dep := range waiting.Dependencies {
|
|
|
|
if ok && taskID == dep {
|
2019-07-22 10:43:59 +00:00
|
|
|
waiting.DepStatus[dep] = status
|
2019-07-09 14:23:56 +00:00
|
|
|
}
|
|
|
|
}
|
|
|
|
}
|
2019-06-16 13:26:45 +00:00
|
|
|
}
|
2019-06-19 06:36:13 +00:00
|
|
|
|
|
|
|
func (q *fifo) removeFromPending(taskID string) {
|
2021-10-12 07:25:13 +00:00
|
|
|
log.Debug().Msgf("queue: trying to remove %s", taskID)
|
2019-06-19 06:36:13 +00:00
|
|
|
var next *list.Element
|
|
|
|
for e := q.pending.Front(); e != nil; e = next {
|
|
|
|
next = e.Next()
|
2024-01-12 01:01:02 +00:00
|
|
|
task, _ := e.Value.(*model.Task)
|
2019-06-19 06:36:13 +00:00
|
|
|
if task.ID == taskID {
|
2021-10-12 07:25:13 +00:00
|
|
|
log.Debug().Msgf("queue: %s is removed from pending", taskID)
|
2019-06-19 06:36:13 +00:00
|
|
|
q.pending.Remove(e)
|
|
|
|
return
|
|
|
|
}
|
|
|
|
}
|
|
|
|
}
|