// WaitForCacheSync is a wrapper around cache.WaitForCacheSync that generates log messages
// indicating that the controller identified by controllerName is waiting for syncs, followed by
// either a successful or failed sync.
549188
func WaitForCacheSync(controllerName string, stopCh <-chan struct{}, cacheSyncs ...cache.InformerSynced) bool {
	klog.Infof("Waiting for caches to sync for %s controller", controllerName)

	if !cache.WaitForCacheSync(stopCh, cacheSyncs...) {
		utilruntime.HandleError(fmt.Errorf("unable to sync caches for %s controller", controllerName))
		return false
	}

	klog.Infof("Caches are synced for %s controller", controllerName)
	return true
}
783979
func (p *Process) waitForSync(downstream *discoverd.Instance) {
	p.Logger.Debug("waiting for downstream sync")

	stopCh := make(chan struct{})
	doneCh := make(chan struct{})

	var once sync.Once
	p.cancelSyncWait = func() {
		once.Do(func() { close(stopCh); <-doneCh })
	}

	go p.waitForSyncInner(downstream, stopCh, doneCh)
}
718125
func (w *KubeWaiter) WaitForStaticPodHashChange(nodeName, component, previousHash string) error {
	return wait.PollImmediate(constants.APICallRetryInterval, w.timeout, func() (bool, error) {

		hash, err := getStaticPodSingleHash(w.client, nodeName, component)
		if err != nil {
			return false, nil
		}
		// We should continue polling until the UID changes
		if hash == previousHash {
			return false, nil
		}

		return true, nil
	})
}
526852
func (helper *CacheHelper) DoResync(resyncEv datasync.ResyncEvent) error {
	var wasError error
	//idx.Put()
	ifaces, found := resyncEv.GetValues()[helper.Prefix]
	if found {
		// Step 1: fill the existing things
		resyncNames := map[string]interface{}{}
		for {
			item, stop := ifaces.GetNext()
			if stop {
				break
			}
			ifaceName, err := helper.ParseName(item.GetKey())
			if err != nil {
				wasError = err
			} else {
				current := proto.Clone(helper.DataPrototype)
				item.GetValue(current)
				helper.IDX.Put(ifaceName, current)
				resyncNames[ifaceName] = nil
			}
		}

		// Step 2:
		existingNames := []string{} //TODO
		for _, existingName := range existingNames {
			if _, found := resyncNames[existingName]; !found {
				helper.IDX.Delete(existingName)
			}
		}
	}
	return wasError
}
841916
func (app *App) CacheFor(handler interface{}, ttl time.Duration) func(ctx *Context) {
	opts := app.getCacheOpts()

	opts.TTL = ttl
	return app.Cache(handler, opts)
}
742170
func (w *Waiter) WaitForStaticPodSingleHash(_ string, _ string) (string, error) {
	return "", nil
}
109814
func (a *flexVolumeAttacher) WaitForAttach(spec *volume.Spec, devicePath string, _ *v1.Pod, timeout time.Duration) (string, error) {
	call := a.plugin.NewDriverCallWithTimeout(waitForAttachCmd, timeout)
	call.Append(devicePath)
	call.AppendSpec(spec, a.plugin.host, nil)

	status, err := call.Run()
	if isCmdNotSupportedErr(err) {
		return (*attacherDefaults)(a).WaitForAttach(spec, devicePath, timeout)
	} else if err != nil {
		return "", err
	}
	return status.DevicePath, nil
}
1258791
func (w *watchCache) WaitUntilFreshAndGet(resourceVersion uint64, key string, trace *utiltrace.Trace) (interface{}, bool, uint64, error) {
	err := w.waitUntilFreshAndBlock(resourceVersion, trace)
	defer w.RUnlock()
	if err != nil {
		return nil, false, 0, err
	}
	value, exists, err := w.store.GetByKey(key)
	return value, exists, w.resourceVersion, err
}
632250
func (c *LeaseCache) checkCacheForRequest(id string) (*SendResponse, error) {
	index, err := c.db.Get(cachememdb.IndexNameID, id)
	if err != nil {
		return nil, err
	}

	if index == nil {
		return nil, nil
	}

	// Cached request is found, deserialize the response
	reader := bufio.NewReader(bytes.NewReader(index.Response))
	resp, err := http.ReadResponse(reader, nil)
	if err != nil {
		c.logger.Error("failed to deserialize response", "error", err)
		return nil, err
	}

	sendResp, err := NewSendResponse(&api.Response{Response: resp}, index.Response)
	if err != nil {
		c.logger.Error("failed to create new send response", "error", err)
		return nil, err
	}
	sendResp.CacheMeta.Hit = true

	respTime, err := http.ParseTime(resp.Header.Get("Date"))
	if err != nil {
		c.logger.Error("failed to parse cached response date", "error", err)
		return nil, err
	}
	sendResp.CacheMeta.Age = time.Now().Sub(respTime)

	return sendResp, nil
}
965883
func (s *ClusterHealthService) WaitForStatus(waitForStatus string) *ClusterHealthService {
	s.waitForStatus = waitForStatus
	return s
}
920496
func waitForState(errCh chan<- error, target string, refresh stateRefreshFunc) error {
	ctx := context.TODO()
	err := retry.Config{
		RetryDelay: (&retry.Backoff{InitialBackoff: 2 * time.Second, MaxBackoff: 2 * time.Second, Multiplier: 2}).Linear,
	}.Run(ctx, func(ctx context.Context) error {
		state, err := refresh()
		if err != nil {
			return err
		}
		if state == target {
			return nil
		}
		return fmt.Errorf("retrying for state %s, got %s", target, state)
	})
	errCh <- err
	return err
}
376620
func (m *PluginGroup) WaitForFirstFingerprint(ctx context.Context) (<-chan struct{}, error) {
	m.mLock.Lock()
	defer m.mLock.Unlock()
	if m.shutdown {
		return nil, fmt.Errorf("plugin group already shutdown")
	}

	var wg sync.WaitGroup
	for i := range m.managers {
		manager, ok := m.managers[i].(FingerprintingPluginManager)
		if !ok {
			continue
		}
		logger := m.logger.With("plugin-type", manager.PluginType())
		wg.Add(1)
		go func() {
			defer wg.Done()
			logger.Debug("waiting on plugin manager initial fingerprint")
			select {
			case <-manager.WaitForFirstFingerprint(ctx):
				select {
				case <-ctx.Done():
					logger.Warn("timeout waiting for plugin manager to be ready")
				default:
					logger.Debug("finished plugin manager initial fingerprint")
				}
			}
		}()
	}

	ret := make(chan struct{})
	go func() {
		wg.Wait()
		close(ret)
	}()
	return ret, nil
}
903959
func (c *Controller) Wait(d time.Duration) {
	wg := sync.WaitGroup{}
	wg.Add(len(c.mockers))
	for _, m := range c.mockers {
		go func(m Mocker) {
			defer wg.Done()
			m.MinimockWait(d)
		}(m)
	}

	wg.Wait()
}
268290
func (s *DeleteByQueryService) WaitForCompletion(waitForCompletion bool) *DeleteByQueryService {
	s.waitForCompletion = &waitForCompletion
	return s
}
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
}
573695
func (s *TasksGetTaskService) WaitForCompletion(waitForCompletion bool) *TasksGetTaskService {
	s.waitForCompletion = &waitForCompletion
	return s
}