forked from cockroachdb/cockroach
/
scanner.go
242 lines (221 loc) · 7.93 KB
/
scanner.go
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
// Copyright 2014 The Cockroach Authors.
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
// implied. See the License for the specific language governing
// permissions and limitations under the License. See the AUTHORS file
// for names of contributors.
//
// Author: Spencer Kimball (spencer.kimball@gmail.com)
package storage
import (
"sync"
"time"
"github.com/cockroachdb/cockroach/proto"
"github.com/cockroachdb/cockroach/storage/engine"
"github.com/cockroachdb/cockroach/util/hlc"
"github.com/cockroachdb/cockroach/util/log"
"github.com/cockroachdb/cockroach/util/stop"
)
// A replicaQueue is a prioritized queue of replicas for which work is
// scheduled. For example, there's a GC queue for replicas which are due
// for garbage collection, a rebalance queue to move replicas from full
// or busy stores, a recovery queue for replicas of ranges with dead replicas,
// etc.
type replicaQueue interface {
// Start launches a goroutine to process the contents of the queue.
// The provided stopper is used to signal that the goroutine should exit.
Start(*hlc.Clock, *stop.Stopper)
// MaybeAdd adds the replica to the queue if the replica meets
// the queue's inclusion criteria and the queue is not already
// too full, etc.
MaybeAdd(*Replica, proto.Timestamp)
// MaybeRemove removes the replica from the queue if it is present.
MaybeRemove(*Replica)
}
// A replicaSet provides access to a sequence of replicas to consider
// for inclusion in replica queues. There are no requirements for the
// ordering of the iteration.
type replicaSet interface {
// Visit calls the given function for every replica in the set btree
// until the function returns false.
Visit(func(*Replica) bool)
// EstimatedCount returns the number of replicas estimated to remain
// in the iteration. This value does not need to be exact.
EstimatedCount() int
}
// A storeStats holds statistics over the entire store. Stats is an
// aggregation of MVCC stats across all replicas in the store.
type storeStats struct {
RangeCount int
MVCC engine.MVCCStats
}
// A replicaScanner iterates over replicas at a measured pace in order to
// complete approximately one full scan per target interval in a large
// store (in small stores it may complete faster than the target
// interval). Each replica is tested for inclusion in a sequence of
// prioritized replica queues.
type replicaScanner struct {
targetInterval time.Duration // Target duration interval for scan loop
maxIdleTime time.Duration // Max idle time for scan loop
replicas replicaSet // Replicas to be scanned
queues []replicaQueue // Replica queues managed by this scanner
removed chan *Replica // Replicas to remove from queues
// Count of times and total duration through the scanning loop but locked by the completedScan
// mutex.
completedScan *sync.Cond
count int64
total time.Duration
}
// newReplicaScanner creates a new replica scanner with the provided loop intervals,
// replica set, and replica queues. If scanFn is not nil, after a complete
// loop that function will be called.
func newReplicaScanner(targetInterval, maxIdleTime time.Duration, replicas replicaSet) *replicaScanner {
return &replicaScanner{
targetInterval: targetInterval,
maxIdleTime: maxIdleTime,
replicas: replicas,
removed: make(chan *Replica, 10),
completedScan: sync.NewCond(&sync.Mutex{}),
}
}
// AddQueues adds a variable arg list of queues to the replica scanner.
// This method may only be called before Start().
func (rs *replicaScanner) AddQueues(queues ...replicaQueue) {
rs.queues = append(rs.queues, queues...)
}
// Start spins up the scanning loop. Call Stop() to exit the loop.
func (rs *replicaScanner) Start(clock *hlc.Clock, stopper *stop.Stopper) {
for _, queue := range rs.queues {
queue.Start(clock, stopper)
}
rs.scanLoop(clock, stopper)
}
// Count returns the number of times the scanner has cycled through
// all replicas.
func (rs *replicaScanner) Count() int64 {
rs.completedScan.L.Lock()
defer rs.completedScan.L.Unlock()
return rs.count
}
// avgScan returns the average scan time of each scan cycle. Used in unittests.
func (rs *replicaScanner) avgScan() time.Duration {
rs.completedScan.L.Lock()
defer rs.completedScan.L.Unlock()
return time.Duration(rs.total.Nanoseconds() / int64(rs.count))
}
// RemoveReplica removes a replica from any replica queues the scanner may
// have placed it in. This method should be called by the Store
// when a replica is removed (e.g. rebalanced or merged).
func (rs *replicaScanner) RemoveReplica(repl *Replica) {
rs.removed <- repl
}
// WaitForScanCompletion waits until the end of the next scan and returns the
// total number of scans completed so far.
func (rs *replicaScanner) WaitForScanCompletion() int64 {
rs.completedScan.L.Lock()
defer rs.completedScan.L.Unlock()
initalValue := rs.count
for rs.count == initalValue {
rs.completedScan.Wait()
}
return rs.count
}
// paceInterval returns a duration between iterations to allow us to pace
// the scan.
func (rs *replicaScanner) paceInterval(start, now time.Time) time.Duration {
elapsed := now.Sub(start)
remainingNanos := rs.targetInterval.Nanoseconds() - elapsed.Nanoseconds()
if remainingNanos < 0 {
remainingNanos = 0
}
count := rs.replicas.EstimatedCount()
if count < 1 {
count = 1
}
interval := time.Duration(remainingNanos / int64(count))
if rs.maxIdleTime > 0 && interval > rs.maxIdleTime {
interval = rs.maxIdleTime
}
return interval
}
// waitAndProcess waits for the pace interval and processes the replica
// if repl is not nil. The method returns true when the scanner needs
// to be stopped. The method also removes a replica from queues when it
// is signaled via the removed channel.
func (rs *replicaScanner) waitAndProcess(start time.Time, clock *hlc.Clock, stopper *stop.Stopper,
repl *Replica) bool {
waitInterval := rs.paceInterval(start, time.Now())
nextTime := time.After(waitInterval)
if log.V(6) {
log.Infof("Wait time interval set to %s", waitInterval)
}
for {
select {
case <-nextTime:
if repl == nil {
return false
}
return !stopper.RunTask(func() {
// Try adding replica to all queues.
for _, q := range rs.queues {
q.MaybeAdd(repl, clock.Now())
}
})
case repl := <-rs.removed:
// Remove replica from all queues as applicable.
for _, q := range rs.queues {
q.MaybeRemove(repl)
}
if log.V(6) {
log.Infof("removed replica %s", repl)
}
case <-stopper.ShouldStop():
return true
}
}
}
// scanLoop loops endlessly, scanning through replicas available via
// the replica set, or until the scanner is stopped. The iteration
// is paced to complete a full scan in approximately the scan interval.
func (rs *replicaScanner) scanLoop(clock *hlc.Clock, stopper *stop.Stopper) {
stopper.RunWorker(func() {
start := time.Now()
for {
var shouldStop bool
if rs.replicas.EstimatedCount() == 0 {
// Just wait without processing any replica.
shouldStop = rs.waitAndProcess(start, clock, stopper, nil)
} else {
shouldStop = true
rs.replicas.Visit(func(repl *Replica) bool {
shouldStop = rs.waitAndProcess(start, clock, stopper, repl)
return !shouldStop
})
}
shouldStop = shouldStop || !stopper.RunTask(func() {
// Increment iteration count.
rs.completedScan.L.Lock()
rs.count++
rs.total += time.Now().Sub(start)
rs.completedScan.Broadcast()
rs.completedScan.L.Unlock()
if log.V(6) {
log.Infof("reset replica scan iteration")
}
// Reset iteration and start time.
start = time.Now()
})
if shouldStop {
return
}
}
})
}