From 724684a4f11203536fea95b0faf84494dc8d67a9 Mon Sep 17 00:00:00 2001 From: Marty Schoch Date: Sun, 20 Mar 2016 11:02:13 -0400 Subject: [PATCH] additional firestorm fixes for 64-bit alignment part of #359 --- index/firestorm/dict_updater.go | 6 +++--- index/firestorm/firestorm.go | 10 +++++----- index/firestorm/lookup.go | 10 +++++----- index/firestorm/warmup.go | 2 +- 4 files changed, 14 insertions(+), 14 deletions(-) diff --git a/index/firestorm/dict_updater.go b/index/firestorm/dict_updater.go index 053ba0d4..c30fce26 100644 --- a/index/firestorm/dict_updater.go +++ b/index/firestorm/dict_updater.go @@ -22,6 +22,9 @@ const DefaultDictUpdateThreshold = 10 var DefaultDictUpdateSleep = 1 * time.Second type DictUpdater struct { + batchesStarted uint64 + batchesFlushed uint64 + f *Firestorm dictUpdateSleep time.Duration quit chan struct{} @@ -30,9 +33,6 @@ type DictUpdater struct { mutex sync.RWMutex workingSet map[string]int64 closeWait sync.WaitGroup - - batchesStarted uint64 - batchesFlushed uint64 } func NewDictUpdater(f *Firestorm) *DictUpdater { diff --git a/index/firestorm/firestorm.go b/index/firestorm/firestorm.go index 023ef439..78181959 100644 --- a/index/firestorm/firestorm.go +++ b/index/firestorm/firestorm.go @@ -27,14 +27,15 @@ const Name = "firestorm" var UnsafeBatchUseDetected = fmt.Errorf("bleve.Batch is NOT thread-safe, modification after execution detected") type Firestorm struct { + highDocNumber uint64 + docCount uint64 + storeName string storeConfig map[string]interface{} store store.KVStore compensator *Compensator analysisQueue *index.AnalysisQueue fieldCache *index.FieldCache - highDocNumber uint64 - docCount *uint64 garbageCollector *GarbageCollector lookuper *Lookuper dictUpdater *DictUpdater @@ -42,14 +43,13 @@ type Firestorm struct { } func NewFirestorm(storeName string, storeConfig map[string]interface{}, analysisQueue *index.AnalysisQueue) (index.Index, error) { - initialCount := uint64(0) rv := Firestorm{ storeName: storeName, storeConfig: storeConfig, compensator: NewCompensator(), analysisQueue: analysisQueue, fieldCache: index.NewFieldCache(), - docCount: &initialCount, + docCount: 0, highDocNumber: 0, stats: &indexStat{}, } @@ -130,7 +130,7 @@ func (f *Firestorm) Close() error { } func (f *Firestorm) DocCount() (uint64, error) { - count := atomic.LoadUint64(f.docCount) + count := atomic.LoadUint64(&f.docCount) return count, nil } diff --git a/index/firestorm/lookup.go b/index/firestorm/lookup.go index d58640e6..fee6c828 100644 --- a/index/firestorm/lookup.go +++ b/index/firestorm/lookup.go @@ -19,13 +19,13 @@ import ( const channelBufferSize = 1000 type Lookuper struct { + tasksQueued uint64 + tasksDone uint64 + f *Firestorm workChan chan []*InFlightItem quit chan struct{} closeWait sync.WaitGroup - - tasksQueued uint64 - tasksDone uint64 } func NewLookuper(f *Firestorm) *Lookuper { @@ -117,10 +117,10 @@ func (l *Lookuper) lookup(item *InFlightItem) { l.f.compensator.Migrate(item.docID, item.docNum, oldDocNums) if len(oldDocNums) == 0 && item.docNum != 0 { // this was an add, not an update - atomic.AddUint64(l.f.docCount, 1) + atomic.AddUint64(&l.f.docCount, 1) } else if len(oldDocNums) > 0 && item.docNum == 0 { // this was a delete (and it previously existed) - atomic.AddUint64(l.f.docCount, ^uint64(0)) + atomic.AddUint64(&l.f.docCount, ^uint64(0)) } } diff --git a/index/firestorm/warmup.go b/index/firestorm/warmup.go index 02e3b21a..4e20575a 100644 --- a/index/firestorm/warmup.go +++ b/index/firestorm/warmup.go @@ -90,7 +90,7 @@ func (f *Firestorm) warmup(reader store.KVReader) error { lastDocNumbers = append(lastDocNumbers, docNum) } else { // new doc id - atomic.AddUint64(f.docCount, 1) + atomic.AddUint64(&f.docCount, 1) // last docID had multiple doc numbers if len(lastDocNumbers) > 1 {