// Run all jobs with delay seconds
139
func (s *Scheduler) RunAllwithDelay(d int) {
	for i := 0; i < s.size; i++ {
		s.jobs[i].run()
		time.Sleep(time.Duration(d))
	}
}
42805
func (mod *TimeBee) Run(eventChan chan bees.Event) {
	mod.eventChan = eventChan
	for {
		select {
		case <-mod.SigChan:
			return

		default:
		}

		mod.timer()
		time.Sleep(500 * time.Millisecond)
	}
}
84635
func (a *HorizontalController) Run(stopCh <-chan struct{}) {
	defer utilruntime.HandleCrash()
	defer a.queue.ShutDown()

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

	if !controller.WaitForCacheSync("HPA", stopCh, a.hpaListerSynced, a.podListerSynced) {
		return
	}

	// start a single worker (we may wish to start more in the future)
	go wait.Until(a.worker, time.Second, stopCh)

	<-stopCh
}
814525
func Run(workers []Worker, duration time.Duration) *Result {
	debug = os.Getenv("DEBUG") != ""
	c := len(workers)
	log.Printf("starting benchmark: concurrency: %d, time: %s, GOMAXPROCS: %d", c, duration, runtime.GOMAXPROCS(0))
	startCh := make(chan bool, c)
	readyCh := make(chan bool, c)
	var stopFlag int32
	scoreCh := make(chan int, c)
	var wg sync.WaitGroup

	// spawn worker goroutines
	for i, w := range workers {
		debugLog("spwan worker[%d]", i)
		go func(n int, worker Worker) {
			wg.Add(1)
			defer wg.Done()
			score := 0
			worker.Setup()
			readyCh <- true // ready of worker:n
			<-startCh       // notified go benchmark from Runner
			debugLog("worker[%d] starting Benchmark()", n)
			for atomic.LoadInt32(&stopFlag) == 0 {
				score += worker.Process()
			}
			scoreCh <- score
			debugLog("worker[%d] done Benchmark() score: %d", n, score)
			worker.Teardown()
			debugLog("worker[%d] exit", n)
		}(i, w)
	}

	// wait for ready of workres
	debugLog("waiting for all workers finish Setup()")
	for i := 0; i < c; i++ {
		<-readyCh
	}

	// notify "start" to workers
	close(startCh)
	start := time.Now()

	// wait for catching signal or timed out
	signalCh := make(chan os.Signal, 1)
	signal.Notify(signalCh, TrapSignals...)
	select {
	case s := <-signalCh:
		switch sig := s.(type) {
		case syscall.Signal:
			log.Printf("Got signal: %s(%d)", sig, sig)
		default:
			log.Printf("interrupted %s", s)
			break
		}
	case <-time.After(duration):
		debugLog("timed out")
		break
	}

	// notify "stop" to workers
	atomic.StoreInt32(&stopFlag, 1)

	// collect scores from workers
	totalScore := 0
	for i := 0; i < c; i++ {
		totalScore += <-scoreCh
	}
	end := time.Now()
	elapsed := end.Sub(start)
	log.Printf("done benchmark: score %d, elapsed %s = %f / sec\n", totalScore, elapsed, float64(totalScore)/float64(elapsed)*float64(time.Second))

	wg.Wait()
	return &Result{Score: totalScore, Elapsed: elapsed}
}
419382
func (ac *AuthorizationCache) Run(period time.Duration) {
	ac.skip = &statelessSkipSynchronizer{}

	go utilwait.Forever(func() { ac.synchronize() }, period)
}
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
}
406983
func (dc *DeploymentController) Run(workers int, stopCh <-chan struct{}) {
	defer utilruntime.HandleCrash()
	defer dc.queue.ShutDown()

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

	if !controller.WaitForCacheSync("deployment", stopCh, dc.dListerSynced, dc.rsListerSynced, dc.podListerSynced) {
		return
	}

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

	<-stopCh
}
644668
func (s *ScheduledImageStreamController) Run(stopCh <-chan struct{}) {
	defer utilruntime.HandleCrash()

	klog.Infof("Starting scheduled import controller")

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

	go s.scheduler.RunUntil(stopCh)

	metrics.InitializeImportCollector(true, s.importCounter.Collect)

	<-stopCh
	klog.Infof("Shutting down image stream controller")
}
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
}
1166739
func (nm *NamespaceController) Run(workers int, stopCh <-chan struct{}) {
	defer utilruntime.HandleCrash()
	defer nm.queue.ShutDown()

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

	if !controller.WaitForCacheSync("namespace", stopCh, nm.listerSynced) {
		return
	}

	klog.V(5).Info("Starting workers of namespace controller")
	for i := 0; i < workers; i++ {
		go wait.Until(nm.worker, time.Second, stopCh)
	}
	<-stopCh
}
419760
func (tc *Controller) Run(workers int, stopCh <-chan struct{}) {
	defer utilruntime.HandleCrash()
	defer tc.queue.ShutDown()

	klog.Infof("Starting TTL after finished controller")
	defer klog.Infof("Shutting down TTL after finished controller")

	if !controller.WaitForCacheSync("TTL after finished", stopCh, tc.jListerSynced) {
		return
	}

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

	<-stopCh
}
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
}
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()
				}()
			}
		}
	}()
}
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
}
250497
func (msdw *MultiSplitDiffWorker) Run(ctx context.Context) error {
	resetVars()
	err := msdw.run(ctx)

	msdw.SetState(WorkerStateCleanUp)
	cerr := msdw.cleaner.CleanUp(msdw.wr)
	if cerr != nil {
		if err != nil {
			msdw.wr.Logger().Errorf("CleanUp failed in addition to job error: %v", cerr)
		} else {
			err = cerr
		}
	}
	if err != nil {
		msdw.wr.Logger().Errorf("Run() error: %v", err)
		msdw.SetState(WorkerStateError)
		return err
	}
	msdw.SetState(WorkerStateDone)
	return nil
}
850892
func (c *Cron) run() {

	var effective time.Time
	now := time.Now().Local()

	// to figure next trig time for entries, referenced from now
	for _, e := range c.entries {
		e.Next = e.Schedule.Next(now)
	}

	for {
		sort.Sort(byTime(c.entries))
		if len(c.entries) > 0 {
			effective = c.entries[0].Next
		} else {
			effective = now.AddDate(15, 0, 0) // to prevent phantom jobs.
		}

		select {
		case now = <-after(effective.Sub(now)):
			// entries with same time gets run.
			for _, entry := range c.entries {
				if entry.Next != effective {
					break
				}
				entry.Prev = now
				entry.Next = entry.Schedule.Next(now)
				go entry.Job.Run()
			}
		case e := <-c.add:
			e.Next = e.Schedule.Next(time.Now())
			c.entries = append(c.entries, e)
		case <-c.stop:
			return // terminate go-routine.
		}
	}
}
1367944
func (pq *PriorityQueue) Run(ctx context.Context, f func(interface{})) {
	top := len(pq.Queues) - 1
	p := top
READ:
	for {
		q := pq.Queues[p]
		select {
		case <-ctx.Done():
			return
		case x := <-q:
			val := x.(struct {
				v interface{}
				t time.Time
			})
			f(val.v)
			metrics.GetOrRegisterResettingTimer("pq.run", nil).UpdateSince(val.t)
			p = top
		default:
			if p > 0 {
				p--
				continue READ
			}
			p = top
			select {
			case <-ctx.Done():
				return
			case <-pq.wakeup:
			}
		}
	}
}