// loop is somewhat out of the ordinary, because an api.Connection
// *does* maintain an internal workeresque heartbeat goroutine, but it
// doesn't implement Worker.
186833
func (w *apiConnWorker) loop() (err error) {
	// TODO(fwereade): we should make this rational at some point.

	defer func() {
		// Since we can't tell for sure what error killed the connection, any
		// error out of Close is more important and relevant than any error we
		// might return in the loop.
		if closeErr := w.conn.Close(); closeErr != nil {
			err = closeErr
		}
	}()

	// Note that we should never return a nil error from this loop. If we're
	// shut down deliberately we should return ErrDying, to be overwritten by
	// any non-nil Close error and otherwise not perturb the tomb; and if the
	// connection closes on its own there's surely *something* wrong even if
	// there's no error reported from Close. (sample problem: *someone else*
	// closed the conn that we're meant to be in control of).
	for {
		select {
		case <-w.tomb.Dying():
			return tomb.ErrDying
		case <-w.conn.Broken():
			return errors.New("api connection broken unexpectedly")
		}
	}
}
45940
func (m *Manager) loop() {

	var once sync.Once

	for {
		select {
		case request := <-m.requests:
			switch request.op {
			case managerOpNotify:
				var failed bool
				for _, svc := range m.services {
					if svc.Notify != nil && svc.IsRunning() {
						if err := svc.Notify(svc, request.val); err != nil {
							log.WithFields(logrus.Fields{
								"isvc": svc.Name,
							}).WithError(err).Error("Unable to notify internal service")
							failed = true
						}
					}
				}
				if failed {
					request.response <- ErrNotifyFailed
				} else {
					request.response <- nil
				}
			case managerOpExit:
				request.response <- nil
				return // this will exit the loop()
			case managerOpStart:
				var err error
				once.Do(func() {
					if err = m.loadImages(); err != nil {
						return
					} else if err = m.allImagesExist(); err != nil {
						return
					}
				})
				if err != nil {
					request.response <- err
					continue
				}

				// Start each group of services in group-number order
				unstartedServices := 0
				for _, group := range m.orderedStartGroups() {
					unstartedServices += m.startServiceGroup(group)
				}

				if unstartedServices > 0 {
					request.response <- StartError(unstartedServices)
				} else {
					request.response <- nil
				}

			case managerOpStop:
				// track the number of services that haven't stopped
				var noStop = make([]int, len(m.services))

				// stop services in parallel
				var wg sync.WaitGroup
				index := 0
				for name, svc := range m.services {
					if svc.IsRunning() {
						wg.Add(1)
						go func(svc *IService, i int) {
							defer wg.Done()
							if err := svc.Stop(); err != nil {
								log.WithFields(logrus.Fields{
									"isvc": svc.Name,
								}).WithError(err).Error("Unable to stop internal service")
								noStop[i] = 1
								return
							}
						}(m.services[name], index)
					}
					index++
				}
				wg.Wait()
				count := 0
				for _, i := range noStop {
					count += i
				}
				if count > 0 {
					request.response <- StopError(count)
				} else {
					request.response <- nil
				}
			case managerOpRegisterContainer:
				svc, ok := request.val.(*IService)
				if !ok {
					request.response <- ErrManagerUnknownArg
				} else {
					m.services[svc.Name] = svc
					m.addServiceToStartGroup(svc)
					request.response <- nil
				}
			case managerOpInit:
				request.response <- nil
			default:
				request.response <- ErrManagerUnknownOp
			}
		}
	}
}
1063315
func (pt *pingTimeout) loop() error {
	for {
		select {
		case <-pt.tomb.Dying():
			return tomb.ErrDying
		case <-pt.reset:
		case <-pt.clock.After(pt.timeout):
			go pt.action()
			return errors.New("ping timeout")
		}
	}
}
1107030
func (eq *eventQueue) run() {
	for {
		block := eq.next()

		if block == nil {
			return // nil block means event queue is closed.
		}

		if err := eq.sink.Write(block...); err != nil {
			logrus.Warnf("eventqueue: error writing events to %v, these events will be lost: %v", eq.sink, err)
		}

		for _, listener := range eq.listeners {
			listener.egress(block...)
		}
	}
}
1016733
func (l *loop) run() {
	l.status = Running
	// Finalize the loop.
	defer l.finalizeTermination()
	// Create a loop wrapper containing the recovering control.
	loopWrapper := func() {
		defer func() {
			// Check for recovering.
			if reason := recover(); reason != nil {
				l.checkTermination(reason)
			}
		}()
		l.checkTermination(l.loopF(l))
	}
	// Now start running the loop wrappr.
	l.startedC <- struct{}{}
	for l.status == Running {
		loopWrapper()
	}
}
460458
func (manager *Manager) loop() error {
	if collector, ok := manager.config.Store.(prometheus.Collector); ok && manager.config.PrometheusRegisterer != nil {
		// The store implements the collector interface, but the lease.Store
		// does not expose those.
		_ = manager.config.PrometheusRegisterer.Register(collector)
		defer manager.config.PrometheusRegisterer.Unregister(collector)
	}

	defer manager.wg.Wait()
	blocks := make(blocks)
	manager.setupInitialTimer()
	for {
		if err := manager.choose(blocks); err != nil {
			manager.config.Logger.Tracef("[%s] exiting main loop with error: %v", manager.logContext, err)
			return errors.Trace(err)
		}
	}
}
745224
func Loop(lmid *[curl.StateSize]uint64, hmid *[curl.StateSize]uint64, m int, cancelled *bool, checkFun CheckFunc, loopCnt int) (nonce Trits, rate int64, foundIndex int) {
	var lcpy, hcpy [curl.StateSize]uint64
	var i int64

	for i = 0; !*cancelled; i++ {
		copy(lcpy[:], lmid[:])
		copy(hcpy[:], hmid[:])
		transform64(&lcpy, &hcpy, loopCnt)

		if n := checkFun(&lcpy, &hcpy, m); n >= 0 {
			nonce := seri(lmid, hmid, uint(n))
			return nonce, i * 64, n
		}
		incr(lmid, hmid)
	}
	return nil, i * 64, -1
}
292169
func (s *DefaultServer) Loop() {
	defer func() {
		if r := recover(); r != nil {
			if r != io.EOF {
				log.Println("Recovered from runtime panic:", r)
				log.Println("Panic location: ", identifyPanic())
			}

			abort(s.conns, fmt.Errorf("Runtime panic: %v", r))
		}
	}()

	for {
		request, reqType, start, err := s.rp.Parse()
		if err != nil {
			if err == common.ErrBadRequest ||
				err == common.ErrBadLength ||
				err == common.ErrBadFlags ||
				err == common.ErrBadExptime {
				s.orca.Error(nil, common.RequestUnknown, err)
				continue
			} else {
				// Otherwise IO error. Abort!
				abort(s.conns, err)
				return
			}
		}

		metrics.IncCounter(MetricCmdTotal)

		// TODO: handle nil
		switch reqType {
		case common.RequestSet:
			metrics.IncCounter(MetricCmdSet)
			err = s.orca.Set(request.(common.SetRequest))
		case common.RequestAdd:
			metrics.IncCounter(MetricCmdAdd)
			err = s.orca.Add(request.(common.SetRequest))
		case common.RequestReplace:
			metrics.IncCounter(MetricCmdReplace)
			err = s.orca.Replace(request.(common.SetRequest))
		case common.RequestAppend:
			metrics.IncCounter(MetricCmdAppend)
			err = s.orca.Append(request.(common.SetRequest))
		case common.RequestPrepend:
			metrics.IncCounter(MetricCmdPrepend)
			err = s.orca.Prepend(request.(common.SetRequest))
		case common.RequestDelete:
			metrics.IncCounter(MetricCmdDelete)
			err = s.orca.Delete(request.(common.DeleteRequest))
		case common.RequestTouch:
			metrics.IncCounter(MetricCmdTouch)
			err = s.orca.Touch(request.(common.TouchRequest))
		case common.RequestGet:
			metrics.IncCounter(MetricCmdGet)
			err = s.orca.Get(request.(common.GetRequest))
		case common.RequestGetE:
			metrics.IncCounter(MetricCmdGetE)
			err = s.orca.GetE(request.(common.GetRequest))
		case common.RequestGat:
			metrics.IncCounter(MetricCmdGat)
			err = s.orca.Gat(request.(common.GATRequest))
		case common.RequestNoop:
			metrics.IncCounter(MetricCmdNoop)
			err = s.orca.Noop(request.(common.NoopRequest))
		case common.RequestQuit:
			metrics.IncCounter(MetricCmdQuit)
			s.orca.Quit(request.(common.QuitRequest))
			abort(s.conns, err)
			return
		case common.RequestVersion:
			metrics.IncCounter(MetricCmdVersion)
			err = s.orca.Version(request.(common.VersionRequest))
		case common.RequestUnknown:
			metrics.IncCounter(MetricCmdUnknown)
			err = s.orca.Unknown(request)
		}

		if err != nil {
			if common.IsAppError(err) {
				if err != common.ErrKeyNotFound {
					metrics.IncCounter(MetricErrAppError)
				}
				s.orca.Error(request, reqType, err)
			} else {
				metrics.IncCounter(MetricErrUnrecoverable)
				abort(s.conns, err)
				return
			}
		}

		dur := timer.Since(start)
		switch reqType {
		case common.RequestSet:
			metrics.ObserveHist(HistSet, dur)
		case common.RequestAdd:
			metrics.ObserveHist(HistAdd, dur)
		case common.RequestReplace:
			metrics.ObserveHist(HistReplace, dur)
		case common.RequestDelete:
			metrics.ObserveHist(HistDelete, dur)
		case common.RequestTouch:
			metrics.ObserveHist(HistTouch, dur)
		case common.RequestGet:
			metrics.ObserveHist(HistGet, dur)
		case common.RequestGetE:
			metrics.ObserveHist(HistGetE, dur)
		case common.RequestGat:
			metrics.ObserveHist(HistGat, dur)
		}
	}
}
1107480
func (h *Heartbeat) Run() error {
	defer func() {
		h.reset(HeartbeatStateInit)
		h.checkTicker.Stop()
	}()
	for {
		if err := h.fetchAndAnnounce(); err != nil {
			h.Warningf("Heartbeat failed %v.", err)
		}
		select {
		case <-h.checkTicker.C:
		case <-h.sendC:
			h.Debugf("Asked check out of cycle")
		case <-h.cancelCtx.Done():
			h.Debugf("Heartbeat exited.")
			return nil
		}
	}
}
931091
func (idx *TruncIndex) Iterate(handler func(id string)) {
	idx.Lock()
	defer idx.Unlock()
	idx.trie.Visit(func(prefix patricia.Prefix, item patricia.Item) error {
		handler(string(prefix))
		return nil
	})
}
222553
func (sync *NodeSync) Loop(done chan struct{}) {
	klog.V(2).Infof("Starting sync loop for node %q", sync.nodeName)

	defer func() {
		if done != nil {
			close(done)
		}
	}()

	timeout := sync.c.ResyncTimeout()
	delayTimer := time.NewTimer(timeout)
	klog.V(4).Infof("Resync node %q in %v", sync.nodeName, timeout)

	for {
		select {
		case op, more := <-sync.opChan:
			if !more {
				klog.V(2).Infof("Stopping sync loop")
				return
			}
			sync.c.ReportResult(op.run(sync))
			if !delayTimer.Stop() {
				<-delayTimer.C
			}
		case <-delayTimer.C:
			klog.V(4).Infof("Running resync for node %q", sync.nodeName)
			sync.c.ReportResult((&updateOp{}).run(sync))
		}

		timeout := sync.c.ResyncTimeout()
		delayTimer.Reset(timeout)
		klog.V(4).Infof("Resync node %q in %v", sync.nodeName, timeout)
	}
}
983394
func (forwarder *MetricForwarder) loop() {
	routes := []rest.Route{
		rest.Route{
			HttpMethod: "POST",
			PathExp:    "/api/metrics/store",
			Func:       postAPIMetricsStore(forwarder.metricsRedirectURL),
		},
	}

	handler := rest.ResourceHandler{}
	handler.SetRoutes(routes...)
	http.Serve(*forwarder.listener, &handler)
}
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)
	}()
}
908958
func (w *Worker) Run() {

	// TODO: make this dynamic
	for i := 0; i < w.numGoroutines; i++ {
		log.Println("Starting handler routine ", i+1)
		go w.handlerLoop()
	}

	for {
		client, err := w.pool.Get()
		defer client.Close()
		if err != nil {
			log.Println("tasque: could not get client")
			select {
			case <-w.stopch:
				return
			case <-time.After(100 * time.Millisecond):
			}

		}

		for {
			select {
			case <-w.stopch:
				return
			default:
			}

			job, err := client.Get(getTimeout, w.channels...)

			if err == nil {

				if job.Data != nil {

					task := new(Task)
					if err := json.Unmarshal(job.Data, task); err != nil {
						log.Printf("tasque: Could not unmarshal message data: %s", err)
						continue
					}
					task.jobId = job.Id()

					select {
					case w.workchan <- task:
					case <-w.stopch:
						return
					}
				}

			} else {
				log.Printf("Error receiving: %s", err)
				client.Close()
				break
			}
		}
	}

}
1007453
func (b *BackgroundEphemeralPurger) loop(shutdownCh chan struct{}) {
	bgctx := context.Background()
	b.Debug(bgctx, "loop: starting for %s", b.uid)

	for {
		select {
		case <-b.purgeTimer.C:
			b.Debug(bgctx, "loop: looping for %s", b.uid)
			b.queuePurges(bgctx)
		case <-shutdownCh:
			b.Debug(bgctx, "loop: shutting down for %s", b.uid)
			return
		}
	}
}
688173
func (ListAsyncJobs) Each(resp interface{}, callback IterateItemFunc) {
	items, ok := resp.(*ListAsyncJobsResponse)
	if !ok {
		callback(nil, fmt.Errorf("wrong type, ListAsyncJobsResponse was expected, got %T", resp))
		return
	}

	for i := range items.AsyncJob {
		if !callback(&items.AsyncJob[i], nil) {
			break
		}
	}
}
10733
func (feed *UprFeed) run() {
	retryInterval := initialRetryInterval
	bucketOK := true
	for {
		// Connect to the UPR feed of each server node:
		if bucketOK {
			// Run until one of the sub-feeds fails:
			select {
			case <-feed.killSwitch:
			case <-feed.quit:
				return
			}
			//feed.closeNodeFeeds()
			retryInterval = initialRetryInterval
		}

		if feed.closing == true {
			// we have been asked to shut down
			return
		}

		// On error, try to refresh the bucket in case the list of nodes changed:
		logging.Infof("go-couchbase: UPR connection lost; reconnecting to bucket %q in %v",
			feed.bucket.Name, retryInterval)

		if err := feed.bucket.Refresh(); err != nil {
			// if we fail to refresh the bucket, exit the feed
			// MB-14917
			logging.Infof("Unable to refresh bucket %s ", err.Error())
			close(feed.output)
			feed.outputClosed = true
			feed.closeNodeFeeds()
			return
		}

		// this will only connect to nodes that are not connected or changed
		// user will have to reconnect the stream
		err := feed.connectToNodes()
		if err != nil {
			logging.Infof("Unable to connect to nodes..exit ")
			close(feed.output)
			feed.outputClosed = true
			feed.closeNodeFeeds()
			return
		}
		bucketOK = err == nil

		select {
		case <-time.After(retryInterval):
		case <-feed.quit:
			return
		}
		if retryInterval *= 2; retryInterval > maximumRetryInterval {
			retryInterval = maximumRetryInterval
		}
	}
}