revert code on server-level mc & add default value for method-level mc

src/brpc/details/server_private_accessor.h

@@ -35,22 +35,22 @@ public:
     }
 
     void AddError() {
-        _server->_nerror << 1;
+        _server->_nerror_bvar << 1;
     }
 
     // Returns true if the `max_concurrency' limit is not reached.
     bool AddConcurrency(Controller* c) {
-        if (NULL != _server->_cl) {
-            c->add_flag(Controller::FLAGS_CONCURRENCY_LIMITER_REQUESTED);
-            return _server->_cl->OnRequested();
-        } 
-        return true;
+        if (_server->options().max_concurrency <= 0) {
+            return true;
+        }
+        c->add_flag(Controller::FLAGS_ADDED_CONCURRENCY);
+        return (butil::subtle::NoBarrier_AtomicIncrement(&_server->_concurrency, 1)
+                <= _server->options().max_concurrency);
     }
 
     void RemoveConcurrency(const Controller* c) {
-        if (c->has_flag(Controller::FLAGS_CONCURRENCY_LIMITER_REQUESTED)){
-            CHECK(_server->_cl != NULL);
-            _server->_cl->OnResponded(c->ErrorCode(), c->latency_us());
+        if (c->has_flag(Controller::FLAGS_ADDED_CONCURRENCY)) {
+            butil::subtle::NoBarrier_AtomicIncrement(&_server->_concurrency, -1);
         }
     }
 

src/brpc/server.cpp

@@ -272,7 +272,8 @@ void* Server::UpdateDerivedVars(void* arg) {
     std::vector<SocketId> conns;
     std::vector<SocketId> internal_conns;
     
-    server->_nerror.expose_as(prefix, "error");
+    server->_nerror_bvar.expose_as(prefix, "error");
+    //server->_nrefused_bvar.expose_as(prefix, "refused");
 
     bvar::PassiveStatus<timeval> uptime_st(
         prefix, "uptime", GetUptime, (void*)(intptr_t)start_us);
@@ -382,7 +383,9 @@ Server::Server(ProfilerLinker)
     , _last_start_time(0)
     , _derivative_thread(INVALID_BTHREAD)
     , _keytable_pool(NULL)
-    , _cl(NULL) {
+    , _concurrency(0) {
+    BAIDU_CASSERT(offsetof(Server, _concurrency) % 64 == 0,	
+                  Server_concurrency_must_be_aligned_by_cacheline);
 }
 
 Server::~Server() {
@@ -423,10 +426,6 @@ Server::~Server() {
         delete _options.auth;
         _options.auth = NULL;
     }
-    if (_cl) {
-        _cl->Destroy();
-        _cl = NULL;
-    }
 }
 
 int Server::AddBuiltinServices() {
@@ -664,6 +663,27 @@ static int get_port_from_fd(int fd) {
     return ntohs(addr.sin_port);
 }
 
+static bool CreateConcurrencyLimiter(const AdaptiveMaxConcurrency& amc,
+                                     ConcurrencyLimiter** out) {
+    if (amc.type() == AdaptiveMaxConcurrency::UNLIMITED()) {
+        *out = NULL;
+        return true;
+    }
+    const ConcurrencyLimiter* cl =
+        ConcurrencyLimiterExtension()->Find(amc.type().c_str());
+    if (cl == NULL) {
+        LOG(ERROR) << "Fail to find ConcurrencyLimiter by `" << amc.value() << "'";
+        return false;
+    }
+    ConcurrencyLimiter* cl_copy = cl->New(amc);
+    if (cl_copy == NULL) {
+        LOG(ERROR) << "Fail to new ConcurrencyLimiter";
+        return false;
+    }
+    *out = cl_copy;
+    return true;
+}
+
 static AdaptiveMaxConcurrency g_default_max_concurrency_of_method = 0;
 
 int Server::StartInternal(const butil::ip_t& ip,
@@ -835,6 +855,8 @@ int Server::StartInternal(const butil::ip_t& ip,
             }
         }
     }
+
+    _concurrency = 0;
     
     if (_options.has_builtin_services &&
         _builtin_service_count <= 0 &&
@@ -872,28 +894,21 @@ int Server::StartInternal(const butil::ip_t& ip,
         bthread_setconcurrency(_options.num_threads);
     }
 
-    if (NULL != _cl) {
-        _cl->Destroy();
-        _cl = NULL;
-    }
-    if (_options.max_concurrency != "constant" || 
-        static_cast<int>(_options.max_concurrency) != 0) {
-        _cl = ConcurrencyLimiter::CreateConcurrencyLimiterOrDie(
-            _options.max_concurrency);
-        _cl->Expose("Server_Concurrency_Limiter");
-    }
-
     for (MethodMap::iterator it = _method_map.begin();
         it != _method_map.end(); ++it) {
         if (it->second.is_builtin_service) {
             it->second.status->SetConcurrencyLimiter(NULL);
-        } else if (it->second.max_concurrency == "constant" &&
-            static_cast<int>(it->second.max_concurrency) == 0) {
-            it->second.status->SetConcurrencyLimiter(NULL);
         } else {
-            it->second.status->SetConcurrencyLimiter(
-                ConcurrencyLimiter::CreateConcurrencyLimiterOrDie(
-                    it->second.max_concurrency));
+            const AdaptiveMaxConcurrency* amc = &it->second.max_concurrency;
+            if (amc->type() == AdaptiveMaxConcurrency::UNLIMITED()) {
+                amc = &_options.method_max_concurrency;
+            }
+            ConcurrencyLimiter* cl = NULL;
+            if (!CreateConcurrencyLimiter(*amc, &cl)) {
+                LOG(ERROR) << "Fail to create ConcurrencyLimiter for method";
+                return -1;
+            }
+            it->second.status->SetConcurrencyLimiter(cl);
         }
     }
     
@@ -1986,16 +2001,13 @@ bool Server::ClearCertMapping(CertMaps& bg) {
 }
 
 int Server::ResetMaxConcurrency(int max_concurrency) {
-    LOG(WARNING) << "ResetMaxConcurrency is already deprecated";
-    return 0;
-}
-
-int Server::max_concurrency() const {
-    if (NULL != _cl) {
-        return _cl->max_concurrency();
-    } else {
-        return g_default_max_concurrency_of_method;
+    if (!IsRunning()) {
+        LOG(WARNING) << "ResetMaxConcurrency is only allowd for a Running Server";
+        return -1;
     }
+    // Assume that modifying int32 is atomical in X86
+    _options.max_concurrency = max_concurrency;
+    return 0;
 }
 
 AdaptiveMaxConcurrency& Server::MaxConcurrencyOf(MethodProperty* mp) {

src/brpc/server.h

@@ -36,7 +36,6 @@
 #include "brpc/builtin/tabbed.h"
 #include "brpc/details/profiler_linker.h"
 #include "brpc/health_reporter.h"
-#include "brpc/concurrency_limiter.h"
 #include "brpc/adaptive_max_concurrency.h"
 
 extern "C" {
@@ -101,14 +100,14 @@ struct ServerOptions {
     // Default: #cpu-cores
     int num_threads;
 
-    // Limit number of requests processed in parallel. To limit the max
-    // concurrency of a method, use server.MaxConcurrencyOf("xxx") instead.
+    // Server-level max concurrency.
+    // "concurrency" = "number of requests processed in parallel"
     //
     // In a traditional server, number of pthread workers also limits
     // concurrency. However brpc runs requests in bthreads which are
     // mapped to pthread workers, when a bthread context switches, it gives
     // the pthread worker to another bthread, yielding a higher concurrency
-    // than number of pthreads. In some situation, higher concurrency may
+    // than number of pthreads. In some situations, higher concurrency may
     // consume more resources, to protect the server from running out of
     // resources, you may set this option.
     // If the server reaches the limitation, it responds client with ELIMIT
@@ -116,12 +115,11 @@ struct ServerOptions {
     // shall try another server.
     // NOTE: accesses to builtin services are not limited by this option.
     // Default: 0 (unlimited)
-    // NOTE: Once you have chosen the automatic concurrency limit strategy, brpc
-    // ONLY limits concurrency at the method level, And each method will use
-    // the strategy you set in ServerOptions to limit the maximum concurrency, 
-    // even if you have set a maximum concurrency through `MaxConcurrencyOf`.
- 
-    AdaptiveMaxConcurrency max_concurrency;
+    int max_concurrency;
+
+    // Default value of method-level max concurrencies,
+    // Overridable by Server.MaxConcurrencyOf().
+    AdaptiveMaxConcurrency method_max_concurrency;
 
     // -------------------------------------------------------
     // Differences between session-local and thread-local data
@@ -484,13 +482,9 @@ public:
     // current_tab_name is the tab highlighted.
     void PrintTabsBody(std::ostream& os, const char* current_tab_name) const;
 
-    
     // This method is already deprecated.You should NOT call it anymore.
     int ResetMaxConcurrency(int max_concurrency);
 
-    // Server's current max concurrency
-    int max_concurrency() const;
-
     // Get/set max_concurrency associated with a method.
     // Example:
     //    server.MaxConcurrencyOf("example.EchoService.Echo") = 10;
@@ -659,11 +653,10 @@ friend class Controller;
     
     bthread_keytable_pool_t* _keytable_pool;
 
-    // FIXME: Temporarily for `ServerPrivateAccessor' to change this bvar
-    //        Replace `ServerPrivateAccessor' with other private-access
-    //        mechanism
-    mutable bvar::Adder<int64_t> _nerror;
-    ConcurrencyLimiter* _cl;
+    // mutable is required for `ServerPrivateAccessor' to change this bvar
+    mutable bvar::Adder<int64_t> _nerror_bvar;
+    mutable int32_t BAIDU_CACHELINE_ALIGNMENT _concurrency;
+
 };
 
 // Get the data attached to current searching thread. The data is created by