// startTableWorker launchs some background goroutines which pick tasks from workCh and execute the task.
770140
func (e *IndexLookUpExecutor) startTableWorker(ctx context.Context, workCh <-chan *lookupTableTask) {
	lookupConcurrencyLimit := e.ctx.GetSessionVars().IndexLookupConcurrency
	e.tblWorkerWg.Add(lookupConcurrencyLimit)
	for i := 0; i < lookupConcurrencyLimit; i++ {
		worker := &tableWorker{
			idxLookup:      e,
			workCh:         workCh,
			finished:       e.finished,
			buildTblReader: e.buildTableReader,
			keepOrder:      e.keepOrder,
			handleIdx:      e.handleIdx,
			isCheckOp:      e.isCheckOp,
			memTracker:     memory.NewTracker(tableWorkerLabel, -1),
		}
		worker.memTracker.AttachTo(e.memTracker)
		ctx1, cancel := context.WithCancel(ctx)
		go func() {
			worker.pickAndExecTask(ctx1)
			cancel()
			e.tblWorkerWg.Done()
		}()
	}
}
253635
func (c *Client) AsyncStart() error {
	if err := c.Listen(); err != nil {
		return err
	}
	var started sync.WaitGroup
	started.Add(1)
	go func() {
		started.Done()
		c.Serve()
	}()
	started.Wait()
	return nil
}
545134
func NewCeleryWorker(broker CeleryBroker, backend CeleryBackend, numWorkers int) *CeleryWorker {
	return &CeleryWorker{
		broker:          broker,
		backend:         backend,
		numWorkers:      numWorkers,
		registeredTasks: map[string]interface{}{},
		rateLimitPeriod: 100 * time.Millisecond,
	}
}
1337878
func (q *SQSDispatcher) Start(handle func(context.Context, *sqs.Message) error) {
	var wg sync.WaitGroup
	for i := 0; i < q.NumWorkers; i++ {
		wg.Add(1)
		go func() {
			defer wg.Done()
			q.start(handle)
		}()
	}

	wg.Wait()
}
1207329
func (t *CliLogBoxTask) Start(wg *sync.WaitGroup) golib.StopChan {
	if t.Update == nil {
		return golib.NewStoppedChan(errors.New("CliLogBoxTask.Update cannot be nil"))
	}
	t.InterceptLoggers()
	t.updateTask = &golib.LoopTask{
		Description: "CliLogBoxTask",
		StopHook: func() {
			err := t.updateBox() // One last screen refresh to make sure no messages get lost.
			t.RestoreLoggers()
			golib.Printerr(err)
		},
		Loop: func(stop golib.StopChan) (err error) {
			err = t.updateBox()
			if err == nil {
				// Wait between t.MinUpdateInterval and t.UpdateInterval,
				// but wake up from stop.WaitChan() and t.updateTrigger.
				sleepStart := time.Now()
				select {
				case <-time.After(t.UpdateInterval):
				case <-stop.WaitChan():
				case <-t.updateTrigger:
				}
				sleepTime := time.Now().Sub(sleepStart)
				if diff := t.MinUpdateInterval - sleepTime; diff > 0 {
					select {
					// Don't wait for t.updateTrigger here
					case <-time.After(diff):
					case <-stop.WaitChan():
					}
				}
			}
			return
		},
	}
	return t.updateTask.Start(wg)
}
258570
func startFunc(a agent.Agent) dependency.StartFunc {
	return func(_ dependency.Context) (worker.Worker, error) {
		w := &agentWorker{agent: a}
		w.tomb.Go(func() error {
			<-w.tomb.Dying()
			return nil
		})
		return w, nil
	}
}
704038
func (w *WorkerPool) Start() {
	w.mu.Lock()
	defer w.mu.Unlock()

	for i := range w.workers {
		w.workers[i] = NewWorker(w.c, w.WorkMap)
		w.workers[i].Interval = w.Interval
		w.workers[i].Queue = w.Queue
		go w.workers[i].Work()
	}
}
729602
func (w *worker) start() {
	for {
		select {
		case q := <-w.rcv:
			// assert that deadline is not exceeded
			if chrono.Chrono.Now().Before(q.Job().Deadline()) {
				q.Job().Run()
			} else {
				// the deadline was exceeded and this job will not run
				q.Job().AddErrors(errors.New("Worker refused to run overdue job."))
			}

			// mark the job complete
			q.Promise().Complete(q.Job().Errors())

		// the single kill-channel -- used when resizing worker pools
		case <-w.kamikaze:
			return

		//the broadcast that kills all workers
		case <-workerKillChan:
			return
		}
	}
}
145534
func NewWorker(srv *Server) (*Worker, error) {
	w := &Worker{
		srv:    srv,
		logger: srv.logger.ResetNamed("worker"),
		start:  time.Now(),
	}
	w.pauseCond = sync.NewCond(&w.pauseLock)
	go w.run()
	return w, nil
}
852089
func (s *RouterService) StartRouter(p *StartRouterParams) (*StartRouterResponse, error) {
	resp, err := s.cs.newRequest("startRouter", p.toURLValues())
	if err != nil {
		return nil, err
	}

	var r StartRouterResponse
	if err := json.Unmarshal(resp, &r); err != nil {
		return nil, err
	}

	// If we have a async client, we need to wait for the async result
	if s.cs.async {
		b, err := s.cs.GetAsyncJobResult(r.JobID, s.cs.timeout)
		if err != nil {
			if err == AsyncTimeoutErr {
				return &r, err
			}
			return nil, err
		}

		b, err = getRawValue(b)
		if err != nil {
			return nil, err
		}

		if err := json.Unmarshal(b, &r); err != nil {
			return nil, err
		}
	}

	return &r, nil
}
633816
func (c *ServiceWorker) StartWorkerWithParams(v *ServiceWorkerStartWorkerParams) (*gcdmessage.ChromeResponse, error) {
	return gcdmessage.SendDefaultRequest(c.target, c.target.GetSendCh(), &gcdmessage.ParamRequest{Id: c.target.GetId(), Method: "ServiceWorker.startWorker", Params: v})
}
387842
func (w *DefaultWorker) Start() (<-chan *Task, <-chan error) {
	w.smu.Lock()
	defer w.smu.Unlock()

	w.state = open
	w.lifecycle.SetQueues(w.availableTasks, w.workingTasks)
	w.janitorRunner = newJanitorRunner(w.pool, w.detector, w.janitor, w.availableTasks)

	errs1 := w.janitorRunner.Start()
	tasks, errs2 := w.lifecycle.Listen()

	return tasks, concatErrs(errs1, errs2)
}
185043
func (gs *Server) StartRoutine() {
	gs.wg.Add(1)
	atomic.AddInt32(&gs.routinesCount, 1)
}
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}
}
1371956
func startCleaner(cache *Cache, interval time.Duration) {
	cleaner := &cleaner{
		interval: interval,
		stop:     make(chan bool),
	}

	cache.cleaner = cleaner
	go cleaner.Run(cache)
}
1271646
func (config FlagManifoldConfig) start(context dependency.Context) (worker.Worker, error) {
	var gate Waiter
	if err := context.Get(config.GateName, &gate); err != nil {
		return nil, errors.Trace(err)
	}
	worker, err := config.NewWorker(gate)
	if err != nil {
		return nil, errors.Trace(err)
	}
	return worker, nil

}
1031452
func (w *GCWorker) Start() {
	var ctx context.Context
	ctx, w.cancel = context.WithCancel(context.Background())
	var wg sync.WaitGroup
	wg.Add(1)
	go w.start(ctx, &wg)
	wg.Wait() // Wait create session finish in worker, some test code depend on this to avoid race.
}