go func() {
			defer waitGroup.Done()
			wait.Until(c.runWorker, time.Second, ctx.Done())
		}()
	}

	klog.Infof("Started %v workers for %v", c.options.Workers, c.options.Name)

	// Wait for context cancel.
	<-ctx.Done()

	// Forcefully shutdown workqueue. Drop any enqueued items.
	c.queue.ShutDown()

	// Workqueue shutdown signals for workers to stop. Wait for all workers to
	// clean up
	waitGroup.Wait()

	}
}

func retryWithExponentialBackOff(initialDuration time.Duration, fn wait.ConditionFunc) error {
	backoff := wait.Backoff{
		Duration: initialDuration,
		Factor:   3,
		Jitter:   0,
		Steps:    6,
	}
	return wait.ExponentialBackoff(backoff, fn)
}

func TestPluginRegistration(t *testing.T) {
	defer cleanup(t)

				w := xioutil.NewDeadlineWorker(globalDriveConfig.GetMaxTimeout())
				return w.Run(func() error {
					wait := deleteCleanupSleeper.Timer(ctx)
					removeAll(pathJoin(drivePath, minioMetaTmpDeletedBucket, ddir))
					wait()
					return nil
				})
			})
		}(disk)
	}
	wg.Wait()
}

// nsScanner will start scanning buckets and send updated totals as they are traversed.

		t.Fatalf("error starting test binary: %v", err)
	}

	pid := cmd.Process.Pid

	err = cmd.Wait()
	t.Logf("child output:\n%s", output.String())
	if err == nil {
		t.Fatalf("Wait succeeded, want SIGABRT")
	}
	ee, ok := err.(*exec.ExitError)
	if !ok {
		t.Fatalf("Wait err got %T %v, want exec.ExitError", ee, ee)
	}
	ws, ok := ee.Sys().(syscall.WaitStatus)
	if !ok {

	if client.Discovery().RESTClient() == nil {
		fmt.Printf("[upgrade/health] Would create the Job with the prefix %q in namespace %q and wait until it completes\n", prefix, ns)
		return nil
	}

	// Check if there is at least one Node where a Job's Pod can schedule. If not, skip this preflight check.
	err = wait.PollUntilContextTimeout(ctx, time.Second*1, timeout, true, func(_ context.Context) (bool, error) {

	epsHandler := NewEndpointSliceHandlerMock()
	epsConfig.RegisterEventHandler(epsHandler)

	stopCh := make(chan struct{})
	defer close(stopCh)
	sharedInformers.Start(stopCh)

	err := wait.PollImmediate(time.Millisecond*10, wait.ForeverTestTimeout, func() (bool, error) {
		svcHandler.lock.Lock()
		defer svcHandler.lock.Unlock()
		if reflect.DeepEqual(svcHandler.state, expectedSvcState) {
			return true, nil
		}

limitations under the License.
*/

package cacher

import (
	"context"
	"sync"
	"time"

	"google.golang.org/grpc/metadata"

	utilruntime "k8s.io/apimachinery/pkg/util/runtime"
	"k8s.io/apimachinery/pkg/util/wait"

	"k8s.io/klog/v2"
	"k8s.io/utils/clock"
)

const (
	// progressRequestPeriod determines period of requesting progress
	// from etcd when there is a request waiting for watch cache to be fresh.

    // TODO(user): Consider scaling by number of mutator threads,
    // e.g. using Thread#activeCount()
    return Math.max(10L, Runtime.getRuntime().totalMemory() / (32L * 1024L * 1024L));
  }

  /**
   * Waits until the given future {@linkplain Future#isDone is done}, invoking the garbage collector
   * as necessary to try to ensure that this will happen.
   *
   * @throws RuntimeException if timed out or interrupted while waiting

			default:
				cacher.Lock()
				cacher.bookmarkWatchers.popExpiredWatchersThreadUnsafe()
				cacher.Unlock()
			}
		}
	}()

	// wait for adding/removing watchers to end
	wg.Wait()

	if watchErr != nil {
		t.Fatal(watchErr)
	}

	// wait out the expiration period and pop expired watchers
	time.Sleep(2 * time.Second)
	cacher.Lock()
	defer cacher.Unlock()

		i++
		return false, nil
	})
	if err != ErrWaitTimeout || i != 0 {
		t.Errorf("unexpected error: %v", err)
	}

	opts := Backoff{Factor: 1.0, Steps: 3}

	// waits up to steps
	i = 0
	err = ExponentialBackoff(opts, func() (bool, error) {
		i++
		return false, nil
	})
	if err != ErrWaitTimeout || i != opts.Steps {
		t.Errorf("unexpected error: %v", err)
	}