experimental refactoring of channel communication

index/scorch/communication.go

@@ -0,0 +1,55 @@
+//  Copyright (c) 2020 Couchbase, Inc.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+// 		http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package scorch
+
+type notificationChan chan struct{}
+
+type epochWatcher struct {
+	epoch    uint64
+	notifyCh notificationChan
+}
+
+type epochWatchers []*epochWatcher
+
+func (e *epochWatchers) Add(watcher *epochWatcher) {
+	*e = append(*e, watcher)
+}
+
+func (e *epochWatchers) NotifySatisfiedWatchers(epoch uint64) {
+	var epochWatchersNext epochWatchers
+	for _, w := range *e {
+		if w.epoch < epoch {
+			close(w.notifyCh)
+		} else {
+			epochWatchersNext.Add(w)
+		}
+	}
+	*e = epochWatchersNext
+}
+
+type watcherChan chan *epochWatcher
+
+func (w watcherChan) NotifyUsAfter(epoch uint64, closeCh chan struct{}) (*epochWatcher, error) {
+	ew := &epochWatcher{
+		epoch:    epoch,
+		notifyCh: make(notificationChan, 1),
+	}
+	select {
+	case <-closeCh:
+		return nil, ErrClosed
+	case w <- ew:
+	}
+	return ew, nil
+}

index/scorch/introducer.go

@@ -40,13 +40,8 @@ type persistIntroduction struct {
 	applied   notificationChan
 }
 
-type epochWatcher struct {
-	epoch    uint64
-	notifyCh notificationChan
-}
-
 func (s *Scorch) introducerLoop() {
-	var epochWatchers []*epochWatcher
+	var introduceWatchers epochWatchers
 OUTER:
 	for {
 		atomic.AddUint64(&s.stats.TotIntroduceLoop, 1)
@@ -56,7 +51,7 @@ OUTER:
 			break OUTER
 
 		case epochWatcher := <-s.introducerNotifier:
-			epochWatchers = append(epochWatchers, epochWatcher)
+			introduceWatchers.Add(epochWatcher)
 
 		case nextMerge := <-s.merges:
 			s.introduceMerge(nextMerge)
@@ -78,15 +73,7 @@ OUTER:
 			epochCurr = s.root.epoch
 		}
 		s.rootLock.RUnlock()
-		var epochWatchersNext []*epochWatcher
-		for _, w := range epochWatchers {
-			if w.epoch < epochCurr {
-				close(w.notifyCh)
-			} else {
-				epochWatchersNext = append(epochWatchersNext, w)
-			}
-		}
-		epochWatchers = epochWatchersNext
+		introduceWatchers.NotifySatisfiedWatchers(epochCurr)
 	}
 
 	s.asyncTasks.Done()

index/scorch/merge.go

@@ -27,12 +27,20 @@ import (
 	"github.com/blevesearch/bleve/index/scorch/segment"
 )
 
-func (s *Scorch) mergerLoop() {
+func (s *Scorch) mergerLoop(initialEpoch uint64) {
+	defer s.asyncTasks.Done()
+
 	var lastEpochMergePlanned uint64
 	mergePlannerOptions, err := s.parseMergePlannerOptions()
 	if err != nil {
 		s.fireAsyncError(fmt.Errorf("mergePlannerOption json parsing err: %v", err))
-		s.asyncTasks.Done()
+		return
+	}
+
+	// tell the persister we're waiting for anything after the initialEpoch
+	var ew *epochWatcher
+	ew, err = s.persisterNotifier.NotifyUsAfter(initialEpoch, s.closeCh)
+	if err != nil {
 		return
 	}
 
@@ -44,7 +52,7 @@ OUTER:
 		case <-s.closeCh:
 			break OUTER
 
-		default:
+		case <-ew.notifyCh:
 			// check to see if there is a new snapshot to persist
 			s.rootLock.Lock()
 			ourSnapshot := s.root
@@ -78,32 +86,16 @@ OUTER:
 			}
 			_ = ourSnapshot.DecRef()
 
-			// tell the persister we're waiting for changes
-			// first make a epochWatcher chan
-			ew := &epochWatcher{
-				epoch:    lastEpochMergePlanned,
-				notifyCh: make(notificationChan, 1),
-			}
-
-			// give it to the persister
-			select {
-			case <-s.closeCh:
-				break OUTER
-			case s.persisterNotifier <- ew:
-			}
-
-			// now wait for persister (but also detect close)
-			select {
-			case <-s.closeCh:
+			// update the persister, that we're now waiting for something
+			// after lastEpochMergePlanned
+			ew, err = s.persisterNotifier.NotifyUsAfter(lastEpochMergePlanned, s.closeCh)
+			if err != nil {
 				break OUTER
-			case <-ew.notifyCh:
 			}
 		}
 
 		atomic.AddUint64(&s.stats.TotFileMergeLoopEnd, 1)
 	}
-
-	s.asyncTasks.Done()
 }
 
 func (s *Scorch) parseMergePlannerOptions() (*mergeplan.MergePlanOptions,