msg/async: add async msg perf counter

Signed-off-by: Xinze Chi <xinze@xsky.com>
Updated to account for the new ProtocolV1/ProtocolV2

Signed-off-by: Greg Farnum <gfarnum@redhat.com>

diff --git a/src/msg/Message.h b/src/msg/Message.h
index 1ed032f501641b95d032eb4df473f1e07302796f..115356c465aedb5ad6dc4d19cfe7f8d995af1062 100644 (file)
--- a/src/msg/Message.h
+++ b/src/msg/Message.h
@@ -280,6 +280,7 @@ public:
                                   boost::intrusive::list_member_hook<>,
                                   &Message::dispatch_q>> Queue;
 
+  ceph::mono_time queue_start;
 protected:
   CompletionHook* completion_hook = nullptr; // owned by Messenger
 

diff --git a/src/msg/async/ProtocolV1.cc b/src/msg/async/ProtocolV1.cc
index a14eb337cbb06eb677baf94dca2bc8fed718ed91..8879767453684b415fcad7b3f6e6485d3d4dc624 100644 (file)
--- a/src/msg/async/ProtocolV1.cc
+++ b/src/msg/async/ProtocolV1.cc
@@ -242,6 +242,7 @@ void ProtocolV1::send_message(Message *m) {
                    << " Drop message " << m << dendl;
     m->put();
   } else {
+    m->queue_start = ceph::mono_clock::now();
     m->trace.event("async enqueueing message");
     out_q[m->get_priority()].emplace_back(std::move(bl), m);
     ldout(cct, 15) << __func__ << " inline write is denied, reschedule m=" << m
@@ -338,6 +339,11 @@ void ProtocolV1::write_event() {
         prepare_send_message(connection->get_features(), m, data);
       }
 
+      if (m->queue_start != ceph::mono_time()) {
+        connection->logger->tinc(l_msgr_send_messages_queue_lat,
+                                ceph::mono_clock::now() - m->queue_start);
+      }
+
       r = write_message(m, data, more);
 
       connection->write_lock.lock();
@@ -590,6 +596,7 @@ CtPtr ProtocolV1::handle_tag_ack(char *buffer, int r) {
   // trim sent list
   static const int max_pending = 128;
   int i = 0;
+  auto now = ceph::mono_clock::now();
   Message *pending[max_pending];
   connection->write_lock.lock();
   while (!sent.empty() && sent.front()->get_seq() <= seq && i < max_pending) {
@@ -601,6 +608,7 @@ CtPtr ProtocolV1::handle_tag_ack(char *buffer, int r) {
                    << dendl;
   }
   connection->write_lock.unlock();
+  connection->logger->tinc(l_msgr_handle_ack_lat, ceph::mono_clock::now() - now);
   for (int k = 0; k < i; k++) {
     pending[k]->put();
   }

diff --git a/src/msg/async/ProtocolV2.cc b/src/msg/async/ProtocolV2.cc
index 22e60620e46517c30fbb254db31bdea2c9a64e8d..8fb38a84cd0f530e925021be1cee5a89a228ce34 100644 (file)
--- a/src/msg/async/ProtocolV2.cc
+++ b/src/msg/async/ProtocolV2.cc
@@ -418,6 +418,7 @@ void ProtocolV2::send_message(Message *m) {
   } else {
     ldout(cct, 5) << __func__ << " enqueueing message m=" << m
                   << " type=" << m->get_type() << " " << *m << dendl;
+    m->queue_start = ceph::mono_clock::now();
     m->trace.event("async enqueueing message");
     out_queue[m->get_priority()].emplace_back(
       out_queue_entry_t{is_prepared, m});
@@ -562,6 +563,7 @@ void ProtocolV2::handle_message_ack(uint64_t seq) {
   static const int max_pending = 128;
   int i = 0;
   Message *pending[max_pending];
+  auto now = ceph::mono_clock::now();
   connection->write_lock.lock();
   while (!sent.empty() && sent.front()->get_seq() <= seq && i < max_pending) {
     Message *m = sent.front();
@@ -572,6 +574,7 @@ void ProtocolV2::handle_message_ack(uint64_t seq) {
                    << dendl;
   }
   connection->write_lock.unlock();
+  connection->logger->tinc(l_msgr_handle_ack_lat, ceph::mono_clock::now() - now);
   for (int k = 0; k < i; k++) {
     pending[k]->put();
   }
@@ -609,6 +612,12 @@ void ProtocolV2::write_event() {
         prepare_send_message(connection->get_features(), out_entry.m);
       }
 
+      if (out_entry.m->queue_start != ceph::mono_time()) {
+        connection->logger->tinc(l_msgr_send_messages_queue_lat,
+                                ceph::mono_clock::now() -
+                                out_entry.m->queue_start);
+      }
+
       r = write_message(out_entry.m, more);
 
       connection->write_lock.lock();

diff --git a/src/msg/async/Stack.h b/src/msg/async/Stack.h
index a093dadbc64a4d531d6c9448f0a6bb25aa931702..d05f934fd093aba035b43d83d1a363e347555b21 100644 (file)
--- a/src/msg/async/Stack.h
+++ b/src/msg/async/Stack.h
@@ -211,6 +211,9 @@ enum {
   l_msgr_running_recv_time,
   l_msgr_running_fast_dispatch_time,
 
+  l_msgr_send_messages_queue_lat,
+  l_msgr_handle_ack_lat,
+
   l_msgr_last,
 };
 
@@ -251,6 +254,9 @@ class Worker {
     plb.add_time(l_msgr_running_recv_time, "msgr_running_recv_time", "The total time of message receiving");
     plb.add_time(l_msgr_running_fast_dispatch_time, "msgr_running_fast_dispatch_time", "The total time of fast dispatch");
 
+    plb.add_time_avg(l_msgr_send_messages_queue_lat, "msgr_send_messages_queue_lat", "Network sent messages lat");
+    plb.add_time_avg(l_msgr_handle_ack_lat, "msgr_handle_ack_lat", "Connection handle ack lat");
+
     perf_logger = plb.create_perf_counters();
     cct->get_perfcounters_collection()->add(perf_logger);
   }