}
    for (existingCall in readyAsyncCalls) {
      if (existingCall.host == host) return existingCall
    }
    return null
  }

  /**
   * Cancel all calls currently enqueued or executing. Includes calls executed both
   * [synchronously][Call.execute] and [asynchronously][Call.enqueue].
   */
  fun cancelAll() {
    this.withLock {
      for (call in readyAsyncCalls) {

      "FINE: Q10000 scheduled after 100 µs: task",
      "FINE: Q10000 starting              : task",
      "FINE: Q10000 finished run in   0 µs: task",
    )
  }

  /** Inspect how many runnables have been enqueued. If none then we're truly sequential. */
  @Test fun singleQueueIsSerial() {
    redQueue.execute("task one", 100.µs) {
      log += "one:run@${taskFaker.nanoTime} parallel=${taskFaker.isParallel}"
    }

	set := newStatefulSet(3)
	pod := newStatefulSetPod(set, 0)
	ssc.addPod(logger, pod)
	ssc.queue.ShutDown()
	key, _ := ssc.queue.Get()
	if key != "" {
		t.Errorf("StatefulSet enqueued key for Pod with no Set %s", key)
	}
}

func TestStatefulSetControllerUpdatePod(t *testing.T) {
	logger, ctx := ktesting.NewTestContext(t)
	ssc, _, om, _ := newFakeStatefulSetController(ctx)

			manager.enqueueDaemonSet = func(ds *apps.DaemonSet) {
				if ds.Name == "ds" {
					enqueued = true
				}
			}

			enqueued = false
			manager.updateNode(logger, c.oldNode, c.newNode)
			if enqueued != c.shouldEnqueue {
				t.Errorf("Test case: '%s', expected: %t, got: %t", c.test, c.shouldEnqueue, enqueued)
			}
		}
	}
}

    }

    /**
     * Returns true if the request contains conditions that save the server from sending a response
     * that the client has locally. When a request is enqueued with its own conditions, the built-in
     * response cache won't be used.
     */
    private fun hasConditions(request: Request): Boolean =

}

func (gc *GarbageCollector) runAttemptToDeleteWorker(ctx context.Context) {
	for gc.processAttemptToDeleteWorker(ctx) {
	}
}

var enqueuedVirtualDeleteEventErr = goerrors.New("enqueued virtual delete event")

var namespacedOwnerOfClusterScopedObjectErr = goerrors.New("cluster-scoped objects cannot refer to namespaced owners")

			newPVC, ok := new.(*v1.PersistentVolumeClaim)
			if !ok {
				return
			}
			newReq := newPVC.Spec.Resources.Requests[v1.ResourceStorage]
			newCap := newPVC.Status.Capacity[v1.ResourceStorage]
			// PVC will be enqueued under 2 circumstances
			// 1. User has increased PVC's request capacity --> volume needs to be expanded

        }
      }
    }
  }

  internal class CallReference(
    referent: RealCall,
    /**
     * Captures the stack trace at the time the Call is executed or enqueued. This is helpful for
     * identifying the origin of connection leaks.
     */
    val callStackTrace: Any?,
  ) : WeakReference<RealCall>(referent)

    // Also ideally this host callback should be on source stream rather than
    // destination stream, but when this function returns, the send requests
    // might not be enqueued to the stream yet, we put it on destination stream.
    TensorReference input_reference(*input);
    std::move(return_substream).release();
    return dst_device_to_device_stream->DoHostCallback(

	// initialization doesn't have dependencies that will block, there is no need to add any timeout
	// here. Prior to this explicit wait, we were implicitly waiting by receive() not sending to
	// reqChannel and the connection not being enqueued for pushes to pushChannel until the
	// initialization is complete.
	<-con.initialized

	for {
		// Go select{} statements are not ordered; the same channel can be chosen many times.