Commit 2cb180f3 authored by Ge Jun's avatar Ge Jun

Simplify ConstantConcurrencyLimiter

parent 7c65df38
...@@ -19,29 +19,25 @@ ...@@ -19,29 +19,25 @@
namespace brpc { namespace brpc {
namespace policy { namespace policy {
bool ConstantConcurrencyLimiter::OnRequested() { ConstantConcurrencyLimiter::ConstantConcurrencyLimiter(int max_concurrency)
const int32_t current_concurrency = : _max_concurrency(max_concurrency) {
_current_concurrency.fetch_add(1, butil::memory_order_relaxed);
if (_max_concurrency != 0 && current_concurrency >= _max_concurrency) {
return false;
}
return true;
} }
void ConstantConcurrencyLimiter::OnResponded(int error_code, int64_t latency) { bool ConstantConcurrencyLimiter::OnRequested(int current_concurrency) {
_current_concurrency.fetch_sub(1, butil::memory_order_relaxed); return current_concurrency <= _max_concurrency;
} }
int ConstantConcurrencyLimiter::Expose(const butil::StringPiece& prefix) { void ConstantConcurrencyLimiter::OnResponded(int error_code, int64_t latency) {
return 0;
} }
ConstantConcurrencyLimiter* ConstantConcurrencyLimiter::New() const { int ConstantConcurrencyLimiter::MaxConcurrency() {
return new (std::nothrow) ConstantConcurrencyLimiter; return _max_concurrency.load(butil::memory_order_relaxed);
} }
void ConstantConcurrencyLimiter::Destroy() { ConstantConcurrencyLimiter*
delete this; ConstantConcurrencyLimiter::New(const AdaptiveMaxConcurrency& amc) const {
CHECK_EQ(amc.type(), AdaptiveMaxConcurrency::CONSTANT());
return new ConstantConcurrencyLimiter(static_cast<int>(amc));
} }
} // namespace policy } // namespace policy
......
...@@ -24,19 +24,18 @@ namespace policy { ...@@ -24,19 +24,18 @@ namespace policy {
class ConstantConcurrencyLimiter : public ConcurrencyLimiter { class ConstantConcurrencyLimiter : public ConcurrencyLimiter {
public: public:
ConstantConcurrencyLimiter() : _current_concurrency(0) {} explicit ConstantConcurrencyLimiter(int max_concurrency);
~ConstantConcurrencyLimiter() {} bool OnRequested(int current_concurrency) override;
bool OnRequested() override;
void OnResponded(int error_code, int64_t latency_us) override; void OnResponded(int error_code, int64_t latency_us) override;
int Expose(const butil::StringPiece& prefix) override; int MaxConcurrency() override;
ConstantConcurrencyLimiter* New() const override;
void Destroy() override; ConstantConcurrencyLimiter* New(const AdaptiveMaxConcurrency&) const override;
private: private:
butil::atomic<int32_t> _current_concurrency; butil::atomic<int> _max_concurrency;
}; };
} // namespace policy } // namespace policy
......
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment