MCPcopy Index your code
hub / github.com/flant/shell-operator / AddLast

Method AddLast

pkg/task/queue/task_queue.go:327–393  ·  view source on GitHub ↗

addLast adds a new tail element. It implements the merging logic for HookRun tasks by scanning the whole queue.

(tasks ...task.Task)

Source from the content-addressed store, hash-verified

325// addLast adds a new tail element.
326// It implements the merging logic for HookRun tasks by scanning the whole queue.
327func (q *TaskQueue) AddLast(tasks ...task.Task) {
328 defer q.MeasureActionTime("AddLast")()
329
330 for _, t := range tasks {
331 q.lazydebug("adding task to queue", func() []any {
332 return []any{
333 slog.String(pkg.LogKeyQueue, q.Name),
334 slog.String(pkg.LogKeyTaskID, t.GetId()),
335 slog.String(pkg.LogKeyTaskType, string(t.GetType())),
336 slog.String(pkg.LogKeyTaskDescription, t.GetDescription()),
337 slog.Int(pkg.LogKeyQueueLengthBefore, q.storage.Length()),
338 }
339 })
340
341 if q.storage.Get(t.GetId()) != nil {
342 q.logger.Warn("task collision detected, unexpected behavior possible", slog.String(pkg.LogKeyQueue, q.Name), slog.String(pkg.LogKeyTaskID, t.GetId()))
343 }
344
345 q.storage.AddLast(t)
346 q.queueTasksCounter.Add(t)
347
348 // TODO: skip compactable check if is already compactable
349
350 taskType := t.GetType()
351 if _, ok := q.compactableTypes[taskType]; ok {
352 q.lazydebug("task is mergeable, marking queue as dirty", func() []any {
353 return []any{
354 slog.String(pkg.LogKeyQueue, q.Name),
355 slog.String(pkg.LogKeyTaskID, t.GetId()),
356 slog.String(pkg.LogKeyTaskType, string(taskType)),
357 slog.Int(pkg.LogKeyQueueLength, q.storage.Length()),
358 slog.Bool(pkg.LogKeyQueueIsDirty, q.queueTasksCounter.IsAnyCapReached()),
359 }
360 })
361
362 // Only trigger compaction if queue is getting long and we have mergeable tasks
363 if q.storage.Length() > compactionThreshold && q.queueTasksCounter.IsAnyCapReached() {
364 q.lazydebug("triggering compaction due to queue length", func() []any {
365 return []any{
366 slog.String(pkg.LogKeyQueue, q.Name),
367 slog.Int(pkg.LogKeyQueueLength, q.storage.Length()),
368 slog.Int(pkg.LogKeyCompactionThreshold, compactionThreshold),
369 }
370 })
371
372 currentQueue := q.storage.Length()
373 q.compaction(q.queueTasksCounter.GetReachedCap())
374
375 q.lazydebug("compaction finished", func() []any {
376 return []any{
377 slog.String(pkg.LogKeyQueue, q.Name),
378 slog.Int(pkg.LogKeyQueueLengthBefore, currentQueue),
379 slog.Int(pkg.LogKeyQueueLengthAfter, q.storage.Length()),
380 }
381 })
382 }
383 } else {
384 q.lazydebug("task is not mergeable", func() []any {

Callers

nothing calls this directly

Calls 13

MeasureActionTimeMethod · 0.95
lazydebugMethod · 0.95
compactionMethod · 0.95
IsAnyCapReachedMethod · 0.80
GetReachedCapMethod · 0.80
GetIdMethod · 0.65
GetTypeMethod · 0.65
GetDescriptionMethod · 0.65
GetMethod · 0.65
AddLastMethod · 0.65
AddMethod · 0.65
StringMethod · 0.45

Tested by

no test coverage detected