Fail the RPC ASSP if the sending socket is overcrowded

src/brpc/policy/baidu_rpc_protocol.cpp

@@ -386,12 +386,19 @@ void ProcessRpcRequest(InputMessageBase* msg_base) {
             cntl->SetFailed(ELOGOFF, "Server is stopping");
             break;
         }
+
+        if (socket->is_overcrowded()) {
+            cntl->SetFailed(EOVERCROWDED, "Connection to %s is overcrowded",
+                            butil::endpoint2str(socket->remote_side()).c_str());
+            break;
+        }
         
         if (!server_accessor.AddConcurrency(cntl.get())) {
             cntl->SetFailed(ELIMIT, "Reached server's max_concurrency=%d",
                             server->options().max_concurrency);
             break;
         }
+
         if (FLAGS_usercode_in_pthread && TooManyUserCode()) {
             cntl->SetFailed(ELIMIT, "Too many user code to run when"
                             " -usercode_in_pthread is on");

src/brpc/policy/http_rpc_protocol.cpp

@@ -1183,6 +1183,11 @@ void ProcessHttpRequest(InputMessageBase *msg) {
     // NOTE: accesses to builtin services are not counted as part of
     // concurrency, therefore are not limited by ServerOptions.max_concurrency.
     if (!sp->is_builtin_service && !sp->params.is_tabbed) {
+        if (socket->is_overcrowded()) {
+            cntl->SetFailed(EOVERCROWDED, "Connection to %s is overcrowded",
+                            butil::endpoint2str(socket->remote_side()).c_str());
+            return SendHttpResponse(cntl.release(), server, method_status);
+        }
         if (!server_accessor.AddConcurrency(cntl.get())) {
             cntl->SetFailed(ELIMIT, "Reached server's max_concurrency=%d",
                             server->options().max_concurrency);

src/brpc/policy/hulu_pbrpc_protocol.cpp

@@ -418,6 +418,12 @@ void ProcessHuluRequest(InputMessageBase* msg_base) {
             break;
         }
 
+        if (socket->is_overcrowded()) {
+            cntl->SetFailed(EOVERCROWDED, "Connection to %s is overcrowded",
+                            butil::endpoint2str(socket->remote_side()).c_str());
+            break;
+        }
+
         if (!server_accessor.AddConcurrency(cntl.get())) {
             cntl->SetFailed(ELIMIT, "Reached server's max_concurrency=%d",
                             server->options().max_concurrency);

src/brpc/policy/nshead_protocol.cpp

@@ -290,6 +290,11 @@ void ProcessNsheadRequest(InputMessageBase* msg_base) {
             cntl->SetFailed(ELOGOFF, "Server is stopping");
             break;
         }
+        if (socket->is_overcrowded()) {
+            cntl->SetFailed(EOVERCROWDED, "Connection to %s is overcrowded",
+                            butil::endpoint2str(socket->remote_side()).c_str());
+            break;
+        }
         if (!server_accessor.AddConcurrency(cntl)) {
             cntl->SetFailed(ELIMIT, "Reached server's max_concurrency=%d",
                             server->options().max_concurrency);

src/brpc/policy/sofa_pbrpc_protocol.cpp

@@ -382,6 +382,12 @@ void ProcessSofaRequest(InputMessageBase* msg_base) {
             break;
         }
 
+        if (socket->is_overcrowded()) {
+            cntl->SetFailed(EOVERCROWDED, "Connection to %s is overcrowded",
+                            butil::endpoint2str(socket->remote_side()).c_str());
+            break;
+        }
+
         if (!server_accessor.AddConcurrency(cntl.get())) {
             cntl->SetFailed(ELIMIT, "Reached server's max_concurrency=%d",
                             server->options().max_concurrency);

src/brpc/socket.h

@@ -452,6 +452,9 @@ public:
     // A brief description of this socket, consistent with os << *this
     std::string description() const;
 
+    // Returns true if the remote side is overcrowded.
+    bool is_overcrowded() const { return _overcrowded; }
+
 private:
     DISALLOW_COPY_AND_ASSIGN(Socket);