// Flood is a long-running goroutine that floods servers from the LAN to the
// given global Serf instance, such as the WAN. This will exit once either of
// the Serf instances are shut down.
489558
func (s *Server) Flood(addrFn router.FloodAddrFn, portFn router.FloodPortFn, global *serf.Serf) {
	s.floodLock.Lock()
	floodCh := make(chan struct{})
	s.floodCh = append(s.floodCh, floodCh)
	s.floodLock.Unlock()

	ticker := time.NewTicker(s.config.SerfFloodInterval)
	defer ticker.Stop()
	defer func() {
		s.floodLock.Lock()
		defer s.floodLock.Unlock()

		for i, ch := range s.floodCh {
			if ch == floodCh {
				s.floodCh = append(s.floodCh[:i], s.floodCh[i+1:]...)
				return
			}
		}
		panic("flood channels out of sync")
	}()

	for {
		select {
		case <-s.serfLAN.ShutdownCh():
			return

		case <-global.ShutdownCh():
			return

		case <-ticker.C:
			goto FLOOD

		case <-floodCh:
			goto FLOOD
		}

	FLOOD:
		router.FloodJoins(s.logger, addrFn, portFn, s.config.Datacenter, s.serfLAN, global)
	}
}
14193
func (s *Server) processRemoteServerShutdown(sid string) {
	s.accounts.Range(func(k, v interface{}) bool {
		a := v.(*Account)
		a.mu.Lock()
		prev := a.strack[sid]
		delete(a.strack, sid)
		a.nrclients -= prev.conns
		a.nrleafs -= prev.leafs
		a.mu.Unlock()
		return true
	})
}
256198
func (o *DrainCmdOptions) RunDrain() error {
	if err := o.RunCordonOrUncordon(true); err != nil {
		return err
	}

	printObj, err := o.ToPrinter("drained")
	if err != nil {
		return err
	}

	drainedNodes := sets.NewString()
	var fatal error

	for _, info := range o.nodeInfos {
		var err error
		if !o.drainer.DryRun {
			err = o.deleteOrEvictPodsSimple(info)
		}
		if err == nil || o.drainer.DryRun {
			drainedNodes.Insert(info.Name)
			printObj(info.Object, o.Out)
		} else {
			fmt.Fprintf(o.ErrOut, "error: unable to drain node %q, aborting command...\n\n", info.Name)
			remainingNodes := []string{}
			fatal = err
			for _, remainingInfo := range o.nodeInfos {
				if drainedNodes.Has(remainingInfo.Name) {
					continue
				}
				remainingNodes = append(remainingNodes, remainingInfo.Name)
			}

			if len(remainingNodes) > 0 {
				fmt.Fprintf(o.ErrOut, "There are pending nodes to be drained:\n")
				for _, nodeName := range remainingNodes {
					fmt.Fprintf(o.ErrOut, " %s\n", nodeName)
				}
			}
			break
		}
	}

	return fatal
}
260413
func FloodFillRegionCovering(region Region, start CellID) []CellID {
	var output []CellID
	all := map[CellID]bool{
		start: true,
	}
	frontier := []CellID{start}
	for len(frontier) > 0 {
		id := frontier[len(frontier)-1]
		frontier = frontier[:len(frontier)-1]
		if !region.IntersectsCell(CellFromCellID(id)) {
			continue
		}
		output = append(output, id)
		for _, nbr := range id.EdgeNeighbors() {
			if !all[nbr] {
				all[nbr] = true
				frontier = append(frontier, nbr)
			}
		}
	}

	return output
}
31884
func (acc *Accumulator) tearDownPools() {
	data := make([]byte, 0, numPools*sha256d.Size)

	acc.poolMutex.Lock()
	for i := 0; i < numPools; i++ {
		data = acc.pool[i].Sum(data)
		acc.pool[i] = nil
	}
	acc.poolZeroSize = 0 // prevent accidential last-minute reseeding
	acc.poolMutex.Unlock()

	acc.genMutex.Lock()
	acc.gen.Reseed(data)
	acc.genMutex.Unlock()
}
10148
func (s *Stream) Flush() (error) {
	var err C.int
	_ = C.pa_simple_flush(s.simple, &err)
	if err == C.PA_OK {
		return nil
	}
	return errorFromCode(err)
}
233209
func (b *NoStorage) Flush() error {
	if f, ok := b.w.(http.Flusher); ok {
		f.Flush()
	}
	return nil
}
258150
func (mc *Connection) FlushAll() (err error) {
	defer handleError(&err)
	mc.setDeadline()
	// flush_all [delay] [noreply]\r\n
	mc.writestrings("flush_all\r\n")
	response := mc.readline()
	if !strings.Contains(response, "OK") {
		panic(NewError(fmt.Sprintf("Error in FlushAll %v", response)))
	}
	return nil
}
1118865
func (f *asyncFlusher) uploaderLoop() {
	for chunk := range f.chunks {
		ctx := f.ctx(chunk.index)
		logging.Infof(ctx, "Chunk flush starting")
		sent, err := f.upload(ctx, chunk)
		f.mu.Lock()
		if err == nil {
			f.rowsSent += sent
		} else {
			f.errs = append(f.errs, err)
		}
		f.mu.Unlock()
		logging.Infof(ctx, "Chunk flush finished")
	}
}
880324
func (c *Host) StreamVolumes(ch chan *volume.Event) (stream.Stream, error) {
	return c.c.ResumingStream("GET", "/storage/volumes", ch)
}
264327
func (f *FileSystem) stream(conn io.ReadWriteCloser) {
	defer metrics.MeasureSince([]string{"client", "file_system", "stream"}, time.Now())
	defer conn.Close()

	// Decode the arguments
	var req cstructs.FsStreamRequest
	decoder := codec.NewDecoder(conn, structs.MsgpackHandle)
	encoder := codec.NewEncoder(conn, structs.MsgpackHandle)

	if err := decoder.Decode(&req); err != nil {
		f.handleStreamResultError(err, helper.Int64ToPtr(500), encoder)
		return
	}

	// Check read permissions
	if aclObj, err := f.c.ResolveToken(req.QueryOptions.AuthToken); err != nil {
		f.handleStreamResultError(err, nil, encoder)
		return
	} else if aclObj != nil && !aclObj.AllowNsOp(req.Namespace, acl.NamespaceCapabilityReadFS) {
		f.handleStreamResultError(structs.ErrPermissionDenied, nil, encoder)
		return
	}

	// Validate the arguments
	if req.AllocID == "" {
		f.handleStreamResultError(allocIDNotPresentErr, helper.Int64ToPtr(400), encoder)
		return
	}
	if req.Path == "" {
		f.handleStreamResultError(pathNotPresentErr, helper.Int64ToPtr(400), encoder)
		return
	}
	switch req.Origin {
	case "start", "end":
	case "":
		req.Origin = "start"
	default:
		f.handleStreamResultError(invalidOrigin, helper.Int64ToPtr(400), encoder)
		return
	}

	fs, err := f.c.GetAllocFS(req.AllocID)
	if err != nil {
		code := helper.Int64ToPtr(500)
		if structs.IsErrUnknownAllocation(err) {
			code = helper.Int64ToPtr(404)
		}

		f.handleStreamResultError(err, code, encoder)
		return
	}

	// Calculate the offset
	fileInfo, err := fs.Stat(req.Path)
	if err != nil {
		f.handleStreamResultError(err, helper.Int64ToPtr(400), encoder)
		return
	}
	if fileInfo.IsDir {
		f.handleStreamResultError(
			fmt.Errorf("file %q is a directory", req.Path),
			helper.Int64ToPtr(400), encoder)
		return
	}

	// If offsetting from the end subtract from the size
	if req.Origin == "end" {
		req.Offset = fileInfo.Size - req.Offset
		if req.Offset < 0 {
			req.Offset = 0
		}
	}

	frames := make(chan *sframer.StreamFrame, streamFramesBuffer)
	errCh := make(chan error)
	var buf bytes.Buffer
	frameCodec := codec.NewEncoder(&buf, structs.JsonHandle)

	// Create the framer
	framer := sframer.NewStreamFramer(frames, streamHeartbeatRate, streamBatchWindow, streamFrameSize)
	framer.Run()
	defer framer.Destroy()

	// If we aren't following end as soon as we hit EOF
	var eofCancelCh chan error
	if !req.Follow {
		eofCancelCh = make(chan error)
		close(eofCancelCh)
	}

	ctx, cancel := context.WithCancel(context.Background())
	defer cancel()

	// Start streaming
	go func() {
		if err := f.streamFile(ctx, req.Offset, req.Path, req.Limit, fs, framer, eofCancelCh); err != nil {
			select {
			case errCh <- err:
			case <-ctx.Done():
			}
		}

		framer.Destroy()
	}()

	// Create a goroutine to detect the remote side closing
	go func() {
		for {
			if _, err := conn.Read(nil); err != nil {
				if err == io.EOF || err == io.ErrClosedPipe {
					// One end of the pipe was explicitly closed, exit cleanly
					cancel()
					return
				}
				select {
				case errCh <- err:
				case <-ctx.Done():
					return
				}
			}
		}
	}()

	var streamErr error
OUTER:
	for {
		select {
		case streamErr = <-errCh:
			break OUTER
		case frame, ok := <-frames:
			if !ok {
				// frame may have been closed when an error
				// occurred. Check once more for an error.
				select {
				case streamErr = <-errCh:
					// There was a pending error!
				default:
					// No error, continue on
				}

				break OUTER
			}

			var resp cstructs.StreamErrWrapper
			if req.PlainText {
				resp.Payload = frame.Data
			} else {
				if err = frameCodec.Encode(frame); err != nil {
					streamErr = err
					break OUTER
				}

				resp.Payload = buf.Bytes()
				buf.Reset()
			}

			if err := encoder.Encode(resp); err != nil {
				streamErr = err
				break OUTER
			}
			encoder.Reset(conn)
		case <-ctx.Done():
			break OUTER
		}
	}

	if streamErr != nil {
		f.handleStreamResultError(streamErr, helper.Int64ToPtr(500), encoder)
		return
	}
}
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
		}
	}
}
1186669
func (p *Pubsub) work() {
	var (
		cnx   redis.Conn
		read  *readPump
		write *writePump
	)

	teardown := func() {
		read.Close()
		write.Close()
		cnx.Close()
		cnx = nil
	}

	defer teardown()

	for {
		if cnx == nil {
			cnx = p.pool.Get()
			read = newReadPump(cnx)
			write = newWritePump(cnx)
			go write.Work()
			p.resubscribe(write)

			go read.Work()
		}

		select {
		case <-p.closer:
			return
		case data := <-p.send:
			write.Data() <- data
			data.written <- struct{}{}
		case event := <-read.Data():
			go p.handleEvent(event)
		case err := <-read.Errs():
			teardown()
			p.errs <- err
		case err := <-write.Errs():
			teardown()
			p.errs <- err
		}
	}
}
793941
func (logger *Logger) Flush() {
	b := &backgroundFlusher{
		logger:     logger,
		updateChan: make(chan struct{}),
	}
	transitChannel <- b
	<-b.updateChan
}
213517
func (s *serfQueries) stream() {
	for {
		select {
		case e := <-s.inCh:
			// Check if this is a query we should process
			if q, ok := e.(*Query); ok && strings.HasPrefix(q.Name, InternalQueryPrefix) {
				go s.handleQuery(q)

			} else if s.outCh != nil {
				s.outCh <- e
			}

		case <-s.shutdownCh:
			return
		}
	}
}
1326128
func (c *conn) Flush() error {
	c.mu.Lock()
	defer c.mu.Unlock()
	return c.buf.Flush()
}
841842
func (stream *AudioStream) Flush() (err error) {
	ret := int(C.SDL_AudioStreamFlush(stream.cptr()))
	err = errorFromInt(ret)
	return
}