// Run the main goroutine responsible for watching and syncing jobs.
768573
func (cc *CertificateController) Run(workers int, stopCh <-chan struct{}) {
	defer utilruntime.HandleCrash()
	defer cc.queue.ShutDown()

	klog.Infof("Starting certificate controller")
	defer klog.Infof("Shutting down certificate controller")

	if !controller.WaitForCacheSync("certificate", stopCh, cc.csrsSynced) {
		return
	}

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

	<-stopCh
}
246174
func (m *Manager) Run() {
	m.lock.Lock()
	if m.runState != managerStateIdle {
		m.lock.Unlock()
		return
	}

	// Set the state to running
	m.runState = managerStateRunning
	m.lock.Unlock()

	// Start a goroutine that just waits for a stop request
	stopCh := make(chan struct{})
	go func() {
		defer close(stopCh)
		m.lock.Lock()
		defer m.lock.Unlock()

		// We wait for anything not running, just so we're more resilient
		// in the face of state machine issues. Basically any state change
		// will cause us to quit.
		for m.runState == managerStateRunning {
			m.cond.Wait()
		}
	}()

	// When we exit, we set the state to stopped and broadcast to any
	// waiting Close functions that they can return.
	defer func() {
		m.lock.Lock()
		m.runState = managerStateStopped
		m.cond.Broadcast()
		m.lock.Unlock()
	}()

	// Register for proxy catalog change notifications
	notifyCh := make(chan struct{}, 1)
	m.State.NotifyProxy(notifyCh)
	defer m.State.StopNotifyProxy(notifyCh)

	// Start the timer for snapshots. We don't use a ticker because disk
	// IO can be slow and we don't want overlapping notifications. So we only
	// reset the timer once the snapshot is complete rather than continuously.
	snapshotTimer := time.NewTimer(m.SnapshotPeriod)
	defer snapshotTimer.Stop()

	m.Logger.Println("[DEBUG] agent/proxy: managed Connect proxy manager started")
SYNC:
	for {
		// Sync first, before waiting on further notifications so that
		// we can start with a known-current state.
		m.sync()

		// Note for these variables we don't use a time.Timer because both
		// periods are relatively short anyways so they end up being eligible
		// for GC very quickly, so overhead is not a concern.
		var quiescent, quantum <-chan time.Time

		// Start a loop waiting for events from the local state store. This
		// loops rather than just `select` so we can coalesce many state
		// updates over a period of time.
		for {
			select {
			case <-notifyCh:
				// If this is our first notification since the last sync,
				// reset the quantum timer which is the max time we'll wait.
				if quantum == nil {
					quantum = time.After(m.CoalescePeriod)
				}

				// Always reset the quiescent timer
				quiescent = time.After(m.QuiescentPeriod)

			case <-quantum:
				continue SYNC

			case <-quiescent:
				continue SYNC

			case <-snapshotTimer.C:
				// Perform a snapshot
				if path := m.SnapshotPath(); path != "" {
					if err := m.snapshot(path, true); err != nil {
						m.Logger.Printf("[WARN] agent/proxy: failed to snapshot state: %s", err)
					}
				}

				// Reset
				snapshotTimer.Reset(m.SnapshotPeriod)

			case <-stopCh:
				// Stop immediately, no cleanup
				m.Logger.Println("[DEBUG] agent/proxy: Stopping managed Connect proxy manager")
				return
			}
		}
	}
}
1184939
func (ic *IngressIPController) Run(stopCh <-chan struct{}) {
	defer utilruntime.HandleCrash()
	defer ic.queue.ShutDown()

	klog.V(5).Infof("Waiting for the initial sync to be completed")
	if !cache.WaitForCacheSync(stopCh, ic.hasSynced) {
		return
	}

	if !ic.processInitialSync() {
		return
	}

	klog.V(5).Infof("Initial sync completed, starting worker")
	for ic.work() {
		var done bool
		select {
		case _, ok := <-stopCh:
			done = !ok
		default:
		}
		if done {
			break
		}
	}

	klog.V(1).Infof("Shutting down ingress ip controller")
}
918956
func (t *ThrottledExecutor) Run(ctx context.Context, job func()) {
	t.wg.Add(1)
	go func() {
		defer t.wg.Done()
		select {
		case t.limiter <- struct{}{}:
			defer func() {
				<-t.limiter
			}()
			job()
		case <-ctx.Done():
			return
		}
	}()
}
706543
func (rsc *ReplicaSetController) Run(workers int, stopCh <-chan struct{}) {
	defer utilruntime.HandleCrash()
	defer rsc.queue.ShutDown()

	controllerName := strings.ToLower(rsc.Kind)
	klog.Infof("Starting %v controller", controllerName)
	defer klog.Infof("Shutting down %v controller", controllerName)

	if !controller.WaitForCacheSync(rsc.Kind, stopCh, rsc.podListerSynced, rsc.rsListerSynced) {
		return
	}

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

	<-stopCh
}
942981
func (c *ImageStreamController) Run(workers int, stopCh <-chan struct{}) {
	defer utilruntime.HandleCrash()
	defer c.queue.ShutDown()

	klog.Infof("Starting image stream controller")

	// Wait for the stream store to sync before starting any work in this controller.
	if !cache.WaitForCacheSync(stopCh, c.listerSynced) {
		utilruntime.HandleError(fmt.Errorf("timed out waiting for caches to sync"))
		return
	}

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

	metrics.InitializeImportCollector(false, c.importCounter.Collect)

	<-stopCh
	klog.Infof("Shutting down image stream controller")
}
560379
func Run(jobs []Tasker) (err error) {
	// []T does not convert to []Tasker implicitly even is T implements
	// Tasker. We need to iterate on []Tasker making an explicit cast.
	// http://golang.org/doc/faq#convert_slice_of_interface
	prematureEnd := make(chan struct{})
	jobsQueue := make(chan Tasker, workersNumber)
	done := make(chan struct{}, workersNumber)
	var totalDone int
	go populateQueue(jobsQueue, jobs, prematureEnd)
	go parallelizeWorkers(jobsQueue, done)
	// TODO add a case timeout that returns error.
	for {
		select {
		case <-done:
			totalDone++
		case <-prematureEnd:
			err = ErrTasksNotCompleted
		}
		if totalDone == workersNumber {
			// We can assume that jobsQueue is closed and
			// that no goroutine is operating on []Tasker.
			break
		}
	}
	return
}
642671
func (s *Scheduler) Run(ctx context.Context) error {
	defer close(s.doneChan)

	updates, cancel, err := store.ViewAndWatch(s.store, s.setupTasksList)
	if err != nil {
		log.G(ctx).WithError(err).Errorf("snapshot store update failed")
		return err
	}
	defer cancel()

	// Validate resource for tasks from preassigned tasks
	// do this before other tasks because preassigned tasks like
	// global service should start before other tasks
	s.processPreassignedTasks(ctx)

	// Queue all unassigned tasks before processing changes.
	s.tick(ctx)

	const (
		// commitDebounceGap is the amount of time to wait between
		// commit events to debounce them.
		commitDebounceGap = 50 * time.Millisecond
		// maxLatency is a time limit on the debouncing.
		maxLatency = time.Second
	)
	var (
		debouncingStarted     time.Time
		commitDebounceTimer   *time.Timer
		commitDebounceTimeout <-chan time.Time
	)

	tickRequired := false

	schedule := func() {
		if len(s.pendingPreassignedTasks) > 0 {
			s.processPreassignedTasks(ctx)
		}
		if tickRequired {
			s.tick(ctx)
			tickRequired = false
		}
	}

	// Watch for changes.
	for {
		select {
		case event := <-updates:
			switch v := event.(type) {
			case api.EventCreateTask:
				if s.createTask(ctx, v.Task) {
					tickRequired = true
				}
			case api.EventUpdateTask:
				if s.updateTask(ctx, v.Task) {
					tickRequired = true
				}
			case api.EventDeleteTask:
				if s.deleteTask(v.Task) {
					// deleting tasks may free up node resource, pending tasks should be re-evaluated.
					tickRequired = true
				}
			case api.EventCreateNode:
				s.createOrUpdateNode(v.Node)
				tickRequired = true
			case api.EventUpdateNode:
				s.createOrUpdateNode(v.Node)
				tickRequired = true
			case api.EventDeleteNode:
				s.nodeSet.remove(v.Node.ID)
			case state.EventCommit:
				if commitDebounceTimer != nil {
					if time.Since(debouncingStarted) > maxLatency {
						commitDebounceTimer.Stop()
						commitDebounceTimer = nil
						commitDebounceTimeout = nil
						schedule()
					} else {
						commitDebounceTimer.Reset(commitDebounceGap)
					}
				} else {
					commitDebounceTimer = time.NewTimer(commitDebounceGap)
					commitDebounceTimeout = commitDebounceTimer.C
					debouncingStarted = time.Now()
				}
			}
		case <-commitDebounceTimeout:
			schedule()
			commitDebounceTimer = nil
			commitDebounceTimeout = nil
		case <-s.stopChan:
			return nil
		}
	}
}
837188
func (sdw *SplitDiffWorker) Run(ctx context.Context) error {
	resetVars()
	err := sdw.run(ctx)

	sdw.SetState(WorkerStateCleanUp)
	cerr := sdw.cleaner.CleanUp(sdw.wr)
	if cerr != nil {
		if err != nil {
			sdw.wr.Logger().Errorf2(cerr, "CleanUp failed in addition to job error")
		} else {
			err = cerr
		}
	}
	if err != nil {
		sdw.wr.Logger().Errorf2(err, "Run() error")
		sdw.SetState(WorkerStateError)
		return err
	}
	sdw.SetState(WorkerStateDone)
	return nil
}
44638
func (s *Supervisor) run() {
	defer s.stopWg.Done()
	for {
		select {
		case <-s.watcherErrorC:
			s.watcherWg.Wait()
			s.watcherErrorC = nil
		case <-s.stopC:
			close(s.watcherCancelC)
			s.engine.Close()
			s.watcherWg.Wait()
			if s.proxy != nil {
				s.proxy.Stop(true)
			}
			return
		}

		err := s.init()
		// In case of an error keep trying to initialize making pauses between
		// attempts.
		for err != nil {
			log.Errorf("sup failed to reinit, err=%v", err)
			select {
			case <-time.After(retryPeriod):
			case <-s.stopC:
				return
			}
			err = s.init()
		}
	}
}
911124
func (e *DockerRegistryServiceController) Run(workers int, stopCh <-chan struct{}) {
	defer utilruntime.HandleCrash()
	defer e.registryLocationQueue.ShutDown()

	klog.Infof("Starting DockerRegistryServiceController controller")
	defer klog.Infof("Shutting down DockerRegistryServiceController controller")

	// Wait for the store to sync before starting any work in this controller.
	ready := make(chan struct{})
	go e.waitForDockerURLs(ready, stopCh)
	select {
	case <-ready:
	case <-stopCh:
		return
	}
	klog.V(1).Infof("caches synced")

	go wait.Until(e.watchForDockerURLChanges, time.Second, stopCh)
	for i := 0; i < workers; i++ {
		go wait.Until(e.watchForDockercfgSecretUpdates, time.Second, stopCh)
	}
	<-stopCh
}
646371
func (k8s *Kubernetes) Run(qorJob worker.QorJobInterface) error {
	job := qorJob.GetJob()

	if job.Handler != nil {
		return job.Handler(qorJob.GetSerializableArgument(qorJob), qorJob)
	}

	return errors.New("no handler found for job " + job.Name)
}
1346561
func (ar *allocRunner) Run() {
	// Close the wait channel on return
	defer close(ar.waitCh)

	// Start the task state update handler
	go ar.handleTaskStateUpdates()

	// Start the alloc update handler
	go ar.handleAllocUpdates()

	// If task update chan has been closed, that means we've been shutdown.
	select {
	case <-ar.taskStateUpdateHandlerCh:
		return
	default:
	}

	// Run the prestart hooks if non-terminal
	if ar.shouldRun() {
		if err := ar.prerun(); err != nil {
			ar.logger.Error("prerun failed", "error", err)
			goto POST
		}
	}

	// Run the runners (blocks until they exit)
	ar.runTasks()

POST:
	// Run the postrun hooks
	if err := ar.postrun(); err != nil {
		ar.logger.Error("postrun failed", "error", err)
	}

}
95643
func (j *resizeJob) run() error {
	j.Logger.Printf("run resizeJob")
	// Set job state to RUNNING.
	j.setState(resizeJobStateRunning)

	// Job can be considered done in the case where it doesn't require any action.
	if !j.nodesArePending() {
		j.Logger.Printf("resizeJob contains no pending tasks; mark as done")
		j.result <- resizeJobStateDone
		return nil
	}

	j.Logger.Printf("distribute tasks for resizeJob")
	err := j.distributeResizeInstructions()
	if err != nil {
		j.result <- resizeJobStateAborted
		return errors.Wrap(err, "distributing instructions")
	}
	return nil
}
359736
func (bw *BlockWorker) Run(ctx context.Context) error {
	resetVars()
	err := bw.run(ctx)

	bw.SetState(WorkerStateCleanUp)
	if err != nil {
		bw.SetState(WorkerStateError)
		return err
	}
	bw.SetState(WorkerStateDone)
	return nil
}
177660
func (m *Main) Run() error {
	conf := sarama.NewConfig()
	conf.Version = sarama.V0_10_0_0
	conf.Producer.Return.Successes = true
	producer, err := sarama.NewSyncProducer(m.Hosts, conf)
	if err != nil {
		return errors.Wrap(err, "getting new producer")
	}
	defer producer.Close()

	for ticker := time.NewTicker(m.Rate); true; <-ticker.C {
		ev := fake.GenEvent()
		msg := &sarama.ProducerMessage{Topic: m.Topic, Value: JSONEvent(*ev)}
		_, _, err := producer.SendMessage(msg)
		if err != nil {
			log.Printf("Error sending message: '%v', backing off", err)
			time.Sleep(time.Second * 10)
		}
	}
	return nil
}
393355
func (s *Server) Run() {
	go s.updater()
	go s.notifier()
	go s.manager()

	<-s.quit
}