// Start starts syncing the RuntimeClass cache with the apiserver.
func (m *Manager) Start(stopCh <-chan struct{}) {
	m.informerFactory.Start(stopCh)
}

// WaitForCacheSync exposes the WaitForCacheSync method on the informer factory for testing
// purposes.
func (m *Manager) WaitForCacheSync(stopCh <-chan struct{}) {
	m.informerFactory.WaitForCacheSync(stopCh)
}

	fakePodControl := controller.FakePodControl{}
	stopCh := make(chan struct{})
	defer close(stopCh)
	manager, informers := testNewReplicaSetControllerFromClient(t, client, stopCh, BurstReplicas)

	// 2 running pods, a controller with 2 replicas, sync is a no-op
	labelMap := map[string]string{"foo": "bar"}

		handler,
	)
	controller.csrsSynced = func() bool { return true }

	stopCh := make(chan struct{})
	defer close(stopCh)
	informerFactory.Start(stopCh)
	informerFactory.WaitForCacheSync(stopCh)
	wait.PollUntil(10*time.Millisecond, func() (bool, error) {
		return controller.queue.Len() >= 1, nil
	}, stopCh)
	controller.processNextWorkItem(ctx)

	actions := client.Actions()
	if len(actions) != 1 {

}

func (lb *tcpLB) handleConnection(in net.Conn, stopCh chan struct{}) {
	out, err := net.Dial("tcp", lb.serverURL)
	if err != nil {
		lb.t.Log(err)
		return
	}
	go io.Copy(out, in)
	go io.Copy(in, out)
	<-stopCh
	if err := out.Close(); err != nil {
		lb.t.Fatalf("failed to close connection: %v", err)
	}
}

func (lb *tcpLB) serve(stopCh chan struct{}) {
	conn, err := lb.ln.Accept()

	out, err := m.readyWorkloads()
	m.mutex.Unlock()
	return out, err
}

func (m *workloadManager) Start() error {
	// Run the pod controller.
	go m.podController.Run(m.stopCh)

	// Wait for the cache to sync.
	if !m.podController.WaitForSync(m.stopCh) {
		return fmt.Errorf(
			"failed syncing cache for echo %s/%s: controller stopping",
			m.cfg.Namespace.Name(),
			m.cfg.Service)
	}

	// stopCh is closed, it is supposed to stop them. Run will be called before the first call to ProcessEvents.
	Run(stopCh <-chan struct{}) error

	// Shutdown will synchronously shut down the backend while making sure that all pending
	// events are delivered. It can be assumed that this method is called after
	// the stopCh channel passed to the Run method has been closed.
	Shutdown()

const (
	maxConcurrencyPerForward = 20
)

type executor struct {
	totalRequests  *atomic.Uint64
	activeRequests *atomic.Uint64
	stopCh         chan struct{}
}

func newExecutor() *executor {
	e := &executor{
		totalRequests:  atomic.NewUint64(0),
		activeRequests: atomic.NewUint64(0),
		stopCh:         make(chan struct{}),
	}

	return e
}

func (e *executor) ActiveRequests() uint64 {
	return e.activeRequests.Load()

	c.restMapper = mapper
}

func (c *Plugin[H]) SetDynamicClient(client dynamic.Interface) {
	c.dynamicClient = client
}

func (c *Plugin[H]) SetDrainedNotification(stopCh <-chan struct{}) {
	c.stopCh = stopCh
}

func (c *Plugin[H]) SetAuthorizer(authorizer authorizer.Authorizer) {
	c.authorizer = authorizer
}

func (c *Plugin[H]) SetMatcher(matcher *matching.Matcher) {
	c.matcher = matcher

	// lock is to ensure the access and modify of lastAccessTime, stopped, and immutable are thread safety,
	// and protecting from closing stopCh multiple times.
	lock           sync.Mutex
	lastAccessTime time.Time
	stopped        bool
	immutable      bool
	stopCh         chan struct{}
}

func (i *objectCacheItem) stop() bool {
	i.lock.Lock()
	defer i.lock.Unlock()
	return i.stopThreadUnsafe()
}

		wg.Wait()
		c <- struct{}{}
	}()
	select {
	case <-c:
		return true
	case <-time.After(timeout):
		return false
	}
}

func TestSendPushesManyPushes(t *testing.T) {
	stopCh := make(chan struct{})
	defer close(stopCh)

	semaphore := make(chan struct{}, 2)
	queue := NewPushQueue()
	defer queue.ShutDown()

	proxies := createProxies(5)

	pushes := make(map[string]int)
	pushesMu := &sync.Mutex{}