// ClientLoop connects to websocket server, the keep send and recv dataS
421472
func ClientLoop() {
	for {
		time.Sleep(time.Second)
		err := ConnectWebSocket()
		if err != nil {
			fmt.Println("failed to connect websocket", err.Error())
			continue
		}
		// time.Sleep(time.Second)
		err = SendMessage("2", "all", "join", "dummy2")
		go sendUntilErr(1)
		recvUntilErr()
		err = CloseWebSocket()
		if err != nil {
			fmt.Println("failed to close websocket", err.Error())
		}
	}

}
850407
func (g *gossipRegistry) eventLoop() {
	for {
		select {
		// return when done
		case <-g.done:
			return
		case ev := <-g.events:
			// TODO: nonblocking update
			g.Lock()
			if _, ok := g.members[ev.node]; ok {
				g.members[ev.node] = ev.action
			}
			g.Unlock()
		}
	}
}
964837
func (c *Crontab) backendLoop(l loop.Loop) error {
	ticker := time.NewTicker(c.frequency)
	for {
		select {
		case <-l.ShallStop():
			return nil
		case cmd := <-c.commandChan:
			if cmd.add {
				c.jobs[cmd.id] = cmd.job
			} else {
				delete(c.jobs, cmd.id)
			}
		case now := <-ticker.C:
			for id, job := range c.jobs {
				c.do(id, job, now)
			}
		}
	}
}
916793
func (env *Zlisp) FindLoop(target *Loop) (int, error) {
	if env.curfunc.user {
		panic(fmt.Errorf("impossible in user-defined-function to find a loop '%s'", target.stmtname.name))
	}
	instruc := env.curfunc.fun
	for i := range instruc {
		switch loop := instruc[i].(type) {
		case LoopStartInstr:
			if loop.loop == target {
				return i, nil
			}
		}
	}
	return -1, fmt.Errorf("could not find loop target '%s'", target.stmtname.name)
}
72671
func (proxier *Proxier) SyncLoop() {
	t := time.NewTicker(proxier.syncPeriod)
	defer t.Stop()
	for {
		<-t.C
		klog.V(6).Infof("Periodic sync")
		proxier.Sync()
	}
}
270752
func (sq *servingQueue) threadCountLoop() {
	var threadCountTarget int
	for {
		for threadCountTarget > sq.threadCount {
			sq.wg.Add(1)
			go sq.threadController()
			sq.threadCount++
		}
		if threadCountTarget < sq.threadCount {
			select {
			case threadCountTarget = <-sq.setThreadsCh:
			case sq.stopThreadCh <- struct{}{}:
				sq.threadCount--
			case <-sq.quit:
				sq.wg.Done()
				return
			}
		} else {
			select {
			case threadCountTarget = <-sq.setThreadsCh:
			case <-sq.quit:
				sq.wg.Done()
				return
			}
		}
	}
}
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...)
}
1190790
func (s *Scroller) backendLoop(l loop.Loop) error {
	// Initial positioning.
	if err := s.seekInitial(); err != nil {
		return err
	}
	// Polling loop.
	timer := time.NewTimer(0)
	for {
		select {
		case <-l.ShallStop():
			return nil
		case <-timer.C:
			for {
				line, readErr := s.readLine()
				_, writeErr := s.writer.Write(line)
				if writeErr != nil {
					return writeErr
				}
				if readErr != nil {
					if readErr != io.EOF {
						return readErr
					}
					break
				}
			}
			if writeErr := s.writer.Flush(); writeErr != nil {
				return writeErr
			}
			timer.Reset(s.pollTime)
		}
	}
}
586118
func (t *udp) readLoop() {
	defer t.conn.Close()
	// Discovery packets are defined to be no larger than 1280 bytes.
	// Packets larger than this size will be cut at the end and treated
	// as invalid because their hash won't match.
	buf := make([]byte, 1280)
	for {
		nbytes, from, err := t.conn.ReadFromUDP(buf)
		ingressTrafficMeter.Mark(int64(nbytes))
		if netutil.IsTemporaryError(err) {
			// Ignore temporary read errors.
			log.Debug(fmt.Sprintf("Temporary read error: %v", err))
			continue
		} else if err != nil {
			// Shut down the loop for permament errors.
			log.Debug(fmt.Sprintf("Read error: %v", err))
			return
		}
		t.handlePacket(from, buf[:nbytes])
	}
}
614150
func (t *transacter) receiveLoop(connIndex int) {
	c := t.conns[connIndex]
	defer t.endingWg.Done()
	for {
		_, _, err := c.ReadMessage()
		if err != nil {
			if !websocket.IsCloseError(err, websocket.CloseNormalClosure) {
				t.logger.Error(
					fmt.Sprintf("failed to read response on conn %d", connIndex),
					"err",
					err,
				)
			}
			return
		}
		if t.stopped || t.connsBroken[connIndex] {
			return
		}
	}
}
1027446
func (s *LesServer) startEventLoop() {
	s.protocolManager.wg.Add(1)

	var processing bool
	blockProcFeed := make(chan bool, 100)
	s.protocolManager.blockchain.(*core.BlockChain).SubscribeBlockProcessingEvent(blockProcFeed)
	totalRechargeCh := make(chan uint64, 100)
	totalRecharge := s.costTracker.subscribeTotalRecharge(totalRechargeCh)
	totalCapacityCh := make(chan uint64, 100)
	updateRecharge := func() {
		if processing {
			s.protocolManager.servingQueue.setThreads(s.thcBlockProcessing)
			s.fcManager.SetRechargeCurve(flowcontrol.PieceWiseLinear{{0, 0}, {totalRecharge, totalRecharge}})
		} else {
			s.protocolManager.servingQueue.setThreads(s.thcNormal)
			s.fcManager.SetRechargeCurve(flowcontrol.PieceWiseLinear{{0, 0}, {totalRecharge / 10, totalRecharge}, {totalRecharge, totalRecharge}})
		}
	}
	updateRecharge()
	totalCapacity := s.fcManager.SubscribeTotalCapacity(totalCapacityCh)
	s.priorityClientPool.setLimits(s.maxPeers, totalCapacity)

	go func() {
		for {
			select {
			case processing = <-blockProcFeed:
				updateRecharge()
			case totalRecharge = <-totalRechargeCh:
				updateRecharge()
			case totalCapacity = <-totalCapacityCh:
				s.priorityClientPool.setLimits(s.maxPeers, totalCapacity)
			case <-s.protocolManager.quitSync:
				s.protocolManager.wg.Done()
				return
			}
		}
	}()
}
485135
func (c *ClientConn) mainLoop() {
	defer c.Close()

	// Build the map of available server messages
	typeMap := make(map[uint8]ServerMessage)

	defaultMessages := []ServerMessage{
		new(FramebufferUpdateMessage),
		new(SetColorMapEntriesMessage),
		new(BellMessage),
		new(ServerCutTextMessage),
	}

	for _, msg := range defaultMessages {
		typeMap[msg.Type()] = msg
	}

	if c.config.ServerMessages != nil {
		for _, msg := range c.config.ServerMessages {
			typeMap[msg.Type()] = msg
		}
	}

	for {
		var messageType uint8
		if err := binary.Read(c.c, binary.BigEndian, &messageType); err != nil {
			break
		}

		msg, ok := typeMap[messageType]
		if !ok {
			// Unsupported message type! Bad!
			break
		}

		parsedMsg, err := msg.Read(c, c.c)
		if err != nil {
			break
		}

		if c.config.ServerMessageCh == nil {
			continue
		}

		c.config.ServerMessageCh <- parsedMsg
	}
}
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)
}
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
}
1298759
func (c *EchoClient) ReadLoop() error {
	// start loop to read msg from server
	for {
		msg, err := c.conn.Read() // if you want set a read deadline, you can use 'connector.ReadTimeout(timeout)'
		if err != nil {
			fmt.Printf("read msg from server<%s> failure", c.serverAddr)
			return err
		}

		fmt.Printf("receive a msg<%s> from <%s>", msg, c.serverAddr)
	}
}
233431
func Client(only bool) Option {
	return func(o *Options) error {
		o.Client = only
		return nil
	}
}
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()
			}

		}
	}
}