You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@qpid.apache.org by as...@apache.org on 2010/10/12 18:04:48 UTC
svn commit: r1021819 - in /qpid/trunk/qpid/cpp/src/qpid:
client/RdmaConnector.cpp sys/rdma/RdmaIO.cpp sys/rdma/RdmaIO.h
Author: astitcher
Date: Tue Oct 12 16:04:47 2010
New Revision: 1021819
URL: http://svn.apache.org/viewvc?rev=1021819&view=rev
Log:
Serialise close into the data callbacks:
Rejig Rdma::ConnectionManager to have a stop function with a callback and
use this to ensure that the Rdma::Connector used by qpid::sys::RdmaConnector
is correctly deleted only after it has been actually stopped
Modified:
qpid/trunk/qpid/cpp/src/qpid/client/RdmaConnector.cpp
qpid/trunk/qpid/cpp/src/qpid/sys/rdma/RdmaIO.cpp
qpid/trunk/qpid/cpp/src/qpid/sys/rdma/RdmaIO.h
Modified: qpid/trunk/qpid/cpp/src/qpid/client/RdmaConnector.cpp
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/qpid/client/RdmaConnector.cpp?rev=1021819&r1=1021818&r2=1021819&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/qpid/client/RdmaConnector.cpp (original)
+++ qpid/trunk/qpid/cpp/src/qpid/client/RdmaConnector.cpp Tue Oct 12 16:04:47 2010
@@ -73,7 +73,7 @@ class RdmaConnector : public Connector,
framing::OutputHandler* output;
Rdma::AsynchIO* aio;
- std::auto_ptr<Rdma::Connector> acon;
+ Rdma::Connector* acon;
sys::Poller::shared_ptr poller;
std::auto_ptr<qpid::sys::SecurityLayer> securityLayer;
@@ -82,7 +82,6 @@ class RdmaConnector : public Connector,
// Callbacks
void connected(sys::Poller::shared_ptr, Rdma::Connection::intrusive_ptr, const Rdma::ConnectionParams&);
void connectionError(sys::Poller::shared_ptr, Rdma::Connection::intrusive_ptr, Rdma::ErrorType);
- void disconnectAction();
void disconnected();
void rejected(sys::Poller::shared_ptr, Rdma::Connection::intrusive_ptr, const Rdma::ConnectionParams&);
@@ -91,7 +90,8 @@ class RdmaConnector : public Connector,
void writeDataBlock(const framing::AMQDataBlock& data);
void dataError(Rdma::AsynchIO&);
void drained();
- void stopped(Rdma::AsynchIO* aio=0);
+ void connectionStopped(Rdma::Connector* acon);
+ void dataStopped(Rdma::AsynchIO* aio);
std::string identifier;
@@ -147,6 +147,7 @@ RdmaConnector::RdmaConnector(Poller::sha
polling(false),
shutdownHandler(0),
aio(0),
+ acon(0),
poller(p)
{
QPID_LOG(debug, "RdmaConnector created for " << version);
@@ -156,14 +157,20 @@ namespace {
void deleteAsynchIO(Rdma::AsynchIO& aio) {
delete &aio;
}
+
+ void deleteConnector(Rdma::ConnectionManager& con) {
+ delete &con;
+ }
}
RdmaConnector::~RdmaConnector() {
QPID_LOG(debug, "~RdmaConnector " << identifier);
- close();
if (aio) {
aio->stop(deleteAsynchIO);
}
+ if (acon) {
+ acon->stop(deleteConnector);
+ }
if (shutdownHandler) {
shutdownHandler->shutdown();
}
@@ -173,12 +180,12 @@ void RdmaConnector::connect(const std::s
Mutex::ScopedLock l(pollingLock);
assert(!polling);
- acon.reset(new Rdma::Connector(
+ acon = new Rdma::Connector(
Rdma::ConnectionParams(maxFrameSize, Rdma::DEFAULT_WR_ENTRIES),
boost::bind(&RdmaConnector::connected, this, poller, _1, _2),
boost::bind(&RdmaConnector::connectionError, this, poller, _1, _2),
boost::bind(&RdmaConnector::disconnected, this),
- boost::bind(&RdmaConnector::rejected, this, poller, _1, _2)));
+ boost::bind(&RdmaConnector::rejected, this, poller, _1, _2));
polling = true;
@@ -211,11 +218,10 @@ void RdmaConnector::connected(Poller::sh
}
{
Mutex::ScopedLock l(pollingLock);
- // If we're closed already then we'll get to stopped() anyway
- if (!polling) return;
+ assert(polling);
polling = false;
}
- stopped();
+ connectionStopped(acon);
}
void RdmaConnector::connectionError(sys::Poller::shared_ptr, Rdma::Connection::intrusive_ptr, Rdma::ErrorType) {
@@ -226,10 +232,10 @@ void RdmaConnector::connectionError(sys:
if (!polling) return;
polling = false;
}
- stopped();
+ connectionStopped(acon);
}
-void RdmaConnector::disconnectAction() {
+void RdmaConnector::disconnected() {
QPID_LOG(debug, "Connection disconnected " << identifier);
{
Mutex::ScopedLock l(pollingLock);
@@ -237,11 +243,8 @@ void RdmaConnector::disconnectAction() {
if (!polling) return;
polling = false;
}
- drained();
-}
-
-void RdmaConnector::disconnected() {
- aio->requestCallback(boost::bind(&RdmaConnector::disconnectAction, this));
+ // Make sure that all the disconnected actions take place on the data "thread"
+ aio->requestCallback(boost::bind(&RdmaConnector::drained, this));
}
void RdmaConnector::rejected(sys::Poller::shared_ptr, Rdma::Connection::intrusive_ptr, const Rdma::ConnectionParams& cp) {
@@ -252,7 +255,7 @@ void RdmaConnector::rejected(sys::Poller
if (!polling) return;
polling = false;
}
- stopped();
+ connectionStopped(acon);
}
void RdmaConnector::dataError(Rdma::AsynchIO&) {
@@ -266,35 +269,48 @@ void RdmaConnector::dataError(Rdma::Asyn
drained();
}
-void RdmaConnector::stopped(Rdma::AsynchIO* a) {
- QPID_LOG(debug, "RdmaConnector::stopped " << identifier);
- assert(!polling);
- aio = 0;
- delete a;
- if (shutdownHandler) {
- ShutdownHandler* s = shutdownHandler;
- shutdownHandler = 0;
- s->shutdown();
+void RdmaConnector::close() {
+ QPID_LOG(debug, "RdmaConnector::close " << identifier);
+ {
+ Mutex::ScopedLock l(pollingLock);
+ if (!polling) return;
+ polling = false;
}
+ if (aio) aio->drainWriteQueue(boost::bind(&RdmaConnector::drained, this));
}
void RdmaConnector::drained() {
QPID_LOG(debug, "RdmaConnector::drained " << identifier);
assert(!polling);
- acon->stop();
if (aio) {
Rdma::AsynchIO* a = aio;
aio = 0;
- a->stop(boost::bind(&RdmaConnector::stopped, this, a));
+ a->stop(boost::bind(&RdmaConnector::dataStopped, this, a));
}
}
-void RdmaConnector::close() {
- QPID_LOG(debug, "RdmaConnector::close " << identifier);
- Mutex::ScopedLock l(pollingLock);
- if (!polling) return;
- polling = false;
- if (aio) aio->drainWriteQueue(boost::bind(&RdmaConnector::drained, this));
+void RdmaConnector::dataStopped(Rdma::AsynchIO* a) {
+ QPID_LOG(debug, "RdmaConnector::dataStopped " << identifier);
+ assert(!polling);
+ aio = 0;
+ delete a;
+ if (acon) {
+ Rdma::Connector* c = acon;
+ acon = 0;
+ c->stop(boost::bind(&RdmaConnector::connectionStopped, this, c));
+ }
+}
+
+void RdmaConnector::connectionStopped(Rdma::Connector* c) {
+ QPID_LOG(debug, "RdmaConnector::connectionStopped " << identifier);
+ assert(!polling);
+ acon = 0;
+ delete c;
+ if (shutdownHandler) {
+ ShutdownHandler* s = shutdownHandler;
+ shutdownHandler = 0;
+ s->shutdown();
+ }
}
void RdmaConnector::setInputHandler(InputHandler* handler){
Modified: qpid/trunk/qpid/cpp/src/qpid/sys/rdma/RdmaIO.cpp
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/qpid/sys/rdma/RdmaIO.cpp?rev=1021819&r1=1021818&r2=1021819&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/qpid/sys/rdma/RdmaIO.cpp (original)
+++ qpid/trunk/qpid/cpp/src/qpid/sys/rdma/RdmaIO.cpp Tue Oct 12 16:04:47 2010
@@ -518,6 +518,7 @@ namespace Rdma {
ErrorCallback errc,
DisconnectedCallback dc
) :
+ state(IDLE),
ci(Connection::make()),
handle(*ci, boost::bind(&ConnectionManager::event, this, _1), 0, 0),
errorCallback(errc),
@@ -537,11 +538,23 @@ namespace Rdma {
handle.startWatch(poller);
}
- void ConnectionManager::stop() {
+ void ConnectionManager::doStoppedCallback() {
+ // Ensure we can't get any more callbacks (except for the stopped callback)
handle.stopWatch();
+
+ NotifyCallback nc;
+ nc.swap(notifyCallback);
+ nc(*this);
+ }
+
+ void ConnectionManager::stop(NotifyCallback nc) {
+ state = STOPPED;
+ notifyCallback = nc;
+ handle.call(boost::bind(&ConnectionManager::doStoppedCallback, this));
}
void ConnectionManager::event(DispatchHandle&) {
+ if (state.get() == STOPPED) return;
connectionEvent(ci);
}
Modified: qpid/trunk/qpid/cpp/src/qpid/sys/rdma/RdmaIO.h
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/qpid/sys/rdma/RdmaIO.h?rev=1021819&r1=1021818&r2=1021819&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/qpid/sys/rdma/RdmaIO.h (original)
+++ qpid/trunk/qpid/cpp/src/qpid/sys/rdma/RdmaIO.h Tue Oct 12 16:04:47 2010
@@ -26,7 +26,6 @@
#include "qpid/sys/AtomicValue.h"
#include "qpid/sys/Dispatcher.h"
#include "qpid/sys/DispatchHandle.h"
-#include "qpid/sys/Mutex.h"
#include "qpid/sys/SocketAddress.h"
#include <netinet/in.h>
@@ -163,14 +162,19 @@ namespace Rdma {
typedef boost::function1<void, Rdma::Connection::intrusive_ptr> DisconnectedCallback;
class ConnectionManager {
+ typedef boost::function1<void, ConnectionManager&> NotifyCallback;
+
+ enum State {IDLE, STOPPED};
+ qpid::sys::AtomicValue<State> state;
Connection::intrusive_ptr ci;
qpid::sys::DispatchHandleRef handle;
+ NotifyCallback notifyCallback;
protected:
ErrorCallback errorCallback;
DisconnectedCallback disconnectedCallback;
- public:
+ public:
ConnectionManager(
ErrorCallback errc,
DisconnectedCallback dc
@@ -179,10 +183,11 @@ namespace Rdma {
virtual ~ConnectionManager();
void start(qpid::sys::Poller::shared_ptr poller, const qpid::sys::SocketAddress& addr);
- void stop();
+ void stop(NotifyCallback);
private:
void event(qpid::sys::DispatchHandle& handle);
+ void doStoppedCallback();
virtual void startConnection(Connection::intrusive_ptr ci, const qpid::sys::SocketAddress& addr) = 0;
virtual void connectionEvent(Connection::intrusive_ptr ci) = 0;
---------------------------------------------------------------------
Apache Qpid - AMQP Messaging Implementation
Project: http://qpid.apache.org
Use/Interact: mailto:commits-subscribe@qpid.apache.org