// Run implements the Worker interface.
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
}
518960
func (j *Job) run() (result []reflect.Value, err error) {
	f := reflect.ValueOf(j.funcs[j.jobFunc])
	params := j.fparams[j.jobFunc]
	if len(params) != f.Type().NumIn() {
		err = errors.New("the number of param is not adapted")
		return
	}
	in := make([]reflect.Value, len(params))
	for k, param := range params {
		in[k] = reflect.ValueOf(param)
	}
	result = f.Call(in)
	j.lastRun = time.Now()
	j.scheduleNextRun()
	return
}
547761
func (e *Echo) Run(eng engine.Engine, handler ...engine.Handler) error {
	err := e.setEngine(eng).start(handler...)
	if err != nil {
		fmt.Println(err)
	}
	return err
}
383639
func (c *registerCommand) Run(ctx *cmd.Context) error {
	err := c.run(ctx)
	if err != nil && c.onRunError != nil {
		c.onRunError()
	}
	return err
}
84786
func (worker *copIteratorWorker) run(ctx context.Context) {
	defer worker.wg.Done()
	for task := range worker.taskCh {
		respCh := worker.respChan
		if respCh == nil {
			respCh = task.respChan
		}

		bo := NewBackoffer(ctx, copNextMaxBackoff).WithVars(worker.vars)
		worker.handleTask(bo, task, respCh)
		if bo.totalSleep > 0 {
			metrics.TiKVBackoffHistogram.Observe(float64(bo.totalSleep) / 1000)
		}
		close(task.respChan)
		select {
		case <-worker.finishCh:
			return
		default:
		}
	}
}
1310649
func (c *APIServiceRegistrationController) Run(stopCh <-chan struct{}) {
	defer utilruntime.HandleCrash()
	defer c.queue.ShutDown()

	klog.Infof("Starting APIServiceRegistrationController")
	defer klog.Infof("Shutting down APIServiceRegistrationController")

	if !controllers.WaitForCacheSync("APIServiceRegistrationController", stopCh, c.apiServiceSynced) {
		return
	}

	// only start one worker thread since its a slow moving API and the aggregation server adding bits
	// aren't threadsafe
	go wait.Until(c.runWorker, time.Second, stopCh)

	<-stopCh
}
398680
func (c *autoRegisterController) Run(threadiness int, stopCh <-chan struct{}) {
	// don't let panics crash the process
	defer utilruntime.HandleCrash()
	// make sure the work queue is shutdown which will trigger workers to end
	defer c.queue.ShutDown()

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

	// wait for your secondary caches to fill before starting your work
	if !controllers.WaitForCacheSync("autoregister", stopCh, c.apiServiceSynced) {
		return
	}

	// record APIService objects that existed when we started
	if services, err := c.apiServiceLister.List(labels.Everything()); err == nil {
		for _, service := range services {
			c.apiServicesAtStart[service.Name] = true
		}
	}

	// start up your worker threads based on threadiness.  Some controllers have multiple kinds of workers
	for i := 0; i < threadiness; i++ {
		// runWorker will loop until "something bad" happens.  The .Until will then rekick the worker
		// after one second
		go wait.Until(c.runWorker, time.Second, stopCh)
	}

	// wait until we're told to stop
	<-stopCh
}
1196161
func (h *Interface) run(jobs chan job, ctx context.Context) {

	ListenAndServeIf(h.Name, h, jobs, ctx)
}
81914
func Run(eng engine.Engine, handler ...engine.Handler) error {
	return Default.Run(eng, handler...)
}
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()
		}
	}
}
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
		}
	}()
}
964777
func (c *TemplateInstanceController) Run(workers int, stopCh <-chan struct{}) {
	defer utilruntime.HandleCrash()
	defer c.queue.ShutDown()

	if !cache.WaitForCacheSync(stopCh, c.informer.HasSynced) {
		return
	}

	klog.V(2).Infof("Starting TemplateInstance controller")

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

	<-stopCh
}
81814
func (c *SetMeterStatusCommand) Run(ctx *cmd.Context) error {
	client, err := newClient(c.ModelCommandBase)
	if err != nil {
		return errors.Trace(err)
	}
	defer client.Close()
	err = client.SetMeterStatus(c.Tag.String(), c.Status, c.StatusInfo)
	if err != nil {
		return errors.Trace(err)
	}
	return nil
}
947337
func (c *applicationVersionSetCommand) Run(ctx *cmd.Context) error {
	return c.ctx.SetUnitWorkloadVersion(c.version)
}
148778
func (t *Timer) Run() {
	log.Print("Timer is running now.")
	go func() {
		for {
			select {
			case <-t.ticker.C:
				go func() {
					log.Print("Worker is doing its jobs.")
					t.worker.Work()
				}()
			}
		}
	}()
}
272024
func (bc *BuildController) Run(workers int, stopCh <-chan struct{}) {
	defer utilruntime.HandleCrash()
	defer bc.buildQueue.ShutDown()
	defer bc.buildConfigQueue.ShutDown()
	defer bc.controllerConfigQueue.ShutDown()

	// Wait for the controller stores to sync before starting any work in this controller.
	if !cache.WaitForCacheSync(stopCh,
		bc.buildStoreSynced,
		bc.podStoreSynced,
		bc.secretStoreSynced,
		bc.imageStreamStoreSynced,
		bc.configMapStoreSynced) {
		utilruntime.HandleError(fmt.Errorf("timed out waiting for caches to sync"))
		return
	}

	// Integration tests currently do not support cache sync for operator-installed custom resource definitions
	if os.Getenv("OS_INTEGRATION_TEST") != "true" {
		if !cache.WaitForCacheSync(stopCh,
			bc.buildControllerConfigStoreSynced,
			bc.imageConfigStoreSynced) {
			utilruntime.HandleError(fmt.Errorf("timed out waiting for caches to sync"))
			return
		}
	}

	// Sync build controller config on start up.
	if errs := bc.handleControllerConfig(); len(errs) > 0 {
		utilruntime.HandleError(fmt.Errorf("errors syncing build controller config: %v", errs))
	}

	klog.Infof("Starting build controller")

	// Only need one worker to manage config
	go wait.Until(bc.controllerConfigWorker, time.Second, stopCh)

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

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

	metrics.IntializeMetricsCollector(bc.buildLister)

	<-stopCh
	klog.Infof("Shutting down build controller")
}
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
}