JitterUntil(f, period, 0.0, true, stopCh)
}

// UntilWithContext loops until context is done, running f every period.
//
// UntilWithContext is syntactic sugar on top of JitterUntilWithContext
// with zero jitter factor and with sliding = true (which means the timer
// for period starts after the f completes).
func UntilWithContext(ctx context.Context, f func(context.Context), period time.Duration) {

	for i := 0; i < workers; i++ {
		go wait.UntilWithContext(ctx, rq.worker(rq.queue), time.Second)
		go wait.UntilWithContext(ctx, rq.worker(rq.missingUsageQueue), time.Second)
	}
	// the timer for how often we do a full recalculation across all quotas
	if rq.resyncPeriod() > 0 {
		go wait.UntilWithContext(ctx, rq.enqueueAll, rq.resyncPeriod())
	} else {

	// handles storageversion creation/update with non-existing id. The latter should rarely
	// happen. It's okay for the two workers to conflict on update.
	go wait.UntilWithContext(ctx, c.runLeaseWorker, time.Second)
	go wait.UntilWithContext(ctx, c.runStorageVersionWorker, time.Second)

	<-ctx.Done()
}

func (c *Controller) runLeaseWorker(ctx context.Context) {
	for c.processNextLease(ctx) {
	}
}

		return
	}

	logger := klog.FromContext(ctx)
	logger.V(5).Info("Starting workers")
	for i := 0; i < workers; i++ {
		go wait.UntilWithContext(ctx, e.syncServiceAccount, 0)
		go wait.UntilWithContext(ctx, e.syncSecret, 0)
	}
	<-ctx.Done()
	logger.V(1).Info("Shutting down")
}

func (e *TokensController) queueServiceAccountSync(obj interface{}) {

	if !cache.WaitForNamedCacheSync("legacy-service-account-token-cleaner", ctx.Done(), tc.saInformerSynced, tc.secretInformerSynced, tc.podInformerSynced) {
		return
	}

	go wait.UntilWithContext(ctx, tc.evaluateSATokens, tc.syncInterval)

	<-ctx.Done()
}

func (tc *LegacySATokenCleaner) evaluateSATokens(ctx context.Context) {
	logger := klog.FromContext(ctx)

	now := tc.clock.Now().UTC()

		return
	}

	ctrl.initializeCaches(logger, ctrl.volumeLister, ctrl.claimLister)

	go wait.Until(func() { ctrl.resync(ctx) }, ctrl.resyncPeriod, ctx.Done())
	go wait.UntilWithContext(ctx, ctrl.volumeWorker, time.Second)
	go wait.UntilWithContext(ctx, ctrl.claimWorker, time.Second)

	metrics.Register(ctrl.volumes.store, ctrl.claims, &ctrl.volumePluginMgr)

	<-ctx.Done()
}

	if !cache.WaitForNamedCacheSync("TTL after finished", ctx.Done(), tc.jListerSynced) {
		return
	}

	for i := 0; i < workers; i++ {
		go wait.UntilWithContext(ctx, tc.worker, time.Second)
	}

	<-ctx.Done()
}

func (tc *Controller) addJob(logger klog.Logger, obj interface{}) {
	job := obj.(*batch.Job)
	logger.V(4).Info("Adding job", "job", klog.KObj(job))

	// shutted down, so we need to shut it down in a separate goroutine.
	go func() {
		defer utilruntime.HandleCrash()
		defer qm.resourceChanges.ShutDown()

		<-ctx.Done()
	}()
	wait.UntilWithContext(ctx, qm.runProcessResourceChanges, 1*time.Second)

	// Stop any running monitors.
	qm.monitorLock.Lock()
	defer qm.monitorLock.Unlock()
	monitors := qm.monitors
	stopped := 0

	defer gcc.nodeQueue.ShutDown()
	defer logger.Info("Shutting down GC controller")

	if !cache.WaitForNamedCacheSync("GC", ctx.Done(), gcc.podListerSynced, gcc.nodeListerSynced) {
		return
	}

	go wait.UntilWithContext(ctx, gcc.gc, gcc.gcCheckPeriod)

	<-ctx.Done()
}

func (gcc *PodGCController) gc(ctx context.Context) {
	pods, err := gcc.podLister.List(labels.Everything())
	if err != nil {

	defer logger.Info("Shutting down expand controller")

	if !cache.WaitForNamedCacheSync("expand", ctx.Done(), expc.pvcsSynced) {
		return
	}

	for i := 0; i < defaultWorkerCount; i++ {
		go wait.UntilWithContext(ctx, expc.runWorker, time.Second)
	}

	<-ctx.Done()
}

func (expc *expandController) runWorker(ctx context.Context) {
	for expc.processNextWorkItem(ctx) {
	}
}