// Run watches for resource changes annotated with application label on all registered clusters and schedule corresponding app refresh.
968112
func (c *liveStateCache) Run(ctx context.Context) {
	util.RetryUntilSucceed(func() error {
		clusterEventCallback := func(event *db.ClusterEvent) {
			c.lock.Lock()
			defer c.lock.Unlock()
			if cluster, ok := c.clusters[event.Cluster.Server]; ok {
				if event.Type == watch.Deleted {
					cluster.invalidate()
					delete(c.clusters, event.Cluster.Server)
				} else if event.Type == watch.Modified {
					cluster.cluster = event.Cluster
					cluster.invalidate()
				}
			} else if event.Type == watch.Added && isClusterHasApps(c.appInformer.GetStore().List(), event.Cluster) {
				go func() {
					// warm up cache for cluster with apps
					_, _ = c.getSyncedCluster(event.Cluster.Server)
				}()
			}
		}

		return c.db.WatchClusters(ctx, clusterEventCallback)

	}, "watch clusters", ctx, clusterRetryTimeout)

	<-ctx.Done()
}
1305136
func (s *Scheduler) Run() {
	s.mu.Lock()
	s.state = stateRunning
	s.mu.Unlock()

	s.startProcessRequestsLoop()
}
947337
func (c *applicationVersionSetCommand) Run(ctx *cmd.Context) error {
	return c.ctx.SetUnitWorkloadVersion(c.version)
}
1328169
func (c *TriggerController) Run(workers int, stopCh <-chan struct{}) {
	defer utilruntime.HandleCrash()
	defer c.queue.ShutDown()

	klog.Infof("Starting trigger controller")

	if !cache.WaitForCacheSync(stopCh, c.syncs...) {
		utilruntime.HandleError(fmt.Errorf("timed out waiting for caches to sync"))
		return
	}

	for i := 0; i < workers; i++ {
		go wait.Until(c.imageStreamWorker, time.Second, stopCh)
	}

	for i := 0; i < workers; i++ {
		go wait.Until(c.resourceWorker, time.Second, stopCh)
	}

	<-stopCh
	klog.Infof("Shutting down trigger controller")
}
133040
func (jm *Controller) Run(stopCh <-chan struct{}) {
	defer utilruntime.HandleCrash()
	klog.Infof("Starting CronJob Manager")
	// Check things every 10 second.
	go wait.Until(jm.syncAll, 10*time.Second, stopCh)
	<-stopCh
	klog.Infof("Shutting down CronJob Manager")
}
12264
func (c *CloudNodeLifecycleController) Run(stopCh <-chan struct{}) {
	defer utilruntime.HandleCrash()

	// The following loops run communicate with the APIServer with a worst case complexity
	// of O(num_nodes) per cycle. These functions are justified here because these events fire
	// very infrequently. DO NOT MODIFY this to perform frequent operations.

	// Start a loop to periodically check if any nodes have been
	// deleted or shutdown from the cloudprovider
	wait.Until(c.MonitorNodes, c.nodeMonitorPeriod, stopCh)
}
666394
func (x *xdsBalancer) run() {
	for {
		select {
		case update := <-x.grpcUpdate:
			x.handleGRPCUpdate(update)
		case update := <-x.xdsClientUpdate:
			x.handleXDSClientUpdate(update)
		case <-x.timer.C: // x.timer.C will block if we are not in fallback monitoring stage.
			x.switchFallback()
		case <-x.noSubConnAlert: // x.noSubConnAlert will block if we are not in fallback monitoring stage.
			x.switchFallback()
		case <-x.ctx.Done():
			if x.client != nil {
				x.client.close()
			}
			if x.xdsLB != nil {
				x.xdsLB.Close()
			}
			if x.fallbackLB != nil {
				x.fallbackLB.Close()
			}
			return
		}
	}
}
393355
func (s *Server) Run() {
	go s.updater()
	go s.notifier()
	go s.manager()

	<-s.quit
}
91726
func (r *Reflector) Run(stopCh <-chan struct{}) {
	klog.V(3).Infof("Starting reflector %v (%s) from %s", r.expectedType, r.resyncPeriod, r.name)
	wait.Until(func() {
		if err := r.ListAndWatch(stopCh); err != nil {
			utilruntime.HandleError(err)
		}
	}, r.period, stopCh)
}
496263
func (ec *ServiceManager) Run() (err error) {

	// Launch services
	for _, s := range ec.services {
		ec.wg.Add(1)
		go ec.execute(s)
	}

	// Start event loop
	go ec.eventLoop()

	return nil
}
239945
func (m *Monitor) Run(looper director.Looper) {
	looper.Loop(func() error {
		log.Debugf("Running checks")

		var wg sync.WaitGroup

		// Make immutable copy of m.Checks (checks are still mutable)
		m.RLock()
		checks := make(map[string]*Check, len(m.Checks))
		for k, v := range m.Checks {
			checks[k] = v
		}
		m.RUnlock()

		wg.Add(len(checks))
		for _, check := range checks {
			// Run all checks in parallel in goroutines
			resultChan := make(chan checkResult, 1)

			go func(check *Check, resultChan chan checkResult) {
				result, err := check.Command.Run(check.Args)
				resultChan <- checkResult{result, err}
			}(check, resultChan) // copy check pointer for the goroutine

			go func(check *Check, resultChan chan checkResult) {
				defer wg.Done()

				// We make the call but we time out if it gets too close to the
				// m.CheckInterval.
				select {
				case result := <-resultChan:
					check.UpdateStatus(result.status, result.err)
				case <-time.After(m.CheckInterval - 1*time.Millisecond):
					log.Errorf("Error, check %s timed out! (%v)", check.ID, check.Args)
					check.UpdateStatus(UNKNOWN, errors.New("Timed out!"))
				}
			}(check, resultChan) // copy check pointer for the goroutine
		}

		// Let's make sure we don't continue to spool up
		// huge quantities of goroutines. Wait on all of them
		// to complete before moving on. This could slow down
		// our check loop if something doesn't time out properly.
		wg.Wait()

		return nil
	})
}
835857
func (w *checkWatcher) Run(ctx context.Context) {
	defer close(w.done)

	// map of check IDs to their metadata
	checks := map[string]*checkRestart{}

	// timer for check polling
	checkTimer := time.NewTimer(0)
	defer checkTimer.Stop() // ensure timer is never leaked

	stopTimer := func() {
		checkTimer.Stop()
		select {
		case <-checkTimer.C:
		default:
		}
	}

	// disable by default
	stopTimer()

	// Main watch loop
	for {
		// disable polling if there are no checks
		if len(checks) == 0 {
			stopTimer()
		}

		select {
		case update := <-w.checkUpdateCh:
			if update.remove {
				// Remove a check
				delete(checks, update.checkID)
				continue
			}

			// Add/update a check
			checks[update.checkID] = update.checkRestart
			w.logger.Debug("watching check", "alloc_id", update.checkRestart.allocID,
				"task", update.checkRestart.taskName, "check", update.checkRestart.checkName)

			// if first check was added make sure polling is enabled
			if len(checks) == 1 {
				stopTimer()
				checkTimer.Reset(w.pollFreq)
			}

		case <-ctx.Done():
			return

		case <-checkTimer.C:
			checkTimer.Reset(w.pollFreq)

			// Set "now" as the point in time the following check results represent
			now := time.Now()

			results, err := w.consul.Checks()
			if err != nil {
				if !w.lastErr {
					w.lastErr = true
					w.logger.Error("failed retrieving health checks", "error", err)
				}
				continue
			}

			w.lastErr = false

			// Keep track of tasks restarted this period so they
			// are only restarted once and all of their checks are
			// removed.
			restartedTasks := map[string]struct{}{}

			// Loop over watched checks and update their status from results
			for cid, check := range checks {
				// Shortcircuit if told to exit
				if ctx.Err() != nil {
					return
				}

				if _, ok := restartedTasks[check.taskKey]; ok {
					// Check for this task already restarted; remove and skip check
					delete(checks, cid)
					continue
				}

				result, ok := results[cid]
				if !ok {
					// Only warn if outside grace period to avoid races with check registration
					if now.After(check.graceUntil) {
						w.logger.Warn("watched check not found in Consul", "check", check.checkName, "check_id", cid)
					}
					continue
				}

				restarted := check.apply(ctx, now, result.Status)
				if restarted {
					// Checks are registered+watched on
					// startup, so it's safe to remove them
					// whenever they're restarted
					delete(checks, cid)

					restartedTasks[check.taskKey] = struct{}{}
				}
			}

			// Ensure even passing checks for restartedTasks are removed
			if len(restartedTasks) > 0 {
				for cid, check := range checks {
					if _, ok := restartedTasks[check.taskKey]; ok {
						delete(checks, cid)
					}
				}
			}
		}
	}
}
633862
func (sdn *OpenShiftSDN) Run(c *cobra.Command, errout io.Writer, stopCh chan struct{}) {
	err := injectKubeAPIEnv(sdn.URLOnlyKubeConfigFilePath)
	if err != nil {
		klog.Fatal(err)
	}

	// Parse config file, build config objects
	err = sdn.ValidateAndParse()
	if err != nil {
		if kerrors.IsInvalid(err) {
			if details := err.(*kerrors.StatusError).ErrStatus.Details; details != nil {
				fmt.Fprintf(errout, "Invalid %s %s\n", details.Kind, details.Name)
				for _, cause := range details.Causes {
					fmt.Fprintf(errout, "  %s: %s\n", cause.Field, cause.Message)
				}
				os.Exit(255)
			}
		}
		klog.Fatal(err)
	}

	// Set up a watch on our config file; if it changes, we should exit -
	// (we don't have the ability to dynamically reload config changes).
	if err := watchForChanges(sdn.ConfigFilePath, stopCh); err != nil {
		klog.Fatalf("unable to setup configuration watch: %v", err)
	}

	// Build underlying network objects
	err = sdn.Init()
	if err != nil {
		klog.Fatalf("Failed to initialize sdn: %v", err)
	}

	err = sdn.Start(stopCh)
	if err != nil {
		klog.Fatalf("Failed to start sdn: %v", err)
	}

	<-stopCh
	time.Sleep(500 * time.Millisecond) // gracefully shut down
	os.Exit(1)
}
631665
func (ssc *StatefulSetController) Run(workers int, stopCh <-chan struct{}) {
	defer utilruntime.HandleCrash()
	defer ssc.queue.ShutDown()

	klog.Infof("Starting stateful set controller")
	defer klog.Infof("Shutting down statefulset controller")

	if !controller.WaitForCacheSync("stateful set", stopCh, ssc.podListerSynced, ssc.setListerSynced, ssc.pvcListerSynced, ssc.revListerSynced) {
		return
	}

	for i := 0; i < workers; i++ {
		go wait.Until(ssc.worker, time.Second, stopCh)
	}

	<-stopCh
}
605744
func (c *ClusterQuotaReconcilationController) Run(workers int, stopCh <-chan struct{}) {
	defer utilruntime.HandleCrash()

	klog.Infof("Starting the cluster quota reconciliation controller")

	// the controllers that replenish other resources to respond rapidly to state changes
	go c.quotaMonitor.Run(stopCh)

	if !controller.WaitForCacheSync("cluster resource quota", stopCh, c.informerSyncedFuncs...) {
		return
	}

	// the workers that chug through the quota calculation backlog
	for i := 0; i < workers; i++ {
		go wait.Until(c.worker, time.Second, stopCh)
	}

	// the timer for how often we do a full recalculation across all quotas
	go wait.Until(func() { c.calculateAll() }, c.resyncPeriod, stopCh)

	<-stopCh
	klog.Infof("Shutting down ClusterQuotaReconcilationController")
	c.queue.ShutDown()
}
1317970
func (c *ServiceController) Run() {
	glog.Infof("starting service controller")

	stopCh := c.StopChannel()
	go c.runWatcher(stopCh)

	<-stopCh
	glog.Infof("shutting down service controller")
}
1211552
func (app *Application) Run(addr string) {
	err := http.ListenAndServe(addr, app)
	if err != nil {
		panic(err)
	}
}