// Only run on the nodes
183673
func (node *OsdnNode) watchSubnets() {
	subnets := make(map[string]*osapi.HostSubnet)
	eventQueue := node.registry.RunEventQueue(HostSubnets)

	for {
		eventType, obj, err := eventQueue.Pop()
		if err != nil {
			utilruntime.HandleError(fmt.Errorf("EventQueue failed for subnets: %v", err))
			return
		}
		hs := obj.(*osapi.HostSubnet)

		if hs.HostIP == node.localIP {
			continue
		}

		log.V(5).Infof("Watch %s event for HostSubnet %q", strings.Title(string(eventType)), hs.ObjectMeta.Name)
		switch eventType {
		case watch.Added, watch.Modified:
			oldSubnet, exists := subnets[string(hs.UID)]
			if exists {
				if oldSubnet.HostIP == hs.HostIP {
					continue
				} else {
					// Delete old subnet rules
					if err := node.DeleteHostSubnetRules(oldSubnet); err != nil {
						log.Error(err)
					}
				}
			}
			if err := node.registry.ValidateNodeIP(hs.HostIP); err != nil {
				log.Errorf("Ignoring invalid subnet for node %s: %v", hs.HostIP, err)
				continue
			}

			if err := node.AddHostSubnetRules(hs); err != nil {
				log.Error(err)
				continue
			}
			subnets[string(hs.UID)] = hs
		case watch.Deleted:
			delete(subnets, string(hs.UID))

			if err := node.DeleteHostSubnetRules(hs); err != nil {
				log.Error(err)
			}
		}
	}
}
1301742
func (e *PGPPurge) Run(m libkb.MetaContext) error {
	me, err := libkb.LoadMe(libkb.NewLoadUserPubOptionalArg(e.G()))
	if err != nil {
		return err
	}
	e.me = me

	// get all PGP blocks in keyring
	ring, err := m.ActiveDevice().Keyring(m)
	if err != nil {
		return err
	}
	blocks, err := ring.AllPGPBlocks()
	if err != nil {
		return err
	}

	// export each one to a file
	if err := e.exportBlocks(m, blocks); err != nil {
		return err
	}

	if e.arg.DoPurge {
		// if purge flag set, remove all PGP blocks from keyring and save it
		err = ring.RemoveAllPGPBlocks()
		if err != nil {
			return err
		}
		err = ring.Save()
		if err != nil {
			return err
		}
	}

	return nil
}
250697
func (ce *ConstraintEnforcer) Run() {
	defer close(ce.doneChan)

	watcher, cancelWatch := state.Watch(ce.store.WatchQueue(), api.EventUpdateNode{})
	defer cancelWatch()

	var (
		nodes []*api.Node
		err   error
	)
	ce.store.View(func(readTx store.ReadTx) {
		nodes, err = store.FindNodes(readTx, store.All)
	})
	if err != nil {
		log.L.WithError(err).Error("failed to check nodes for noncompliant tasks")
	} else {
		for _, node := range nodes {
			ce.rejectNoncompliantTasks(node)
		}
	}

	for {
		select {
		case event := <-watcher:
			node := event.(api.EventUpdateNode).Node
			ce.rejectNoncompliantTasks(node)
		case <-ce.stopChan:
			return
		}
	}
}
1091375
func (toks *Toks) Iterate(t *html.Node) {
	for _, tok := range *toks {
		if tok.Match(t.Type, t.DataAtom) {
			tok.Callback(t)
		}
	}
}
521826
func (c *ProjectFinalizerController) Run(stopCh <-chan struct{}, workers int) {
	defer runtime.HandleCrash()
	defer c.queue.ShutDown()

	// Wait for the stores to fill
	if !cache.WaitForCacheSync(stopCh, c.cacheSynced) {
		return
	}

	klog.V(5).Infof("Starting workers")
	for i := 0; i < workers; i++ {
		go c.worker()
	}
	<-stopCh
	klog.V(1).Infof("Shutting down")
}
860771
func (p *EvaluatePool) run(stopCh chan struct{}) {
	for {
		select {
		case req := <-p.req:
			fit, reason, err := evaluateNodePlan(req.snap, req.plan, req.nodeID)
			p.res <- evaluateResult{req.nodeID, fit, reason, err}

		case <-stopCh:
			return
		}
	}
}
907592
func (cnc *CloudNodeController) Run(stopCh <-chan struct{}) {
	defer utilruntime.HandleCrash()

	// The following loops run communicate with the APIServer with a worst case complexity
	// of O(num_nodes) per cycle. These functions are justified here because these events fire
	// very infrequently. DO NOT MODIFY this to perform frequent operations.

	// Start a loop to periodically update the node addresses obtained from the cloud
	wait.Until(cnc.UpdateNodeStatus, cnc.nodeStatusUpdateFrequency, stopCh)
}
843327
func (s *Simple) run() {
	for {
		select {
		case <-time.After(s.config.GCPeriodicity):
			s.GC()
		case <-s.halt:
			s.closeError <- nil
			// there is no cleanup required, so we just return
			return
		}
	}
}
471192
func (g *lockedNetNSGoroutine) run(f func()) {
	done := make(chan struct{})
	g.funcC <- func() {
		defer close(done)
		f()
	}
	<-done
}
236765
func (c *connectionBenchmark) run() *result {
	var err error
	if c.requestRate == 0 {
		c.elapsed, err = c.runFullThrottle()
	} else {
		c.elapsed, err = c.runRateLimited()
	}
	return &result{summary: c.summarize(), err: err}
}
1211035
func (brw *blockRetrievalWorker) run() {
	defer close(brw.doneCh)
	for {
		err := brw.HandleRequest()
		// Only io.EOF is relevant to the loop; other errors are handled in
		// FinalizeRequest
		if err == io.EOF {
			return
		}
	}
}
914467
func (r RevelHooks) Run() {
	serverLogger.Infof("There is %d hooks need to run ...", len(r))
	sort.Sort(r)
	for i, hook := range r {
		utilLog.Infof("Run the %d hook ...", i+1)
		hook.f()
	}
}
1242209
func (t *stringTree) DoAll(f func(v string) error) error {
	return t.container.root.doAll(func(dn *node) error {
		return f(dn.content.value().(string))
	})
}
1174906
func (m *cloudResourceSyncManager) Run(stopCh <-chan struct{}) {
	wait.Until(m.syncNodeAddresses, m.syncPeriod, stopCh)
}
730849
func Nodes(sel interface{}, nodes *[]*cdp.Node, opts ...QueryOption) Action {
	if nodes == nil {
		panic("nodes cannot be nil")
	}

	return QueryAfter(sel, func(ctx context.Context, n ...*cdp.Node) error {
		*nodes = n
		return nil
	}, opts...)
}
281164
func (m *Manager) DeleteAllNodes() {
	m.mutex.Lock()
	for _, entry := range m.nodes {
		entry.mutex.Lock()
		m.datapath.NodeDelete(entry.node)
		entry.mutex.Unlock()
	}
	m.nodes = map[node.Identity]*nodeEntry{}
	m.mutex.Unlock()
}
867811
func (nodeInfo *executionNode) renderNodes(w *tabwriter.Writer, wf *wfv1.Workflow, depth int, nodePrefix string, childPrefix string, outFmt string) {
	filtered, _ := filterNode(nodeInfo.getNodeStatus(wf))
	if !filtered {
		printNode(w, wf, nodeInfo.getNodeStatus(wf), depth, nodePrefix, childPrefix, outFmt)
	}
}