summaryrefslogtreecommitdiff
path: root/cpp/src/qpid/store/ms-clfs/Messages.cpp
diff options
context:
space:
mode:
Diffstat (limited to 'cpp/src/qpid/store/ms-clfs/Messages.cpp')
-rw-r--r--cpp/src/qpid/store/ms-clfs/Messages.cpp302
1 files changed, 188 insertions, 114 deletions
diff --git a/cpp/src/qpid/store/ms-clfs/Messages.cpp b/cpp/src/qpid/store/ms-clfs/Messages.cpp
index 98cc19a7f7..6ab00d560a 100644
--- a/cpp/src/qpid/store/ms-clfs/Messages.cpp
+++ b/cpp/src/qpid/store/ms-clfs/Messages.cpp
@@ -24,6 +24,7 @@
#include "Messages.h"
#include "Lsn.h"
#include "qpid/store/StoreException.h"
+#include <boost/foreach.hpp>
namespace qpid {
namespace store {
@@ -67,37 +68,25 @@ Messages::enqueue(uint64_t msgId, uint64_t queueId, Transaction::shared_ptr& t)
THROW_STORE_EXCEPTION("Message does not exist");
p = i->second;
}
- // If transacted, it still needs to be counted as enqueued to ensure it
- // is not deleted. Remember the transacted operation so it can be properly
- // resolved later.
- ::InterlockedIncrement(&p->enqueuedCount);
- uint64_t transactionId = 0;
- if (t.get() != 0)
- transactionId = t->getId();
- if (transactionId != 0) {
- qpid::sys::ScopedLock<qpid::sys::Mutex> l(p->transOpsLock);
- p->transOps[t].push_back(MessageInfo::TRANSACTION_ENQUEUE);
- t->enroll(msgId);
- }
- try {
- log.recordEnqueue(msgId, queueId, transactionId);
- }
- catch (...) {
- // Undo the record-keeping if the log wasn't written correctly.
- ::InterlockedDecrement(&p->enqueuedCount);
- if (transactionId != 0) {
- t->unenroll(msgId);
- qpid::sys::ScopedLock<qpid::sys::Mutex> l(p->transOpsLock);
- std::vector<MessageInfo::TransType> &oplist = p->transOps[t];
- std::vector<MessageInfo::TransType>::iterator i;
- for (i = oplist.begin(); i < oplist.end(); ++i) {
- if (*i == MessageInfo::TRANSACTION_ENQUEUE) {
- oplist.erase(i);
- break;
- }
- }
+ MessageInfo::Location loc(queueId, t, MessageInfo::TRANSACTION_ENQUEUE);
+ {
+ qpid::sys::ScopedLock<qpid::sys::Mutex> l(p->whereLock);
+ p->where.push_back(loc);
+ uint64_t transactionId = 0;
+ if (t.get() != 0) {
+ transactionId = t->getId();
+ t->enroll(msgId);
+ }
+ try {
+ log.recordEnqueue(msgId, queueId, transactionId);
+ }
+ catch (...) {
+ // Undo the record-keeping if the log wasn't written correctly.
+ if (transactionId != 0)
+ t->unenroll(msgId);
+ p->where.pop_back();
+ throw;
}
- throw;
}
}
@@ -112,38 +101,46 @@ Messages::dequeue(uint64_t msgId, uint64_t queueId, Transaction::shared_ptr& t)
THROW_STORE_EXCEPTION("Message does not exist");
p = i->second;
}
- // Remember the transacted operation so it can be properly resolved later.
- uint64_t transactionId = 0;
- if (t.get() != 0)
- transactionId = t->getId();
- if (transactionId != 0) {
- qpid::sys::ScopedLock<qpid::sys::Mutex> l(p->transOpsLock);
- p->transOps[t].push_back(MessageInfo::TRANSACTION_DEQUEUE);
- t->enroll(msgId);
- }
- try {
- log.recordDequeue(msgId, queueId, transactionId);
- }
- catch(...) {
+ {
+ // Locate the 'where' entry for the specified queue. Once this operation
+ // is recorded in the log, update the 'where' entry to reflect it.
+ // Note that an existing entry in 'where' that refers to a transaction
+ // is not eligible for this operation.
+ qpid::sys::ScopedLock<qpid::sys::Mutex> l(p->whereLock);
+ std::list<MessageInfo::Location>::iterator i;
+ for (i = p->where.begin(); i != p->where.end(); ++i) {
+ if (i->queueId == queueId && i->transaction.get() == 0)
+ break;
+ }
+ if (i == p->where.end())
+ THROW_STORE_EXCEPTION("Message not on queue");
+ uint64_t transactionId = 0;
+ if (t.get() != 0) {
+ transactionId = t->getId();
+ t->enroll(msgId);
+ }
+ try {
+ log.recordDequeue(msgId, queueId, transactionId);
+ }
+ catch (...) {
+ // Undo the record-keeping if the log wasn't written correctly.
+ if (transactionId != 0)
+ t->unenroll(msgId);
+ throw;
+ }
+ // Ok, logged successfully. If this is a transactional op, note
+ // the transaction. If non-transactional, remove the 'where' entry.
if (transactionId != 0) {
- t->unenroll(msgId);
- qpid::sys::ScopedLock<qpid::sys::Mutex> l(p->transOpsLock);
- std::vector<MessageInfo::TransType> &oplist = p->transOps[t];
- std::vector<MessageInfo::TransType>::iterator i;
- for (i = oplist.begin(); i < oplist.end(); ++i) {
- if (*i == MessageInfo::TRANSACTION_DEQUEUE) {
- oplist.erase(i);
- break;
- }
- }
+ i->transaction = t;
+ i->disposition = MessageInfo::TRANSACTION_DEQUEUE;
+ }
+ else {
+ p->where.erase(i);
+ // If the message doesn't exist on any other queues, remove it.
+ if (p->where.empty())
+ remove(msgId);
}
- throw;
}
-
- // If transacted, leave the reference until the transaction commits.
- if (transactionId == 0)
- if (::InterlockedDecrement(&p->enqueuedCount) == 0)
- remove(msgId);
}
// Commit a previous provisional enqueue or dequeue of a particular message
@@ -161,22 +158,23 @@ Messages::commit(uint64_t msgId, Transaction::shared_ptr& t)
p = i->second;
}
{
- qpid::sys::ScopedLock<qpid::sys::Mutex> l(p->transOpsLock);
- std::vector<MessageInfo::TransType> &oplist = p->transOps[t];
- std::vector<MessageInfo::TransType>::iterator i;
- for (i = oplist.begin(); i < oplist.end(); ++i) {
- // Transactional dequeues left the ref count alone until commit
- // while transaction enqueues already incremented it.
- if (*i == MessageInfo::TRANSACTION_DEQUEUE)
- ::InterlockedDecrement(&p->enqueuedCount);
+ qpid::sys::ScopedLock<qpid::sys::Mutex> l(p->whereLock);
+ std::list<MessageInfo::Location>::iterator i;
+ for (i = p->where.begin(); i != p->where.end(); ++i) {
+ if (i->transaction != t)
+ continue;
+ // Transactional dequeues can now remove the item from the
+ // where list; enqueues just clear the transaction reference.
+ if (i->disposition == MessageInfo::TRANSACTION_DEQUEUE)
+ i = p->where.erase(i);
+ else
+ i->transaction.reset();
}
- // Remember, last deref of Transaction::shared_ptr deletes Transaction.
- p->transOps.erase(t);
}
// If committing results in this message having no further enqueue
// references, delete it. If the delete fails, swallow the exception
// and let recovery take care of removing it later.
- if (::InterlockedCompareExchange(&p->enqueuedCount, 0, 0) == 0) {
+ if (p->where.empty()) {
try {
remove(msgId);
}
@@ -199,22 +197,28 @@ Messages::abort(uint64_t msgId, Transaction::shared_ptr& t)
p = i->second;
}
{
- qpid::sys::ScopedLock<qpid::sys::Mutex> l(p->transOpsLock);
- std::vector<MessageInfo::TransType> &oplist = p->transOps[t];
- std::vector<MessageInfo::TransType>::iterator i;
- for (i = oplist.begin(); i < oplist.end(); ++i) {
- // Transactional enqueues incremented the ref count when seen;
- // while transaction dequeues left it alone.
- if (*i == MessageInfo::TRANSACTION_ENQUEUE)
- ::InterlockedDecrement(&p->enqueuedCount);
+ qpid::sys::ScopedLock<qpid::sys::Mutex> l(p->whereLock);
+ std::list<MessageInfo::Location>::iterator i = p->where.begin();
+ while (i != p->where.end()) {
+ if (i->transaction != t) {
+ ++i;
+ continue;
+ }
+ // Aborted transactional dequeues result in the message remaining
+ // enqueued like before the operation; enqueues clear the
+ // message from the where list - like the enqueue never happened.
+ if (i->disposition == MessageInfo::TRANSACTION_ENQUEUE)
+ i = p->where.erase(i);
+ else {
+ i->transaction.reset();
+ ++i;
+ }
}
- // Remember, last deref of Transaction::shared_ptr deletes Transaction.
- p->transOps.erase(t);
}
- // If committing results in this message having no further enqueue
+ // If aborting results in this message having no further enqueue
// references, delete it. If the delete fails, swallow the exception
// and let recovery take care of removing it later.
- if (::InterlockedCompareExchange(&p->enqueuedCount, 0, 0) == 0) {
+ if (p->where.empty()) {
try {
remove(msgId);
}
@@ -237,9 +241,10 @@ Messages::loadContent(uint64_t msgId,
// the log.
void
Messages::recover(qpid::broker::RecoveryManager& recoverer,
+ const std::set<uint64_t> &validQueues,
+ const std::map<uint64_t, Transaction::shared_ptr>& transMap,
qpid::store::MessageMap& messageMap,
- qpid::store::MessageQueueMap& messageQueueMap,
- const std::map<uint64_t, Transaction::shared_ptr>& transMap)
+ qpid::store::MessageQueueMap& messageQueueMap)
{
std::map<uint64_t, std::vector<MessageLog::RecoveredMsgOp> > messageOps;
log.recover(recoverer, messageMap, messageOps);
@@ -264,9 +269,16 @@ Messages::recover(qpid::broker::RecoveryManager& recoverer,
std::vector<MessageLog::RecoveredMsgOp>::const_iterator op;
for (op = ops.begin(); op != ops.end(); ++op) {
QueueEntry entry(op->queueId);
+ MessageInfo::Location loc(op->queueId);
std::string dir =
op->op == MessageLog::RECOVERED_ENQUEUE ? "enqueue"
: "dequeue";
+ if (validQueues.find(op->queueId) == validQueues.end()) {
+ QPID_LOG(info,
+ "Message " << msgId << dir << " on non-existant queue "
+ << op->queueId << "; dropped");
+ continue;
+ }
if (op->txnId != 0) {
// Be sure to enroll this message in the transaction even if
// it has committed or aborted. This ensures that the
@@ -276,22 +288,18 @@ Messages::recover(qpid::broker::RecoveryManager& recoverer,
// it couldn't be recovered again.
//
// Recall what is being reconstructed; 2 things:
- // 1. This class's 'messages' list which only keeps track
- // of how many queues reference each message (though NOT
- // which queues) and the transactions each message is
- // enrolled in. For this, aborted transactions cause the
- // result of the operation to be ignored, but the
- // message does need to be enrolled in the transaction
- // to properly maintain the transaction references until
- // the message is deleted.
- // 2. The StorageProvider's MessageQueueMap, which DOES
- // have an entry for each queue each message is on and
+ // 1. This class's 'messages' list which keeps track
+ // of the queues each message is on and the transactions
+ // each message is enrolled in. For this, aborted
+ // transactions cause the result of the operation to be
+ // ignored, but the message does need to be enrolled in
+ // the transaction to properly maintain the transaction
+ // references until the message is deleted.
+ // 2. The StorageProvider's MessageQueueMap, which also
+ // has an entry for each queue each message is on and
// its TPL status and associated xid.
const Transaction::shared_ptr &t =
transMap.find(op->txnId)->second;
- // Adds t to map, ensuring a reference to Transaction, even if
- // no ops are added to the TransType vector.
- std::vector<MessageInfo::TransType>& tOps = m->transOps[t];
// Prepared transactions cause the operation to be
// provisionally acted on, and the message to be enrolled in
// the transaction for when it commits/aborts. This is
@@ -304,13 +312,14 @@ Messages::recover(qpid::broker::RecoveryManager& recoverer,
THROW_STORE_EXCEPTION("Invalid transaction state");
t->enroll(msgId);
entry.xid = tpct->getXid();
+ loc.transaction = t;
if (op->op == MessageLog::RECOVERED_ENQUEUE) {
- tOps.push_back(MessageInfo::TRANSACTION_ENQUEUE);
entry.tplStatus = QueueEntry::ADDING;
+ loc.disposition = MessageInfo::TRANSACTION_ENQUEUE;
}
else {
- tOps.push_back(MessageInfo::TRANSACTION_DEQUEUE);
entry.tplStatus = QueueEntry::REMOVING;
+ loc.disposition = MessageInfo::TRANSACTION_DEQUEUE;
}
}
else if (t->getState() != Transaction::TRANS_COMMITTED) {
@@ -329,42 +338,107 @@ Messages::recover(qpid::broker::RecoveryManager& recoverer,
// it if the current op is non-transactional; if it's a prepared
// transaction then replace the existing entry with the current
// one that notes the message is enqueued but being removed under
- // a prepared transaciton.
+ // a prepared transaction.
QPID_LOG(debug, dir + " at queue " << entry.queueId);
if (op->op == MessageLog::RECOVERED_ENQUEUE) {
entries.push_back(entry);
+ m->where.push_back(loc);
}
else {
std::vector<QueueEntry>::iterator i = entries.begin();
while (i != entries.end()) {
if (i->queueId == entry.queueId) {
- *i = entry;
+ if (entry.tplStatus != QueueEntry::NONE)
+ *i = entry;
+ else
+ entries.erase(i);
break;
}
++i;
}
+ std::list<MessageInfo::Location>::iterator w = m->where.begin();
+ while (w != m->where.end()) {
+ if (w->queueId == loc.queueId) {
+ if (loc.transaction.get() != 0)
+ *w = loc;
+ else
+ m->where.erase(w);
+ }
+ }
}
}
- // Now that all the queue entries have been set correctly, the
- // enqueuedCount that MessageInfo keeps track of is simply the
- // number of queue map entries. If there are none, add this
- // message to the homeless list to be deleted from the log after
- // the recovery is done.
- if ((m->enqueuedCount = entries.size()) == 0) {
+ // Now that all the queue entries have been set correctly, see if
+ // there are any entries; they may have all been removed during
+ // recovery. If there are none, add this message to the homeless
+ // list to be deleted from the log after the recovery is done.
+ if (m->where.size() == 0) {
homeless.push_back(msgId);
messageMap.erase(msgId);
messageQueueMap.erase(msgId);
}
- std::pair<uint64_t, MessageInfo::shared_ptr> p(msgId, m);
- messages.insert(p);
+ else {
+ std::pair<uint64_t, MessageInfo::shared_ptr> p(msgId, m);
+ messages.insert(p);
+ }
}
QPID_LOG(debug, "Message log recovery done.");
// Done! Ok, go back and delete all the homeless messages.
- for (std::vector<uint64_t>::iterator i = homeless.begin();
- i != homeless.end();
- ++i) {
- QPID_LOG(debug, "Deleting homeless message " << *i);
- remove(*i);
+ BOOST_FOREACH(uint64_t msg, homeless) {
+ QPID_LOG(debug, "Deleting homeless message " << msg);
+ remove(msg);
+ }
+}
+
+// Expunge is called when a queue is deleted. All references to that
+// queue must be expunged from all messages. 'Dequeue' log records are
+// written for each queue entry removed, but any errors are swallowed.
+// On recovery there's a list of valid queues passed in. The deleted
+// queue will not be on that list so if any references to it are
+// recovered they'll get weeded out then.
+void
+Messages::expunge(uint64_t queueId)
+{
+ std::vector<uint64_t> toBeDeleted; // Messages to be deleted later.
+
+ {
+ // Lock everybody out since all messages are possibly in play.
+ // There also may be other threads already working on particular
+ // messages so individual message mutex still must be acquired.
+ qpid::sys::ScopedWlock<qpid::sys::RWlock> l(lock);
+ MessageMap::iterator m;
+ for (m = messages.begin(); m != messages.end(); ++m) {
+ MessageInfo::shared_ptr p = m->second;
+ {
+ qpid::sys::ScopedLock<qpid::sys::Mutex> ml(p->whereLock);
+ std::list<MessageInfo::Location>::iterator i = p->where.begin();
+ while (i != p->where.end()) {
+ if (i->queueId != queueId) {
+ ++i;
+ continue;
+ }
+ // If this entry is involved in a transaction, unenroll it.
+ // Then remove the entry.
+ if (i->transaction.get() != 0)
+ i->transaction->unenroll(m->first);
+ i = p->where.erase(i);
+ try {
+ log.recordDequeue(m->first, queueId, 0);
+ }
+ catch(...) {
+ }
+ }
+ if (p->where.size() == 0)
+ toBeDeleted.push_back(m->first);
+ }
+ }
+ }
+ // Swallow any exceptions during this; don't care. Recover it later
+ // if needed.
+ try {
+ BOOST_FOREACH(uint64_t msg, toBeDeleted)
+ remove(msg);
+ }
+ catch(...) {
}
}
View Lorry Controller Status