if (IsNextIteration(e->src())) {
          ++num_ready_inputs[node->id()];
        }
      }
    }
  }

  // dequeue is used to ensure that the nodes are first-in-first-out.  This
  // order guarantees that the exits in the ready queue are visited before
  // nodes that will become ready in the future.
  std::deque<Node*> ready;
  ready.push_back(src_node);

package timer

import (
	"container/heap"
	"sync"
	"time"
)

type Queue struct {
	heap            timerHeap
	mutex           sync.Mutex
	stopCh          chan struct{}
	resetTimerCh    chan struct{}
	stopping        bool
	timer           *time.Timer
	currentDeadline time.Time
}

func NewQueue() *Queue {
	q := &Queue{
		heap:         make(timerHeap, 0),
		timer:        time.NewTimer(1 * time.Minute),

}

type basicWorkQueue struct {
	clock clock.Clock
	lock  sync.Mutex
	queue map[types.UID]time.Time
}

var _ WorkQueue = &basicWorkQueue{}

// NewBasicWorkQueue returns a new basic WorkQueue with the provided clock
func NewBasicWorkQueue(clock clock.Clock) WorkQueue {
	queue := make(map[types.UID]time.Time)
	return &basicWorkQueue{queue: queue, clock: clock}
}

  private val log = mutableListOf<String>()
  private val redQueue = taskRunner.newQueue()
  private val blueQueue = taskRunner.newQueue()
  private val greenQueue = taskRunner.newQueue()

  @AfterEach
  internal fun tearDown() {
    taskFaker.close()
  }

  @Test fun executeDelayed() {
    redQueue.execute("task", 100.µs) {
      log += "run@${taskFaker.nanoTime}"
    }

// requeueStuckDeployment checks whether the provided deployment needs to be synced for a progress
// check. It returns the time after the deployment will be requeued for the progress check, 0 if it
// will be requeued now, or -1 if it does not need to be requeued.
func (dc *DeploymentController) requeueStuckDeployment(ctx context.Context, d *apps.Deployment, newStatus apps.DeploymentStatus) time.Duration {
	logger := klog.FromContext(ctx)

	// only queue deleted things that haven't been finalized by us
	if newCRD.DeletionTimestamp.IsZero() || !apiextensionshelpers.CRDHasFinalizer(newCRD, apiextensionsv1.CustomResourceCleanupFinalizer) {
		return
	}

	// always requeue resyncs just in case
	if oldCRD.ResourceVersion == newCRD.ResourceVersion {
		c.enqueue(newCRD)
		return
	}

	key, quit := rsc.queue.Get()
	if quit {
		return false
	}
	defer rsc.queue.Done(key)

	err := rsc.syncHandler(ctx, key)
	if err == nil {
		rsc.queue.Forget(key)
		return true
	}

	utilruntime.HandleError(fmt.Errorf("sync %q failed with %v", key, err))
	rsc.queue.AddRateLimited(key)

	return true
}

			ObjectTransform: kubelib.StripNodeUnusedFields,
		}),
	}
	c.queue = controllers.NewQueue("ingress status",
		controllers.WithReconciler(c.Reconcile),
		controllers.WithMaxAttempts(5))

	// For any ingress change, enqueue it - we may need to update the status.
	c.ingresses.AddEventHandler(controllers.ObjectHandler(c.queue.AddObject))

func WithGenericReconciler(f func(key any) error) func(q *Queue) {
	return func(q *Queue) {
		q.workFn = func(key any) error {
			return f(key)
		}
	}
}

// NewQueue creates a new queue
func NewQueue(name string, options ...func(*Queue)) Queue {
	q := Queue{
		name:        name,
		closed:      make(chan struct{}),

	key, quit := rv.queue.Get()
	if quit {
		return false
	}
	defer rv.queue.Done(key)

	err := rv.sync(ctx, key)
	if err == nil {
		rv.queue.Forget(key)
		return true
	}

	klog.FromContext(ctx).V(2).Info("Error syncing SVM resource, retrying", "svm", key, "err", err)
	rv.queue.AddRateLimited(key)

	return true
}