Function execute

pindex_bleve.go:2532–2578

(bdp []*BleveDestPartition, bdpMaxSeqNums []uint64,
	bindex bleve.Index, batch *bleve.Batch)

}

func execute(bdp []*BleveDestPartition, bdpMaxSeqNums []uint64,
	bindex bleve.Index, batch *bleve.Batch) (bool, error) {
	if batch == nil {
		return false, fmt.Errorf("pindex_bleve: executeBatch batch nil")
	}

	if bindex == nil {
		return false, fmt.Errorf("pindex_bleve: executeBatch bindex already closed")
	}

	batchTotalDocsSize := batch.TotalDocsSize()
	atomic.AddUint64(&BatchBytesAdded, batchTotalDocsSize)

	err := cbgt.Timer(func() error {
		atomic.AddUint64(&aggregateBDPStats.TotExecuteBatchBeg, 1)
		err := bindex.Batch(batch)
		atomic.AddUint64(&aggregateBDPStats.TotExecuteBatchEnd, 1)
		if err != nil && err != bleve.ErrorIndexClosed {
			log.Errorf("pindex_bleve: executeBatch over `%v`, err: %+v ",
				bindex.Name(), err)
		}
		return err
	}, bdp[0].bdest.stats.TimerBatchStore)

	if err != nil {
		return false, err
	}

	atomic.AddUint64(&BatchBytesRemoved, batchTotalDocsSize)

	for i, t := range bdp {
		t.m.Lock()
		if bdpMaxSeqNums[i] > atomic.LoadUint64(&t.seqMaxBatch) {
			atomic.StoreUint64(&t.seqMaxBatch, bdpMaxSeqNums[i])
		}
		for t.cwrQueue.Len() > 0 &&
			t.cwrQueue[0].ConsistencySeq <= atomic.LoadUint64(&t.seqMaxBatch) {
			cwr := heap.Pop(&t.cwrQueue).(*cbgt.ConsistencyWaitReq)
			if cwr != nil && cwr.DoneCh != nil {
				close(cwr.DoneCh)
			}
		}
		t.m.Unlock()
	}

	return true, nil
}

// ---------------------------------------------------------