Author: aconway Date: Wed Jan 21 07:44:42 2009 New Revision: 736326 URL: http://svn.apache.org/viewvc?rev=736326&view=rev Log: Additional latency measurement points.
Modified: qpid/trunk/qpid/cpp/src/qpid/cluster/Cluster.cpp qpid/trunk/qpid/cpp/src/qpid/cluster/Connection.cpp qpid/trunk/qpid/cpp/src/qpid/cluster/OutputInterceptor.cpp qpid/trunk/qpid/cpp/src/qpid/cluster/OutputInterceptor.h qpid/trunk/qpid/cpp/src/qpid/framing/AMQFrame.h qpid/trunk/qpid/cpp/src/qpid/sys/LatencyMetric.cpp qpid/trunk/qpid/cpp/src/qpid/sys/LatencyMetric.h qpid/trunk/qpid/cpp/src/tests/.valgrind.supp qpid/trunk/qpid/cpp/src/tests/run_test Modified: qpid/trunk/qpid/cpp/src/qpid/cluster/Cluster.cpp URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/qpid/cluster/Cluster.cpp?rev=736326&r1=736325&r2=736326&view=diff ============================================================================== --- qpid/trunk/qpid/cpp/src/qpid/cluster/Cluster.cpp (original) +++ qpid/trunk/qpid/cpp/src/qpid/cluster/Cluster.cpp Wed Jan 21 07:44:42 2009 @@ -238,6 +238,7 @@ } } } + QPID_LATENCY_RECORD("decode+execute", e); } struct AddrList { Modified: qpid/trunk/qpid/cpp/src/qpid/cluster/Connection.cpp URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/qpid/cluster/Connection.cpp?rev=736326&r1=736325&r2=736326&view=diff ============================================================================== --- qpid/trunk/qpid/cpp/src/qpid/cluster/Connection.cpp (original) +++ qpid/trunk/qpid/cpp/src/qpid/cluster/Connection.cpp Wed Jan 21 07:44:42 2009 @@ -39,6 +39,7 @@ #include "qpid/framing/ConnectionCloseBody.h" #include "qpid/framing/ConnectionCloseOkBody.h" #include "qpid/log/Statement.h" +#include "qpid/sys/LatencyMetric.h" #include <boost/current_function.hpp> @@ -73,7 +74,7 @@ void Connection::init() { QPID_LOG(debug, cluster << " new connection: " << *this); - if (isLocal() && !isCatchUp()) { + if (isLocal() && !isCatchUp() && cluster.getReadMax()) { output.giveReadCredit(cluster.getReadMax()); } } @@ -137,6 +138,7 @@ // Delivered from cluster. void Connection::delivered(framing::AMQFrame& f) { QPID_LOG(trace, cluster << " RECV: " << *this << ": " << f); + QPID_LATENCY_INIT(f); assert(!catchUp); currentChannel = f.getChannel(); if (!framing::invoke(*this, *f.getBody()).wasHandled() // Connection contol. Modified: qpid/trunk/qpid/cpp/src/qpid/cluster/OutputInterceptor.cpp URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/qpid/cluster/OutputInterceptor.cpp?rev=736326&r1=736325&r2=736326&view=diff ============================================================================== --- qpid/trunk/qpid/cpp/src/qpid/cluster/OutputInterceptor.cpp (original) +++ qpid/trunk/qpid/cpp/src/qpid/cluster/OutputInterceptor.cpp Wed Jan 21 07:44:42 2009 @@ -46,6 +46,7 @@ } if (!parent.isCatchUp()) sent += f.encodedSize(); + QPID_LATENCY_RECORD("on write queue", f); } void OutputInterceptor::activateOutput() { @@ -77,6 +78,7 @@ // which tranfers frames to the codec for writing. // void OutputInterceptor::deliverDoOutput(size_t requested) { + QPID_LATENCY_RECORD("deliver do-output", *this); size_t buf = getBuffered(); if (parent.isLocal()) writeEstimate.delivered(requested, sent, buf); // Update the estimate. @@ -101,7 +103,7 @@ // Send a doOutput request if one is not already in flight. void OutputInterceptor::sendDoOutput() { if (!parent.isLocal()) return; - + QPID_LATENCY_INIT(*this); doingOutput = true; size_t request = writeEstimate.sending(getBuffered()); Modified: qpid/trunk/qpid/cpp/src/qpid/cluster/OutputInterceptor.h URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/qpid/cluster/OutputInterceptor.h?rev=736326&r1=736325&r2=736326&view=diff ============================================================================== --- qpid/trunk/qpid/cpp/src/qpid/cluster/OutputInterceptor.h (original) +++ qpid/trunk/qpid/cpp/src/qpid/cluster/OutputInterceptor.h Wed Jan 21 07:44:42 2009 @@ -25,6 +25,7 @@ #include "WriteEstimate.h" #include "qpid/sys/ConnectionOutputHandler.h" #include "qpid/broker/ConnectionFactory.h" +#include "qpid/sys/LatencyMetric.h" #include <boost/function.hpp> namespace qpid { @@ -36,7 +37,7 @@ /** * Interceptor for connection OutputHandler, manages outgoing message replication. */ -class OutputInterceptor : public sys::ConnectionOutputHandler { +class OutputInterceptor : public sys::ConnectionOutputHandler, sys::LatencyMetricTimestamp { public: OutputInterceptor(cluster::Connection& p, sys::ConnectionOutputHandler& h); Modified: qpid/trunk/qpid/cpp/src/qpid/framing/AMQFrame.h URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/qpid/framing/AMQFrame.h?rev=736326&r1=736325&r2=736326&view=diff ============================================================================== --- qpid/trunk/qpid/cpp/src/qpid/framing/AMQFrame.h (original) +++ qpid/trunk/qpid/cpp/src/qpid/framing/AMQFrame.h Wed Jan 21 07:44:42 2009 @@ -27,6 +27,7 @@ #include "AMQHeartbeatBody.h" #include "ProtocolVersion.h" #include "BodyHolder.h" +#include "qpid/sys/LatencyMetric.h" #include <boost/intrusive_ptr.hpp> #include <boost/cast.hpp> @@ -36,7 +37,7 @@ class BodyHolder; -class AMQFrame : public AMQDataBlock +class AMQFrame : public AMQDataBlock, public sys::LatencyMetricTimestamp { public: AMQFrame(boost::intrusive_ptr<BodyHolder> b=0) : body(b) { init(); } Modified: qpid/trunk/qpid/cpp/src/qpid/sys/LatencyMetric.cpp URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/qpid/sys/LatencyMetric.cpp?rev=736326&r1=736325&r2=736326&view=diff ============================================================================== --- qpid/trunk/qpid/cpp/src/qpid/sys/LatencyMetric.cpp (original) +++ qpid/trunk/qpid/cpp/src/qpid/sys/LatencyMetric.cpp Wed Jan 21 07:44:42 2009 @@ -39,7 +39,6 @@ LatencyMetric::~LatencyMetric() { report(); } void LatencyMetric::record(const LatencyMetricTimestamp& start) { - Mutex::ScopedLock l(lock); // FIXME aconway 2009-01-20: atomics? if (!start.latency_metric_timestamp) return; // Ignore 0 timestamps. if (skip) { if (++skipped < skip) return; Modified: qpid/trunk/qpid/cpp/src/qpid/sys/LatencyMetric.h URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/qpid/sys/LatencyMetric.h?rev=736326&r1=736325&r2=736326&view=diff ============================================================================== --- qpid/trunk/qpid/cpp/src/qpid/sys/LatencyMetric.h (original) +++ qpid/trunk/qpid/cpp/src/qpid/sys/LatencyMetric.h Wed Jan 21 07:44:42 2009 @@ -25,7 +25,6 @@ #ifdef QPID_LATENCY_METRIC #include "qpid/sys/IntegerTypes.h" -#include "qpid/sys/Mutex.h" namespace qpid { namespace sys { @@ -53,7 +52,6 @@ private: void report(); - Mutex lock; const char* message; int64_t ignore, count, total, skipped, skip; }; Modified: qpid/trunk/qpid/cpp/src/tests/.valgrind.supp URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/tests/.valgrind.supp?rev=736326&r1=736325&r2=736326&view=diff ============================================================================== --- qpid/trunk/qpid/cpp/src/tests/.valgrind.supp (original) +++ qpid/trunk/qpid/cpp/src/tests/.valgrind.supp Wed Jan 21 07:44:42 2009 @@ -1,4 +1,12 @@ { + Benign leak in CPG - patched version. + Memcheck:Leak + fun:* + fun:openais_service_connect + fun:cpg_initialize +} + +{ Benign error in libcpg. Memcheck:Param socketcall.sendmsg(msg.msg_iov[i]) Modified: qpid/trunk/qpid/cpp/src/tests/run_test URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/tests/run_test?rev=736326&r1=736325&r2=736326&view=diff ============================================================================== --- qpid/trunk/qpid/cpp/src/tests/run_test (original) +++ qpid/trunk/qpid/cpp/src/tests/run_test Wed Jan 21 07:44:42 2009 @@ -24,7 +24,7 @@ # Output nothing if test passes, show the output if it fails and # leave output in <test>.log for examination. # -# If qpidd.port exists run test with QPID_PORT=`cat qpidd.port` +# If qpidd.port exists and is not empty run test with QPID_PORT=`cat qpidd.port` # # If $VALGRIND if is set run under valgrind. If there are valgrind # erros show valgrind output, also leave it in <test>.valgrind for @@ -38,7 +38,7 @@ export VALGRIND srcdir # Set QPID_PORT if qpidd.port exists. -test -f qpidd.port && QPID_PORT=`cat qpidd.port` +test -s qpidd.port && QPID_PORT=`cat qpidd.port` export QPID_PORT # Avoid silly libtool error messages if these are not defined --------------------------------------------------------------------- Apache Qpid - AMQP Messaging Implementation Project: http://qpid.apache.org Use/Interact: mailto:commits-subscr...@qpid.apache.org