t.Errorf("failed to create iptables canaries: %v", err)
	}

	// If we close the stop channel, it should stop running
	close(stopCh)

	if err := waitForNoReload(&reloads, 2); err != nil {
		t.Errorf("got unexpected number of reloads after stop: %v", err)
	}
	if !ensureNoChains(mfe) {
		t.Errorf("canaries still exist after stopping monitor")
	}

		}
	})
}

// StartMaxInFlightWatermarkMaintenance starts the goroutines to observe and maintain watermarks for max-in-flight
// requests.
func StartMaxInFlightWatermarkMaintenance(stopCh <-chan struct{}) {
	startWatermarkMaintenance(watermark, stopCh)

	serviceCIDRsSynced  cache.InformerSynced

	interval time.Duration
}

// Start will not return until the default ServiceCIDR exists or stopCh is closed.
func (c *Controller) Start(ctx context.Context) {
	defer utilruntime.HandleCrash()
	stopCh := ctx.Done()

	c.eventBroadcaster = record.NewBroadcaster(record.WithContext(ctx))

	LoopbackClientConfig *restclient.Config
	// StopCh is the channel that will be closed when the server stops.
	//
	// Deprecated: use the PostStartHookContext itself instead, it contains a context that
	// gets cancelled when the server stops. StopCh keeps getting provided for existing code.
	StopCh <-chan struct{}
	// Context gets cancelled when the server stops.
	context.Context
}

func (pw *ProxyWatcher) StopChan() <-chan struct{} {
	return pw.stopCh
}

// MockWatcher implements watch.Interface with mockable functions.
type MockWatcher struct {
	StopFunc       func()
	ResultChanFunc func() <-chan Event
}

var _ Interface = &MockWatcher{}

// Stop calls StopFunc
func (mw MockWatcher) Stop() {
	mw.StopFunc()
}

// ResultChan calls ResultChanFunc

		// Set the request delay so the manager timeouts and collects the node addresses later
		RequestDelay: 200 * time.Millisecond,
	}
	stopCh := make(chan struct{})
	defer close(stopCh)

	manager := NewSyncManager(cloud, "defaultNode", syncPeriod).(*cloudResourceSyncManager)
	go manager.Run(stopCh)

	nodeAddresses, err := manager.NodeAddresses()
	if err != nil {
		t.Errorf("Unexpected err: %q\n", err)
	}

	c.services.AddEventHandler(controllers.EventHandler[controllers.Object]{AddFunc: c.queue.AddObject})

	return c
}

func (c *autoServiceExportController) Run(stopCh <-chan struct{}) {
	kube.WaitForCacheSync("auto service export", stopCh, c.services.HasSynced)
	c.queue.Run(stopCh)
	c.services.ShutdownHandlers()
}

func (c *autoServiceExportController) logPrefix() string {
	return "AutoServiceExport (cluster=" + c.ClusterID.String() + ") "

func (d *CacheDebugger) ListenForSignal(ctx context.Context) {
	logger := klog.FromContext(ctx)
	stopCh := ctx.Done()
	ch := make(chan os.Signal, 1)
	signal.Notify(ch, compareSignal)

	go func() {
		for {
			select {
			case <-stopCh:
				return
			case <-ch:
				d.Comparer.Compare(logger)
				d.Dumper.DumpAll(logger)
			}
		}
	}()

	handlers            map[string]cache.PluginHandler
	sync.RWMutex
}

var _ Reconciler = &reconciler{}

func (rc *reconciler) Run(stopCh <-chan struct{}) {
	wait.Until(func() {
		rc.reconcile()
	},
		rc.loopSleepDuration,
		stopCh)
}

func (rc *reconciler) AddHandler(pluginType string, pluginHandler cache.PluginHandler) {
	rc.Lock()
	defer rc.Unlock()

// limitations under the License.

package concurrent

import (
	"time"

	"istio.io/istio/pkg/util/sets"
)

type Debouncer[T comparable] struct{}

func (d *Debouncer[T]) Run(ch chan T, stopCh <-chan struct{}, debounceMinInterval, debounceMaxInterval time.Duration, pushFn func(sets.Set[T])) {
	var timeChan <-chan time.Time
	var startDebounce time.Time
	var lastConfigUpdateTime time.Time

	pushCounter := 0