summaryrefslogtreecommitdiff
path: root/cpp/src/qpid/cluster
diff options
context:
space:
mode:
Diffstat (limited to 'cpp/src/qpid/cluster')
-rw-r--r--cpp/src/qpid/cluster/Cluster.cpp2
-rw-r--r--cpp/src/qpid/cluster/EventFrame.h5
-rw-r--r--cpp/src/qpid/cluster/OutputInterceptor.cpp3
3 files changed, 8 insertions, 2 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.