// Loop starts the Pulse.
777808
func (p *Pulse) Loop(id ID, pulseCh chan Update, consumerStopCh <-chan struct{}) {
	log.Infof("starting pulse for %s", id)

	// Randomize the first health-check to avoid thundering herd syndrome.
	interval := time.Duration(rng.Int63n(int64(p.interval)))

	for {
		select {
		case <-time.After(interval):
			select {
			// Recalculate metrics and statistics and send them to Context.
			case pulseCh <- Update{id, p.metrics.Update(p.driver.Check())}:
			case <-consumerStopCh:
				// prevent blocking if the consumer stops before us
			}
		case <-p.stopCh:
			log.Infof("stopping pulse for %s", id)
			pulseCh <- Update{id, p.metrics.Update(StatusRemoved)}
			return
		}

		// TODO(@kobolog): Add exponential back-offs, thresholds.
		interval = p.interval

		log.Debugf("current pulse for %s: %s", id, p.metrics.Status.String())
	}
}
380862
func (u *user) run() {
	done := false
	for !done {
		select {
		case ca := <-u.checkCh:
			u.check(ca)
		case <-u.stopCh:
			u.ca.log.Debug("Stopping user loop for %s", u)
			done = true
		case <-u.ca.shutdownCh:
			done = true
		}
	}
	u.ca.eng.Evicted(u.uid)
}
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")
		}
	}
}
561522
func (d *DockerDiscovery) Run(looper director.Looper) {
	connQuitChan := make(chan bool)

	go d.manageConnection(connQuitChan)

	go func() {
		// Loop around, process any events which came in, and
		// periodically fetch the whole container list
		looper.Loop(func() error {
			select {
			case event := <-d.events:
				if event == nil {
					// This usually happens because of a Docker restart.
					// Sleep, let us reconnect in the background, then loop.
					return nil
				}
				log.Debugf("Event: %#v\n", event)
				d.handleEvent(*event)
			case <-time.After(d.sleepInterval):
				d.getContainers()
			case <-time.After(CacheDrainInterval):
				d.containerCache.Drain(len(d.services))
			}

			return nil
		})

		// Propagate quit channel message
		close(connQuitChan)
	}()
}
360010
func Loop(count int, s StreamSeeker) Streamer {
	return &loop{
		s:       s,
		remains: count,
	}
}
1358199
func (w *Watcher) loop(period time.Duration) error {
	next := time.After(period)
	w.needSync = true
	if err := w.initLastId(); err != nil {
		return errors.Trace(err)
	}
	for {
		if w.needSync {
			if err := w.sync(); err != nil {
				// If the txn log collection overflows from underneath us,
				// the easiest cause of action to recover is to cause the
				// agen tto restart.
				if errors.Cause(err) == cappedPositionLostError {
					// Ideally we'd not import the worker package but that's
					// where all the errors are defined.
					return jworker.ErrRestartAgent
				}
				return errors.Trace(err)
			}
			w.flush()
			next = time.After(period)
		}
		select {
		case <-w.tomb.Dying():
			return errors.Trace(tomb.ErrDying)
		case <-next:
			next = time.After(period)
			w.needSync = true
		case req := <-w.request:
			w.handle(req)
			w.flush()
		}
	}
}
673717
func Loop() {
	go func() {
		p := fmt.Sprintf(":%d", Configuration.Service.Port)
		LoggingClient.Info(fmt.Sprintf("Starting Export Distro %s", p))
		messageErrors <- http.ListenAndServe(p, httpServer())
	}()

	registrations := make(map[string]*registrationInfo)

	allRegs, err := getRegistrations()

	for allRegs == nil {
		LoggingClient.Info("Waiting for client microservice")
		select {
		case e := <-messageErrors:
			LoggingClient.Error(fmt.Sprintf("exit msg: %s", e.Error()))
			if err != nil {
				LoggingClient.Error(fmt.Sprintf("with error: %s", err.Error()))
			}
			return
		case <-time.After(time.Second):
		}
		allRegs, err = getRegistrations()
	}

	// Create new goroutines for each registration
	for _, reg := range allRegs {
		regInfo := newRegistrationInfo()
		if regInfo.update(reg) {
			registrations[reg.Name] = regInfo
			go registrationLoop(regInfo)
		}
	}

	LoggingClient.Info("Starting registration loop")
	for {
		select {
		case e := <-messageErrors:
			// kill all registration goroutines
			for k, reg := range registrations {
				if !reg.deleteFlag {
					// Do not write in channel that will not be read
					reg.chRegistration <- nil
				}
				delete(registrations, k)
			}
			LoggingClient.Error(fmt.Sprintf("exit msg: %s", e.Error()))
			return

		case update := <-registrationChanges:
			LoggingClient.Info("Registration changes")
			err := updateRunningRegistrations(registrations, update)
			if err != nil {
				LoggingClient.Error(err.Error())
				LoggingClient.Warn(fmt.Sprintf("Error updating registration %s", update.Name))
			}

		case msgEnvelope := <-messageEnvelopes:
			LoggingClient.Info(fmt.Sprintf("Event received on message queue. Topic: %s, Correlation-id: %s ", Configuration.MessageQueue.Topic, msgEnvelope.CorrelationID))
			if msgEnvelope.ContentType != clients.ContentTypeJSON {
				LoggingClient.Error(fmt.Sprintf("Incorrect content type for event message. Received: %s, Expected: %s", msgEnvelope.ContentType, clients.ContentTypeJSON))
				continue
			}
			str := string(msgEnvelope.Payload)
			event := parseEvent(str)
			if event == nil {
				continue
			}

			for k, reg := range registrations {
				if reg.deleteFlag {
					delete(registrations, k)
				} else {
					// TODO only sent event if it is not blocking
					reg.chEvent <- event
				}
			}
		}
	}
}
1279151
func (pool *TxPool) loop() {
	defer pool.wg.Done()

	// Start the stats reporting and transaction eviction tickers
	var prevPending, prevQueued, prevStales int

	report := time.NewTicker(statsReportInterval)
	defer report.Stop()

	evict := time.NewTicker(evictionInterval)
	defer evict.Stop()

	journal := time.NewTicker(pool.config.Rejournal)
	defer journal.Stop()

	// Track the previous head headers for transaction reorgs
	head := pool.chain.CurrentBlock()

	// Keep waiting for and reacting to the various events
	for {
		select {
		// Handle ChainHeadEvent
		case ev := <-pool.chainHeadCh:
			if ev.Block != nil {
				pool.mu.Lock()
				if pool.chainconfig.IsHomestead(ev.Block.Number()) {
					pool.homestead = true
				}
				pool.reset(head.Header(), ev.Block.Header())
				head = ev.Block

				pool.mu.Unlock()
			}
		// Be unsubscribed due to system stopped
		case <-pool.chainHeadSub.Err():
			return

		// Handle stats reporting ticks
		case <-report.C:
			pool.mu.RLock()
			pending, queued := pool.stats()
			stales := pool.priced.stales
			pool.mu.RUnlock()

			if pending != prevPending || queued != prevQueued || stales != prevStales {
				log.Debug("Transaction pool status report", "executable", pending, "queued", queued, "stales", stales)
				prevPending, prevQueued, prevStales = pending, queued, stales
			}

		// Handle inactive account transaction eviction
		case <-evict.C:
			pool.mu.Lock()
			for addr := range pool.queue {
				// Skip local transactions from the eviction mechanism
				if pool.locals.contains(addr) {
					continue
				}
				// Any non-locals old enough should be removed
				if time.Since(pool.beats[addr]) > pool.config.Lifetime {
					for _, tx := range pool.queue[addr].Flatten() {
						pool.removeTx(tx.Hash(), true)
					}
				}
			}
			pool.mu.Unlock()

		// Handle local transaction journal rotation
		case <-journal.C:
			if pool.journal != nil {
				pool.mu.Lock()
				if err := pool.journal.rotate(pool.local()); err != nil {
					log.Warn("Failed to rotate local tx journal", "err", err)
				}
				pool.mu.Unlock()
			}
		}
	}
}
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)
}
250773
func (loop Loop) Close() error {
	errc := make(chan error)
	loop.closing <- errc
	return <-errc
}
1035401
func (p *progressLogger) loopB(tick *time.Ticker, ch <-chan progress.Report) error {
	var r progress.Report
	var ok bool
	var err error

	for ok = true; ok; {
		select {
		case r, ok = <-ch:
			if !ok {
				break
			}
			err = r.Error()
		case <-tick.C:
			line := fmt.Sprintf("\r%s", p.prefix)
			if r != nil {
				line += fmt.Sprintf("(%.0f%%", r.Percentage())
				detail := r.Detail()
				if detail != "" {
					line += fmt.Sprintf(", %s", detail)
				}
				line += ")"
			}
			p.flag.Log(line)
		}
	}

	return err
}
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)
	}
}
383357
func (w *watcher) start() {
	if w.starting || w.running {
		return
	}
	w.starting = true
	go w.loop()
}
1128323
func (d *MultiDiscovery) Run(looper director.Looper) {
	var loopers []director.Looper

	for _, disco := range d.Discoverers {
		l := director.NewFreeLooper(director.FOREVER, make(chan error))
		loopers = append(loopers, l)
		disco.Run(l)
	}

	// Waiting for a quit on the Looper's channel
	looper.Loop(func() error {
		return nil
	})

	for _, l := range loopers {
		l.Quit()
	}
}
709384
func (l *loop) Wait() error {
	<-l.doneC
	l.mux.Lock()
	defer l.mux.Unlock()
	err := l.err
	return err
}
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
		}
	}
}
152779
func (tab *Table) loop() {
	var (
		revalidate     = time.NewTimer(tab.nextRevalidateTime())
		refresh        = time.NewTicker(refreshInterval)
		copyNodes      = time.NewTicker(copyNodesInterval)
		refreshDone    = make(chan struct{})           // where doRefresh reports completion
		revalidateDone chan struct{}                   // where doRevalidate reports completion
		waiting        = []chan struct{}{tab.initDone} // holds waiting callers while doRefresh runs
	)
	defer refresh.Stop()
	defer revalidate.Stop()
	defer copyNodes.Stop()

	// Start initial refresh.
	go tab.doRefresh(refreshDone)

loop:
	for {
		select {
		case <-refresh.C:
			tab.seedRand()
			if refreshDone == nil {
				refreshDone = make(chan struct{})
				go tab.doRefresh(refreshDone)
			}
		case req := <-tab.refreshReq:
			waiting = append(waiting, req)
			if refreshDone == nil {
				refreshDone = make(chan struct{})
				go tab.doRefresh(refreshDone)
			}
		case <-refreshDone:
			for _, ch := range waiting {
				close(ch)
			}
			waiting, refreshDone = nil, nil
		case <-revalidate.C:
			revalidateDone = make(chan struct{})
			go tab.doRevalidate(revalidateDone)
		case <-revalidateDone:
			revalidate.Reset(tab.nextRevalidateTime())
			revalidateDone = nil
		case <-copyNodes.C:
			go tab.copyLiveNodes()
		case <-tab.closeReq:
			break loop
		}
	}

	if refreshDone != nil {
		<-refreshDone
	}
	for _, ch := range waiting {
		close(ch)
	}
	if revalidateDone != nil {
		<-revalidateDone
	}
	close(tab.closed)
}