add server processed time in controller

src/brpc/controller.cpp

@@ -252,6 +252,7 @@ void Controller::ResetPods() {
     _request_stream = INVALID_STREAM_ID;
     _response_stream = INVALID_STREAM_ID;
     _remote_stream_settings = NULL;
+    _received_us = 0;
 }
 
 Controller::Call::Call(Controller::Call* rhs)

src/brpc/controller.h

@@ -175,6 +175,9 @@ public:
     // Get latency of the RPC call.
     int64_t latency_us() const { return _end_time_us - _begin_time_us; }
 
+    // Get processed time of the RPC call since received
+    int64_t processed_us() const { return butil::cpuwide_time_us()- _received_us; }
+
     // Response of the RPC call (passed to CallMethod)
     google::protobuf::Message* response() const { return _response; }
 
@@ -729,6 +732,9 @@ private:
 
     // Thrift method name, only used when thrift protocol enabled
     std::string _thrift_method_name;
+
+    // Received time of message
+    int64_t _received_us;
 };
 
 // Advises the RPC system that the caller desires that the RPC call be

src/brpc/policy/baidu_rpc_protocol.cpp

@@ -351,6 +351,7 @@ void ProcessRpcRequest(InputMessageBase* msg_base) {
         .set_local_side(socket->local_side())
         .set_auth_context(socket->auth_context())
         .set_request_protocol(PROTOCOL_BAIDU_STD)
+        .set_received_us(msg_base->received_us())
         .move_in_server_receiving_sock(socket_guard);
 
     if (meta.has_stream_settings()) {

src/brpc/policy/http_rpc_protocol.cpp

@@ -1237,6 +1237,7 @@ void ProcessHttpRequest(InputMessageBase *msg) {
         .set_local_side(socket->local_side())
         .set_auth_context(socket->auth_context())
         .set_request_protocol(PROTOCOL_HTTP)
+        .set_received_us(msg->received_us())
         .move_in_server_receiving_sock(socket_guard);
     
     // Read log-id. errno may be set when input to strtoull overflows.

src/brpc/policy/hulu_pbrpc_protocol.cpp

@@ -384,6 +384,7 @@ void ProcessHuluRequest(InputMessageBase* msg_base) {
         .set_local_side(socket->local_side())
         .set_auth_context(socket->auth_context())
         .set_request_protocol(PROTOCOL_HULU_PBRPC)
+        .set_received_us(msg->received_us())
         .move_in_server_receiving_sock(socket_guard);
 
     if (meta.has_user_data()) {

src/brpc/policy/mongo_protocol.cpp

@@ -206,6 +206,7 @@ void ProcessMongoRequest(InputMessageBase* msg_base) {
         .set_local_side(socket->local_side())
         .set_auth_context(socket->auth_context())
         .set_request_protocol(PROTOCOL_MONGO)
+        .set_received_us(msg_base->received_us())
         .move_in_server_receiving_sock(socket_guard);
 
     // Tag the bthread with this server's key for

src/brpc/policy/nshead_protocol.cpp

@@ -262,6 +262,7 @@ void ProcessNsheadRequest(InputMessageBase* msg_base) {
         .set_remote_side(socket->remote_side())
         .set_local_side(socket->local_side())
         .set_request_protocol(PROTOCOL_NSHEAD)
+        .set_received_us(msg_base->received_us())
         .move_in_server_receiving_sock(socket_guard);
 
     // Tag the bthread with this server's key for thread_local_data().

src/brpc/policy/sofa_pbrpc_protocol.cpp

@@ -352,6 +352,7 @@ void ProcessSofaRequest(InputMessageBase* msg_base) {
         .set_local_side(socket->local_side())
         .set_auth_context(socket->auth_context())
         .set_request_protocol(PROTOCOL_SOFA_PBRPC)
+        .set_received_us(msg_base->received_us())
         .move_in_server_receiving_sock(socket_guard);
 
     // Tag the bthread with this server's key for thread_local_data().

src/brpc/policy/thrift_protocol.cpp

@@ -465,6 +465,7 @@ void ProcessThriftRequest(InputMessageBase* msg_base) {
         .set_remote_side(socket->remote_side())
         .set_local_side(socket->local_side())
         .set_request_protocol(PROTOCOL_THRIFT)
+        .set_received_us(msg_base->received_us())
         .move_in_server_receiving_sock(socket_guard);
 
     uint32_t seq_id;