search indexing process coupling

domain/search/queue.go

@@ -0,0 +1,142 @@
+// Copyright 2016 Documize Inc. <legal@documize.com>. All rights reserved.
+//
+// This software (Documize Community Edition) is licensed under
+// GNU AGPL v3 http://www.gnu.org/licenses/agpl-3.0.en.html
+//
+// You can operate outside the AGPL restrictions by purchasing
+// Documize Enterprise Edition and obtaining a commercial license
+// by contacting <sales@documize.com>.
+//
+// https://documize.com
+
+package search
+
+import (
+	"errors"
+	"fmt"
+	"sync"
+
+	"github.com/documize/community/core/env"
+	"github.com/documize/community/domain"
+	"github.com/documize/community/model/page"
+)
+
+// Indexer type provides the datastructure for the queues of activity to be serialized through a single background goroutine.
+// NOTE if the queue becomes full, the system will trigger the rebuilding entire files in order to clear the backlog.
+type Indexer struct {
+	queue        chan queueEntry
+	rebuild      map[string]bool
+	rebuildLock  sync.RWMutex
+	givenWarning bool
+	runtime      *env.Runtime
+	store        *domain.Store
+}
+
+type queueEntry struct {
+	action    func(domain.RequestContext, page.Page) error
+	isRebuild bool
+	page.Page
+	ctx domain.RequestContext
+}
+
+const searchQueueLength = 2048 // NOTE the largest 15Mb docx in the test set generates 2142 queue entries, but the queue is constantly emptied
+
+// NewIndexer provides background search indexer
+func NewIndexer(rt *env.Runtime, s *domain.Store) (i Indexer) {
+	i = Indexer{}
+	i.queue = make(chan queueEntry, searchQueueLength) // provide some decoupling
+	i.rebuild = make(map[string]bool)
+	i.runtime = rt
+	i.store = s
+
+	go i.processQueue()
+
+	return
+}
+
+// processQueue is run as a goroutine, it processes the queue of search index update requests.
+func (m *Indexer) processQueue() {
+	for {
+		//fmt.Println("DEBUG queue length=", len(Searches.queue))
+		if len(m.queue) <= searchQueueLength/20 { // on a busy server, the queue may never get to zero - so use 5%
+			m.rebuildLock.Lock()
+			for docid := range m.rebuild {
+				m.queue <- queueEntry{
+					action:    m.store.Search.Rebuild,
+					isRebuild: true,
+					Page:      page.Page{DocumentID: docid},
+				}
+				delete(m.rebuild, docid)
+			}
+			m.rebuildLock.Unlock()
+		}
+
+		qe := <-m.queue
+		doit := true
+
+		if len(qe.DocumentID) > 0 {
+			m.rebuildLock.RLock()
+			if m.rebuild[qe.DocumentID] {
+				doit = false // don't execute an action on a document queued to be rebuilt
+			}
+			m.rebuildLock.RUnlock()
+		}
+
+		if doit {
+			tx, err := m.runtime.Db.Beginx()
+			if err != nil {
+			} else {
+				ctx := qe.ctx
+				ctx.Transaction = tx
+				err = qe.action(ctx, qe.Page)
+				if err != nil {
+					tx.Rollback()
+					// This action has failed, so re-build indexes for the entire document,
+					// provided it was not a re-build command that failed and we know the documentId.
+					if !qe.isRebuild && len(qe.DocumentID) > 0 {
+						m.rebuildLock.Lock()
+						m.rebuild[qe.DocumentID] = true
+						m.rebuildLock.Unlock()
+					}
+				} else {
+					tx.Commit()
+				}
+			}
+		}
+	}
+}
+
+func (m *Indexer) addQueue(qe queueEntry) error {
+	lsq := len(m.queue)
+
+	if lsq >= (searchQueueLength - 1) {
+		if qe.DocumentID != "" {
+			m.rebuildLock.Lock()
+			if !m.rebuild[qe.DocumentID] {
+				m.runtime.Log.Info(fmt.Sprintf("WARNING: Search Queue Has No Space! Marked rebuild index for document id %s", qe.DocumentID))
+			}
+			m.rebuild[qe.DocumentID] = true
+			m.rebuildLock.Unlock()
+		} else {
+			m.runtime.Log.Error("addQueue", errors.New("WARNING: Search Queue Has No Space! But unable to index unknown document id"))
+		}
+
+		return nil
+	}
+
+	if lsq > ((8 * searchQueueLength) / 10) {
+		if !m.givenWarning {
+			m.runtime.Log.Info(fmt.Sprintf("WARNING: Searches.queue length %d exceeds 80%% of capacity", lsq))
+			m.givenWarning = true
+		}
+	} else {
+		if m.givenWarning {
+			m.runtime.Log.Info(fmt.Sprintf("INFO: Searches.queue length %d now below 80%% of capacity", lsq))
+			m.givenWarning = false
+		}
+	}
+
+	m.queue <- qe
+
+	return nil
+}