Skip to content

Commit f6f7beb

Browse files
lafrikslafriks
committed
Fix locking and correct context usage
1 parent 09ed40d commit f6f7beb

File tree

17 files changed

+118
-68
lines changed

17 files changed

+118
-68
lines changed

‎models/issue.go‎

Lines changed: 3 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -1859,7 +1859,7 @@ func GetRepoIssueStats(repoID, uid int64, filterMode int, isPull bool) (numOpen,
18591859
}
18601860

18611861
// SearchIssueIDsByKeyword search issues on database
1862-
func SearchIssueIDsByKeyword(kw string, repoIDs []int64, limit, start int) (int64, []int64, error) {
1862+
func SearchIssueIDsByKeyword(ctx context.Context, kw string, repoIDs []int64, limit, start int) (int64, []int64, error) {
18631863
repoCond := builder.In("repo_id", repoIDs)
18641864
subQuery := builder.Select("id").From("issue").Where(repoCond)
18651865
kw = strings.ToUpper(kw)
@@ -1884,7 +1884,7 @@ func SearchIssueIDsByKeyword(kw string, repoIDs []int64, limit, start int) (int6
18841884
ID int64
18851885
UpdatedUnix int64
18861886
}, 0, limit)
1887-
err := db.GetEngine(db.DefaultContext).Distinct("id", "updated_unix").Table("issue").Where(cond).
1887+
err := db.GetEngine(ctx).Distinct("id", "updated_unix").Table("issue").Where(cond).
18881888
OrderBy("`updated_unix` DESC").Limit(limit, start).
18891889
Find(&res)
18901890
if err != nil {
@@ -1894,7 +1894,7 @@ func SearchIssueIDsByKeyword(kw string, repoIDs []int64, limit, start int) (int6
18941894
ids = append(ids, r.ID)
18951895
}
18961896

1897-
total, err := db.GetEngine(db.DefaultContext).Distinct("id").Table("issue").Where(cond).Count()
1897+
total, err := db.GetEngine(ctx).Distinct("id").Table("issue").Where(cond).Count()
18981898
if err != nil {
18991899
return 0, nil, err
19001900
}

‎models/issue_test.go‎

Lines changed: 6 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -5,6 +5,7 @@
55
package models
66

77
import (
8+
"context"
89
"fmt"
910
"sort"
1011
"sync"
@@ -303,23 +304,23 @@ func TestIssue_loadTotalTimes(t *testing.T) {
303304

304305
func TestIssue_SearchIssueIDsByKeyword(t *testing.T) {
305306
assert.NoError(t, unittest.PrepareTestDatabase())
306-
total, ids, err := SearchIssueIDsByKeyword("issue2", []int64{1}, 10, 0)
307+
total, ids, err := SearchIssueIDsByKeyword(context.TODO(), "issue2", []int64{1}, 10, 0)
307308
assert.NoError(t, err)
308309
assert.EqualValues(t, 1, total)
309310
assert.EqualValues(t, []int64{2}, ids)
310311

311-
total, ids, err = SearchIssueIDsByKeyword("first", []int64{1}, 10, 0)
312+
total, ids, err = SearchIssueIDsByKeyword(context.TODO(), "first", []int64{1}, 10, 0)
312313
assert.NoError(t, err)
313314
assert.EqualValues(t, 1, total)
314315
assert.EqualValues(t, []int64{1}, ids)
315316

316-
total, ids, err = SearchIssueIDsByKeyword("for", []int64{1}, 10, 0)
317+
total, ids, err = SearchIssueIDsByKeyword(context.TODO(), "for", []int64{1}, 10, 0)
317318
assert.NoError(t, err)
318319
assert.EqualValues(t, 5, total)
319320
assert.ElementsMatch(t, []int64{1, 2, 3, 5, 11}, ids)
320321

321322
// issue1's comment id 2
322-
total, ids, err = SearchIssueIDsByKeyword("good", []int64{1}, 10, 0)
323+
total, ids, err = SearchIssueIDsByKeyword(context.TODO(), "good", []int64{1}, 10, 0)
323324
assert.NoError(t, err)
324325
assert.EqualValues(t, 1, total)
325326
assert.EqualValues(t, []int64{1}, ids)
@@ -464,7 +465,7 @@ func TestCorrectIssueStats(t *testing.T) {
464465
wg.Wait()
465466

466467
// Now we will get all issueID's that match the "Bugs are nasty" query.
467-
total, ids, err := SearchIssueIDsByKeyword("Bugs are nasty", []int64{1}, issueAmount, 0)
468+
total, ids, err := SearchIssueIDsByKeyword(context.TODO(), "Bugs are nasty", []int64{1}, issueAmount, 0)
468469

469470
// Just to be sure.
470471
assert.NoError(t, err)

‎modules/indexer/code/bleve.go‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -328,7 +328,7 @@ func (b *BleveIndexer) Delete(repoID int64) error {
328328

329329
// Search searches for files in the specified repo.
330330
// Returns the matching file-paths
331-
func (b *BleveIndexer) Search(repoIDs []int64, language, keyword string, page, pageSize int, isMatch bool) (int64, []*SearchResult, []*SearchResultLanguages, error) {
331+
func (b *BleveIndexer) Search(ctx context.Context, repoIDs []int64, language, keyword string, page, pageSize int, isMatch bool) (int64, []*SearchResult, []*SearchResultLanguages, error) {
332332
var (
333333
indexerQuery query.Query
334334
keywordQuery query.Query
@@ -381,7 +381,7 @@ func (b *BleveIndexer) Search(repoIDs []int64, language, keyword string, page, p
381381
searchRequest.AddFacet("languages", bleve.NewFacetRequest("Language", 10))
382382
}
383383

384-
result, err := b.indexer.Search(searchRequest)
384+
result, err := b.indexer.SearchInContext(ctx, searchRequest)
385385
if err != nil {
386386
return 0, nil, nil, err
387387
}

‎modules/indexer/code/elastic_search.go‎

Lines changed: 36 additions & 15 deletions
Original file line numberDiff line numberDiff line change
@@ -13,12 +13,14 @@ import (
1313
"net"
1414
"strconv"
1515
"strings"
16+
"sync"
1617
"time"
1718

1819
repo_model "code.gitea.io/gitea/models/repo"
1920
"code.gitea.io/gitea/modules/analyze"
2021
"code.gitea.io/gitea/modules/charset"
2122
"code.gitea.io/gitea/modules/git"
23+
"code.gitea.io/gitea/modules/graceful"
2224
"code.gitea.io/gitea/modules/json"
2325
"code.gitea.io/gitea/modules/log"
2426
"code.gitea.io/gitea/modules/setting"
@@ -46,6 +48,7 @@ type ElasticSearchIndexer struct {
4648
available bool
4749
availabilityCallback func(bool)
4850
stopTimer chan struct{}
51+
lock sync.RWMutex
4952
}
5053

5154
type elasticLogger struct {
@@ -144,7 +147,7 @@ func (b *ElasticSearchIndexer) realIndexerName() string {
144147

145148
// Init will initialize the indexer
146149
func (b *ElasticSearchIndexer) init() (bool, error) {
147-
ctx := context.Background()
150+
ctx := graceful.GetManager().HammerContext()
148151
exists, err := b.client.IndexExists(b.realIndexerName()).Do(ctx)
149152
if err != nil {
150153
return false, b.checkError(err)
@@ -198,11 +201,15 @@ func (b *ElasticSearchIndexer) init() (bool, error) {
198201

199202
// SetAvailabilityChangeCallback sets callback that will be triggered when availability changes
200203
func (b *ElasticSearchIndexer) SetAvailabilityChangeCallback(callback func(bool)) {
204+
b.lock.Lock()
205+
defer b.lock.Unlock()
201206
b.availabilityCallback = callback
202207
}
203208

204209
// Ping checks if elastic is available
205210
func (b *ElasticSearchIndexer) Ping() bool {
211+
b.lock.RLock()
212+
defer b.lock.RUnlock()
206213
return b.available
207214
}
208215

@@ -305,7 +312,7 @@ func (b *ElasticSearchIndexer) Index(ctx context.Context, repo *repo_model.Repos
305312
_, err := b.client.Bulk().
306313
Index(b.indexerAliasName).
307314
Add(reqs...).
308-
Do(context.Background())
315+
Do(ctx)
309316
return b.checkError(err)
310317
}
311318
return nil
@@ -315,7 +322,7 @@ func (b *ElasticSearchIndexer) Index(ctx context.Context, repo *repo_model.Repos
315322
func (b *ElasticSearchIndexer) Delete(repoID int64) error {
316323
_, err := b.client.DeleteByQuery(b.indexerAliasName).
317324
Query(elastic.NewTermsQuery("repo_id", repoID)).
318-
Do(context.Background())
325+
Do(graceful.GetManager().HammerContext())
319326
return b.checkError(err)
320327
}
321328

@@ -397,7 +404,7 @@ func extractAggs(searchResult *elastic.SearchResult) []*SearchResultLanguages {
397404
}
398405

399406
// Search searches for codes and language stats by given conditions.
400-
func (b *ElasticSearchIndexer) Search(repoIDs []int64, language, keyword string, page, pageSize int, isMatch bool) (int64, []*SearchResult, []*SearchResultLanguages, error) {
407+
func (b *ElasticSearchIndexer) Search(ctx context.Context, repoIDs []int64, language, keyword string, page, pageSize int, isMatch bool) (int64, []*SearchResult, []*SearchResultLanguages, error) {
401408
searchType := esMultiMatchTypeBestFields
402409
if isMatch {
403410
searchType = esMultiMatchTypePhrasePrefix
@@ -438,7 +445,7 @@ func (b *ElasticSearchIndexer) Search(repoIDs []int64, language, keyword string,
438445
).
439446
Sort("repo_id", true).
440447
From(start).Size(pageSize).
441-
Do(context.Background())
448+
Do(ctx)
442449
if err != nil {
443450
return 0, nil, nil, b.checkError(err)
444451
}
@@ -452,7 +459,7 @@ func (b *ElasticSearchIndexer) Search(repoIDs []int64, language, keyword string,
452459
Aggregation("language", aggregation).
453460
Query(query).
454461
Size(0). // We only needs stats information
455-
Do(context.Background())
462+
Do(ctx)
456463
if err != nil {
457464
return 0, nil, nil, b.checkError(err)
458465
}
@@ -469,7 +476,7 @@ func (b *ElasticSearchIndexer) Search(repoIDs []int64, language, keyword string,
469476
).
470477
Sort("repo_id", true).
471478
From(start).Size(pageSize).
472-
Do(context.Background())
479+
Do(ctx)
473480
if err != nil {
474481
return 0, nil, nil, b.checkError(err)
475482
}
@@ -486,27 +493,41 @@ func (b *ElasticSearchIndexer) Close() {
486493

487494
func (b *ElasticSearchIndexer) checkError(err error) error {
488495
var opErr *net.OpError
489-
if b.available && (elastic.IsConnErr(err) || (errors.As(err, &opErr) && (opErr.Op == "dial" || opErr.Op == "read"))) {
490-
b.available = false
491-
if b.availabilityCallback != nil {
492-
b.availabilityCallback(b.available)
493-
}
496+
if !(elastic.IsConnErr(err) || (errors.As(err, &opErr) && (opErr.Op == "dial" || opErr.Op == "read"))) {
497+
return err
494498
}
499+
500+
b.setAvailability(false)
501+
495502
return err
496503
}
497504

498505
func (b *ElasticSearchIndexer) checkAvailability() {
499-
if b.available {
506+
if b.Ping() {
500507
return
501508
}
502509

503510
// Request cluster state to check if elastic is available again
504-
_, err := b.client.ClusterState().Do(context.Background())
511+
_, err := b.client.ClusterState().Do(graceful.GetManager().ShutdownContext())
505512
if err != nil {
513+
b.setAvailability(false)
514+
return
515+
}
516+
517+
b.setAvailability(true)
518+
}
519+
520+
func (b *ElasticSearchIndexer) setAvailability(available bool) {
521+
b.lock.Lock()
522+
defer b.lock.Unlock()
523+
524+
if b.available == available {
506525
return
507526
}
508-
b.available = true
527+
528+
b.available = available
509529
if b.availabilityCallback != nil {
530+
// Call the callback from within the lock to ensure that the ordering remains correct
510531
b.availabilityCallback(b.available)
511532
}
512533
}

‎modules/indexer/code/indexer.go‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -46,7 +46,7 @@ type Indexer interface {
4646
SetAvailabilityChangeCallback(callback func(bool))
4747
Index(ctx context.Context, repo *repo_model.Repository, sha string, changes *repoChanges) error
4848
Delete(repoID int64) error
49-
Search(repoIDs []int64, language, keyword string, page, pageSize int, isMatch bool) (int64, []*SearchResult, []*SearchResultLanguages, error)
49+
Search(ctx context.Context, repoIDs []int64, language, keyword string, page, pageSize int, isMatch bool) (int64, []*SearchResult, []*SearchResultLanguages, error)
5050
Close()
5151
}
5252

‎modules/indexer/code/indexer_test.go‎

Lines changed: 2 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -5,6 +5,7 @@
55
package code
66

77
import (
8+
"context"
89
"path/filepath"
910
"testing"
1011

@@ -65,7 +66,7 @@ func testIndexer(name string, t *testing.T, indexer Indexer) {
6566

6667
for _, kw := range keywords {
6768
t.Run(kw.Keyword, func(t *testing.T) {
68-
total, res, langs, err := indexer.Search(kw.RepoIDs, "", kw.Keyword, 1, 10, false)
69+
total, res, langs, err := indexer.Search(context.TODO(), kw.RepoIDs, "", kw.Keyword, 1, 10, false)
6970
assert.NoError(t, err)
7071
assert.EqualValues(t, len(kw.IDs), total)
7172
assert.Len(t, langs, kw.Langs)

‎modules/indexer/code/search.go‎

Lines changed: 3 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -6,6 +6,7 @@ package code
66

77
import (
88
"bytes"
9+
"context"
910
"strings"
1011

1112
"code.gitea.io/gitea/modules/highlight"
@@ -106,12 +107,12 @@ func searchResult(result *SearchResult, startIndex, endIndex int) (*Result, erro
106107
}
107108

108109
// PerformSearch perform a search on a repository
109-
func PerformSearch(repoIDs []int64, language, keyword string, page, pageSize int, isMatch bool) (int, []*Result, []*SearchResultLanguages, error) {
110+
func PerformSearch(ctx context.Context, repoIDs []int64, language, keyword string, page, pageSize int, isMatch bool) (int, []*Result, []*SearchResultLanguages, error) {
110111
if len(keyword) == 0 {
111112
return 0, nil, nil, nil
112113
}
113114

114-
total, results, resultLanguages, err := indexer.Search(repoIDs, language, keyword, page, pageSize, isMatch)
115+
total, results, resultLanguages, err := indexer.Search(ctx, repoIDs, language, keyword, page, pageSize, isMatch)
115116
if err != nil {
116117
return 0, nil, nil, err
117118
}

‎modules/indexer/code/wrapped.go‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -93,12 +93,12 @@ func (w *wrappedIndexer) Delete(repoID int64) error {
9393
return indexer.Delete(repoID)
9494
}
9595

96-
func (w *wrappedIndexer) Search(repoIDs []int64, language, keyword string, page, pageSize int, isMatch bool) (int64, []*SearchResult, []*SearchResultLanguages, error) {
96+
func (w *wrappedIndexer) Search(ctx context.Context, repoIDs []int64, language, keyword string, page, pageSize int, isMatch bool) (int64, []*SearchResult, []*SearchResultLanguages, error) {
9797
indexer, err := w.get()
9898
if err != nil {
9999
return 0, nil, nil, err
100100
}
101-
return indexer.Search(repoIDs, language, keyword, page, pageSize, isMatch)
101+
return indexer.Search(ctx, repoIDs, language, keyword, page, pageSize, isMatch)
102102
}
103103

104104
func (w *wrappedIndexer) Close() {

‎modules/indexer/issues/bleve.go‎

Lines changed: 3 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -5,6 +5,7 @@
55
package issues
66

77
import (
8+
"context"
89
"fmt"
910
"os"
1011
"strconv"
@@ -238,7 +239,7 @@ func (b *BleveIndexer) Delete(ids ...int64) error {
238239

239240
// Search searches for issues by given conditions.
240241
// Returns the matching issue IDs
241-
func (b *BleveIndexer) Search(keyword string, repoIDs []int64, limit, start int) (*SearchResult, error) {
242+
func (b *BleveIndexer) Search(ctx context.Context, keyword string, repoIDs []int64, limit, start int) (*SearchResult, error) {
242243
var repoQueriesP []*query.NumericRangeQuery
243244
for _, repoID := range repoIDs {
244245
repoQueriesP = append(repoQueriesP, numericEqualityQuery(repoID, "RepoID"))
@@ -258,7 +259,7 @@ func (b *BleveIndexer) Search(keyword string, repoIDs []int64, limit, start int)
258259
search := bleve.NewSearchRequestOptions(indexerQuery, limit, start, false)
259260
search.SortBy([]string{"-_score"})
260261

261-
result, err := b.indexer.Search(search)
262+
result, err := b.indexer.SearchInContext(ctx, search)
262263
if err != nil {
263264
return nil, err
264265
}

‎modules/indexer/issues/bleve_test.go‎

Lines changed: 2 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -5,6 +5,7 @@
55
package issues
66

77
import (
8+
"context"
89
"os"
910
"testing"
1011

@@ -84,7 +85,7 @@ func TestBleveIndexAndSearch(t *testing.T) {
8485
}
8586

8687
for _, kw := range keywords {
87-
res, err := indexer.Search(kw.Keyword, []int64{2}, 10, 0)
88+
res, err := indexer.Search(context.TODO(), kw.Keyword, []int64{2}, 10, 0)
8889
assert.NoError(t, err)
8990

9091
ids := make([]int64, 0, len(res.Hits))

0 commit comments

Comments
 (0)