Coverage Report

Created: 2026-07-28 20:20

next uncovered line (L), next uncovered region (R), next uncovered branch (B)
/root/doris/common/cpp/s3_rate_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 "s3_rate_limiter.h"
19
20
#include <glog/logging.h> // IWYU pragma: export
21
22
#include <atomic>
23
#include <chrono>
24
#include <mutex>
25
#include <thread>
26
27
#if defined(__APPLE__)
28
#include <ctime>
29
#endif
30
31
namespace doris {
32
// Just 10^6.
33
static constexpr auto NS = 1000000000UL;
34
35
class S3RateLimiter::SimpleSpinLock {
36
public:
37
8
    SimpleSpinLock() = default;
38
    ~SimpleSpinLock() = default;
39
40
5
    void lock() {
41
5
        int spin_count = 0;
42
5
        static constexpr int MAX_SPIN_COUNT = 50;
43
5
        while (_flag.test_and_set(std::memory_order_acq_rel)) {
44
0
            spin_count++;
45
0
            if (spin_count >= MAX_SPIN_COUNT) {
46
0
                LOG(WARNING) << "Warning: Excessive spinning detected while acquiring lock. Spin "
47
0
                                "count: "
48
0
                             << spin_count;
49
0
                spin_count = 0;
50
0
            }
51
            // Spin until we acquire the lock
52
0
        }
53
5
    }
54
55
5
    void unlock() { _flag.clear(std::memory_order_release); }
56
57
private:
58
    std::atomic_flag _flag = ATOMIC_FLAG_INIT;
59
};
60
61
S3RateLimiter::S3RateLimiter(size_t max_speed, size_t max_burst, size_t limit)
62
8
        : _max_speed(max_speed),
63
8
          _max_burst(max_burst),
64
8
          _limit(limit),
65
8
          _mutex(std::make_unique<S3RateLimiter::SimpleSpinLock>()),
66
8
          _remain_tokens(max_burst) {}
67
68
8
S3RateLimiter::~S3RateLimiter() = default;
69
70
2
S3RateLimiterHolder::~S3RateLimiterHolder() = default;
71
72
5
std::pair<size_t, double> S3RateLimiter::_update_remain_token(long now, size_t amount) {
73
    // Values obtained under lock to be checked after release
74
5
    size_t count_value;
75
5
    double tokens_value;
76
5
    {
77
5
        std::lock_guard<SimpleSpinLock> lock(*_mutex);
78
5
        now = (now < _prev_ns_count) ? _prev_ns_count : now;
79
5
        if (_max_speed) {
80
5
            double delta_seconds =
81
5
                    _prev_ns_count ? static_cast<double>(now - _prev_ns_count) / NS : 0;
82
5
            _remain_tokens = std::min<double>(_remain_tokens + _max_speed * delta_seconds - amount,
83
5
                                              _max_burst);
84
5
        }
85
5
        _count += amount;
86
5
        count_value = _count;
87
5
        tokens_value = _remain_tokens;
88
5
        _prev_ns_count = now;
89
5
    }
90
5
    return {count_value, tokens_value};
91
5
}
92
93
5
int64_t S3RateLimiter::add(size_t amount) {
94
    // Values obtained under lock to be checked after release
95
5
    auto duration = std::chrono::steady_clock::now().time_since_epoch();
96
5
    auto time_nano_count = std::chrono::duration_cast<std::chrono::nanoseconds>(duration).count();
97
5
    auto [count_value, tokens_value] = _update_remain_token(time_nano_count, amount);
98
99
5
    if (_limit && count_value > _limit) {
100
        // CK would throw exception
101
1
        return -1;
102
1
    }
103
104
    // Wait unless there is positive amount of remain_tokens - throttling
105
4
    int64_t sleep_time_ns = 0;
106
4
    if (_max_speed && tokens_value < 0) {
107
0
        sleep_time_ns = static_cast<int64_t>(-tokens_value / _max_speed * NS);
108
0
        std::this_thread::sleep_for(std::chrono::nanoseconds(sleep_time_ns));
109
0
    }
110
111
4
    return sleep_time_ns;
112
5
}
113
114
S3RateLimiterHolder::S3RateLimiterHolder(size_t max_speed, size_t max_burst, size_t limit,
115
                                         std::function<void(int64_t)> metric_func)
116
2
        : rate_limiter(std::make_unique<S3RateLimiter>(max_speed, max_burst, limit)),
117
2
          metric_func(std::move(metric_func)) {}
118
119
5
int64_t S3RateLimiterHolder::add(size_t amount) {
120
5
    int64_t sleep;
121
5
    {
122
5
        std::shared_lock read {rate_limiter_rw_lock};
123
5
        sleep = rate_limiter->add(amount);
124
5
    }
125
5
    metric_func(sleep);
126
5
    return sleep;
127
5
}
128
129
6
int S3RateLimiterHolder::reset(size_t max_speed, size_t max_burst, size_t limit) {
130
6
    {
131
6
        std::unique_lock write {rate_limiter_rw_lock};
132
6
        rate_limiter = std::make_unique<S3RateLimiter>(max_speed, max_burst, limit);
133
6
    }
134
6
    return 0;
135
6
}
136
137
0
std::string to_string(S3RateLimitType type) {
138
0
    switch (type) {
139
0
    case S3RateLimitType::GET:
140
0
        return "get";
141
0
    case S3RateLimitType::PUT:
142
0
        return "put";
143
0
    default:
144
0
        return std::to_string(static_cast<size_t>(type));
145
0
    }
146
0
}
147
148
0
S3RateLimitType string_to_s3_rate_limit_type(std::string_view value) {
149
0
    if (value == "get") {
150
0
        return S3RateLimitType::GET;
151
0
    } else if (value == "put") {
152
0
        return S3RateLimitType::PUT;
153
0
    }
154
0
    return S3RateLimitType::UNKNOWN;
155
0
}
156
} // namespace doris