Line data Source code
1 : // Copyright 2023 The LevelDB-Go and Pebble Authors. All rights reserved. Use
2 : // of this source code is governed by a BSD-style license that can be found in
3 : // the LICENSE file.
4 :
5 : package tool
6 :
7 : import (
8 : "context"
9 : "fmt"
10 : "io"
11 : "math"
12 : "math/rand"
13 : "sort"
14 : "strconv"
15 : "strings"
16 : "sync"
17 : "time"
18 :
19 : "github.com/cockroachdb/errors"
20 : "github.com/cockroachdb/pebble"
21 : "github.com/cockroachdb/pebble/internal/base"
22 : "github.com/cockroachdb/pebble/objstorage"
23 : "github.com/spf13/cobra"
24 : )
25 :
26 : type benchIO struct {
27 : readableIdx int
28 : ofs int64
29 : size int
30 : // elapsed time for the IO, filled out by performIOs.
31 : elapsed time.Duration
32 : }
33 :
34 : const maxIOSize = 1024 * 1024
35 :
36 : // runIOBench runs an IO benchmark against the current sstables of a database.
37 : // The workload is random IO, with various IO sizes. The main goal of the
38 : // benchmark is to establish the relationship between IO size and latency,
39 : // especially against shared object storage.
40 0 : func (d *dbT) runIOBench(cmd *cobra.Command, args []string) {
41 0 : stdout := cmd.OutOrStdout()
42 0 :
43 0 : ioSizes, err := parseIOSizes(d.ioSizes)
44 0 : if err != nil {
45 0 : fmt.Fprintf(stdout, "error parsing io-sizes: %s\n", err)
46 0 : return
47 0 : }
48 :
49 0 : db, err := d.openDB(args[0])
50 0 : if err != nil {
51 0 : fmt.Fprintf(stdout, "%s\n", err)
52 0 : return
53 0 : }
54 0 : defer d.closeDB(stdout, db)
55 0 :
56 0 : readables, err := d.openBenchTables(db)
57 0 : if err != nil {
58 0 : fmt.Fprintf(stdout, "%s\n", err)
59 0 : return
60 0 : }
61 :
62 0 : defer func() {
63 0 : for _, r := range readables {
64 0 : r.Close()
65 0 : }
66 : }()
67 :
68 0 : ios := genBenchIOs(stdout, readables, d.ioCount, ioSizes)
69 0 :
70 0 : levels := "L5,L6"
71 0 : if d.allLevels {
72 0 : levels = "all"
73 0 : }
74 0 : fmt.Fprintf(stdout, "IO count: %d Parallelism: %d Levels: %s\n", d.ioCount, d.ioParallelism, levels)
75 0 :
76 0 : var wg sync.WaitGroup
77 0 : wg.Add(d.ioParallelism)
78 0 : remainingIOs := ios
79 0 : for i := 0; i < d.ioParallelism; i++ {
80 0 : // We want to distribute the IOs among d.ioParallelism goroutines. At each
81 0 : // step, we look at the number of IOs remaining and take the average (across
82 0 : // the goroutines that are left); this deals with any rounding issues.
83 0 : n := len(remainingIOs) / (d.ioParallelism - i)
84 0 : go func(workerIdx int, ios []benchIO) {
85 0 : defer wg.Done()
86 0 : if err := performIOs(readables, ios); err != nil {
87 0 : fmt.Fprintf(stdout, "worker %d encountered error: %v", workerIdx, err)
88 0 : }
89 : }(i, remainingIOs[:n])
90 0 : remainingIOs = remainingIOs[n:]
91 : }
92 0 : wg.Wait()
93 0 :
94 0 : elapsed := make([]time.Duration, d.ioCount)
95 0 : for _, ioSize := range ioSizes {
96 0 : elapsed = elapsed[:0]
97 0 : for i := range ios {
98 0 : if ios[i].size == ioSize {
99 0 : elapsed = append(elapsed, ios[i].elapsed)
100 0 : }
101 : }
102 0 : fmt.Fprintf(stdout, "%4dKB -- %s\n", ioSize/1024, getStats(elapsed))
103 : }
104 : }
105 :
106 : // genBenchIOs generates <count> IOs for each given size. All IOs (across all
107 : // sizes) are in random order.
108 : func genBenchIOs(
109 : stdout io.Writer, readables []objstorage.Readable, count int, sizes []int,
110 0 : ) []benchIO {
111 0 : // size[i] is the size of the object, in blocks of maxIOSize.
112 0 : size := make([]int, len(readables))
113 0 : // sum[i] is the sum (size[0] + ... + size[i]).
114 0 : sum := make([]int, len(readables))
115 0 : total := 0
116 0 : for i, r := range readables {
117 0 : size[i] = int(r.Size() / maxIOSize)
118 0 : total += size[i]
119 0 : sum[i] = total
120 0 : }
121 0 : fmt.Fprintf(stdout, "Opened %d objects; total size %d MB.\n", len(readables), total*maxIOSize/(1024*1024))
122 0 :
123 0 : // To avoid a lot of overlap between the reads, the total size should be a
124 0 : // factor larger than the size we will actually read (for the largest IO
125 0 : // size).
126 0 : const sizeFactor = 2
127 0 : if total*maxIOSize < count*sizes[len(sizes)-1]*sizeFactor {
128 0 : fmt.Fprintf(stdout, "Warning: store too small for the given IO count and sizes.\n")
129 0 : }
130 :
131 : // Choose how many IOs we do for each object, by selecting a random block
132 : // across all file blocks.
133 : // The choice of objects will be the same across all IO sizes.
134 0 : b := make([]int, count)
135 0 : for i := range b {
136 0 : b[i] = rand.Intn(total)
137 0 : }
138 : // For each b[i], find the index such that sum[idx-1] <= b < sum[idx].
139 : // Sorting b makes this easier: we can "merge" the sorted arrays b and sum.
140 0 : sort.Ints(b)
141 0 : rIdx := make([]int, count)
142 0 : currIdx := 0
143 0 : for i := range b {
144 0 : for b[i] >= sum[currIdx] {
145 0 : currIdx++
146 0 : }
147 0 : rIdx[i] = currIdx
148 : }
149 :
150 0 : res := make([]benchIO, 0, count*len(sizes))
151 0 : for _, ioSize := range sizes {
152 0 : for _, idx := range rIdx {
153 0 : // Random ioSize aligned offset.
154 0 : ofs := ioSize * rand.Intn(size[idx]*maxIOSize/ioSize)
155 0 :
156 0 : res = append(res, benchIO{
157 0 : readableIdx: idx,
158 0 : ofs: int64(ofs),
159 0 : size: ioSize,
160 0 : })
161 0 : }
162 : }
163 0 : rand.Shuffle(len(res), func(i, j int) {
164 0 : res[i], res[j] = res[j], res[i]
165 0 : })
166 0 : return res
167 : }
168 :
169 : // openBenchTables opens the sstables for the benchmark and returns them as a
170 : // list of Readables.
171 : //
172 : // By default, only L5/L6 sstables are used; all levels are used if the
173 : // allLevels flag is set.
174 : //
175 : // Note that only sstables that are at least maxIOSize (1MB) are used.
176 0 : func (d *dbT) openBenchTables(db *pebble.DB) ([]objstorage.Readable, error) {
177 0 : tables, err := db.SSTables()
178 0 : if err != nil {
179 0 : return nil, err
180 0 : }
181 0 : startLevel := 5
182 0 : if d.allLevels {
183 0 : startLevel = 0
184 0 : }
185 :
186 0 : var nums []base.DiskFileNum
187 0 : numsMap := make(map[base.DiskFileNum]struct{})
188 0 : for l := startLevel; l < len(tables); l++ {
189 0 : for _, t := range tables[l] {
190 0 : n := t.BackingSSTNum.DiskFileNum()
191 0 : if _, ok := numsMap[n]; !ok {
192 0 : nums = append(nums, n)
193 0 : numsMap[n] = struct{}{}
194 0 : }
195 : }
196 : }
197 :
198 0 : p := db.ObjProvider()
199 0 : var res []objstorage.Readable
200 0 : for _, n := range nums {
201 0 : r, err := p.OpenForReading(context.Background(), base.FileTypeTable, n, objstorage.OpenOptions{})
202 0 : if err != nil {
203 0 : for _, r := range res {
204 0 : _ = r.Close()
205 0 : }
206 0 : return nil, err
207 : }
208 0 : if r.Size() < maxIOSize {
209 0 : _ = r.Close()
210 0 : continue
211 : }
212 0 : res = append(res, r)
213 : }
214 0 : if len(res) == 0 {
215 0 : return nil, errors.Errorf("no sstables (with size at least %d)", maxIOSize)
216 0 : }
217 :
218 0 : return res, nil
219 : }
220 :
221 : // parseIOSizes parses a comma-separated list of IO sizes, in KB.
222 0 : func parseIOSizes(sizes string) ([]int, error) {
223 0 : var res []int
224 0 : for _, s := range strings.Split(sizes, ",") {
225 0 : n, err := strconv.Atoi(s)
226 0 : if err != nil {
227 0 : return nil, err
228 0 : }
229 0 : ioSize := n * 1024
230 0 : if ioSize > maxIOSize {
231 0 : return nil, errors.Errorf("IO sizes over %d not supported", maxIOSize)
232 0 : }
233 0 : if maxIOSize%ioSize != 0 {
234 0 : return nil, errors.Errorf("IO size must be a divisor of %d", maxIOSize)
235 0 : }
236 0 : res = append(res, ioSize)
237 : }
238 0 : if len(res) == 0 {
239 0 : return nil, errors.Errorf("no IO sizes specified")
240 0 : }
241 0 : sort.Ints(res)
242 0 : return res, nil
243 : }
244 :
245 : // performIOs performs the given list of IOs and populates the elapsed fields.
246 0 : func performIOs(readables []objstorage.Readable, ios []benchIO) error {
247 0 : ctx := context.Background()
248 0 : rh := make([]objstorage.ReadHandle, len(readables))
249 0 : for i := range rh {
250 0 : rh[i] = readables[i].NewReadHandle(ctx)
251 0 : }
252 0 : defer func() {
253 0 : for i := range rh {
254 0 : rh[i].Close()
255 0 : }
256 : }()
257 :
258 0 : buf := make([]byte, maxIOSize)
259 0 : startTime := time.Now()
260 0 : var firstErr error
261 0 : var nOtherErrs int
262 0 : for i := range ios {
263 0 : if err := rh[ios[i].readableIdx].ReadAt(ctx, buf[:ios[i].size], ios[i].ofs); err != nil {
264 0 : if firstErr == nil {
265 0 : firstErr = err
266 0 : } else {
267 0 : nOtherErrs++
268 0 : }
269 : }
270 0 : endTime := time.Now()
271 0 : ios[i].elapsed = endTime.Sub(startTime)
272 0 : startTime = endTime
273 : }
274 0 : if nOtherErrs > 0 {
275 0 : return errors.Errorf("%v; plus %d more errors", firstErr, nOtherErrs)
276 0 : }
277 0 : return firstErr
278 : }
279 :
280 : // getStats calculates various statistics given a list of elapsed times.
281 0 : func getStats(d []time.Duration) string {
282 0 : sort.Slice(d, func(i, j int) bool { return d[i] < d[j] })
283 :
284 0 : factor := 1.0 / float64(len(d))
285 0 : var mean float64
286 0 : for i := range d {
287 0 : mean += float64(d[i]) * factor
288 0 : }
289 0 : var variance float64
290 0 : for i := range d {
291 0 : delta := float64(d[i]) - mean
292 0 : variance += delta * delta * factor
293 0 : }
294 :
295 0 : toStr := func(d time.Duration) string {
296 0 : if d < 10*time.Millisecond {
297 0 : return fmt.Sprintf("%1.2fms", float64(d)/float64(time.Millisecond))
298 0 : }
299 0 : if d < 100*time.Millisecond {
300 0 : return fmt.Sprintf("%2.1fms", float64(d)/float64(time.Millisecond))
301 0 : }
302 0 : return fmt.Sprintf("%4dms", d/time.Millisecond)
303 : }
304 :
305 0 : return fmt.Sprintf(
306 0 : "avg %s stddev %s p10 %s p50 %s p90 %s p95 %s p99 %s",
307 0 : toStr(time.Duration(mean)),
308 0 : toStr(time.Duration(math.Sqrt(variance))),
309 0 : toStr(d[len(d)*10/100]),
310 0 : toStr(d[len(d)*50/100]),
311 0 : toStr(d[len(d)*90/100]),
312 0 : toStr(d[len(d)*95/100]),
313 0 : toStr(d[len(d)*99/100]),
314 0 : )
315 : }
|
