* 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 <matti@mdranta.net>master
@@ -5,6 +5,8 @@ | |||||
package code | package code | ||||
import ( | import ( | ||||
"context" | |||||
"os" | |||||
"time" | "time" | ||||
"code.gitea.io/gitea/modules/graceful" | "code.gitea.io/gitea/modules/graceful" | ||||
@@ -12,10 +14,6 @@ import ( | |||||
"code.gitea.io/gitea/modules/setting" | "code.gitea.io/gitea/modules/setting" | ||||
) | ) | ||||
var ( | |||||
indexer Indexer | |||||
) | |||||
// SearchResult result of performing a search in a repo | // SearchResult result of performing a search in a repo | ||||
type SearchResult struct { | type SearchResult struct { | ||||
RepoID int64 | RepoID int64 | ||||
@@ -36,28 +34,45 @@ type Indexer interface { | |||||
// Init initialize the repo indexer | // Init initialize the repo indexer | ||||
func Init() { | func Init() { | ||||
if !setting.Indexer.RepoIndexerEnabled { | if !setting.Indexer.RepoIndexerEnabled { | ||||
indexer.Close() | |||||
return | 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) | waitChannel := make(chan time.Duration) | ||||
go func() { | go func() { | ||||
start := time.Now() | 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 err != nil { | ||||
if bleveIndexer != nil { | |||||
bleveIndexer.Close() | |||||
} | |||||
cancel() | |||||
indexer.Close() | 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) | go processRepoIndexerOperationQueue(indexer) | ||||
if created { | if created { | ||||
go populateRepoIndexer() | go populateRepoIndexer() | ||||
} | } | ||||
select { | |||||
case waitChannel <- time.Since(start): | |||||
case <-graceful.GetManager().IsShutdown(): | |||||
} | |||||
waitChannel <- time.Since(start) | |||||
close(waitChannel) | |||||
}() | }() | ||||
if setting.Indexer.StartupTimeout > 0 { | if setting.Indexer.StartupTimeout > 0 { | ||||
@@ -67,9 +82,21 @@ func Init() { | |||||
timeout += setting.GracefulHammerTime | timeout += setting.GracefulHammerTime | ||||
} | } | ||||
select { | 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) | log.Info("Repository Indexer Initialization took %v", duration) | ||||
case <-time.After(timeout): | case <-time.After(timeout): | ||||
cancel() | |||||
indexer.Close() | |||||
log.Fatal("Repository Indexer Initialization Timed-Out after: %v", timeout) | log.Fatal("Repository Indexer Initialization Timed-Out after: %v", timeout) | ||||
} | } | ||||
}() | }() | ||||
@@ -22,8 +22,6 @@ type repoIndexerOperation struct { | |||||
var repoIndexerOperationQueue chan repoIndexerOperation | var repoIndexerOperationQueue chan repoIndexerOperation | ||||
func processRepoIndexerOperationQueue(indexer Indexer) { | func processRepoIndexerOperationQueue(indexer Indexer) { | ||||
defer indexer.Close() | |||||
repoIndexerOperationQueue = make(chan repoIndexerOperation, setting.Indexer.UpdateQueueLength) | repoIndexerOperationQueue = make(chan repoIndexerOperation, setting.Indexer.UpdateQueueLength) | ||||
for { | for { | ||||
select { | select { | ||||
@@ -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() | |||||
} | |||||
} |