fork of https://github.com/sourcegraph/zoekt
shards: implement markReady on shardedSearcher (#493)
This is the final commit implementing a non-zero Crashes value while
shardedSearcher is startin up. This is meant to communicate when a zoekt
is partially available. When landed, Sourcegraph's APIs will communicate
that the user may have partial results and should retry.
Test Plan: go test. Additionally added a large sleep to loader.load to
simulate a slow startup. Then did searches in zoekt-webserver and
observed it returning non-zero crash counts until all shards loaded.
author
Keegan Carruthers-Smith
committer
GitHub
date
(Dec 6, 2022, 9:57 PM +0200)
commit
cfd410d2
cfd410d245651c689837a904c22be8b5f015907a
parent
9899a9b3
9899a9b3f475ef066ed70c395f8b303268f5d00c
+39
-13
+14
-2
shards/shards.go
+14
-2
shards/shards.go
···
25
25
"sort"
26
26
"strconv"
27
27
"sync"
28
-
"sync/atomic"
29
28
"time"
30
29
31
30
"golang.org/x/sync/semaphore"
32
31
33
32
"github.com/prometheus/client_golang/prometheus"
34
33
"github.com/prometheus/client_golang/prometheus/promauto"
34
+
"go.uber.org/atomic"
35
35
36
36
"github.com/sourcegraph/zoekt"
37
37
"github.com/sourcegraph/zoekt/query"
···
196
196
mu sync.Mutex // protects writes to shards
197
197
shards map[string]*rankedShard
198
198
199
+
ready atomic.Bool
199
200
ranked atomic.Value
200
201
}
201
202
···
265
266
}
266
267
267
268
func (tl *loader) load(keys ...string) {
269
+
// This is called with all keys on startup, so once this function has
270
+
// finished running shardedSearcher will be ready.
271
+
defer tl.ss.markReady()
272
+
268
273
var (
269
274
mu sync.Mutex // synchronizes writes to the shards map
270
275
wg sync.WaitGroup // used to wait for all shards to load
···
980
985
func (s *shardedSearcher) getLoaded() loaded {
981
986
// next commit will store the true value of this, for now we keep the
982
987
// backwards compatible behaviour.
983
-
ready := true
988
+
ready := s.ready.Load()
984
989
// ranked is loaded after ready to avoid a race were ready is true but
985
990
// ranked is still not the final set of shards.
986
991
ranked, _ := s.ranked.Load().([]*rankedShard)
···
1020
1025
repos: repos,
1021
1026
priority: maxPriority,
1022
1027
}
1028
+
}
1029
+
1030
+
// markReady should be called once all shards have been passed into replace on
1031
+
// startup. Once s is marked as ready it stops reporting a Crash in the
1032
+
// response Stats.
1033
+
func (s *shardedSearcher) markReady() {
1034
+
s.ready.CompareAndSwap(false, true)
1023
1035
}
1024
1036
1025
1037
func (s *shardedSearcher) replace(shards map[string]zoekt.Searcher) {
+25
-11
shards/shards_test.go
+25
-11
shards/shards_test.go
···
65
65
ss := newShardedSearcher(2)
66
66
ss.ranked.Store([]*rankedShard{{Searcher: &crashSearcher{}}})
67
67
68
-
q := &query.Substring{Pattern: "hoi"}
69
-
opts := &zoekt.SearchOptions{}
70
-
if res, err := ss.Search(context.Background(), q, opts); err != nil {
71
-
t.Fatalf("Search: %v", err)
72
-
} else if res.Stats.Crashes != 1 {
73
-
t.Errorf("got stats %#v, want crashes = 1", res.Stats)
68
+
var wantCrashes int
69
+
test := func(t *testing.T) {
70
+
q := &query.Substring{Pattern: "hoi"}
71
+
opts := &zoekt.SearchOptions{}
72
+
if res, err := ss.Search(context.Background(), q, opts); err != nil {
73
+
t.Fatalf("Search: %v", err)
74
+
} else if res.Stats.Crashes != wantCrashes {
75
+
t.Errorf("got stats %#v, want crashes = %d", res.Stats, wantCrashes)
76
+
}
77
+
78
+
if res, err := ss.List(context.Background(), q, nil); err != nil {
79
+
t.Fatalf("List: %v", err)
80
+
} else if res.Crashes != wantCrashes {
81
+
t.Errorf("got result %#v, want crashes = %d", res, wantCrashes)
82
+
}
74
83
}
75
84
76
-
if res, err := ss.List(context.Background(), q, nil); err != nil {
77
-
t.Fatalf("List: %v", err)
78
-
} else if res.Crashes != 1 {
79
-
t.Errorf("got result %#v, want crashes = 1", res)
80
-
}
85
+
// Before we are marked as ready we have one extra crash
86
+
wantCrashes = 2
87
+
t.Run("loading", test)
88
+
89
+
// After marking as ready we should only have the crashSearcher
90
+
// contributing.
91
+
ss.markReady()
92
+
wantCrashes = 1
93
+
t.Run("ready", test)
81
94
}
82
95
83
96
type rankSearcher struct {
···
376
389
"3": searcherForTest(t, testIndexBuilder(t, repos[1], doc)),
377
390
"4": searcherForTest(t, testIndexBuilder(t, repos[1])),
378
391
})
392
+
ss.markReady()
379
393
380
394
stats := zoekt.RepoStats{
381
395
Shards: 2,