firestorm DictUpdater NotifyBatch is more async

index/firestorm/dict_updater.go

@@ -25,6 +25,7 @@ type DictUpdater struct {
 	f               *Firestorm
 	dictUpdateSleep time.Duration
 	quit            chan struct{}
+	incoming        chan map[string]int64
 
 	mutex      sync.RWMutex
 	workingSet map[string]int64
@@ -41,6 +42,7 @@ func NewDictUpdater(f *Firestorm) *DictUpdater {
 		workingSet:      make(map[string]int64),
 		batchesStarted:  1,
 		quit:            make(chan struct{}),
+		incoming:        make(chan map[string]int64, 8),
 	}
 	return &rv
 }
@@ -52,15 +54,12 @@ func (d *DictUpdater) Notify(term string, usage int64) {
 }
 
 func (d *DictUpdater) NotifyBatch(termUsages map[string]int64) {
-	d.mutex.Lock()
+	d.incoming <- termUsages
-	defer d.mutex.Unlock()
-	for term, usage := range termUsages {
-		d.workingSet[term] += usage
-	}
 }
 
 func (d *DictUpdater) Start() {
 	d.closeWait.Add(1)
+    go d.runIncoming()
 	go d.run()
 }
 
@@ -69,6 +68,24 @@ func (d *DictUpdater) Stop() {
 	d.closeWait.Wait()
 }
 
+func (d *DictUpdater) runIncoming() {
+	for {
+		select {
+		case <-d.quit:
+			return
+		case termUsages, ok := <-d.incoming:
+			if !ok {
+				return
+			}
+			d.mutex.Lock()
+			for term, usage := range termUsages {
+				d.workingSet[term] += usage
+			}
+			d.mutex.Unlock()
+		}
+	}
+}
+
 func (d *DictUpdater) run() {
 	tick := time.Tick(d.dictUpdateSleep)
 	for {

index/firestorm/dict_updater_test.go

@@ -10,6 +10,7 @@
 package firestorm
 
 import (
+	"runtime"
 	"testing"
 
 	"github.com/blevesearch/bleve/index"
@@ -38,6 +39,9 @@ func TestDictUpdater(t *testing.T) {
 	f.(*Firestorm).dictUpdater.NotifyBatch(dictBatch)
 
 	// invoke updater manually
+	for len(f.(*Firestorm).dictUpdater.incoming) > 0 {
+		runtime.Gosched()
+	}
 	f.(*Firestorm).dictUpdater.update()
 
 	// assert that dictionary rows are correct
@@ -77,6 +81,9 @@ func TestDictUpdater(t *testing.T) {
 	f.(*Firestorm).dictUpdater.NotifyBatch(dictBatch)
 
 	// invoke updater manually
+	for len(f.(*Firestorm).dictUpdater.incoming) > 0 {
+		runtime.Gosched()
+	}
 	f.(*Firestorm).dictUpdater.update()
 
 	// assert that dictionary rows are correct
@@ -116,6 +123,9 @@ func TestDictUpdater(t *testing.T) {
 	f.(*Firestorm).dictUpdater.NotifyBatch(dictBatch)
 
 	// invoke updater manually
+	for len(f.(*Firestorm).dictUpdater.incoming) > 0 {
+		runtime.Gosched()
+	}
 	f.(*Firestorm).dictUpdater.update()
 
 	// assert that dictionary rows are correct