gibheer/bleve
0
0
Fork 0
Code

upside_down retrieves backindex concurrently with analysis

Browse Source 
Start backindex reading concurrently with analysi to try to utilize
more I/O bandwidth.

The analysis time vs indexing time stats tracking are also now "off",
since there's now concurrency between those actiivties.

One tradeoff is that the lock area in upside_down Batch() is increased
as part of this change.
This commit is contained in:
Steve Yen 2016-01-10 09:41:12 -08:00
parent bff95eef70
commit d3dd40d334
1 changed files with 46 additions and 32 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

78
index/upside_down/upside_down.go
View File

@ -710,7 +710,7 @@ func (udc *UpsideDownCouch) termFieldVectorsFromTermVectors(in []*TermVector) []
func (udc *UpsideDownCouch) Batch(batch *index.Batch) (err error) {
analysisStart := time.Now()
resultChan := make(chan *index.AnalysisResult)
resultChan := make(chan *index.AnalysisResult, len(batch.IndexOps))
var numUpdates uint64
for _, doc := range batch.IndexOps {
@ -740,8 +740,40 @@ func (udc *UpsideDownCouch) Batch(batch *index.Batch) (err error) {
}
}()
newRowsMap := make(map[string][]index.IndexRow)
udc.writeMutex.Lock()
defer udc.writeMutex.Unlock()
var backIndexRows map[string]*BackIndexRow
backindexReaderCh := make(chan error)
go func() {
defer close(backindexReaderCh)
// open a reader for backindex lookup
var kvreader store.KVReader
kvreader, err = udc.store.Reader()
if err != nil {
backindexReaderCh <- err
return
}
// first lookup all the back index rows
backIndexRows, err = udc.backIndexRowsForBatch(kvreader, batch)
if err != nil {
_ = kvreader.Close()
backindexReaderCh <- err
return
}
err = kvreader.Close()
if err != nil {
backindexReaderCh <- err
return
}
}()
// wait for the result
newRowsMap := make(map[string][]index.IndexRow)
var itemsDeQueued uint64
for itemsDeQueued < numUpdates {
result := <-resultChan
@ -750,6 +782,11 @@ func (udc *UpsideDownCouch) Batch(batch *index.Batch) (err error) {
}
close(resultChan)
backindexReaderErr := <-backindexReaderCh
if backindexReaderErr != nil {
return backindexReaderErr
}
detectedUnsafeMutex.RLock()
defer detectedUnsafeMutex.RUnlock()
if detectedUnsafe {
@ -760,36 +797,6 @@ func (udc *UpsideDownCouch) Batch(batch *index.Batch) (err error) {
indexStart := time.Now()
udc.writeMutex.Lock()
defer udc.writeMutex.Unlock()
// open a reader for backindex lookup
var kvreader store.KVReader
kvreader, err = udc.store.Reader()
if err != nil {
return
}
// first lookup all the back index rows
var backIndexRows map[string]*BackIndexRow
backIndexRows, err = udc.backIndexRowsForBatch(kvreader, batch)
if err != nil {
_ = kvreader.Close()
return
}
err = kvreader.Close()
if err != nil {
return
}
// start a writer for this batch
var kvwriter store.KVWriter
kvwriter, err = udc.store.Writer()
if err != nil {
return
}
// prepare a list of rows
addRows := make([]UpsideDownCouchRow, 0)
updateRows := make([]UpsideDownCouchRow, 0)
@ -823,6 +830,13 @@ func (udc *UpsideDownCouch) Batch(batch *index.Batch) (err error) {
}
}
// start a writer for this batch
var kvwriter store.KVWriter
kvwriter, err = udc.store.Writer()
if err != nil {
return
}
err = udc.batchRows(kvwriter, addRows, updateRows, deleteRows)
if err != nil {
_ = kvwriter.Close()