// Flush generates InterMetrics for the current state of the Histo. percentiles
// indicates what percentiles should be exported from the histogram.
503966
func (h *Histo) Flush(interval time.Duration, percentiles []float64, aggregates HistogramAggregates, global bool) []InterMetric {
	now := time.Now().Unix()
	metrics := make([]InterMetric, 0, aggregates.Count+len(percentiles))
	sinks := routeInfo(h.Tags)

	// The second clause in this if statement can be confusing.
	//
	// The infinity check ensures we don't submit mixed scope histogram summaries. These values
	// are initialized to infinity and reset when a value is sampled locally. This never happens
	// for mixed scope values since we only consume them through the merge path.
	//
	// The global check overrides this for global scope histograms. In this case, we're going to
	// use values from the merged histogram struct.
	//
	// The proliferation of flags here suggests this method is the wrong abstraction, since it
	// requires a lot of forking for different paths. (Read: https://www.sandimetz.com/blog/2016/1/20/the-wrong-abstraction)
	//
	// Think twice before adding to complexity here--it may be worth refactoring how histograms
	// are modeled in mixed scope before continuing with new features.
	if (aggregates.Value&AggregateMax) == AggregateMax && (!math.IsInf(h.LocalMax, 0) || global) {
		// Defensively recopy tags to avoid aliasing bugs in case multiple InterMetrics share the same
		// tag array in the future
		tags := make([]string, len(h.Tags))
		copy(tags, h.Tags)

		val := float64(h.LocalMax)
		if global {
			val = h.Value.Max()
		}
		metrics = append(metrics, InterMetric{
			Name:      fmt.Sprintf("%s.max", h.Name),
			Timestamp: now,
			Value:     val,
			Tags:      tags,
			Type:      GaugeMetric,
			Sinks:     sinks,
		})
	}
	if (aggregates.Value&AggregateMin) == AggregateMin && (!math.IsInf(h.LocalMin, 0) || global) {
		tags := make([]string, len(h.Tags))
		copy(tags, h.Tags)
		val := float64(h.LocalMin)
		if global {
			val = h.Value.Min()
		}
		metrics = append(metrics, InterMetric{
			Name:      fmt.Sprintf("%s.min", h.Name),
			Timestamp: now,
			Value:     val,
			Tags:      tags,
			Type:      GaugeMetric,
			Sinks:     sinks,
		})
	}

	if (aggregates.Value&AggregateSum) == AggregateSum && (h.LocalSum != 0 || global) {
		tags := make([]string, len(h.Tags))
		copy(tags, h.Tags)
		val := float64(h.LocalSum)
		if global {
			val = h.Value.Sum()
		}
		metrics = append(metrics, InterMetric{
			Name:      fmt.Sprintf("%s.sum", h.Name),
			Timestamp: now,
			Value:     val,
			Tags:      tags,
			Type:      GaugeMetric,
			Sinks:     sinks,
		})
	}

	if (aggregates.Value&AggregateAverage) == AggregateAverage && (global || h.LocalSum != 0 && h.LocalWeight != 0) {
		// we need both a rate and a non-zero sum before it will make sense
		// to submit an average
		tags := make([]string, len(h.Tags))
		copy(tags, h.Tags)
		val := float64(h.LocalSum / h.LocalWeight)
		if global {
			val = h.Value.Sum() / h.Value.Count()
		}
		metrics = append(metrics, InterMetric{
			Name:      fmt.Sprintf("%s.avg", h.Name),
			Timestamp: now,
			Value:     val,
			Tags:      tags,
			Type:      GaugeMetric,
			Sinks:     sinks,
		})
	}

	if (aggregates.Value&AggregateCount) == AggregateCount && (h.LocalWeight != 0 || global) {
		// if we haven't received any local samples, then leave this sparse,
		// otherwise it can lead to some misleading zeroes in between the
		// flushes of downstream instances
		tags := make([]string, len(h.Tags))
		copy(tags, h.Tags)
		val := float64(h.LocalWeight)
		if global {
			val = h.Value.Count()
		}
		metrics = append(metrics, InterMetric{
			Name:      fmt.Sprintf("%s.count", h.Name),
			Timestamp: now,
			Value:     val,
			Tags:      tags,
			Type:      CounterMetric,
			Sinks:     sinks,
		})
	}

	if (aggregates.Value & AggregateMedian) == AggregateMedian {
		tags := make([]string, len(h.Tags))
		copy(tags, h.Tags)
		metrics = append(
			metrics,
			InterMetric{
				Name:      fmt.Sprintf("%s.median", h.Name),
				Timestamp: now,
				Value:     float64(h.Value.Quantile(0.5)),
				Tags:      tags,
				Type:      GaugeMetric,
				Sinks:     sinks,
			},
		)
	}

	if (aggregates.Value&AggregateHarmonicMean) == AggregateHarmonicMean && (global || h.LocalReciprocalSum != 0 && h.LocalWeight != 0) {
		// we need both a rate and a non-zero sum before it will make sense
		// to submit an average
		tags := make([]string, len(h.Tags))
		copy(tags, h.Tags)
		val := float64(h.LocalWeight / h.LocalReciprocalSum)
		if global {
			val = h.Value.Count() / h.Value.ReciprocalSum()
		}
		metrics = append(metrics, InterMetric{
			Name:      fmt.Sprintf("%s.hmean", h.Name),
			Timestamp: now,
			Value:     val,
			Tags:      tags,
			Type:      GaugeMetric,
			Sinks:     sinks,
		})
	}

	for _, p := range percentiles {
		tags := make([]string, len(h.Tags))
		copy(tags, h.Tags)
		metrics = append(
			metrics,
			// TODO Fix to allow for p999, etc
			InterMetric{
				Name:      fmt.Sprintf("%s.%dpercentile", h.Name, int(p*100)),
				Timestamp: now,
				Value:     float64(h.Value.Quantile(p)),
				Tags:      tags,
				Type:      GaugeMetric,
				Sinks:     sinks,
			},
		)
	}

	return metrics
}
153679
func (s *KVStore) Flush(ctx context.Context) {
	s.mu.Lock()
	defer s.mu.Unlock()
	for _, b := range s.buckets {
		b.btree.Clear(false)
	}
}
118896
func (x *XRaySpanSink) Flush() {
	x.log.WithFields(logrus.Fields{
		"flushed_spans": atomic.LoadInt64(&x.spansHandled),
		"dropped_spans": atomic.LoadInt64(&x.spansDropped),
	}).Debug("Checkpointing flushed spans for X-Ray")
	metrics.ReportBatch(x.traceClient, []*ssf.SSFSample{
		ssf.Count(sinks.MetricKeyTotalSpansFlushed, float32(atomic.SwapInt64(&x.spansHandled, 0)), map[string]string{"sink": x.Name()}),
		ssf.Count(sinks.MetricKeyTotalSpansDropped, float32(atomic.SwapInt64(&x.spansDropped, 0)), map[string]string{"sink": x.Name()}),
	})
}
980812
func (w *TimeWait) Flush() {
	w.doonce.Do(func() {
		close(w.closer)
		w.action.Fullfill(0)
		atomic.StoreInt64(&w.hits, 0)
	})
}
995944
func (w *Writer) Flush() error {
	if _, err := w.w.Write(w.b); err != nil {
		return err
	}
	w.b = w.b[:0]
	return nil
}
491261
func (n *NodeDrainer) flush(state *state.StateStore) {
	// Cancel anything that may be running.
	if n.exitFn != nil {
		n.exitFn()
	}

	// Store the new state
	if state != nil {
		n.state = state
	}

	n.ctx, n.exitFn = context.WithCancel(context.Background())
	n.jobWatcher = n.jobFactory(n.ctx, n.queryLimiter, n.state, n.logger)
	n.nodeWatcher = n.nodeFactory(n.ctx, n.queryLimiter, n.state, n.logger, n)
	n.deadlineNotifier = n.deadlineNotifierFactory(n.ctx)
	n.nodes = make(map[string]*drainingNode, 32)
}
373926
func (w *compressWriter) Flush() {
	if w.encoder != nil {
		w.encoder.Flush()
	}
	if w, ok := w.ResponseWriter.(http.Flusher); ok {
		w.Flush()
	}
}
62820
func (w *GatedWriter) Flush() {
	w.lock.Lock()
	w.flush = true
	w.lock.Unlock()

	for _, p := range w.buf {
		w.Write(p)
	}
	w.buf = nil
}
196076
func Flush(c context.Context) error {
	return GetState(c).Flush(c, nil)
}
1275402
func (b *BlockedEvals) Flush() {
	b.l.Lock()
	defer b.l.Unlock()

	// Reset the blocked eval tracker.
	b.stats.TotalEscaped = 0
	b.stats.TotalBlocked = 0
	b.stats.TotalQuotaLimit = 0
	b.captured = make(map[string]wrappedEval)
	b.escaped = make(map[string]wrappedEval)
	b.jobs = make(map[structs.NamespacedID]string)
	b.unblockIndexes = make(map[string]uint64)
	b.timetable = nil
	b.duplicates = nil
	b.capacityChangeCh = make(chan *capacityUpdate, unblockBuffer)
	b.stopCh = make(chan struct{})
	b.duplicateCh = make(chan struct{}, 1)
}
525898
func Flush(ctx context.Context) {
	if r, ok := FromContext(ctx); ok {
		if f, ok := r.(flusher); ok {
			f.Flush()
		}
	}
}
483841
func (e *Encoder) Flush() error {
	_, err := e.w.Write(FlushPkt)
	return err
}
100430
func (w *Worker) Flush() WorkerMetrics {
	// This is a critical spot. The worker can't process metrics while this
	// mutex is held! So we try and minimize it by copying the maps of values
	// and assigning new ones.
	wm := NewWorkerMetrics()
	w.mutex.Lock()
	ret := w.wm
	processed := w.processed
	imported := w.imported

	w.wm = wm
	w.processed = 0
	w.imported = 0
	w.mutex.Unlock()

	w.stats.Count("worker.metrics_processed_total", processed, []string{}, 1.0)
	w.stats.Count("worker.metrics_imported_total", imported, []string{}, 1.0)

	return ret
}
840482
func (mem *Mempool) Flush() {
	mem.proxyMtx.Lock()
	defer mem.proxyMtx.Unlock()

	mem.cache.Reset()

	for e := mem.txs.Front(); e != nil; e = e.Next() {
		mem.txs.Remove(e)
		e.DetachPrev()
	}

	mem.txsMap = sync.Map{}
	_ = atomic.SwapInt64(&mem.txsBytes, 0)
}
352847
func (r *RegionSet) flush() {
	for i := 1; i < len(r.regions); i++ {
		ov := r.overlaps(r.regions[i], 0, i)
		if len(ov) == 0 {
			continue
		}
		r.merge(ov[0], append(ov[1:], i))
	}
}
156349
func (s *LedisStore) Flush() error {
	s.lock.Lock()
	defer s.lock.Unlock()

	s.data = make(map[interface{}]interface{})
	return nil
}
1188142
func (agg *Aggregator) flush() {
	if agg.minMetric != nil {
		agg.minMetric.Add(agg.currentBoundary, agg.agg.Min)
	}
	if agg.maxMetric != nil {
		agg.maxMetric.Add(agg.currentBoundary, agg.agg.Max)
	}
	if agg.sumMetric != nil {
		agg.sumMetric.Add(agg.currentBoundary, agg.agg.Sum)
	}
	if agg.cntMetric != nil {
		agg.cntMetric.Add(agg.currentBoundary, agg.agg.Cnt)
	}
	if agg.lstMetric != nil {
		agg.lstMetric.Add(agg.currentBoundary, agg.agg.Lst)
	}
	//msg := fmt.Sprintf("flushed cnt %v sum %f min %f max %f, reset the block", agg.agg.cnt, agg.agg.sum, agg.agg.min, agg.agg.max)
	agg.agg.Reset()
}