common/cpp/token_bucket_rate_limiter.h
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 | | #pragma once |
19 | | #include <bvar/bvar.h> |
20 | | |
21 | | #include <functional> |
22 | | #include <memory> |
23 | | #include <shared_mutex> |
24 | | #include <string> |
25 | | #include <string_view> |
26 | | |
27 | | namespace doris { |
28 | | enum class S3RateLimitType : int { |
29 | | GET = 0, |
30 | | PUT, |
31 | | UNKNOWN, |
32 | | }; |
33 | | |
34 | | extern std::string to_string(S3RateLimitType type); |
35 | | extern S3RateLimitType string_to_s3_rate_limit_type(std::string_view value); |
36 | | |
37 | | inline auto metric_func_factory(bvar::Adder<int64_t>& sleep_ns_bvar, |
38 | | bvar::Adder<int64_t>& sleep_count_bvar, |
39 | 16 | bvar::Adder<int64_t>* rejected_count_bvar = nullptr) { |
40 | 16 | return [&, rejected_count_bvar](int64_t ns) { |
41 | 0 | if (ns > 0) { |
42 | 0 | sleep_ns_bvar << ns; |
43 | 0 | sleep_count_bvar << 1; |
44 | 0 | } else if (ns < 0 && rejected_count_bvar != nullptr) { |
45 | 0 | *rejected_count_bvar << 1; |
46 | 0 | } |
47 | 0 | }; |
48 | 16 | } |
49 | | |
50 | | class TokenBucketRateLimiter { |
51 | | public: |
52 | | static constexpr size_t default_burst_seconds = 1; |
53 | | |
54 | | TokenBucketRateLimiter(size_t max_speed, size_t max_burst, size_t limit); |
55 | | ~TokenBucketRateLimiter(); |
56 | | |
57 | | // Use `amount` remain_tokens, sleeps if required or throws exception on limit overflow. |
58 | | // Returns duration of sleep in nanoseconds (to distinguish sleeping on different kinds of S3RateLimiters for metrics) |
59 | | int64_t add(size_t amount); |
60 | | |
61 | 0 | size_t get_max_speed() const { return _max_speed; } |
62 | | |
63 | 0 | size_t get_max_burst() const { return _max_burst; } |
64 | | |
65 | 0 | size_t get_limit() const { return _limit; } |
66 | | |
67 | | private: |
68 | | std::pair<size_t, double> _update_remain_token(long now, size_t amount); |
69 | | size_t _count {0}; |
70 | | const size_t _max_speed {0}; // in tokens per second. which indicates the QPS |
71 | | const size_t _max_burst {0}; // in tokens. which indicates the token bucket size |
72 | | const uint64_t _limit {0}; // 0 - not limited. |
73 | | class SimpleSpinLock; |
74 | | std::unique_ptr<SimpleSpinLock> _mutex; |
75 | | // Amount of remain_tokens available in token bucket. Updated in `add` method. |
76 | | double _remain_tokens {0}; |
77 | | long _prev_ns_count {0}; // Previous `add` call time (in nanoseconds). |
78 | | }; |
79 | | |
80 | | class TokenBucketRateLimiterHolder { |
81 | | public: |
82 | | TokenBucketRateLimiterHolder(size_t max_speed, size_t max_burst, size_t limit, |
83 | | std::function<void(int64_t)> metric_func); |
84 | | ~TokenBucketRateLimiterHolder(); |
85 | | |
86 | | int64_t add(size_t amount); |
87 | | |
88 | | int reset(size_t max_speed, size_t max_burst, size_t limit); |
89 | | |
90 | 0 | size_t get_max_speed() const { return rate_limiter->get_max_speed(); } |
91 | | |
92 | 0 | size_t get_max_burst() const { return rate_limiter->get_max_burst(); } |
93 | | |
94 | 0 | size_t get_limit() const { return rate_limiter->get_limit(); } |
95 | | |
96 | | private: |
97 | | std::shared_mutex rate_limiter_rw_lock; |
98 | | std::unique_ptr<TokenBucketRateLimiter> rate_limiter; |
99 | | // Record the correspoding sleeping time(unit is ms) |
100 | | std::function<void(int64_t)> metric_func; |
101 | | }; |
102 | | |
103 | | using S3RateLimiter = TokenBucketRateLimiter; |
104 | | using S3RateLimiterHolder = TokenBucketRateLimiterHolder; |
105 | | |
106 | | std::function<void(int64_t)> s3_rate_limiter_metric_func(S3RateLimitType type); |
107 | | int64_t apply_s3_rate_limit(S3RateLimitType type, S3RateLimiterHolder* rate_limiter, |
108 | | int64_t log_interval); |
109 | | } // namespace doris |