// runLoop processes the queue of prunable images until empty. It makes the workers busy and updates the graph
// with each change.
770141
func (p *pruner) runLoop(
	jobChan chan<- *Job,
	resultChan <-chan JobResult,
) (deletions []Deletion, failures []Failure) {
	imgUpdateChan := p.imageWatcher.ResultChan()
	isUpdateChan := p.imageStreamWatcher.ResultChan()
	for {
		// make workers busy
		for len(p.processedImages) < p.numWorkers {
			job, blocked := p.getNextJob()
			if blocked {
				break
			}
			if job == nil {
				if len(p.processedImages) == 0 {
					return
				}
				break
			}
			jobChan <- job
			p.processedImages[job.Image] = job
		}

		select {
		case res := <-resultChan:
			p.updateGraphWithResult(&res)
			for _, deletion := range res.Deletions {
				deletions = append(deletions, deletion)
			}
			for _, failure := range res.Failures {
				failures = append(failures, failure)
			}
			delete(p.processedImages, res.Job.Image)
		case <-isUpdateChan:
			// TODO: fix gonum/graph to not reuse IDs of deleted nodes and reenable event handling
			//p.handleImageStreamEvent(event)
		case <-imgUpdateChan:
			// TODO: fix gonum/graph to not reuse IDs of deleted nodes and reenable event handling
			//p.handleImageEvent(event)
		}
	}
}
525347
func (q *ConvexHullQuery) AddLoop(l *Loop) {
	q.bound = q.bound.Union(l.RectBound())
	if l.isEmptyOrFull() {
		return
	}
	q.points = append(q.points, l.vertices...)
}
883683
func (o *GetOptions) Run(f cmdutil.Factory, cmd *cobra.Command, args []string) error {
	if len(o.Raw) > 0 {
		return o.raw(f)
	}
	if o.Watch || o.WatchOnly {
		return o.watch(f, cmd, args)
	}

	chunkSize := o.ChunkSize
	if o.Sort {
		// TODO(juanvallejo): in the future, we could have the client use chunking
		// to gather all results, then sort them all at the end to reduce server load.
		chunkSize = 0
	}

	r := f.NewBuilder().
		Unstructured().
		NamespaceParam(o.Namespace).DefaultNamespace().AllNamespaces(o.AllNamespaces).
		FilenameParam(o.ExplicitNamespace, &o.FilenameOptions).
		LabelSelectorParam(o.LabelSelector).
		FieldSelectorParam(o.FieldSelector).
		ExportParam(o.Export).
		RequestChunksOf(chunkSize).
		ResourceTypeOrNameArgs(true, args...).
		ContinueOnError().
		Latest().
		Flatten().
		TransformRequests(o.transformRequests).
		Do()

	if o.IgnoreNotFound {
		r.IgnoreErrors(kapierrors.IsNotFound)
	}
	if err := r.Err(); err != nil {
		return err
	}

	if !o.IsHumanReadablePrinter {
		return o.printGeneric(r)
	}

	allErrs := []error{}
	errs := sets.NewString()
	infos, err := r.Infos()
	if err != nil {
		allErrs = append(allErrs, err)
	}
	printWithKind := multipleGVKsRequested(infos)

	objs := make([]runtime.Object, len(infos))
	for ix := range infos {
		objs[ix] = infos[ix].Object
	}

	sorting, err := cmd.Flags().GetString("sort-by")
	if err != nil {
		return err
	}

	var positioner OriginalPositioner
	if o.Sort {
		sorter := NewRuntimeSorter(objs, sorting)
		if err := sorter.Sort(); err != nil {
			return err
		}
		positioner = sorter
	}

	var printer printers.ResourcePrinter
	var lastMapping *meta.RESTMapping

	// track if we write any output
	trackingWriter := &trackingWriterWrapper{Delegate: o.Out}

	w := utilprinters.GetNewTabWriter(trackingWriter)
	for ix := range objs {
		var mapping *meta.RESTMapping
		var info *resource.Info
		if positioner != nil {
			info = infos[positioner.OriginalPosition(ix)]
			mapping = info.Mapping
		} else {
			info = infos[ix]
			mapping = info.Mapping
		}

		printWithNamespace := o.AllNamespaces

		if mapping != nil && mapping.Scope.Name() == meta.RESTScopeNameRoot {
			printWithNamespace = false
		}

		if shouldGetNewPrinterForMapping(printer, lastMapping, mapping) {
			w.Flush()
			w.SetRememberedWidths(nil)

			// TODO: this doesn't belong here
			// add linebreak between resource groups (if there is more than one)
			// skip linebreak above first resource group
			if lastMapping != nil && !o.NoHeaders {
				fmt.Fprintln(o.ErrOut)
			}

			printer, err = o.ToPrinter(mapping, printWithNamespace, printWithKind)
			if err != nil {
				if !errs.Has(err.Error()) {
					errs.Insert(err.Error())
					allErrs = append(allErrs, err)
				}
				continue
			}

			lastMapping = mapping
		}

		// ensure a versioned object is passed to the custom-columns printer
		// if we are using OpenAPI columns to print
		if o.PrintWithOpenAPICols {
			printer.PrintObj(info.Object, w)
			continue
		}

		internalObj, err := legacyscheme.Scheme.ConvertToVersion(info.Object, info.Mapping.GroupVersionKind.GroupKind().WithVersion(runtime.APIVersionInternal).GroupVersion())
		if err != nil {
			// if there's an error, try to print what you have (mirrors old behavior).
			klog.V(1).Info(err)
			printer.PrintObj(info.Object, w)
		} else {
			printer.PrintObj(internalObj, w)
		}
	}
	w.Flush()
	if trackingWriter.Written == 0 && !o.IgnoreNotFound && len(allErrs) == 0 {
		// if we wrote no output, and had no errors, and are not ignoring NotFound, be sure we output something
		fmt.Fprintln(o.ErrOut, "No resources found.")
	}
	return utilerrors.NewAggregate(allErrs)
}
254668
func (c *Client) runAllocs(update *allocUpdates) {
	// Get the existing allocs
	c.allocLock.RLock()
	existing := make(map[string]uint64, len(c.allocs))
	for id, ar := range c.allocs {
		existing[id] = ar.Alloc().AllocModifyIndex
	}
	c.allocLock.RUnlock()

	// Diff the existing and updated allocations
	diff := diffAllocs(existing, update)
	c.logger.Debug("allocation updates", "added", len(diff.added), "removed", len(diff.removed),
		"updated", len(diff.updated), "ignored", len(diff.ignore))

	errs := 0

	// Remove the old allocations
	for _, remove := range diff.removed {
		c.removeAlloc(remove)
	}

	// Update the existing allocations
	for _, update := range diff.updated {
		c.logger.Trace("updating alloc", "alloc_id", update.ID, "index", update.AllocModifyIndex)
		c.updateAlloc(update)
	}

	// Make room for new allocations before running
	if err := c.garbageCollector.MakeRoomFor(diff.added); err != nil {
		c.logger.Error("error making room for new allocations", "error", err)
		errs++
	}

	// Start the new allocations
	for _, add := range diff.added {
		migrateToken := update.migrateTokens[add.ID]
		if err := c.addAlloc(add, migrateToken); err != nil {
			c.logger.Error("error adding alloc", "error", err, "alloc_id", add.ID)
			errs++
			// We mark the alloc as failed and send an update to the server
			// We track the fact that creating an allocrunner failed so that we don't send updates again
			if add.ClientStatus != structs.AllocClientStatusFailed {
				c.handleInvalidAllocs(add, err)
			}
		}
	}

	// Trigger the GC once more now that new allocs are started that could
	// have caused thresholds to be exceeded
	c.garbageCollector.Trigger()
	c.logger.Debug("allocation updates applied", "added", len(diff.added), "removed", len(diff.removed),
		"updated", len(diff.updated), "ignored", len(diff.ignore), "errors", errs)
}
415087
func (c *Client) run() {
	err := c.readLoop()
	if err != nil {
		c.LocalKite.Log.Debug("readloop err: %s", err)
	}

	// falls here when connection disconnects
	c.callOnDisconnectHandlers()

	// let others know that the client has disconnected
	c.disconnectMu.Lock()
	if c.disconnect != nil {
		close(c.disconnect)
		c.disconnect = nil
	}
	c.disconnectMu.Unlock()

	if c.reconnect() {
		// we override it so it doesn't get selected next time. Because we are
		// redialing, so after redial if a new method is called, the disconnect
		// channel is being read and the local "disconnect" message will be the
		// final response. This shouldn't be happen for redials.
		c.disconnectMu.Lock()
		c.disconnect = make(chan struct{}, 1)
		c.disconnectMu.Unlock()
		go c.dialForever(nil)
	}
}
1282953
func (i *ImagesList) Run(out io.Writer) error {
	imgs := images.GetControlPlaneImages(&i.cfg.ClusterConfiguration)
	for _, img := range imgs {
		fmt.Fprintln(out, img)
	}

	return nil
}
875054
func (pool *serverPool) eventLoop() {
	lookupCnt := 0
	var convTime mclock.AbsTime
	if pool.discSetPeriod != nil {
		pool.discSetPeriod <- time.Millisecond * 100
	}

	// disconnect updates service quality statistics depending on the connection time
	// and disconnection initiator.
	disconnect := func(req *disconnReq, stopped bool) {
		// Handle peer disconnection requests.
		entry := req.entry
		if entry.state == psRegistered {
			connAdjust := float64(mclock.Now()-entry.regTime) / float64(targetConnTime)
			if connAdjust > 1 {
				connAdjust = 1
			}
			if stopped {
				// disconnect requested by ourselves.
				entry.connectStats.add(1, connAdjust)
			} else {
				// disconnect requested by server side.
				entry.connectStats.add(connAdjust, 1)
			}
		}
		entry.state = psNotConnected

		if entry.knownSelected {
			pool.knownSelected--
		} else {
			pool.newSelected--
		}
		pool.setRetryDial(entry)
		pool.connWg.Done()
		close(req.done)
	}

	for {
		select {
		case entry := <-pool.timeout:
			if !entry.removed {
				pool.checkDialTimeout(entry)
			}

		case entry := <-pool.enableRetry:
			if !entry.removed {
				entry.delayedRetry = false
				pool.updateCheckDial(entry)
			}

		case adj := <-pool.adjustStats:
			switch adj.adjustType {
			case pseBlockDelay:
				adj.entry.delayStats.add(float64(adj.time), 1)
			case pseResponseTime:
				adj.entry.responseStats.add(float64(adj.time), 1)
				adj.entry.timeoutStats.add(0, 1)
			case pseResponseTimeout:
				adj.entry.timeoutStats.add(1, 1)
			}

		case node := <-pool.discNodes:
			if pool.trustedNodes[node.ID()] == nil {
				entry := pool.findOrNewNode(node)
				pool.updateCheckDial(entry)
			}

		case conv := <-pool.discLookups:
			if conv {
				if lookupCnt == 0 {
					convTime = mclock.Now()
				}
				lookupCnt++
				if pool.fastDiscover && (lookupCnt == 50 || time.Duration(mclock.Now()-convTime) > time.Minute) {
					pool.fastDiscover = false
					if pool.discSetPeriod != nil {
						pool.discSetPeriod <- time.Minute
					}
				}
			}

		case req := <-pool.connCh:
			if pool.trustedNodes[req.p.ID()] != nil {
				// ignore trusted nodes
				req.result <- nil
			} else {
				// Handle peer connection requests.
				entry := pool.entries[req.p.ID()]
				if entry == nil {
					entry = pool.findOrNewNode(req.node)
				}
				if entry.state == psConnected || entry.state == psRegistered {
					req.result <- nil
					continue
				}
				pool.connWg.Add(1)
				entry.peer = req.p
				entry.state = psConnected
				addr := &poolEntryAddress{
					ip:       req.node.IP(),
					port:     uint16(req.node.TCP()),
					lastSeen: mclock.Now(),
				}
				entry.lastConnected = addr
				entry.addr = make(map[string]*poolEntryAddress)
				entry.addr[addr.strKey()] = addr
				entry.addrSelect = *newWeightedRandomSelect()
				entry.addrSelect.update(addr)
				req.result <- entry
			}

		case req := <-pool.registerCh:
			// Handle peer registration requests.
			entry := req.entry
			entry.state = psRegistered
			entry.regTime = mclock.Now()
			if !entry.known {
				pool.newQueue.remove(entry)
				entry.known = true
			}
			pool.knownQueue.setLatest(entry)
			entry.shortRetry = shortRetryCnt
			close(req.done)

		case req := <-pool.disconnCh:
			// Handle peer disconnection requests.
			disconnect(req, req.stopped)

		case <-pool.quit:
			if pool.discSetPeriod != nil {
				close(pool.discSetPeriod)
			}

			// Spawn a goroutine to close the disconnCh after all connections are disconnected.
			go func() {
				pool.connWg.Wait()
				close(pool.disconnCh)
			}()

			// Handle all remaining disconnection requests before exit.
			for req := range pool.disconnCh {
				disconnect(req, true)
			}
			pool.saveNodes()
			pool.wg.Done()
			return
		}
	}
}
295244
func (s *Session) recvLoop() {
	var hdr rawHeader

	for {
		for atomic.LoadInt32(&s.bucket) <= 0 && !s.IsClosed() {
			select {
			case <-s.bucketNotify:
			case <-s.die:
				return
			}
		}

		// read header first
		if _, err := io.ReadFull(s.conn, hdr[:]); err == nil {
			atomic.StoreInt32(&s.dataReady, 1)
			if hdr.Version() != version {
				s.Close()
				return
			}
			sid := hdr.StreamID()
			switch hdr.Cmd() {
			case cmdNOP:
			case cmdSYN:
				s.streamLock.Lock()
				if _, ok := s.streams[sid]; !ok {
					stream := newStream(sid, s.config.MaxFrameSize, s)
					s.streams[sid] = stream
					select {
					case s.chAccepts <- stream:
					case <-s.die:
					}
				}
				s.streamLock.Unlock()
			case cmdFIN:
				s.streamLock.Lock()
				if stream, ok := s.streams[sid]; ok {
					stream.markRST()
					stream.notifyReadEvent()
				}
				s.streamLock.Unlock()
			case cmdPSH:
				if hdr.Length() > 0 {
					newbuf := make([]byte, hdr.Length())
					if written, err := io.ReadFull(s.conn, newbuf); err == nil {
						s.streamLock.Lock()
						if stream, ok := s.streams[sid]; ok {
							stream.pushBytes(newbuf)
							atomic.AddInt32(&s.bucket, -int32(written))
							stream.notifyReadEvent()
						}
						s.streamLock.Unlock()
					} else {
						s.Close()
						return
					}
				}
			default:
				s.Close()
				return
			}
		} else {
			s.Close()
			return
		}
	}
}
374901
func (api *PublicDownloaderAPI) eventLoop() {
	var (
		sub               = api.mux.Subscribe(StartEvent{}, DoneEvent{}, FailedEvent{})
		syncSubscriptions = make(map[chan interface{}]struct{})
	)

	for {
		select {
		case i := <-api.installSyncSubscription:
			syncSubscriptions[i] = struct{}{}
		case u := <-api.uninstallSyncSubscription:
			delete(syncSubscriptions, u.c)
			close(u.uninstalled)
		case event := <-sub.Chan():
			if event == nil {
				return
			}

			var notification interface{}
			switch event.Data.(type) {
			case StartEvent:
				notification = &SyncingResult{
					Syncing: true,
					Status:  api.d.Progress(),
				}
			case DoneEvent, FailedEvent:
				notification = false
			}
			// broadcast
			for c := range syncSubscriptions {
				c <- notification
			}
		}
	}
}
1269418
func (p *gateway) runningLoop(ctx context.Context) (err error) {
	// getEtcdStopCh returns the gateway's etcd server's stop notify channel or returns a blocking channel if etcd is nil
	getEtcdStopCh := func() <-chan struct{} {
		if p.etcdServer != nil && p.etcdServer.Server != nil {
			return p.etcdServer.Server.StopNotify()
		}
		return make(chan struct{})
	}
	// main loop
	for {
		select {
		case <-ctx.Done():
			return err
		case <-p.signalChan: // shutdown the gateway if the gateway is signaled
			return p.stop(ctx)
		case <-getEtcdStopCh(): // shutdown the gateway if the etcd server goes down
			// TODO: try to relaod the etcd server some # of times if it fails
			//  instead of shutting down the whole gateway.  There is an err chan
			//  on etcd server that we could use to identify if we've errored out
			// signal to the running routine to close the gateway
			p.signalChan <- syscall.SIGTERM
			p.logger.Log(log.Msg, "etcd server has stopped")
		}
	}
}
942981
func (c *ImageStreamController) Run(workers int, stopCh <-chan struct{}) {
	defer utilruntime.HandleCrash()
	defer c.queue.ShutDown()

	klog.Infof("Starting image stream controller")

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

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

	metrics.InitializeImportCollector(false, c.importCounter.Collect)

	<-stopCh
	klog.Infof("Shutting down image stream controller")
}
602423
func (n *NodeIPTables) syncLoop() {
	t := time.NewTicker(n.syncPeriod)
	defer t.Stop()
	for {
		<-t.C
		klog.V(6).Infof("Periodic openshift iptables sync")
		err := n.syncIPTableRules()
		if err != nil {
			utilruntime.HandleError(fmt.Errorf("Syncing openshift iptables failed: %v", err))
		}
	}
}
1076993
func goLoop(lf LoopFunc, rf RecoverFunc, o Observable, s *sentinel, d string) *loop {
	l := &loop{
		descr:    d,
		loopF:    lf,
		recoverF: rf,
		startedC: make(chan struct{}),
		stopC:    make(chan struct{}),
		doneC:    make(chan struct{}),
		owner:    o,
		sentinel: s,
	}
	// Check owner, at least we should own ourself.
	if l.owner == nil {
		l.owner = l
	}
	// Start the loop.
	l.logf(false, "loop %q starts", l)
	go l.run()
	<-l.startedC
	return l
}
930976
func (w *Worker) handlerLoop() {

	for task := range w.workchan {
		handler, found := w.getHandler(task)
		if !found {
			log.Printf("tasque: ERROR no handler for task %s", task.Name)
			continue
		}

		// call a safe runner func
		err := w.runHandler(handler, task)
		if err != nil {
			log.Println("tasque: Error handling task %s: %s", task.Name, err)
		} else {
			client, err := w.pool.Get()
			if err != nil {
				log.Println("Error getting client: %s", err)
			} else {
				client.Ack(task.JobId())
				client.Close()
			}

		}
	}
}
998305
func (loop *Loop) Run(path string) {
	if path == "" {
		path = "./"
	}
	// lifecycle
	loop.lifecycle = jobsync.NewLifecycle(workers.DefaultTimeout, true)
	if loop.scope != nil {
		loop.scope.On(app.ErrorEvent, loop.KillSlot)
		loop.scope.On(app.KillEvent, loop.KillSlot)
	}
	// producer
	producentMaxJob := loop.loopData.Producents
	if producentMaxJob == 0 || producentMaxJob > workers.MaxJob {
		producentMaxJob = workers.MaxJob
	}
	producerPool := jobsync.NewPool(producentMaxJob)
	producer := &Producer{
		lifecycle: loop.lifecycle,
		pool:      producerPool,
		loopData:  loop.loopData,
		path:      path,
	}
	producerCounter := producerPool.Add(1)
	if producerCounter != 1 {
		panic("filesystem.Loop it is not possible to start producer job")
	}
	go producer.Loop()
	// consumer
	consumerMaxJob := loop.loopData.Consumers
	if consumerMaxJob == 0 || consumerMaxJob > workers.MaxJob {
		consumerMaxJob = workers.MaxJob
	}
	loop.consumerPool = jobsync.NewPool(consumerMaxJob)
	consumer := &Consumer{
		lifecycle: loop.lifecycle,
		pool:      loop.consumerPool,
		loopData:  loop.loopData,
	}
	consumerCounter := loop.consumerPool.Add(workers.MaxJob)
	for i := 0; i < consumerCounter; i++ {
		go consumer.Loop()
	}
	// lifecycle
	go func() {
		producerPool.Wait()
		loop.lifecycle.NextStep(StepClose)
		close(loop.loopData.chans.dirChan)
		close(loop.loopData.chans.fileChan)
	}()
}
1117289
func (c *Client) readLoop() error {
	for {
		p, err := c.receiveData()

		c.LocalKite.Log.Debug("readloop received: %s %v", p, err)

		if err != nil {
			return err
		}

		msg, fn, err := c.processMessage(p)
		if err != nil {
			if _, ok := err.(dnode.CallbackNotFoundError); !ok {
				c.LocalKite.Log.Warning("error processing message err: %s message: %s", err, msg)
			}
		}

		switch v := fn.(type) {
		case *Method: // invoke method
			if c.Concurrent {
				go c.runMethod(v, msg.Arguments)
			} else {
				c.runMethod(v, msg.Arguments)
			}
		case func(*dnode.Partial): // invoke callback
			if c.Concurrent && c.ConcurrentCallbacks {
				go c.runCallback(v, msg.Arguments)
			} else {
				c.runCallback(v, msg.Arguments)
			}
		}
	}
}
1187902
func (gb *GraphBuilder) Run(stopCh <-chan struct{}) {
	klog.Infof("GraphBuilder running")
	defer klog.Infof("GraphBuilder stopping")

	// Set up the stop channel.
	gb.monitorLock.Lock()
	gb.stopCh = stopCh
	gb.running = true
	gb.monitorLock.Unlock()

	// Start monitors and begin change processing until the stop channel is
	// closed.
	gb.startMonitors()
	wait.Until(gb.runProcessGraphChanges, 1*time.Second, stopCh)

	// Stop any running monitors.
	gb.monitorLock.Lock()
	defer gb.monitorLock.Unlock()
	monitors := gb.monitors
	stopped := 0
	for _, monitor := range monitors {
		if monitor.stopCh != nil {
			stopped++
			close(monitor.stopCh)
		}
	}

	// reset monitors so that the graph builder can be safely re-run/synced.
	gb.monitors = nil
	klog.Infof("stopped %d of %d monitors", stopped, len(monitors))
}