QPID-5388: Segmentation fault when recovering empty queue. The recovery could not handle uninitialized journal files, which exist prior to the first write of a record to the journal. The recovery now correctly utilizes the uninitialized file.

git-svn-id: https://svn.apache.org/repos/asf/qpid/trunk@1547921 13f79535-47bb-0310-9956-ffa450edef68

7 files changed, 53 insertions, 19 deletions

diff --git a/qpid/cpp/src/qpid/linearstore/ISSUES b/qpid/cpp/src/qpid/linearstore/ISSUES
index c511517ce8..3e1216f8a7 100644
--- a/qpid/cpp/src/qpid/linearstore/ISSUES
+++ b/qpid/cpp/src/qpid/linearstore/ISSUES
@@ -3,13 +3,13 @@ LinearStore issues:
 Store:
 ------
 
-1. (SOLVED) Overwrite identity: When recovering a previously used file, if the write boundary coincides with old record
+1. (FIXED) Overwrite identity: When recovering a previously used file, if the write boundary coincides with old record
    start, no way of discriminating old from new at boundary (used to use OWI).
 
-2. (SOLVED) QPID-5357: Recycling files while in use not working, however, files are recovered to EFP during recovery. Must solve
+2. (FIXED) QPID-5357: Recycling files while in use not working, however, files are recovered to EFP during recovery. Must solve
    #1 first.
 
-3. (SOLVED) QPID-5358: Checksum not implemented in record tail, not checked during read.
+3. (FIXED) QPID-5358: Checksum not implemented in record tail, not checked during read.
 
 4. QPID-5359: Rework qpid management parameters and controls (QMF).
 
@@ -31,7 +31,7 @@ Current bugs and performance issues:
 ------------------------------------
 1. RH Bugzilla 1035843 - Slow performance for producers
 2. (FIXED) QPID-5387 (BZ 1036071) - Crash when deleting queue
-3. RH Bugzilla 1035802 - Broker won't recover with durable queue
+3. (FIXED) QPID-5388 (BZ 1035802) - Segmentation fault when recovering empty queue
 4. RH Bugzilla 1036026 - Unable to create durable queue - framing error
 
 Code tidy-up
diff --git a/qpid/cpp/src/qpid/linearstore/JournalLogImpl.cpp b/qpid/cpp/src/qpid/linearstore/JournalLogImpl.cpp
index 36e7c7e410..c3e631a6ca 100644
--- a/qpid/cpp/src/qpid/linearstore/JournalLogImpl.cpp
+++ b/qpid/cpp/src/qpid/linearstore/JournalLogImpl.cpp
@@ -48,13 +48,13 @@ JournalLogImpl::log(const qpid::linearstore::journal::JournalLog::log_level_t le
                     const std::string& jid,
                     const std::string& log_stmt) const {
     switch (level) {
-      case LOG_CRITICAL: QPID_LOG(critical, "Linear Store: Journal \'" << jid << "\":" << log_stmt); break;
-      case LOG_ERROR: QPID_LOG(error, "Linear Store: Journal \'" << jid << "\":" << log_stmt); break;
-      case LOG_WARN: QPID_LOG(warning, "Linear Store: Journal \'" << jid << "\":" << log_stmt); break;
-      case LOG_NOTICE: QPID_LOG(notice, "Linear Store: Journal \'" << jid << "\":" << log_stmt); break;
-      case LOG_INFO: QPID_LOG(info, "Linear Store: Journal \'" << jid << "\":" << log_stmt); break;
-      case LOG_DEBUG: QPID_LOG(debug, "Linear Store: Journal \'" << jid << "\":" << log_stmt); break;
-      default: QPID_LOG(trace, "Linear Store: Journal \'" << jid << "\":" << log_stmt);
+      case LOG_CRITICAL: QPID_LOG(critical, "Linear Store: Journal \"" << jid << "\": " << log_stmt); break;
+      case LOG_ERROR: QPID_LOG(error, "Linear Store: Journal \"" << jid << "\": " << log_stmt); break;
+      case LOG_WARN: QPID_LOG(warning, "Linear Store: Journal \"" << jid << "\": " << log_stmt); break;
+      case LOG_NOTICE: QPID_LOG(notice, "Linear Store: Journal \"" << jid << "\": " << log_stmt); break;
+      case LOG_INFO: QPID_LOG(info, "Linear Store: Journal \"" << jid << "\": " << log_stmt); break;
+      case LOG_DEBUG: QPID_LOG(debug, "Linear Store: Journal \"" << jid << "\": " << log_stmt); break;
+      default: QPID_LOG(trace, "Linear Store: Journal \"" << jid << "\": " << log_stmt);
     }
 }
 
diff --git a/qpid/cpp/src/qpid/linearstore/journal/LinearFileController.cpp b/qpid/cpp/src/qpid/linearstore/journal/LinearFileController.cpp
index aa0f6da0de..f962976daf 100644
--- a/qpid/cpp/src/qpid/linearstore/journal/LinearFileController.cpp
+++ b/qpid/cpp/src/qpid/linearstore/journal/LinearFileController.cpp
@@ -91,6 +91,10 @@ void LinearFileController::pullEmptyFileFromEfp() {
     addJournalFile(ef, emptyFilePoolPtr_->getIdentity(), getNextFileSeqNum(), 0);
 }
 
+void LinearFileController::restoreEmptyFile(const std::string& fileName) {
+    addJournalFile(fileName, emptyFilePoolPtr_->getIdentity(), getNextFileSeqNum(), 0);
+}
+
 void LinearFileController::purgeEmptyFilesToEfp() {
     slock l(journalFileListMutex_);
     purgeEmptyFilesToEfpNoLock();
diff --git a/qpid/cpp/src/qpid/linearstore/journal/LinearFileController.h b/qpid/cpp/src/qpid/linearstore/journal/LinearFileController.h
index fa8327007a..933b9792a4 100644
--- a/qpid/cpp/src/qpid/linearstore/journal/LinearFileController.h
+++ b/qpid/cpp/src/qpid/linearstore/journal/LinearFileController.h
@@ -68,6 +68,7 @@ public:
     efpFileSize_sblks_t fileSize_sblks() const;
     uint64_t getNextRecordId();
     void pullEmptyFileFromEfp();
+    void restoreEmptyFile(const std::string& fileName);
     void purgeEmptyFilesToEfp();
 
     // Functions for manipulating counts of non-current JournalFile instances in journalFileList_
diff --git a/qpid/cpp/src/qpid/linearstore/journal/RecoveryManager.cpp b/qpid/cpp/src/qpid/linearstore/journal/RecoveryManager.cpp
index a8d24366de..66ac7c3a2d 100644
--- a/qpid/cpp/src/qpid/linearstore/journal/RecoveryManager.cpp
+++ b/qpid/cpp/src/qpid/linearstore/journal/RecoveryManager.cpp
@@ -242,11 +242,17 @@ bool RecoveryManager::readNextRemainingRecord(void** const dataPtrPtr,
 
 void RecoveryManager::setLinearFileControllerJournals(lfcAddJournalFileFn fnPtr,
                                                       LinearFileController* lfcPtr) {
-    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);
+    if (journalEmptyFlag_) {
+        if (uninitializedJournal_.size() > 0) {
+            lfcPtr->restoreEmptyFile(uninitializedJournal_);
+        }
+    } 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);
+        }
     }
 }
 
@@ -332,7 +338,17 @@ void RecoveryManager::analyzeJournalFileHeaders(efpIdentity_t& efpIdentity) {
     jdir::read_dir(journalDirectory_, directoryList, false, true, false, true);
     for (directoryListConstItr_t i = directoryList.begin(); i != directoryList.end(); ++i) {
         readJournalFileHeader(*i, fileHeader, headerQueueName);
-        if (headerQueueName.compare(queueName_) != 0) {
+        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());
+            }
+        } 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());
@@ -344,9 +360,18 @@ void RecoveryManager::analyzeJournalFileHeaders(efpIdentity_t& efpIdentity) {
             }
         }
     }
+
+    // TODO: Logic weak here for detecting error conditions in journal, specifically when no
+    // valid files exist, or files from mixed EFPs. Currently last read file header determines
+    // efpIdentity.
     efpIdentity.pn_ = fileHeader._efp_partition;
     efpIdentity.ds_ = fileHeader._data_size_kib;
-    currentJournalFileConstItr_ = fileNumberMap_.begin();
+
+    if (fileNumberMap_.empty()) {
+        journalEmptyFlag_ = true;
+    } else {
+        currentJournalFileConstItr_ = fileNumberMap_.begin();
+    }
 }
 
 void RecoveryManager::checkFileStreamOk(bool checkEof) {
@@ -471,6 +496,9 @@ bool RecoveryManager::getFile(const uint64_t fileNumber, bool jumpToFirstRecordO
 }
 
 bool RecoveryManager::getNextFile(bool jumpToFirstRecordOffsetFlag) {
+    if (fileNumberMap_.empty()) {
+        return false;
+    }
     if (inFileStream_.is_open()) {
         inFileStream_.close();
 //std::cout << " .f=" << getCurrentFileName() << "]" << std::flush; // DEBUG
diff --git a/qpid/cpp/src/qpid/linearstore/journal/RecoveryManager.h b/qpid/cpp/src/qpid/linearstore/journal/RecoveryManager.h
index 153b0f0511..a32fdc1853 100644
--- a/qpid/cpp/src/qpid/linearstore/journal/RecoveryManager.h
+++ b/qpid/cpp/src/qpid/linearstore/journal/RecoveryManager.h
@@ -71,6 +71,7 @@ protected:
     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
 
     // State for recovery of individual enqueued records
     uint64_t currentSerial_;
diff --git a/qpid/cpp/src/qpid/linearstore/journal/utils/file_hdr.c b/qpid/cpp/src/qpid/linearstore/journal/utils/file_hdr.c
index 35b4ea219e..07f4719a8e 100644
--- a/qpid/cpp/src/qpid/linearstore/journal/utils/file_hdr.c
+++ b/qpid/cpp/src/qpid/linearstore/journal/utils/file_hdr.c
@@ -57,7 +57,7 @@ int file_hdr_init(void* dest, const uint64_t dest_len, const uint16_t uflag, con
 
 int file_hdr_check(file_hdr_t* hdr, const uint32_t magic, const uint16_t version, const uint64_t data_size_kib) {
     int res = rec_hdr_check_base(&hdr->_rhdr, magic, version);
-    if (res != 0) return 0;
+    if (res != 0) return res;
     if (hdr->_data_size_kib != data_size_kib) return 3;
     return 0;
 }