Add monitoring item acc_requests

src/brpc/builtin/connections_service.cpp

@@ -122,7 +122,6 @@ void ConnectionsService::PrintConnections(
         os << "<th>SSL</th>"
             "<th>Protocol</th>"
             "<th>fd</th>"
-            "<th>acc_errors</th>"
             "<th>InBytes/s</th>"
             "<th>In/s</th>"
             "<th>InBytes/m</th>"
@@ -139,7 +138,7 @@ void ConnectionsService::PrintConnections(
         if (need_local) {
             os << "Local|";
         }
-        os << "SSL|Protocol    |fd   |acc_errors|"
+        os << "SSL|Protocol    |fd   |"
             "InBytes/s|In/s  |InBytes/m |In/m    |"
             "OutBytes/s|Out/s |OutBytes/m|Out/m   |"
             "Rtt/Var(ms)|SocketId\n";
@@ -178,7 +177,6 @@ void ConnectionsService::PrintConnections(
             os << min_width("-", 3) << bar
                << min_width("-", 12) << bar
                << min_width("-", 5) << bar
-               << min_width(ptr->acc_errors(), 10) << bar
                << min_width("-", 9) << bar
                << min_width("-", 6) << bar
                << min_width("-", 10) << bar
@@ -290,8 +288,7 @@ void ConnectionsService::PrintConnections(
             } else {
                 os << min_width("-", 5) << bar;
             }
-            os << min_width(ptr->acc_errors(), 10) << bar
-               << min_width(stat.in_size_s, 9) << bar
+            os << min_width(stat.in_size_s, 9) << bar
                << min_width(stat.in_num_messages_s, 6) << bar
                << min_width(stat.in_size_m, 10) << bar
                << min_width(stat.in_num_messages_m, 8) << bar

src/brpc/controller.cpp

@@ -696,14 +696,23 @@ inline bool does_error_affect_main_socket(int error_code) {
 //      entire RPC (specified by c->FailedInline()).
 void Controller::Call::OnComplete(
         Controller* c, int error_code/*note*/, bool responded, bool end_of_rpc) {
-    if (enable_circuit_breaker && sending_sock) {
-        sending_sock->FeedbackCircuitBreaker(error_code, 
-            butil::gettimeofday_us() - begin_time_us);
+    if (stream_user_data) {
+        stream_user_data->DestroyStreamUserData(sending_sock, c, error_code, end_of_rpc);
+        stream_user_data = NULL;
     }
-    if (error_code != 0 && sending_sock) {
-        sending_sock->AddErrorCount();
+
+    if (sending_sock) {
+        sending_sock->AddRequestCount();
+        if (error_code != 0) {
+            sending_sock->AddErrorCount();
+        }
+
+        if (enable_circuit_breaker) {
+            sending_sock->FeedbackCircuitBreaker(error_code, 
+                butil::gettimeofday_us() - begin_time_us);
+        }
     }
- 
+
     switch (c->connection_type()) {
     case CONNECTION_TYPE_UNKNOWN:
         break;
@@ -776,11 +785,6 @@ void Controller::Call::OnComplete(
         c->_lb->Feedback(info);
     }
 
-    if (stream_user_data) {
-        stream_user_data->DestroyStreamUserData(sending_sock, c, error_code, end_of_rpc);
-        stream_user_data = NULL;
-    }
-
     // Release the `Socket' we used to send/receive data
     sending_sock.reset(NULL);
 }

src/brpc/socket.cpp

@@ -179,6 +179,7 @@ public:
     CircuitBreaker circuit_breaker;
 
     butil::atomic<uint64_t> acc_errors;
+    butil::atomic<uint64_t> acc_requests;
 
     explicit SharedPart(SocketId creator_socket_id);
     ~SharedPart();
@@ -806,7 +807,17 @@ int Socket::ReleaseAdditionalReference() {
 }
 
 void Socket::AddErrorCount() {
-    GetOrNewSharedPart()->acc_errors.fetch_add(1, butil::memory_order_relaxed);
+    SharedPart* sp = GetSharedPart();
+    if (sp) {
+        sp->acc_errors.fetch_add(1, butil::memory_order_relaxed);
+    }
+}
+
+void Socket::AddRequestCount() {
+    SharedPart* sp = GetSharedPart();
+    if (sp) {
+        sp->acc_requests.fetch_add(1, butil::memory_order_relaxed);
+    }
 }
 
 uint64_t Socket::acc_errors() const {
@@ -2137,6 +2148,7 @@ void Socket::DebugSocket(std::ostream& os, SocketId id) {
            << "\n  out_num_messages=" << sp->out_num_messages.load(butil::memory_order_relaxed)
            << "\n  health_score=" << sp->circuit_breaker.health_score()
            << "\n  isolated_times=" << sp->circuit_breaker.isolated_times()
+           << "\n  acc_requests=" << sp->acc_requests.load(butil::memory_order_relaxed)
            << "\n  acc_errors=" << sp->acc_errors.load(butil::memory_order_relaxed)
            << "\n}";
     }

src/brpc/socket.h

@@ -319,7 +319,7 @@ public:
 
     void AddErrorCount();
 
-    uint64_t acc_errors() const;
+    void AddRequestCount();
 
     void FeedbackCircuitBreaker(int error_code, int64_t latency_us);