/*
FlushAllThreads is synced by IW fullFlushLock. Flushing all threads
is a two stage operation; the caller must ensure (in try/finally)
that finishFLush is called after this method, to release the flush
lock in DWFlushControl
*/
484912
func (dw *DocumentsWriter) flushAllThreads(indexWriter *IndexWriter) (bool, error) {
	if dw.infoStream.IsEnabled("DW") {
		dw.infoStream.Message("DW", "startFullFlush")
	}

	flushingDeleteQueue := func() *DocumentsWriterDeleteQueue {
		dw.Lock()
		defer dw.Unlock()
		dw.pendingChangesInCurrentFullFlush = dw.anyChanges()
		dq := dw.deleteQueue
		// Cut over to a new delete queue. This must be synced on the
		// flush control otherwise a new DWPT could sneak into the loop
		// with an already flushing delete queue
		dw.flushControl.markForFullFlush() // swaps the delQueue synced on FlushControl
		dw.currentFullFlushDelQueue = dq
		return dq
	}()
	assert(dw.currentFullFlushDelQueue != nil)
	assert(dw.currentFullFlushDelQueue != dw.deleteQueue)

	return func() (bool, error) {
		anythingFlushed := false
		defer func() { assert(flushingDeleteQueue == dw.currentFullFlushDelQueue) }()

		flushingDWPT := dw.flushControl.nextPendingFlush()
		for flushingDWPT != nil {
			flushed, err := dw.doFlush(flushingDWPT)
			if err != nil {
				return false, err
			}
			if flushed {
				anythingFlushed = true
			}
			next := dw.flushControl.nextPendingFlush()
			assert(next != flushingDWPT)
			flushingDWPT = next
		}

		// If a concurrent flush is still in flight wait for it
		dw.flushControl.waitForFlush()
		if !anythingFlushed && flushingDeleteQueue.anyChanges() {
			// apply deletes if we did not flush any document
			if dw.infoStream.IsEnabled("DW") {
				dw.infoStream.Message("DW", "flush naked frozen global deletes")
			}
			err := dw.ticketQueue.addDeletes(flushingDeleteQueue)
			if err != nil {
				return false, err
			}
		}
		_, err := dw.ticketQueue.forcePurge(indexWriter)
		if err != nil {
			return false, err
		}
		assert(!flushingDeleteQueue.anyChanges() && !dw.ticketQueue.hasTickets())
		return anythingFlushed, nil
	}()
}
901962
func (l *Ledis) FlushAll() error {
	l.wLock.Lock()
	defer l.wLock.Unlock()

	return l.flushAll()
}
600651
func (evc *EventCache) Flush() {
	for _, ei := range evc.events {
		evc.evsw.FireEvent(ei.event, ei.data)
	}
	// Clear the buffer, since we only add to it with append it's safe to just set it to nil and maybe safe an allocation
	evc.events = nil
}
1086263
func (rww *responseWriterWrapper) Flush() {
	if f, ok := rww.w.(http.Flusher); ok {
		f.Flush()
	} else {
		panic(httpserver.NonFlusherError{Underlying: rww.w}) // should be recovered at the beginning of middleware stack
	}
}
49819
func (s *statusCodeWithoutCloseNotify) Flush() {
	if flusher, ok := s.ResponseWriter.(http.Flusher); ok {
		flusher.Flush()
	}
}
828251
func (fc *DocumentsWriterFlushControl) pruneBlockedQueue(flushingQueue *DocumentsWriterDeleteQueue) {
	for e := fc.blockedFlushes.Front(); e != nil; e = e.Next() {
		if blockedFlush := e.Value.(*BlockedFlush); blockedFlush.dwpt.deleteQueue == flushingQueue {
			fc.blockedFlushes.Remove(e)
			_, ok := fc.flushingWriters[blockedFlush.dwpt]
			assert2(!ok, "DWPT is already flushing")
			// Record the flushing DWPT to reduce flushBytes in doAfterFlush
			fc.flushingWriters[blockedFlush.dwpt] = blockedFlush.bytes
			// don't decr pending here - its already done when DWPT is blocked
			fc.flushQueue.PushBack(blockedFlush.dwpt)
		}
	}
}
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
}
156349
func (s *LedisStore) Flush() error {
	s.lock.Lock()
	defer s.lock.Unlock()

	s.data = make(map[interface{}]interface{})
	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
}
159229
func (s *StatusCheck) Flush() []InterMetric {
	s.Timestamp = time.Now().Unix()
	s.Type = StatusMetric
	s.Sinks = routeInfo(s.Tags)
	return []InterMetric{s.InterMetric}
}
1264556
func (avroWriter *Writer) Flush() error {
	if avroWriter.nextBlockRecords == 0 {
		return nil
	}

	// Write out all of the buffered records as a new block
	// Must be called before closing to ensure the last block is written
	if fwWriter, ok := avroWriter.compressedWriter.(CloseableResettableWriter); ok {
		fwWriter.Close()
		fwWriter.Reset(avroWriter.blockBuffer)
	}

	if avroWriter.nextBlockRecords > 0 {
		block := &avro.AvroContainerBlock{
			NumRecords:  avroWriter.nextBlockRecords,
			RecordBytes: avroWriter.blockBuffer.Bytes(),
			Sync:        avroWriter.syncMarker,
		}
		err := block.Serialize(avroWriter.writer)
		if err != nil {
			return err
		}
	}

	avroWriter.blockBuffer.Reset()
	avroWriter.nextBlockRecords = 0

	return nil
}
744237
func (w *Watcher) flush() {
	// refreshEvents are stored newest first.
	for i := len(w.syncEvents) - 1; i >= 0; i-- {
		e := &w.syncEvents[i]
		for e.ch != nil {
			change := Change{
				C:     e.key.c,
				Id:    e.key.id,
				Revno: e.revno,
			}
			select {
			case <-w.tomb.Dying():
				return
			case req := <-w.request:
				w.handle(req)
				continue
			case e.ch <- change:
			}
			break
		}
	}
	// requestEvents are stored oldest first, and
	// may grow during the loop.
	for i := 0; i < len(w.requestEvents); i++ {
		e := &w.requestEvents[i]
		for e.ch != nil {
			change := Change{
				C:     e.key.c,
				Id:    e.key.id,
				Revno: e.revno,
			}
			select {
			case <-w.tomb.Dying():
				return
			case req := <-w.request:
				w.handle(req)
				continue
			case e.ch <- change:
			}
			break
		}
	}
	w.syncEvents = w.syncEvents[:0]
	w.requestEvents = w.requestEvents[:0]
}
322049
func (p *PeriodicDispatch) flush() {
	p.updateCh = make(chan struct{}, 1)
	p.tracked = make(map[structs.NamespacedID]*structs.Job)
	p.heap = NewPeriodicHeap()
	p.stopFn = nil
}
291110
func (c *Client) Flush() error {
	if c == nil {
		return fmt.Errorf("Client is nil")
	}
	c.Lock()
	defer c.Unlock()
	return c.flushLocked()
}
628192
func (bw *BufferedFileWriter) Flush() {
	bw.mu.Lock()
	// could add check if bw.done already
	//  should never happen
	bw.bufWriter.Flush()
	bw.origWriter.f.Sync()
	bw.mu.Unlock()
}
858933
func flush(stream FSEventStreamRef, sync bool) {
	if sync {
		C.FSEventStreamFlushSync(stream)
	} else {
		C.FSEventStreamFlushAsync(stream)
	}
}
1155695
func (ew *EventWorker) Flush() []ssf.SSFSample {
	ew.mutex.Lock()

	retsamples := ew.samples
	// these slices will be allocated again at append time
	ew.samples = nil

	ew.mutex.Unlock()
	if len(retsamples) != 0 {
		ew.stats.Count("worker.other_samples_flushed_total", int64(len(retsamples)), nil, 1.0)
	}
	return retsamples
}