common/cpp/token_bucket_rate_limiter.cpp

Source
// Licensed to the Apache Software Foundation (ASF) under one
// or more contributor license agreements.  See the NOTICE file
// distributed with this work for additional information
// regarding copyright ownership.  The ASF licenses this file
// to you 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.

#include "token_bucket_rate_limiter.h"

#include <bthread/bthread.h>
#include <glog/logging.h> // IWYU pragma: export

#include <atomic>
#include <chrono>
#include <mutex>
#include <thread>

#if defined(__APPLE__)
#include <ctime>
#endif

namespace doris {
// Just 10^6.
static constexpr auto NS = 1000000000UL;

bvar::Adder<int64_t> s3_get_rate_limit_sleep_ns("s3_get_rate_limit_sleep_ns");
bvar::Adder<int64_t> s3_get_rate_limit_sleep_count("s3_get_rate_limit_sleep_count");
bvar::Adder<int64_t> s3_get_rate_limit_rejected_count("s3_get_rate_limit_rejected_count");
bvar::Adder<int64_t> s3_put_rate_limit_sleep_ns("s3_put_rate_limit_sleep_ns");
bvar::Adder<int64_t> s3_put_rate_limit_sleep_count("s3_put_rate_limit_sleep_count");
bvar::Adder<int64_t> s3_put_rate_limit_rejected_count("s3_put_rate_limit_rejected_count");

static std::atomic<int64_t> s3_get_rate_limit_sleep_log_count {0};
static std::atomic<int64_t> s3_get_rate_limit_rejected_log_count {0};
static std::atomic<int64_t> s3_put_rate_limit_sleep_log_count {0};
static std::atomic<int64_t> s3_put_rate_limit_rejected_log_count {0};

class TokenBucketRateLimiter::SimpleSpinLock {
public:
    SimpleSpinLock() = default;
    ~SimpleSpinLock() = default;

    void lock() {
        int spin_count = 0;
        static constexpr int MAX_SPIN_COUNT = 50;
        while (_flag.test_and_set(std::memory_order_acq_rel)) {
            spin_count++;
            if (spin_count >= MAX_SPIN_COUNT) {
                LOG(WARNING) << "Warning: Excessive spinning detected while acquiring lock. Spin "
                                "count: "
                             << spin_count;
                spin_count = 0;
            }
            // Spin until we acquire the lock
        }
    }

    void unlock() { _flag.clear(std::memory_order_release); }

private:
    std::atomic_flag _flag = ATOMIC_FLAG_INIT;
};

TokenBucketRateLimiter::TokenBucketRateLimiter(size_t max_speed, size_t max_burst, size_t limit)
        : _max_speed(max_speed),
          _max_burst(max_burst),
          _limit(limit),
          _mutex(std::make_unique<TokenBucketRateLimiter::SimpleSpinLock>()),
          _remain_tokens(max_burst) {}

TokenBucketRateLimiter::~TokenBucketRateLimiter() = default;

TokenBucketRateLimiterHolder::~TokenBucketRateLimiterHolder() = default;

std::pair<size_t, double> TokenBucketRateLimiter::_update_remain_token(long now, size_t amount) {
    // Values obtained under lock to be checked after release
    size_t count_value;
    double tokens_value;
    {
        std::lock_guard<SimpleSpinLock> lock(*_mutex);
        now = (now < _prev_ns_count) ? _prev_ns_count : now;
        if (_max_speed) {
            double delta_seconds =
                    _prev_ns_count ? static_cast<double>(now - _prev_ns_count) / NS : 0;
            _remain_tokens = std::min<double>(_remain_tokens + _max_speed * delta_seconds - amount,
                                              _max_burst);
        }
        _count += amount;
        count_value = _count;
        if (_limit && count_value > _limit) {
            // Keep rejection side-effect free. Roll back before releasing the lock so
            // concurrent callers cannot observe debt from a request that will not run.
            _count -= amount;
            if (_max_speed) {
                _remain_tokens = std::min<double>(_remain_tokens + amount, _max_burst);
            }
        }
        tokens_value = _remain_tokens;
        _prev_ns_count = now;
    }
    return {count_value, tokens_value};
}

int64_t TokenBucketRateLimiter::add(size_t amount) {
    int64_t sleep_time_ns = reserve(amount);
    if (sleep_time_ns > 0) {
        bthread_usleep(sleep_time_ns / 1000);
    }
    return sleep_time_ns;
}

int64_t TokenBucketRateLimiter::reserve(size_t amount) {
    // Values obtained under lock to be checked after release
    auto duration = std::chrono::steady_clock::now().time_since_epoch();
    auto time_nano_count = std::chrono::duration_cast<std::chrono::nanoseconds>(duration).count();
    auto [count_value, tokens_value] = _update_remain_token(time_nano_count, amount);

    if (_limit && count_value > _limit) {
        // CK would throw exception
        return -1;
    }

    // Wait unless there is positive amount of remain_tokens - throttling
    int64_t sleep_time_ns = 0;
    if (_max_speed && tokens_value < 0) {
        sleep_time_ns = static_cast<int64_t>(-tokens_value / _max_speed * NS);
    }

    return sleep_time_ns;
}

void TokenBucketRateLimiter::refund(size_t amount) {
    std::lock_guard<SimpleSpinLock> lock(*_mutex);
    if (_max_speed) {
        _remain_tokens = std::min<double>(_remain_tokens + amount, _max_burst);
    }
    _count = (_count >= amount) ? _count - amount : 0;
}

TokenBucketRateLimiterHolder::TokenBucketRateLimiterHolder(size_t max_speed, size_t max_burst,
                                                           size_t limit,
                                                           std::function<void(int64_t)> metric_func)
        : rate_limiter(std::make_shared<TokenBucketRateLimiter>(max_speed, max_burst, limit)),
          _enabled(max_speed > 0 || limit > 0),
          metric_func(std::move(metric_func)) {}

int64_t TokenBucketRateLimiterHolder::add(size_t amount) {
    return add_with_config(amount).sleep_duration;
}

TokenBucketRateLimiterResult TokenBucketRateLimiterHolder::add_with_config(size_t amount) {
    return _consume_with_config(amount, true);
}

TokenBucketRateLimiterResult TokenBucketRateLimiterHolder::reserve_with_config(size_t amount) {
    return _consume_with_config(amount, false);
}

TokenBucketRateLimiterResult TokenBucketRateLimiterHolder::_consume_with_config(size_t amount,
                                                                                bool wait) {
    // Snapshot the current limiter and consume outside the read lock. The waiting
    // path may sleep for a long time when throttled, and holding the read lock across
    // the sleep would block reset() (dynamic config update) for the whole duration.
    std::shared_ptr<TokenBucketRateLimiter> limiter;
    {
        std::shared_lock read {rate_limiter_rw_lock};
        limiter = rate_limiter;
    }
    TokenBucketRateLimiterResult result = {.sleep_duration = wait ? limiter->add(amount)
                                                                  : limiter->reserve(amount),
                                           .max_speed = limiter->get_max_speed(),
                                           .max_burst = limiter->get_max_burst(),
                                           .limit = limiter->get_limit()};
    metric_func(result.sleep_duration);
    return result;
}

std::shared_ptr<TokenBucketRateLimiter> TokenBucketRateLimiterHolder::charge(size_t amount) {
    std::shared_ptr<TokenBucketRateLimiter> limiter;
    {
        std::shared_lock read {rate_limiter_rw_lock};
        limiter = rate_limiter;
    }
    int64_t sleep_duration = limiter->add(amount);
    metric_func(sleep_duration);
    return sleep_duration < 0 ? nullptr : limiter;
}

int TokenBucketRateLimiterHolder::reset(size_t max_speed, size_t max_burst, size_t limit) {
    auto new_rate_limiter = std::make_shared<TokenBucketRateLimiter>(max_speed, max_burst, limit);
    {
        std::unique_lock write {rate_limiter_rw_lock};
        rate_limiter = std::move(new_rate_limiter);
        _enabled.store(max_speed > 0 || limit > 0, std::memory_order_release);
    }
    return 0;
}

size_t TokenBucketRateLimiterHolder::get_max_speed() const {
    std::shared_lock read {rate_limiter_rw_lock};
    return rate_limiter->get_max_speed();
}

size_t TokenBucketRateLimiterHolder::get_max_burst() const {
    std::shared_lock read {rate_limiter_rw_lock};
    return rate_limiter->get_max_burst();
}

size_t TokenBucketRateLimiterHolder::get_limit() const {
    std::shared_lock read {rate_limiter_rw_lock};
    return rate_limiter->get_limit();
}

std::string to_string(S3RateLimitType type) {
    switch (type) {
    case S3RateLimitType::GET:
        return "get";
    case S3RateLimitType::PUT:
        return "put";
    default:
        return std::to_string(static_cast<size_t>(type));
    }
}

S3RateLimitType string_to_s3_rate_limit_type(std::string_view value) {
    if (value == "get") {
        return S3RateLimitType::GET;
    } else if (value == "put") {
        return S3RateLimitType::PUT;
    }
    return S3RateLimitType::UNKNOWN;
}

std::function<void(int64_t)> s3_rate_limiter_metric_func(S3RateLimitType type) {
    switch (type) {
    case S3RateLimitType::GET:
        return metric_func_factory(s3_get_rate_limit_sleep_ns, s3_get_rate_limit_sleep_count,
                                   &s3_get_rate_limit_rejected_count);
    case S3RateLimitType::PUT:
        return metric_func_factory(s3_put_rate_limit_sleep_ns, s3_put_rate_limit_sleep_count,
                                   &s3_put_rate_limit_rejected_count);
    default:
        return [](int64_t) {};
    }
}

int64_t apply_s3_rate_limit(S3RateLimitType type, S3RateLimiterHolder* rate_limiter,
                            int64_t log_interval) {
    auto result = rate_limiter->add_with_config(1);
    auto sleep_duration = result.sleep_duration;
    if (log_interval <= 0 || sleep_duration == 0) {
        return sleep_duration;
    }

    auto is_get = type == S3RateLimitType::GET;
    auto* sleep_log_count =
            is_get ? &s3_get_rate_limit_sleep_log_count : &s3_put_rate_limit_sleep_log_count;
    auto* rejected_log_count =
            is_get ? &s3_get_rate_limit_rejected_log_count : &s3_put_rate_limit_rejected_log_count;

    if (sleep_duration > 0) {
        int64_t count = sleep_log_count->fetch_add(1, std::memory_order_relaxed) + 1;
        if (count == 1 || count % log_interval == 0) {
            LOG(INFO) << "S3 " << to_string(type) << " request is throttled by local rate limiter"
                      << ", sleep_ms=" << sleep_duration / 1000000 << ", sleep_count=" << count
                      << ", token_per_second=" << result.max_speed
                      << ", bucket_tokens=" << result.max_burst << ", token_limit=" << result.limit;
        }
    } else {
        int64_t count = rejected_log_count->fetch_add(1, std::memory_order_relaxed) + 1;
        if (count == 1 || count % log_interval == 0) {
            LOG(WARNING) << "S3 " << to_string(type) << " request is rejected by local rate limiter"
                         << ", rejected_count=" << count
                         << ", token_per_second=" << result.max_speed
                         << ", bucket_tokens=" << result.max_burst
                         << ", token_limit=" << result.limit;
        }
    }
    return sleep_duration;
}
} // namespace doris

Coverage Report

Created: 2026-09-29 12:58

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 "token_bucket_rate_limiter.h"
19
20		#include <bthread/bthread.h>
21		#include <glog/logging.h> // IWYU pragma: export
22
23		#include <atomic>
24		#include <chrono>
25		#include <mutex>
26		#include <thread>
27
28		#if defined(__APPLE__)
29		#include <ctime>
30		#endif
31
32		namespace doris {
33		// Just 10^6.
34		static constexpr auto NS = 1000000000UL;
35
36		bvar::Adder<int64_t> s3_get_rate_limit_sleep_ns("s3_get_rate_limit_sleep_ns");
37		bvar::Adder<int64_t> s3_get_rate_limit_sleep_count("s3_get_rate_limit_sleep_count");
38		bvar::Adder<int64_t> s3_get_rate_limit_rejected_count("s3_get_rate_limit_rejected_count");
39		bvar::Adder<int64_t> s3_put_rate_limit_sleep_ns("s3_put_rate_limit_sleep_ns");
40		bvar::Adder<int64_t> s3_put_rate_limit_sleep_count("s3_put_rate_limit_sleep_count");
41		bvar::Adder<int64_t> s3_put_rate_limit_rejected_count("s3_put_rate_limit_rejected_count");
42
43		static std::atomic<int64_t> s3_get_rate_limit_sleep_log_count {0};
44		static std::atomic<int64_t> s3_get_rate_limit_rejected_log_count {0};
45		static std::atomic<int64_t> s3_put_rate_limit_sleep_log_count {0};
46		static std::atomic<int64_t> s3_put_rate_limit_rejected_log_count {0};
47
48		class TokenBucketRateLimiter::SimpleSpinLock {
49		public:
50	3.44k	SimpleSpinLock() = default;
51		~SimpleSpinLock() = default;
52
53	1.21k	void lock() {
54	1.21k	int spin_count = 0;
55	1.21k	static constexpr int MAX_SPIN_COUNT = 50;
56	1.29k	while (_flag.test_and_set(std::memory_order_acq_rel)) {
57	77	spin_count++;
58	77	if (spin_count >= MAX_SPIN_COUNT) {
59	1	LOG(WARNING) << "Warning: Excessive spinning detected while acquiring lock. Spin "
60	1	"count: "
61	1	<< spin_count;
62	1	spin_count = 0;
63	1	}
64		// Spin until we acquire the lock
65	77	}
66	1.21k	}
67
68	1.21k	void unlock() { _flag.clear(std::memory_order_release); }
69
70		private:
71		std::atomic_flag _flag = ATOMIC_FLAG_INIT;
72		};
73
74		TokenBucketRateLimiter::TokenBucketRateLimiter(size_t max_speed, size_t max_burst, size_t limit)
75	3.44k	: _max_speed(max_speed),
76	3.44k	_max_burst(max_burst),
77	3.44k	_limit(limit),
78	3.44k	_mutex(std::make_unique<TokenBucketRateLimiter::SimpleSpinLock>()),
79	3.44k	_remain_tokens(max_burst) {}
80
81	3.44k	TokenBucketRateLimiter::~TokenBucketRateLimiter() = default;
82
83	3.27k	TokenBucketRateLimiterHolder::~TokenBucketRateLimiterHolder() = default;
84
85	1.21k	std::pair<size_t, double> TokenBucketRateLimiter::_update_remain_token(long now, size_t amount) {
86		// Values obtained under lock to be checked after release
87	1.21k	size_t count_value;
88	1.21k	double tokens_value;
89	1.21k	{
90	1.21k	std::lock_guard<SimpleSpinLock> lock(*_mutex);
91	1.21k	now = (now < _prev_ns_count) ? _prev_ns_count : now;
92	1.21k	if (_max_speed) {
93	1.18k	double delta_seconds =
94	1.18k	_prev_ns_count ? static_cast<double>(now - _prev_ns_count) / NS : 0;
95	1.18k	_remain_tokens = std::min<double>(_remain_tokens + _max_speed * delta_seconds - amount,
96	1.18k	_max_burst);
97	1.18k	}
98	1.21k	_count += amount;
99	1.21k	count_value = _count;
100	1.21k	if (_limit && count_value > _limit) {
101		// Keep rejection side-effect free. Roll back before releasing the lock so
102		// concurrent callers cannot observe debt from a request that will not run.
103	44	_count -= amount;
104	44	if (_max_speed) {
105	30	_remain_tokens = std::min<double>(_remain_tokens + amount, _max_burst);
106	30	}
107	44	}
108	1.21k	tokens_value = _remain_tokens;
109	1.21k	_prev_ns_count = now;
110	1.21k	}
111	1.21k	return {count_value, tokens_value};
112	1.21k	}
113
114	1.17k	int64_t TokenBucketRateLimiter::add(size_t amount) {
115	1.17k	int64_t sleep_time_ns = reserve(amount);
116	1.17k	if (sleep_time_ns > 0) {
117	576	bthread_usleep(sleep_time_ns / 1000);
118	576	}
119	1.17k	return sleep_time_ns;
120	1.17k	}
121
122	1.21k	int64_t TokenBucketRateLimiter::reserve(size_t amount) {
123		// Values obtained under lock to be checked after release
124	1.21k	auto duration = std::chrono::steady_clock::now().time_since_epoch();
125	1.21k	auto time_nano_count = std::chrono::duration_cast<std::chrono::nanoseconds>(duration).count();
126	1.21k	auto [count_value, tokens_value] = _update_remain_token(time_nano_count, amount);
127
128	1.21k	if (_limit && count_value > _limit) {
129		// CK would throw exception
130	44	return -1;
131	44	}
132
133		// Wait unless there is positive amount of remain_tokens - throttling
134	1.16k	int64_t sleep_time_ns = 0;
135	1.16k	if (_max_speed && tokens_value < 0) {
136	594	sleep_time_ns = static_cast<int64_t>(-tokens_value / _max_speed * NS);
137	594	}
138
139	1.16k	return sleep_time_ns;
140	1.21k	}
141
142	6	void TokenBucketRateLimiter::refund(size_t amount) {
143	6	std::lock_guard<SimpleSpinLock> lock(*_mutex);
144	6	if (_max_speed) {
145	6	_remain_tokens = std::min<double>(_remain_tokens + amount, _max_burst);
146	6	}
147	6	_count = (_count >= amount) ? _count - amount : 0;
148	6	}
149
150		TokenBucketRateLimiterHolder::TokenBucketRateLimiterHolder(size_t max_speed, size_t max_burst,
151		size_t limit,
152		std::function<void(int64_t)> metric_func)
153	3.27k	: rate_limiter(std::make_shared<TokenBucketRateLimiter>(max_speed, max_burst, limit)),
154	3.27k	_enabled(max_speed > 0 \|\| limit > 0),
155	3.27k	metric_func(std::move(metric_func)) {}
156
157	88	int64_t TokenBucketRateLimiterHolder::add(size_t amount) {
158	88	return add_with_config(amount).sleep_duration;
159	88	}
160
161	1.15k	TokenBucketRateLimiterResult TokenBucketRateLimiterHolder::add_with_config(size_t amount) {
162	1.15k	return _consume_with_config(amount, true);
163	1.15k	}
164
165	42	TokenBucketRateLimiterResult TokenBucketRateLimiterHolder::reserve_with_config(size_t amount) {
166	42	return _consume_with_config(amount, false);
167	42	}
168
169		TokenBucketRateLimiterResult TokenBucketRateLimiterHolder::_consume_with_config(size_t amount,
170	1.19k	bool wait) {
171		// Snapshot the current limiter and consume outside the read lock. The waiting
172		// path may sleep for a long time when throttled, and holding the read lock across
173		// the sleep would block reset() (dynamic config update) for the whole duration.
174	1.19k	std::shared_ptr<TokenBucketRateLimiter> limiter;
175	1.19k	{
176	1.19k	std::shared_lock read {rate_limiter_rw_lock};
177	1.19k	limiter = rate_limiter;
178	1.19k	}
179	1.19k	TokenBucketRateLimiterResult result = {.sleep_duration = wait ? limiter->add(amount)
180	1.19k	: limiter->reserve(amount),
181	1.19k	.max_speed = limiter->get_max_speed(),
182	1.19k	.max_burst = limiter->get_max_burst(),
183	1.19k	.limit = limiter->get_limit()};
184	1.19k	metric_func(result.sleep_duration);
185	1.19k	return result;
186	1.19k	}
187
188	14	std::shared_ptr<TokenBucketRateLimiter> TokenBucketRateLimiterHolder::charge(size_t amount) {
189	14	std::shared_ptr<TokenBucketRateLimiter> limiter;
190	14	{
191	14	std::shared_lock read {rate_limiter_rw_lock};
192	14	limiter = rate_limiter;
193	14	}
194	14	int64_t sleep_duration = limiter->add(amount);
195	14	metric_func(sleep_duration);
196	14	return sleep_duration < 0 ? nullptr : limiter;
197	14	}
198
199	170	int TokenBucketRateLimiterHolder::reset(size_t max_speed, size_t max_burst, size_t limit) {
200	170	auto new_rate_limiter = std::make_shared<TokenBucketRateLimiter>(max_speed, max_burst, limit);
201	170	{
202	170	std::unique_lock write {rate_limiter_rw_lock};
203	170	rate_limiter = std::move(new_rate_limiter);
204	170	_enabled.store(max_speed > 0 \|\| limit > 0, std::memory_order_release);
205	170	}
206	170	return 0;
207	170	}
208
209	458	size_t TokenBucketRateLimiterHolder::get_max_speed() const {
210	458	std::shared_lock read {rate_limiter_rw_lock};
211	458	return rate_limiter->get_max_speed();
212	458	}
213
214	136	size_t TokenBucketRateLimiterHolder::get_max_burst() const {
215	136	std::shared_lock read {rate_limiter_rw_lock};
216	136	return rate_limiter->get_max_burst();
217	136	}
218
219	120	size_t TokenBucketRateLimiterHolder::get_limit() const {
220	120	std::shared_lock read {rate_limiter_rw_lock};
221	120	return rate_limiter->get_limit();
222	120	}
223
224	106	std::string to_string(S3RateLimitType type) {
225	106	switch (type) {
226	78	case S3RateLimitType::GET:
227	78	return "get";
228	28	case S3RateLimitType::PUT:
229	28	return "put";
230	0	default:
231	0	return std::to_string(static_cast<size_t>(type));
232	106	}
233	106	}
234
235	0	S3RateLimitType string_to_s3_rate_limit_type(std::string_view value) {
236	0	if (value == "get") {
237	0	return S3RateLimitType::GET;
238	0	} else if (value == "put") {
239	0	return S3RateLimitType::PUT;
240	0	}
241	0	return S3RateLimitType::UNKNOWN;
242	0	}
243
244	4	std::function<void(int64_t)> s3_rate_limiter_metric_func(S3RateLimitType type) {
245	4	switch (type) {
246	2	case S3RateLimitType::GET:
247	2	return metric_func_factory(s3_get_rate_limit_sleep_ns, s3_get_rate_limit_sleep_count,
248	2	&s3_get_rate_limit_rejected_count);
249	2	case S3RateLimitType::PUT:
250	2	return metric_func_factory(s3_put_rate_limit_sleep_ns, s3_put_rate_limit_sleep_count,
251	2	&s3_put_rate_limit_rejected_count);
252	0	default:
253	0	return [](int64_t) {};
254	4	}
255	4	}
256
257		int64_t apply_s3_rate_limit(S3RateLimitType type, S3RateLimiterHolder* rate_limiter,
258	14	int64_t log_interval) {
259	14	auto result = rate_limiter->add_with_config(1);
260	14	auto sleep_duration = result.sleep_duration;
261	14	if (log_interval <= 0 \|\| sleep_duration == 0) {
262	8	return sleep_duration;
263	8	}
264
265	6	auto is_get = type == S3RateLimitType::GET;
266	6	auto* sleep_log_count =
267	6	is_get ? &s3_get_rate_limit_sleep_log_count : &s3_put_rate_limit_sleep_log_count;
268	6	auto* rejected_log_count =
269	6	is_get ? &s3_get_rate_limit_rejected_log_count : &s3_put_rate_limit_rejected_log_count;
270
271	6	if (sleep_duration > 0) {
272	0	int64_t count = sleep_log_count->fetch_add(1, std::memory_order_relaxed) + 1;
273	0	if (count == 1 \|\| count % log_interval == 0) {
274	0	LOG(INFO) << "S3 " << to_string(type) << " request is throttled by local rate limiter"
275	0	<< ", sleep_ms=" << sleep_duration / 1000000 << ", sleep_count=" << count
276	0	<< ", token_per_second=" << result.max_speed
277	0	<< ", bucket_tokens=" << result.max_burst << ", token_limit=" << result.limit;
278	0	}
279	6	} else {
280	6	int64_t count = rejected_log_count->fetch_add(1, std::memory_order_relaxed) + 1;
281	6	if (count == 1 \|\| count % log_interval == 0) {
282		LOG(WARNING) << "S3 " << to_string(type) << " request is rejected by local rate limiter"
283	2	<< ", rejected_count=" << count
284	2	<< ", token_per_second=" << result.max_speed
285	2	<< ", bucket_tokens=" << result.max_burst
286	2	<< ", token_limit=" << result.limit;
287	2	}
288	6	}
289	6	return sleep_duration;
290	14	}
291		} // namespace doris