Merge pull request #538 from zyearn/add_server_processed_time

add server processed time in controller

src/brpc/controller.h

@@ -172,8 +172,15 @@ public:
     // True if a backup request was sent during the RPC.
     bool has_backup_request() const { return has_flag(FLAGS_BACKUP_REQUEST); }
 
-    // Get latency of the RPC call.
-    int64_t latency_us() const { return _end_time_us - _begin_time_us; }
+    // This function has different meanings in client and server side.
+    // In client side it gets latency of the RPC call. While in server side,
+    // it gets queue time before server processes the RPC call.
+    int64_t latency_us() const {
+        if (_end_time_us == UNSET_MAGIC_NUM) {
+            return butil::cpuwide_time_us() - _begin_time_us;
+        }
+        return _end_time_us - _begin_time_us;
+    }
 
     // Response of the RPC call (passed to CallMethod)
     google::protobuf::Message* response() const { return _response; }

src/brpc/details/controller_private_accessor.h

@@ -132,6 +132,12 @@ public:
     // side is properly set in the RPC sending path.
     void set_deadline_us(int64_t deadline_us) { _cntl->_deadline_us = deadline_us; }
 
+    ControllerPrivateAccessor& set_begin_time_us(int64_t begin_time_us) {
+        _cntl->_begin_time_us = begin_time_us;
+        _cntl->_end_time_us = UNSET_MAGIC_NUM;
+        return *this;
+    }
+
 private:
     Controller* _cntl;
 };

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_begin_time_us(msg->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_begin_time_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_begin_time_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_begin_time_us(msg->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_begin_time_us(msg->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_begin_time_us(msg->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_begin_time_us(msg_base->received_us())
         .move_in_server_receiving_sock(socket_guard);
 
     uint32_t seq_id;