From 30181d459da79ef57c58bb3789e2ce7130d08b77 Mon Sep 17 00:00:00 2001 From: zeripath Date: Tue, 24 Dec 2019 07:26:34 +0000 Subject: [PATCH] Wrap the code indexer (#9476) * Wrap the code indexer In order to prevent a data race in the code indexer it must be wrapped with a holder otherwise it is possible to Search/Index on an incompletely initialised indexer, and search will fail with a nil pointer until the repository indexer is initialised. Further a completely initialised repository indexer should not be closed until Termination otherwise actions in Hammer/Shutdown phases could block or be lost. Finally, there is a complex dance of shutdown etiquette should the index initialisation fail. This PR restores that. * Always return err if closed whilst waiting Co-authored-by: techknowlogick --- modules/indexer/code/indexer.go | 49 +++++++++++++---- modules/indexer/code/queue.go | 2 - modules/indexer/code/wrapped.go | 94 +++++++++++++++++++++++++++++++++ 3 files changed, 132 insertions(+), 13 deletions(-) create mode 100644 modules/indexer/code/wrapped.go diff --git a/modules/indexer/code/indexer.go b/modules/indexer/code/indexer.go index c68c7c2d7..04d556f54 100644 --- a/modules/indexer/code/indexer.go +++ b/modules/indexer/code/indexer.go @@ -5,6 +5,8 @@ package code import ( + "context" + "os" "time" "code.gitea.io/gitea/modules/graceful" @@ -12,10 +14,6 @@ import ( "code.gitea.io/gitea/modules/setting" ) -var ( - indexer Indexer -) - // SearchResult result of performing a search in a repo type SearchResult struct { RepoID int64 @@ -36,28 +34,45 @@ type Indexer interface { // Init initialize the repo indexer func Init() { if !setting.Indexer.RepoIndexerEnabled { + indexer.Close() return } + ctx, cancel := context.WithCancel(context.Background()) + + graceful.GetManager().RunAtTerminate(ctx, func() { + log.Debug("Closing repository indexer") + indexer.Close() + log.Info("PID: %d Repository Indexer closed", os.Getpid()) + }) + waitChannel := make(chan time.Duration) go func() { start := time.Now() - log.Info("Initializing Repository Indexer") - var created bool - var err error - indexer, created, err = NewBleveIndexer(setting.Indexer.RepoPath) + log.Info("PID: %d Initializing Repository Indexer at: %s", os.Getpid(), setting.Indexer.RepoPath) + bleveIndexer, created, err := NewBleveIndexer(setting.Indexer.RepoPath) if err != nil { + if bleveIndexer != nil { + bleveIndexer.Close() + } + cancel() indexer.Close() - log.Fatal("indexer.Init: %v", err) + close(waitChannel) + log.Fatal("PID: %d Unable to initialize the Repository Indexer at path: %s Error: %v", os.Getpid(), setting.Indexer.RepoPath, err) } + indexer.set(bleveIndexer) go processRepoIndexerOperationQueue(indexer) if created { go populateRepoIndexer() } + select { + case waitChannel <- time.Since(start): + case <-graceful.GetManager().IsShutdown(): + } - waitChannel <- time.Since(start) + close(waitChannel) }() if setting.Indexer.StartupTimeout > 0 { @@ -67,9 +82,21 @@ func Init() { timeout += setting.GracefulHammerTime } select { - case duration := <-waitChannel: + case <-graceful.GetManager().IsShutdown(): + log.Warn("Shutdown before Repository Indexer completed initialization") + cancel() + indexer.Close() + case duration, ok := <-waitChannel: + if !ok { + log.Warn("Repository Indexer Initialization failed") + cancel() + indexer.Close() + return + } log.Info("Repository Indexer Initialization took %v", duration) case <-time.After(timeout): + cancel() + indexer.Close() log.Fatal("Repository Indexer Initialization Timed-Out after: %v", timeout) } }() diff --git a/modules/indexer/code/queue.go b/modules/indexer/code/queue.go index 90f7d1bb1..82cd8ded5 100644 --- a/modules/indexer/code/queue.go +++ b/modules/indexer/code/queue.go @@ -22,8 +22,6 @@ type repoIndexerOperation struct { var repoIndexerOperationQueue chan repoIndexerOperation func processRepoIndexerOperationQueue(indexer Indexer) { - defer indexer.Close() - repoIndexerOperationQueue = make(chan repoIndexerOperation, setting.Indexer.UpdateQueueLength) for { select { diff --git a/modules/indexer/code/wrapped.go b/modules/indexer/code/wrapped.go new file mode 100644 index 000000000..6a2088398 --- /dev/null +++ b/modules/indexer/code/wrapped.go @@ -0,0 +1,94 @@ +// Copyright 2019 The Gitea Authors. All rights reserved. +// Use of this source code is governed by a MIT-style +// license that can be found in the LICENSE file. + +package code + +import ( + "fmt" + "sync" +) + +var ( + indexer = newWrappedIndexer() +) + +// ErrWrappedIndexerClosed is the error returned if the indexer was closed before it was ready +var ErrWrappedIndexerClosed = fmt.Errorf("Indexer closed before ready") + +type wrappedIndexer struct { + internal Indexer + lock sync.RWMutex + cond *sync.Cond + closed bool +} + +func newWrappedIndexer() *wrappedIndexer { + w := &wrappedIndexer{} + w.cond = sync.NewCond(w.lock.RLocker()) + return w +} + +func (w *wrappedIndexer) set(indexer Indexer) { + w.lock.Lock() + defer w.lock.Unlock() + if w.closed { + // Too late! + indexer.Close() + } + w.internal = indexer + w.cond.Broadcast() +} + +func (w *wrappedIndexer) get() (Indexer, error) { + w.lock.RLock() + defer w.lock.RUnlock() + if w.internal == nil { + if w.closed { + return nil, ErrWrappedIndexerClosed + } + w.cond.Wait() + if w.closed { + return nil, ErrWrappedIndexerClosed + } + } + return w.internal, nil +} + +func (w *wrappedIndexer) Index(repoID int64) error { + indexer, err := w.get() + if err != nil { + return err + } + return indexer.Index(repoID) +} + +func (w *wrappedIndexer) Delete(repoID int64) error { + indexer, err := w.get() + if err != nil { + return err + } + return indexer.Delete(repoID) +} + +func (w *wrappedIndexer) Search(repoIDs []int64, keyword string, page, pageSize int) (int64, []*SearchResult, error) { + indexer, err := w.get() + if err != nil { + return 0, nil, err + } + return indexer.Search(repoIDs, keyword, page, pageSize) + +} + +func (w *wrappedIndexer) Close() { + w.lock.Lock() + defer w.lock.Unlock() + if w.closed { + return + } + w.closed = true + w.cond.Broadcast() + if w.internal != nil { + w.internal.Close() + } +}