// NewLoop creates a loop object that executes the cancelable function according to the
// given policy
186370
func NewLoop(policy Policy, cancelable func(chan chan error) chan error) Loop {
	loop := Loop{
		startTime:   time.Now(),
		retryPolicy: policy,
		cancelable:  cancelable,
		waiting:     make(chan error, 1),
		closing:     make(chan chan error),
	}
	go loop.loop()
	return loop
}
487315
func (c *Canceller) NewContext() (context.Context, context.CancelFunc) {
	return context.WithCancel(c.ctx)
}
1133485
func eventLoop(ctx context.Context, config Config, eventDecoder encoding.Decoder) (err error) {
	for err == nil {
		select {
		case <-ctx.Done():
			err = ctx.Err()
			return
		default:
		}

		var e scheduler.Event
		if err = eventDecoder.Decode(&e); err != nil {
			return
		}

		err = config.handler.HandleEvent(ctx, &e)
	}
	return
}
1324929
func (k *KBFSService) listenLoop(l net.Listener) error {
	go func() {
		<-k.stopCh
		l.Close()
	}()
	defer l.Close()
	for {
		c, err := l.Accept()
		if err != nil {
			if libkb.IsSocketClosedError(err) {
				err = nil
			}

			k.log.Debug("listenLoop() done, error: %+v", err)
			return err
		}
		go k.handle(c)
	}
}
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
}
977200
func (c *client) writeLoop() {
	defer c.srv.grWG.Done()

	// Used to check that we did flush from last wake up.
	waitOk := true

	// Main loop. Will wait to be signaled and then will use
	// buffered outbound structure for efficient writev to the underlying socket.
	for {
		c.mu.Lock()
		owtf := c.out.fsp > 0 && c.out.pb < maxBufSize && c.out.fsp < maxFlushPending
		if waitOk && (c.out.pb == 0 || owtf) && !c.flags.isSet(clearConnection) {
			// Wait on pending data.
			c.out.sgw = true
			c.out.sg.Wait()
			c.out.sgw = false
		}
		// Flush data
		// TODO(dlc) - This could spin if another go routine in flushOutbound waiting on a slow IO.
		waitOk = c.flushOutbound()
		isClosed := c.flags.isSet(clearConnection)
		c.mu.Unlock()

		if isClosed {
			return
		}
	}
}
125126
func (c *cache) backendLoop(l loop.Loop) error {
	// Prepare ticker for lifetime check.
	checker := time.NewTicker(c.interval)
	defer checker.Stop()
	// Run loop.
	for {
		select {
		case <-l.ShallStop():
			return nil
		case do := <-c.taskc:
			if err := do(c); err != nil {
				return err
			}
		case lenc := <-c.lenc:
			lenc <- len(c.buckets)
		case <-checker.C:
			if err := c.cleanup(); err != nil {
				return err
			}
		}
	}
}
1230836
func (b *stdBackend) backendLoop(l loop.Loop) error {
	// Init the monitor.
	b.init()
	// Run loop.
	for {
		select {
		case <-l.ShallStop():
			return nil
		case measuring := <-b.measuringC:
			// Received a new measuring.
			if mp, ok := b.etmData[measuring.id]; ok {
				mp.update(measuring)
			} else {
				b.etmData[measuring.id] = newStdMeasuringPoint(measuring)
			}
		case ssvChange := <-b.ssvChangeC:
			// Received a new change.
			if ssv, ok := b.ssvData[ssvChange.id]; ok {
				ssv.update(ssvChange)
			} else {
				b.ssvData[ssvChange.id] = newStdStaySetVariable(ssvChange)
			}
		case registration := <-b.retrieverRegistrationC:
			// Received a new retriever for registration.
			b.dsrData[registration.id] = registration.dsr
		case cmd := <-b.commandC:
			// Received a command to process.
			b.processCommand(cmd)
		}
	}
}
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")
		}
	}
}
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)
			}
		}
	}
}
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
		}
	}
}
394254
func (g *gossipRegistry) connectLoop() {
	// try every second
	ticker := time.NewTicker(1 * time.Second)
	defer ticker.Stop()

	for {
		select {
		case <-g.done:
			return
		case <-g.options.Context.Done():
			g.Stop()
			return
		case <-ticker.C:
			var addrs []string

			g.RLock()

			// only process if we have a memberlist
			if g.member == nil {
				g.RUnlock()
				continue
			}

			// self
			local := g.member.LocalNode().Address()

			// operate on each member
			for node, action := range g.members {
				switch action {
				// process leave event
				case nodeActionLeave:
					// don't process self
					if node == local {
						continue
					}
					addrs = append(addrs, node)
				}
			}

			g.RUnlock()

			// connect to all the members
			// TODO: only connect to new members
			if len(addrs) > 0 {
				g.connect(addrs)
			}
		}
	}
}
362013
func (w *writer) readLoop() {
	for {
		if _, _, err := w.conn.NextReader(); err != nil {
			w.conn.Close()
			break
		}
	}
}
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
			}
		}
	}()
}
767490
func (a *Agent) eventLoop() {
	serfShutdownCh := a.serf.ShutdownCh()
	for {
		select {
		case e := <-a.eventCh:
			a.logger.Printf("[INFO] agent: Received event: %s", e.String())
			a.eventHandlersLock.Lock()
			handlers := a.eventHandlerList
			a.eventHandlersLock.Unlock()
			for _, eh := range handlers {
				eh.HandleEvent(e)
			}

		case <-serfShutdownCh:
			a.logger.Printf("[WARN] agent: Serf shutdown detected, quitting")
			a.Shutdown()
			return

		case <-a.shutdownCh:
			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
	}
}
1251585
func (d *DiskStorer) cleanerLoop() {
	defer d.wg.Done()

	t, c := timerTestHarness(d.cleanInterval)

	select {
	case <-c:
		d.Clean()
		t.Reset(d.cleanInterval)
	case <-d.quit:
		t.Stop()
		return
	}
}