// doFlush actually sends the metrics to the monitor.
500156
func (s *State) doFlush(c context.Context, state *tsmon.State, settings *tsmonSettings) error {
	var mon monitor.Monitor
	var err error

	if s.testingMonitor != nil {
		mon = s.testingMonitor
	} else if s.IsDevMode || settings.ProdXAccount == "" {
		mon = monitor.NewDebugMonitor("")
	} else {
		logging.Infof(c, "Sending metrics to ProdX using %s", settings.ProdXAccount)
		transport, err := auth.GetRPCTransport(
			c,
			auth.AsActor,
			auth.WithServiceAccount(settings.ProdXAccount),
			auth.WithScopes(monitor.ProdxmonScopes...))
		if err != nil {
			return err
		}
		endpoint, err := url.Parse(prodXEndpoint)
		if err != nil {
			return err
		}
		mon, err = monitor.NewHTTPMonitor(c, &http.Client{Transport: transport}, endpoint)
		if err != nil {
			return err
		}
	}

	defer mon.Close()
	if err = state.Flush(c, mon); err != nil {
		return err
	}

	return nil
}
408899
func Flush() error {
	update_size_maybe()
	prepare_diff_messages()
	for _, diff := range diffbuf {
		r := small_rect{
			left:   0,
			top:    diff.pos,
			right:  term_size.x - 1,
			bottom: diff.pos + diff.lines - 1,
		}
		write_console_output(out, diff.chars, r)
	}
	if !is_cursor_hidden(cursor_x, cursor_y) {
		move_cursor(cursor_x, cursor_y)
	}
	return nil
}
1334024
func NotifyFlush(dom *domain.Domain, pluginName string) error {
	p := getByName(pluginName)
	if p == nil || p.Manifest.flushWatcher == nil || p.State != Ready {
		return errors.Errorf("plugin %s doesn't exists or unsupported flush or doesn't start with PD", pluginName)
	}
	_, err := dom.GetEtcdClient().KV.Put(context.Background(), p.Manifest.flushWatcher.path, "")
	if err != nil {
		return err
	}
	return nil
}
1015864
func (w *IndexWriter) flush(triggerMerge bool, applyAllDeletes bool) error {
	// NOTE: this method cannot be sync'd because
	// maybeMerge() in turn calls mergeScheduler.merge which
	// in turn can take a long time to run and we don't want
	// to hold the lock for that.  In the case of
	// ConcurrentMergeScheduler this can lead to deadlock
	// when it stalls due to too many running merges.

	// We can be called during close, when closing==true, so we must pass false to ensureOpen:
	w.ClosingControl.ensureOpen(false)
	ok, err := w.doFlush(applyAllDeletes)
	if err != nil {
		return err
	}
	if ok && triggerMerge {
		return w.maybeMerge(w.config.MergePolicy(), MERGE_TRIGGER_FULL_FLUSH, UNBOUNDED_MAX_MERGE_SEGMENTS)
	}
	return nil
}
758669
func (pcond *ProxyConds) DoFunc(f func(resp *http.Response, ctx *ProxyCtx) *http.Response) {
	pcond.Do(FuncRespHandler(f))
}
989061
func (w *Writer) flush(events []*cloudwatchlogs.InputLogEvent) error {
	resp, err := w.client.PutLogEvents(&cloudwatchlogs.PutLogEventsInput{
		LogEvents:     events,
		LogGroupName:  w.group,
		LogStreamName: w.stream,
		SequenceToken: w.sequenceToken,
	})
	if err != nil {
		return err
	}

	if resp.RejectedLogEventsInfo != nil {
		w.err = &RejectedLogEventsInfoError{Info: resp.RejectedLogEventsInfo}
		return w.err
	}

	w.sequenceToken = resp.NextSequenceToken

	return nil
}
940832
func (sfx *SignalFxSink) Flush(ctx context.Context, interMetrics []samplers.InterMetric) error {
	span, subCtx := trace.StartSpanFromContext(ctx, "")
	defer span.ClientFinish(sfx.traceClient)

	flushStart := time.Now()
	coll := sfx.newPointCollection()
	numPoints := 0
	countSkipped := 0
	countStatusMetrics := 0

METRICLOOP: // Convenience label so that inner nested loops and `continue` easily
	for _, metric := range interMetrics {
		if !sinks.IsAcceptableMetric(metric, sfx) {
			countSkipped++
			continue
		}
		if len(sfx.metricNamePrefixDrops) > 0 {
			for _, pre := range sfx.metricNamePrefixDrops {
				if strings.HasPrefix(metric.Name, pre) {
					countSkipped++
					continue METRICLOOP
				}
			}
		}
		if len(sfx.metricTagPrefixDrops) > 0 {
			for _, dropTag := range sfx.metricTagPrefixDrops {
				for _, tag := range metric.Tags {
					if strings.HasPrefix(tag, dropTag) {
						countSkipped++
						continue METRICLOOP
					}
				}
			}
		}
		dims := map[string]string{}
		// Set the hostname as a tag, since SFx doesn't have a first-class hostname field
		dims[sfx.hostnameTag] = sfx.hostname
		for _, tag := range metric.Tags {
			kv := strings.SplitN(tag, ":", 2)
			key := kv[0]

			if len(kv) == 1 {
				dims[key] = ""
			} else {
				dims[key] = kv[1]
			}
		}
		// Copy common dimensions
		for k, v := range sfx.commonDimensions {
			dims[k] = v
		}
		metricKey := ""
		if sfx.varyBy != "" {
			if val, ok := dims[sfx.varyBy]; ok {
				metricKey = val
			}
		}

		for k := range sfx.excludedTags {
			delete(dims, k)
		}
		delete(dims, "veneursinkonly")

		var point *datapoint.Datapoint
		switch metric.Type {
		case samplers.GaugeMetric:
			point = sfxclient.GaugeF(metric.Name, dims, metric.Value)
		case samplers.CounterMetric:
			// TODO I am not certain if this should be a Counter or a Cumulative
			point = sfxclient.Counter(metric.Name, dims, int64(metric.Value))
		case samplers.StatusMetric:
			countStatusMetrics++
			point = sfxclient.GaugeF(metric.Name, dims, metric.Value)
		}
		coll.addPoint(metricKey, point)
		numPoints++
	}
	tags := map[string]string{"sink": "signalfx"}
	span.Add(ssf.Count(sinks.MetricKeyTotalMetricsSkipped, float32(countSkipped), tags))
	err := coll.submit(subCtx, sfx.traceClient, sfx.maxPointsInBatch)
	if err != nil {
		span.Error(err)
	}
	span.Add(ssf.Timing(sinks.MetricKeyMetricFlushDuration, time.Since(flushStart), time.Nanosecond, tags))
	span.Add(ssf.Count(sinks.MetricKeyTotalMetricsFlushed, float32(numPoints), tags))
	sfx.log.WithFields(logrus.Fields{
		"metrics": len(interMetrics),
		"success": err == nil,
	}).Info("Completed flush to SignalFx")

	return err
}
688199
func (p *BulkProcessor) Flush() error {
	p.statsMu.Lock()
	p.stats.Flushed++
	p.statsMu.Unlock()

	for _, w := range p.workers {
		w.flushC <- struct{}{}
		<-w.flushAckC // wait for completion
	}
	return nil
}
592370
func (ef *File) monitorChunks(stop chan bool, sleep time.Duration) (cc chan Chunk) {
	cc = make(chan Chunk, 4)
	sleepTime := sleep
	markedChunks := datastructs.NewSyncedSet()

	// Main routine to feed the Chunk Channel
	go func() {
		defer close(cc)
		firstLoopFlag := !ef.monitorExisting
		for {
			// Parse the file header again to get the updates in the file
			ef.ParseFileHeader()

			// check if we should stop or not
			select {
			case <-stop:
				return
			default:
				// go through
			}
			curChunks := datastructs.NewSyncedSet()
			//cs := make(ChunkSorter, 0, ef.Header.ChunkCount)
			ss := datastructs.NewSortedSlice(0, int(ef.Header.ChunkCount))
			for i := uint16(0); i < ef.Header.ChunkCount; i++ {
				offsetChunk := int64(ef.Header.ChunkDataOffset) + int64(ChunkSize)*int64(i)
				chunk, err := ef.FetchRawChunk(offsetChunk)
				curChunks.Add(chunk.Header.FirstEventRecID, chunk.Header.LastEventRecID)
				// We append only the Chunks whose EventRecordIds have not been treated yet
				if markedChunks.Contains(chunk.Header.FirstEventRecID) && markedChunks.Contains(chunk.Header.LastEventRecID) {
					continue
				}
				switch {
				case err != nil && err != io.EOF:
					panic(err)
				case err == nil:
					markedChunks.Add(chunk.Header.FirstEventRecID)
					markedChunks.Add(chunk.Header.LastEventRecID)
					if !firstLoopFlag {
						//cs = append(cs, chunk)
						ss.Insert(chunk)
					}
				}
			}

			// Cleanup the useless cache entries (consider putting in go routine if worth)
			markedChunks = datastructs.NewSyncedSet(markedChunks.Intersect(&curChunks))

			// We flag out of first loop
			firstLoopFlag = false
			// We sort out the chunks
			//sort.Stable(cs)
			//for _, rc := range cs {
			for rc := range ss.ReversedIter() {
				chunk, err := ef.FetchChunk(rc.(Chunk).Offset)
				switch {
				case err != nil && err != io.EOF:
					panic(err)
				case err == nil:
					cc <- chunk
				}
			}

			// Check if we should quit
			if ef.Header.ChunkCount >= math.MaxUint16 {
				log.Info("Monitoring stopped: maximum chunk number reached")
				break
			}

			// Sleep between loops
			time.Sleep(sleepTime)
		}
	}()
	return
}
392885
func kbfsLibdokanFlushFileBuffers(
	fname C.LPCWSTR,
	pfi C.PDOKAN_FILE_INFO) C.NTSTATUS {
	debugf("FlushFileBuffers '%v' %v", d16{fname}, *pfi)
	ctx, cancel := getContext(pfi)
	if cancel != nil {
		defer cancel()
	}
	err := getfi(pfi).FlushFileBuffers(ctx, makeFI(fname, pfi))
	return errToNT(err)
}
224904
func (dwpt *DocumentsWriterPerThread) sealFlushedSegment(flushedSegment *FlushedSegment) error {
	assert(flushedSegment != nil)

	newSegment := flushedSegment.segmentInfo

	setDiagnostics(newSegment.Info, SOURCE_FLUSH)

	segSize, err := newSegment.SizeInBytes()
	if err != nil {
		return err
	}
	context := store.NewIOContextForFlush(&store.FlushInfo{
		newSegment.Info.DocCount(),
		segSize,
	})

	var success = false
	defer func() {
		if !success {
			if dwpt.infoStream.IsEnabled("DWPT") {
				dwpt.infoStream.Message(
					"DWPT", "hit error relating compound file for newly flushed segment %v",
					newSegment.Info.Name)
			}
		}
	}()

	if dwpt.indexWriterConfig.UseCompoundFile() {
		files, err := createCompoundFile(
			dwpt.infoStream, dwpt.directory,
			CheckAbortNone(0), newSegment.Info, context)
		if err != nil {
			return err
		}
		for _, file := range files {
			dwpt.filesToDelete[file] = true
		}
		newSegment.Info.SetUseCompoundFile(true)
	}

	// Have codec write SegmentInfo. Must do this after creating CFS so
	// that 1) .si isn't slurped into CFS, and 2) .si reflects
	// useCompoundFile=true change above:
	err = dwpt.codec.SegmentInfoFormat().SegmentInfoWriter().Write(
		dwpt.directory,
		newSegment.Info,
		flushedSegment.fieldInfos,
		context)
	if err != nil {
		return err
	}

	// TODO: ideally we would freeze newSegment here!!
	// because any changes after writing the .si will be lost...

	// Must write deleted docs after the CFS so we don't slurp the del
	// file into CFS:
	if flushedSegment.liveDocs != nil {
		delCount := flushedSegment.delCount
		assert(delCount > 0)
		if dwpt.infoStream.IsEnabled("DWPT") {
			dwpt.infoStream.Message("DWPT", "flush: write %v deletes gen=%v",
				delCount, flushedSegment.segmentInfo.DelGen())
		}

		// TODO: we should prune the segment if it's 100% deleted... but
		// merge will also catch it.

		// TODO: in the NRT case id'd be better to hand this del vector
		// over to the shortly-to-be-opened SegmentReader and let it
		// carry the changes; there's no reason to use filesystem as
		// intermediary here.

		info := flushedSegment.segmentInfo
		codec := info.Info.Codec().(Codec)
		err = codec.LiveDocsFormat().WriteLiveDocs(flushedSegment.liveDocs,
			dwpt.directory, info, delCount, context)
		if err != nil {
			return err
		}
		newSegment.SetDelCount(delCount)
		newSegment.AdvanceDelGen()
	}

	success = true
	return nil
}
1208509
func (batch *MessageBatch) AfterFlushDo(callback func() error) error {
	batch.flushing.IncWhenDone()
	defer batch.flushing.Done()
	return callback()
}
489806
func (c *ConfigLocal) DoBackgroundFlushes() bool {
	if !c.Mode().BackgroundFlushesEnabled() {
		return false
	}

	c.lock.RLock()
	defer c.lock.RUnlock()
	return !c.noBGFlush
}
835780
func (fs *InterruptFS) EnableFlushBlocking() {
	fs.mu.Lock()
	defer fs.mu.Unlock()

	fs.blockForFlushes = true
}
366938
func (queue *BufferedQueue) FlushBuffer() {
	flushing := make(chan bool, 1)
	queue.flushStatus <- flushing
	queue.flushCommand <- true
	<-flushing
	return
}
1309836
func (h *Holder) monitorCacheFlush() {
	ticker := time.NewTicker(h.cacheFlushInterval)
	defer ticker.Stop()

	for {
		select {
		case <-h.closing:
			return
		case <-ticker.C:
			h.flushCaches()
		}
	}
}
651612
func (dwpt *DocumentsWriterPerThread) flush() (fs *FlushedSegment, err error) {
	assert(dwpt.numDocsInRAM > 0)
	assert2(dwpt.deleteSlice.isEmpty(), "all deletes must be applied in prepareFlush")
	dwpt.segmentInfo.SetDocCount(dwpt.numDocsInRAM)
	numBytesUsed := dwpt.bytesUsed()
	flushState := NewSegmentWriteState(dwpt.infoStream, dwpt.directory,
		dwpt.segmentInfo, dwpt.fieldInfos.Finish(),
		dwpt.indexWriterConfig.TermIndexInterval(), dwpt.pendingUpdates,
		store.NewIOContextForFlush(&store.FlushInfo{dwpt.numDocsInRAM, numBytesUsed}))
	startMBUsed := float64(numBytesUsed) / 1024 / 1024

	// Apply delete-by-docID now (delete-byDocID only happens when an
	// error is hit processing that doc, e.g., if analyzer has some
	// problem with the text):
	if delCount := len(dwpt.pendingUpdates.docIDs); delCount > 0 {
		flushState.LiveDocs = dwpt.codec.LiveDocsFormat().NewLiveDocs(dwpt.numDocsInRAM)
		for _, delDocID := range dwpt.pendingUpdates.docIDs {
			flushState.LiveDocs.Clear(delDocID)
		}
		flushState.DelCountOnFlush = delCount
		atomic.AddInt64(&dwpt.pendingUpdates.bytesUsed, -int64(delCount)*BYTES_PER_DEL_DOCID)
		dwpt.pendingUpdates.docIDs = nil
	}

	if dwpt.aborting {
		if dwpt.infoStream.IsEnabled("DWPT") {
			dwpt.infoStream.Message("DWPT", "flush: skip because aborting is set")
		}
		return nil, nil
	}

	if dwpt.infoStream.IsEnabled("DWPT") {
		dwpt.infoStream.Message("DWPT", "flush postings as segment %v numDocs=%v",
			flushState.SegmentInfo.Name, dwpt.numDocsInRAM)
	}

	var success = false
	defer func() {
		if !success {
			dwpt.abort(dwpt.filesToDelete)
		}
	}()

	err = dwpt.consumer.flush(flushState)
	if err != nil {
		return nil, err
	}
	dwpt.pendingUpdates.terms = make(map[*Term]int)
	files := make(map[string]bool)
	dwpt.directory.EachCreatedFiles(func(name string) {
		files[name] = true
	})
	dwpt.segmentInfo.SetFiles(files)

	info := NewSegmentCommitInfo(dwpt.segmentInfo, 0, -1, -1, -1)
	if dwpt.infoStream.IsEnabled("DWPT") {
		dwpt.infoStream.Message("DWPT", "new segment has %v deleted docs",
			check(flushState.LiveDocs == nil, 0,
				flushState.SegmentInfo.DocCount()-flushState.DelCountOnFlush))
		dwpt.infoStream.Message("DWPT", "new segment has %v; %v; %v; %v; %v",
			check(flushState.FieldInfos.HasVectors, "vectors", "no vectors"),
			check(flushState.FieldInfos.HasNorms, "norms", "no norms"),
			check(flushState.FieldInfos.HasDocValues, "docValues", "no docValues"),
			check(flushState.FieldInfos.HasProx, "prox", "no prox"),
			check(flushState.FieldInfos.HasFreq, "freqs", "no freqs"))
		dwpt.infoStream.Message("DWPT", "flushedFiles=%v", info.Files())
		dwpt.infoStream.Message("DWPT", "flushed codec=%v", dwpt.codec)
	}

	var segmentUpdates *BufferedUpdates
	if len(dwpt.pendingUpdates.queries) > 0 {
		segmentUpdates = dwpt.pendingUpdates
	}

	if dwpt.infoStream.IsEnabled("DWPT") {
		numBytes, err := info.SizeInBytes()
		if err != nil {
			return nil, err
		}
		newSegmentSize := float64(numBytes) / 1024 / 1024
		dwpt.infoStream.Message("DWPT",
			"flushed: segment=%v ramUsed=%v MB newFlushedSize(includes docstores)=%v MB docs/MB=%v",
			dwpt.segmentInfo.Name, startMBUsed, newSegmentSize,
			float64(flushState.SegmentInfo.DocCount())/newSegmentSize)
	}

	assert(dwpt.segmentInfo != nil)

	fs = newFlushedSegment(info, flushState.FieldInfos, segmentUpdates,
		flushState.LiveDocs, flushState.DelCountOnFlush)
	err = dwpt.sealFlushedSegment(fs)
	if err != nil {
		return nil, err
	}
	success = true

	return fs, nil
}