// WaitForAttach implements Attacher.WaitForAttach. It's called by kubelet to
// attach volume onto the node.
// This method is idempotent, callers are responsible for retrying on failure.
345751
func (attacher *rbdAttacher) WaitForAttach(spec *volume.Spec, devicePath string, pod *v1.Pod, timeout time.Duration) (string, error) {
	klog.V(4).Infof("rbd: waiting for attach volume (name: %s) for pod (name: %s, uid: %s)", spec.Name(), pod.Name, pod.UID)
	mounter, err := attacher.plugin.createMounterFromVolumeSpecAndPod(spec, pod)
	if err != nil {
		klog.Warningf("failed to create mounter: %v", spec)
		return "", err
	}
	realDevicePath, err := attacher.manager.AttachDisk(*mounter)
	if err != nil {
		return "", err
	}
	klog.V(3).Infof("rbd: successfully wait for attach volume (spec: %s, pool: %s, image: %s) at %s", spec.Name(), mounter.Pool, mounter.Image, realDevicePath)
	return realDevicePath, nil
}
933545
func (c *sioClient) WaitForDetachedDevice(token string) error {
	if token == "" {
		return fmt.Errorf("invalid detach token")
	}

	// wait for attach.Token to show up in local device list
	ticker := time.NewTicker(time.Second)
	defer ticker.Stop()
	timer := time.NewTimer(30 * time.Second)
	defer timer.Stop()

	for {
		select {
		case <-ticker.C:
			devMap, err := c.Devs()
			if err != nil {
				klog.Error(log("failed while waiting for volume to unmap/detach: %v", err))
				return err
			}
			go func() {
				klog.V(4).Info(log("waiting for volume %s to be unmapped/detached", token))
			}()
			// cant find vol id, then ok.
			if _, ok := devMap[token]; !ok {
				return nil
			}
		case <-timer.C:
			klog.Error(log("timed out while waiting for volume %s to be unmapped/detached", token))
			return fmt.Errorf("volume detach timeout")
		}
	}
}
490576
func (s *persistentRemotes) WaitSelect(ctx context.Context) <-chan api.Peer {
	c := make(chan api.Peer, 1)
	s.RLock()
	done := make(chan struct{})
	go func() {
		select {
		case <-ctx.Done():
			s.c.Broadcast()
		case <-done:
		}
	}()
	go func() {
		defer s.RUnlock()
		defer close(c)
		defer close(done)
		for {
			if ctx.Err() != nil {
				return
			}
			p, err := s.Select()
			if err == nil {
				c <- p
				return
			}
			s.c.Wait()
		}
	}()
	return c
}
962539
func (s *gceOps) waitForDetach(
	diskURL string,
	timeout time.Duration,
) error {

	_, err := task.DoRetryWithTimeout(
		func() (interface{}, bool, error) {
			inst, err := s.describeinstance()
			if err != nil {
				return nil, true, err
			}

			for _, d := range inst.Disks {
				if d.Source == diskURL {
					return nil, true,
						fmt.Errorf("disk: %s is still attached to instance: %s",
							diskURL, s.inst.name)
				}
			}

			return nil, false, nil

		},
		storageops.ProviderOpsTimeout,
		storageops.ProviderOpsRetryInterval)

	return err
}
513416
func (s *UpdateByQueryService) WaitForCompletion(waitForCompletion bool) *UpdateByQueryService {
	s.waitForCompletion = &waitForCompletion
	return s
}
639719
func (c *MockSSHClient) WaitForSSH(maxWait time.Duration) error {
	if c.MockWaitForSSH != nil {
		return c.MockWaitForSSH(maxWait)
	}
	return ErrNotImplemented
}
1227624
func WaitForCluster(ctx context.Context, n *Node) (cluster *api.Cluster, err error) {
	watch, cancel := state.Watch(n.MemoryStore().WatchQueue(), api.EventCreateCluster{})
	defer cancel()

	var clusters []*api.Cluster
	n.MemoryStore().View(func(readTx store.ReadTx) {
		clusters, err = store.FindClusters(readTx, store.ByName(store.DefaultClusterName))
	})

	if err != nil {
		return nil, err
	}

	if len(clusters) == 1 {
		cluster = clusters[0]
	} else {
		select {
		case e := <-watch:
			cluster = e.(api.EventCreateCluster).Cluster
		case <-ctx.Done():
			return nil, ctx.Err()
		}
	}

	return cluster, nil
}
209723
func (a *attacherDefaults) WaitForAttach(spec *volume.Spec, devicePath string, timeout time.Duration) (string, error) {
	klog.Warning(logPrefix(a.plugin.flexVolumePlugin), "using default WaitForAttach for volume ", spec.Name(), ", device ", devicePath)
	return devicePath, nil
}
966408
func (c *cmd) waitForJob() int {
	// Although the session destroy is already deferred, we do it again here,
	// because invalidation of the session before destroyData() ensures there is
	// no race condition allowing an agent to upload data (the acquire will fail).
	defer c.destroySession()
	start := time.Now()
	ackCh := make(chan rExecAck, 128)
	heartCh := make(chan rExecHeart, 128)
	outputCh := make(chan rExecOutput, 128)
	exitCh := make(chan rExecExit, 128)
	doneCh := make(chan struct{})
	errCh := make(chan struct{}, 1)
	defer close(doneCh)
	go c.streamResults(doneCh, ackCh, heartCh, outputCh, exitCh, errCh)
	target := &TargetedUI{UI: c.UI}

	var ackCount, exitCount, badExit int
OUTER:
	for {
		// Determine wait time. We provide a larger window if we know about
		// nodes which are still working.
		waitIntv := c.conf.wait
		if ackCount > exitCount {
			waitIntv *= 2
		}

		select {
		case e := <-ackCh:
			ackCount++
			if c.conf.verbose {
				target.Target = e.Node
				target.Info("acknowledged")
			}

		case h := <-heartCh:
			if c.conf.verbose {
				target.Target = h.Node
				target.Info("heartbeat received")
			}

		case e := <-outputCh:
			target.Target = e.Node
			target.Output(string(e.Output))

		case e := <-exitCh:
			exitCount++
			target.Target = e.Node
			target.Info(fmt.Sprintf("finished with exit code %d", e.Code))
			if e.Code != 0 {
				badExit++
			}

		case <-time.After(waitIntv):
			c.UI.Info(fmt.Sprintf("%d / %d node(s) completed / acknowledged", exitCount, ackCount))
			if c.conf.verbose {
				c.UI.Info(fmt.Sprintf("Completed in %0.2f seconds",
					float64(time.Since(start))/float64(time.Second)))
			}
			if exitCount < ackCount {
				badExit++
			}
			break OUTER

		case <-errCh:
			return 1

		case <-c.shutdownCh:
			return 1
		}
	}

	if badExit > 0 {
		return 2
	}
	return 0
}
105593
func (client *Client) WaitForAssembly(ctx context.Context, assembly *AssemblyInfo) (*AssemblyInfo, error) {
	for {
		res, err := client.GetAssembly(ctx, assembly.AssemblySSLURL)
		if err != nil {
			return nil, err
		}

		// Abort polling if the assembly has entered an error state
		if res.Error != "" {
			return res, nil
		}

		// The polling is done if the assembly is not uploading or executing anymore.
		if res.Ok != "ASSEMBLY_UPLOADING" && res.Ok != "ASSEMBLY_EXECUTING" {
			return res, nil
		}

		select {
		case <-ctx.Done():
			return nil, ctx.Err()
		case <-time.After(time.Second):
			continue
		}
	}
}
260885
func waitForOperation(timeout int, funcOperation func() (*googlecloud.Operation, error)) error {
	var op *googlecloud.Operation
	var err error

	for i := 0; i < timeout; i++ {
		op, err = funcOperation()
		if err != nil {
			return err
		}

		if op.Status == "DONE" {
			if op.Error != nil {
				return fmt.Errorf("operation error: %v", *op.Error.Errors[0])
			}
			return nil
		}
		time.Sleep(1 * time.Second)
	}

	return fmt.Errorf("operation timeout, operations status: %v", op.Status)
}
319701
func (c *Cluster) WaitContainer(id string) (int, error) {
	node, err := c.getNodeForContainer(id)
	if err != nil {
		return -1, err
	}
	node.setPersistentClient()
	code, err := node.WaitContainer(id)
	return code, wrapError(node, err)
}
873008
func waitForFile(p string, to time.Duration) []error {
	done := time.After(to)
	for {
		select {
		case <-done:
			return []error{
				fmt.Errorf("timed out waiting for %s", p),
			}
		case <-time.After(1):
			if ok, _ := fileExists(p); ok {
				return nil
			}
		}
	}
}
965883
func (s *ClusterHealthService) WaitForStatus(waitForStatus string) *ClusterHealthService {
	s.waitForStatus = waitForStatus
	return s
}
332941
func (s *Server) waitFor(search string) (err error) {
	var line string

	scanner := bufio.NewScanner(s.stdout)
	for scanner.Scan() {
		line = scanner.Text()
		fmt.Fprintf(&s.stdoutBuf, "%s\n", line)
		if strings.Contains(line, search) {
			return nil
		}
	}
	err = scanner.Err()
	if err == nil {
		err = io.EOF
	}
	return err
}
680240
func (d *driver) waitVolumeComplete(
	ctx types.Context, volumeID, action string) error {
	// no volume id inputted
	if volumeID == "" {
		return errMissingVolID
	}

	f := func() (interface{}, error) {
		attached := awsec2.VolumeAttachmentStateAttached
		duration := d.statusDelay
		for i := 1; i <= d.maxAttempts; i++ {
			// update volume
			volumes, err := d.getVolume(ctx, volumeID, "")
			if err != nil {
				return nil, goof.WithFieldE("volumeID",
					volumeID, "error getting volume", err)
			}

			// check retrieved volume
			switch action {
			case waitVolumeCreate:
				if *volumes[0].State == awsec2.VolumeStateAvailable {
					return nil, nil
				}
			case waitVolumeDetach:
				if len(volumes[0].Attachments) == 0 {
					return nil, nil
				}
			case waitVolumeAttach:
				if len(volumes[0].Attachments) == 1 &&
					*volumes[0].Attachments[0].State == attached {
					return nil, nil
				}
			}

			ctx.WithField("action", action).Debug(
				"still waiting for action",
			)
			time.Sleep(time.Duration(duration) * time.Nanosecond)
			duration = int64(2) * duration
		}
		return nil, goof.WithField("maxAttempts", d.maxAttempts,
			"Status attempts exhausted")
	}

	_, ok, err := apiUtils.WaitFor(f, d.statusTimeout)
	if !ok {
		return goof.WithFields(goof.Fields{
			"volumeID":      volumeID,
			"statusTimeout": d.statusTimeout},
			"Timeout occured waiting for storage action")
	}
	if err != nil {
		return goof.WithFieldE("volumeID", volumeID,
			"Error while waiting for storage action to finish", err)
	}
	return nil
}
298846
func waitForPods(k8s config.KubernetesConfig, quiet bool) error {
	// Do not wait for "k8s-app" pods in the case of CNI, as they are managed
	// by a CNI plugin which is usually started after minikube has been brought
	// up. Otherwise, minikube won't start, as "k8s-app" pods are not ready.
	componentsOnly := k8s.NetworkPlugin == "cni"

	if !quiet {
		console.OutStyle("waiting-pods", "Waiting for pods:")
	}
	client, err := util.GetClient()
	if err != nil {
		return errors.Wrap(err, "k8s client")
	}

	for _, p := range PodsByLayer {
		if componentsOnly && p.key != "component" {
			continue
		}

		if !quiet {
			console.Out(" %s", p.name)
		}
		selector := labels.SelectorFromSet(labels.Set(map[string]string{p.key: p.value}))
		if err := util.WaitForPodsWithLabelRunning(client, "kube-system", selector); err != nil {
			return errors.Wrap(err, fmt.Sprintf("waiting for %s=%s", p.key, p.value))
		}
	}
	if !quiet {
		console.OutLn("")
	}
	return nil
}