// Workers returns a summary of workers connected to gearman.
512871
func (ga GearmanAdmin) Workers() ([]Worker, error) {
	var workers []Worker
	fmt.Fprintf(ga.conn, "workers\n")
	scanner := bufio.NewScanner(ga.conn)
	for scanner.Scan() && scanner.Text() != "." {
		toks := strings.Split(scanner.Text(), " ")
		if len(toks) < 4 {
			return workers, fmt.Errorf("unexpected worker: '%s'", scanner.Text())
		}
		workers = append(workers, Worker{
			Fd:        toks[0],
			IPAddress: toks[1],
			ClientID:  toks[2],
			Functions: toks[4:],
		})
	}
	return workers, scanner.Err()
}
789966
func NewWorker(sm *source.Manager, vm *storage.VendorCollectionManager, rm *storage.RecordCollectionManager) *Worker {
	return &Worker{
		sm: sm,
		vm: vm,
		rm: rm,
	}
}
37737
func (r *Resident) cleanupWorkers() error {
	var errs []string
	for id, w := range r.workers {
		if err := worker.Stop(w); err != nil {
			errs = append(errs, errors.Annotatef(err, "worker %d", id).Error())
		}
	}

	if len(errs) != 0 {
		return errors.Errorf("worker cleanup errors:\n\t%s", strings.Join(errs, "\n\t"))
	}
	return nil
}
103062
func (p *ParallelManager) addWorker() {
	if atomic.LoadUint32(&p.workersNum) >= maxParallelWorkers {
		// Number of maximum workers is reached, no need to
		// to create a new one.
		return
	}

	// Update number of threads
	atomic.AddUint32(&p.workersNum, 1)

	// Start a new worker
	p.wg.Add(1)
	go func() {
		for {
			// Wait for jobs
			fn, ok := <-p.queueCh
			if !ok {
				// No more tasks, quit
				p.wg.Done()
				return
			}
			// Execute the task and send the result
			// to result channel.
			p.resultCh <- fn()
		}
	}()
}
922165
func (r *Resident) deregisterWorker(id uint64) {
	r.mu.Lock()
	delete(r.workers, id)
	r.mu.Unlock()
}
1088474
func (p *Mx) Workers() int {
	p.mu.Lock()
	defer p.mu.Unlock()
	return p.workers
}
1098102
func (prod *SimpleProducer) SetWorkerWaitGroup(workers *sync.WaitGroup) {
	prod.runState.SetWorkerWaitGroup(workers)
}
754904
func (w *Worker) Work() {
	for {
		select {
		case m := <-w.PacketChan:
			w.ProcessMetric(&m)
		case m := <-w.ImportChan:
			for _, j := range m {
				w.ImportMetric(j)
			}
		case ms := <-w.ImportMetricChan:
			for _, m := range ms {
				w.ImportMetricGRPC(m)
			}
		case <-w.QuitChan:
			// We have been asked to stop.
			log.WithField("worker", w.id).Error("Stopping")
			return
		}
	}
}
438679
func (c *connections) Agents() []string {
	agents := set.NewStrings()
	for _, value := range c.values {
		agents.Add(value.Agent)
	}
	return agents.Values()
}
830329
func NewWorkerManager(id string, config *ConsumerConfig, topicPartition TopicAndPartition, metrics *ConsumerMetrics, closeConsumer chan bool) *WorkerManager {
	workers := make([]*Worker, config.NumWorkers)
	availableWorkers := make(chan *Worker, config.NumWorkers)
	for i := 0; i < config.NumWorkers; i++ {
		workers[i] = &Worker{
			InputChannel:         make(chan *TaskAndStrategy),
			OutputChannel:        make(chan WorkerResult),
			HandlerInputChannel:  make(chan *TaskAndStrategy),
			HandlerOutputChannel: make(chan WorkerResult),
			TaskTimeout:          config.WorkerTaskTimeout,
		}
		workers[i].Start()
		availableWorkers <- workers[i]
	}

	return &WorkerManager{
		id:                  id,
		config:              config,
		availableWorkers:    availableWorkers,
		workers:             workers,
		inputChannel:        make(chan []*Message),
		currentBatch:        newTaskBatch(),
		batchOrder:          make([]TaskId, 0),
		topicPartition:      topicPartition,
		largestOffset:       InvalidOffset,
		lastCommittedOffset: InvalidOffset,
		failCounter:         NewFailureCounter(config.WorkerRetryThreshold, config.WorkerThresholdTimeWindow),
		batchProcessed:      make(chan bool),
		managerStop:         make(chan bool),
		processingStop:      make(chan bool),
		commitStop:          make(chan bool),
		metrics:             metrics,
		closeConsumer:       closeConsumer,
	}
}
1260783
func (m *manager) workerCount() int {
	m.workerLock.RLock()
	defer m.workerLock.RUnlock()
	return len(m.workers)
}
610158
func (lbc *loadBalancerController) worker() {
	for {
		key, _ := lbc.queue.Get()
		glog.Infof("Sync triggered by service %v", key)
		if err := lbc.sync(false); err != nil {
			glog.Warningf("Requeuing %v because of error: %v", key, err)
			lbc.queue.Add(key)
		}
		lbc.queue.Done(key)
	}
}
1250817
func (w *workManager) sendToWorker(j queuedJob) {
	switch j.Job().Type() {
	case collectJobType:
		w.collectchan <- j
	case publishJobType:
		w.publishchan <- j
	case processJobType:
		w.processchan <- j
	}
}
1239379
func (r resources) registerAgentWorkers() {
	if !markRegistered(resource.ComponentName, "agent-workers") {
		return
	}

	charmrevisionupdater.RegisterLatestCharmHandler("resources", resourceadapters.NewLatestCharmHandler)
}
449338
func NewWorker(st *keyupdater.State, agentConfig agent.Config) (worker.Worker, error) {
	machineTag, ok := agentConfig.Tag().(names.MachineTag)
	if !ok {
		return nil, errors.NotValidf("machine tag %v", agentConfig.Tag())
	}
	if os.HostOS() == os.Windows {
		return jworker.NewNoOpWorker(), nil
	}
	w, err := watcher.NewNotifyWorker(watcher.NotifyConfig{
		Handler: &keyupdaterWorker{
			st:  st,
			tag: machineTag,
		},
	})
	if err != nil {
		return nil, errors.Trace(err)
	}
	return w, nil
}
40315
func GetAllWorkers() []*Worker {
	o := orm.NewOrm()

	var workers []*Worker
	num, err := o.QueryTable("worker").All(&workers)
	fmt.Printf("Returned Rows Num: %s, %s", num, err)
	return workers
}
560672
func (w *worker) sendRangeTaskToWorkers(t table.Table, workers []*addIndexWorker, reorgInfo *reorgInfo, totalAddedCount *int64, kvRanges []kv.KeyRange) ([]kv.KeyRange, error) {
	batchTasks := make([]*reorgIndexTask, 0, len(workers))
	physicalTableID := reorgInfo.PhysicalTableID

	// Build reorg indices tasks.
	for _, keyRange := range kvRanges {
		startHandle, endHandle, err := decodeHandleRange(keyRange)
		if err != nil {
			return nil, errors.Trace(err)
		}

		endKey := t.RecordKey(endHandle)
		endIncluded := false
		if endKey.Cmp(keyRange.EndKey) < 0 {
			endIncluded = true
		}
		task := &reorgIndexTask{physicalTableID, startHandle, endHandle, endIncluded}
		batchTasks = append(batchTasks, task)

		if len(batchTasks) >= len(workers) {
			break
		}
	}

	if len(batchTasks) == 0 {
		return nil, nil
	}

	// Wait tasks finish.
	err := w.handleReorgTasks(reorgInfo, totalAddedCount, workers, batchTasks)
	if err != nil {
		return nil, errors.Trace(err)
	}

	if len(batchTasks) < len(kvRanges) {
		// there are kvRanges not handled.
		remains := kvRanges[len(batchTasks):]
		return remains, nil
	}

	return nil, nil
}