func (fc *DocumentsWriterFlushControl) waitIfStalled() {
	if fc.infoStream.IsEnabled("DWFC") {
		fc.infoStream.Message(
			"DWFC", "waitIfStalled: numFlushesPending: %v netBytes: %v flushingBytes: %v fullFlush: %v",
			fc.flushQueue.Len(), fc.netBytes(), fc.flushBytes(), fc.fullFlush)
	}
	fc.DocumentsWriterStallControl.waitIfStalled()
}
/*
This mehtod will block if too many DWPT are currently flushing and no
checked out DWPT are available.
package index
import (
	"container/list"
	"fmt"
	"github.com/balzaczyy/golucene/core/util"
	"math"
	"sync"
	"sync/atomic"
)
type DocumentsWriterFlushControl struct {
	sync.Locker
	condFlushWait *sync.Cond
	hardMaxBytesPerDWPT int64
	_activeBytes        int64
	_flushBytes         int64
	numPending          int  
	numDocsSinceStalled int
	flushDeletes        int32  
	fullFlush           bool
	flushQueue          *list.List
	blockedFlushes  *list.List
	flushingWriters map[*DocumentsWriterPerThread]int64
	maxConfiguredRamBuffer float64
	peakActiveBytes        int64  
	peakFlushBytes         int64  
	peakNetBytes           int64  
	peakDelta              int64  
	flushByRAMWasDisabled  bool   
	*DocumentsWriterStallControl  
	perThreadPool *DocumentsWriterPerThreadPool
	flushPolicy   FlushPolicy
	closed        bool
	documentsWriter       *DocumentsWriter
	config                LiveIndexWriterConfig
	bufferedUpdatesStream *BufferedUpdatesStream
	infoStream            util.InfoStream
	fullFlushBuffer []*DocumentsWriterPerThread
}
func newDocumentsWriterFlushControl(documentsWriter *DocumentsWriter,
	config LiveIndexWriterConfig,
	bufferedUpdatesStream *BufferedUpdatesStream) *DocumentsWriterFlushControl {
	objLock := &sync.Mutex{}
	ans := &DocumentsWriterFlushControl{
		Locker:                objLock,
		condFlushWait:         sync.NewCond(objLock),
		flushQueue:            list.New(),
		blockedFlushes:        list.New(),
		flushingWriters:       make(map[*DocumentsWriterPerThread]int64),
		infoStream:            config.InfoStream(),
		perThreadPool:         documentsWriter.perThreadPool,
		flushPolicy:           documentsWriter.flushPolicy,
		config:                config,
		hardMaxBytesPerDWPT:   int64(config.RAMPerThreadHardLimitMB()) * 1024 * 1024,
		documentsWriter:       documentsWriter,
		bufferedUpdatesStream: bufferedUpdatesStream,
	}
	ans.DocumentsWriterStallControl = newDocumentsWriterStallControl()
	return ans
}
func (fc *DocumentsWriterFlushControl) activeBytes() int64 {
	fc.Lock()
	defer fc.Unlock()
	return fc._activeBytes
}
func (fc *DocumentsWriterFlushControl) flushBytes() int64 {
	fc.Lock()
	defer fc.Unlock()
	return fc._flushBytes
}
func (fc *DocumentsWriterFlushControl) netBytes() int64 {
	fc.Lock()
	defer fc.Unlock()
	return fc._netBytes()
}
func (fc *DocumentsWriterFlushControl) _netBytes() int64 {
	return fc._activeBytes + fc._flushBytes
}
func (fc *DocumentsWriterFlushControl) assertMemory() {
	maxRamMB := fc.config.RAMBufferSizeMB()
	if fc.flushByRAMWasDisabled || maxRamMB == DISABLE_AUTO_FLUSH {
		fc.flushByRAMWasDisabled = true
		return
	}
	if maxRamMB > fc.maxConfiguredRamBuffer {
		fc.maxConfiguredRamBuffer = maxRamMB
	}
	ram := fc._flushBytes + fc._activeBytes
	ramBufferBytes := int64(fc.maxConfiguredRamBuffer) * 1024 * 1024
	expected := 2*ramBufferBytes + int64(fc.numPending+len(fc.flushingWriters)+
		fc.blockedFlushes.Len())*fc.peakDelta +
		int64(fc.numDocsSinceStalled)*fc.peakDelta
	if fc.peakDelta < (ramBufferBytes >> 1) {
		assertn(ram <= expected,
			"actual mem: %v, expected mem: %v, flush mem: %v, active mem: %v, "+
				"pending DWPT: %v, flushing DWPT: %v, blocked DWPT: %v, "+
				"peakDelta mem: %v bytes, ramBufferBytes=%v, maxConfiguredRamBuffer=%v",
			ram, expected, fc._flushBytes, fc._activeBytes, fc.numPending,
			len(fc.flushingWriters), fc.blockedFlushes.Len(), fc.peakDelta,
			ramBufferBytes, fc.maxConfiguredRamBuffer)
	}
}
func (fc *DocumentsWriterFlushControl) commitPerThreadBytes(perThread *ThreadState) {
	delta := perThread.dwpt.bytesUsed() - perThread.bytesUsed
	perThread.bytesUsed += delta
	if perThread.flushPending {
		fc._flushBytes += delta
	} else {
		fc._activeBytes += delta
	}
	fc.assertUpdatePeaks(delta)
}
func (fc *DocumentsWriterFlushControl) assertUpdatePeaks(delta int64) {
	if fc.peakActiveBytes < fc._activeBytes {
		fc.peakActiveBytes = fc._activeBytes
	}
	if fc.peakFlushBytes < fc._flushBytes {
		fc.peakFlushBytes = fc._flushBytes
	}
	if n := fc._netBytes(); fc.peakNetBytes < n {
		fc.peakNetBytes = n
	}
	if fc.peakDelta < delta {
		fc.peakDelta = delta
	}
}
func (fc *DocumentsWriterFlushControl) doAfterDocument(perThread *ThreadState, isUpdate bool) *DocumentsWriterPerThread {
	fc.Lock()
	defer fc.Unlock()
	defer func() {
		stalled := fc.updateStallState()
		fc.assertNumDocsSinceStalled(stalled)
		fc.assertMemory()
	}()
	fc.commitPerThreadBytes(perThread)
	if !perThread.flushPending {
		if isUpdate {
			fc.flushPolicy.onUpdate(fc, perThread)
		} else {
			fc.flushPolicy.onInsert(fc, perThread)
		}
		if !perThread.flushPending && perThread.bytesUsed > fc.hardMaxBytesPerDWPT {
			fc._setFlushPending(perThread)
		}
	}
	var flushingDWPT *DocumentsWriterPerThread
	if fc.fullFlush {
		if perThread.flushPending {
			fc.checkoutAndBlock(perThread)
			flushingDWPT = fc.nextPendingFlush()
		}
	} else {
		flushingDWPT = fc._tryCheckoutForFlush(perThread)
	}
	return flushingDWPT
}
func (fc *DocumentsWriterFlushControl) assertNumDocsSinceStalled(stalled bool) {
	if fc.stalled {
		fc.numDocsSinceStalled++
	} else {
		fc.numDocsSinceStalled = 0
	}
}
func (fc *DocumentsWriterFlushControl) doAfterFlush(dwpt *DocumentsWriterPerThread) {
	fc.Lock()
	defer fc.Unlock()
	bytes, ok := fc.flushingWriters[dwpt]
	assert(ok)
	defer func() {
		defer func() {
			fc.condFlushWait.Signal()
		}()
		fc.updateStallState()
	}()
	delete(fc.flushingWriters, dwpt)
	fc._flushBytes -= bytes
	fc.assertMemory()
}
func (fc *DocumentsWriterFlushControl) updateStallState() bool {
	var limit int64 = math.MaxInt64
	if maxRamMB := fc.config.RAMBufferSizeMB(); maxRamMB != DISABLE_AUTO_FLUSH {
		limit = int64(2 * 1024 * 1024 * maxRamMB)
	}
	stall := (fc._activeBytes+fc._flushBytes) > limit &&
		fc._activeBytes < limit &&
		!fc.closed
	fc.updateStalled(stall)
	return stall
}
func (fc *DocumentsWriterFlushControl) waitForFlush() {
	fc.Lock()
	defer fc.Unlock()
	for len(fc.flushingWriters) > 0 {
		fc.condFlushWait.Wait()
	}
}
func (fc *DocumentsWriterFlushControl) setFlushPending(perThread *ThreadState) {
	fc.Lock()
	defer fc.Unlock()
	fc._setFlushPending(perThread)
}
func (fc *DocumentsWriterFlushControl) _setFlushPending(perThread *ThreadState) {
	assert(!perThread.flushPending)
	if perThread.dwpt.numDocsInRAM > 0 {
		perThread.flushPending = true  
		bytes := perThread.bytesUsed
		fc._flushBytes += bytes
		fc._activeBytes -= bytes
		fc.numPending++  
		fc.assertMemory()
	}
}
func (fc *DocumentsWriterFlushControl) doOnAbort(state *ThreadState) {
	fc.Lock()
	defer fc.Unlock()
	defer fc.updateStallState()
	if state.flushPending {
		fc._flushBytes -= state.bytesUsed
	} else {
		fc._activeBytes -= state.bytesUsed
	}
	fc.assertMemory()
	fc.perThreadPool.reset(state, fc.closed)
}
func (fc *DocumentsWriterFlushControl) tryCheckoutForFlush(perThread *ThreadState) *DocumentsWriterPerThread {
	fc.Lock()
	defer fc.Unlock()
	return fc._tryCheckoutForFlush(perThread)
}
func (fc *DocumentsWriterFlushControl) _tryCheckoutForFlush(perThread *ThreadState) *DocumentsWriterPerThread {
	if perThread.flushPending {
		return fc.internalTryCheckOutForFlush(perThread)
	}
	return nil
}
func (fc *DocumentsWriterFlushControl) checkoutAndBlock(perThread *ThreadState) {
	assert2(perThread.flushPending, "can not block non-pending threadstate")
	assert2(fc.fullFlush, "can not block if fullFlush == false")
	bytes := perThread.bytesUsed
	dwpt := fc.perThreadPool.reset(perThread, fc.closed)
	fc.numPending--
	fc.blockedFlushes.PushBack(&BlockedFlush{dwpt, bytes})
}
func (fc *DocumentsWriterFlushControl) internalTryCheckOutForFlush(perThread *ThreadState) *DocumentsWriterPerThread {
	assert(perThread.flushPending)
	defer fc.updateStallState()
	if perThread.isActive && perThread.dwpt != nil {
		bytes := perThread.bytesUsed  
		dwpt := fc.perThreadPool.reset(perThread, fc.closed)
		_, ok := fc.flushingWriters[dwpt]
		assert2(!ok, "DWPT is already flushing")
		fc.flushingWriters[dwpt] = bytes
		fc.numPending--  
		return dwpt
	}
	return nil
}
func (fc *DocumentsWriterFlushControl) String() string {
	return fmt.Sprintf("DocumentsWriterFlushControl [activeBytes=%v, flushBytes=%v]",
		fc.activeBytes, fc.flushBytes)
}
func (fc *DocumentsWriterFlushControl) close() {
	fc.Lock()
	defer fc.Unlock()
	if !fc.closed {
		fc.closed = true
	}
}
func (fc *DocumentsWriterFlushControl) deleteBytesUsed() int64 {
	return fc.documentsWriter.deleteQueue.RamBytesUsed() + fc.bufferedUpdatesStream.RamBytesUsed()
}
func (fc *DocumentsWriterFlushControl) obtainAndLock() *ThreadState {
	perThread := fc.perThreadPool.lockAny()
	var success = false
	defer func() {
		if !success {
			fc.perThreadPool.release(perThread)
		}
	}()
	if perThread.isActive &&
		perThread.dwpt != nil &&
		perThread.dwpt.deleteQueue != fc.documentsWriter.deleteQueue {
		fc.addFlushableState(perThread)
	}
	success = true
	return perThread
}
func (fc *DocumentsWriterFlushControl) markForFullFlush() {
	flushingQueue := func() *DocumentsWriterDeleteQueue {
		fc.Lock()
		defer fc.Unlock()
		assert2(!fc.fullFlush, "called DWFC#markForFullFlush() while full flush is still running")
		assertn(len(fc.fullFlushBuffer) == 0, "full flush buffer should be empty: ", fc.fullFlushBuffer)
		fc.fullFlush = true
		res := fc.documentsWriter.deleteQueue
		fc.documentsWriter.deleteQueue = newDocumentsWriterDeleteQueueWithGeneration(res.generation + 1)
		return res
	}()
	fc.perThreadPool.foreach(func(next *ThreadState) {
		if !next.isActive || next.dwpt == nil {
			if fc.closed && next.isActive {
				next.deactivate()
			}
			return
		}
		assertn(next.dwpt.deleteQueue == flushingQueue ||
			next.dwpt.deleteQueue == fc.documentsWriter.deleteQueue,
			" flushingQueue: %v currentQueue: %v perThread queue: %v numDocsInRAM: %v",
			flushingQueue, fc.documentsWriter.deleteQueue, next.dwpt.deleteQueue,
			next.dwpt.numDocsInRAM)
		if next.dwpt.deleteQueue != flushingQueue {
			return
		}
		fc.addFlushableState(next)
	})
	func() {
		fc.Lock()
		defer fc.Unlock()
		fc.pruneBlockedQueue(flushingQueue)
		fc.assertBlockedFlushes(fc.documentsWriter.deleteQueue)
		for _, dwpt := range fc.fullFlushBuffer {
			fc.flushQueue.PushBack(dwpt)
		}
		fc.fullFlushBuffer = nil
		fc.updateStallState()
	}()
	fc.assertActiveDeleteQueue(fc.documentsWriter.deleteQueue)
}
func (fc *DocumentsWriterFlushControl) assertActiveDeleteQueue(queue *DocumentsWriterDeleteQueue) {
	fc.perThreadPool.foreach(func(next *ThreadState) {
		n := 0
		if next.dwpt != nil {
			n = next.dwpt.numDocsInRAM
		}
		assertn(!next.isActive || next.dwpt == nil || next.dwpt.deleteQueue == queue,
			"isInitialized: %v numDocs: %v", next.isActive && next.dwpt != nil, n)
	})
}
func (fc *DocumentsWriterFlushControl) nextPendingFlush() *DocumentsWriterPerThread {
	numPending, fullFlush, dwpt := func() (int, bool, *DocumentsWriterPerThread) {
		fc.Lock()
		defer fc.Unlock()
		if e := fc.flushQueue.Front(); e != nil {
			fc.flushQueue.Remove(e)
			pool := e.Value.(*DocumentsWriterPerThread)
			fc.updateStallState()
			return 0, false, pool
		}
		return fc.numPending, fc.fullFlush, nil
	}()
	if dwpt != nil {
		return dwpt
	}
	if numPending > 0 && !fullFlush {
		dwpt = fc.perThreadPool.find(func(next *ThreadState) interface{} {
			if numPending > 0 && next.flushPending {
				if dwpt := fc.tryCheckoutForFlush(next); dwpt != nil {
					return dwpt
				}
			}
			return nil
		}).(*DocumentsWriterPerThread)
	}
	return dwpt
}
func (fc *DocumentsWriterFlushControl) addFlushableState(perThread *ThreadState) {
	if fc.infoStream.IsEnabled("DWFC") {
		fc.infoStream.Message("DWFC", "addFlushableState %v", perThread.dwpt)
	}
	dwpt := perThread.dwpt
	assert(perThread.isActive && perThread.dwpt != nil)
	assert(fc.fullFlush)
	assert(dwpt.deleteQueue != fc.documentsWriter.deleteQueue)
	if dwpt.numDocsInRAM > 0 {
		func() {
			fc.Lock()
			defer fc.Unlock()
			if !perThread.flushPending {
				fc._setFlushPending(perThread)
			}
			flushingDWPT := fc.internalTryCheckOutForFlush(perThread)
			assert2(flushingDWPT != nil, "DWPT must never be null here since we hold the lock and it holds documents")
			assert2(dwpt == flushingDWPT, "flushControl returned different DWPT")
			fc.fullFlushBuffer = append(fc.fullFlushBuffer, flushingDWPT)
		}()
	} else {
		fc.perThreadPool.reset(perThread, fc.closed)  
	}
}
func (fc *DocumentsWriterFlushControl) getAndResetApplyAllDeletes() bool {
	return atomic.SwapInt32(&fc.flushDeletes, 0) == 1
}
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")
			fc.flushingWriters[blockedFlush.dwpt] = blockedFlush.bytes
			fc.flushQueue.PushBack(blockedFlush.dwpt)
		}
	}
}
func (fc *DocumentsWriterFlushControl) finishFullFlush() {
	fc.Lock()
	defer fc.Unlock()
	assert(fc.fullFlush)
	assert(fc.flushQueue.Len() == 0)
	assert(len(fc.flushingWriters) == 0)
	defer func() { fc.fullFlush = false }()
	if fc.blockedFlushes.Len() > 0 {
		fc.assertBlockedFlushes(fc.documentsWriter.deleteQueue)
		fc.pruneBlockedQueue(fc.documentsWriter.deleteQueue)
		assert(fc.blockedFlushes.Len() == 0)
	}
}
func (fc *DocumentsWriterFlushControl) abortFullFlushes(newFiles map[string]bool) {
	fc.Lock()
	defer fc.Unlock()
	defer func() { fc.fullFlush = false }()
}
func (fc *DocumentsWriterFlushControl) assertBlockedFlushes(flushingQueue *DocumentsWriterDeleteQueue) {
	for e := fc.blockedFlushes.Front(); e != nil; e = e.Next() {
		blockedFlush := e.Value.(*BlockedFlush)
		assert(blockedFlush.dwpt.deleteQueue == flushingQueue)
	}
}
func (fc *DocumentsWriterFlushControl) abortPendingFlushes(newFiles map[string]bool) {
	fc.Lock()
	defer fc.Unlock()
	defer func() {
		fc.flushQueue.Init()
		fc.blockedFlushes.Init()
		fc.updateStallState()
	}()
	for e := fc.flushQueue.Front(); e != nil; e = e.Next() {
		dwpt := e.Value.(*DocumentsWriterPerThread)
		fc.documentsWriter.subtractFlushedNumDocs(dwpt.numDocsInRAM)
		fc.doAfterFlush(dwpt)
	}
	for e := fc.blockedFlushes.Front(); e != nil; e = e.Next() {
		blockedFlush := e.Value.(*BlockedFlush)
		fc.flushingWriters[blockedFlush.dwpt] = blockedFlush.bytes
		fc.documentsWriter.subtractFlushedNumDocs(blockedFlush.dwpt.numDocsInRAM)
		blockedFlush.dwpt.abort(newFiles)
		fc.doAfterFlush(blockedFlush.dwpt)
	}
}
func (fc *DocumentsWriterFlushControl) numQueuedFlushes() int {
	fc.Lock()
	defer fc.Unlock()
	return fc.flushQueue.Len()
}
type BlockedFlush struct {
	dwpt  *DocumentsWriterPerThread
	bytes int64
}