// mainEventLoop runs the main event loop with an optional ping channel.
427364
func mainEventLoop(xu *xgbutil.XUtil,
	pingBefore, pingAfter, pingQuit chan struct{}) {
	for {
		if Quitting(xu) {
			if pingQuit != nil {
				pingQuit <- struct{}{}
			}
			break
		}

		// Gobble up as many events as possible (into the queue).
		// If there are no events, we block.
		Read(xu, true)

		// Now process every event/error in the queue.
		processEventQueue(xu, pingBefore, pingAfter)
	}
}
417810
func (c *client) mainloop(params *LookupParams) {
	// start listening for responses
	msgCh := make(chan *dns.Msg, 32)
	if c.ipv4conn != nil {
		go c.recv(c.ipv4conn, msgCh)
	}
	if c.ipv6conn != nil {
		go c.recv(c.ipv6conn, msgCh)
	}

	// Iterate through channels from listeners goroutines
	var entries, sentEntries map[string]*ServiceEntry
	sentEntries = make(map[string]*ServiceEntry)
	for !c.closed {
		select {
		case <-c.closedCh:
			c.shutdown()
		case msg := <-msgCh:
			entries = make(map[string]*ServiceEntry)
			sections := append(msg.Answer, msg.Ns...)
			sections = append(sections, msg.Extra...)
			for _, answer := range sections {
				switch rr := answer.(type) {
				case *dns.PTR:
					if params.ServiceName() != rr.Hdr.Name {
						continue
					}
					if params.ServiceInstanceName() != "" && params.ServiceInstanceName() != rr.Ptr {
						continue
					}
					if _, ok := entries[rr.Ptr]; !ok {
						entries[rr.Ptr] = NewServiceEntry(
							trimDot(strings.Replace(rr.Ptr, rr.Hdr.Name, "", -1)),
							params.Service,
							params.Domain)
					}
					entries[rr.Ptr].TTL = rr.Hdr.Ttl
				case *dns.SRV:
					if params.ServiceInstanceName() != "" && params.ServiceInstanceName() != rr.Hdr.Name {
						continue
					} else if !strings.HasSuffix(rr.Hdr.Name, params.ServiceName()) {
						continue
					}
					if _, ok := entries[rr.Hdr.Name]; !ok {
						entries[rr.Hdr.Name] = NewServiceEntry(
							trimDot(strings.Replace(rr.Hdr.Name, params.ServiceName(), "", 1)),
							params.Service,
							params.Domain)
					}
					entries[rr.Hdr.Name].HostName = rr.Target
					entries[rr.Hdr.Name].Port = int(rr.Port)
					entries[rr.Hdr.Name].TTL = rr.Hdr.Ttl
				case *dns.TXT:
					if params.ServiceInstanceName() != "" && params.ServiceInstanceName() != rr.Hdr.Name {
						continue
					} else if !strings.HasSuffix(rr.Hdr.Name, params.ServiceName()) {
						continue
					}
					if _, ok := entries[rr.Hdr.Name]; !ok {
						entries[rr.Hdr.Name] = NewServiceEntry(
							trimDot(strings.Replace(rr.Hdr.Name, params.ServiceName(), "", 1)),
							params.Service,
							params.Domain)
					}
					entries[rr.Hdr.Name].Text = rr.Txt
					entries[rr.Hdr.Name].TTL = rr.Hdr.Ttl
				case *dns.A:
					for k, e := range entries {
						if e.HostName == rr.Hdr.Name && entries[k].AddrIPv4 == nil {
							entries[k].AddrIPv4 = rr.A
						}
					}
				case *dns.AAAA:
					for k, e := range entries {
						if e.HostName == rr.Hdr.Name && entries[k].AddrIPv6 == nil {
							entries[k].AddrIPv6 = rr.AAAA
						}
					}
				}
			}
		}

		if len(entries) > 0 {
			for k, e := range entries {
				if e.TTL == 0 {
					delete(entries, k)
					delete(sentEntries, k)
					continue
				}
				if _, ok := sentEntries[k]; ok {
					continue
				}
				params.Entries <- e
				sentEntries[k] = e
			}
			// reset entries
			entries = make(map[string]*ServiceEntry)
		}
	}
}
379888
func (psh *PubSubHub) sendLoop(conn *connection) {
	// If returning because the WebSocketHub sent a quit signal, the receive
	// loop may still be waiting for a message, so it is necessary to close the
	// websocket.Conn in this case.
	ws := conn.ws
	defer closeWS(ws)

	// sendLoop should be started after conn.Add(1), and before a conn.Wait().
	defer conn.Done()

	// Use this client's unique channel to receive signals from the
	// WebSocketHub, which broadcasts signals to all clients.
	updateSigChan := *conn.client.c
	clientData := conn.client.cl
	buff := new(bytes.Buffer)

loop:
	for {
		// Wait for signal from the WebSocketHub to update.
		select {
		case sig, ok := <-updateSigChan:
			// Check if the update channel was closed. Either the websocket
			// hub will do it after unregistering the client, or forcibly in
			// response to (http.CloseNotifier).CloseNotify() and only then
			// if the hub has somehow lost track of the client.
			if !ok {
				return
			}

			if !sig.IsValid() {
				log.Errorf("invalid signal to send: %s / %d", sig.Signal.String(), int(sig.Signal))
				continue loop
			}

			if !clientData.isSubscribed(sig) {
				log.Errorf("Client not subscribed for %s events. "+
					"WebSocketHub should have caught this.", sig.Signal.String())
				continue loop // break
			}

			log.Tracef("signaling client: %p", conn.client.c) // ID by address

			// Respond to the websocket client.
			pushMsg := pstypes.WebSocketMessage{
				EventId: sig.Signal.String(),
				// Message is set in switch statement below.
			}

			// JSON encoder for the Message.
			buff.Reset()
			enc := json.NewEncoder(buff)

			switch sig.Signal {
			case sigAddressTx:
				// sig was already validated, but do it again here in case the
				// type changed without changing the type assertion here.
				am, ok := sig.Msg.(*pstypes.AddressMessage)
				if !ok {
					log.Errorf("sigAddressTx did not store a *AddressMessage in Msg.")
					continue loop
				}
				err := enc.Encode(am)
				if err != nil {
					log.Warnf("Encode(AddressMessage) failed: %v", err)
				}

				pushMsg.Message = buff.String()
			case sigNewBlock:
				psh.state.mtx.RLock()
				if psh.state.BlockInfo == nil {
					psh.state.mtx.RUnlock()
					break // from switch to send empty message
				}
				err := enc.Encode(exptypes.WebsocketBlock{
					Block: psh.state.BlockInfo,
					Extra: psh.state.GeneralInfo,
				})
				psh.state.mtx.RUnlock()
				if err != nil {
					log.Warnf("Encode(WebsocketBlock) failed: %v", err)
				}

				pushMsg.Message = buff.String()

			case sigMempoolUpdate:
				// You probably want the sigNewTX event. sigMempoolUpdate sends
				// a summary of mempool contents, and the NumLatestMempoolTxns
				// latest transactions.
				inv := psh.MempoolInventory()
				if inv == nil {
					break // from switch to send empty message
				}
				inv.RLock()
				err := enc.Encode(inv.MempoolShort)
				inv.RUnlock()
				if err != nil {
					log.Warnf("Encode(MempoolShort) failed: %v", err)
				}

				pushMsg.Message = buff.String()

			case sigPingAndUserCount:
				// ping and send user count
				pushMsg.Message = strconv.Itoa(psh.wsHub.NumClients())

			case sigNewTxs:
				// Marshal this client's tx buffer if it is not empty.
				clientData.newTxs.Lock()
				if len(clientData.newTxs.t) == 0 {
					clientData.newTxs.Unlock()
					continue loop // break sigselect
				}
				err := enc.Encode(clientData.newTxs.t)

				// Reinit the tx buffer.
				clientData.newTxs.t = make(pstypes.TxList, 0, NewTxBufferSize)
				clientData.newTxs.Unlock()
				if err != nil {
					log.Warnf("Encode([]*exptypes.MempoolTx) failed: %v", err)
				}

				pushMsg.Message = buff.String()

			// case sigSyncStatus:
			// 	err := enc.Encode(explorer.SyncStatus())
			// 	if err != nil {
			// 		log.Warnf("Encode(SyncStatus()) failed: %v", err)
			// 	}
			// 	pushMsg.Message = buff.String()

			default:
				log.Errorf("Not sending a %v to the client.", sig)
				continue loop // break sigselect
			} // switch sig

			// Send the message.
			err := ws.SetWriteDeadline(time.Now().Add(wsWriteTimeout))
			if err != nil && !pstypes.IsWSClosedErr(err) {
				log.Warnf("SetWriteDeadline failed: %v", err)
			}
			if err = websocket.JSON.Send(ws, pushMsg); err != nil {
				// Do not log the error if the connection is just closed.
				if !pstypes.IsWSClosedErr(err) {
					log.Debugf("Failed to encode WebSocketMessage (push) %v: %v", sig, err)
				}
				// If the send failed, the client is probably gone, quit the
				// send loop, unregistering the client from the websocket hub.
				log.Errorf("websocket.JSON.Send of %v failed: %v", pushMsg, err)
				return
			}

		// end of case sig, ok := <-updateSigChan

		case <-psh.wsHub.quitWSHandler:
			return
		} // select
	} // for { a.k.a. loop:
}
135180
func (e *EventSubscriber) loop() {
	defer close(e.Events)
	for {
		select {
		case <-e.stop:
			return
		case event := <-e.queue:
			e.Events <- event
		}
	}
}
985127
func (t *UDPv4) readLoop(unhandled chan<- ReadPacket) {
	defer t.wg.Done()
	if unhandled != nil {
		defer close(unhandled)
	}

	buf := make([]byte, maxPacketSize)
	for {
		nbytes, from, err := t.conn.ReadFromUDP(buf)
		if netutil.IsTemporaryError(err) {
			// Ignore temporary read errors.
			t.log.Debug("Temporary UDP read error", "err", err)
			continue
		} else if err != nil {
			// Shut down the loop for permament errors.
			if err != io.EOF {
				t.log.Debug("UDP read error", "err", err)
			}
			return
		}
		if t.handlePacket(from, buf[:nbytes]) != nil && unhandled != nil {
			select {
			case unhandled <- ReadPacket{buf[:nbytes], from}:
			default:
			}
		}
	}
}
282606
func (s *Session) recvLoop() error {
	defer close(s.recvDoneCh)
	hdr := header(make([]byte, headerSize))
	for {
		// Read the header
		if _, err := io.ReadFull(s.bufRead, hdr); err != nil {
			if err != io.EOF && !strings.Contains(err.Error(), "closed") && !strings.Contains(err.Error(), "reset by peer") {
				s.logger.Printf("[ERR] yamux: Failed to read header: %v", err)
			}
			return err
		}

		// Verify the version
		if hdr.Version() != protoVersion {
			s.logger.Printf("[ERR] yamux: Invalid protocol version: %d", hdr.Version())
			return ErrInvalidVersion
		}

		mt := hdr.MsgType()
		if mt < typeData || mt > typeGoAway {
			return ErrInvalidMsgType
		}

		if err := handlers[mt](s, hdr); err != nil {
			return err
		}
	}
}
793728
func (c *Connection) healthCheckLoop(connChan chan ConnectionEvent) {
	// create a separate API channel for health check probes
	ch, err := c.newAPIChannel(1, 1)
	if err != nil {
		log.Error("Failed to create health check API channel, health check will be disabled:", err)
		return
	}

	var (
		sinceLastReply time.Duration
		failedChecks   int
	)

	// send health check probes until an error or timeout occurs
	for {
		// sleep until next health check probe period
		time.Sleep(HealthCheckProbeInterval)

		if atomic.LoadUint32(&c.vppConnected) == 0 {
			// Disconnect has been called in the meantime, return the healthcheck - reconnect loop
			log.Debug("Disconnected on request, exiting health check loop.")
			return
		}

		// try draining probe replies from previous request before sending next one
		select {
		case <-ch.replyChan:
			log.Debug("drained old probe reply from reply channel")
		default:
		}

		// send the control ping request
		ch.reqChan <- &vppRequest{msg: msgControlPing}

		for {
			// expect response within timeout period
			select {
			case vppReply := <-ch.replyChan:
				err = vppReply.err

			case <-time.After(HealthCheckReplyTimeout):
				err = ErrProbeTimeout

				// check if time since last reply from any other
				// channel is less than health check reply timeout
				c.lastReplyLock.Lock()
				sinceLastReply = time.Since(c.lastReply)
				c.lastReplyLock.Unlock()

				if sinceLastReply < HealthCheckReplyTimeout {
					log.Warnf("VPP health check probe timing out, but some request on other channel was received %v ago, continue waiting!", sinceLastReply)
					continue
				}
			}
			break
		}

		if err == ErrProbeTimeout {
			failedChecks++
			log.Warnf("VPP health check probe timed out after %v (%d. timeout)", HealthCheckReplyTimeout, failedChecks)
			if failedChecks > HealthCheckThreshold {
				// in case of exceeded failed check treshold, assume VPP disconnected
				log.Errorf("VPP health check exceeded treshold for timeouts (>%d), assuming disconnect", HealthCheckThreshold)
				connChan <- ConnectionEvent{Timestamp: time.Now(), State: Disconnected}
				break
			}
		} else if err != nil {
			// in case of error, assume VPP disconnected
			log.Errorf("VPP health check probe failed: %v", err)
			connChan <- ConnectionEvent{Timestamp: time.Now(), State: Disconnected, Error: err}
			break
		} else if failedChecks > 0 {
			// in case of success after failed checks, clear failed check counter
			failedChecks = 0
			log.Infof("VPP health check probe OK")
		}
	}

	// cleanup
	ch.Close()
	c.disconnectVPP()

	// we are now disconnected, start connect loop
	c.connectLoop(connChan)
}
717830
func (f *lightFetcher) syncLoop() {
	requesting := false
	defer f.pm.wg.Done()
	for {
		select {
		case <-f.pm.quitSync:
			return
		// when a new announce is received, request loop keeps running until
		// no further requests are necessary or possible
		case newAnnounce := <-f.requestChn:
			f.lock.Lock()
			s := requesting
			requesting = false
			var (
				rq      *distReq
				reqID   uint64
				syncing bool
			)

			if !f.syncing && !(newAnnounce && s) {
				rq, reqID, syncing = f.nextRequest()
			}
			f.lock.Unlock()

			if rq != nil {
				requesting = true
				if _, ok := <-f.pm.reqDist.queue(rq); ok {
					if syncing {
						f.lock.Lock()
						f.syncing = true
						f.lock.Unlock()
					} else {
						go func() {
							time.Sleep(softRequestTimeout)
							f.reqMu.Lock()
							req, ok := f.requested[reqID]
							if ok {
								req.timeout = true
								f.requested[reqID] = req
							}
							f.reqMu.Unlock()
							// keep starting new requests while possible
							f.requestChn <- false
						}()
					}
				} else {
					f.requestChn <- false
				}
			}
		case reqID := <-f.timeoutChn:
			f.reqMu.Lock()
			req, ok := f.requested[reqID]
			if ok {
				delete(f.requested, reqID)
			}
			f.reqMu.Unlock()
			if ok {
				f.pm.serverPool.adjustResponseTime(req.peer.poolEntry, time.Duration(mclock.Now()-req.sent), true)
				req.peer.Log().Debug("Fetching data timed out hard")
				go f.pm.removePeer(req.peer.id)
			}
		case resp := <-f.deliverChn:
			f.reqMu.Lock()
			req, ok := f.requested[resp.reqID]
			if ok && req.peer != resp.peer {
				ok = false
			}
			if ok {
				delete(f.requested, resp.reqID)
			}
			f.reqMu.Unlock()
			if ok {
				f.pm.serverPool.adjustResponseTime(req.peer.poolEntry, time.Duration(mclock.Now()-req.sent), req.timeout)
			}
			f.lock.Lock()
			if !ok || !(f.syncing || f.processResponse(req, resp)) {
				resp.peer.Log().Debug("Failed processing response")
				go f.pm.removePeer(resp.peer.id)
			}
			f.lock.Unlock()
		case p := <-f.syncDone:
			f.lock.Lock()
			p.Log().Debug("Done synchronising with peer")
			f.checkSyncedHeaders(p)
			f.syncing = false
			f.lock.Unlock()
			f.requestChn <- false
		}
	}
}
1300394
func (s *Server) lanEventHandler() {
	for {
		select {
		case e := <-s.eventChLAN:
			switch e.EventType() {
			case serf.EventMemberJoin:
				s.lanNodeJoin(e.(serf.MemberEvent))
				s.localMemberEvent(e.(serf.MemberEvent))

			case serf.EventMemberLeave, serf.EventMemberFailed, serf.EventMemberReap:
				s.lanNodeFailed(e.(serf.MemberEvent))
				s.localMemberEvent(e.(serf.MemberEvent))

			case serf.EventUser:
				s.localEvent(e.(serf.UserEvent))
			case serf.EventMemberUpdate:
				s.localMemberEvent(e.(serf.MemberEvent))
			case serf.EventQuery: // Ignore
			default:
				s.logger.Printf("[WARN] consul: Unhandled LAN Serf Event: %#v", e)
			}

		case <-s.shutdownCh:
			return
		}
	}
}
959397
func RunLoop(netplugin *plugin.NetPlugin, ctrlIP, vtepIP, hostname string) error {
	// Register ourselves
	err := registerService(ObjdbClient, ctrlIP, vtepIP, hostname, netplugin.PluginConfig.Instance.VxlanUDPPort)

	// Start peer discovery loop
	go peerDiscoveryLoop(netplugin, ObjdbClient, ctrlIP, vtepIP)

	return err
}
822863
func (p *Pool) queryLoop() error {
	if err := p.sendNextJobs(p.config.BatchSize); err != nil {
		return err
	}
	for {
		minWait := time.After(p.config.MinWait)
		select {
		case <-p.exit:
			// Close the channel to tell workers to stop executing new jobs
			close(p.jobs)
			return nil
		case <-minWait:
			if err := p.sendNextJobs(p.config.BatchSize); err != nil {
				return err
			}
		}
	}
	return nil
}
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
}
1049436
func (c *ChainIndexer) eventLoop(currentHeader *types.Header, events chan ChainHeadEvent, sub event.Subscription) {
	// Mark the chain indexer as active, requiring an additional teardown
	atomic.StoreUint32(&c.active, 1)

	defer sub.Unsubscribe()

	// Fire the initial new head event to start any outstanding processing
	c.newHead(currentHeader.Number.Uint64(), false)

	var (
		prevHeader = currentHeader
		prevHash   = currentHeader.Hash()
	)
	for {
		select {
		case errc := <-c.quit:
			// Chain indexer terminating, report no failure and abort
			errc <- nil
			return

		case ev, ok := <-events:
			// Received a new event, ensure it's not nil (closing) and update
			if !ok {
				errc := <-c.quit
				errc <- nil
				return
			}
			header := ev.Block.Header()
			if header.ParentHash != prevHash {
				// Reorg to the common ancestor if needed (might not exist in light sync mode, skip reorg then)
				// TODO(karalabe, zsfelfoldi): This seems a bit brittle, can we detect this case explicitly?

				if rawdb.ReadCanonicalHash(c.chainDb, prevHeader.Number.Uint64()) != prevHash {
					if h := rawdb.FindCommonAncestor(c.chainDb, prevHeader, header); h != nil {
						c.newHead(h.Number.Uint64(), true)
					}
				}
			}
			c.newHead(header.Number.Uint64(), false)

			prevHeader, prevHash = header, header.Hash()
		}
	}
}
312482
func (e *endpoint) protocolListenLoop(rcvWnd seqnum.Size) *tcpip.Error {
	defer func() {
		// Mark endpoint as closed. This will prevent goroutines running
		// handleSynSegment() from attempting to queue new connections
		// to the endpoint.
		e.mu.Lock()
		e.state = stateClosed

		// Do cleanup if needed.
		e.completeWorkerLocked()

		if e.drainDone != nil {
			close(e.drainDone)
		}
		e.mu.Unlock()

		// Notify waiters that the endpoint is shutdown.
		e.waiterQueue.Notify(waiter.EventIn | waiter.EventOut)
	}()

	e.mu.Lock()
	v6only := e.v6only
	e.mu.Unlock()

	ctx := newListenContext(e.stack, rcvWnd, v6only, e.netProto)

	s := sleep.Sleeper{}
	s.AddWaker(&e.notificationWaker, wakerForNotification)
	s.AddWaker(&e.newSegmentWaker, wakerForNewSegment)
	for {
		switch index, _ := s.Fetch(true); index {
		case wakerForNotification:
			n := e.fetchNotifications()
			if n&notifyClose != 0 {
				return nil
			}
			if n&notifyDrain != 0 {
				for !e.segmentQueue.empty() {
					s := e.segmentQueue.dequeue()
					e.handleListenSegment(ctx, s)
					s.decRef()
				}
				synRcvdCount.pending.Wait()
				close(e.drainDone)
				<-e.undrain
			}

		case wakerForNewSegment:
			// Process at most maxSegmentsPerWake segments.
			mayRequeue := true
			for i := 0; i < maxSegmentsPerWake; i++ {
				s := e.segmentQueue.dequeue()
				if s == nil {
					mayRequeue = false
					break
				}

				e.handleListenSegment(ctx, s)
				s.decRef()
			}

			// If the queue is not empty, make sure we'll wake up
			// in the next iteration.
			if mayRequeue && !e.segmentQueue.empty() {
				e.newSegmentWaker.Assert()
			}
		}
	}
}
683122
func (p *grpcPeer) monitorLoopWait(ctx context.Context, cancel context.CancelFunc, connectivityState connectivity.State) (connectivity.State, bool) {
	changedC := make(chan bool, 1)
	go func() { changedC <- p.clientConn.WaitForStateChange(ctx, connectivityState) }()

	loop := false
	select {
	case changed := <-changedC:
		if cancel != nil {
			cancel()
		}
		if changed {
			connectivityState = p.clientConn.GetState()
		}
		loop = true
	case <-p.stoppingC:
	case <-p.t.once.Stopping():
		if cancel != nil {
			cancel()
		}
	}
	return connectivityState, loop
}
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))
		}
	}
}
890711
func (soc *Socket) GetMcastLoop() (bool, error) {
	v, e := soc.getInt64(C.ZMQ_MCAST_LOOP)
	if v == 0 {
		return false, e
	}
	return true, e
}