// device. Device attribute is used to infer the appropriate sharding
        // within TPUs for this op.
        // TODO(b/183598857): Use explicit sharding ops from the front-end.
        // For example, dequeue ops generated by
        // tensorflow/python/tpu/tpu_feed.py
        if (!tensorflow::IsTPUReplicatedCore(attr.getValue())) {
          inner_op->removeAttr(kDeviceAttr);
        }
      }
    });
  }

:r.Callbacks("once memory").add(function(){W.remove(a,[b+"queue",c])})})}}),r.fn.extend({queue:function(a,b){var c=2;return"string"!=typeof a&&(b=a,a="fx",c--),arguments.length<c?r.queue(this[0],a):void 0===b?this:this.each(function(){var c=r.queue(this,a,b);r._queueHooks(this,a),"fx"===a&&"inprogress"!==c[0]&&r.dequeue(this,a)})},dequeue:function(a){return this.each(function(){r.dequeue(this,a)})},clearQueue:function(a){return this.queue(a||"fx",[])},promise:function(a,b){var c,d=1,e=r.Deferred...

func blockTimerChan(c *hchan) {
	t := c.timer
	t.lock()
	t.trace("blockTimerChan")
	if !t.isChan {
		badTimer()
	}

	t.blocked++

	// If this is the first enqueue after a recent dequeue,
	// the timer may still be in the heap but marked as a zombie.
	// Unmark it in this case, if the timer is still pending.
	if t.state&timerHeaped != 0 && t.state&timerZombie != 0 && t.when > 0 {

}

func (t *trackingWorkqueue[T]) Add(item T) {
	t.queue(item)
	t.limiter.Add(item)
}
func (t *trackingWorkqueue[T]) AddAfter(item T, duration time.Duration) {
	t.Add(item)
}
func (t *trackingWorkqueue[T]) AddRateLimited(item T) {
	t.Add(item)
}
func (t *trackingWorkqueue[T]) Get() (T, bool) {
	item, shutdown := t.limiter.Get()
	t.dequeue(item)
	return item, shutdown
}

        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);

	idle    chan struct{} // if non-nil, closed when active becomes 0
}

// NewQueue returns a Queue that executes up to maxActive items in parallel.
//
// maxActive must be positive.
func NewQueue(maxActive int) *Queue {
	if maxActive < 1 {
		panic(fmt.Sprintf("par.NewQueue called with nonpositive limit (%d)", maxActive))
	}

	q := &Queue{
		maxActive: maxActive,
		st:        make(chan queueState, 1),
	}

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)