// libnetwork side of agent depends on the keys. On the first receipt of
// keys setup the agent. For subsequent key set handle the key change
177606
func (c *controller) SetKeys(keys []*types.EncryptionKey) error {
	subsysKeys := make(map[string]int)
	for _, key := range keys {
		if key.Subsystem != subsysGossip &&
			key.Subsystem != subsysIPSec {
			return fmt.Errorf("key received for unrecognized subsystem")
		}
		subsysKeys[key.Subsystem]++
	}
	for s, count := range subsysKeys {
		if count != keyringSize {
			return fmt.Errorf("incorrect number of keys for subsystem %v", s)
		}
	}

	agent := c.getAgent()

	if agent == nil {
		c.Lock()
		c.keys = keys
		c.Unlock()
		return nil
	}
	return c.handleKeyChange(keys)
}
1064206
func (c *BuildConfigController) work() bool {
	key, quit := c.queue.Get()
	if quit {
		return true
	}

	defer c.queue.Done(key)

	bc, err := c.getBuildConfigByKey(key.(string))
	if err != nil {
		c.handleError(err, key)
		return false
	}
	if bc == nil {
		return false
	}

	err = c.handleBuildConfig(bc)
	c.handleError(err, key)

	return false
}
533824
func (a *Allocator) getNodeNetworks(nodeID string) ([]*api.Network, error) {
	var (
		// no need to initialize networks. we only append to it, and appending
		// to a nil slice is valid. this has the added bonus of making this nil
		// if we return an error
		networks []*api.Network
		err      error
	)
	a.store.View(func(tx store.ReadTx) {
		// get all tasks currently assigned to this node. it's no big deal if
		// the tasks change in the meantime, there's no race to clean up
		// unneeded network attachments on a node.
		var tasks []*api.Task
		tasks, err = store.FindTasks(tx, store.ByNodeID(nodeID))
		if err != nil {
			return
		}
		// we need to keep track of network IDs that we've already added to the
		// list of networks we're going to return. we could do
		// map[string]*api.Network and then convert to []*api.Network and
		// return that, but it seems cleaner to have a separate set and list.
		networkIDs := map[string]struct{}{}
		for _, task := range tasks {
			// we don't need to check if a task is before the Assigned state.
			// the only way we have a task with a NodeID that isn't yet in
			// Assigned is if it's a global service task. this check is not
			// necessary:
			// if task.Status.State < api.TaskStateAssigned {
			//     continue
			// }
			if task.Status.State > api.TaskStateRunning {
				// we don't need to have network attachments for a task that's
				// already in a terminal state
				continue
			}

			// now go through the task's network attachments and find all of
			// the networks
			for _, attachment := range task.Networks {
				// if the network is an overlay network, and the network ID is
				// not yet in the set of network IDs, then add it to the set
				// and add the network to the list of networks we'll be
				// returning
				if _, ok := networkIDs[attachment.Network.ID]; isOverlayNetwork(attachment.Network) && !ok {
					networkIDs[attachment.Network.ID] = struct{}{}
					// we don't need to worry about retrieving the network from
					// the store, because the network in the attachment is an
					// identical copy of the network in the store.
					networks = append(networks, attachment.Network)
				}
			}
		}
	})

	// finally, we need the ingress network if one exists.
	if a.netCtx != nil && a.netCtx.ingressNetwork != nil {
		networks = append(networks, a.netCtx.ingressNetwork)
	}

	return networks, err
}
472998
func (c *NamespaceSCCAllocationController) work() bool {
	key, quit := c.queue.Get()
	if quit {
		return false
	}
	defer c.queue.Done(key)

	if err := c.syncNamespace(key.(string)); err == nil {
		// this means the request was successfully handled.  We should "forget" the item so that any retry
		// later on is reset
		c.queue.Forget(key)
	} else {
		// if we had an error it means that we didn't handle it, which means that we want to requeue the work
		utilruntime.HandleError(fmt.Errorf("error syncing namespace, it will be retried: %v", err))
		c.queue.AddRateLimited(key)
	}
	return true
}
92269
func (g *BlockGate) GetChainWork(hash *chainhash.Hash) (string, error) {
	return GetChainWork(g.client, hash)
}
1171325
func (nm *NamespaceController) worker() {
	workFunc := func() bool {
		key, quit := nm.queue.Get()
		if quit {
			return true
		}
		defer nm.queue.Done(key)

		err := nm.syncNamespaceFromKey(key.(string))
		if err == nil {
			// no error, forget this entry and return
			nm.queue.Forget(key)
			return false
		}

		if estimate, ok := err.(*deletion.ResourcesRemainingError); ok {
			t := estimate.Estimate/2 + 1
			klog.V(4).Infof("Content remaining in namespace %s, waiting %d seconds", key, t)
			nm.queue.AddAfter(key, time.Duration(t)*time.Second)
		} else {
			// rather than wait for a full resync, re-add the namespace to the queue to be processed
			nm.queue.AddRateLimited(key)
			utilruntime.HandleError(err)
		}
		return false
	}

	for {
		quit := workFunc()

		if quit {
			return
		}
	}
}
172019
func NewGCWorker(store tikv.Storage, pdClient pd.Client) (tikv.GCHandler, error) {
	ver, err := store.CurrentVersion()
	if err != nil {
		return nil, errors.Trace(err)
	}
	hostName, err := os.Hostname()
	if err != nil {
		hostName = "unknown"
	}
	worker := &GCWorker{
		uuid:        strconv.FormatUint(ver.Ver, 16),
		desc:        fmt.Sprintf("host:%s, pid:%d, start at %s", hostName, os.Getpid(), time.Now()),
		store:       store,
		pdClient:    pdClient,
		gcIsRunning: false,
		lastFinish:  time.Now(),
		done:        make(chan error),
	}
	return worker, nil
}
1315588
func NewDbWorker(db *DbGorp, workInfo DbWorkInfo, numWorkers int) (container *DbWorkerContainer) {

	container = &DbWorkerContainer{
		SharedWorker: SharedWorker{
			InputChannel:   make(chan interface{}, numWorkers),
			OutputChannel:  make(chan interface{}, numWorkers),
			ControlChannel: make(chan func() (WorkerPhase, *DbWorker), numWorkers),
			workInfo:       workInfo,
		},
		NumWorkers:       numWorkers,
		Db:               db,
		StartWorkTimeout: 0,
		LongWorkTimeout:  0,
	}
	return
}
24885
func (c *Client) GetWorkSubmitAsync(data string) FutureGetWorkSubmit {
	cmd := btcjson.NewGetWorkCmd(&data)
	return c.sendCmd(cmd)
}
1265439
func worker_task(i int) {
	worker, _ := zmq.NewSocket(zmq.REQ)
	defer worker.Close()
	worker.Connect("ipc://" + self + "-localbe.ipc")

	//  Tell broker we're ready for work
	worker.SendMessage(WORKER_READY)

	//  Process messages as they arrive
	for {
		msg, err := worker.RecvMessage(0)
		if err != nil {
			break //  Interrupted
		}

		//  Workers are busy for 0/1 seconds
		time.Sleep(time.Duration(rand.Intn(2)) * time.Second)
		n := len(msg) - 1
		worker.SendMessage(msg[:n], fmt.Sprintf("%s %s-%d", msg[n], self, i))
	}
}
674314
func (w *workManager) AddPublishWorker() {
	nw := newWorker(w.publishchan)
	go nw.start()
	w.publishWkrs = append(w.publishWkrs, nw)
	w.publishWkrSize++
}
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
		}
	}
}
460285
func newAgent(net, addr string, worker *Worker) (a *agent, err error) {
	a = &agent{
		net:    net,
		addr:   addr,
		worker: worker,
		in:     make(chan []byte, queueSize),
	}
	return
}
1259689
func (config ManifoldConfig) newWorker(a agent.Agent, apiCaller base.APICaller) (worker.Worker, error) {
	cfg := a.CurrentConfig()
	// Grab the tag and ensure that it's for a machine.
	tag, ok := cfg.Tag().(names.MachineTag)
	if !ok {
		return nil, errors.New("agent's tag is not a machine tag")
	}

	// Get the machine agent's jobs.
	// TODO(fwereade): this functionality should be on the
	// deployer facade instead.
	agentFacade, err := apiagent.NewState(apiCaller)
	if err != nil {
		return nil, errors.Trace(err)
	}
	entity, err := agentFacade.Entity(tag)
	if err != nil {
		return nil, errors.Trace(err)
	}

	var isUnitHoster bool
	for _, job := range entity.Jobs() {
		if job == multiwatcher.JobHostUnits {
			isUnitHoster = true
			break
		}
	}

	if !isUnitHoster {
		return nil, dependency.ErrUninstall
	}

	deployerFacade := apideployer.NewState(apiCaller)
	context := config.NewDeployContext(deployerFacade, cfg)
	w, err := NewDeployer(deployerFacade, context)
	if err != nil {
		return nil, errors.Annotate(err, "cannot start unit agent deployer worker")
	}
	return w, nil
}
792220
func (p *AgentClient) EmitZipkinBatch(spans []*zipkincore.Span) (err error) {
	if err = p.sendEmitZipkinBatch(spans); err != nil {
		return
	}
	return
}
229110
func parseNetworkGenericOptions(data interface{}) (*configuration, error) {
	var (
		err    error
		config *configuration
	)
	switch opt := data.(type) {
	case *configuration:
		config = opt
	case map[string]string:
		config = &configuration{}
		err = config.fromOptions(opt)
	case options.Generic:
		var opaqueConfig interface{}
		if opaqueConfig, err = options.GenerateFromModel(opt, config); err == nil {
			config = opaqueConfig.(*configuration)
		}
	default:
		err = types.BadRequestErrorf("unrecognized network configuration format: %v", opt)
	}
	return config, err
}
920087
func parseNetwork(addr net.Addr) string {
	switch addr := addr.(type) {
	// TCP addresses resolved through net.ResolveTCPAddr give a default
	// network of "tcp", so we'll map back the correct network for the given
	// address. This ensures that we can listen on the correct interface
	// (IPv4 vs IPv6).
	case *net.TCPAddr:
		if addr.IP.To4() != nil {
			return "tcp4"
		}
		return "tcp6"

	default:
		return addr.Network()
	}
}