// waitForCacheSync waits for k8s caches to be synchronized for the given
// resource. Returns once all resourcesNames are synchronized with cilium-agent.
774840
func (d *Daemon) waitForCacheSync(resourceNames ...string) {
	for _, resourceName := range resourceNames {
		d.k8sResourceSyncedMu.RLock()
		c, ok := d.k8sResourceSynced[resourceName]
		d.k8sResourceSyncedMu.RUnlock()
		if !ok {
			continue
		}
		<-c
	}
}
565668
func (o *oracleVolumeSource) waitForResourceStatus(
	fetch func(name string, desiredStatus string) (complete bool, err error),
	name, state string,
	timeout time.Duration,
) error {

	timeoutTimer := o.clock.NewTimer(timeout)
	defer timeoutTimer.Stop()

	retryTimer := o.clock.NewTimer(0)
	defer retryTimer.Stop()

	for {
		select {
		case <-retryTimer.Chan():
			done, err := fetch(name, state)
			if err != nil {
				return err
			}
			if done {
				return nil
			}
			retryTimer.Reset(2 * time.Second)
		case <-timeoutTimer.Chan():
			return errors.Errorf(
				"timed out waiting for resource %q to transition to %v",
				name, state,
			)
		}
	}
}
1073945
func (t *tokenWithProvider) fetchFromCache(ctx context.Context) error {
	key, err := t.provider.CacheKey(ctx)
	if err != nil {
		return err
	}
	tok, err := t.cache.GetToken(key)
	if err != nil {
		return err
	}
	t.token = tok
	return nil
}
841916
func (app *App) CacheFor(handler interface{}, ttl time.Duration) func(ctx *Context) {
	opts := app.getCacheOpts()

	opts.TTL = ttl
	return app.Cache(handler, opts)
}
885799
func (t *Transaction) WaitForSync(waitForSync bool) *Transaction {
	t.waitForSync = &waitForSync
	return t
}
879920
func (c *cache) start(a *Allocator) waitChan {
	listDone := make(waitChan)

	c.mutex.Lock()

	// start with a fresh nextCache
	c.nextCache = idMap{}
	c.nextKeyCache = keyMap{}
	c.mutex.Unlock()

	c.stopWatchWg.Add(1)

	go func() {
		<-c.backend.Connected()
		logger := c.getLogger()
		logger.Info("Starting to watch allocation changes")

		watcher := c.backend.ListAndWatch(c.prefix, c.prefix, 512)

		for {
			select {
			case event, ok := <-watcher.Events:
				if !ok {
					goto abort
				}
				if event.Typ == kvstore.EventTypeListDone {
					c.mutex.Lock()
					// nextCache is valid, point the live cache to it
					c.cache = c.nextCache
					c.keyCache = c.nextKeyCache
					c.mutex.Unlock()

					// report that the list operation has
					// been completed and the allocator is
					// ready to use
					close(listDone)
					continue
				}

				id := c.keyToID(event.Key, c.deleteInvalidPrefixes)
				if id != 0 {
					c.mutex.Lock()

					var key AllocatorKey

					if len(event.Value) > 0 {
						var err error
						key, err = a.keyType.PutKey(string(event.Value))
						if err != nil {
							c.getLogger().WithError(err).WithField(fieldKey, event.Value).
								Warning("Unable to unmarshal allocator key")
						}
					}
					debugFields := c.getLogger().WithFields(logrus.Fields{fieldKey: key, fieldID: id})

					switch event.Typ {
					case kvstore.EventTypeCreate:
						kvstore.Trace("Adding id to cache", nil, debugFields.Data)
						c.nextCache[id] = key
						if key != nil {
							c.nextKeyCache[key.GetKey()] = id
						}
						a.idPool.Remove(id)

					case kvstore.EventTypeModify:
						kvstore.Trace("Modifying id in cache", nil, debugFields.Data)
						if k, ok := c.nextCache[id]; ok {
							delete(c.nextKeyCache, k.GetKey())
						}

						c.nextCache[id] = key
						if key != nil {
							c.nextKeyCache[key.GetKey()] = id
						}

					case kvstore.EventTypeDelete:
						kvstore.Trace("Removing id from cache", nil, debugFields.Data)

						if a.enableMasterKeyProtection {
							if value := a.localKeys.lookupID(id); value != "" {
								a.recreateMasterKey(id, value, true)
								break
							}
						}

						if k, ok := c.nextCache[id]; ok && k != nil {
							delete(c.nextKeyCache, k.GetKey())
						}

						delete(c.nextCache, id)
						a.idPool.Insert(id)
					}
					c.mutex.Unlock()

					if a.events != nil {
						a.events <- AllocatorEvent{
							Typ: event.Typ,
							ID:  idpool.ID(id),
							Key: key,
						}
					}
				}

			case <-c.stopChan:
				goto abort
			}
		}

	abort:
		watcher.Stop()
		// Signal that watcher is done
		c.stopWatchWg.Done()
	}()

	return listDone
}
316868
func (s *SnapshotCreateService) WaitForCompletion(waitForCompletion bool) *SnapshotCreateService {
	s.waitForCompletion = &waitForCompletion
	return s
}
983069
func (w *Waiter) WaitForStaticPodControlPlaneHashes(_ string) (map[string]string, error) {
	return map[string]string{
		constants.KubeAPIServer:         "",
		constants.KubeControllerManager: "",
		constants.KubeScheduler:         "",
	}, nil
}
513002
func (kubemarkController *KubemarkController) WaitForCacheSync(stopCh chan struct{}) bool {
	return controller.WaitForCacheSync("kubemark", stopCh,
		kubemarkController.externalCluster.rcSynced,
		kubemarkController.externalCluster.podSynced,
		kubemarkController.kubemarkCluster.nodeSynced)
}
1046211
func WaitFunc(wait func(context.Context, *cdp.Frame, ...cdp.NodeID) ([]*cdp.Node, error)) QueryOption {
	return func(s *Selector) {
		s.wait = wait
	}
}
141400
func (cache *DiskBlockCacheLocal) WaitUntilStarted() error {
	select {
	case <-cache.startedCh:
		return nil
	case <-cache.startErrCh:
		return DiskBlockCacheError{"error starting channel"}
	}
}
1070328
func (a *Allocator) WaitForInitialSync(ctx context.Context) error {
	select {
	case <-a.initialListDone:
	case <-ctx.Done():
		return fmt.Errorf("identity sync with kvstore was cancelled: %s", ctx.Err())
	}

	return nil
}
200138
func (e *edge) slowCacheFunc(dep *dep) ResultBasedCacheFunc {
	if e.cacheMap == nil {
		return nil
	}
	return e.cacheMap.Deps[int(dep.index)].ComputeDigestFunc
}
495169
func WaitForRCToStabilize(c kubernetes.Interface, ns, name string, timeout time.Duration) error {
	options := metav1.ListOptions{FieldSelector: fields.Set{
		"metadata.name":      name,
		"metadata.namespace": ns,
	}.AsSelector().String()}
	w, err := c.Core().ReplicationControllers(ns).Watch(options)
	if err != nil {
		return err
	}
	_, err = watch.Until(timeout, w, func(event watch.Event) (bool, error) {
		switch event.Type {
		case watch.Deleted:
			return false, apierrs.NewNotFound(schema.GroupResource{Resource: "replicationcontrollers"}, "")
		}
		switch rc := event.Object.(type) {
		case *v1.ReplicationController:
			if rc.Name == name && rc.Namespace == ns &&
				rc.Generation <= rc.Status.ObservedGeneration &&
				*(rc.Spec.Replicas) == rc.Status.Replicas {
				return true, nil
			}
			glog.Infof("Waiting for rc %s to stabilize, generation %v observed generation %v spec.replicas %d status.replicas %d",
				name, rc.Generation, rc.Status.ObservedGeneration, *(rc.Spec.Replicas), rc.Status.Replicas)
		}
		return false, nil
	})
	return err
}
493666
func WaitForService(c kubernetes.Interface, namespace, name string, exist bool, interval, timeout time.Duration) error {
	err := wait.PollImmediate(interval, timeout, func() (bool, error) {
		_, err := c.Core().Services(namespace).Get(name, metav1.GetOptions{})
		switch {
		case err == nil:
			glog.Infof("Service %s in namespace %s found.", name, namespace)
			return exist, nil
		case apierrs.IsNotFound(err):
			glog.Infof("Service %s in namespace %s disappeared.", name, namespace)
			return !exist, nil
		case !IsRetryableAPIError(err):
			glog.Info("Non-retryable failure while getting service.")
			return false, err
		default:
			glog.Infof("Get service %s in namespace %s failed: %v", name, namespace, err)
			return false, nil
		}
	})
	if err != nil {
		stateMsg := map[bool]string{true: "to appear", false: "to disappear"}
		return fmt.Errorf("error waiting for service %s/%s %s: %v", namespace, name, stateMsg[exist], err)
	}
	return nil
}
230835
func (tt *timeoutThread) waitForNonEmptyQueue() bool {
	tt.mu.Lock()
	queueNotEmpty := tt.queueNotEmpty
	tt.mu.Unlock()

	select {
	// a) Always check these channels, regardless of the state.
	case <-tt.maxDuration.C:
		// Max duration is up. Stop buffering. Do not error out entries explicitly.
		tt.sb.stopBufferingDueToMaxDuration()
		return true
	case <-tt.stopChan:
		// Failover ended before timeout. Do nothing.
		return true
	// b) State-specific check.
	case <-queueNotEmpty:
		// At least one entry present. Check its timeout in the next iteration.
		return false
	}
}
15879
func (lk *etcdLock) waitForLock() {
	// If timeout is not specified, set it to high value
	timeoutIntvl := time.Second * time.Duration(20000)
	if lk.timeout != 0 {
		timeoutIntvl = time.Second * time.Duration(lk.timeout)
	}

	log.Infof("Waiting to acquire lock (%s/%s)", lk.name, lk.myID)

	// Create a timer
	timer := time.NewTimer(timeoutIntvl)
	defer timer.Stop()

	// Wait for changes
	for {
		// wait on watch channel for holder to release the lock
		select {
		case <-timer.C:
			lk.mutex.Lock()
			if lk.timeout != 0 {
				lk.mutex.Unlock()
				log.Infof("Lock timeout on lock %s/%s", lk.name, lk.myID)

				lk.eventChan <- LockEvent{EventType: LockAcquireTimeout}

				log.Infof("Lock acquire timed out. Stopping lock")

				lk.watchCancel()

				// Release the lock
				lk.Release()

				return
			}
			lk.mutex.Unlock()
		case watchResp := <-lk.watchCh:
			if watchResp != nil {
				log.Debugf("Received watch notification(%s/%s): %+v", lk.name, lk.myID, watchResp)

				if watchResp.Action == "expire" || watchResp.Action == "delete" ||
					watchResp.Action == "compareAndDelete" {
					log.Infof("Retrying to acquire lock")
					return
				}
			}
		case <-lk.stopChan:
			log.Infof("Stopping lock")
			lk.watchCancel()
			return
		}
	}
}