More latency metrics.

git-svn-id: https://svn.apache.org/repos/asf/qpid/trunk/qpid@736752 13f79535-47bb-0310-9956-ffa450edef68

5 files changed, 20 insertions, 3 deletions

diff --git a/cpp/src/qpid/cluster/Cluster.cpp b/cpp/src/qpid/cluster/Cluster.cpp
index 05a7a9efff..0a40493350 100644
--- a/cpp/src/qpid/cluster/Cluster.cpp
+++ b/cpp/src/qpid/cluster/Cluster.cpp
@@ -242,6 +242,7 @@ void Cluster::deliveredEvent(const Event& e) {
 }
 
 void Cluster::deliveredFrame(const EventFrame& e) {
+    QPID_LATENCY_RECORD("delivered frame queue", e.frame);
     QPID_LOG(trace, *this << " DLVR: " << e.frame);
     if (e.connection)   {
         e.connection->deliveredFrame(e);
@@ -252,6 +253,7 @@ void Cluster::deliveredFrame(const EventFrame& e) {
         if (!framing::invoke(dispatch, *e.frame.getBody()).wasHandled())
             throw Exception(QPID_MSG("Invalid cluster control"));
     }
+    QPID_LATENCY_RECORD("processed", e.frame);
 }
   
 struct AddrList {
diff --git a/cpp/src/qpid/cluster/EventFrame.h b/cpp/src/qpid/cluster/EventFrame.h
index a6d0ebbf0e..28ec6e4dc0 100644
--- a/cpp/src/qpid/cluster/EventFrame.h
+++ b/cpp/src/qpid/cluster/EventFrame.h
@@ -24,6 +24,7 @@
 
 #include "types.h"
 #include "qpid/framing/AMQFrame.h"
+#include "qpid/sys/LatencyMetric.h"
 #include <boost/intrusive_ptr.hpp>
 
 namespace qpid {
@@ -38,7 +39,9 @@ struct EventFrame
 {
     // Connection event frame
     EventFrame(const boost::intrusive_ptr<Connection>& c, const MemberId& m, const framing::AMQFrame& f, int rc=0)
-        : connection(c), member(m), frame(f), readCredit(rc) {}
+        : connection(c), member(m), frame(f), readCredit(rc) {
+        QPID_LATENCY_INIT(frame);
+    }
 
     bool isCluster() const { return !connection; }
     bool isConnection() const { return connection; }
diff --git a/cpp/src/qpid/cluster/OutputInterceptor.cpp b/cpp/src/qpid/cluster/OutputInterceptor.cpp
index 472fb2e6c0..dbed75beed 100644
--- a/cpp/src/qpid/cluster/OutputInterceptor.cpp
+++ b/cpp/src/qpid/cluster/OutputInterceptor.cpp
@@ -46,7 +46,7 @@ void OutputInterceptor::send(framing::AMQFrame& f) {
     }
     if (!parent.isCatchUp())
         sent += f.encodedSize();
-    QPID_LATENCY_RECORD("on write queue", f);
+    QPID_LATENCY_RECORD("up to write queue", f);
 }
 
 void OutputInterceptor::activateOutput() {
@@ -79,6 +79,7 @@ bool  OutputInterceptor::doOutput() {
 // 
 void OutputInterceptor::deliverDoOutput(size_t requested) {
     QPID_LATENCY_RECORD("deliver do-output", *this);
+    QPID_LATENCY_CLEAR(*this);
     size_t buf = getBuffered();
     if (parent.isLocal())
         writeEstimate.delivered(requested, sent, buf); // Update the estimate.
diff --git a/cpp/src/qpid/sys/LatencyMetric.cpp b/cpp/src/qpid/sys/LatencyMetric.cpp
index 6a52425706..caa221def4 100644
--- a/cpp/src/qpid/sys/LatencyMetric.cpp
+++ b/cpp/src/qpid/sys/LatencyMetric.cpp
@@ -32,6 +32,10 @@ void LatencyMetricTimestamp::initialize(const LatencyMetricTimestamp& ts) {
     const_cast<int64_t&>(ts.latency_metric_timestamp) = Duration(now());
 }
 
+void LatencyMetricTimestamp::clear(const LatencyMetricTimestamp& ts) {
+    const_cast<int64_t&>(ts.latency_metric_timestamp) = 0;
+}
+
 LatencyMetric::LatencyMetric(const char* msg, int64_t skip_) : 
     message(msg), count(0), total(0), skipped(0), skip(skip_)
 {}
diff --git a/cpp/src/qpid/sys/LatencyMetric.h b/cpp/src/qpid/sys/LatencyMetric.h
index ff679ef6a8..63b5020db4 100644
--- a/cpp/src/qpid/sys/LatencyMetric.h
+++ b/cpp/src/qpid/sys/LatencyMetric.h
@@ -33,6 +33,7 @@ namespace sys {
 struct LatencyMetricTimestamp {
     LatencyMetricTimestamp() : latency_metric_timestamp(0) {}
     static void initialize(const LatencyMetricTimestamp&);
+    static void clear(const LatencyMetricTimestamp&);
     int64_t latency_metric_timestamp;
 };
     
@@ -53,15 +54,19 @@ class LatencyMetric {
   private:
     void report();
     const char* message;
-    int64_t ignore, count, total, skipped, skip;
+    int64_t count, total, skipped, skip;
 };
 
 }} // namespace qpid::sys
 
 #define QPID_LATENCY_INIT(x) ::qpid::sys::LatencyMetricTimestamp::initialize(x)
+#define QPID_LATENCY_CLEAR(x) ::qpid::sys::LatencyMetricTimestamp::clear(x)
 #define QPID_LATENCY_RECORD(msg, x) do {                                 \
         static ::qpid::sys::LatencyMetric metric__(msg); metric__.record(x); \
     } while (false)
+#define QPID_LATENCY_RECORD_SKIP(msg, x, skip) do {                          \
+        static ::qpid::sys::LatencyMetric metric__(msg, skip); metric__.record(x); \
+    } while (false)
 
 
 #else  /* defined QPID_LATENCY_METRIC */
@@ -71,7 +76,9 @@ class LatencyMetricTimestamp {};
 }}
 
 #define QPID_LATENCY_INIT(x) (void)x
+#define QPID_LATENCY_CLEAR(x) (void)x
 #define QPID_LATENCY_RECORD(msg, x) (void)x
+#define QPID_LATENCY_RECORD_SKIP(msg, x, skip) (void)x
 
 #endif /* defined QPID_LATENCY_METRIC */