| /* |
| * |
| * 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 <qpid/log/Statement.h> |
| |
| #include "Messages.h" |
| #include "Lsn.h" |
| #include "qpid/store/StoreException.h" |
| #include <boost/foreach.hpp> |
| |
| namespace qpid { |
| namespace store { |
| namespace ms_clfs { |
| |
| void |
| Messages::openLog(const std::string& path, const Log::TuningParameters& params) |
| { |
| log.open (path, params); |
| } |
| |
| void |
| Messages::add(const boost::intrusive_ptr<qpid::broker::PersistableMessage>& msg) |
| { |
| uint64_t id = log.add(msg); |
| msg->setPersistenceId(id); |
| std::auto_ptr<MessageInfo> autom(new MessageInfo); |
| MessageInfo::shared_ptr m(autom); |
| std::pair<uint64_t, MessageInfo::shared_ptr> p(id, m); |
| { |
| qpid::sys::ScopedWlock<qpid::sys::RWlock> l(lock); |
| messages.insert(p); |
| // If there's only this one message there, move the tail to it. |
| // This prevents the log from continually growing when messages |
| // are added and removed one at a time. |
| if (messages.size() == 1) { |
| CLFS_LSN newTail = idToLsn(id); |
| log.moveTail(newTail); |
| } |
| } |
| } |
| |
| void |
| Messages::enqueue(uint64_t msgId, uint64_t queueId, Transaction::shared_ptr& t) |
| { |
| MessageInfo::shared_ptr p; |
| { |
| qpid::sys::ScopedRlock<qpid::sys::RWlock> l(lock); |
| MessageMap::const_iterator i = messages.find(msgId); |
| if (i == messages.end()) |
| THROW_STORE_EXCEPTION("Message does not exist"); |
| p = i->second; |
| } |
| 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; |
| } |
| } |
| } |
| |
| void |
| Messages::dequeue(uint64_t msgId, uint64_t queueId, Transaction::shared_ptr& t) |
| { |
| MessageInfo::shared_ptr p; |
| { |
| qpid::sys::ScopedRlock<qpid::sys::RWlock> l(lock); |
| MessageMap::const_iterator i = messages.find(msgId); |
| if (i == messages.end()) |
| THROW_STORE_EXCEPTION("Message does not exist"); |
| p = i->second; |
| } |
| { |
| // 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) { |
| 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); |
| } |
| } |
| } |
| |
| // Commit a previous provisional enqueue or dequeue of a particular message |
| // actions under a specified transaction. If this results in the message's |
| // being removed from all queues, it is deleted. |
| void |
| Messages::commit(uint64_t msgId, Transaction::shared_ptr& t) |
| { |
| MessageInfo::shared_ptr p; |
| { |
| qpid::sys::ScopedRlock<qpid::sys::RWlock> l(lock); |
| MessageMap::const_iterator i = messages.find(msgId); |
| if (i == messages.end()) |
| THROW_STORE_EXCEPTION("Message does not exist"); |
| p = i->second; |
| } |
| { |
| 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(); |
| } |
| } |
| // 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 (p->where.empty()) { |
| try { |
| remove(msgId); |
| } |
| catch(...) {} |
| } |
| } |
| |
| // Abort a previous provisional enqueue or dequeue of a particular message |
| // actions under a specified transaction. If this results in the message's |
| // being removed from all queues, it is deleted. |
| void |
| Messages::abort(uint64_t msgId, Transaction::shared_ptr& t) |
| { |
| MessageInfo::shared_ptr p; |
| { |
| qpid::sys::ScopedRlock<qpid::sys::RWlock> l(lock); |
| MessageMap::const_iterator i = messages.find(msgId); |
| if (i == messages.end()) |
| THROW_STORE_EXCEPTION("Message does not exist"); |
| p = i->second; |
| } |
| { |
| 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; |
| } |
| } |
| } |
| // 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 (p->where.empty()) { |
| try { |
| remove(msgId); |
| } |
| catch(...) {} |
| } |
| } |
| |
| // Load part or all of a message's content from previously stored |
| // log record(s). |
| void |
| Messages::loadContent(uint64_t msgId, |
| std::string& data, |
| uint64_t offset, |
| uint32_t length) |
| { |
| log.loadContent(msgId, data, offset, length); |
| } |
| |
| // Recover the current set of messages and where they're queued from |
| // 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) |
| { |
| std::map<uint64_t, std::vector<MessageLog::RecoveredMsgOp> > messageOps; |
| log.recover(recoverer, messageMap, messageOps); |
| // Now read through the messageOps replaying the operations with the |
| // knowledge of which transactions committed, aborted, etc. A transaction |
| // should not be deleted until there are no messages referencing it so |
| // a message operation with a transaction id not found in transMap is |
| // a serious problem. |
| QPID_LOG(debug, "Beginning CLFS-recovered message operation replay"); |
| // Keep track of any messages that are recovered from the log but don't |
| // have any place to be. This can happen, for example, if the broker |
| // crashes while logging a message deletion. After all the recovery is |
| // done, delete all the homeless messages. |
| std::vector<uint64_t> homeless; |
| std::map<uint64_t, std::vector<MessageLog::RecoveredMsgOp> >::const_iterator msg; |
| for (msg = messageOps.begin(); msg != messageOps.end(); ++msg) { |
| uint64_t msgId = msg->first; |
| const std::vector<MessageLog::RecoveredMsgOp>& ops = msg->second; |
| QPID_LOG(debug, "Message " << msgId << "; " << ops.size() << " op(s)"); |
| MessageInfo::shared_ptr m(new MessageInfo); |
| std::vector<QueueEntry>& entries = messageQueueMap[msgId]; |
| 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 |
| // transaction isn't removed from the log while finalizing the |
| // recovery. If it were to be removed and the broker failed |
| // again before removing this message during normal operation, |
| // it couldn't be recovered again. |
| // |
| // Recall what is being reconstructed; 2 things: |
| // 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; |
| // 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 |
| // noted in the QueueEntry for the StorageProvider's map. |
| if (t->getState() == Transaction::TRANS_PREPARED) { |
| QPID_LOG(debug, dir << " for queue " << op->queueId << |
| ", prepared txn " << op->txnId); |
| TPCTransaction::shared_ptr tpct(boost::dynamic_pointer_cast<TPCTransaction>(t)); |
| if (tpct.get() == 0) |
| THROW_STORE_EXCEPTION("Invalid transaction state"); |
| t->enroll(msgId); |
| entry.xid = tpct->getXid(); |
| loc.transaction = t; |
| if (op->op == MessageLog::RECOVERED_ENQUEUE) { |
| entry.tplStatus = QueueEntry::ADDING; |
| loc.disposition = MessageInfo::TRANSACTION_ENQUEUE; |
| } |
| else { |
| entry.tplStatus = QueueEntry::REMOVING; |
| loc.disposition = MessageInfo::TRANSACTION_DEQUEUE; |
| } |
| } |
| else if (t->getState() != Transaction::TRANS_COMMITTED) { |
| QPID_LOG(debug, dir << " for queue " << op->queueId << |
| ", txn " << op->txnId << ", rolling back"); |
| continue; |
| } |
| } |
| // Here for non-transactional and prepared transactional operations |
| // to set up the messageQueueMap entries. Note that at this point |
| // a committed transactional operation looks like a |
| // non-transactional one as far as the QueueEntry is |
| // concerned - just do it. If this is an entry enqueuing a |
| // message, just add it to the entries list. If it's a dequeue |
| // operation, locate the matching entry for the queue and delete |
| // 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 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) { |
| 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; |
| ++w; |
| } |
| else { |
| w = m->where.erase(w); |
| } |
| } |
| } |
| } |
| } |
| // 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); |
| } |
| 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. |
| 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(...) { |
| } |
| } |
| |
| // Remove a specified message from those controlled by this object. |
| void |
| Messages::remove(uint64_t messageId) |
| { |
| uint64_t newFirstId = 0; |
| { |
| qpid::sys::ScopedWlock<qpid::sys::RWlock> l(lock); |
| messages.erase(messageId); |
| // May have deleted the first entry; if so the log can release that. |
| // If this message being deleted results in an empty list of |
| // messages, move the tail up to this message's LSN. This may |
| // result in one or more messages being stranded in the log |
| // until there's more activity. If a restart happens while these |
| // unneeded log records are there, the presence of the MessageDelete |
| // entry will cause the message(s) to be ignored anyway. |
| if (messages.empty()) |
| newFirstId = messageId; |
| else if (messages.begin()->first > messageId) |
| newFirstId = messages.begin()->first; |
| } |
| log.deleteMessage(messageId, newFirstId); |
| } |
| |
| }}} // namespace qpid::store::ms_clfs |
