Coverage Report

Created: 2026-06-03 13:59

next uncovered line (L), next uncovered region (R), next uncovered branch (B)
be/src/storage/compaction/compaction_permit_limiter.cpp
Line
Count
Source
1
// Licensed to the Apache Software Foundation (ASF) under one
2
// or more contributor license agreements.  See the NOTICE file
3
// distributed with this work for additional information
4
// regarding copyright ownership.  The ASF licenses this file
5
// to you under the Apache License, Version 2.0 (the
6
// "License"); you may not use this file except in compliance
7
// with the License.  You may obtain a copy of the License at
8
//
9
//   http://www.apache.org/licenses/LICENSE-2.0
10
//
11
// Unless required by applicable law or agreed to in writing,
12
// software distributed under the License is distributed on an
13
// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
14
// KIND, either express or implied.  See the License for the
15
// specific language governing permissions and limitations
16
// under the License.
17
18
#include "storage/compaction/compaction_permit_limiter.h"
19
20
#include "common/config.h"
21
#include "common/metrics/doris_metrics.h"
22
23
namespace doris {
24
25
378
CompactionPermitLimiter::CompactionPermitLimiter() : _used_permits(0), _binlog_used_permits(0) {}
26
27
0
bool CompactionPermitLimiter::try_request(int64_t permits, CompactionType compaction_type) {
28
0
    const int64_t total_permits = config::total_permits_for_compaction_score;
29
0
    std::unique_lock<std::mutex> lock(_permits_mutex);
30
0
    if (_used_permits + permits > total_permits) {
31
0
        return false;
32
0
    }
33
34
0
    if (compaction_type == CompactionType::BINLOG_COMPACTION) {
35
0
        const int64_t binlog_total_permits =
36
0
                (total_permits * config::binlog_compaction_permits_percent + 99) / 100;
37
0
        if (_binlog_used_permits + permits > binlog_total_permits) {
38
0
            return false;
39
0
        }
40
0
        _binlog_used_permits += permits;
41
0
        DorisMetrics::instance()->binlog_compaction_used_permits->set_value(_binlog_used_permits);
42
0
    }
43
44
0
    _used_permits += permits;
45
0
    DorisMetrics::instance()->compaction_used_permits->set_value(_used_permits);
46
0
    return true;
47
0
}
48
49
32
void CompactionPermitLimiter::request(int64_t permits) {
50
32
    DorisMetrics::instance()->compaction_waitting_permits->set_value(permits);
51
    // 1. config::total_permits_for_compaction_score = 20000
52
    // 2. Thread-B requests permits 11000， used_permits = 11000
53
    // 3. Thread-A requests permits 12000，wait for used_permits + 12000 <= 20000
54
    // 4. adjust config::total_permits_for_compaction_score = 10000
55
    // 5. Thread-B releases permits，used_permits = 0，notify Thread-A，used_permits + 12000 <= 10000
56
    // we need to initialize total_permits instead of using the config.
57
32
    int64_t total_permits = config::total_permits_for_compaction_score;
58
32
    if (permits > total_permits) {
59
        // when tablet's compaction score is larger than "config::total_permits_for_compaction_score",
60
        // it's necessary to do compaction for this tablet because this tablet will not get "permits"
61
        // anyway. otherwise, compaction task for this tablet will not be executed forever.
62
0
        std::unique_lock<std::mutex> lock(_permits_mutex);
63
0
        _permits_cv.wait(lock, [permits, total_permits, this] {
64
0
            return _used_permits == 0 || _used_permits + permits <= total_permits;
65
0
        });
66
32
    } else {
67
32
        if (_used_permits + permits > total_permits) {
68
0
            std::unique_lock<std::mutex> lock(_permits_mutex);
69
0
            _permits_cv.wait(lock, [permits, total_permits, this] {
70
0
                return _used_permits + permits <= total_permits;
71
0
            });
72
0
        }
73
32
    }
74
32
    _used_permits += permits;
75
32
    DorisMetrics::instance()->compaction_waitting_permits->set_value(0);
76
32
    DorisMetrics::instance()->compaction_used_permits->set_value(_used_permits);
77
32
}
78
79
19
void CompactionPermitLimiter::release(int64_t permits, CompactionType compaction_type) {
80
19
    std::unique_lock<std::mutex> lock(_permits_mutex);
81
19
    if (compaction_type == CompactionType::BINLOG_COMPACTION) {
82
0
        _binlog_used_permits -= permits;
83
0
        DorisMetrics::instance()->binlog_compaction_used_permits->set_value(_binlog_used_permits);
84
0
    }
85
19
    _used_permits -= permits;
86
19
    _permits_cv.notify_one();
87
19
    DorisMetrics::instance()->compaction_used_permits->set_value(_used_permits);
88
19
}
89
} // namespace doris