From 650dc8949488e88797a61908723619aca5fc8909 Mon Sep 17 00:00:00 2001 From: Alan Conway Date: Fri, 5 Sep 2008 19:53:44 +0000 Subject: Fixed cluster membership notification. Cluster events with RefCountedBuffers for queueing. PollableQueue clears bacth immediately. Improved perfdist: clients hit multiple brokers in a cluster. git-svn-id: https://svn.apache.org/repos/asf/incubator/qpid/trunk@692521 13f79535-47bb-0310-9956-ffa450edef68 --- qpid/cpp/src/Makefile.am | 2 + qpid/cpp/src/cluster.mk | 4 +- qpid/cpp/src/qpid/RefCountedBuffer.cpp | 45 ++++++++++++++++++++ qpid/cpp/src/qpid/RefCountedBuffer.h | 68 +++++++++++++++++++++++++++++++ qpid/cpp/src/qpid/cluster/Cluster.cpp | 51 +++++++++++++++++------ qpid/cpp/src/qpid/cluster/Cluster.h | 1 - qpid/cpp/src/qpid/cluster/Cpg.cpp | 30 -------------- qpid/cpp/src/qpid/cluster/Cpg.h | 10 ----- qpid/cpp/src/qpid/cluster/Event.cpp | 52 +++++++++++++++++++++++ qpid/cpp/src/qpid/cluster/Event.h | 65 +++++++++++++++++++++++++++++ qpid/cpp/src/qpid/cluster/PollableQueue.h | 10 +++-- qpid/cpp/src/tests/benchmark | 52 ++++++++++++++--------- qpid/cpp/src/tests/perfdist | 38 ++++++++++------- qpid/cpp/src/tests/start_cluster_hosts | 16 ++++---- 14 files changed, 344 insertions(+), 100 deletions(-) create mode 100644 qpid/cpp/src/qpid/RefCountedBuffer.cpp create mode 100644 qpid/cpp/src/qpid/RefCountedBuffer.h create mode 100644 qpid/cpp/src/qpid/cluster/Event.cpp create mode 100644 qpid/cpp/src/qpid/cluster/Event.h (limited to 'qpid/cpp') diff --git a/qpid/cpp/src/Makefile.am b/qpid/cpp/src/Makefile.am index 25c27549c3..d52994a6e4 100644 --- a/qpid/cpp/src/Makefile.am +++ b/qpid/cpp/src/Makefile.am @@ -237,6 +237,8 @@ libqpidcommon_la_SOURCES = \ qpid/DataDir.cpp \ qpid/Options.cpp \ qpid/log/Options.cpp \ + qpid/RefCountedBuffer.h \ + qpid/RefCountedBuffer.cpp \ qpid/log/Selector.cpp \ qpid/log/Statement.cpp \ qpid/pointer_to_other.h diff --git a/qpid/cpp/src/cluster.mk b/qpid/cpp/src/cluster.mk index d5be6dfc57..cb07804550 100644 --- a/qpid/cpp/src/cluster.mk +++ b/qpid/cpp/src/cluster.mk @@ -25,7 +25,9 @@ libqpidcluster_la_SOURCES = \ qpid/cluster/WriteEstimate.cpp \ qpid/cluster/OutputInterceptor.h \ qpid/cluster/OutputInterceptor.cpp \ - qpid/cluster/ProxyInputHandler.h + qpid/cluster/ProxyInputHandler.h \ + qpid/cluster/Event.h \ + qpid/cluster/Event.cpp libqpidcluster_la_LIBADD= -lcpg libqpidbroker.la diff --git a/qpid/cpp/src/qpid/RefCountedBuffer.cpp b/qpid/cpp/src/qpid/RefCountedBuffer.cpp new file mode 100644 index 0000000000..3a52b94412 --- /dev/null +++ b/qpid/cpp/src/qpid/RefCountedBuffer.cpp @@ -0,0 +1,45 @@ +/* + * + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + * + */ + +#include "RefCountedBuffer.h" + +namespace qpid { + +RefCountedBuffer::RefCountedBuffer() : count(0) {} + +void RefCountedBuffer::destroy() const { + this->~RefCountedBuffer(); + ::delete[] reinterpret_cast(this); +} + +char* RefCountedBuffer::addr() const { + return const_cast(reinterpret_cast(this)+sizeof(RefCountedBuffer)); +} + +RefCountedBuffer::intrusive_ptr RefCountedBuffer::create(size_t n) { + char* store=::new char[n+sizeof(RefCountedBuffer)]; + new(store) RefCountedBuffer; + return reinterpret_cast(store); +} + +} // namespace qpid + + diff --git a/qpid/cpp/src/qpid/RefCountedBuffer.h b/qpid/cpp/src/qpid/RefCountedBuffer.h new file mode 100644 index 0000000000..af46cbb92a --- /dev/null +++ b/qpid/cpp/src/qpid/RefCountedBuffer.h @@ -0,0 +1,68 @@ +#ifndef QPID_REFCOUNTEDBUFFER_H +#define QPID_REFCOUNTEDBUFFER_H + +/* + * + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + * + */ + +#include +#include +#include + +namespace qpid { + +/** + * Reference-counted byte buffer. + * No alignment guarantees. + */ +class RefCountedBuffer : boost::noncopyable { + mutable boost::detail::atomic_count count; + RefCountedBuffer(); + void destroy() const; + char* addr() const; + +public: + + typedef boost::intrusive_ptr intrusive_ptr; + + /** Create a reference counted buffer of size n */ + static intrusive_ptr create(size_t n); + + /** Get a pointer to the start of the buffer. */ + char* get() { return addr(); } + const char* get() const { return addr(); } + char& operator[](size_t i) { return get()[i]; } + const char& operator[](size_t i) const { return get()[i]; } + + void addRef() const { ++count; } + void release() const { if (--count==0) destroy(); } + long refCount() { return count; } +}; + +} // namespace qpid + +// intrusive_ptr support. +namespace boost { +inline void intrusive_ptr_add_ref(const qpid::RefCountedBuffer* p) { p->addRef(); } +inline void intrusive_ptr_release(const qpid::RefCountedBuffer* p) { p->release(); } +} + + +#endif /*!QPID_REFCOUNTEDBUFFER_H*/ diff --git a/qpid/cpp/src/qpid/cluster/Cluster.cpp b/qpid/cpp/src/qpid/cluster/Cluster.cpp index f93203acbf..4d54a837ca 100644 --- a/qpid/cpp/src/qpid/cluster/Cluster.cpp +++ b/qpid/cpp/src/qpid/cluster/Cluster.cpp @@ -38,6 +38,7 @@ #include #include #include +#include namespace qpid { namespace cluster { @@ -67,11 +68,8 @@ Cluster::Cluster(const std::string& name_, const Url& url_, broker::Broker& b) : ) { broker->addFinalizer(boost::bind(&Cluster::leave, this)); - QPID_LOG(trace, "Joining cluster: " << name << " as " << self); + QPID_LOG(notice, "Joining cluster: " << name.str() << " as " << self); cpg.join(name); - mcastFrame(AMQFrame(in_place(ProtocolVersion(), url.str())), - ConnectionId(self,0)); - // Start dispatching from the poller. cpgDispatchHandle.startWatch(poller); } @@ -94,7 +92,7 @@ void Cluster::leave() { // Leave is called by from Broker destructor after the poller has // been shut down. No dispatches can occur. - QPID_LOG(debug, "Leaving cluster " << name.str()); + QPID_LOG(notice, "Leaving cluster " << name.str()); cpg.leave(name); // broker= is set to 0 when the final config-change is delivered. while(broker) { @@ -158,7 +156,7 @@ boost::intrusive_ptr Cluster::getConnection(const ConnectionId& id) if (i == connections.end()) { // New shadow connection. assert(id.getMember() != self); std::ostringstream mgmtId; - mgmtId << name << ":" << id; + mgmtId << name.str() << ":" << id; ConnectionMap::value_type value(id, new Connection(*this, shadowOut, mgmtId.str(), id)); i = connections.insert(value).first; } @@ -205,22 +203,50 @@ void Cluster::deliver( } } +struct AddrList { + const cpg_address* addrs; + int count; + AddrList(const cpg_address* a, int n) : addrs(a), count(n) {} +}; + +ostream& operator<<(ostream& o, const AddrList& a) { + for (const cpg_address* p = a.addrs; p < a.addrs+a.count; ++p) { + const char* reasonString; + switch (p->reason) { + case CPG_REASON_JOIN: reasonString = " joined "; break; + case CPG_REASON_LEAVE: reasonString = " left ";break; + case CPG_REASON_NODEDOWN: reasonString = " node-down ";break; + case CPG_REASON_NODEUP: reasonString = " node-up ";break; + case CPG_REASON_PROCDOWN: reasonString = " process-down ";break; + default: reasonString = " "; + } + qpid::cluster::MemberId member(*p); + o << member << reasonString; + } + return o; +} + void Cluster::configChange( cpg_handle_t /*handle*/, cpg_name */*group*/, cpg_address *current, int nCurrent, cpg_address *left, int nLeft, - cpg_address */*joined*/, int /*nJoined*/) + cpg_address *joined, int nJoined) { - QPID_LOG(debug, "Cluster change: " - << std::make_pair(current, nCurrent) - << std::make_pair(left, nLeft)); + QPID_LOG(notice, "Cluster of " << nCurrent << ": " << AddrList(current, nCurrent) << ".\n Changes: " + << AddrList(joined, nJoined) << AddrList(left, nLeft)); + + if (nJoined) // Notfiy new members of my URL. + mcastFrame( + AMQFrame(in_place(ProtocolVersion(), url.str())), + ConnectionId(self,0)); + Mutex::ScopedLock l(lock); for (int i = 0; i < nLeft; ++i) urls.erase(left[i]); // Add new members when their URL notice arraives. - if (std::find(left, left+nLeft, self) != left+nLeft) + if (find(left, left+nLeft, self) != left+nLeft) broker = 0; // We have left the group, this is the final config change. lock.notifyAll(); // Threads waiting for membership changes. } @@ -236,7 +262,8 @@ void Cluster::disconnect(sys::DispatchHandle& h) { broker->shutdown(); } -void Cluster::urlNotice(const MemberId& m, const std::string& url) { +void Cluster::urlNotice(const MemberId& m, const string& url) { + QPID_LOG(notice, "Cluster member " << m << " has URL " << url); urls.insert(UrlMap::value_type(m,Url(url))); } diff --git a/qpid/cpp/src/qpid/cluster/Cluster.h b/qpid/cpp/src/qpid/cluster/Cluster.h index 4963400e10..630de97093 100644 --- a/qpid/cpp/src/qpid/cluster/Cluster.h +++ b/qpid/cpp/src/qpid/cluster/Cluster.h @@ -19,7 +19,6 @@ * */ -#include "qpid/cluster/types.h" #include "qpid/cluster/Cpg.h" #include "qpid/cluster/PollableQueue.h" #include "qpid/cluster/NoOpConnectionOutputHandler.h" diff --git a/qpid/cpp/src/qpid/cluster/Cpg.cpp b/qpid/cpp/src/qpid/cluster/Cpg.cpp index 754b4abd58..96a5b3da43 100644 --- a/qpid/cpp/src/qpid/cluster/Cpg.cpp +++ b/qpid/cpp/src/qpid/cluster/Cpg.cpp @@ -184,33 +184,3 @@ ostream& operator<<(ostream& o, const ConnectionId& c) { } }} // namespace qpid::cluster - - -// In proper namespace for ADL. - -std::ostream& operator<<(std::ostream& o, const ::cpg_address& a) { - const char* reasonString; - switch (a.reason) { - case CPG_REASON_JOIN: reasonString = " joined"; break; - case CPG_REASON_LEAVE: reasonString = " left";break; - case CPG_REASON_NODEDOWN: reasonString = " node-down";break; - case CPG_REASON_NODEUP: reasonString = " node-up";break; - case CPG_REASON_PROCDOWN: reasonString = " process-down";break; - default: reasonString = ""; - } - return o << qpid::cluster::MemberId(a.nodeid, a.pid) << reasonString; -} - -std::ostream& operator<<(std::ostream& o, const cpg_name& name) { - return o << std::string(name.value, name.length); -} - -namespace std { -ostream& operator<<(ostream& o, std::pair a) { - for (cpg_address* p = a.first; p < a.first+a.second; ++p) - o << *p << " "; - return o; -} -} - - diff --git a/qpid/cpp/src/qpid/cluster/Cpg.h b/qpid/cpp/src/qpid/cluster/Cpg.h index fdc451fbbc..5ffd42e12a 100644 --- a/qpid/cpp/src/qpid/cluster/Cpg.h +++ b/qpid/cpp/src/qpid/cluster/Cpg.h @@ -158,8 +158,6 @@ class Cpg : public sys::IOHandle { bool isShutdown; }; -std::ostream& operator <<(std::ostream& out, const MemberId& id); - inline bool operator==(const cpg_name& a, const cpg_name& b) { return a.length==b.length && strncmp(a.value, b.value, a.length) == 0; } @@ -167,12 +165,4 @@ inline bool operator!=(const cpg_name& a, const cpg_name& b) { return !(a == b); }} // namespace qpid::cluster -// In proper namespaces for ADL -std::ostream& operator <<(std::ostream& out, const cpg_name& name); -std::ostream& operator<<(std::ostream& o, const cpg_address& a); -namespace std { -std::ostream& operator <<(std::ostream& out, std::pair addresses); -} - - #endif /*!CPG_H*/ diff --git a/qpid/cpp/src/qpid/cluster/Event.cpp b/qpid/cpp/src/qpid/cluster/Event.cpp new file mode 100644 index 0000000000..ff558842e4 --- /dev/null +++ b/qpid/cpp/src/qpid/cluster/Event.cpp @@ -0,0 +1,52 @@ +/* + * + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + * + */ +#include "types.h" +#include "Event.h" +#include "Cpg.h" +#include "qpid/framing/Buffer.h" + +namespace qpid { +namespace cluster { +using framing::Buffer; + +const size_t Event::OVERHEAD = 1 /*type*/ + 8 /*64-bit pointr*/; + +Event::Event(EventType t, const ConnectionId c, const size_t s) + : type(t), connection(c), size(s), data(RefCountedBuffer::create(s)) {} + +Event::Event(const MemberId& m, const char* d, size_t s) + : connection(m, 0), size(s-OVERHEAD), data(RefCountedBuffer::create(size)) +{ + memcpy(data->get(), d, s); +} + +void Event::mcast(const Cpg::Name& name, Cpg& cpg) { + char header[OVERHEAD]; + Buffer b; + b.putOctet(type); + b.putLongLong(reinterpret_cast(connection.getConnectionPtr())); + iovec iov[] = { { header, b.getPosition() }, { data.get(), size } }; + cpg.mcast(name, iov, sizeof(iov)/sizeof(*iov)); +} + + + +}} // namespace qpid::cluster diff --git a/qpid/cpp/src/qpid/cluster/Event.h b/qpid/cpp/src/qpid/cluster/Event.h new file mode 100644 index 0000000000..3e4a19f7f3 --- /dev/null +++ b/qpid/cpp/src/qpid/cluster/Event.h @@ -0,0 +1,65 @@ +#ifndef QPID_CLUSTER_EVENT_H +#define QPID_CLUSTER_EVENT_H + +/* + * + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + * + */ + +#include "types.h" +#include "Cpg.h" +#include "qpid/RefCountedBuffer.h" + +namespace qpid { +namespace cluster { + +// TODO aconway 2008-09-03: more efficient solution for shared +// byte-stream data. +// + +/** + * Events are sent to/received from the cluster. + * Refcounted so they can be stored on queues. + */ +struct Event { + public: + /** Create an event with for mcasting, with size bytes of space. */ + Event(EventType t, const ConnectionId c, size_t size); + + /** Create an event from delivered data. */ + Event(const MemberId& m, const char* data, size_t size); + + void mcast(const Cpg::Name& name, Cpg& cpg); + + EventType getType() const { return type; } + ConnectionId getConnection() const { return connection; } + size_t getSize() const { return size; } + char* getData() { return data->get(); } + + private: + static const size_t OVERHEAD; + EventType type; + ConnectionId connection; + size_t size; + RefCountedBuffer::intrusive_ptr data; +}; + +}} // namespace qpid::cluster + +#endif /*!QPID_CLUSTER_EVENT_H*/ diff --git a/qpid/cpp/src/qpid/cluster/PollableQueue.h b/qpid/cpp/src/qpid/cluster/PollableQueue.h index 29891da344..74da2df750 100644 --- a/qpid/cpp/src/qpid/cluster/PollableQueue.h +++ b/qpid/cpp/src/qpid/cluster/PollableQueue.h @@ -89,9 +89,13 @@ template void PollableQueue::dispatch(sys::DispatchHandle& h) { batch.clear(); batch.swap(queue); condition.clear(); - ScopedUnlock u(lock); - callback(batch.begin(), batch.end()); // Process outside the lock to allow concurrent push. - h.rewatch(); + { + // Process outside the lock to allow concurrent push. + ScopedUnlock u(lock); + callback(batch.begin(), batch.end()); + h.rewatch(); + } + batch.clear(); } }} // namespace qpid::cluster diff --git a/qpid/cpp/src/tests/benchmark b/qpid/cpp/src/tests/benchmark index 9e8023043b..5381fd69fa 100755 --- a/qpid/cpp/src/tests/benchmark +++ b/qpid/cpp/src/tests/benchmark @@ -21,33 +21,45 @@ # and latency against a single cluster member while they are replicating. # # Must be run in the qpid src/tests build directory. -# +# +usage() { +cat < -- +usage: $0 -- [ --- ] +Client & broker hosts can also be set in env vars CLIENTS and BROKERS. -Run perftest with clients running on the listed hosts. Clients are -assigned to hosts publishers first, then subscribers the host list is -used round-robin if there are more clients than hosts. perftest-args should -include a --host flag (and --port if necessary). +Run perftest with clients running on the clients and brokers running +on the specified hosts. Clients are assigned to client hosts round +robin: publishers first, then subscribers. If there are multiple +brokers (for cluster tests) clients connect to them round robin. -Do not pass preftest action flags: --setup, --control, --publish, --subscribe. -The script will pass them to the appropriate client processes. - -Note all perftest args must come before --. +Broker hosts can be listed with -b in perftest-args or after --- +at the end of the arguments. Error: $* EOF @@ -36,19 +35,28 @@ while test $# -gt 0; do --npubs) collect $1 $2; NPUBS=$2; shift 2 ;; --nsubs) collect $1 $2; NSUBS=$2; shift 2 ;; -s|--summary) collect $1; QUIET=yes; shift 1 ;; - --) COLLECT=HOSTS; shift ;; + -b|--broker) BROKERS="$BROKERS $2"; shift 2;; + --) COLLECT=CLIENTARG; shift ;; + ---) COLLECT=BROKERARG; shift;; *) collect $1; shift ;; esac done -if [ -z "$HOSTS" ]; then usage "No hosts listed after --"; fi +CLIENTS=${CLIENTARG:-$CLIENTS} +if [ -z "$CLIENTS" ]; then usage "No client hosts listed after --"; fi +BROKERS=${BROKERARG:-$BROKERS} +if [ -z "$BROKERS" ]; then usage "No brokers specified"; fi + PERFTEST="$TESTDIR/perftest $ARGS" -HOSTS=($HOSTS) +CLIENTS=($CLIENTS) +BROKERS=($BROKERS) start() { - HOST=${HOSTS[i % ${#HOSTS[*]}]} - test -z "$QUIET" && echo "Client $i on $HOST $*" - ssh -fT $HOST "PATH=$ADDPATH:\$PATH" $PERFTEST "$@" + CLIENT=${CLIENTS[i % ${#CLIENTS[*]}]} + BROKER=${BROKERS[i % ${#BROKERS[*]}]} + ARGS="$* --broker $BROKER" + test -z "$QUIET" && echo "Client $i on $CLIENT: $ARGS" + ssh -fT $CLIENT $PERFTEST "$@" } $PERFTEST --setup diff --git a/qpid/cpp/src/tests/start_cluster_hosts b/qpid/cpp/src/tests/start_cluster_hosts index 37dda882ca..683798453b 100755 --- a/qpid/cpp/src/tests/start_cluster_hosts +++ b/qpid/cpp/src/tests/start_cluster_hosts @@ -16,14 +16,14 @@ QPIDD=${QPIDD:-$PWD/../qpidd} LIBQPIDCLUSTER=${LIBQPIDCLUSTER:-$PWD/../.libs/libqpidcluster.so} -CLUSTER=$USER # User name is default cluster name. +NAME=$USER # User name is default cluster name. RESTART=NO -while getopts "kp:c:q:r" ARG ; do +while getopts "kp:n:q:r" ARG ; do case $ARG in k) KILL=yes ;; p) PORT="$OPTARG" ;; - c) CLUSTER=$OPTARG ;; + n) NAME=$OPTARG ;; q) QPIDD=$OPTARG ;; l) LIBQPIDCLUSTER=$OPTARG ;; r) RESTART=yes ;; @@ -33,17 +33,17 @@ done shift `expr $OPTIND - 1` test -n "$PORT" && PORTOPT="-p $PORT" test "$KILL" = yes && KILL="$QPIDD -q $PORTOPT ;" -test -z "$*" && { echo Must specify at least one host; exit 1; } +CLUSTER=${*:-$CLUSTER} # Use args or env +test -z "$CLUSTER" && { echo Must specify at least one host; exit 1; } -OPTS="-d $PORTOPT --load-module $LIBQPIDCLUSTER --cluster-name=$CLUSTER --no-data-dir --auth=no --log-output=syslog" +OPTS="-d $PORTOPT --load-module $LIBQPIDCLUSTER --cluster-name=$NAME --no-data-dir --auth=no --log-output=syslog --log-enable=info+" num=0 -for h in $*; do +for h in $CLUSTER; do num=`expr $num + 1` # Give a unique log prefix to each node. cmd="$KILL $QPIDD $OPTS --log-prefix $num.$h" - echo == $h - out=`echo "$cmd" | ssh $h newgrp ais` || { echo $out ; exit 1; } + out=`echo "$cmd" | ssh $h newgrp ais` || { echo == $h error: $out ; exit 1; } if [ "$PORT" = 0 ] ; then p=$out; else p=$PORT; fi echo "$h $p" done -- cgit v1.2.1