Record latency of EventDispatcher (#2897)

* Record latency of EventDispatcher

* Delete global bvar

Author: chenBright

src/brpc/event_dispatcher.cpp

@@ -21,9 +21,9 @@
 #include "butil/fd_utility.h"                         // make_close_on_exec
 #include "butil/logging.h"                            // LOG
 #include "butil/third_party/murmurhash3/murmurhash3.h"// fmix32
+#include "bvar/latency_recorder.h"                    // bvar::LatencyRecorder
 #include "bthread/bthread.h"                          // bthread_start_background
 #include "brpc/event_dispatcher.h"
-#include "brpc/reloadable_flags.h"
 
 DECLARE_int32(task_group_ntags);
 
@@ -37,6 +37,8 @@ DEFINE_bool(usercode_in_coroutine, false,
             "User's callback are run in coroutine, no bthread or pthread blocking call");
 
 static EventDispatcher* g_edisp = NULL;
+static bvar::LatencyRecorder* g_edisp_read_lantency = NULL;
+static bvar::LatencyRecorder* g_edisp_write_lantency = NULL;
 static pthread_once_t g_edisp_once = PTHREAD_ONCE_INIT;
 
 static void StopAndJoinGlobalDispatchers() {
@@ -46,8 +48,13 @@ static void StopAndJoinGlobalDispatchers() {
             g_edisp[i * FLAGS_event_dispatcher_num + j].Join();
         }
     }
+    delete g_edisp_read_lantency;
+    delete g_edisp_write_lantency;
 }
 void InitializeGlobalDispatchers() {
+    g_edisp_read_lantency = new bvar::LatencyRecorder("event_dispatcher_read_latency");
+    g_edisp_write_lantency = new bvar::LatencyRecorder("event_dispatcher_write_latency");
+
     g_edisp = new EventDispatcher[FLAGS_task_group_ntags * FLAGS_event_dispatcher_num];
     for (int i = 0; i < FLAGS_task_group_ntags; ++i) {
         for (int j = 0; j < FLAGS_event_dispatcher_num; ++j) {

src/brpc/event_dispatcher_epoll.cpp

@@ -222,14 +222,18 @@ void EventDispatcher::Run() {
                 || (e[i].events & has_epollrdhup)
 #endif
                 ) {
+                int64_t start_ns = butil::cpuwide_time_ns();
                 // We don't care about the return value.
                 CallInputEventCallback(e[i].data.u64, e[i].events, _thread_attr);
+                (*g_edisp_read_lantency) << (butil::cpuwide_time_ns() - start_ns);
             }
         }
         for (int i = 0; i < n; ++i) {
             if (e[i].events & (EPOLLOUT | EPOLLERR | EPOLLHUP)) {
+                int64_t start_ns = butil::cpuwide_time_ns();
                 // We don't care about the return value.
                 CallOutputEventCallback(e[i].data.u64, e[i].events, _thread_attr);
+                (*g_edisp_write_lantency) << (butil::cpuwide_time_ns() - start_ns);
             }
         }
     }

src/brpc/event_dispatcher_kqueue.cpp

@@ -205,16 +205,20 @@ void EventDispatcher::Run() {
         }
         for (int i = 0; i < n; ++i) {
             if ((e[i].flags & EV_ERROR) || e[i].filter == EVFILT_READ) {
+                int64_t start_ns = butil::cpuwide_time_ns();
                 // We don't care about the return value.
                 CallInputEventCallback((IOEventDataId)e[i].udata,
                                        e[i].filter, _thread_attr);
+                (*g_edisp_read_lantency) << (butil::cpuwide_time_ns() - start_ns);
             }
         }
         for (int i = 0; i < n; ++i) {
             if ((e[i].flags & EV_ERROR) || e[i].filter == EVFILT_WRITE) {
+                int64_t start_ns = butil::cpuwide_time_ns();
                 // We don't care about the return value.
                 CallOutputEventCallback((IOEventDataId)e[i].udata,
                                         e[i].filter, _thread_attr);
+                (*g_edisp_write_lantency) << (butil::cpuwide_time_ns() - start_ns);
             }
         }
     }