You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@qpid.apache.org by ac...@apache.org on 2006/12/20 18:54:07 UTC
svn commit: r489159 [1/2] - in
/incubator/qpid/branches/event-queue-2006-12-20/cpp: ./ lib/broker/
lib/client/ lib/common/ lib/common/framing/ lib/common/sys/
lib/common/sys/apr/ lib/common/sys/posix/ tests/
Author: aconway
Date: Wed Dec 20 09:54:05 2006
New Revision: 489159
URL: http://svn.apache.org/viewvc?view=rev&rev=489159
Log:
Current state of event channel code.
Modified:
incubator/qpid/branches/event-queue-2006-12-20/cpp/configure.ac
incubator/qpid/branches/event-queue-2006-12-20/cpp/lib/broker/AutoDelete.cpp
incubator/qpid/branches/event-queue-2006-12-20/cpp/lib/broker/Broker.cpp
incubator/qpid/branches/event-queue-2006-12-20/cpp/lib/broker/BrokerChannel.cpp
incubator/qpid/branches/event-queue-2006-12-20/cpp/lib/broker/BrokerChannel.h
incubator/qpid/branches/event-queue-2006-12-20/cpp/lib/broker/BrokerMessage.cpp
incubator/qpid/branches/event-queue-2006-12-20/cpp/lib/broker/BrokerQueue.cpp
incubator/qpid/branches/event-queue-2006-12-20/cpp/lib/broker/TopicExchange.cpp
incubator/qpid/branches/event-queue-2006-12-20/cpp/lib/client/ResponseHandler.cpp
incubator/qpid/branches/event-queue-2006-12-20/cpp/lib/common/Exception.cpp
incubator/qpid/branches/event-queue-2006-12-20/cpp/lib/common/Exception.h
incubator/qpid/branches/event-queue-2006-12-20/cpp/lib/common/ExceptionHolder.cpp
incubator/qpid/branches/event-queue-2006-12-20/cpp/lib/common/ExceptionHolder.h
incubator/qpid/branches/event-queue-2006-12-20/cpp/lib/common/Makefile.am
incubator/qpid/branches/event-queue-2006-12-20/cpp/lib/common/QpidError.cpp
incubator/qpid/branches/event-queue-2006-12-20/cpp/lib/common/QpidError.h
incubator/qpid/branches/event-queue-2006-12-20/cpp/lib/common/framing/AMQFrame.h
incubator/qpid/branches/event-queue-2006-12-20/cpp/lib/common/sys/Acceptor.h
incubator/qpid/branches/event-queue-2006-12-20/cpp/lib/common/sys/AtomicCount.h
incubator/qpid/branches/event-queue-2006-12-20/cpp/lib/common/sys/Runnable.cpp
incubator/qpid/branches/event-queue-2006-12-20/cpp/lib/common/sys/Runnable.h
incubator/qpid/branches/event-queue-2006-12-20/cpp/lib/common/sys/Socket.h
incubator/qpid/branches/event-queue-2006-12-20/cpp/lib/common/sys/Thread.h
incubator/qpid/branches/event-queue-2006-12-20/cpp/lib/common/sys/Time.h
incubator/qpid/branches/event-queue-2006-12-20/cpp/lib/common/sys/apr/APRAcceptor.cpp
incubator/qpid/branches/event-queue-2006-12-20/cpp/lib/common/sys/apr/LFProcessor.cpp
incubator/qpid/branches/event-queue-2006-12-20/cpp/lib/common/sys/posix/EventChannel.cpp
incubator/qpid/branches/event-queue-2006-12-20/cpp/lib/common/sys/posix/EventChannel.h
incubator/qpid/branches/event-queue-2006-12-20/cpp/lib/common/sys/posix/EventChannelAcceptor.cpp
incubator/qpid/branches/event-queue-2006-12-20/cpp/lib/common/sys/posix/EventChannelThreads.cpp
incubator/qpid/branches/event-queue-2006-12-20/cpp/lib/common/sys/posix/EventChannelThreads.h
incubator/qpid/branches/event-queue-2006-12-20/cpp/lib/common/sys/posix/PosixAcceptor.cpp
incubator/qpid/branches/event-queue-2006-12-20/cpp/lib/common/sys/posix/Socket.cpp
incubator/qpid/branches/event-queue-2006-12-20/cpp/lib/common/sys/posix/check.cpp
incubator/qpid/branches/event-queue-2006-12-20/cpp/lib/common/sys/posix/check.h
incubator/qpid/branches/event-queue-2006-12-20/cpp/tests/EventChannelTest.cpp
incubator/qpid/branches/event-queue-2006-12-20/cpp/tests/EventChannelThreadsTest.cpp
incubator/qpid/branches/event-queue-2006-12-20/cpp/tests/Makefile.am
incubator/qpid/branches/event-queue-2006-12-20/cpp/tests/run-python-tests
incubator/qpid/branches/event-queue-2006-12-20/cpp/tests/topic_publisher.cpp
incubator/qpid/branches/event-queue-2006-12-20/cpp/tests/topictest
Modified: incubator/qpid/branches/event-queue-2006-12-20/cpp/configure.ac
URL: http://svn.apache.org/viewvc/incubator/qpid/branches/event-queue-2006-12-20/cpp/configure.ac?view=diff&rev=489159&r1=489158&r2=489159
==============================================================================
--- incubator/qpid/branches/event-queue-2006-12-20/cpp/configure.ac (original)
+++ incubator/qpid/branches/event-queue-2006-12-20/cpp/configure.ac Wed Dec 20 09:54:05 2006
@@ -25,14 +25,6 @@
AM_MISSING_PROG([HELP2MAN], [help2man])
-AC_ARG_ENABLE(warnings,
-[ --enable-warnings turn on lots of compiler warnings (recommended)],
-[case "${enableval}" in
- yes|no) ;;
- *) AC_MSG_ERROR([bad value ${enableval} for warnings option]) ;;
- esac],
- [enableval=yes])
-
# Turn on this automake conditional if we are in a qpid
# hierarchy (i.e. with gentools/ and specs/ sibling directories),
# and if we have working java + javac.
@@ -53,7 +45,12 @@
# -Wunreachable-code -Wpadded -Winline
# -Wshadow - warns about boost headers.
-if test "${enableval}" = yes; then
+AC_ARG_ENABLE(warnings,
+ [AS_HELP_STRING([--disable-warnings],
+ [Disable compiler warnings (default enabled)])],
+ [], [enable_warnings=yes])
+
+if test "${enable_warnings}" = yes; then
gl_COMPILER_FLAGS(-Werror)
gl_COMPILER_FLAGS(-pedantic)
gl_COMPILER_FLAGS(-Wall)
@@ -106,7 +103,7 @@
yes|no) ;;
*) AC_MSG_ERROR([invalid APR enable/disable value: $enable_APR]) ;;
esac],
-[enable_APR=yes])
+[enable_APR=no])
APR_MINIMUM_VERSION=1.2.2
AC_SUBST(APR_MINIMUM_VERSION)
Modified: incubator/qpid/branches/event-queue-2006-12-20/cpp/lib/broker/AutoDelete.cpp
URL: http://svn.apache.org/viewvc/incubator/qpid/branches/event-queue-2006-12-20/cpp/lib/broker/AutoDelete.cpp?view=diff&rev=489159&r1=489158&r2=489159
==============================================================================
--- incubator/qpid/branches/event-queue-2006-12-20/cpp/lib/broker/AutoDelete.cpp (original)
+++ incubator/qpid/branches/event-queue-2006-12-20/cpp/lib/broker/AutoDelete.cpp Wed Dec 20 09:54:05 2006
@@ -63,7 +63,7 @@
Monitor::ScopedLock l(monitor);
while(!stopped){
process();
- monitor.wait(period*TIME_MSEC);
+ monitor.wait(now() + period*TIME_MSEC);
}
}
Modified: incubator/qpid/branches/event-queue-2006-12-20/cpp/lib/broker/Broker.cpp
URL: http://svn.apache.org/viewvc/incubator/qpid/branches/event-queue-2006-12-20/cpp/lib/broker/Broker.cpp?view=diff&rev=489159&r1=489158&r2=489159
==============================================================================
--- incubator/qpid/branches/event-queue-2006-12-20/cpp/lib/broker/Broker.cpp (original)
+++ incubator/qpid/branches/event-queue-2006-12-20/cpp/lib/broker/Broker.cpp Wed Dec 20 09:54:05 2006
@@ -47,7 +47,7 @@
}
void Broker::run() {
- acceptor->run(&factory);
+ acceptor->run(factory);
}
void Broker::shutdown() {
Modified: incubator/qpid/branches/event-queue-2006-12-20/cpp/lib/broker/BrokerChannel.cpp
URL: http://svn.apache.org/viewvc/incubator/qpid/branches/event-queue-2006-12-20/cpp/lib/broker/BrokerChannel.cpp?view=diff&rev=489159&r1=489158&r2=489159
==============================================================================
--- incubator/qpid/branches/event-queue-2006-12-20/cpp/lib/broker/BrokerChannel.cpp (original)
+++ incubator/qpid/branches/event-queue-2006-12-20/cpp/lib/broker/BrokerChannel.cpp Wed Dec 20 09:54:05 2006
@@ -18,11 +18,15 @@
* under the License.
*
*/
-#include <BrokerChannel.h>
-#include <QpidError.h>
+#include <assert.h>
+
#include <iostream>
#include <sstream>
-#include <assert.h>
+
+#include <boost/bind.hpp>
+
+#include "BrokerChannel.h"
+#include "QpidError.h"
using std::mem_fun_ref;
using std::bind2nd;
@@ -50,11 +54,18 @@
}
bool Channel::exists(const string& consumerTag){
+ Mutex::ScopedLock l(lock);
return consumers.find(consumerTag) != consumers.end();
}
-void Channel::consume(string& tag, Queue::shared_ptr queue, bool acks, bool exclusive, ConnectionToken* const connection, const FieldTable*){
- if(tag.empty()) tag = tagGenerator.generate();
+void Channel::consume(
+ string& tag, Queue::shared_ptr queue, bool acks,
+ bool exclusive, ConnectionToken* const connection, const FieldTable*)
+{
+ Mutex::ScopedLock l(lock);
+ if(tag.empty()) tag = tagGenerator.generate();
+ // TODO aconway 2006-12-13: enforce ownership of consumer
+ // with auto_ptr.
ConsumerImpl* c(new ConsumerImpl(this, tag, queue, connection, acks));
try{
queue->consume(c, exclusive);//may throw exception
@@ -65,7 +76,8 @@
}
}
-void Channel::cancel(consumer_iterator i){
+void Channel::cancel(consumer_iterator i) {
+ // Private, must be called with lock held.
ConsumerImpl* c = i->second;
consumers.erase(i);
if(c){
@@ -75,6 +87,7 @@
}
void Channel::cancel(const string& tag){
+ Mutex::ScopedLock l(lock);
consumer_iterator i = consumers.find(tag);
if(i != consumers.end()){
cancel(i);
@@ -82,11 +95,14 @@
}
void Channel::close(){
- //cancel all consumers
- for(consumer_iterator i = consumers.begin(); i != consumers.end(); i = consumers.begin() ){
- cancel(i);
+ {
+ Mutex::ScopedLock l(lock);
+ while(!consumers.empty()) {
+ cancel(consumers.begin());
+ }
}
- //requeue:
+ // TODO aconway 2006-12-13: does recovery need to be atomic with
+ // cancelling all consumers?
recover(true);
}
@@ -109,20 +125,21 @@
}
void Channel::deliver(Message::shared_ptr& msg, const string& consumerTag, Queue::shared_ptr& queue, bool ackExpected){
- Mutex::ScopedLock locker(deliveryLock);
-
- u_int64_t deliveryTag = currentDeliveryTag++;
- if(ackExpected){
- unacked.push_back(DeliveryRecord(msg, queue, consumerTag, deliveryTag));
- outstanding.size += msg->contentSize();
- outstanding.count++;
+ u_int64_t deliveryTag;
+ {
+ Mutex::ScopedLock l(lock);
+ deliveryTag = currentDeliveryTag++;
+ if(ackExpected){
+ unacked.push_back(
+ DeliveryRecord(msg, queue, consumerTag, deliveryTag));
+ outstanding.size += msg->contentSize();
+ outstanding.count++;
+ }
}
- //send deliver method, header and content(s)
msg->deliver(out, id, consumerTag, deliveryTag, framesize);
}
bool Channel::checkPrefetch(Message::shared_ptr& msg){
- Mutex::ScopedLock locker(deliveryLock);
bool countOk = !prefetchCount || prefetchCount > unacked.size();
bool sizeOk = !prefetchSize || prefetchSize > msg->contentSize() + outstanding.size || unacked.empty();
return countOk && sizeOk;
@@ -187,71 +204,99 @@
}
exchange.reset();
}else{
- std::cout << "Exchange not known in Channel::complete(Message::shared_ptr&)" << std::endl;
+ std::cout << "Exchange not known in" << BOOST_CURRENT_FUNCTION
+ << std::endl;
}
}
-void Channel::ack(u_int64_t deliveryTag, bool multiple){
+void Channel::ack(u_int64_t deliveryTag, bool multiple) {
if(transactional){
+ Mutex::ScopedLock locker(lock);
accumulatedAck.update(deliveryTag, multiple);
- //TODO: I think the outstanding prefetch size & count should be updated at this point...
+ //TODO: I think the outstanding prefetch size & count should
+ //be updated at this point...
//TODO: ...this may then necessitate dispatching to consumers
- }else{
- Mutex::ScopedLock locker(deliveryLock);//need to synchronize with possible concurrent delivery
-
- ack_iterator i = find_if(unacked.begin(), unacked.end(), bind2nd(mem_fun_ref(&DeliveryRecord::matches), deliveryTag));
- if(i == unacked.end()){
- throw InvalidAckException();
- }else if(multiple){
- ack_iterator end = ++i;
- for_each(unacked.begin(), end, mem_fun_ref(&DeliveryRecord::discard));
- unacked.erase(unacked.begin(), end);
-
- //recalculate the prefetch:
- outstanding.reset();
- for_each(unacked.begin(), unacked.end(), bind2nd(mem_fun_ref(&DeliveryRecord::addTo), &outstanding));
- }else{
- i->discard();
- i->subtractFrom(&outstanding);
- unacked.erase(i);
+ }
+ else {
+ {
+ Mutex::ScopedLock locker(lock);
+ ack_iterator i = find_if(
+ unacked.begin(), unacked.end(),
+ boost::bind(&DeliveryRecord::matches, _1, deliveryTag));
+ if(i == unacked.end()) {
+ throw InvalidAckException();
+ }
+ else if(multiple) {
+ ack_iterator end = ++i;
+ for_each(unacked.begin(), end,
+ mem_fun_ref(&DeliveryRecord::discard));
+ unacked.erase(unacked.begin(), end);
+
+ //recalculate the prefetch:
+ outstanding.reset();
+ for_each(
+ unacked.begin(), unacked.end(),
+ boost::bind(&DeliveryRecord::addTo, _1, &outstanding));
+ }
+ else {
+ i->discard();
+ i->subtractFrom(&outstanding);
+ unacked.erase(i);
+ }
}
-
//if the prefetch limit had previously been reached, there may
//be messages that can be now be delivered
- for(consumer_iterator j = consumers.begin(); j != consumers.end(); j++){
+
+ // TODO aconway 2006-12-13: Does this need to be atomic?
+ // If so we need a redesign, requestDispatch re-enters
+ // Channel::dispatch.
+ //
+ for(consumer_iterator j = consumers.begin(); j != consumers.end(); j++){
j->second->requestDispatch();
}
}
}
-void Channel::recover(bool requeue){
- Mutex::ScopedLock locker(deliveryLock);//need to synchronize with possible concurrent delivery
-
- if(requeue){
- outstanding.reset();
- std::list<DeliveryRecord> copy = unacked;
- unacked.clear();
- for_each(copy.begin(), copy.end(), mem_fun_ref(&DeliveryRecord::requeue));
- }else{
- for_each(unacked.begin(), unacked.end(), bind2nd(mem_fun_ref(&DeliveryRecord::redeliver), this));
+void Channel::recover(bool requeue) {
+ std::list<DeliveryRecord> copyUnacked;
+ boost::function1<void, DeliveryRecord&> recoverFn;
+ {
+ Mutex::ScopedLock l(lock);
+ if(requeue) {
+ outstanding.reset();
+ copyUnacked.swap(unacked);
+ recoverFn = boost::bind(&DeliveryRecord::requeue, _1);
+ }
+ else {
+ copyUnacked = unacked;
+ recoverFn = boost::bind(&DeliveryRecord::redeliver, _1, this);
+ }
}
+ // TODO aconway 2006-12-13: Does recovery of copyUnacked have to
+ // be atomic with extracting the list?
+ for_each(copyUnacked.begin(), copyUnacked.end(), recoverFn);
}
bool Channel::get(Queue::shared_ptr queue, bool ackExpected){
+ Mutex::ScopedLock l(lock);
+ // TODO aconway 2006-12-13: Nasty to have all these external calls
+ // inside a critical.section but none appear to have blocking potential.
+ // sendGetOk does non-blocking IO
+ //
Message::shared_ptr msg = queue->dequeue();
- if(msg){
- Mutex::ScopedLock locker(deliveryLock);
+ if(msg) {
u_int64_t myDeliveryTag = currentDeliveryTag++;
- msg->sendGetOk(out, id, queue->getMessageCount() + 1, myDeliveryTag, framesize);
+ u_int32_t count = queue->getMessageCount();
+ msg->sendGetOk(out, id, count + 1, myDeliveryTag, framesize);
if(ackExpected){
unacked.push_back(DeliveryRecord(msg, queue, myDeliveryTag));
}
return true;
- }else{
- return false;
}
+ return false;
}
-void Channel::deliver(Message::shared_ptr& msg, const string& consumerTag, u_int64_t deliveryTag){
+void Channel::deliver(Message::shared_ptr& msg, const string& consumerTag,
+ u_int64_t deliveryTag){
msg->deliver(out, id, consumerTag, deliveryTag, framesize);
}
Modified: incubator/qpid/branches/event-queue-2006-12-20/cpp/lib/broker/BrokerChannel.h
URL: http://svn.apache.org/viewvc/incubator/qpid/branches/event-queue-2006-12-20/cpp/lib/broker/BrokerChannel.h?view=diff&rev=489159&r1=489158&r2=489159
==============================================================================
--- incubator/qpid/branches/event-queue-2006-12-20/cpp/lib/broker/BrokerChannel.h (original)
+++ incubator/qpid/branches/event-queue-2006-12-20/cpp/lib/broker/BrokerChannel.h Wed Dec 20 09:54:05 2006
@@ -79,10 +79,10 @@
u_int32_t prefetchSize;
u_int16_t prefetchCount;
Prefetch outstanding;
- u_int32_t framesize;
+ const u_int32_t framesize;
NameGenerator tagGenerator;
std::list<DeliveryRecord> unacked;
- qpid::sys::Mutex deliveryLock;
+ qpid::sys::Mutex lock;
TxBuffer txBuffer;
AccumulatedAck accumulatedAck;
MessageStore* const store;
Modified: incubator/qpid/branches/event-queue-2006-12-20/cpp/lib/broker/BrokerMessage.cpp
URL: http://svn.apache.org/viewvc/incubator/qpid/branches/event-queue-2006-12-20/cpp/lib/broker/BrokerMessage.cpp?view=diff&rev=489159&r1=489158&r2=489159
==============================================================================
--- incubator/qpid/branches/event-queue-2006-12-20/cpp/lib/broker/BrokerMessage.cpp (original)
+++ incubator/qpid/branches/event-queue-2006-12-20/cpp/lib/broker/BrokerMessage.cpp Wed Dec 20 09:54:05 2006
@@ -143,7 +143,8 @@
{
u_int64_t expected = expectedContentSize();
if (expected != buffer.available()) {
- std::cout << "WARN: Expected " << expectedContentSize() << " bytes, got " << buffer.available() << std::endl;
+ std::cout << "WARN: Expected " << expectedContentSize()
+ << " bytes, got " << buffer.available() << std::endl;
throw Exception("Cannot decode content, buffer not large enough.");
}
Modified: incubator/qpid/branches/event-queue-2006-12-20/cpp/lib/broker/BrokerQueue.cpp
URL: http://svn.apache.org/viewvc/incubator/qpid/branches/event-queue-2006-12-20/cpp/lib/broker/BrokerQueue.cpp?view=diff&rev=489159&r1=489158&r2=489159
==============================================================================
--- incubator/qpid/branches/event-queue-2006-12-20/cpp/lib/broker/BrokerQueue.cpp (original)
+++ incubator/qpid/branches/event-queue-2006-12-20/cpp/lib/broker/BrokerQueue.cpp Wed Dec 20 09:54:05 2006
@@ -83,7 +83,8 @@
return false;
}else if(exclusive){
if(!exclusive->deliver(msg)){
- std::cout << "WARNING: Dropping undeliverable message from queue with exclusive consumer." << std::endl;
+ std::cout << "WARNING: Dropping undeliverable message "
+ << "from queue with exclusive consumer." << std::endl;
}
return true;
}else{
Modified: incubator/qpid/branches/event-queue-2006-12-20/cpp/lib/broker/TopicExchange.cpp
URL: http://svn.apache.org/viewvc/incubator/qpid/branches/event-queue-2006-12-20/cpp/lib/broker/TopicExchange.cpp?view=diff&rev=489159&r1=489158&r2=489159
==============================================================================
--- incubator/qpid/branches/event-queue-2006-12-20/cpp/lib/broker/TopicExchange.cpp (original)
+++ incubator/qpid/branches/event-queue-2006-12-20/cpp/lib/broker/TopicExchange.cpp Wed Dec 20 09:54:05 2006
@@ -81,10 +81,7 @@
namespace {
-// TODO aconway 2006-09-20: Ineficient to convert every routingKey to a string.
-// Need StringRef class that operates on a string in place witout copy.
-// Should be applied everywhere strings are extracted from frames.
-//
+
bool do_match(Tokens::const_iterator pattern_begin, Tokens::const_iterator pattern_end, Tokens::const_iterator target_begin, Tokens::const_iterator target_end)
{
// Invariant: [pattern_begin..p) matches [target_begin..t)
Modified: incubator/qpid/branches/event-queue-2006-12-20/cpp/lib/client/ResponseHandler.cpp
URL: http://svn.apache.org/viewvc/incubator/qpid/branches/event-queue-2006-12-20/cpp/lib/client/ResponseHandler.cpp?view=diff&rev=489159&r1=489158&r2=489159
==============================================================================
--- incubator/qpid/branches/event-queue-2006-12-20/cpp/lib/client/ResponseHandler.cpp (original)
+++ incubator/qpid/branches/event-queue-2006-12-20/cpp/lib/client/ResponseHandler.cpp Wed Dec 20 09:54:05 2006
@@ -28,32 +28,33 @@
qpid::client::ResponseHandler::~ResponseHandler(){}
-bool qpid::client::ResponseHandler::validate(const qpid::framing::AMQMethodBody& expected){
+bool qpid::client::ResponseHandler::validate(
+ const qpid::framing::AMQMethodBody& expected)
+{
return expected.match(response.get());
}
void qpid::client::ResponseHandler::waitForResponse(){
Monitor::ScopedLock l(monitor);
- if(waiting){
+ while(waiting)
monitor.wait();
- }
}
-void qpid::client::ResponseHandler::signalResponse(qpid::framing::AMQMethodBody::shared_ptr _response){
- response = _response;
+void qpid::client::ResponseHandler::signalResponse(
+ qpid::framing::AMQMethodBody::shared_ptr _response)
+{
Monitor::ScopedLock l(monitor);
+ response = _response;
waiting = false;
monitor.notify();
}
void qpid::client::ResponseHandler::receive(const qpid::framing::AMQMethodBody& expected){
Monitor::ScopedLock l(monitor);
- if(waiting){
+ while(waiting)
monitor.wait();
- }
- if(!validate(expected)){
+ if(!validate(expected))
THROW_QPID_ERROR(PROTOCOL_ERROR, "Protocol Error");
- }
}
void qpid::client::ResponseHandler::expect(){
Modified: incubator/qpid/branches/event-queue-2006-12-20/cpp/lib/common/Exception.cpp
URL: http://svn.apache.org/viewvc/incubator/qpid/branches/event-queue-2006-12-20/cpp/lib/common/Exception.cpp?view=diff&rev=489159&r1=489158&r2=489159
==============================================================================
--- incubator/qpid/branches/event-queue-2006-12-20/cpp/lib/common/Exception.cpp (original)
+++ incubator/qpid/branches/event-queue-2006-12-20/cpp/lib/common/Exception.cpp Wed Dec 20 09:54:05 2006
@@ -20,6 +20,7 @@
*/
#include <Exception.h>
+#include <iostream>
namespace qpid {
@@ -29,14 +30,32 @@
Exception::Exception(const char* str) throw() : whatStr(str) {}
+Exception::Exception(const std::exception& e) throw() : whatStr(e.what()) {}
+
Exception::~Exception() throw() {}
const char* Exception::what() const throw() { return whatStr.c_str(); }
std::string Exception::toString() const throw() { return whatStr; }
-Exception* Exception::clone() const throw() { return new Exception(*this); }
+Exception::auto_ptr Exception::clone() const throw() {
+ return Exception::auto_ptr(new Exception(*this));
+}
void Exception::throwSelf() const { throw *this; }
+
+const char* Exception::defaultMessage = "Unexpected exception";
+
+void Exception::log(const char* what, const char* message) {
+ std::cout << message << ": " << what << std::endl;
+}
+
+void Exception::log(const std::exception& e, const char* message) {
+ log(e.what(), message);
+}
+
+void Exception::logUnknown(const char* message) {
+ log("unknown exception.", message);
+}
} // namespace qpid
Modified: incubator/qpid/branches/event-queue-2006-12-20/cpp/lib/common/Exception.h
URL: http://svn.apache.org/viewvc/incubator/qpid/branches/event-queue-2006-12-20/cpp/lib/common/Exception.h?view=diff&rev=489159&r1=489158&r2=489159
==============================================================================
--- incubator/qpid/branches/event-queue-2006-12-20/cpp/lib/common/Exception.h (original)
+++ incubator/qpid/branches/event-queue-2006-12-20/cpp/lib/common/Exception.h Wed Dec 20 09:54:05 2006
@@ -26,6 +26,7 @@
#include <string>
#include <memory>
#include <boost/shared_ptr.hpp>
+#include <boost/function.hpp>
namespace qpid
{
@@ -38,6 +39,10 @@
std::string whatStr;
public:
+ typedef boost::shared_ptr<Exception> shared_ptr;
+ typedef boost::shared_ptr<const Exception> shared_ptr_const;
+ typedef std::auto_ptr<Exception> auto_ptr;
+
Exception() throw();
Exception(const std::string& str) throw();
Exception(const char* str) throw();
@@ -48,14 +53,65 @@
virtual const char* what() const throw();
virtual std::string toString() const throw();
- virtual Exception* clone() const throw();
+ virtual std::auto_ptr<Exception> clone() const throw();
virtual void throwSelf() const;
- typedef boost::shared_ptr<Exception> shared_ptr;
-};
-
+ /** Default message: "Unknown exception" or something like it. */
+ static const char* defaultMessage;
-}
+ /**
+ * Log a message of the form "message: what"
+ *@param what Exception's what() message.
+ *@param message Prefix message.
+ */
+ static void log(const char* what, const char* message = defaultMessage);
+
+ /**
+ * Log an exception.
+ *@param e Exception to log.
+
+ */
+ static void log(
+ const std::exception& e, const char* message = defaultMessage);
+
+
+ /**
+ * Log an unknown exception - use in catch(...)
+ *@param message Prefix message.
+ */
+ static void logUnknown(const char* message = defaultMessage);
+
+ /**
+ * Wrapper template function to call another function inside
+ * try/catch and log any exception. Use boost::bind to wrap
+ * member function calls or functions with arguments.
+ *
+ *@param f Function to call in try block.
+ *@param retrhow If true the exception is rethrown.
+ *@param message Prefix message.
+ */
+ template <class T>
+ static T tryCatchLog(boost::function0<T> f, bool rethrow=true,
+ const char* message=defaultMessage)
+ {
+ try {
+ return f();
+ }
+ catch (const std::exception& e) {
+ log(e, message);
+ if (rethrow)
+ throw;
+ }
+ catch (...) {
+ logUnknown(message);
+ if (rethrow)
+ throw;
+ }
+ }
+
+};
+
+} // namespace qpid
#endif /*!_Exception_*/
Modified: incubator/qpid/branches/event-queue-2006-12-20/cpp/lib/common/ExceptionHolder.cpp
URL: http://svn.apache.org/viewvc/incubator/qpid/branches/event-queue-2006-12-20/cpp/lib/common/ExceptionHolder.cpp?view=diff&rev=489159&r1=489158&r2=489159
==============================================================================
--- incubator/qpid/branches/event-queue-2006-12-20/cpp/lib/common/ExceptionHolder.cpp (original)
+++ incubator/qpid/branches/event-queue-2006-12-20/cpp/lib/common/ExceptionHolder.cpp Wed Dec 20 09:54:05 2006
@@ -1,32 +0,0 @@
-/*
- *
- * Copyright (c) 2006 The Apache Software Foundation
- *
- * Licensed 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 "ExceptionHolder.h"
-
-namespace qpid {
-
-ExceptionHolder::ExceptionHolder(const std::exception& e) {
- const Exception* ex = dynamic_cast<const Exception*>(&e);
- if (ex) {
- reset(ex->clone());
- } else {
- reset(new Exception(e.what()));
- }
-}
-
-}
Modified: incubator/qpid/branches/event-queue-2006-12-20/cpp/lib/common/ExceptionHolder.h
URL: http://svn.apache.org/viewvc/incubator/qpid/branches/event-queue-2006-12-20/cpp/lib/common/ExceptionHolder.h?view=diff&rev=489159&r1=489158&r2=489159
==============================================================================
--- incubator/qpid/branches/event-queue-2006-12-20/cpp/lib/common/ExceptionHolder.h (original)
+++ incubator/qpid/branches/event-queue-2006-12-20/cpp/lib/common/ExceptionHolder.h Wed Dec 20 09:54:05 2006
@@ -1,62 +0,0 @@
-#ifndef _qpid_ExceptionHolder_h
-#define _qpid_ExceptionHolder_h
-
-/*
- *
- * Copyright (c) 2006 The Apache Software Foundation
- *
- * Licensed 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 <Exception.h>
-#include <boost/shared_ptr.hpp>
-
-namespace qpid {
-
-/**
- * Holder for a heap-allocated exc eption that can be stack allocated
- * and thrown safely.
- *
- * Basically this is a shared_ptr with the Exception functions added
- * so the catcher need not be aware that it is a pointer rather than a
- * reference.
- *
- * shared_ptr is chosen over auto_ptr because it has normal
- * copy semantics.
- */
-class ExceptionHolder : public Exception, public boost::shared_ptr<Exception>
-{
- public:
- typedef boost::shared_ptr<Exception> shared_ptr;
-
- ExceptionHolder() throw() {}
- ExceptionHolder(Exception* p) throw() : shared_ptr(p) {}
- ExceptionHolder(shared_ptr p) throw() : shared_ptr(p) {}
-
- ExceptionHolder(const Exception& e) throw() : shared_ptr(e.clone()) {}
- ExceptionHolder(const std::exception& e);
-
- ~ExceptionHolder() throw() {}
-
- const char* what() const throw() { return (*this)->what(); }
- std::string toString() const throw() { return (*this)->toString(); }
- virtual Exception* clone() const throw() { return (*this)->clone(); }
- virtual void throwSelf() const { (*this)->throwSelf(); }
-};
-
-} // namespace qpid
-
-
-
-#endif /*!_qpid_ExceptionHolder_h*/
Modified: incubator/qpid/branches/event-queue-2006-12-20/cpp/lib/common/Makefile.am
URL: http://svn.apache.org/viewvc/incubator/qpid/branches/event-queue-2006-12-20/cpp/lib/common/Makefile.am?view=diff&rev=489159&r1=489158&r2=489159
==============================================================================
--- incubator/qpid/branches/event-queue-2006-12-20/cpp/lib/common/Makefile.am (original)
+++ incubator/qpid/branches/event-queue-2006-12-20/cpp/lib/common/Makefile.am Wed Dec 20 09:54:05 2006
@@ -24,21 +24,23 @@
$(apr)/LFSessionContext.h
posix = sys/posix
-posix_src = \
- $(posix)/PosixAcceptor.cpp \
- $(posix)/Socket.cpp \
- $(posix)/Thread.cpp \
- $(posix)/check.cpp \
- $(posix)/EventChannel.cpp \
- $(posix)/EventChannelThreads.cpp
-posix_hdr = \
- $(posix)/check.h \
- $(posix)/EventChannel.h \
- $(posix)/EventChannelThreads.h
-
-EXTRA_DIST=$(posix_src) $(posix_hdr)
-platform_src = $(apr_src)
-platform_hdr = $(apr_hdr)
+posix_src = \
+ $(posix)/EventChannelAcceptor.cpp \
+ $(posix)/Socket.cpp \
+ $(posix)/Thread.cpp \
+ $(posix)/check.cpp \
+ $(posix)/EventChannel.cpp \
+ $(posix)/EventChannelThreads.cpp \
+ $(posix)/EventChannelConnection.cpp
+posix_hdr = \
+ $(posix)/check.h \
+ $(posix)/EventChannel.h \
+ $(posix)/EventChannelThreads.h \
+ $(posix)/EventChannelConnection.h
+
+EXTRA_DIST=$(apr_src) $(apr_hdr)
+platform_src = $(posix_src)
+platform_hdr = $(posix_hdr)
framing = framing
gen = $(srcdir)/../../gen
@@ -76,7 +78,6 @@
$(gen)/AMQP_MethodVersionMap.cpp \
$(gen)/AMQP_ServerProxy.cpp \
Exception.cpp \
- ExceptionHolder.cpp \
QpidError.cpp \
sys/Runnable.cpp \
sys/Time.cpp
@@ -107,7 +108,6 @@
$(framing)/amqp_types.h \
$(framing)/AMQP_HighestVersion.h \
Exception.h \
- ExceptionHolder.h \
QpidError.h \
SharedObject.h \
sys/Acceptor.h \
Modified: incubator/qpid/branches/event-queue-2006-12-20/cpp/lib/common/QpidError.cpp
URL: http://svn.apache.org/viewvc/incubator/qpid/branches/event-queue-2006-12-20/cpp/lib/common/QpidError.cpp?view=diff&rev=489159&r1=489158&r2=489159
==============================================================================
--- incubator/qpid/branches/event-queue-2006-12-20/cpp/lib/common/QpidError.cpp (original)
+++ incubator/qpid/branches/event-queue-2006-12-20/cpp/lib/common/QpidError.cpp Wed Dec 20 09:54:05 2006
@@ -22,23 +22,41 @@
#include <QpidError.h>
#include <sstream>
-using namespace qpid;
+namespace qpid {
QpidError::QpidError() : code(0) {}
-QpidError::QpidError(int _code, const std::string& _msg,
- const SrcLine& _loc) throw()
+QpidError::QpidError(
+ int _code, const std::string& _msg, Location _loc) throw()
: code(_code), msg(_msg), location(_loc)
{
- std::ostringstream os;
- os << "Error [" << code << "] " << msg << " ("
- << location.file << ":" << location.line << ")";
- whatStr = os.str();
+ setWhat();
+}
+
+QpidError::QpidError(
+ int _code, const char* _msg, Location _loc) throw()
+ : code(_code), msg(_msg), location(_loc)
+{
+ setWhat();
}
QpidError::~QpidError() throw() {}
-Exception* QpidError::clone() const throw() { return new QpidError(*this); }
+Exception::auto_ptr QpidError::clone() const throw() {
+ return Exception::auto_ptr(new QpidError(*this));
+}
void QpidError::throwSelf() const { throw *this; }
+void QpidError::setWhat() {
+ std::ostringstream os;
+ os << "Error [" << code << "] " << msg;
+ if (location.file) {
+ os << " (" ;
+ os << location.file << ":" << location.line;
+ os << ")";
+ }
+ whatStr = os.str();
+}
+
+} // namespace qpid
Modified: incubator/qpid/branches/event-queue-2006-12-20/cpp/lib/common/QpidError.h
URL: http://svn.apache.org/viewvc/incubator/qpid/branches/event-queue-2006-12-20/cpp/lib/common/QpidError.h?view=diff&rev=489159&r1=489158&r2=489159
==============================================================================
--- incubator/qpid/branches/event-queue-2006-12-20/cpp/lib/common/QpidError.h (original)
+++ incubator/qpid/branches/event-queue-2006-12-20/cpp/lib/common/QpidError.h Wed Dec 20 09:54:05 2006
@@ -24,37 +24,45 @@
#include <memory>
#include <ostream>
#include <Exception.h>
+#include <boost/current_function.hpp>
namespace qpid {
-struct SrcLine {
- public:
- SrcLine(const std::string& file_="", int line_=0) :
- file(file_), line(line_) {}
-
- std::string file;
- int line;
-};
-
class QpidError : public Exception {
public:
+ // Use macro QPID_LOCATION to construct a location.
+ struct Location {
+ Location(const char* function_=0, const char* file_=0, int line_=0) :
+ function(function_), file(file_), line(line_) {}
+ const char* function;
+ const char* file;
+ int line;
+ };
+
const int code;
const std::string msg;
- const SrcLine location;
+ const Location location;
QpidError();
- QpidError(int _code, const std::string& _msg, const SrcLine& _loc) throw();
+ QpidError(int _code, const char* _msg, const Location _loc) throw();
+ QpidError(int _code, const std::string& _msg, const Location _loc) throw();
+
~QpidError() throw();
- Exception* clone() const throw();
+ Exception::auto_ptr clone() const throw();
void throwSelf() const;
+
+ private:
+ void setWhat();
};
} // namespace qpid
-#define SRCLINE ::qpid::SrcLine(__FILE__, __LINE__)
+#define QPID_ERROR_LOCATION \
+ ::qpid::QpidError::Location(BOOST_CURRENT_FUNCTION, __FILE__, __LINE__)
-#define QPID_ERROR(CODE, MESSAGE) ::qpid::QpidError((CODE), (MESSAGE), SRCLINE)
+#define QPID_ERROR(CODE, MESSAGE) \
+ ::qpid::QpidError((CODE), (MESSAGE), QPID_ERROR_LOCATION)
#define THROW_QPID_ERROR(CODE, MESSAGE) throw QPID_ERROR(CODE,MESSAGE)
Modified: incubator/qpid/branches/event-queue-2006-12-20/cpp/lib/common/framing/AMQFrame.h
URL: http://svn.apache.org/viewvc/incubator/qpid/branches/event-queue-2006-12-20/cpp/lib/common/framing/AMQFrame.h?view=diff&rev=489159&r1=489158&r2=489159
==============================================================================
--- incubator/qpid/branches/event-queue-2006-12-20/cpp/lib/common/framing/AMQFrame.h (original)
+++ incubator/qpid/branches/event-queue-2006-12-20/cpp/lib/common/framing/AMQFrame.h Wed Dec 20 09:54:05 2006
@@ -40,9 +40,9 @@
static AMQP_MethodVersionMap versionMap;
u_int16_t channel;
- u_int8_t type;//used if the body is decoded separately from the 'head'
+ u_int8_t type;//used if body decoded separately from 'head'
AMQBody::shared_ptr body;
- AMQBody::shared_ptr createMethodBody(Buffer& buffer);
+ AMQBody::shared_ptr createMethodBody(Buffer& buffer);
public:
AMQFrame();
Modified: incubator/qpid/branches/event-queue-2006-12-20/cpp/lib/common/sys/Acceptor.h
URL: http://svn.apache.org/viewvc/incubator/qpid/branches/event-queue-2006-12-20/cpp/lib/common/sys/Acceptor.h?view=diff&rev=489159&r1=489158&r2=489159
==============================================================================
--- incubator/qpid/branches/event-queue-2006-12-20/cpp/lib/common/sys/Acceptor.h (original)
+++ incubator/qpid/branches/event-queue-2006-12-20/cpp/lib/common/sys/Acceptor.h Wed Dec 20 09:54:05 2006
@@ -33,10 +33,11 @@
class Acceptor : public qpid::SharedObject<Acceptor>
{
public:
- static Acceptor::shared_ptr create(int16_t port, int backlog, int threads, bool trace = false);
- virtual ~Acceptor() = 0;
- virtual int16_t getPort() const = 0;
- virtual void run(qpid::sys::SessionHandlerFactory* factory) = 0;
+ static Acceptor::shared_ptr create(
+ int16_t port, int backlog, int threads, bool trace = false);
+ virtual ~Acceptor();
+ virtual int getPort() const = 0;
+ virtual void run(SessionHandlerFactory& factory) = 0;
virtual void shutdown() = 0;
};
Modified: incubator/qpid/branches/event-queue-2006-12-20/cpp/lib/common/sys/AtomicCount.h
URL: http://svn.apache.org/viewvc/incubator/qpid/branches/event-queue-2006-12-20/cpp/lib/common/sys/AtomicCount.h?view=diff&rev=489159&r1=489158&r2=489159
==============================================================================
--- incubator/qpid/branches/event-queue-2006-12-20/cpp/lib/common/sys/AtomicCount.h (original)
+++ incubator/qpid/branches/event-queue-2006-12-20/cpp/lib/common/sys/AtomicCount.h Wed Dec 20 09:54:05 2006
@@ -21,36 +21,52 @@
#include <boost/detail/atomic_count.hpp>
#include <boost/noncopyable.hpp>
+#include <boost/function.hpp>
namespace qpid {
namespace sys {
/**
+ * Increment counter in constructor and decrement in destructor.
+ * Optionally call a function if the decremented counter value is 0.
+ * Note the function must not throw, it is called in the destructor.
+ */
+template <class Count>
+class ScopedIncrement : boost::noncopyable {
+ public:
+ ScopedIncrement(Count& c, boost::function0<void> f=0)
+ : count(c), callback(f) { ++count; }
+ ~ScopedIncrement() { if (--count == 0 && callback) callback(); }
+
+ private:
+ Count& count;
+ boost::function0<void> callback;
+};
+
+/** Decrement counter in constructor and increment in destructor. */
+template <class Count>
+class ScopedDecrement : boost::noncopyable {
+ public:
+ ScopedDecrement(Count& c) : count(c) { value = --count; }
+ ~ScopedDecrement() { ++count; }
+
+ /** Return the value after the decrement. */
+ operator long() { return value; }
+
+ private:
+ Count& count;
+ long value;
+};
+
+
+/**
* Atomic counter.
*/
class AtomicCount : boost::noncopyable {
public:
- class ScopedDecrement : boost::noncopyable {
- public:
- /** Decrement counter in constructor and increment in destructor. */
- ScopedDecrement(AtomicCount& c) : count(c) { value = --count; }
- ~ScopedDecrement() { ++count; }
- /** Return the value returned by the decrement. */
- operator long() { return value; }
- private:
- AtomicCount& count;
- long value;
- };
-
- class ScopedIncrement : boost::noncopyable {
- public:
- /** Increment counter in constructor and increment in destructor. */
- ScopedIncrement(AtomicCount& c) : count(c) { ++count; }
- ~ScopedIncrement() { --count; }
- private:
- AtomicCount& count;
- };
-
+ typedef ScopedIncrement<AtomicCount> ScopedIncrement;
+ typedef ScopedDecrement<AtomicCount> ScopedDecrement;
+
AtomicCount(long value = 0) : count(value) {}
void operator++() { ++count ; }
Modified: incubator/qpid/branches/event-queue-2006-12-20/cpp/lib/common/sys/Runnable.cpp
URL: http://svn.apache.org/viewvc/incubator/qpid/branches/event-queue-2006-12-20/cpp/lib/common/sys/Runnable.cpp?view=diff&rev=489159&r1=489158&r2=489159
==============================================================================
--- incubator/qpid/branches/event-queue-2006-12-20/cpp/lib/common/sys/Runnable.cpp (original)
+++ incubator/qpid/branches/event-queue-2006-12-20/cpp/lib/common/sys/Runnable.cpp Wed Dec 20 09:54:05 2006
@@ -29,4 +29,8 @@
return boost::bind(&Runnable::run, this);
}
+void FunctorRunnable::run() {
+ f();
+}
+
}}
Modified: incubator/qpid/branches/event-queue-2006-12-20/cpp/lib/common/sys/Runnable.h
URL: http://svn.apache.org/viewvc/incubator/qpid/branches/event-queue-2006-12-20/cpp/lib/common/sys/Runnable.h?view=diff&rev=489159&r1=489158&r2=489159
==============================================================================
--- incubator/qpid/branches/event-queue-2006-12-20/cpp/lib/common/sys/Runnable.h (original)
+++ incubator/qpid/branches/event-queue-2006-12-20/cpp/lib/common/sys/Runnable.h Wed Dec 20 09:54:05 2006
@@ -44,7 +44,16 @@
Functor functor();
};
-}}
+/** Runnable wrapper for a functor */
+class FunctorRunnable : public Runnable {
+ public:
+ explicit FunctorRunnable(const Runnable::Functor& runMe) : f(runMe) {}
+ void run();
+ private:
+ Runnable::Functor f;
+};
+
+}} // namespace qpid::sys
#endif
Modified: incubator/qpid/branches/event-queue-2006-12-20/cpp/lib/common/sys/Socket.h
URL: http://svn.apache.org/viewvc/incubator/qpid/branches/event-queue-2006-12-20/cpp/lib/common/sys/Socket.h?view=diff&rev=489159&r1=489158&r2=489159
==============================================================================
--- incubator/qpid/branches/event-queue-2006-12-20/cpp/lib/common/sys/Socket.h (original)
+++ incubator/qpid/branches/event-queue-2006-12-20/cpp/lib/common/sys/Socket.h Wed Dec 20 09:54:05 2006
@@ -70,8 +70,11 @@
*/
int listen(int port = 0, int backlog = 10);
+ /** Accept a connection. This socket must be listening */
+ Socket accept();
+
/** Get file descriptor */
- int fd();
+ int fd() const;
private:
#ifdef USE_APR
Modified: incubator/qpid/branches/event-queue-2006-12-20/cpp/lib/common/sys/Thread.h
URL: http://svn.apache.org/viewvc/incubator/qpid/branches/event-queue-2006-12-20/cpp/lib/common/sys/Thread.h?view=diff&rev=489159&r1=489158&r2=489159
==============================================================================
--- incubator/qpid/branches/event-queue-2006-12-20/cpp/lib/common/sys/Thread.h (original)
+++ incubator/qpid/branches/event-queue-2006-12-20/cpp/lib/common/sys/Thread.h Wed Dec 20 09:54:05 2006
@@ -116,7 +116,8 @@
}
void Thread::join(){
- QPID_POSIX_THROW_IF(pthread_join(thread, 0));
+ if (thread != 0)
+ QPID_POSIX_THROW_IF(pthread_join(thread, 0));
}
long Thread::id() {
Modified: incubator/qpid/branches/event-queue-2006-12-20/cpp/lib/common/sys/Time.h
URL: http://svn.apache.org/viewvc/incubator/qpid/branches/event-queue-2006-12-20/cpp/lib/common/sys/Time.h?view=diff&rev=489159&r1=489158&r2=489159
==============================================================================
--- incubator/qpid/branches/event-queue-2006-12-20/cpp/lib/common/sys/Time.h (original)
+++ incubator/qpid/branches/event-queue-2006-12-20/cpp/lib/common/sys/Time.h Wed Dec 20 09:54:05 2006
@@ -22,6 +22,7 @@
*
*/
+#include <limits>
#include <stdint.h>
#ifdef USE_APR
@@ -33,7 +34,7 @@
namespace qpid {
namespace sys {
-/** Time in nanoseconds */
+/** Time in nanoseconds. */
typedef int64_t Time;
Time now();
@@ -47,6 +48,9 @@
/** Nanoseconds per nanosecond. */
const Time TIME_NSEC = 1;
+/** Value to represent an infinite timeout */
+const Time TIME_INFINITE = std::numeric_limits<Time>::max();
+
#ifndef USE_APR
struct timespec toTimespec(const Time& t);
struct timespec& toTimespec(struct timespec& ts, const Time& t);
Modified: incubator/qpid/branches/event-queue-2006-12-20/cpp/lib/common/sys/apr/APRAcceptor.cpp
URL: http://svn.apache.org/viewvc/incubator/qpid/branches/event-queue-2006-12-20/cpp/lib/common/sys/apr/APRAcceptor.cpp?view=diff&rev=489159&r1=489158&r2=489159
==============================================================================
--- incubator/qpid/branches/event-queue-2006-12-20/cpp/lib/common/sys/apr/APRAcceptor.cpp (original)
+++ incubator/qpid/branches/event-queue-2006-12-20/cpp/lib/common/sys/apr/APRAcceptor.cpp Wed Dec 20 09:54:05 2006
@@ -32,20 +32,16 @@
{
public:
APRAcceptor(int16_t port, int backlog, int threads, bool trace);
- virtual int16_t getPort() const;
- virtual void run(qpid::sys::SessionHandlerFactory* factory);
+ virtual int getPort() const;
+ virtual void run(qpid::sys::SessionHandlerFactory& factory);
virtual void shutdown();
private:
- void shutdownImpl();
-
- private:
int16_t port;
bool trace;
LFProcessor processor;
apr_socket_t* socket;
volatile bool running;
- Mutex shutdownLock;
};
// Define generic Acceptor::create() to return APRAcceptor.
@@ -69,13 +65,13 @@
CHECK_APR_SUCCESS(apr_socket_listen(socket, backlog));
}
-int16_t APRAcceptor::getPort() const {
+int APRAcceptor::getPort() const {
apr_sockaddr_t* address;
CHECK_APR_SUCCESS(apr_socket_addr_get(&address, APR_LOCAL, socket));
return address->port;
}
-void APRAcceptor::run(SessionHandlerFactory* factory) {
+void APRAcceptor::run(SessionHandlerFactory& factory) {
running = true;
processor.start();
std::cout << "Listening on port " << getPort() << "..." << std::endl;
@@ -90,31 +86,23 @@
CHECK_APR_SUCCESS(apr_socket_opt_set(client, APR_SO_SNDBUF, 32768));
CHECK_APR_SUCCESS(apr_socket_opt_set(client, APR_SO_RCVBUF, 32768));
LFSessionContext* session = new LFSessionContext(APRPool::get(), client, &processor, trace);
- session->init(factory->create(session));
+ session->init(factory.create(session));
}else{
- Mutex::ScopedLock locker(shutdownLock);
- if(running) {
- if(status != APR_EINTR){
- std::cout << "ERROR: " << get_desc(status) << std::endl;
- }
- shutdownImpl();
+ running = false;
+ if(status != APR_EINTR){
+ std::cout << "ERROR: " << get_desc(status) << std::endl;
}
}
}
+ shutdown();
}
void APRAcceptor::shutdown() {
- Mutex::ScopedLock locker(shutdownLock);
if (running) {
- shutdownImpl();
+ running = false;
+ processor.stop();
+ CHECK_APR_SUCCESS(apr_socket_close(socket));
}
-}
-
-void APRAcceptor::shutdownImpl() {
- Mutex::ScopedLock locker(shutdownLock);
- running = false;
- processor.stop();
- CHECK_APR_SUCCESS(apr_socket_close(socket));
}
Modified: incubator/qpid/branches/event-queue-2006-12-20/cpp/lib/common/sys/apr/LFProcessor.cpp
URL: http://svn.apache.org/viewvc/incubator/qpid/branches/event-queue-2006-12-20/cpp/lib/common/sys/apr/LFProcessor.cpp?view=diff&rev=489159&r1=489158&r2=489159
==============================================================================
--- incubator/qpid/branches/event-queue-2006-12-20/cpp/lib/common/sys/apr/LFProcessor.cpp (original)
+++ incubator/qpid/branches/event-queue-2006-12-20/cpp/lib/common/sys/apr/LFProcessor.cpp Wed Dec 20 09:54:05 2006
@@ -27,8 +27,6 @@
using namespace qpid::sys;
using qpid::QpidError;
-// TODO aconway 2006-10-12: stopped is read outside locks.
-//
LFProcessor::LFProcessor(apr_pool_t* pool, int _workers, int _size, int _timeout) :
size(_size),
Modified: incubator/qpid/branches/event-queue-2006-12-20/cpp/lib/common/sys/posix/EventChannel.cpp
URL: http://svn.apache.org/viewvc/incubator/qpid/branches/event-queue-2006-12-20/cpp/lib/common/sys/posix/EventChannel.cpp?view=diff&rev=489159&r1=489158&r2=489159
==============================================================================
--- incubator/qpid/branches/event-queue-2006-12-20/cpp/lib/common/sys/posix/EventChannel.cpp (original)
+++ incubator/qpid/branches/event-queue-2006-12-20/cpp/lib/common/sys/posix/EventChannel.cpp Wed Dec 20 09:54:05 2006
@@ -1,4 +1,4 @@
-/*
+/*
*
* Copyright (c) 2006 The Apache Software Foundation
*
@@ -16,6 +16,13 @@
*
*/
+// TODO aconway 2006-12-15: Locking review.
+
+// TODO aconway 2006-12-15: use Descriptor pointers everywhere,
+// get them from channel, pass them to Event constructors.
+// Eliminate lookup.
+
+
#include <mqueue.h>
#include <string.h>
#include <iostream>
@@ -29,10 +36,10 @@
#include <queue>
#include <boost/ptr_container/ptr_map.hpp>
-#include <boost/current_function.hpp>
+#include <boost/noncopyable.hpp>
+#include <boost/bind.hpp>
#include <QpidError.h>
-#include <sys/Monitor.h>
#include "check.h"
#include "EventChannel.h"
@@ -40,127 +47,319 @@
using namespace std;
-// Convenience template to zero out a struct.
-template <class S> struct ZeroStruct : public S {
- ZeroStruct() { memset(this, 0, sizeof(*this)); }
-};
-
namespace qpid {
namespace sys {
+// ================================================================
+// Private class declarations
+
+namespace {
+
+typedef enum { IN, OUT } Direction;
+typedef std::pair<Event*, Event*> EventPair;
+} // namespace
+
/**
- * EventHandler wraps an epoll file descriptor. Acts as private
- * interface between EventChannel and subclasses.
- *
- * Also implements Event interface for events that are not associated
- * with a file descriptor and are passed via the message queue.
- */
-class EventHandler : public Event, private Monitor
+ * Queue of events corresponding to one IO direction (IN or OUT).
+ * Each Descriptor contains two Queues.
+ */
+class EventChannel::Queue : private boost::noncopyable
{
public:
- EventHandler(int epollSize = 256);
- ~EventHandler();
+ Queue(Descriptor& container, Direction dir);
- int getEpollFd() { return epollFd; }
- void epollAdd(int fd, uint32_t epollEvents, Event* event);
- void epollMod(int fd, uint32_t epollEvents, Event* event);
- void epollDel(int fd);
+ /** Called by Event classes in prepare() */
+ void push(Event* e);
- void mqPut(Event* event);
- Event* mqGet();
-
- protected:
- // Should never be called, only complete.
- void prepare(EventHandler&) { assert(0); }
- Event* complete(EventHandler& eh);
+ /** Called when epoll wakes.
+ *@return The next completed event or 0.
+ */
+ Event* wake(uint32_t epollFlags);
+
+ void setBit(uint32_t &epollFlags);
+
+ void shutdown();
private:
+ typedef std::deque<Event*> EventQ;
+
+ inline bool isMyEvent(uint32_t flags) { return flags | myEvent; }
+
+ Mutex& lock; // Shared with Descriptor.
+ Descriptor& descriptor;
+ uint32_t myEvent; // Epoll event flag.
+ EventQ queue;
+};
+
+
+/**
+ * Manages a file descriptor in an epoll set.
+ *
+ * Can be shutdown and re-activated for the same file descriptor.
+ */
+class EventChannel::Descriptor : private boost::noncopyable {
+ public:
+ Descriptor() : epollFd(-1), myFd(-1),
+ inQueue(*this, IN), outQueue(*this, OUT) {}
+
+ void activate(int epollFd_, int myFd_);
+
+ /** Epoll woke up for this descriptor. */
+ EventPair wake(uint32_t epollEvents);
+
+ /** Shut down: close and remove file descriptor.
+ * May be re-activated if fd is reused.
+ */
+ void shutdown();
+
+ // TODO aconway 2006-12-18: Nasty. Need to clean up interaction.
+ void shutdownUnsafe();
+
+ bool isShutdown() { return epollFd == -1; }
+
+ Queue& getQueue(Direction d) { return d==IN ? inQueue : outQueue; }
+
+ private:
+ void update();
+ void epollCtl(int op, uint32_t events);
+
+ Mutex lock;
int epollFd;
- std::string mqName;
- int mqFd;
- std::queue<Event*> mqEvents;
+ int myFd;
+ Queue inQueue, outQueue;
+
+ friend class Queue;
};
-EventHandler::EventHandler(int epollSize)
-{
- epollFd = epoll_create(epollSize);
- if (epollFd < 0) throw QPID_POSIX_ERROR(errno);
+
+/**
+ * Holds the epoll fd, Descriptor map and dispatch queue.
+ * Most of the epoll work is done by the Descriptors.
+ */
+class EventChannel::Impl {
+ public:
+ Impl(int size = 256);
- // Create a POSIX message queue for non-fd events.
- // We write one byte and never read it is always ready for read
- // when we add it to epoll.
- //
- ZeroStruct<struct mq_attr> attr;
- attr.mq_maxmsg = 1;
- attr.mq_msgsize = 1;
- do {
- char tmpnam[L_tmpnam];
- tmpnam_r(tmpnam);
- mqName = tmpnam + 4; // Skip "tmp/"
- mqFd = mq_open(
- mqName.c_str(), O_CREAT|O_EXCL|O_RDWR|O_NONBLOCK, S_IRWXU, &attr);
- if (mqFd < 0) throw QPID_POSIX_ERROR(errno);
- } while (mqFd == EEXIST); // Name already taken, try again.
+ ~Impl();
+
+ /**
+ * Registers fd if not already registered.
+ */
+ Descriptor& getDescriptor(int fd);
+
+ /** Wait for an event, return 0 on timeout */
+ Event* wait(Time timeout);
+
+ Queue& getDispatchQueue() { return *dispatchQueue; }
+
+ private:
+
+ typedef boost::ptr_map<int, Descriptor> DescriptorMap;
+
+ Mutex lock;
+ int epollFd;
+ DescriptorMap descriptors;
+ int pipe[2];
+ Queue* dispatchQueue;
+};
- static char zero = '\0';
- mq_send(mqFd, &zero, 1, 0);
- epollAdd(mqFd, 0, this);
-}
-EventHandler::~EventHandler() {
- mq_close(mqFd);
- mq_unlink(mqName.c_str());
-}
-void EventHandler::mqPut(Event* event) {
- ScopedLock l(*this);
- assert(event != 0);
- mqEvents.push(event);
- epollMod(mqFd, EPOLLIN|EPOLLONESHOT, this);
+// ================================================================
+// EventChannel::Queue::implementation.
+
+static const char* shutdownMsg = "Event queue shut down.";
+
+EventChannel::Queue::Queue(Descriptor& d, Direction dir) : lock(d.lock), descriptor(d),
+ myEvent(dir==IN ? EPOLLIN : EPOLLOUT)
+{}
+
+void EventChannel::Queue::push(Event* e) {
+ Mutex::ScopedLock l(lock);
+ if (descriptor.isShutdown())
+ THROW_QPID_ERROR(INTERNAL_ERROR, shutdownMsg);
+ queue.push_back(e);
+ descriptor.update();
+}
+
+void EventChannel::Queue::setBit(uint32_t &epollFlags) {
+ if (queue.empty())
+ epollFlags &= ~myEvent;
+ else
+ epollFlags |= myEvent;
+}
+
+Event* EventChannel::Queue::wake(uint32_t epollFlags) {
+ // Called with lock held.
+ if (!queue.empty() && (isMyEvent(epollFlags))) {
+ Event* e = queue.front()->complete(descriptor);
+ if (e) {
+ queue.pop_front();
+ return e;
+ }
+ }
+ return 0;
+}
+
+void EventChannel::Queue::shutdown() {
+ // Mark all pending events with a shutdown exception.
+ // The server threads will remove and dispatch the events.
+ //
+ qpid::QpidError ex(INTERNAL_ERROR, shutdownMsg, QPID_ERROR_LOCATION);
+ for_each(queue.begin(), queue.end(),
+ boost::bind(&Event::setException, _1, ex));
}
-Event* EventHandler::mqGet() {
- ScopedLock l(*this);
- if (mqEvents.empty())
- return 0;
- Event* event = mqEvents.front();
- mqEvents.pop();
- if(!mqEvents.empty())
- epollMod(mqFd, EPOLLIN|EPOLLONESHOT, this);
- return event;
+
+// ================================================================
+// Descriptor
+
+
+void EventChannel::Descriptor::activate(int epollFd_, int myFd_) {
+ Mutex::ScopedLock l(lock);
+ assert(myFd < 0 || (myFd == myFd_)); // Can't change fd.
+ if (epollFd < 0) { // Means we're not polling.
+ epollFd = epollFd_;
+ myFd = myFd_;
+ epollCtl(EPOLL_CTL_ADD, 0);
+ }
}
-void EventHandler::epollAdd(int fd, uint32_t epollEvents, Event* event)
-{
- ZeroStruct<struct epoll_event> ee;
- ee.data.ptr = event;
- ee.events = epollEvents;
- if (epoll_ctl(epollFd, EPOLL_CTL_ADD, fd, &ee) < 0)
+void EventChannel::Descriptor::shutdown() {
+ Mutex::ScopedLock l(lock);
+ shutdownUnsafe();
+}
+
+void EventChannel::Descriptor::shutdownUnsafe() {
+ // Caller holds lock.
+ ::close(myFd);
+ epollFd = -1; // Indicate we are not polling.
+ inQueue.shutdown();
+ outQueue.shutdown();
+ epollCtl(EPOLL_CTL_DEL, 0);
+}
+
+void EventChannel::Descriptor::update() {
+ // Caller holds lock.
+ uint32_t events = EPOLLONESHOT | EPOLLERR | EPOLLHUP;
+ inQueue.setBit(events);
+ outQueue.setBit(events);
+ epollCtl(EPOLL_CTL_MOD, events);
+}
+
+void EventChannel::Descriptor::epollCtl(int op, uint32_t events) {
+ // Caller holds lock
+ assert(!isShutdown());
+ struct epoll_event ee;
+ memset(&ee, 0, sizeof(ee));
+ ee.data.ptr = this;
+ ee.events = events;
+ int status = ::epoll_ctl(epollFd, op, myFd, &ee);
+ if (status < 0)
throw QPID_POSIX_ERROR(errno);
+ }
}
+
+
+EventPair EventChannel::Descriptor::wake(uint32_t epollEvents) {
+ Mutex::ScopedLock l(lock);
+ cout << "DEBUG: " << std::hex << epollEvents << std::dec << endl;
+ // If we have an error:
+ if (epollEvents & (EPOLLERR | EPOLLHUP)) {
+ shutdownUnsafe();
+ // Complete both sides on error so the event can fail and
+ // mark itself with an exception.
+ epollEvents |= EPOLLIN | EPOLLOUT;
+ }
+ EventPair ready(inQueue.wake(epollEvents), outQueue.wake(epollEvents));
+ update();
+ return ready;
+}
+
+
+// ================================================================
+// EventChannel::Impl
-void EventHandler::epollMod(int fd, uint32_t epollEvents, Event* event)
+
+EventChannel::Impl::Impl(int epollSize):
+ epollFd(-1), dispatchQueue(0)
{
- ZeroStruct<struct epoll_event> ee;
- ee.data.ptr = event;
- ee.events = epollEvents;
- if (epoll_ctl(epollFd, EPOLL_CTL_MOD, fd, &ee) < 0)
- throw QPID_POSIX_ERROR(errno);
+ // Create the epoll file descriptor.
+ epollFd = epoll_create(epollSize);
+ QPID_POSIX_CHECK(epollFd);
+
+ // Create a pipe and write a single byte. The byte is never
+ // read so the pipes read fd is always ready for read.
+ // We activate the FD when there are messages in the queue.
+ QPID_POSIX_CHECK(::pipe(pipe));
+ static char zero = '\0';
+ QPID_POSIX_CHECK(::write(pipe[1], &zero, 1));
+ dispatchQueue = &getDescriptor(pipe[0]).getQueue(IN);
}
-void EventHandler::epollDel(int fd) {
- if (epoll_ctl(epollFd, EPOLL_CTL_DEL, fd, 0) < 0)
- throw QPID_POSIX_ERROR(errno);
+EventChannel::Impl::~Impl() {
+ close(epollFd);
+ close(pipe[0]);
+ close(pipe[1]);
}
-Event* EventHandler::complete(EventHandler& eh)
+
+/**
+ * Wait for epoll to wake up, return the descriptor or 0 on timeout.
+ */
+Event* EventChannel::Impl::wait(Time timeoutNs)
{
- assert(&eh == this);
- Event* event = mqGet();
- return event==0 ? 0 : event->complete(eh);
+ // No lock, all thread safe calls or local variables:
+ //
+ const long timeoutMs =
+ (timeoutNs == TIME_INFINITE) ? -1 : timeoutNs/TIME_MSEC;
+ struct epoll_event ee;
+ Event* event = 0;
+ bool doSwap = true;
+
+ // Loop till we get a completed event. Some events may repost
+ // themselves and return 0, e.g. incomplete read or write events.
+ //
+ while (!event) {
+ int n = epoll_wait(epollFd, &ee, 1, timeoutMs); // Thread safe.
+ if (n == 0) // Timeout
+ return 0;
+ if (n < 0 && errno != EINTR) // Interrupt, ignore it.
+ continue;
+ if (n < 0)
+ throw QPID_POSIX_ERROR(errno);
+ assert(n == 1);
+ Descriptor* ed =
+ reinterpret_cast<Descriptor*>(ee.data.ptr);
+ assert(ed);
+ EventPair ready = ed->wake(ee.events);
+
+ // We can only return one event so if both completed push one
+ // onto the dispatch queue to be dispatched in another thread.
+ if (ready.first && ready.second) {
+ // Keep it fair: in & out take turns to be returned first.
+ if (doSwap)
+ swap(ready.first, ready.second);
+ doSwap = !doSwap;
+ event = ready.first;
+ dispatchQueue->push(ready.second);
+ }
+ else {
+ event = ready.first ? ready.first : ready.second;
+ }
+ }
+ return event;
}
-
+
+EventChannel::Descriptor& EventChannel::Impl::getDescriptor(int fd) {
+ Mutex::ScopedLock l(lock);
+ Descriptor& ed = descriptors[fd];
+ ed.activate(epollFd, fd);
+ return ed;
+}
+
+
// ================================================================
// EventChannel
@@ -168,157 +367,138 @@
return shared_ptr(new EventChannel());
}
-EventChannel::EventChannel() : handler(new EventHandler()) {}
+EventChannel::EventChannel() : impl(new EventChannel::Impl()) {}
EventChannel::~EventChannel() {}
-void EventChannel::postEvent(Event& e)
+void EventChannel::post(Event& e)
{
- e.prepare(*handler);
+ e.prepare(*impl);
}
-Event* EventChannel::getEvent()
-{
- static const int infiniteTimeout = -1;
- ZeroStruct<struct epoll_event> epollEvent;
-
- // Loop until we can complete the event. Some events may re-post
- // themselves and return 0 from complete, e.g. partial reads. //
- Event* event = 0;
- while (event == 0) {
- int eventCount = epoll_wait(handler->getEpollFd(),
- &epollEvent, 1, infiniteTimeout);
- if (eventCount < 0) {
- if (errno != EINTR) {
- // TODO aconway 2006-11-28: Proper handling/logging of errors.
- cerr << BOOST_CURRENT_FUNCTION << " ignoring error "
- << PosixError::getMessage(errno) << endl;
- assert(0);
- }
- }
- else if (eventCount == 1) {
- event = reinterpret_cast<Event*>(epollEvent.data.ptr);
- assert(event != 0);
- try {
- event = event->complete(*handler);
- }
- catch (const Exception& e) {
- if (event)
- event->setError(e);
- }
- catch (const std::exception& e) {
- if (event)
- event->setError(e);
- }
- }
- }
- return event;
+void EventChannel::post(Event* e) {
+ assert(e);
+ post(*e);
}
-Event::~Event() {}
-
-void Event::prepare(EventHandler& handler)
+Event* EventChannel::wait(Time timeoutNs)
{
- handler.mqPut(this);
+ return impl->wait(timeoutNs);
}
-bool Event::hasError() const {
- return error;
-}
-void Event::throwIfError() throw (Exception) {
- if (hasError())
- error.throwSelf();
+// ================================================================
+// Event and subclasses.
+
+Event::~Event() {}
+
+Exception::shared_ptr_const Event::getException() const {
+ return exception;
}
-Event* Event::complete(EventHandler&)
-{
- return this;
+void Event::throwIfException() {
+ if (getException())
+ exception->throwSelf();
}
void Event::dispatch()
{
+ if (!callback.empty())
+ callback();
+}
+
+void Event::setException(const std::exception& e) {
+ const Exception* ex = dynamic_cast<const Exception*>(&e);
+ if (ex)
+ exception.reset(ex->clone().release());
+ else
+ exception.reset(new Exception(e));
+#ifndef NDEBUG
+ // Throw and re-catch the exception. Has no effect on the
+ // program but it triggers debuggers watching for throw. The
+ // context that sets the exception is more informative for
+ // debugging purposes than the one that ultimately throws it.
+ //
try {
- if (!callback.empty())
- callback();
- } catch (const std::exception&) {
- throw;
- } catch (...) {
- throw QPID_ERROR(INTERNAL_ERROR, "Unknown exception.");
+ throwIfException();
}
+ catch (...) { } // Ignored.
+#endif
}
-void Event::setError(const ExceptionHolder& e) {
- error = e;
+
+void ReadEvent::prepare(EventChannel::Impl& impl) {
+ impl.getDescriptor(descriptor).getQueue(IN).push(this);
}
-void ReadEvent::prepare(EventHandler& handler)
+Event* ReadEvent::complete(EventChannel::Descriptor& ed)
{
- handler.epollAdd(descriptor, EPOLLIN | EPOLLONESHOT, this);
+ ssize_t n = ::read(descriptor,
+ static_cast<char*>(buffer) + bytesRead,
+ size - bytesRead);
+
+ if (n < 0 && errno != EAGAIN) { // Error
+ setException(QPID_POSIX_ERROR(errno));
+ ed.shutdownUnsafe(); // Called with lock held.
+ }
+ else if (n == 0) { // End of file
+ // TODO aconway 2006-12-13: Don't treat EOF as exception
+ // unless we're partway thru a !noWait read.
+ setException(QPID_POSIX_ERROR(ENODATA));
+ ed.shutdownUnsafe(); // Called with lock held.
+ }
+ else {
+ if (n > 0) // possible that n < 0 && errno == EAGAIN
+ bytesRead += n;
+ if (bytesRead < size && !noWait) {
+ // Continue reading, not enough data.
+ return 0;
+ }
+ }
+ return this;
}
-ssize_t ReadEvent::doRead() {
- ssize_t n = ::read(descriptor, static_cast<char*>(buffer) + received,
- size - received);
- if (n > 0) received += n;
- return n;
+
+void WriteEvent::prepare(EventChannel::Impl& impl) {
+ impl.getDescriptor(descriptor).getQueue(OUT).push(this);
}
-Event* ReadEvent::complete(EventHandler& handler)
+
+Event* WriteEvent::complete(EventChannel::Descriptor& ed)
{
- // Read as much as possible without blocking.
- ssize_t n = doRead();
- while (n > 0 && received < size) doRead();
-
- if (received == size) {
- handler.epollDel(descriptor);
- received = 0; // Reset for re-use.
- return this;
- }
- else if (n <0 && (errno == EAGAIN)) {
- // Keep polling for more.
- handler.epollMod(descriptor, EPOLLIN | EPOLLONESHOT, this);
+ ssize_t n = ::write(descriptor,
+ static_cast<const char*>(buffer) + bytesWritten,
+ size - bytesWritten);
+ if(n < 0 && errno == EAGAIN && noWait) {
return 0;
}
- else {
- // Unexpected EOF or error. Throw ENODATA for EOF.
- handler.epollDel(descriptor);
- received = 0; // Reset for re-use.
- throw QPID_POSIX_ERROR((n < 0) ? errno : ENODATA);
+ if (n < 0 || (bytesWritten += n) < size) {
+ setException(QPID_POSIX_ERROR(errno));
+ ed.shutdownUnsafe(); // Called with lock held.
}
+ return this;
}
-void WriteEvent::prepare(EventHandler& handler)
-{
- handler.epollAdd(descriptor, EPOLLOUT | EPOLLONESHOT, this);
+void AcceptEvent::prepare(EventChannel::Impl& impl) {
+ impl.getDescriptor(descriptor).getQueue(IN).push(this);
}
-Event* WriteEvent::complete(EventHandler& handler)
+Event* AcceptEvent::complete(EventChannel::Descriptor& ed)
{
- ssize_t n = write(descriptor, static_cast<const char*>(buffer) + written,
- size - written);
- if (n < 0) throw QPID_POSIX_ERROR(errno);
- written += n;
- if(written < size) {
- // Keep polling.
- handler.epollMod(descriptor, EPOLLOUT | EPOLLONESHOT, this);
- return 0;
+ accepted = ::accept(descriptor, 0, 0);
+ if (accepted < 0) {
+ setException(QPID_POSIX_ERROR(errno));
+ ed.shutdownUnsafe(); // Called with lock held.
}
- written = 0; // Reset for re-use.
- handler.epollDel(descriptor);
return this;
}
-void AcceptEvent::prepare(EventHandler& handler)
-{
- handler.epollAdd(descriptor, EPOLLIN | EPOLLONESHOT, this);
+void DispatchEvent::prepare(EventChannel::Impl& impl) {
+ impl.getDispatchQueue().push(this);
}
-Event* AcceptEvent::complete(EventHandler& handler)
+Event* DispatchEvent::complete(EventChannel::Descriptor&)
{
- handler.epollDel(descriptor);
- accepted = ::accept(descriptor, 0, 0);
- if (accepted < 0) throw QPID_POSIX_ERROR(errno);
return this;
}
Modified: incubator/qpid/branches/event-queue-2006-12-20/cpp/lib/common/sys/posix/EventChannel.h
URL: http://svn.apache.org/viewvc/incubator/qpid/branches/event-queue-2006-12-20/cpp/lib/common/sys/posix/EventChannel.h?view=diff&rev=489159&r1=489158&r2=489159
==============================================================================
--- incubator/qpid/branches/event-queue-2006-12-20/cpp/lib/common/sys/posix/EventChannel.h (original)
+++ incubator/qpid/branches/event-queue-2006-12-20/cpp/lib/common/sys/posix/EventChannel.h Wed Dec 20 09:54:05 2006
@@ -19,8 +19,11 @@
*
*/
-#include <SharedObject.h>
-#include <ExceptionHolder.h>
+#include "SharedObject.h"
+#include "Exception.h"
+#include "sys/Monitor.h"
+#include "sys/Time.h"
+
#include <boost/function.hpp>
#include <memory>
@@ -28,11 +31,47 @@
namespace sys {
class Event;
-class EventHandler;
-class EventChannel;
+
+/**
+ * Channel to post and wait for events.
+ */
+class EventChannel : public qpid::SharedObject<EventChannel>
+{
+ public:
+ static shared_ptr create();
+
+ ~EventChannel();
+
+ /** Post an event to the channel. */
+ void post(Event& event);
+
+ /** Post an event to the channel. Must not be 0. */
+ void post(Event* event);
+
+ /**
+ * Wait for the next complete event, up to timeout.
+ *@return Pointer to event or 0 if timeout elapses.
+ */
+ Event* wait(Time timeout = TIME_INFINITE);
+
+ class Impl;
+ class Queue;
+ class Descriptor;
+
+ private:
+
+ EventChannel();
+
+ Mutex lock;
+ boost::shared_ptr<Impl> impl;
+};
/**
* Base class for all Events.
+ * Derived classes define events representing various async IO operations.
+ * When an event is complete, it is returned by the EventChannel to
+ * a thread calling wait. The thread will call Event::dispatch() to
+ * execute code associated with event completion.
*/
class Event
{
@@ -40,132 +79,134 @@
/** Type for callback when event is dispatched */
typedef boost::function0<void> Callback;
- /**
- * Create an event with optional callback.
- * Instances of Event are sent directly through the channel.
- * Derived classes define additional waiting behaviour.
- *@param cb A callback functor that is invoked when dispatch() is called.
- */
- Event(Callback cb = 0) : callback(cb) {}
-
virtual ~Event();
/** Call the callback provided to the constructor, if any. */
void dispatch();
- /** True if there was an error processing this event */
- bool hasError() const;
+ /**
+ *If there was an exception processing this Event, return it.
+ *@return 0 if there was no exception. Caller must not delete.
+ */
+ qpid::Exception::shared_ptr_const getException() const;
+
+ /** If getException() throw the corresponding exception. */
+ void throwIfException();
+
+ /** Set the dispatch callback. */
+ void setCallback(Callback cb) { callback = cb; }
- /** If hasError() throw the corresponding exception. */
- void throwIfError() throw(Exception);
+ /** Set the exception. */
+ void setException(const std::exception& e);
protected:
- virtual void prepare(EventHandler&);
- virtual Event* complete(EventHandler&);
- void setError(const ExceptionHolder& e);
+ Event(Callback cb=0) : callback(cb) {}
+
+ virtual void prepare(EventChannel::Impl&) = 0;
+ virtual Event* complete(EventChannel::Descriptor&) = 0;
Callback callback;
- ExceptionHolder error;
+ Exception::shared_ptr_const exception;
friend class EventChannel;
- friend class EventHandler;
+ friend class EventChannel::Queue;
};
-template <class BufT>
-class IOEvent : public Event {
+/**
+ * An event that does not wait for anything, it is processed
+ * immediately by one of the channel threads.
+ */
+class DispatchEvent : public Event {
public:
- void getDescriptor() const { return descriptor; }
- size_t getSize() const { return size; }
- BufT getBuffer() const { return buffer; }
-
+ DispatchEvent(Callback cb=0) : Event(cb) {}
+
protected:
- IOEvent(int fd, Callback cb, size_t sz, BufT buf) :
- Event(cb), descriptor(fd), buffer(buf), size(sz) {}
+ void prepare(EventChannel::Impl&);
+ Event* complete(EventChannel::Descriptor&);
+};
+
+// Utility base class.
+class FDEvent : public Event {
+ public:
+ int getDescriptor() const { return descriptor; }
+ protected:
+ FDEvent(Callback cb = 0, int fd = 0)
+ : Event(cb), descriptor(fd) {}
int descriptor;
- BufT buffer;
- size_t size;
};
+// Utility base class
+class IOEvent : public FDEvent {
+ public:
+ size_t getSize() const { return size; }
+
+ protected:
+ IOEvent(Callback cb, int fd, size_t sz, bool noWait_) :
+ FDEvent(cb, fd), size(sz), noWait(noWait_) {}
+
+ size_t size;
+ bool noWait;
+};
+
/** Asynchronous read event */
-class ReadEvent : public IOEvent<void*>
+class ReadEvent : public IOEvent
{
public:
- explicit ReadEvent(int fd=-1, void* buf=0, size_t sz=0, Callback cb=0) :
- IOEvent<void*>(fd, cb, sz, buf), received(0) {}
+ explicit ReadEvent(
+ int fd=-1, void* buf=0, size_t sz=0,
+ Callback cb=0, bool noWait=false
+ ) : IOEvent(cb, fd, sz, noWait), buffer(buf), bytesRead(0) {}
+ void* getBuffer() const { return buffer; }
+ size_t getBytesRead() const { return bytesRead; }
+
private:
- void prepare(EventHandler&);
- Event* complete(EventHandler&);
+ void prepare(EventChannel::Impl&);
+ Event* complete(EventChannel::Descriptor&);
ssize_t doRead();
- size_t received;
+ void* buffer;
+ size_t bytesRead;
};
/** Asynchronous write event */
-class WriteEvent : public IOEvent<const void*>
+class WriteEvent : public IOEvent
{
public:
explicit WriteEvent(int fd=-1, const void* buf=0, size_t sz=0,
Callback cb=0) :
- IOEvent<const void*>(fd, cb, sz, buf), written(0) {}
+ IOEvent(cb, fd, sz, noWait), buffer(buf), bytesWritten(0) {}
- protected:
- void prepare(EventHandler&);
- Event* complete(EventHandler&);
+ const void* getBuffer() const { return buffer; }
+ size_t getBytesWritten() const { return bytesWritten; }
private:
+ void prepare(EventChannel::Impl&);
+ Event* complete(EventChannel::Descriptor&);
ssize_t doWrite();
- size_t written;
+
+ const void* buffer;
+ size_t bytesWritten;
};
+
/** Asynchronous socket accept event */
-class AcceptEvent : public Event
+class AcceptEvent : public FDEvent
{
public:
/** Accept a connection on fd. */
explicit AcceptEvent(int fd=-1, Callback cb=0) :
- Event(cb), descriptor(fd), accepted(0) {}
-
- /** Get descriptor for server socket */
+ FDEvent(cb, fd), accepted(0) {}
+
+ /** Get descriptor for accepted server socket */
int getAcceptedDesscriptor() const { return accepted; }
private:
- void prepare(EventHandler&);
- Event* complete(EventHandler&);
+ void prepare(EventChannel::Impl&);
+ Event* complete(EventChannel::Descriptor&);
- int descriptor;
int accepted;
-};
-
-
-class QueueSet;
-
-/**
- * Channel to post and wait for events.
- */
-class EventChannel : public qpid::SharedObject<EventChannel>
-{
- public:
- static shared_ptr create();
-
- ~EventChannel();
-
- /** Post an event to the channel. */
- void postEvent(Event& event);
-
- /** Post an event to the channel. Must not be 0. */
- void postEvent(Event* event) { postEvent(*event); }
-
- /**
- * Wait for the next complete event.
- *@return Pointer to event. Will never return 0.
- */
- Event* getEvent();
-
- private:
- EventChannel();
- boost::shared_ptr<EventHandler> handler;
};
Modified: incubator/qpid/branches/event-queue-2006-12-20/cpp/lib/common/sys/posix/EventChannelAcceptor.cpp
URL: http://svn.apache.org/viewvc/incubator/qpid/branches/event-queue-2006-12-20/cpp/lib/common/sys/posix/EventChannelAcceptor.cpp?view=diff&rev=489159&r1=489158&r2=489159
==============================================================================
--- incubator/qpid/branches/event-queue-2006-12-20/cpp/lib/common/sys/posix/EventChannelAcceptor.cpp (original)
+++ incubator/qpid/branches/event-queue-2006-12-20/cpp/lib/common/sys/posix/EventChannelAcceptor.cpp Wed Dec 20 09:54:05 2006
@@ -139,11 +139,11 @@
shutdown();
return;
}
- // TODO aconway 2006-11-29: Need to reap closed connections also.
int fd = acceptEvent.getAcceptedDesscriptor();
+ threads->post(acceptEvent); // Keep accepting.
+ // TODO aconway 2006-11-29: Need to reap closed connections also.
connections.push_back(
new EventChannelConnection(threads, *factory, fd, fd, isTrace));
- threads->post(acceptEvent); // Keep accepting.
}
}} // namespace qpid::sys
Modified: incubator/qpid/branches/event-queue-2006-12-20/cpp/lib/common/sys/posix/EventChannelThreads.cpp
URL: http://svn.apache.org/viewvc/incubator/qpid/branches/event-queue-2006-12-20/cpp/lib/common/sys/posix/EventChannelThreads.cpp?view=diff&rev=489159&r1=489158&r2=489159
==============================================================================
--- incubator/qpid/branches/event-queue-2006-12-20/cpp/lib/common/sys/posix/EventChannelThreads.cpp (original)
+++ incubator/qpid/branches/event-queue-2006-12-20/cpp/lib/common/sys/posix/EventChannelThreads.cpp Wed Dec 20 09:54:05 2006
@@ -16,26 +16,40 @@
*
*/
-#include "EventChannelThreads.h"
-#include <sys/Runnable.h>
#include <iostream>
-using namespace std;
+#include <limits>
+
#include <boost/bind.hpp>
+#include <sys/Runnable.h>
+
+#include "EventChannelThreads.h"
+
namespace qpid {
namespace sys {
+const size_t EventChannelThreads::unlimited =
+ std::numeric_limits<size_t>::max();
+
EventChannelThreads::shared_ptr EventChannelThreads::create(
- EventChannel::shared_ptr ec)
+ EventChannel::shared_ptr ec, size_t min, size_t max
+)
{
- return EventChannelThreads::shared_ptr(new EventChannelThreads(ec));
+ return EventChannelThreads::shared_ptr(
+ new EventChannelThreads(ec, min, max));
}
-EventChannelThreads::EventChannelThreads(EventChannel::shared_ptr ec) :
- channel(ec), nWaiting(0), state(RUNNING)
+EventChannelThreads::EventChannelThreads(
+ EventChannel::shared_ptr ec, size_t min, size_t max) :
+ minThreads(std::max(size_t(1), min)),
+ maxThreads(std::min(min, max)),
+ channel(ec),
+ nWaiting(0),
+ state(RUNNING)
{
- // TODO aconway 2006-11-15: Estimate initial threads based on CPUs.
- addThread();
+ Monitor::ScopedLock l(monitor);
+ while (workers.size() < minThreads)
+ workers.push_back(Thread(*this));
}
EventChannelThreads::~EventChannelThreads() {
@@ -43,34 +57,37 @@
join();
}
+// Termination marker event.
+static DispatchEvent terminate;
+
void EventChannelThreads::shutdown()
{
- ScopedLock lock(*this);
+ Monitor::ScopedLock lock(monitor);
if (state != RUNNING) // Already shutting down.
return;
+ state = TERMINATING;
for (size_t i = 0; i < workers.size(); ++i) {
- channel->postEvent(terminate);
+ channel->post(terminate);
}
- state = TERMINATE_SENT;
- notify(); // Wake up one join() thread.
+ monitor.notify(); // Wake up one join() thread.
}
void EventChannelThreads::join()
{
{
- ScopedLock lock(*this);
+ Monitor::ScopedLock lock(monitor);
while (state == RUNNING) // Wait for shutdown to start.
- wait();
+ monitor.wait();
if (state == SHUTDOWN) // Shutdown is complete
return;
if (state == JOINING) {
// Someone else is doing the join.
while (state != SHUTDOWN)
- wait();
+ monitor.wait();
return;
}
// I'm the joining thread
- assert(state == TERMINATE_SENT);
+ assert(state == TERMINATING);
state = JOINING;
} // Drop the lock.
@@ -79,12 +96,13 @@
workers[i].join();
}
state = SHUTDOWN;
- notifyAll(); // Notify other join() threaeds.
+ monitor.notifyAll(); // Notify any other join() threads.
}
void EventChannelThreads::addThread() {
- ScopedLock l(*this);
- workers.push_back(Thread(*this));
+ Monitor::ScopedLock l(monitor);
+ if (workers.size() < maxThreads)
+ workers.push_back(Thread(*this));
}
void EventChannelThreads::run()
@@ -93,26 +111,22 @@
AtomicCount::ScopedIncrement inc(nWaiting);
try {
while (true) {
- Event* e = channel->getEvent();
+ Event* e = channel->wait();
assert(e != 0);
- if (e == &terminate) {
+ if (e == &terminate)
return;
- }
AtomicCount::ScopedDecrement dec(nWaiting);
- // I'm no longer waiting, make sure someone is.
- if (dec == 0)
+ // Make sure there's at least one waiting thread.
+ if (dec == 0 && state == RUNNING)
addThread();
e->dispatch();
}
}
catch (const std::exception& e) {
- // TODO aconway 2006-11-15: need better logging across the board.
- std::cerr << "EventChannelThreads::run() caught: " << e.what()
- << std::endl;
+ Exception::log(e, "Exception in EventChannelThreads::run()");
}
catch (...) {
- std::cerr << "EventChannelThreads::run() caught unknown exception."
- << std::endl;
+ Exception::logUnknown("Exception in EventChannelThreads::run()");
}
}
Modified: incubator/qpid/branches/event-queue-2006-12-20/cpp/lib/common/sys/posix/EventChannelThreads.h
URL: http://svn.apache.org/viewvc/incubator/qpid/branches/event-queue-2006-12-20/cpp/lib/common/sys/posix/EventChannelThreads.h?view=diff&rev=489159&r1=489158&r2=489159
==============================================================================
--- incubator/qpid/branches/event-queue-2006-12-20/cpp/lib/common/sys/posix/EventChannelThreads.h (original)
+++ incubator/qpid/branches/event-queue-2006-12-20/cpp/lib/common/sys/posix/EventChannelThreads.h Wed Dec 20 09:54:05 2006
@@ -20,11 +20,12 @@
*/
#include <vector>
-#include <Exception.h>
-#include <sys/Time.h>
-#include <sys/Monitor.h>
-#include <sys/Thread.h>
-#include <sys/AtomicCount.h>
+#include "Exception.h"
+#include "sys/AtomicCount.h"
+#include "sys/Monitor.h"
+#include "sys/Thread.h"
+#include "sys/Time.h"
+
#include "EventChannel.h"
namespace qpid {
@@ -33,26 +34,36 @@
/**
Dynamic thread pool serving an EventChannel.
- Threads run a loop { e = getEvent(); e->dispatch(); }
+ Threads run a loop { e = wait(); e->dispatch(); }
The size of the thread pool is automatically adjusted to optimal size.
*/
class EventChannelThreads :
public qpid::SharedObject<EventChannelThreads>,
- public sys::Monitor, private sys::Runnable
+ private sys::Runnable
{
public:
- /** Create the thread pool and start initial threads. */
+ /** Constant to represent an unlimited number of threads */
+ static const size_t unlimited;
+
+ /**
+ * Create the thread pool and start initial threads.
+ * @param minThreads Pool will initialy contain minThreads threads and
+ * will never shrink to less until shutdown.
+ * @param maxThreads Pool will never grow to more than maxThreads.
+ */
static EventChannelThreads::shared_ptr create(
- EventChannel::shared_ptr channel
+ EventChannel::shared_ptr channel = EventChannel::create(),
+ size_t minThreads = 1,
+ size_t maxThreads = unlimited
);
~EventChannelThreads();
/** Post event to the underlying channel */
- void postEvent(Event& event) { channel->postEvent(event); }
+ void post(Event& event) { channel->post(event); }
/** Post event to the underlying channel Must not be 0. */
- void postEvent(Event* event) { channel->postEvent(event); }
+ void post(Event* event) { channel->post(event); }
/**
* Terminate all threads.
@@ -68,21 +79,25 @@
private:
typedef std::vector<sys::Thread> Threads;
typedef enum {
- RUNNING, TERMINATE_SENT, JOINING, SHUTDOWN
+ RUNNING, TERMINATING, JOINING, SHUTDOWN
} State;
- EventChannelThreads(EventChannel::shared_ptr underlyingChannel);
+ EventChannelThreads(
+ EventChannel::shared_ptr channel, size_t min, size_t max);
+
void addThread();
void run();
bool keepRunning();
void adjustThreads();
+ Monitor monitor;
+ size_t minThreads;
+ size_t maxThreads;
EventChannel::shared_ptr channel;
Threads workers;
sys::AtomicCount nWaiting;
State state;
- Event terminate;
};
Modified: incubator/qpid/branches/event-queue-2006-12-20/cpp/lib/common/sys/posix/PosixAcceptor.cpp
URL: http://svn.apache.org/viewvc/incubator/qpid/branches/event-queue-2006-12-20/cpp/lib/common/sys/posix/PosixAcceptor.cpp?view=diff&rev=489159&r1=489158&r2=489159
==============================================================================
--- incubator/qpid/branches/event-queue-2006-12-20/cpp/lib/common/sys/posix/PosixAcceptor.cpp (original)
+++ incubator/qpid/branches/event-queue-2006-12-20/cpp/lib/common/sys/posix/PosixAcceptor.cpp Wed Dec 20 09:54:05 2006
@@ -1,48 +0,0 @@
-/*
- *
- * 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 <sys/Acceptor.h>
-#include <Exception.h>
-
-namespace qpid {
-namespace sys {
-
-namespace {
-void fail() { throw qpid::Exception("PosixAcceptor not implemented"); }
-}
-
-class PosixAcceptor : public Acceptor {
- public:
- virtual int16_t getPort() const { fail(); return 0; }
- virtual void run(qpid::sys::SessionHandlerFactory* ) { fail(); }
- virtual void shutdown() { fail(); }
-};
-
-// Define generic Acceptor::create() to return APRAcceptor.
- Acceptor::shared_ptr Acceptor::create(int16_t , int, int, bool)
-{
- return Acceptor::shared_ptr(new PosixAcceptor());
-}
-
-// Must define Acceptor virtual dtor.
-Acceptor::~Acceptor() {}
-
-}}
Modified: incubator/qpid/branches/event-queue-2006-12-20/cpp/lib/common/sys/posix/Socket.cpp
URL: http://svn.apache.org/viewvc/incubator/qpid/branches/event-queue-2006-12-20/cpp/lib/common/sys/posix/Socket.cpp?view=diff&rev=489159&r1=489158&r2=489159
==============================================================================
--- incubator/qpid/branches/event-queue-2006-12-20/cpp/lib/common/sys/posix/Socket.cpp (original)
+++ incubator/qpid/branches/event-queue-2006-12-20/cpp/lib/common/sys/posix/Socket.cpp Wed Dec 20 09:54:05 2006
@@ -96,6 +96,8 @@
int Socket::listen(int port, int backlog)
{
struct sockaddr_in name;
+ static const int ON = 1;
+ setsockopt(socket, SOL_SOCKET, SO_REUSEADDR, &ON, sizeof(ON));
name.sin_family = AF_INET;
name.sin_port = htons(port);
name.sin_addr.s_addr = 0;
@@ -111,8 +113,15 @@
return ntohs(name.sin_port);
}
+Socket Socket::accept() {
+ int accepted = ::accept(socket, 0, 0);
+ if (accepted < 0)
+ throw (QPID_POSIX_ERROR(errno));
+ return Socket(accepted);
+}
+
-int Socket::fd()
+int Socket::fd()const
{
return socket;
}