blob: 2adb4985437b018b6bfa8f2942470769b4138321 [file] [log] [blame]
khenaidoo59ce9dd2019-11-11 13:05:32 -05001// Copyright 2015 The etcd Authors
2//
3// Licensed under the Apache License, Version 2.0 (the "License");
4// you may not use this file except in compliance with the License.
5// You may obtain a copy of the License at
6//
7// http://www.apache.org/licenses/LICENSE-2.0
8//
9// Unless required by applicable law or agreed to in writing, software
10// distributed under the License is distributed on an "AS IS" BASIS,
11// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12// See the License for the specific language governing permissions and
13// limitations under the License.
14
15package mvcc
16
17import (
18 "encoding/binary"
19 "time"
20
21 "go.uber.org/zap"
22)
23
24func (s *store) scheduleCompaction(compactMainRev int64, keep map[revision]struct{}) bool {
25 totalStart := time.Now()
26 defer func() { dbCompactionTotalMs.Observe(float64(time.Since(totalStart) / time.Millisecond)) }()
27 keyCompactions := 0
28 defer func() { dbCompactionKeysCounter.Add(float64(keyCompactions)) }()
29
30 end := make([]byte, 8)
31 binary.BigEndian.PutUint64(end, uint64(compactMainRev+1))
32
33 last := make([]byte, 8+1+8)
34 for {
35 var rev revision
36
37 start := time.Now()
38
39 tx := s.b.BatchTx()
40 tx.Lock()
41 keys, _ := tx.UnsafeRange(keyBucketName, last, end, int64(s.cfg.CompactionBatchLimit))
42 for _, key := range keys {
43 rev = bytesToRev(key)
44 if _, ok := keep[rev]; !ok {
45 tx.UnsafeDelete(keyBucketName, key)
46 }
47 }
48
49 if len(keys) < s.cfg.CompactionBatchLimit {
50 rbytes := make([]byte, 8+1+8)
51 revToBytes(revision{main: compactMainRev}, rbytes)
52 tx.UnsafePut(metaBucketName, finishedCompactKeyName, rbytes)
53 tx.Unlock()
54 if s.lg != nil {
55 s.lg.Info(
56 "finished scheduled compaction",
57 zap.Int64("compact-revision", compactMainRev),
58 zap.Duration("took", time.Since(totalStart)),
59 )
60 } else {
61 plog.Infof("finished scheduled compaction at %d (took %v)", compactMainRev, time.Since(totalStart))
62 }
63 return true
64 }
65
66 // update last
67 revToBytes(revision{main: rev.main, sub: rev.sub + 1}, last)
68 tx.Unlock()
69 // Immediately commit the compaction deletes instead of letting them accumulate in the write buffer
70 s.b.ForceCommit()
71 dbCompactionPauseMs.Observe(float64(time.Since(start) / time.Millisecond))
72
73 select {
74 case <-time.After(10 * time.Millisecond):
75 case <-s.stopc:
76 return false
77 }
78 }
79}