You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@qpid.apache.org by rg...@apache.org on 2014/02/07 16:24:53 UTC
svn commit: r1565688 [2/4] - in
/qpid/branches/java-broker-amqp-1-0-management: ./ cpp/bindings/qmf2/ruby/
cpp/bindings/qpid/perl/ cpp/bindings/qpid/ruby/ cpp/include/qpid/ cpp/src/
cpp/src/qpid/broker/ cpp/src/qpid/client/windows/ cpp/src/qpid/ha/ cpp...
Modified: qpid/branches/java-broker-amqp-1-0-management/cpp/src/qpid/linearstore/journal/RecoveryManager.cpp
URL: http://svn.apache.org/viewvc/qpid/branches/java-broker-amqp-1-0-management/cpp/src/qpid/linearstore/journal/RecoveryManager.cpp?rev=1565688&r1=1565687&r2=1565688&view=diff
==============================================================================
--- qpid/branches/java-broker-amqp-1-0-management/cpp/src/qpid/linearstore/journal/RecoveryManager.cpp (original)
+++ qpid/branches/java-broker-amqp-1-0-management/cpp/src/qpid/linearstore/journal/RecoveryManager.cpp Fri Feb 7 15:24:50 2014
@@ -56,11 +56,15 @@ RecoveredRecordData_t::RecoveredRecordDa
pendingTransaction_(ptxn)
{}
-
bool recordIdListCompare(RecoveredRecordData_t a, RecoveredRecordData_t b) {
return a.recordId_ < b.recordId_;
}
+RecoveredFileData_t::RecoveredFileData_t(JournalFile* journalFilePtr, const uint32_t completedDblkCount) :
+ journalFilePtr_(journalFilePtr),
+ completedDblkCount_(completedDblkCount)
+{}
+
RecoveryManager::RecoveryManager(const std::string& journalDirectory,
const std::string& queuename,
enq_map& enqueueMapRef,
@@ -77,11 +81,17 @@ RecoveryManager::RecoveryManager(const s
highestRecordId_(0ULL),
highestFileNumber_(0ULL),
lastFileFullFlag_(false),
+ initial_fid_(0),
currentSerial_(0),
efpFileSize_kib_(0)
{}
-RecoveryManager::~RecoveryManager() {}
+RecoveryManager::~RecoveryManager() {
+ for (fileNumberMapItr_t i = fileNumberMap_.begin(); i != fileNumberMap_.end(); ++i) {
+ delete i->second;
+ }
+ fileNumberMap_.clear();
+}
void RecoveryManager::analyzeJournals(const std::vector<std::string>* preparedTransactionListPtr,
EmptyFilePoolManager* emptyFilePoolManager,
@@ -92,9 +102,6 @@ void RecoveryManager::analyzeJournals(co
*emptyFilePoolPtrPtr = emptyFilePoolManager->getEmptyFilePool(efpIdentity);
efpFileSize_kib_ = (*emptyFilePoolPtrPtr)->fileSize_kib();
- // Check for file full condition
- lastFileFullFlag_ = endOffset_ == (std::streamoff)(*emptyFilePoolPtrPtr)->fileSize_kib() * 1024;
-
if (!journalEmptyFlag_) {
// Read all records, establish remaining enqueued records
@@ -106,6 +113,9 @@ void RecoveryManager::analyzeJournals(co
inFileStream_.close();
}
+ // Check for file full condition
+ lastFileFullFlag_ = endOffset_ == (std::streamoff)(*emptyFilePoolPtrPtr)->fileSize_kib() * 1024;
+
// Remove leading files which have no enqueued records
removeEmptyFiles(*emptyFilePoolPtrPtr);
@@ -121,7 +131,7 @@ void RecoveryManager::analyzeJournals(co
// Unlock any affected enqueues in emap
for (tdl_itr_t i=tdl.begin(); i<tdl.end(); i++) {
if (i->enq_flag_) { // enq op - decrement enqueue count
- fileNumberMap_[i->pfid_]->decrEnqueuedRecordCount();
+ fileNumberMap_[i->pfid_]->journalFilePtr_->decrEnqueuedRecordCount();
} else if (enqueueMapRef_.is_enqueued(i->drid_, true)) { // deq op - unlock enq record
if (enqueueMapRef_.unlock(i->drid_) < enq_map::EMAP_OK) { // fail
// enq_map::unlock()'s only error is enq_map::EMAP_RID_NOT_FOUND
@@ -174,7 +184,7 @@ bool RecoveryManager::readNextRemainingR
}
} while (!foundRecord);
- if (!inFileStream_.is_open() || currentJournalFileConstItr_->first != recordIdListConstItr_->fileId_) {
+ if (!inFileStream_.is_open() || currentJournalFileItr_->first != recordIdListConstItr_->fileId_) {
if (!getFile(recordIdListConstItr_->fileId_, false)) {
std::ostringstream oss;
oss << "Failed to open file with file-id=" << recordIdListConstItr_->fileId_;
@@ -231,7 +241,6 @@ bool RecoveryManager::readNextRemainingR
::rec_tail_t enqueueTail;
inFileStream_.read((char*)&enqueueTail, sizeof(::rec_tail_t));
uint32_t cs = checksum.getChecksum();
-//std::cout << std::hex << "### rid=0x" << enqueueHeader._rhdr._rid << " rtcs=0x" << enqueueTail._checksum << " cs=0x" << cs << std::dec << std::endl; // DEBUG
uint16_t res = ::rec_tail_check(&enqueueTail, &enqueueHeader._rhdr, cs);
if (res != 0) {
std::stringstream oss;
@@ -266,17 +275,30 @@ bool RecoveryManager::readNextRemainingR
void RecoveryManager::setLinearFileControllerJournals(lfcAddJournalFileFn fnPtr,
LinearFileController* lfcPtr) {
if (journalEmptyFlag_) {
- if (uninitializedJournal_.size() > 0) {
- lfcPtr->restoreEmptyFile(uninitializedJournal_);
+ if (uninitFileList_.size() > 0) {
+ std::string uninitFile = uninitFileList_.back();
+ uninitFileList_.pop_back();
+ lfcPtr->restoreEmptyFile(uninitFile);
}
} else {
for (fileNumberMapConstItr_t i = fileNumberMap_.begin(); i != fileNumberMap_.end(); ++i) {
- uint32_t fileDblkCount = i->first == highestFileNumber_ ? // Is this this last file?
- endOffset_ / QLS_DBLK_SIZE_BYTES : // Last file uses _endOffset
- efpFileSize_kib_ * 1024 / QLS_DBLK_SIZE_BYTES; // All others use file size to make them full
- (lfcPtr->*fnPtr)(i->second, fileDblkCount);
+ (lfcPtr->*fnPtr)(i->second->journalFilePtr_, i->second->completedDblkCount_, i->first == initial_fid_);
}
}
+
+ std::ostringstream oss;
+ bool logFlag = !notNeededFilesList_.empty();
+ if (logFlag) {
+ oss << "Files removed from head of journal: prior truncation during recovery:";
+ }
+ while (!notNeededFilesList_.empty()) {
+ lfcPtr->removeFileToEfp(notNeededFilesList_.back());
+ oss << std::endl << " * " << notNeededFilesList_.back();
+ notNeededFilesList_.pop_back();
+ }
+ if (logFlag) {
+ journalLogRef_.log(JournalLog::LOG_NOTICE, queueName_, oss.str());
+ }
}
std::string RecoveryManager::toString(const std::string& jid) {
@@ -285,7 +307,7 @@ std::string RecoveryManager::toString(co
oss << " Number of journal files = " << fileNumberMap_.size() << std::endl;
oss << " Journal File List:" << std::endl;
for (fileNumberMapConstItr_t k=fileNumberMap_.begin(); k!=fileNumberMap_.end(); ++k) {
- std::string fqFileName = k->second->getFqFileName();
+ std::string fqFileName = k->second->journalFilePtr_->getFqFileName();
oss << " " << k->first << ": " << fqFileName.substr(fqFileName.rfind('/')+1) << std::endl;
}
oss << " Enqueue Counts: [ ";
@@ -293,7 +315,7 @@ std::string RecoveryManager::toString(co
if (l != fileNumberMap_.begin()) {
oss << ", ";
}
- oss << l->second->getEnqueuedRecordCount();
+ oss << l->second->journalFilePtr_->getEnqueuedRecordCount();
}
oss << " ]" << std::endl;
oss << " Journal empty = " << (journalEmptyFlag_ ? "TRUE" : "FALSE") << std::endl;
@@ -330,15 +352,17 @@ std::string RecoveryManager::toLog(const
<< std::setw(10) << "--------"
<< std::endl;
for (fileNumberMapConstItr_t k=fileNumberMap_.begin(); k!=fileNumberMap_.end(); ++k) {
- std::string fqFileName = k->second->getFqFileName();
+ std::string fqFileName = k->second->journalFilePtr_->getFqFileName();
+ std::ostringstream fid;
+ fid << std::hex << "0x" << k->first;
std::ostringstream fro;
- fro << std::hex << "0x" << k->second->getFirstRecordOffset();
- oss << indentStr << std::setw(7) << k->first
+ fro << std::hex << "0x" << k->second->journalFilePtr_->getFirstRecordOffset();
+ oss << indentStr << std::setw(7) << fid.str()
<< std::setw(43) << fqFileName.substr(fqFileName.rfind('/')+1)
<< std::setw(16) << fro.str()
- << std::setw(12) << k->second->getEnqueuedRecordCount()
- << std::setw(5) << k->second->getEfpIdentity().pn_
- << std::setw(9) << k->second->getEfpIdentity().ds_ << "k"
+ << std::setw(12) << k->second->journalFilePtr_->getEnqueuedRecordCount()
+ << std::setw(5) << k->second->journalFilePtr_->getEfpIdentity().pn_
+ << std::setw(9) << k->second->journalFilePtr_->getEfpIdentity().ds_ << "k"
<< std::endl;
}
oss << indentStr << "First record offset in first file = 0x" << std::hex << firstRecordOffset_ <<
@@ -347,7 +371,7 @@ std::string RecoveryManager::toLog(const
(endOffset_/QLS_DBLK_SIZE_BYTES) << " dblks)" << std::endl;
oss << indentStr << "Highest rid found = 0x" << std::hex << highestRecordId_ << std::dec << std::endl;
oss << indentStr << "Last file full = " << (lastFileFullFlag_ ? "TRUE" : "FALSE") << std::endl;
- oss << indentStr << "Enqueued records (txn & non-txn):";
+ //oss << indentStr << "Enqueued records (txn & non-txn):"; // TODO: complete report
}
return oss.str();
}
@@ -357,27 +381,28 @@ std::string RecoveryManager::toLog(const
void RecoveryManager::analyzeJournalFileHeaders(efpIdentity_t& efpIdentity) {
std::string headerQueueName;
::file_hdr_t fileHeader;
- directoryList_t directoryList;
+ stringList_t directoryList;
jdir::read_dir(journalDirectory_, directoryList, false, true, false, true);
- for (directoryListConstItr_t i = directoryList.begin(); i != directoryList.end(); ++i) {
+ for (stringListConstItr_t i = directoryList.begin(); i != directoryList.end(); ++i) {
readJournalFileHeader(*i, fileHeader, headerQueueName);
if (headerQueueName.empty()) {
std::ostringstream oss;
- if (uninitializedJournal_.empty()) {
- oss << "Journal file " << (*i) << " is first uninitialized (not yet written) journal file.";
- journalLogRef_.log(JournalLog::LOG_NOTICE, queueName_, oss.str());
- uninitializedJournal_ = *i;
- } else {
- oss << "Journal file " << (*i) << " is second or greater uninitialized journal file - ignoring";
- journalLogRef_.log(JournalLog::LOG_WARN, queueName_, oss.str());
- }
+ oss << "Journal file " << (*i) << " is uninitialized";
+ journalLogRef_.log(JournalLog::LOG_WARN, queueName_, oss.str());
+ uninitFileList_.push_back(*i);
} else if (headerQueueName.compare(queueName_) != 0) {
std::ostringstream oss;
oss << "Journal file " << (*i) << " belongs to queue \"" << headerQueueName << "\": ignoring";
journalLogRef_.log(JournalLog::LOG_WARN, queueName_, oss.str());
} else {
- JournalFile* jfp = new JournalFile(*i, fileHeader);
- fileNumberMap_[fileHeader._file_number] = jfp;
+ JournalFile* jfp = new JournalFile(*i, fileHeader, queueName_);
+ std::pair<fileNumberMapItr_t, bool> res = fileNumberMap_.insert(
+ std::pair<uint64_t, RecoveredFileData_t*>(fileHeader._file_number, new RecoveredFileData_t(jfp, 0)));
+ if (!res.second) {
+ std::ostringstream oss;
+ oss << "Journal file " << (*i) << " has fid=0x" << std::hex << jfp->getFileSeqNum() << " which already exists for this journal.";
+ throw jexception(oss.str()); // TODO: complete this exception
+ }
if (fileHeader._file_number > highestFileNumber_) {
highestFileNumber_ = fileHeader._file_number;
}
@@ -393,7 +418,7 @@ void RecoveryManager::analyzeJournalFile
if (fileNumberMap_.empty()) {
journalEmptyFlag_ = true;
} else {
- currentJournalFileConstItr_ = fileNumberMap_.begin();
+ currentJournalFileItr_ = fileNumberMap_.begin();
}
}
@@ -408,7 +433,7 @@ void RecoveryManager::checkFileStreamOk(
}
}
-void RecoveryManager::checkJournalAlignment(const std::streampos recordPosition) {
+void RecoveryManager::checkJournalAlignment(const uint64_t start_fid, const std::streampos recordPosition) {
if (recordPosition % QLS_DBLK_SIZE_BYTES != 0) {
std::ostringstream oss;
oss << "Current read pointer not dblk aligned: recordPosition=0x" << std::hex << recordPosition;
@@ -420,12 +445,13 @@ void RecoveryManager::checkJournalAlignm
if (sblkOffset)
{
std::ostringstream oss1;
- oss1 << std::hex << "Bad record alignment found at fid=0x" << getCurrentFileNumber();
+ oss1 << std::hex << "Bad record alignment found at fid=0x" << start_fid;
oss1 << " offs=0x" << currentPosn << " (likely journal overwrite boundary); " << std::dec;
oss1 << (QLS_SBLK_SIZE_DBLKS - (sblkOffset/QLS_DBLK_SIZE_BYTES)) << " filler record(s) required.";
journalLogRef_.log(JournalLog::LOG_WARN, queueName_, oss1.str());
- std::ofstream outFileStream(getCurrentFileName().c_str(), std::ios_base::in | std::ios_base::out | std::ios_base::binary);
+ fileNumberMapConstItr_t fnmItr = fileNumberMap_.find(start_fid);
+ std::ofstream outFileStream(fnmItr->second->journalFilePtr_->getFqFileName().c_str(), std::ios_base::in | std::ios_base::out | std::ios_base::binary);
if (!outFileStream.good()) {
throw jexception(jerrno::JERR__FILEIO, getCurrentFileName(), "RecoveryManager", "checkJournalAlignment");
}
@@ -447,7 +473,7 @@ void RecoveryManager::checkJournalAlignm
throw jexception(jerrno::JERR_RCVM_WRITE, "RecoveryManager", "checkJournalAlignment");
}
std::ostringstream oss2;
- oss2 << std::hex << "Recover phase write: Wrote filler record: fid=0x" << getCurrentFileNumber();
+ oss2 << std::hex << "Recover phase write: Wrote filler record: fid=0x" << start_fid;
oss2 << " offs=0x" << currentPosn;
journalLogRef_.log(JournalLog::LOG_NOTICE, queueName_, oss2.str());
currentPosn = outFileStream.tellp();
@@ -456,16 +482,15 @@ void RecoveryManager::checkJournalAlignm
std::free(writeBuffer);
journalLogRef_.log(JournalLog::LOG_INFO, queueName_, "Bad record alignment fixed.");
}
- endOffset_ = currentPosn;
+ lastRecord(start_fid, currentPosn);
}
bool RecoveryManager::decodeRecord(jrec& record,
std::size_t& cumulativeSizeRead,
::rec_hdr_t& headerRecord,
- std::streampos& fileOffset)
+ const uint64_t start_fid,
+ const std::streampos recordOffset)
{
- std::streampos start_file_offs = fileOffset;
-
if (highestRecordId_ == 0) {
highestRecordId_ = headerRecord._rid;
} else if (headerRecord._rid - highestRecordId_ < 0x8000000000000000ULL) { // RFC 1982 comparison for unsigned 64-bit
@@ -475,7 +500,7 @@ bool RecoveryManager::decodeRecord(jrec&
bool done = false;
while (!done) {
try {
- done = record.decode(headerRecord, &inFileStream_, cumulativeSizeRead);
+ done = record.decode(headerRecord, &inFileStream_, cumulativeSizeRead, recordOffset);
}
catch (const jexception& e) {
if (e.err_code() == jerrno::JERR_JREC_BADRECTAIL) {
@@ -485,11 +510,12 @@ bool RecoveryManager::decodeRecord(jrec&
} else {
journalLogRef_.log(JournalLog::LOG_INFO, queueName_, e.what());
}
- checkJournalAlignment(start_file_offs);
+ checkJournalAlignment(start_fid, recordOffset);
return false;
}
if (!done && needNextFile()) {
if (!getNextFile(false)) {
+ checkJournalAlignment(start_fid, recordOffset);
return false;
}
}
@@ -498,11 +524,11 @@ bool RecoveryManager::decodeRecord(jrec&
}
std::string RecoveryManager::getCurrentFileName() const {
- return currentJournalFileConstItr_->second->getFqFileName();
+ return currentJournalFileItr_->second->journalFilePtr_->getFqFileName();
}
uint64_t RecoveryManager::getCurrentFileNumber() const {
- return currentJournalFileConstItr_->first;
+ return currentJournalFileItr_->first;
}
bool RecoveryManager::getFile(const uint64_t fileNumber, bool jumpToFirstRecordOffsetFlag) {
@@ -511,8 +537,8 @@ bool RecoveryManager::getFile(const uint
//std::cout << " f=" << getCurrentFileName() << "]" << std::flush; // DEBUG
inFileStream_.clear(); // clear eof flag, req'd for older versions of c++
}
- currentJournalFileConstItr_ = fileNumberMap_.find(fileNumber);
- if (currentJournalFileConstItr_ == fileNumberMap_.end()) {
+ currentJournalFileItr_ = fileNumberMap_.find(fileNumber);
+ if (currentJournalFileItr_ == fileNumberMap_.end()) {
return false;
}
inFileStream_.open(getCurrentFileName().c_str(), std::ios_base::in | std::ios_base::binary);
@@ -536,7 +562,8 @@ bool RecoveryManager::getNextFile(bool j
if (inFileStream_.is_open()) {
inFileStream_.close();
//std::cout << " .f=" << getCurrentFileName() << "]" << std::flush; // DEBUG
- if (++currentJournalFileConstItr_ == fileNumberMap_.end()) {
+ currentJournalFileItr_->second->completedDblkCount_ = efpFileSize_kib_ * 1024 / QLS_DBLK_SIZE_BYTES;
+ if (++currentJournalFileItr_ == fileNumberMap_.end()) {
return false;
}
inFileStream_.clear(); // clear eof flag, req'd for older versions of c++
@@ -562,13 +589,15 @@ bool RecoveryManager::getNextRecordHeade
rec_hdr_t h;
bool hdr_ok = false;
- std::streampos file_pos;
+ uint64_t file_id = 0;
+ std::streampos file_pos = 0;
while (!hdr_ok) {
if (needNextFile()) {
if (!getNextFile(true)) {
return false;
}
}
+ file_id = currentJournalFileItr_->second->journalFilePtr_->getFileSeqNum();
file_pos = inFileStream_.tellg();
if (file_pos == std::streampos(-1)) {
std::ostringstream oss;
@@ -587,21 +616,21 @@ bool RecoveryManager::getNextRecordHeade
}
}
+ uint64_t start_fid = getCurrentFileNumber(); // fid may increment in decode() if record folds over file boundary
switch(h._magic) {
case QLS_ENQ_MAGIC:
{
//std::cout << " 0x" << std::hex << file_pos << ".e.0x" << h._rid << std::dec << std::flush; // DEBUG
if (::rec_hdr_check(&h, QLS_ENQ_MAGIC, QLS_JRNL_VERSION, currentSerial_) != 0) {
- endOffset_ = file_pos;
+ lastRecord(file_id, file_pos);
return false;
}
enq_rec er;
- uint64_t start_fid = getCurrentFileNumber(); // fid may increment in decode() if record folds over file boundary
- if (!decodeRecord(er, cum_size_read, h, file_pos)) {
+ if (!decodeRecord(er, cum_size_read, h, start_fid, file_pos)) {
return false;
}
if (!er.is_transient()) { // Ignore transient msgs
- fileNumberMap_[start_fid]->incrEnqueuedRecordCount();
+ fileNumberMap_[start_fid]->journalFilePtr_->incrEnqueuedRecordCount();
if (er.xid_size()) {
er.get_xid(&xidp);
if (xidp == 0) {
@@ -629,12 +658,11 @@ bool RecoveryManager::getNextRecordHeade
{
//std::cout << " 0x" << std::hex << file_pos << ".d.0x" << h._rid << std::dec << std::flush; // DEBUG
if (::rec_hdr_check(&h, QLS_DEQ_MAGIC, QLS_JRNL_VERSION, currentSerial_) != 0) {
- endOffset_ = file_pos;
+ lastRecord(file_id, file_pos);
return false;
}
deq_rec dr;
- uint16_t start_fid = getCurrentFileNumber(); // fid may increment in decode() if record folds over file boundary
- if (!decodeRecord(dr, cum_size_read, h, file_pos)) {
+ if (!decodeRecord(dr, cum_size_read, h, start_fid, file_pos)) {
return false;
}
if (dr.xid_size()) {
@@ -655,7 +683,7 @@ bool RecoveryManager::getNextRecordHeade
} else {
uint64_t enq_fid;
if (enqueueMapRef_.get_remove_pfid(dr.deq_rid(), enq_fid, true) == enq_map::EMAP_OK) { // ignore not found error
- fileNumberMap_[enq_fid]->decrEnqueuedRecordCount();
+ fileNumberMap_[enq_fid]->journalFilePtr_->decrEnqueuedRecordCount();
}
}
}
@@ -664,11 +692,11 @@ bool RecoveryManager::getNextRecordHeade
{
//std::cout << " 0x" << std::hex << file_pos << ".a.0x" << h._rid << std::dec << std::flush; // DEBUG
if (::rec_hdr_check(&h, QLS_TXA_MAGIC, QLS_JRNL_VERSION, currentSerial_) != 0) {
- endOffset_ = file_pos;
+ lastRecord(file_id, file_pos);
return false;
}
txn_rec ar;
- if (!decodeRecord(ar, cum_size_read, h, file_pos)) {
+ if (!decodeRecord(ar, cum_size_read, h, start_fid, file_pos)) {
return false;
}
// Delete this txn from tmap, unlock any locked records in emap
@@ -680,7 +708,7 @@ bool RecoveryManager::getNextRecordHeade
txn_data_list_t tdl = transactionMapRef_.get_remove_tdata_list(xid); // tdl will be empty if xid not found
for (tdl_itr_t itr = tdl.begin(); itr != tdl.end(); itr++) {
if (itr->enq_flag_) {
- fileNumberMap_[itr->pfid_]->decrEnqueuedRecordCount();
+ fileNumberMap_[itr->pfid_]->journalFilePtr_->decrEnqueuedRecordCount();
} else {
enqueueMapRef_.unlock(itr->drid_); // ignore not found error
}
@@ -691,11 +719,11 @@ bool RecoveryManager::getNextRecordHeade
{
//std::cout << " 0x" << std::hex << file_pos << ".c.0x" << h._rid << std::dec << std::flush; // DEBUG
if (::rec_hdr_check(&h, QLS_TXC_MAGIC, QLS_JRNL_VERSION, currentSerial_) != 0) {
- endOffset_ = file_pos;
+ lastRecord(file_id, file_pos);
return false;
}
txn_rec cr;
- if (!decodeRecord(cr, cum_size_read, h, file_pos)) {
+ if (!decodeRecord(cr, cum_size_read, h, start_fid, file_pos)) {
return false;
}
// Delete this txn from tmap, process records into emap
@@ -717,7 +745,7 @@ bool RecoveryManager::getNextRecordHeade
} else { // txn dequeue
uint64_t enq_fid;
if (enqueueMapRef_.get_remove_pfid(itr->drid_, enq_fid, true) == enq_map::EMAP_OK) // ignore not found error
- fileNumberMap_[enq_fid]->decrEnqueuedRecordCount();
+ fileNumberMap_[enq_fid]->journalFilePtr_->decrEnqueuedRecordCount();
}
}
}
@@ -729,7 +757,9 @@ bool RecoveryManager::getNextRecordHeade
inFileStream_.ignore(rec_dblks * QLS_DBLK_SIZE_BYTES - sizeof(::rec_hdr_t));
checkFileStreamOk(false);
if (needNextFile()) {
+ file_pos += rec_dblks * QLS_DBLK_SIZE_BYTES;
if (!getNextFile(false)) {
+ lastRecord(start_fid, file_pos);
return false;
}
}
@@ -737,17 +767,36 @@ bool RecoveryManager::getNextRecordHeade
break;
case 0:
//std::cout << " 0x" << std::hex << file_pos << ".0" << std::dec << std::endl << std::flush; // DEBUG
- checkJournalAlignment(file_pos);
+ checkJournalAlignment(getCurrentFileNumber(), file_pos);
return false;
default:
//std::cout << " 0x" << std::hex << file_pos << ".?" << std::dec << std::endl << std::flush; // DEBUG
// Stop as this is the overwrite boundary.
- checkJournalAlignment(file_pos);
+ checkJournalAlignment(getCurrentFileNumber(), file_pos);
return false;
}
return true;
}
+void RecoveryManager::lastRecord(const uint64_t file_id, const std::streamoff endOffset) {
+ endOffset_ = endOffset;
+ initial_fid_ = file_id;
+ fileNumberMap_[file_id]->completedDblkCount_ = endOffset_ / QLS_DBLK_SIZE_BYTES;
+
+ // Remove any files in fileNumberMap_ beyond initial_fid_
+ fileNumberMapItr_t unwantedFirstItr = fileNumberMap_.find(file_id);
+ if (++unwantedFirstItr != fileNumberMap_.end()) {
+ fileNumberMapItr_t itr = unwantedFirstItr;
+ notNeededFilesList_.push_back(unwantedFirstItr->second->journalFilePtr_->getFqFileName());
+ while (++itr != fileNumberMap_.end()) {
+ notNeededFilesList_.push_back(itr->second->journalFilePtr_->getFqFileName());
+ delete itr->second->journalFilePtr_;
+ delete itr->second;
+ }
+ fileNumberMap_.erase(unwantedFirstItr, fileNumberMap_.end());
+ }
+}
+
bool RecoveryManager::needNextFile() {
if (inFileStream_.is_open()) {
return inFileStream_.eof() || inFileStream_.tellg() >= std::streampos(efpFileSize_kib_ * 1024);
@@ -820,7 +869,7 @@ bool RecoveryManager::readFileHeader() {
currentSerial_ = fhdr._rhdr._serial;
} else {
inFileStream_.close();
- if (currentJournalFileConstItr_ == fileNumberMap_.begin()) {
+ if (currentJournalFileItr_ == fileNumberMap_.begin()) {
journalEmptyFlag_ = true;
}
return false;
@@ -855,9 +904,11 @@ void RecoveryManager::readJournalFileHea
}
void RecoveryManager::removeEmptyFiles(EmptyFilePool* emptyFilePoolPtr) {
- while (fileNumberMap_.begin()->second->getEnqueuedRecordCount() == 0 && fileNumberMap_.size() > 1) {
-//std::cout << "*** File " << i->first << ": " << i->second << " is empty." << std::endl; // DEBUG
- emptyFilePoolPtr->returnEmptyFile(fileNumberMap_.begin()->second->getFqFileName());
+ while (fileNumberMap_.begin()->second->journalFilePtr_->getEnqueuedRecordCount() == 0 && fileNumberMap_.size() > 1) {
+ RecoveredFileData_t* rfdp = fileNumberMap_.begin()->second;
+ emptyFilePoolPtr->returnEmptyFile(rfdp->journalFilePtr_->getFqFileName());
+ delete rfdp->journalFilePtr_;
+ delete rfdp;
fileNumberMap_.erase(fileNumberMap_.begin()->first);
}
}
Modified: qpid/branches/java-broker-amqp-1-0-management/cpp/src/qpid/linearstore/journal/RecoveryManager.h
URL: http://svn.apache.org/viewvc/qpid/branches/java-broker-amqp-1-0-management/cpp/src/qpid/linearstore/journal/RecoveryManager.h?rev=1565688&r1=1565687&r2=1565688&view=diff
==============================================================================
--- qpid/branches/java-broker-amqp-1-0-management/cpp/src/qpid/linearstore/journal/RecoveryManager.h (original)
+++ qpid/branches/java-broker-amqp-1-0-management/cpp/src/qpid/linearstore/journal/RecoveryManager.h Fri Feb 7 15:24:50 2014
@@ -51,15 +51,21 @@ struct RecoveredRecordData_t {
RecoveredRecordData_t(const uint64_t rid, const uint64_t fid, const std::streampos foffs, bool ptxn);
};
+struct RecoveredFileData_t {
+ JournalFile* journalFilePtr_;
+ uint32_t completedDblkCount_;
+ RecoveredFileData_t(JournalFile* journalFilePtr, const uint32_t completedDblkCount);
+};
+
bool recordIdListCompare(RecoveredRecordData_t a, RecoveredRecordData_t b);
class RecoveryManager
{
protected:
// Types
- typedef std::vector<std::string> directoryList_t;
- typedef directoryList_t::const_iterator directoryListConstItr_t;
- typedef std::map<uint64_t, JournalFile*> fileNumberMap_t;
+ typedef std::vector<std::string> stringList_t;
+ typedef stringList_t::const_iterator stringListConstItr_t;
+ typedef std::map<uint64_t, RecoveredFileData_t*> fileNumberMap_t;
typedef fileNumberMap_t::iterator fileNumberMapItr_t;
typedef fileNumberMap_t::const_iterator fileNumberMapConstItr_t;
typedef std::vector<RecoveredRecordData_t> recordIdList_t;
@@ -74,18 +80,20 @@ protected:
// Initial journal analysis data
fileNumberMap_t fileNumberMap_; ///< File number - JournalFilePtr map
+ stringList_t notNeededFilesList_; ///< Files not needed and to be returned to EFP
+ stringList_t uninitFileList_; ///< File name of uninitialized journal files found during header analysis
bool journalEmptyFlag_; ///< Journal data files empty
std::streamoff firstRecordOffset_; ///< First record offset in ffid
std::streamoff endOffset_; ///< End offset (first byte past last record)
uint64_t highestRecordId_; ///< Highest rid found
uint64_t highestFileNumber_; ///< Highest file number found
bool lastFileFullFlag_; ///< Last file is full
- std::string uninitializedJournal_; ///< File name of uninitialized journal found during header analysis
+ uint64_t initial_fid_; ///< File id where initial write after recovery will occur
// State for recovery of individual enqueued records
uint64_t currentSerial_;
uint32_t efpFileSize_kib_;
- fileNumberMapConstItr_t currentJournalFileConstItr_;
+ fileNumberMapConstItr_t currentJournalFileItr_;
std::string currentFileName_;
std::ifstream inFileStream_;
recordIdList_t recordIdList_;
@@ -121,16 +129,18 @@ public:
protected:
void analyzeJournalFileHeaders(efpIdentity_t& efpIdentity);
void checkFileStreamOk(bool checkEof);
- void checkJournalAlignment(const std::streampos recordPosition);
+ void checkJournalAlignment(const uint64_t start_fid, const std::streampos recordPosition);
bool decodeRecord(jrec& record,
std::size_t& cumulativeSizeRead,
::rec_hdr_t& recordHeader,
- std::streampos& fileOffset);
+ const uint64_t start_fid,
+ const std::streampos recordOffset);
std::string getCurrentFileName() const;
uint64_t getCurrentFileNumber() const;
bool getFile(const uint64_t fileNumber, bool jumpToFirstRecordOffsetFlag);
bool getNextFile(bool jumpToFirstRecordOffsetFlag);
bool getNextRecordHeader();
+ void lastRecord(const uint64_t file_id, const std::streamoff endOffset);
bool needNextFile();
void prepareRecordList();
bool readFileHeader();
Modified: qpid/branches/java-broker-amqp-1-0-management/cpp/src/qpid/linearstore/journal/deq_rec.cpp
URL: http://svn.apache.org/viewvc/qpid/branches/java-broker-amqp-1-0-management/cpp/src/qpid/linearstore/journal/deq_rec.cpp?rev=1565688&r1=1565687&r2=1565688&view=diff
==============================================================================
--- qpid/branches/java-broker-amqp-1-0-management/cpp/src/qpid/linearstore/journal/deq_rec.cpp (original)
+++ qpid/branches/java-broker-amqp-1-0-management/cpp/src/qpid/linearstore/journal/deq_rec.cpp Fri Feb 7 15:24:50 2014
@@ -181,7 +181,7 @@ deq_rec::encode(void* wptr, uint32_t rec
}
bool
-deq_rec::decode(::rec_hdr_t& h, std::ifstream* ifsp, std::size_t& rec_offs)
+deq_rec::decode(::rec_hdr_t& h, std::ifstream* ifsp, std::size_t& rec_offs, const std::streampos rec_start)
{
if (rec_offs == 0)
{
@@ -228,7 +228,7 @@ deq_rec::decode(::rec_hdr_t& h, std::ifs
assert(!ifsp->fail() && !ifsp->bad());
return false;
}
- check_rec_tail();
+ check_rec_tail(rec_start);
}
ifsp->ignore(rec_size_dblks() * QLS_DBLK_SIZE_BYTES - rec_size());
assert(!ifsp->fail() && !ifsp->bad());
@@ -274,7 +274,7 @@ deq_rec::rec_size() const
}
void
-deq_rec::check_rec_tail() const {
+deq_rec::check_rec_tail(const std::streampos rec_start) const {
Checksum checksum;
checksum.addData((const unsigned char*)&_deq_hdr, sizeof(::deq_hdr_t));
if (_deq_hdr._xidsize > 0) {
@@ -284,7 +284,7 @@ deq_rec::check_rec_tail() const {
uint16_t res = ::rec_tail_check(&_deq_tail, &_deq_hdr._rhdr, cs);
if (res != 0) {
std::stringstream oss;
- oss << std::hex;
+ oss << std::endl << " Record offset: 0x" << std::hex << rec_start;
if (res & ::REC_TAIL_MAGIC_ERR_MASK) {
oss << std::endl << " Magic: expected 0x" << ~_deq_hdr._rhdr._magic << "; found 0x" << _deq_tail._xmagic;
}
Modified: qpid/branches/java-broker-amqp-1-0-management/cpp/src/qpid/linearstore/journal/deq_rec.h
URL: http://svn.apache.org/viewvc/qpid/branches/java-broker-amqp-1-0-management/cpp/src/qpid/linearstore/journal/deq_rec.h?rev=1565688&r1=1565687&r2=1565688&view=diff
==============================================================================
--- qpid/branches/java-broker-amqp-1-0-management/cpp/src/qpid/linearstore/journal/deq_rec.h (original)
+++ qpid/branches/java-broker-amqp-1-0-management/cpp/src/qpid/linearstore/journal/deq_rec.h Fri Feb 7 15:24:50 2014
@@ -49,7 +49,7 @@ public:
void reset(const uint64_t serial, const uint64_t rid, const uint64_t drid, const void* const xidp,
const std::size_t xidlen, const bool txn_coml_commit);
uint32_t encode(void* wptr, uint32_t rec_offs_dblks, uint32_t max_size_dblks, Checksum& checksum);
- bool decode(::rec_hdr_t& h, std::ifstream* ifsp, std::size_t& rec_offs);
+ bool decode(::rec_hdr_t& h, std::ifstream* ifsp, std::size_t& rec_offs, const std::streampos rec_start);
inline bool is_txn_coml_commit() const { return ::is_txn_coml_commit(&_deq_hdr); }
inline uint64_t rid() const { return _deq_hdr._rhdr._rid; }
@@ -59,7 +59,7 @@ public:
inline std::size_t data_size() const { return 0; } // This record never carries data
std::size_t xid_size() const;
std::size_t rec_size() const;
- void check_rec_tail() const;
+ void check_rec_tail(const std::streampos rec_start) const;
private:
virtual void clean();
Modified: qpid/branches/java-broker-amqp-1-0-management/cpp/src/qpid/linearstore/journal/enq_rec.cpp
URL: http://svn.apache.org/viewvc/qpid/branches/java-broker-amqp-1-0-management/cpp/src/qpid/linearstore/journal/enq_rec.cpp?rev=1565688&r1=1565687&r2=1565688&view=diff
==============================================================================
--- qpid/branches/java-broker-amqp-1-0-management/cpp/src/qpid/linearstore/journal/enq_rec.cpp (original)
+++ qpid/branches/java-broker-amqp-1-0-management/cpp/src/qpid/linearstore/journal/enq_rec.cpp Fri Feb 7 15:24:50 2014
@@ -218,7 +218,7 @@ enq_rec::encode(void* wptr, uint32_t rec
}
bool
-enq_rec::decode(::rec_hdr_t& h, std::ifstream* ifsp, std::size_t& rec_offs)
+enq_rec::decode(::rec_hdr_t& h, std::ifstream* ifsp, std::size_t& rec_offs, const std::streampos rec_start)
{
if (rec_offs == 0)
{
@@ -291,7 +291,7 @@ enq_rec::decode(::rec_hdr_t& h, std::ifs
assert(!ifsp->fail() && !ifsp->bad());
return false;
}
- check_rec_tail();
+ check_rec_tail(rec_start);
}
ifsp->ignore(rec_size_dblks() * QLS_DBLK_SIZE_BYTES - rec_size());
assert(!ifsp->fail() && !ifsp->bad());
@@ -352,7 +352,7 @@ enq_rec::rec_size(const std::size_t xids
}
void
-enq_rec::check_rec_tail() const {
+enq_rec::check_rec_tail(const std::streampos rec_start) const {
Checksum checksum;
checksum.addData((const unsigned char*)&_enq_hdr, sizeof(::enq_hdr_t));
if (_enq_hdr._xidsize > 0) {
@@ -365,7 +365,7 @@ enq_rec::check_rec_tail() const {
uint16_t res = ::rec_tail_check(&_enq_tail, &_enq_hdr._rhdr, cs);
if (res != 0) {
std::stringstream oss;
- oss << std::hex;
+ oss << std::endl << " Record offset: 0x" << std::hex << rec_start;
if (res & ::REC_TAIL_MAGIC_ERR_MASK) {
oss << std::endl << " Magic: expected 0x" << ~_enq_hdr._rhdr._magic << "; found 0x" << _enq_tail._xmagic;
}
Modified: qpid/branches/java-broker-amqp-1-0-management/cpp/src/qpid/linearstore/journal/enq_rec.h
URL: http://svn.apache.org/viewvc/qpid/branches/java-broker-amqp-1-0-management/cpp/src/qpid/linearstore/journal/enq_rec.h?rev=1565688&r1=1565687&r2=1565688&view=diff
==============================================================================
--- qpid/branches/java-broker-amqp-1-0-management/cpp/src/qpid/linearstore/journal/enq_rec.h (original)
+++ qpid/branches/java-broker-amqp-1-0-management/cpp/src/qpid/linearstore/journal/enq_rec.h Fri Feb 7 15:24:50 2014
@@ -51,7 +51,7 @@ public:
void reset(const uint64_t serial, const uint64_t rid, const void* const dbuf, const std::size_t dlen,
const void* const xidp, const std::size_t xidlen, const bool transient, const bool external);
uint32_t encode(void* wptr, uint32_t rec_offs_dblks, uint32_t max_size_dblks, Checksum& checksum);
- bool decode(::rec_hdr_t& h, std::ifstream* ifsp, std::size_t& rec_offs);
+ bool decode(::rec_hdr_t& h, std::ifstream* ifsp, std::size_t& rec_offs, const std::streampos rec_start);
std::size_t get_xid(void** const xidpp);
std::size_t get_data(void** const datapp);
@@ -63,7 +63,7 @@ public:
std::size_t rec_size() const;
static std::size_t rec_size(const std::size_t xidsize, const std::size_t dsize, const bool external);
inline uint64_t rid() const { return _enq_hdr._rhdr._rid; }
- void check_rec_tail() const;
+ void check_rec_tail(const std::streampos rec_start) const;
private:
virtual void clean();
Modified: qpid/branches/java-broker-amqp-1-0-management/cpp/src/qpid/linearstore/journal/jcntl.cpp
URL: http://svn.apache.org/viewvc/qpid/branches/java-broker-amqp-1-0-management/cpp/src/qpid/linearstore/journal/jcntl.cpp?rev=1565688&r1=1565687&r2=1565688&view=diff
==============================================================================
--- qpid/branches/java-broker-amqp-1-0-management/cpp/src/qpid/linearstore/journal/jcntl.cpp (original)
+++ qpid/branches/java-broker-amqp-1-0-management/cpp/src/qpid/linearstore/journal/jcntl.cpp Fri Feb 7 15:24:50 2014
@@ -21,6 +21,7 @@
#include "qpid/linearstore/journal/jcntl.h"
+#include <iomanip>
#include "qpid/linearstore/journal/data_tok.h"
#include "qpid/linearstore/journal/JournalLog.h"
@@ -90,7 +91,7 @@ jcntl::initialize(EmptyFilePool* efpp,
_linearFileController.finalize();
_jdir.clear_dir(); // Clear any existing journal files
_linearFileController.initialize(_jdir.dirname(), efpp, 0ULL);
- _linearFileController.pullEmptyFileFromEfp();
+ _linearFileController.getNextJournalFile();
_wmgr.initialize(cbp, wcache_pgsize_sblks, wcache_num_pages, QLS_WMGR_MAXDTOKPP, QLS_WMGR_MAXWAITUS);
_init_flag = true;
}
@@ -120,6 +121,9 @@ jcntl::recover(EmptyFilePoolManager* efp
_jrnl_log.log(/*LOG_DEBUG*/JournalLog::LOG_INFO, _jid, _recoveryManager.toLog(_jid, 5));
_linearFileController.initialize(_jdir.dirname(), _emptyFilePoolPtr, _recoveryManager.getHighestFileNumber());
_recoveryManager.setLinearFileControllerJournals(&qpid::linearstore::journal::LinearFileController::addJournalFile, &_linearFileController);
+ if (_recoveryManager.isLastFileFull()) {
+ _linearFileController.getNextJournalFile();
+ }
_wmgr.initialize(cbp, wcache_pgsize_sblks, wcache_num_pages, QLS_WMGR_MAXDTOKPP, QLS_WMGR_MAXWAITUS,
(_recoveryManager.isLastFileFull() ? 0 : _recoveryManager.getEndOffset()));
@@ -316,6 +320,20 @@ jcntl::getLinearFileControllerRef() {
return _linearFileController;
}
+// static
+std::string
+jcntl::str2hexnum(const std::string& str) {
+ if (str.empty()) {
+ return "<null>";
+ }
+ std::ostringstream oss;
+ oss << "(" << str.size() << ")0x" << std::hex;
+ for (unsigned i=str.size(); i>0; --i) {
+ oss << std::setfill('0') << std::setw(2) << (uint16_t)(uint8_t)str[i-1];
+ }
+ return oss.str();
+}
+
iores
jcntl::flush(const bool block_till_aio_cmpl)
{
Modified: qpid/branches/java-broker-amqp-1-0-management/cpp/src/qpid/linearstore/journal/jcntl.h
URL: http://svn.apache.org/viewvc/qpid/branches/java-broker-amqp-1-0-management/cpp/src/qpid/linearstore/journal/jcntl.h?rev=1565688&r1=1565687&r2=1565688&view=diff
==============================================================================
--- qpid/branches/java-broker-amqp-1-0-management/cpp/src/qpid/linearstore/journal/jcntl.h (original)
+++ qpid/branches/java-broker-amqp-1-0-management/cpp/src/qpid/linearstore/journal/jcntl.h Fri Feb 7 15:24:50 2014
@@ -537,6 +537,8 @@ public:
inline virtual void instr_incr_outstanding_aio_cnt() {}
inline virtual void instr_decr_outstanding_aio_cnt() {}
+ static std::string str2hexnum(const std::string& str);
+
protected:
static bool _init;
static bool init_statics();
Modified: qpid/branches/java-broker-amqp-1-0-management/cpp/src/qpid/linearstore/journal/jerrno.cpp
URL: http://svn.apache.org/viewvc/qpid/branches/java-broker-amqp-1-0-management/cpp/src/qpid/linearstore/journal/jerrno.cpp?rev=1565688&r1=1565687&r2=1565688&view=diff
==============================================================================
--- qpid/branches/java-broker-amqp-1-0-management/cpp/src/qpid/linearstore/journal/jerrno.cpp (original)
+++ qpid/branches/java-broker-amqp-1-0-management/cpp/src/qpid/linearstore/journal/jerrno.cpp Fri Feb 7 15:24:50 2014
@@ -167,8 +167,8 @@ jerrno::__init()
_err_map[JERR_LFCR_SEQNUMNOTFOUND] = "JERR_LFCR_SEQNUMNOTFOUND: File sequence number not found";
// class jrec, enq_rec, deq_rec, txn_rec
- _err_map[JERR_JREC_BADRECHDR] = "JERR_JREC_BADRECHDR: Invalid data record header.";
- _err_map[JERR_JREC_BADRECTAIL] = "JERR_JREC_BADRECTAIL: Invalid data record tail.";
+ _err_map[JERR_JREC_BADRECHDR] = "JERR_JREC_BADRECHDR: Invalid record header.";
+ _err_map[JERR_JREC_BADRECTAIL] = "JERR_JREC_BADRECTAIL: Invalid record tail.";
// class wmgr
_err_map[JERR_WMGR_BADPGSTATE] = "JERR_WMGR_BADPGSTATE: Page buffer in illegal state for operation.";
Modified: qpid/branches/java-broker-amqp-1-0-management/cpp/src/qpid/linearstore/journal/jrec.h
URL: http://svn.apache.org/viewvc/qpid/branches/java-broker-amqp-1-0-management/cpp/src/qpid/linearstore/journal/jrec.h?rev=1565688&r1=1565687&r2=1565688&view=diff
==============================================================================
--- qpid/branches/java-broker-amqp-1-0-management/cpp/src/qpid/linearstore/journal/jrec.h (original)
+++ qpid/branches/java-broker-amqp-1-0-management/cpp/src/qpid/linearstore/journal/jrec.h Fri Feb 7 15:24:50 2014
@@ -98,7 +98,7 @@ public:
* \returns Number of data-blocks encoded.
*/
virtual uint32_t encode(void* wptr, uint32_t rec_offs_dblks, uint32_t max_size_dblks, Checksum& checksum) = 0;
- virtual bool decode(::rec_hdr_t& h, std::ifstream* ifsp, std::size_t& rec_offs) = 0;
+ virtual bool decode(::rec_hdr_t& h, std::ifstream* ifsp, std::size_t& rec_offs, const std::streampos rec_start) = 0;
virtual std::string& str(std::string& str) const = 0;
virtual std::size_t data_size() const = 0;
Modified: qpid/branches/java-broker-amqp-1-0-management/cpp/src/qpid/linearstore/journal/txn_rec.cpp
URL: http://svn.apache.org/viewvc/qpid/branches/java-broker-amqp-1-0-management/cpp/src/qpid/linearstore/journal/txn_rec.cpp?rev=1565688&r1=1565687&r2=1565688&view=diff
==============================================================================
--- qpid/branches/java-broker-amqp-1-0-management/cpp/src/qpid/linearstore/journal/txn_rec.cpp (original)
+++ qpid/branches/java-broker-amqp-1-0-management/cpp/src/qpid/linearstore/journal/txn_rec.cpp Fri Feb 7 15:24:50 2014
@@ -176,7 +176,7 @@ txn_rec::encode(void* wptr, uint32_t rec
}
bool
-txn_rec::decode(::rec_hdr_t& h, std::ifstream* ifsp, std::size_t& rec_offs)
+txn_rec::decode(::rec_hdr_t& h, std::ifstream* ifsp, std::size_t& rec_offs, const std::streampos rec_start)
{
if (rec_offs == 0)
{
@@ -218,7 +218,7 @@ txn_rec::decode(::rec_hdr_t& h, std::ifs
assert(!ifsp->fail() && !ifsp->bad());
return false;
}
- check_rec_tail();
+ check_rec_tail(rec_start);
}
ifsp->ignore(rec_size_dblks() * QLS_DBLK_SIZE_BYTES - rec_size());
assert(!ifsp->fail() && !ifsp->bad());
@@ -266,7 +266,7 @@ txn_rec::rec_size() const
}
void
-txn_rec::check_rec_tail() const {
+txn_rec::check_rec_tail(const std::streampos rec_start) const {
Checksum checksum;
checksum.addData((const unsigned char*)&_txn_hdr, sizeof(::txn_hdr_t));
if (_txn_hdr._xidsize > 0) {
@@ -276,7 +276,7 @@ txn_rec::check_rec_tail() const {
uint16_t res = ::rec_tail_check(&_txn_tail, &_txn_hdr._rhdr, cs);
if (res != 0) {
std::stringstream oss;
- oss << std::hex;
+ oss << std::endl << " Record offset: 0x" << std::hex << rec_start;
if (res & ::REC_TAIL_MAGIC_ERR_MASK) {
oss << std::endl << " Magic: expected 0x" << ~_txn_hdr._rhdr._magic << "; found 0x" << _txn_tail._xmagic;
}
Modified: qpid/branches/java-broker-amqp-1-0-management/cpp/src/qpid/linearstore/journal/txn_rec.h
URL: http://svn.apache.org/viewvc/qpid/branches/java-broker-amqp-1-0-management/cpp/src/qpid/linearstore/journal/txn_rec.h?rev=1565688&r1=1565687&r2=1565688&view=diff
==============================================================================
--- qpid/branches/java-broker-amqp-1-0-management/cpp/src/qpid/linearstore/journal/txn_rec.h (original)
+++ qpid/branches/java-broker-amqp-1-0-management/cpp/src/qpid/linearstore/journal/txn_rec.h Fri Feb 7 15:24:50 2014
@@ -49,7 +49,7 @@ public:
void reset(const bool commitFlag, const uint64_t serial, const uint64_t rid, const void* const xidp,
const std::size_t xidlen);
uint32_t encode(void* wptr, uint32_t rec_offs_dblks, uint32_t max_size_dblks, Checksum& checksum);
- bool decode(::rec_hdr_t& h, std::ifstream* ifsp, std::size_t& rec_offs);
+ bool decode(::rec_hdr_t& h, std::ifstream* ifsp, std::size_t& rec_offs, const std::streampos rec_start);
std::size_t get_xid(void** const xidpp);
std::string& str(std::string& str) const;
@@ -57,7 +57,7 @@ public:
std::size_t xid_size() const;
std::size_t rec_size() const;
inline uint64_t rid() const { return _txn_hdr._rhdr._rid; }
- void check_rec_tail() const;
+ void check_rec_tail(const std::streampos rec_start) const;
private:
virtual void clean();
Modified: qpid/branches/java-broker-amqp-1-0-management/cpp/src/qpid/linearstore/journal/wmgr.cpp
URL: http://svn.apache.org/viewvc/qpid/branches/java-broker-amqp-1-0-management/cpp/src/qpid/linearstore/journal/wmgr.cpp?rev=1565688&r1=1565687&r2=1565688&view=diff
==============================================================================
--- qpid/branches/java-broker-amqp-1-0-management/cpp/src/qpid/linearstore/journal/wmgr.cpp (original)
+++ qpid/branches/java-broker-amqp-1-0-management/cpp/src/qpid/linearstore/journal/wmgr.cpp Fri Feb 7 15:24:50 2014
@@ -30,8 +30,6 @@
#include "qpid/linearstore/journal/LinearFileController.h"
#include "qpid/linearstore/journal/utils/file_hdr.h"
-//#include <iostream> // DEBUG
-
namespace qpid {
namespace linearstore {
namespace journal {
@@ -49,7 +47,7 @@ wmgr::wmgr(jcntl* jc,
_deq_busy(false),
_abort_busy(false),
_commit_busy(false),
- _txn_pending_set()
+ _txn_pending_map()
{}
wmgr::wmgr(jcntl* jc,
@@ -67,7 +65,7 @@ wmgr::wmgr(jcntl* jc,
_deq_busy(false),
_abort_busy(false),
_commit_busy(false),
- _txn_pending_set()
+ _txn_pending_map()
{}
wmgr::~wmgr()
@@ -281,6 +279,7 @@ wmgr::dequeue(data_tok* dtokp,
_deq_busy = true;
}
//std::cout << "---+++ wmgr::dequeue() DEQ rid=0x" << std::hex << rid << " drid=0x" << dequeue_rid << " " << std::dec << std::flush; // DEBUG
+ std::string xid((const char*)xid_ptr, xid_len);
bool done = false;
Checksum checksum;
while (!done)
@@ -292,9 +291,27 @@ wmgr::dequeue(data_tok* dtokp,
uint32_t ret = _deq_rec.encode(wptr, data_offs_dblks,
(_cache_pgsize_sblks * QLS_SBLK_SIZE_DBLKS) - _pg_offset_dblks, checksum);
- // Remember fid which contains the record header in case record is split over several files
if (data_offs_dblks == 0) {
- dtokp->set_fid(_lfc.getCurrentFileSeqNum());
+ uint64_t fid;
+ short eres = _emap.get_pfid(dtokp->dequeue_rid(), fid);
+ if (eres == enq_map::EMAP_OK) {
+ dtokp->set_fid(fid);
+ } else if (xid_len > 0) {
+ txn_data_list_t tdl = _tmap.get_tdata_list(xid);
+ bool found = false;
+ for (tdl_const_itr_t i=tdl.begin(); i!=tdl.end() && !found; ++i) {
+ if (i->rid_ == dtokp->dequeue_rid()) {
+ found = true;
+ dtokp->set_fid(i->pfid_);
+ break;
+ }
+ }
+ if (!found) {
+ throw jexception("rid found in neither emap nor tmap, transactional");
+ }
+ } else {
+ throw jexception("rid not found in emap, non-transactional");
+ }
}
_pg_offset_dblks += ret;
_cached_offset_dblks += ret;
@@ -325,7 +342,7 @@ wmgr::dequeue(data_tok* dtokp,
if (eres == enq_map::EMAP_RID_NOT_FOUND)
{
std::ostringstream oss;
- oss << std::hex << "rid=0x" << rid;
+ oss << std::hex << "emap: rid=0x" << rid;
throw jexception(jerrno::JERR_MAP_NOTFOUND, oss.str(), "wmgr", "dequeue");
}
if (eres == enq_map::EMAP_LOCKED)
@@ -335,10 +352,6 @@ wmgr::dequeue(data_tok* dtokp,
throw jexception(jerrno::JERR_MAP_LOCKED, oss.str(), "wmgr", "dequeue");
}
}
-//std::cout << "[0x" << std::hex << _lfc.getEnqueuedRecordCount(fid) << std::dec << std::flush; // DEBUG
-//try {
- _lfc.decrEnqueuedRecordCount(fid);
-//} catch (std::exception& e) { std::cout << "***OOPS*** " << e.what() << " cfid=" << _lfc.getCurrentFileSeqNum() << " fid=" << fid << std::flush; throw; }
}
done = true;
@@ -427,14 +440,16 @@ wmgr::abort(data_tok* dtokp,
// Delete this txn from tmap, unlock any locked records in emap
std::string xid((const char*)xid_ptr, xid_len);
txn_data_list_t tdl = _tmap.get_remove_tdata_list(xid); // tdl will be empty if xid not found
+ fidl_t fidl;
for (tdl_itr_t itr = tdl.begin(); itr != tdl.end(); itr++)
{
if (!itr->enq_flag_)
_emap.unlock(itr->drid_); // ignore rid not found error
- if (itr->enq_flag_)
- _lfc.decrEnqueuedRecordCount(itr->pfid_);
+ if (itr->enq_flag_) {
+ fidl.push_back(itr->pfid_);
+ }
}
- std::pair<std::set<std::string>::iterator, bool> res = _txn_pending_set.insert(xid);
+ std::pair<pending_txn_map_itr_t, bool> res = _txn_pending_map.insert(std::pair<std::string, fidl_t>(xid, fidl));
if (!res.second)
{
std::ostringstream oss;
@@ -526,6 +541,7 @@ wmgr::commit(data_tok* dtokp,
// Delete this txn from tmap, process records into emap
std::string xid((const char*)xid_ptr, xid_len);
txn_data_list_t tdl = _tmap.get_remove_tdata_list(xid); // tdl will be empty if xid not found
+ fidl_t fidl;
for (tdl_itr_t itr = tdl.begin(); itr != tdl.end(); itr++)
{
if (itr->enq_flag_) // txn enqueue
@@ -547,20 +563,20 @@ wmgr::commit(data_tok* dtokp,
if (eres == enq_map::EMAP_RID_NOT_FOUND)
{
std::ostringstream oss;
- oss << std::hex << "rid=0x" << rid;
- throw jexception(jerrno::JERR_MAP_NOTFOUND, oss.str(), "wmgr", "dequeue");
+ oss << std::hex << "emap: rid=0x" << itr->drid_;
+ throw jexception(jerrno::JERR_MAP_NOTFOUND, oss.str(), "wmgr", "commit");
}
if (eres == enq_map::EMAP_LOCKED)
{
std::ostringstream oss;
- oss << std::hex << "rid=0x" << rid;
- throw jexception(jerrno::JERR_MAP_LOCKED, oss.str(), "wmgr", "dequeue");
+ oss << std::hex << "rid=0x" << itr->drid_;
+ throw jexception(jerrno::JERR_MAP_LOCKED, oss.str(), "wmgr", "commit");
}
}
- _lfc.decrEnqueuedRecordCount(fid);
+ fidl.push_back(fid);
}
}
- std::pair<std::set<std::string>::iterator, bool> res = _txn_pending_set.insert(xid);
+ std::pair<pending_txn_map_itr_t, bool> res = _txn_pending_map.insert(std::pair<std::string, fidl_t>(xid, fidl));
if (!res.second)
{
std::ostringstream oss;
@@ -695,7 +711,7 @@ wmgr::get_next_file()
{
_pg_cntr = 0;
//std::cout << "&&&&& wmgr::get_next_file(): " << status_str() << std::flush << std::endl; // DEBUG
- _lfc.pullEmptyFileFromEfp();
+ _lfc.getNextJournalFile();
}
int32_t
@@ -757,7 +773,7 @@ wmgr::get_events(timespec* const timeout
data_tok* dtokp = pcbp->_pdtokl->at(k);
if (dtokp->decr_pg_cnt() == 0)
{
- std::set<std::string>::iterator it;
+ pending_txn_map_itr_t it;
switch (dtokp->wstate())
{
case data_tok::ENQ_SUBM:
@@ -770,6 +786,9 @@ wmgr::get_events(timespec* const timeout
_tmap.set_aio_compl(dtokp->xid(), dtokp->rid());
break;
case data_tok::DEQ_SUBM:
+ if (!dtokp->has_xid()) {
+ _lfc.decrEnqueuedRecordCount(dtokp->fid());
+ }
dtokl.push_back(dtokp);
tot_data_toks++;
dtokp->set_wstate(data_tok::DEQ);
@@ -781,31 +800,35 @@ wmgr::get_events(timespec* const timeout
dtokl.push_back(dtokp);
tot_data_toks++;
dtokp->set_wstate(data_tok::ABORTED);
- it = _txn_pending_set.find(dtokp->xid());
- if (it == _txn_pending_set.end())
+ it = _txn_pending_map.find(dtokp->xid());
+ if (it == _txn_pending_map.end())
{
std::ostringstream oss;
- oss << std::hex << "_txn_pending_set: abort xid=\"";
- oss << dtokp->xid() << "\"";
- throw jexception(jerrno::JERR_MAP_NOTFOUND, oss.str(), "wmgr",
- "get_events");
+ oss << std::hex << "_txn_pending_set: abort xid=\""
+ << qpid::linearstore::journal::jcntl::str2hexnum(dtokp->xid()) << "\"";
+ throw jexception(jerrno::JERR_MAP_NOTFOUND, oss.str(), "wmgr", "get_events");
+ }
+ for (fidl_itr_t i=it->second.begin(); i!=it->second.end(); ++i) {
+ _lfc.decrEnqueuedRecordCount(*i);
}
- _txn_pending_set.erase(it);
+ _txn_pending_map.erase(it);
break;
case data_tok::COMMIT_SUBM:
dtokl.push_back(dtokp);
tot_data_toks++;
dtokp->set_wstate(data_tok::COMMITTED);
- it = _txn_pending_set.find(dtokp->xid());
- if (it == _txn_pending_set.end())
+ it = _txn_pending_map.find(dtokp->xid());
+ if (it == _txn_pending_map.end())
{
std::ostringstream oss;
- oss << std::hex << "_txn_pending_set: commit xid=\"";
- oss << dtokp->xid() << "\"";
- throw jexception(jerrno::JERR_MAP_NOTFOUND, oss.str(), "wmgr",
- "get_events");
+ oss << std::hex << "_txn_pending_set: commit xid=\""
+ << qpid::linearstore::journal::jcntl::str2hexnum(dtokp->xid()) << "\"";
+ throw jexception(jerrno::JERR_MAP_NOTFOUND, oss.str(), "wmgr", "get_events");
+ }
+ for (fidl_itr_t i=it->second.begin(); i!=it->second.end(); ++i) {
+ _lfc.decrEnqueuedRecordCount(*i);
}
- _txn_pending_set.erase(it);
+ _txn_pending_map.erase(it);
break;
case data_tok::ENQ_PART:
case data_tok::DEQ_PART:
@@ -858,8 +881,8 @@ wmgr::is_txn_synced(const std::string& x
if (_tmap.is_txn_synced(xid) == txn_map::TMAP_NOT_SYNCED)
return false;
// Check for outstanding commit/aborts
- std::set<std::string>::iterator it = _txn_pending_set.find(xid);
- return it == _txn_pending_set.end();
+ pending_txn_map_itr_t it = _txn_pending_map.find(xid);
+ return it == _txn_pending_map.end();
}
void
@@ -871,7 +894,6 @@ wmgr::initialize(aio_callback* const cbp
pmgr::initialize(cbp, wcache_pgsize_sblks, wcache_num_pages);
wmgr::clean();
_page_cb_arr[0]._state = IN_USE;
- _ddtokl.clear();
_cached_offset_dblks = 0;
_enq_busy = false;
}
Modified: qpid/branches/java-broker-amqp-1-0-management/cpp/src/qpid/linearstore/journal/wmgr.h
URL: http://svn.apache.org/viewvc/qpid/branches/java-broker-amqp-1-0-management/cpp/src/qpid/linearstore/journal/wmgr.h?rev=1565688&r1=1565687&r2=1565688&view=diff
==============================================================================
--- qpid/branches/java-broker-amqp-1-0-management/cpp/src/qpid/linearstore/journal/wmgr.h (original)
+++ qpid/branches/java-broker-amqp-1-0-management/cpp/src/qpid/linearstore/journal/wmgr.h Fri Feb 7 15:24:50 2014
@@ -22,9 +22,11 @@
#ifndef QPID_LINEARSTORE_JOURNAL_WMGR_H
#define QPID_LINEARSTORE_JOURNAL_WMGR_H
+#include <deque>
+#include <map>
#include "qpid/linearstore/journal/enums.h"
#include "qpid/linearstore/journal/pmgr.h"
-#include <set>
+#include <vector>
namespace qpid {
namespace linearstore {
@@ -51,11 +53,15 @@ class LinearFileController;
class wmgr : public pmgr
{
private:
+ typedef std::vector<uint64_t> fidl_t;
+ typedef fidl_t::iterator fidl_itr_t;
+ typedef std::map<std::string, fidl_t> pending_txn_map_t;
+ typedef pending_txn_map_t::iterator pending_txn_map_itr_t;
+
LinearFileController& _lfc; ///< Linear File Controller ref
uint32_t _max_dtokpp; ///< Max data writes per page
uint32_t _max_io_wait_us; ///< Max wait in microseconds till submit
uint32_t _cached_offset_dblks; ///< Amount of unwritten data in page (dblocks)
- std::deque<data_tok*> _ddtokl; ///< Deferred dequeue data_tok list
// TODO: Convert _enq_busy etc into a proper threadsafe lock
// TODO: Convert to enum? Are these encodes mutually exclusive?
@@ -70,7 +76,7 @@ private:
enq_rec _enq_rec; ///< Enqueue record used for encoding/decoding
deq_rec _deq_rec; ///< Dequeue record used for encoding/decoding
txn_rec _txn_rec; ///< Transaction record used for encoding/decoding
- std::set<std::string> _txn_pending_set; ///< Set containing xids of pending commits/aborts
+ pending_txn_map_t _txn_pending_map; ///< Set containing xids of pending commits/aborts
public:
wmgr(jcntl* jc,
Modified: qpid/branches/java-broker-amqp-1-0-management/cpp/src/qpid/log/Logger.cpp
URL: http://svn.apache.org/viewvc/qpid/branches/java-broker-amqp-1-0-management/cpp/src/qpid/log/Logger.cpp?rev=1565688&r1=1565687&r2=1565688&view=diff
==============================================================================
--- qpid/branches/java-broker-amqp-1-0-management/cpp/src/qpid/log/Logger.cpp (original)
+++ qpid/branches/java-broker-amqp-1-0-management/cpp/src/qpid/log/Logger.cpp Fri Feb 7 15:24:50 2014
@@ -106,7 +106,7 @@ void Logger::log(const Statement& s, con
os << s.file << ":";
if (flags&LINE)
os << dec << s.line << ":";
- if (flags&FUNCTION)
+ if ((flags&FUNCTION) && s.function)
os << s.function << ":";
if (flags & (FILE|LINE|FUNCTION))
os << " ";
Modified: qpid/branches/java-broker-amqp-1-0-management/cpp/src/qpid/log/Statement.cpp
URL: http://svn.apache.org/viewvc/qpid/branches/java-broker-amqp-1-0-management/cpp/src/qpid/log/Statement.cpp?rev=1565688&r1=1565687&r2=1565688&view=diff
==============================================================================
--- qpid/branches/java-broker-amqp-1-0-management/cpp/src/qpid/log/Statement.cpp (original)
+++ qpid/branches/java-broker-amqp-1-0-management/cpp/src/qpid/log/Statement.cpp Fri Feb 7 15:24:50 2014
@@ -144,54 +144,33 @@ void Statement::log(const std::string& m
Statement::Initializer::Initializer(Statement& s) : statement(s) {
- // QPID-3891
+ // QPID-3891:
// From the given BOOST_CURRENT_FUNCTION name extract only the
// namespace-qualified-functionName, skipping return and calling args.
// Given:
// <possible return value type> qpid::name::space::Function(args)
// Return:
// "qpid::name::space::Function".
- if (s.function != NULL) {
- bool foundOParen(false);
- const char * opPtr;
- for (opPtr = s.function; *opPtr != '\0'; opPtr++) {
- if (*opPtr == '(') {
- foundOParen = true;
- break;
- }
- }
-
- if (foundOParen) {
- const char * bPtr = opPtr;
- for (bPtr = opPtr; bPtr > s.function; bPtr--) {
- if (bPtr[-1] == ' ') {
- break;
- }
- }
-
- size_t nStoreSize = opPtr - bPtr;
- if (nStoreSize > 0) {
- // Note: the struct into which this name is stored
- // is static and is never deleted.
- char * nStore = new char[nStoreSize + 1];
- std::copy (bPtr, opPtr, nStore);
- nStore[nStoreSize] = '\0';
-
- s.function = nStore;
- } else {
- // Ignore zero length name
- }
- } else {
- // No name found - do nothing
- }
- } else {
- // no function-name pointer to process
+ if (s.function) {
+ const char* end = s.function + strlen(s.function);
+ const char* fEnd = std::find(s.function, end, '(');
+ typedef std::reverse_iterator<const char*> Reverse;
+ const char* fBegin = find(Reverse(fEnd), Reverse(s.function), ' ').base();
+ size_t n = fEnd - fBegin;
+ char* name = new char[n+1];
+ std::copy(fBegin, fEnd, name);
+ name[n] = '\0';
+ s.function = name;
}
-
Statement::categorize(s);
Logger::instance().add(s);
}
+Statement::Initializer::~Initializer() {
+ delete[] const_cast<char*>(statement.function);
+ statement.function = 0;
+}
+
namespace {
const char* names[LevelTraits::COUNT] = {
Modified: qpid/branches/java-broker-amqp-1-0-management/cpp/src/qpid/log/Statement.h
URL: http://svn.apache.org/viewvc/qpid/branches/java-broker-amqp-1-0-management/cpp/src/qpid/log/Statement.h?rev=1565688&r1=1565687&r2=1565688&view=diff
==============================================================================
--- qpid/branches/java-broker-amqp-1-0-management/cpp/src/qpid/log/Statement.h (original)
+++ qpid/branches/java-broker-amqp-1-0-management/cpp/src/qpid/log/Statement.h Fri Feb 7 15:24:50 2014
@@ -114,6 +114,7 @@ struct Statement {
struct Initializer {
QPID_COMMON_EXTERN Initializer(Statement& s);
+ QPID_COMMON_EXTERN ~Initializer();
Statement& statement;
};
};
Modified: qpid/branches/java-broker-amqp-1-0-management/cpp/src/qpid/messaging/amqp/ConnectionContext.cpp
URL: http://svn.apache.org/viewvc/qpid/branches/java-broker-amqp-1-0-management/cpp/src/qpid/messaging/amqp/ConnectionContext.cpp?rev=1565688&r1=1565687&r2=1565688&view=diff
==============================================================================
--- qpid/branches/java-broker-amqp-1-0-management/cpp/src/qpid/messaging/amqp/ConnectionContext.cpp (original)
+++ qpid/branches/java-broker-amqp-1-0-management/cpp/src/qpid/messaging/amqp/ConnectionContext.cpp Fri Feb 7 15:24:50 2014
@@ -819,6 +819,11 @@ qpid::sys::Codec& ConnectionContext::get
return *this;
}
+const qpid::messaging::ConnectionOptions* ConnectionContext::getOptions()
+{
+ return this;
+}
+
std::size_t ConnectionContext::decode(const char* buffer, std::size_t size)
{
qpid::sys::ScopedLock<qpid::sys::Monitor> l(lock);
Modified: qpid/branches/java-broker-amqp-1-0-management/cpp/src/qpid/messaging/amqp/ConnectionContext.h
URL: http://svn.apache.org/viewvc/qpid/branches/java-broker-amqp-1-0-management/cpp/src/qpid/messaging/amqp/ConnectionContext.h?rev=1565688&r1=1565687&r2=1565688&view=diff
==============================================================================
--- qpid/branches/java-broker-amqp-1-0-management/cpp/src/qpid/messaging/amqp/ConnectionContext.h (original)
+++ qpid/branches/java-broker-amqp-1-0-management/cpp/src/qpid/messaging/amqp/ConnectionContext.h Fri Feb 7 15:24:50 2014
@@ -102,6 +102,7 @@ class ConnectionContext : public qpid::s
void activateOutput();
qpid::sys::Codec& getCodec();
+ const qpid::messaging::ConnectionOptions* getOptions();
//ConnectionCodec interface:
std::size_t decode(const char* buffer, std::size_t size);
std::size_t encode(char* buffer, std::size_t size);
Modified: qpid/branches/java-broker-amqp-1-0-management/cpp/src/qpid/messaging/amqp/SslTransport.cpp
URL: http://svn.apache.org/viewvc/qpid/branches/java-broker-amqp-1-0-management/cpp/src/qpid/messaging/amqp/SslTransport.cpp?rev=1565688&r1=1565687&r2=1565688&view=diff
==============================================================================
--- qpid/branches/java-broker-amqp-1-0-management/cpp/src/qpid/messaging/amqp/SslTransport.cpp (original)
+++ qpid/branches/java-broker-amqp-1-0-management/cpp/src/qpid/messaging/amqp/SslTransport.cpp Fri Feb 7 15:24:50 2014
@@ -20,6 +20,7 @@
*/
#include "SslTransport.h"
#include "TransportContext.h"
+#include "qpid/messaging/ConnectionOptions.h"
#include "qpid/sys/ssl/SslSocket.h"
#include "qpid/sys/AsynchIO.h"
#include "qpid/sys/ConnectionCodec.h"
@@ -52,7 +53,14 @@ struct StaticInit
}
-SslTransport::SslTransport(TransportContext& c, boost::shared_ptr<Poller> p) : context(c), connector(0), aio(0), poller(p) {}
+SslTransport::SslTransport(TransportContext& c, boost::shared_ptr<Poller> p) : context(c), connector(0), aio(0), poller(p)
+{
+ const ConnectionOptions* options = context.getOptions();
+ if (options->sslCertName != "") {
+ QPID_LOG(debug, "ssl-cert-name = " << options->sslCertName);
+ socket.setCertName(options->sslCertName);
+ }
+}
void SslTransport::connect(const std::string& host, const std::string& port)
{
Modified: qpid/branches/java-broker-amqp-1-0-management/cpp/src/qpid/messaging/amqp/TransportContext.h
URL: http://svn.apache.org/viewvc/qpid/branches/java-broker-amqp-1-0-management/cpp/src/qpid/messaging/amqp/TransportContext.h?rev=1565688&r1=1565687&r2=1565688&view=diff
==============================================================================
--- qpid/branches/java-broker-amqp-1-0-management/cpp/src/qpid/messaging/amqp/TransportContext.h (original)
+++ qpid/branches/java-broker-amqp-1-0-management/cpp/src/qpid/messaging/amqp/TransportContext.h Fri Feb 7 15:24:50 2014
@@ -26,6 +26,8 @@ namespace sys {
class Codec;
}
namespace messaging {
+class ConnectionOptions;
+
namespace amqp {
/**
@@ -38,6 +40,7 @@ class TransportContext
public:
virtual ~TransportContext() {}
virtual qpid::sys::Codec& getCodec() = 0;
+ virtual const qpid::messaging::ConnectionOptions* getOptions() = 0;
virtual void closed() = 0;
virtual void opened() = 0;
private:
Modified: qpid/branches/java-broker-amqp-1-0-management/cpp/src/qpid/sys/windows/SslAsynchIO.cpp
URL: http://svn.apache.org/viewvc/qpid/branches/java-broker-amqp-1-0-management/cpp/src/qpid/sys/windows/SslAsynchIO.cpp?rev=1565688&r1=1565687&r2=1565688&view=diff
==============================================================================
--- qpid/branches/java-broker-amqp-1-0-management/cpp/src/qpid/sys/windows/SslAsynchIO.cpp (original)
+++ qpid/branches/java-broker-amqp-1-0-management/cpp/src/qpid/sys/windows/SslAsynchIO.cpp Fri Feb 7 15:24:50 2014
@@ -182,15 +182,19 @@ void SslAsynchIO::notifyPendingWrite() {
}
void SslAsynchIO::queueWriteClose() {
- if (state == Negotiating) {
- // Never got going, so don't bother trying to close SSL down orderly.
+ {
+ qpid::sys::Mutex::ScopedLock l(lock);
+ if (state == Negotiating) {
+ // Never got going, so don't bother trying to close SSL down orderly.
+ state = ShuttingDown;
+ aio->queueWriteClose();
+ return;
+ }
+ if (state == ShuttingDown)
+ return;
state = ShuttingDown;
- aio->queueWriteClose();
- return;
}
- state = ShuttingDown;
-
DWORD shutdown = SCHANNEL_SHUTDOWN;
SecBuffer shutBuff;
shutBuff.cbBuffer = sizeof(DWORD);
Modified: qpid/branches/java-broker-amqp-1-0-management/cpp/src/qpid/sys/windows/SslAsynchIO.h
URL: http://svn.apache.org/viewvc/qpid/branches/java-broker-amqp-1-0-management/cpp/src/qpid/sys/windows/SslAsynchIO.h?rev=1565688&r1=1565687&r2=1565688&view=diff
==============================================================================
--- qpid/branches/java-broker-amqp-1-0-management/cpp/src/qpid/sys/windows/SslAsynchIO.h (original)
+++ qpid/branches/java-broker-amqp-1-0-management/cpp/src/qpid/sys/windows/SslAsynchIO.h Fri Feb 7 15:24:50 2014
@@ -26,6 +26,7 @@
#include "qpid/sys/IntegerTypes.h"
#include "qpid/sys/Poller.h"
#include "qpid/CommonImportExport.h"
+#include "qpid/sys/Mutex.h"
#include <boost/function.hpp>
#include <boost/shared_ptr.hpp>
#include <windows.h>
@@ -86,6 +87,7 @@ protected:
// AsynchIO layer below that's actually doing the I/O
qpid::sys::AsynchIO *aio;
+ Mutex lock;
// Track what the state of the SSL session is. Have to know when it's
// time to notify the upper layer that the session is up, and also to
Propchange: qpid/branches/java-broker-amqp-1-0-management/cpp/src/tests/
------------------------------------------------------------------------------
Merged /qpid/trunk/qpid/cpp/src/tests:r1564582-1565686
Modified: qpid/branches/java-broker-amqp-1-0-management/cpp/src/tests/brokertest.py
URL: http://svn.apache.org/viewvc/qpid/branches/java-broker-amqp-1-0-management/cpp/src/tests/brokertest.py?rev=1565688&r1=1565687&r2=1565688&view=diff
==============================================================================
--- qpid/branches/java-broker-amqp-1-0-management/cpp/src/tests/brokertest.py (original)
+++ qpid/branches/java-broker-amqp-1-0-management/cpp/src/tests/brokertest.py Fri Feb 7 15:24:50 2014
@@ -278,7 +278,7 @@ class Broker(Popen):
cmd += ["--log-to-stderr=no"]
# Add default --log-enable arguments unless args already has --log arguments.
- if not next((l for l in args if l.startswith("--log")), None):
+ if not [l for l in args if l.startswith("--log")]:
args += ["--log-enable=info+"]
if test_store: cmd += ["--load-module", BrokerTest.test_store_lib,
Modified: qpid/branches/java-broker-amqp-1-0-management/cpp/src/tests/ha_test.py
URL: http://svn.apache.org/viewvc/qpid/branches/java-broker-amqp-1-0-management/cpp/src/tests/ha_test.py?rev=1565688&r1=1565687&r2=1565688&view=diff
==============================================================================
--- qpid/branches/java-broker-amqp-1-0-management/cpp/src/tests/ha_test.py (original)
+++ qpid/branches/java-broker-amqp-1-0-management/cpp/src/tests/ha_test.py Fri Feb 7 15:24:50 2014
@@ -137,7 +137,7 @@ class HaBroker(Broker):
"--max-negotiate-time=1000",
"--ha-cluster=%s"%ha_cluster]
# Add default --log-enable arguments unless args already has --log arguments.
- if not next((l for l in args if l.startswith("--log")), None):
+ if not [l for l in args if l.startswith("--log")]:
args += ["--log-enable=info+", "--log-enable=debug+:ha::"]
if ha_replicate is not None:
args += [ "--ha-replicate=%s"%ha_replicate ]
Propchange: qpid/branches/java-broker-amqp-1-0-management/java/
------------------------------------------------------------------------------
Merged /qpid/trunk/qpid/java:r1564447-1565686
Propchange: qpid/branches/java-broker-amqp-1-0-management/java/amqp-1-0-client/
------------------------------------------------------------------------------
Merged /qpid/trunk/qpid/java/amqp-1-0-client:r1562456-1565686
Propchange: qpid/branches/java-broker-amqp-1-0-management/java/amqp-1-0-client-jms/
------------------------------------------------------------------------------
Merged /qpid/trunk/qpid/java/amqp-1-0-client-jms:r1562456-1565686
Modified: qpid/branches/java-broker-amqp-1-0-management/java/amqp-1-0-client-jms/pom.xml
URL: http://svn.apache.org/viewvc/qpid/branches/java-broker-amqp-1-0-management/java/amqp-1-0-client-jms/pom.xml?rev=1565688&r1=1565687&r2=1565688&view=diff
==============================================================================
--- qpid/branches/java-broker-amqp-1-0-management/java/amqp-1-0-client-jms/pom.xml (original)
+++ qpid/branches/java-broker-amqp-1-0-management/java/amqp-1-0-client-jms/pom.xml Fri Feb 7 15:24:50 2014
@@ -16,34 +16,35 @@
limitations under the License.
-->
<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/maven-v4_0_0.xsd">
+ <modelVersion>4.0.0</modelVersion>
+
<parent>
<groupId>org.apache.qpid</groupId>
- <artifactId>qpid-project</artifactId>
- <version>0.26-SNAPSHOT</version>
+ <artifactId>qpid-parent</artifactId>
+ <version>1.0-SNAPSHOT</version>
</parent>
- <modelVersion>4.0.0</modelVersion>
<artifactId>qpid-amqp-1-0-client-jms</artifactId>
+ <version>0.28-SNAPSHOT</version>
+ <name>Qpid AMQP 1.0 Client JMS</name>
+ <description>AMQP 1.0 compliant JMS module</description>
<dependencies>
<dependency>
<groupId>org.apache.qpid</groupId>
<artifactId>qpid-amqp-1-0-client</artifactId>
- <version>0.26-SNAPSHOT</version>
- <scope>compile</scope>
+ <version>${project.version}</version>
</dependency>
<dependency>
<groupId>org.apache.qpid</groupId>
<artifactId>qpid-amqp-1-0-common</artifactId>
- <version>0.26-SNAPSHOT</version>
- <scope>compile</scope>
+ <version>${project.version}</version>
</dependency>
<dependency>
<groupId>org.apache.geronimo.specs</groupId>
<artifactId>geronimo-jms_1.1_spec</artifactId>
- <version>1.0</version>
<scope>provided</scope>
</dependency>
</dependencies>
Modified: qpid/branches/java-broker-amqp-1-0-management/java/amqp-1-0-client-websocket/pom.xml
URL: http://svn.apache.org/viewvc/qpid/branches/java-broker-amqp-1-0-management/java/amqp-1-0-client-websocket/pom.xml?rev=1565688&r1=1565687&r2=1565688&view=diff
==============================================================================
--- qpid/branches/java-broker-amqp-1-0-management/java/amqp-1-0-client-websocket/pom.xml (original)
+++ qpid/branches/java-broker-amqp-1-0-management/java/amqp-1-0-client-websocket/pom.xml Fri Feb 7 15:24:50 2014
@@ -16,145 +16,41 @@
limitations under the License.
-->
<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/maven-v4_0_0.xsd">
+ <modelVersion>4.0.0</modelVersion>
+
<parent>
<groupId>org.apache.qpid</groupId>
- <artifactId>qpid-project</artifactId>
- <version>0.26-SNAPSHOT</version>
+ <artifactId>qpid-parent</artifactId>
+ <version>1.0-SNAPSHOT</version>
</parent>
- <modelVersion>4.0.0</modelVersion>
<artifactId>qpid-amqp-1-0-client-websocket</artifactId>
+ <version>0.28-SNAPSHOT</version>
+ <name>Qpid AMQP 1.0 Client WebSocket</name>
+ <description>AMQP 1.0 compliant WebSocket module</description>
<dependencies>
-
<dependency>
<groupId>org.apache.qpid</groupId>
<artifactId>qpid-amqp-1-0-client</artifactId>
- <version>0.26-SNAPSHOT</version>
- <scope>compile</scope>
+ <version>${project.version}</version>
</dependency>
<dependency>
<groupId>org.apache.qpid</groupId>
<artifactId>qpid-amqp-1-0-common</artifactId>
- <version>0.26-SNAPSHOT</version>
- <scope>compile</scope>
+ <version>${project.version}</version>
</dependency>
+ <dependency>
+ <groupId>org.apache.geronimo.specs</groupId>
+ <artifactId>geronimo-servlet_3.0_spec</artifactId>
+ </dependency>
- <dependency>
- <groupId>org.apache.geronimo.specs</groupId>
- <artifactId>geronimo-servlet_3.0_spec</artifactId>
- <version>1.0</version>
- <scope>compile</scope>
- </dependency>
-
- <dependency>
- <groupId>org.eclipse.jetty</groupId>
- <artifactId>jetty-server</artifactId>
- <version>8.1.14.v20131031</version>
- <scope>compile</scope>
- <exclusions>
- <exclusion>
- <groupId>org.eclipse.jetty.orbit</groupId>
- <artifactId>javax.servlet</artifactId>
- </exclusion>
- <exclusion>
- <groupId>org.eclipse.jetty</groupId>
- <artifactId>jetty-continuation</artifactId>
- </exclusion>
- <exclusion>
- <groupId>org.eclipse.jetty</groupId>
- <artifactId>jetty-http</artifactId>
- </exclusion>
- </exclusions>
- </dependency>
-
- <dependency>
- <groupId>org.eclipse.jetty</groupId>
- <artifactId>jetty-continuation</artifactId>
- <version>8.1.14.v20131031</version>
- <scope>compile</scope>
- </dependency>
-
- <dependency>
- <groupId>org.eclipse.jetty</groupId>
- <artifactId>jetty-security</artifactId>
- <version>8.1.14.v20131031</version>
- <scope>compile</scope>
- <exclusions>
- <exclusion>
- <groupId>org.eclipse.jetty</groupId>
- <artifactId>jetty-server</artifactId>
- </exclusion>
- </exclusions>
- </dependency>
-
- <dependency>
- <groupId>org.eclipse.jetty</groupId>
- <artifactId>jetty-http</artifactId>
- <version>8.1.14.v20131031</version>
- <scope>compile</scope>
- <exclusions>
- <exclusion>
- <groupId>org.eclipse.jetty</groupId>
- <artifactId>jetty-io</artifactId>
- </exclusion>
- </exclusions>
- </dependency>
- <dependency>
- <groupId>org.eclipse.jetty</groupId>
- <artifactId>jetty-io</artifactId>
- <version>8.1.14.v20131031</version>
- <scope>compile</scope>
- <exclusions>
- <exclusion>
- <groupId>org.eclipse.jetty</groupId>
- <artifactId>jetty-util</artifactId>
- </exclusion>
- </exclusions>
- </dependency>
-
- <dependency>
- <groupId>org.eclipse.jetty</groupId>
- <artifactId>jetty-servlet</artifactId>
- <version>8.1.14.v20131031</version>
- <scope>compile</scope>
- <exclusions>
- <exclusion>
- <groupId>org.eclipse.jetty</groupId>
- <artifactId>jetty-security</artifactId>
- </exclusion>
- </exclusions>
- </dependency>
-
- <dependency>
- <groupId>org.eclipse.jetty</groupId>
- <artifactId>jetty-util</artifactId>
- <version>8.1.14.v20131031</version>
- <scope>compile</scope>
- </dependency>
-
- <dependency>
- <groupId>org.eclipse.jetty</groupId>
- <artifactId>jetty-websocket</artifactId>
- <version>8.1.14.v20131031</version>
- <scope>compile</scope>
- <exclusions>
- <exclusion>
- <groupId>org.eclipse.jetty</groupId>
- <artifactId>jetty-util</artifactId>
- </exclusion>
- <exclusion>
- <groupId>org.eclipse.jetty</groupId>
- <artifactId>jetty-io</artifactId>
- </exclusion>
- <exclusion>
- <groupId>org.eclipse.jetty</groupId>
- <artifactId>jetty-http</artifactId>
- </exclusion>
- </exclusions>
- </dependency>
+ <dependency>
+ <groupId>org.eclipse.jetty</groupId>
+ <artifactId>jetty-websocket</artifactId>
+ </dependency>
</dependencies>
<build>
Modified: qpid/branches/java-broker-amqp-1-0-management/java/amqp-1-0-client/pom.xml
URL: http://svn.apache.org/viewvc/qpid/branches/java-broker-amqp-1-0-management/java/amqp-1-0-client/pom.xml?rev=1565688&r1=1565687&r2=1565688&view=diff
==============================================================================
--- qpid/branches/java-broker-amqp-1-0-management/java/amqp-1-0-client/pom.xml (original)
+++ qpid/branches/java-broker-amqp-1-0-management/java/amqp-1-0-client/pom.xml Fri Feb 7 15:24:50 2014
@@ -16,21 +16,24 @@
limitations under the License.
-->
<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/maven-v4_0_0.xsd">
+ <modelVersion>4.0.0</modelVersion>
+
<parent>
<groupId>org.apache.qpid</groupId>
- <artifactId>qpid-project</artifactId>
- <version>0.26-SNAPSHOT</version>
+ <artifactId>qpid-parent</artifactId>
+ <version>1.0-SNAPSHOT</version>
</parent>
- <modelVersion>4.0.0</modelVersion>
<artifactId>qpid-amqp-1-0-client</artifactId>
+ <version>0.28-SNAPSHOT</version>
+ <name>Qpid AMQP 1.0 Client</name>
+ <description>AMQP 1.0 compliant client module</description>
<dependencies>
<dependency>
<groupId>org.apache.qpid</groupId>
<artifactId>qpid-amqp-1-0-common</artifactId>
- <version>0.26-SNAPSHOT</version>
- <scope>compile</scope>
+ <version>${project.version}</version>
</dependency>
</dependencies>
Propchange: qpid/branches/java-broker-amqp-1-0-management/java/amqp-1-0-common/
------------------------------------------------------------------------------
Merged /qpid/trunk/qpid/java/amqp-1-0-common:r1562456-1565686
Modified: qpid/branches/java-broker-amqp-1-0-management/java/amqp-1-0-common/pom.xml
URL: http://svn.apache.org/viewvc/qpid/branches/java-broker-amqp-1-0-management/java/amqp-1-0-common/pom.xml?rev=1565688&r1=1565687&r2=1565688&view=diff
==============================================================================
--- qpid/branches/java-broker-amqp-1-0-management/java/amqp-1-0-common/pom.xml (original)
+++ qpid/branches/java-broker-amqp-1-0-management/java/amqp-1-0-common/pom.xml Fri Feb 7 15:24:50 2014
@@ -16,14 +16,18 @@
limitations under the License.
-->
<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/maven-v4_0_0.xsd">
+ <modelVersion>4.0.0</modelVersion>
+
<parent>
<groupId>org.apache.qpid</groupId>
- <artifactId>qpid-project</artifactId>
- <version>0.26-SNAPSHOT</version>
+ <artifactId>qpid-parent</artifactId>
+ <version>1.0-SNAPSHOT</version>
</parent>
- <modelVersion>4.0.0</modelVersion>
<artifactId>qpid-amqp-1-0-common</artifactId>
+ <version>0.28-SNAPSHOT</version>
+ <name>Qpid AMQP 1.0 Common</name>
+ <description>AMQP 1.0 compliant common module</description>
<build>
</build>
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@qpid.apache.org
For additional commands, e-mail: commits-help@qpid.apache.org