1
0
Fork 0
mirror of https://codeberg.org/forgejo/forgejo.git synced 2025-01-09 15:28:22 -05:00

Prevent deadlock during early shutdown of issue indexer

This commit is contained in:
Andrew Thornton 2019-12-23 10:29:36 +00:00
parent 1fb9104009
commit a492b3071c
No known key found for this signature in database
GPG key ID: 3CDE74631F13A748

View file

@ -6,6 +6,7 @@ package issues
import ( import (
"context" "context"
"fmt"
"os" "os"
"sync" "sync"
"time" "time"
@ -51,9 +52,10 @@ type Indexer interface {
} }
type indexerHolder struct { type indexerHolder struct {
indexer Indexer indexer Indexer
mutex sync.RWMutex mutex sync.RWMutex
cond *sync.Cond cond *sync.Cond
cancelled bool
} }
func newIndexerHolder() *indexerHolder { func newIndexerHolder() *indexerHolder {
@ -62,6 +64,13 @@ func newIndexerHolder() *indexerHolder {
return h return h
} }
func (h *indexerHolder) cancel() {
h.mutex.Lock()
defer h.mutex.Unlock()
h.cancelled = true
h.cond.Broadcast()
}
func (h *indexerHolder) set(indexer Indexer) { func (h *indexerHolder) set(indexer Indexer) {
h.mutex.Lock() h.mutex.Lock()
defer h.mutex.Unlock() defer h.mutex.Unlock()
@ -72,7 +81,7 @@ func (h *indexerHolder) set(indexer Indexer) {
func (h *indexerHolder) get() Indexer { func (h *indexerHolder) get() Indexer {
h.mutex.RLock() h.mutex.RLock()
defer h.mutex.RUnlock() defer h.mutex.RUnlock()
if h.indexer == nil { if h.indexer == nil && !h.cancelled {
h.cond.Wait() h.cond.Wait()
} }
return h.indexer return h.indexer
@ -93,6 +102,12 @@ func InitIssueIndexer(syncReindex bool) {
switch setting.Indexer.IssueType { switch setting.Indexer.IssueType {
case "bleve": case "bleve":
handler := func(data ...queue.Data) { handler := func(data ...queue.Data) {
indexer := holder.get()
if indexer == nil {
log.Error("Unable to get indexer!")
return
}
iData := make([]*IndexerData, 0, setting.Indexer.IssueQueueBatchNumber) iData := make([]*IndexerData, 0, setting.Indexer.IssueQueueBatchNumber)
for _, datum := range data { for _, datum := range data {
indexerData, ok := datum.(*IndexerData) indexerData, ok := datum.(*IndexerData)
@ -102,12 +117,12 @@ func InitIssueIndexer(syncReindex bool) {
} }
log.Trace("IndexerData Process: %d %v %t", indexerData.ID, indexerData.IDs, indexerData.IsDelete) log.Trace("IndexerData Process: %d %v %t", indexerData.ID, indexerData.IDs, indexerData.IsDelete)
if indexerData.IsDelete { if indexerData.IsDelete {
_ = holder.get().Delete(indexerData.IDs...) _ = indexer.Delete(indexerData.IDs...)
continue continue
} }
iData = append(iData, indexerData) iData = append(iData, indexerData)
} }
if err := holder.get().Index(iData); err != nil { if err := indexer.Index(iData); err != nil {
log.Error("Error whilst indexing: %v Error: %v", iData, err) log.Error("Error whilst indexing: %v Error: %v", iData, err)
} }
} }
@ -132,6 +147,7 @@ func InitIssueIndexer(syncReindex bool) {
issueIndexer := NewBleveIndexer(setting.Indexer.IssuePath) issueIndexer := NewBleveIndexer(setting.Indexer.IssuePath)
exist, err := issueIndexer.Init() exist, err := issueIndexer.Init()
if err != nil { if err != nil {
holder.cancel()
log.Fatal("Unable to initialize Bleve Issue Indexer: %v", err) log.Fatal("Unable to initialize Bleve Issue Indexer: %v", err)
} }
populate = !exist populate = !exist
@ -153,6 +169,7 @@ func InitIssueIndexer(syncReindex bool) {
issueIndexer := &DBIndexer{} issueIndexer := &DBIndexer{}
holder.set(issueIndexer) holder.set(issueIndexer)
default: default:
holder.cancel()
log.Fatal("Unknown issue indexer type: %s", setting.Indexer.IssueType) log.Fatal("Unknown issue indexer type: %s", setting.Indexer.IssueType)
} }
@ -168,10 +185,14 @@ func InitIssueIndexer(syncReindex bool) {
} }
} }
waitChannel <- time.Since(start) waitChannel <- time.Since(start)
close(waitChannel)
}() }()
if syncReindex { if syncReindex {
<-waitChannel select {
case <-waitChannel:
case <-graceful.GetManager().IsShutdown():
}
} else if setting.Indexer.StartupTimeout > 0 { } else if setting.Indexer.StartupTimeout > 0 {
go func() { go func() {
timeout := setting.Indexer.StartupTimeout timeout := setting.Indexer.StartupTimeout
@ -181,6 +202,8 @@ func InitIssueIndexer(syncReindex bool) {
select { select {
case duration := <-waitChannel: case duration := <-waitChannel:
log.Info("Issue Indexer Initialization took %v", duration) log.Info("Issue Indexer Initialization took %v", duration)
case <-graceful.GetManager().IsShutdown():
log.Warn("Shutdown occurred before issue index initialisation was complete")
case <-time.After(timeout): case <-time.After(timeout):
if shutdownable, ok := issueIndexerQueue.(queue.Shutdownable); ok { if shutdownable, ok := issueIndexerQueue.(queue.Shutdownable); ok {
shutdownable.Terminate() shutdownable.Terminate()
@ -293,7 +316,13 @@ func DeleteRepoIssueIndexer(repo *models.Repository) {
// SearchIssuesByKeyword search issue ids by keywords and repo id // SearchIssuesByKeyword search issue ids by keywords and repo id
func SearchIssuesByKeyword(repoIDs []int64, keyword string) ([]int64, error) { func SearchIssuesByKeyword(repoIDs []int64, keyword string) ([]int64, error) {
var issueIDs []int64 var issueIDs []int64
res, err := holder.get().Search(keyword, repoIDs, 1000, 0) indexer := holder.get()
if indexer == nil {
log.Error("Unable to get indexer!")
return nil, fmt.Errorf("unable to get issue indexer")
}
res, err := indexer.Search(keyword, repoIDs, 1000, 0)
if err != nil { if err != nil {
return nil, err return nil, err
} }