Author: aconway
Date: Mon Feb 13 16:18:46 2012
New Revision: 1243584
URL: http://svn.apache.org/viewvc?rev=1243584&view=rev
Log:
QPID-3603: Fix core dump in Link::requestIOProcessing.
Core dump occuring when a link was closed before being completely opened.
- Merge Link::established and setConnection into one atomic function.
- Moved logic that needs to be executed in a connection thread from ~Link to
Link::destroy
Link::destroy is always called in connection thread, ~Link can be called
later if
another thread is holding a reference.
- Added some asserts to verify functioning as expected.
Modified:
qpid/branches/qpid-3603-2/qpid/cpp/src/qpid/broker/Link.cpp
qpid/branches/qpid-3603-2/qpid/cpp/src/qpid/broker/Link.h
qpid/branches/qpid-3603-2/qpid/cpp/src/qpid/broker/LinkRegistry.cpp
Modified: qpid/branches/qpid-3603-2/qpid/cpp/src/qpid/broker/Link.cpp
URL:
http://svn.apache.org/viewvc/qpid/branches/qpid-3603-2/qpid/cpp/src/qpid/broker/Link.cpp?rev=1243584&r1=1243583&r2=1243584&view=diff
==============================================================================
--- qpid/branches/qpid-3603-2/qpid/cpp/src/qpid/broker/Link.cpp (original)
+++ qpid/branches/qpid-3603-2/qpid/cpp/src/qpid/broker/Link.cpp Mon Feb 13
16:18:46 2012
@@ -103,10 +103,8 @@ Link::Link(LinkRegistry* _links,
Link::~Link ()
{
- timerTask->cancel();
- if (state == STATE_OPERATIONAL && connection != 0)
- connection->close(CLOSE_CODE_CONNECTION_FORCED, "closed by
management");
-
+ assert(state == STATE_CLOSED); // Can only get here after destroy()
+ assert(connection == 0);
if (mgmtObject != 0)
mgmtObject->resourceDestroy ();
}
@@ -134,6 +132,7 @@ void Link::setStateLH (int newState)
void Link::startConnectionLH ()
{
+ assert(state == STATE_WAITING);
try {
// Set the state before calling connect. It is possible that connect
// will fail synchronously and call Link::closed before returning.
@@ -150,7 +149,7 @@ void Link::startConnectionLH ()
}
}
-void Link::established ()
+void Link::established(Connection* c)
{
stringstream addr;
addr << host << ":" << port;
@@ -159,16 +158,19 @@ void Link::established ()
if (!hideManagement() && agent)
agent->raiseEvent(_qmf::EventBrokerLinkUp(addr.str()));
- {
- Mutex::ScopedLock mutex(lock);
- setStateLH(STATE_OPERATIONAL);
- currentInterval = 1;
- visitCount = 0;
- if (closing)
- destroy();
- }
+ Mutex::ScopedLock mutex(lock);
+ assert(state == STATE_CONNECTING);
+ setStateLH(STATE_OPERATIONAL);
+ currentInterval = 1;
+ visitCount = 0;
+ connection = c;
+ if (closing)
+ destroy();
+ else // Process any IO tasks bridges added before established.
+ connection->requestIOProcessing
(boost::bind(&Link::ioThreadProcessing, this));
}
+
void Link::setUrl(const Url& u) {
Mutex::ScopedLock mutex(lock);
url = u;
@@ -222,6 +224,7 @@ void Link::closed(int, std::string text)
destroy();
}
+// Called in connection IO thread.
void Link::destroy ()
{
Bridges toDelete;
@@ -231,7 +234,7 @@ void Link::destroy ()
QPID_LOG (info, "Inter-broker link to " << host << ":" << port << "
removed by management");
if (connection)
connection->close(CLOSE_CODE_CONNECTION_FORCED, "closed by
management");
-
+ connection = 0;
setStateLH(STATE_CLOSED);
// Move the bridges to be deleted into a local vector so there is no
@@ -245,6 +248,8 @@ void Link::destroy ()
for (Bridges::iterator i = created.begin(); i != created.end(); i++)
toDelete.push_back(*i);
created.clear();
+
+ timerTask->cancel();
}
// Now delete all bridges on this link (don't hold the lock for this).
for (Bridges::iterator i = toDelete.begin(); i != toDelete.end(); i++)
@@ -284,7 +289,7 @@ void Link::cancel(Bridge::shared_ptr bri
}
needIOProcessing = !cancellations.empty();
}
- if (needIOProcessing)
+ if (needIOProcessing && connection)
connection->requestIOProcessing
(boost::bind(&Link::ioThreadProcessing, this));
}
@@ -326,14 +331,6 @@ void Link::ioThreadProcessing()
}
}
-void Link::setConnection(Connection* c)
-{
- Mutex::ScopedLock mutex(lock);
- connection = c;
- // Process any IO tasks bridges added before setConnection.
- connection->requestIOProcessing (boost::bind(&Link::ioThreadProcessing,
this));
-}
-
void Link::maintenanceVisit ()
{
Mutex::ScopedLock mutex(lock);
@@ -357,9 +354,8 @@ void Link::maintenanceVisit ()
connection->requestIOProcessing
(boost::bind(&Link::ioThreadProcessing, this));
}
-void Link::reconnect(const Address& a)
+void Link::reconnectLH(const Address& a)
{
- Mutex::ScopedLock mutex(lock);
host = a.host;
port = a.port;
transport = a.protocol;
@@ -378,7 +374,7 @@ bool Link::tryFailoverLH() {
if (next.host != host || next.port != port || next.protocol != transport) {
links->changeAddress(Address(transport, host, port), next);
QPID_LOG(debug, "Inter-broker link failing over to " << next.host <<
":" << next.port);
- reconnect(next);
+ reconnectLH(next);
return true;
}
return false;
Modified: qpid/branches/qpid-3603-2/qpid/cpp/src/qpid/broker/Link.h
URL:
http://svn.apache.org/viewvc/qpid/branches/qpid-3603-2/qpid/cpp/src/qpid/broker/Link.h?rev=1243584&r1=1243583&r2=1243584&view=diff
==============================================================================
--- qpid/branches/qpid-3603-2/qpid/cpp/src/qpid/broker/Link.h (original)
+++ qpid/branches/qpid-3603-2/qpid/cpp/src/qpid/broker/Link.h Mon Feb 13
16:18:46 2012
@@ -121,11 +121,10 @@ class Link : public PersistableConfig, p
void cancel(Bridge::shared_ptr);
void setUrl(const Url&); // Set URL for reconnection.
- void established(); // Called when connection is create
+ void established(Connection*); // Called when connection is create
void opened(); // Called when connection is open (after create)
void closed(int, std::string); // Called when connection goes away
- void setConnection(Connection*); // Set pointer to the AMQP Connection
- void reconnect(const Address&); //called by LinkRegistry
+ void reconnectLH(const Address&); //called by LinkRegistry
void close(); // Close the link from within the broker.
std::string getAuthMechanism() { return authMechanism; }
Modified: qpid/branches/qpid-3603-2/qpid/cpp/src/qpid/broker/LinkRegistry.cpp
URL:
http://svn.apache.org/viewvc/qpid/branches/qpid-3603-2/qpid/cpp/src/qpid/broker/LinkRegistry.cpp?rev=1243584&r1=1243583&r2=1243584&view=diff
==============================================================================
--- qpid/branches/qpid-3603-2/qpid/cpp/src/qpid/broker/LinkRegistry.cpp
(original)
+++ qpid/branches/qpid-3603-2/qpid/cpp/src/qpid/broker/LinkRegistry.cpp Mon Feb
13 16:18:46 2012
@@ -240,8 +240,7 @@ void LinkRegistry::notifyConnection(cons
{
Link::shared_ptr link = findLink(key);
if (link) {
- link->established();
- link->setConnection(c);
+ link->established(c);
c->setUserId(str(format("%1%@%2%") % link->getUsername() % realm));
}
}
---------------------------------------------------------------------
Apache Qpid - AMQP Messaging Implementation
Project: http://qpid.apache.org
Use/Interact: mailto:[email protected]
