| /* |
| * |
| * 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/linearstore/journal/RecoveryManager.h" |
| |
| #include <algorithm> |
| #include <cstdlib> |
| #include <iomanip> |
| #include "qpid/linearstore/journal/Checksum.h" |
| #include "qpid/linearstore/journal/data_tok.h" |
| #include "qpid/linearstore/journal/deq_rec.h" |
| #include "qpid/linearstore/journal/EmptyFilePool.h" |
| #include "qpid/linearstore/journal/EmptyFilePoolManager.h" |
| #include "qpid/linearstore/journal/enq_map.h" |
| #include "qpid/linearstore/journal/enq_rec.h" |
| #include "qpid/linearstore/journal/jcfg.h" |
| #include "qpid/linearstore/journal/jdir.h" |
| #include "qpid/linearstore/journal/JournalFile.h" |
| #include "qpid/linearstore/journal/JournalLog.h" |
| #include "qpid/linearstore/journal/jrec.h" |
| #include "qpid/linearstore/journal/LinearFileController.h" |
| #include "qpid/linearstore/journal/txn_map.h" |
| #include "qpid/linearstore/journal/txn_rec.h" |
| #include "qpid/linearstore/journal/utils/enq_hdr.h" |
| #include "qpid/linearstore/journal/utils/file_hdr.h" |
| #include <sstream> |
| #include <string> |
| #include <unistd.h> |
| #include <vector> |
| |
| namespace qpid { |
| namespace linearstore { |
| namespace journal { |
| |
| RecoveredRecordData_t::RecoveredRecordData_t(const uint64_t rid, const uint64_t fid, const std::streampos foffs, bool ptxn) : |
| recordId_(rid), |
| fileId_(fid), |
| fileOffset_(foffs), |
| 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, |
| txn_map& transactionMapRef, |
| JournalLog& journalLogRef) : |
| journalDirectory_(journalDirectory), |
| queueName_(queuename), |
| enqueueMapRef_(enqueueMapRef), |
| transactionMapRef_(transactionMapRef), |
| journalLogRef_(journalLogRef), |
| journalEmptyFlag_(false), |
| firstRecordOffset_(0), |
| endOffset_(0), |
| highestRecordId_(0ULL), |
| highestFileNumber_(0ULL), |
| lastFileFullFlag_(false), |
| initial_fid_(0), |
| currentSerial_(0), |
| efpFileSize_kib_(0) |
| {} |
| |
| 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, |
| EmptyFilePool** emptyFilePoolPtrPtr) { |
| // Analyze file headers of existing journal files |
| efpIdentity_t efpIdentity; |
| analyzeJournalFileHeaders(efpIdentity); |
| |
| if (journalEmptyFlag_) { |
| if (uninitFileList_.empty()) { |
| *emptyFilePoolPtrPtr = emptyFilePoolManager->getEmptyFilePool(0, 0); // Use default EFP |
| } else { |
| *emptyFilePoolPtrPtr = emptyFilePoolManager->getEmptyFilePool(efpIdentity); |
| } |
| } else { |
| *emptyFilePoolPtrPtr = emptyFilePoolManager->getEmptyFilePool(efpIdentity); |
| if (! *emptyFilePoolPtrPtr) { |
| // TODO: At a later time, this could be used to establish a new pool size provided the partition exists. |
| // If the partition does not exist, this is always an error. For now, throw an exception, as this should |
| // not occur in any practical application. Once multiple partitions and mixed EFPs are supported, this |
| // needs to be resolved. Note that EFP size is always a multiple of QLS_SBLK_SIZE_BYTES (currently 4096 |
| // bytes, any other value cannot be used and should be rejected as an error. |
| std::ostringstream oss; |
| oss << "Invalid EFP identity: Partition=" << efpIdentity.pn_ << " Size=" << efpIdentity.ds_ << "k"; |
| throw jexception(jerrno::JERR_RCVM_INVALIDEFPID, oss.str(), "RecoveryManager", "analyzeJournals"); |
| } |
| efpFileSize_kib_ = (*emptyFilePoolPtrPtr)->fileSize_kib(); |
| |
| // Read all records, establish remaining enqueued records |
| if (inFileStream_.is_open()) { |
| inFileStream_.close(); |
| } |
| while (getNextRecordHeader()) {} |
| if (inFileStream_.is_open()) { |
| 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); |
| |
| // Remove all txns from tmap that are not in the prepared list |
| if (preparedTransactionListPtr) { |
| std::vector<std::string> xidList; |
| transactionMapRef_.xid_list(xidList); |
| for (std::vector<std::string>::iterator itr = xidList.begin(); itr != xidList.end(); itr++) { |
| std::vector<std::string>::const_iterator pitr = |
| std::find(preparedTransactionListPtr->begin(), preparedTransactionListPtr->end(), *itr); |
| if (pitr == preparedTransactionListPtr->end()) { // not found in prepared list |
| txn_data_list_t tdl = transactionMapRef_.get_remove_tdata_list(*itr); // tdl will be empty if xid not found |
| // 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->fid_]->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 |
| std::ostringstream oss; |
| oss << std::hex << "_emap.unlock(): drid=0x\"" << i->drid_; |
| throw jexception(jerrno::JERR_MAP_NOTFOUND, oss.str(), "RecoveryManager", "analyzeJournals"); |
| } |
| } |
| } |
| } |
| } |
| } |
| prepareRecordList(); |
| } |
| } |
| |
| std::streamoff RecoveryManager::getEndOffset() const { |
| return endOffset_; |
| } |
| |
| uint64_t RecoveryManager::getHighestFileNumber() const { |
| return highestFileNumber_; |
| } |
| |
| uint64_t RecoveryManager::getHighestRecordId() const { |
| return highestRecordId_; |
| } |
| |
| bool RecoveryManager::isLastFileFull() const { |
| return lastFileFullFlag_; |
| } |
| |
| bool RecoveryManager::readNextRemainingRecord(void** const dataPtrPtr, |
| std::size_t& dataSize, |
| void** const xidPtrPtr, |
| std::size_t& xidSize, |
| bool& transient, |
| bool& external, |
| data_tok* const dtokp, |
| bool ignore_pending_txns) { |
| bool foundRecord = false; |
| do { |
| if (recordIdListConstItr_ == recordIdList_.end()) { |
| return false; |
| } |
| if (recordIdListConstItr_->pendingTransaction_ && ignore_pending_txns) { // Pending transaction |
| ++recordIdListConstItr_; // ignore, go to next record |
| } else { |
| foundRecord = true; |
| } |
| } while (!foundRecord); |
| |
| 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_; |
| throw jexception(jerrno::JERR__FILEIO, oss.str(), "RecoveryManager", "readNextRemainingRecord"); |
| } |
| } |
| inFileStream_.seekg(recordIdListConstItr_->fileOffset_, std::ifstream::beg); |
| if (!inFileStream_.good()) { |
| std::ostringstream oss; |
| oss << "Could not find offset 0x" << std::hex << recordIdListConstItr_->fileOffset_ << " in file " << getCurrentFileName(); |
| throw jexception(jerrno::JERR__FILEIO, oss.str(), "RecoveryManager", "readNextRemainingRecord"); |
| } |
| |
| ::enq_hdr_t enqueueHeader; |
| inFileStream_.read((char*)&enqueueHeader, sizeof(::enq_hdr_t)); |
| if (inFileStream_.gcount() != sizeof(::enq_hdr_t)) { |
| std::ostringstream oss; |
| oss << "Could not read enqueue header from file " << getCurrentFileName() << " at offset 0x" << std::hex << recordIdListConstItr_->fileOffset_; |
| throw jexception(jerrno::JERR__FILEIO, oss.str(), "RecoveryManager", "readNextRemainingRecord"); |
| } |
| // check flags |
| transient = ::is_enq_transient(&enqueueHeader); |
| external = ::is_enq_external(&enqueueHeader); |
| |
| // read xid |
| xidSize = enqueueHeader._xidsize; |
| *xidPtrPtr = ::malloc(xidSize); |
| if (*xidPtrPtr == 0) { |
| std::ostringstream oss; |
| oss << "xidPtr, size=0x" << std::hex << xidSize; |
| throw jexception(jerrno::JERR__MALLOC, oss.str(), "RecoveryManager", "readNextRemainingRecord"); |
| } |
| readJournalData((char*)*xidPtrPtr, xidSize); |
| |
| // read data |
| dataSize = enqueueHeader._dsize; |
| *dataPtrPtr = ::malloc(dataSize); |
| if (*xidPtrPtr == 0) { |
| std::ostringstream oss; |
| oss << "dataPtr, size=0x" << std::hex << dataSize; |
| throw jexception(jerrno::JERR__MALLOC, oss.str(), "RecoveryManager", "readNextRemainingRecord"); |
| } |
| readJournalData((char*)*dataPtrPtr, dataSize); |
| |
| // Check enqueue record checksum |
| Checksum checksum; |
| checksum.addData((const unsigned char*)&enqueueHeader, sizeof(::enq_hdr_t)); |
| if (xidSize > 0) { |
| checksum.addData((const unsigned char*)*xidPtrPtr, xidSize); |
| } |
| if (dataSize > 0) { |
| checksum.addData((const unsigned char*)*dataPtrPtr, dataSize); |
| } |
| ::rec_tail_t enqueueTail; |
| readJournalData((char*)&enqueueTail, sizeof(::rec_tail_t)); |
| uint32_t cs = checksum.getChecksum(); |
| uint16_t res = ::rec_tail_check(&enqueueTail, &enqueueHeader._rhdr, cs); |
| if (res != 0) { |
| std::stringstream oss; |
| oss << "Bad record tail:" << std::hex; |
| if (res & ::REC_TAIL_MAGIC_ERR_MASK) { |
| oss << std::endl << " Magic: expected 0x" << ~enqueueHeader._rhdr._magic << "; found 0x" << enqueueTail._xmagic; |
| } |
| if (res & ::REC_TAIL_SERIAL_ERR_MASK) { |
| oss << std::endl << " Serial: expected 0x" << enqueueHeader._rhdr._serial << "; found 0x" << enqueueTail._serial; |
| } |
| if (res & ::REC_TAIL_RID_ERR_MASK) { |
| oss << std::endl << " Record Id: expected 0x" << enqueueHeader._rhdr._rid << "; found 0x" << enqueueTail._rid; |
| } |
| if (res & ::REC_TAIL_CHECKSUM_ERR_MASK) { |
| oss << std::endl << " Checksum: expected 0x" << cs << "; found 0x" << enqueueTail._checksum; |
| } |
| throw jexception(jerrno::JERR_JREC_BADRECTAIL, oss.str(), "RecoveryManager", "readNextRemainingRecord"); // TODO: Don't throw exception, log info |
| } |
| |
| // Set data token |
| dtokp->set_wstate(data_tok::ENQ); |
| dtokp->set_rid(enqueueHeader._rhdr._rid); |
| dtokp->set_dsize(dataSize); |
| if (xidSize) { |
| dtokp->set_xid(*xidPtrPtr, xidSize); |
| } |
| |
| ++recordIdListConstItr_; |
| return true; |
| } |
| |
| void RecoveryManager::recoveryComplete() { |
| if(inFileStream_.is_open()) { |
| inFileStream_.close(); |
| } |
| } |
| |
| void RecoveryManager::setLinearFileControllerJournals(lfcAddJournalFileFn fnPtr, |
| LinearFileController* lfcPtr) { |
| if (journalEmptyFlag_) { |
| restoreEmptyFile(lfcPtr); |
| } else { |
| if (initial_fid_ == 0) { |
| throw jexception(jerrno::JERR_RCVM_NULLFID, "RecoveryManager", "setLinearFileControllerJournals"); |
| } |
| for (fileNumberMapConstItr_t i = fileNumberMap_.begin(); i != fileNumberMap_.end(); ++i) { |
| (lfcPtr->*fnPtr)(i->second->journalFilePtr_, i->second->completedDblkCount_, i->first == initial_fid_); |
| } |
| if (lastFileFullFlag_) { |
| restoreEmptyFile(lfcPtr); |
| } |
| } |
| |
| 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()); |
| } |
| } |
| |
| void RecoveryManager::restoreEmptyFile(LinearFileController* lfcPtr) { |
| if (uninitFileList_.size() > 0) { |
| std::string uninitFile = uninitFileList_.back(); |
| uninitFileList_.pop_back(); |
| lfcPtr->restoreEmptyFile(uninitFile); |
| lastFileFullFlag_ = false; |
| endOffset_ = 0; |
| } |
| } |
| |
| std::string RecoveryManager::toString(const std::string& jid, const uint16_t indent) const { |
| std::string indentStr(indent, ' '); |
| std::ostringstream oss; |
| oss << std::endl << indentStr << "Journal recovery analysis (jid=\"" << jid << "\"):" << std::endl; |
| if (journalEmptyFlag_) { |
| oss << indentStr << "<Journal empty, no journal files found>" << std::endl; |
| } else { |
| oss << indentStr << std::setw(7) << "file_id" |
| << std::setw(43) << "file_name" |
| << std::setw(12) << "record_cnt" |
| << std::setw(16) << "fro" |
| << std::setw(12) << "efp_id" |
| << std::endl; |
| oss << indentStr << std::setw(7) << "-------" |
| << std::setw(43) << "-----------------------------------------" |
| << std::setw(12) << "----------" |
| << std::setw(16) << "--------------" |
| << std::setw(12) << "----------" |
| << std::endl; |
| uint32_t totalRecordCount(0UL); |
| for (fileNumberMapConstItr_t k=fileNumberMap_.begin(); k!=fileNumberMap_.end(); ++k) { |
| 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->journalFilePtr_->getFirstRecordOffset(); |
| oss << indentStr << std::setw(7) << fid.str() |
| << std::setw(43) << fqFileName.substr(fqFileName.rfind('/')+1) |
| << std::setw(12) << k->second->journalFilePtr_->getEnqueuedRecordCount() |
| << std::setw(16) << fro.str() |
| << std::setw(12) << k->second->journalFilePtr_->getEfpIdentity() |
| << std::endl; |
| totalRecordCount += k->second->journalFilePtr_->getEnqueuedRecordCount(); |
| } |
| oss << indentStr << std::setw(62) << "----------" << std::endl; |
| oss << indentStr << std::setw(62) << totalRecordCount << std::endl; |
| oss << indentStr << "First record offset in first file = 0x" << std::hex << firstRecordOffset_ << |
| std::dec << " (" << (firstRecordOffset_/QLS_DBLK_SIZE_BYTES) << " dblks)" << std::endl; |
| oss << indentStr << "End offset in last file = 0x" << std::hex << endOffset_ << std::dec << " (" << |
| (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; |
| } |
| return oss.str(); |
| } |
| |
| // --- protected functions --- |
| |
| void RecoveryManager::analyzeJournalFileHeaders(efpIdentity_t& efpIdentity) { |
| std::string headerQueueName; |
| ::file_hdr_t fileHeader; |
| stringList_t directoryList; |
| jdir::read_dir(journalDirectory_, directoryList, false, true, false, true); |
| for (stringListConstItr_t i = directoryList.begin(); i != directoryList.end(); ++i) { |
| bool hdrOk = readJournalFileHeader(*i, fileHeader, headerQueueName); |
| bool hdrEmpty = ::is_file_hdr_reset(&fileHeader); |
| if (!hdrOk) { |
| std::ostringstream oss; |
| oss << "Journal file " << (*i) << " is corrupted or invalid"; |
| journalLogRef_.log(JournalLog::LOG_WARN, queueName_, oss.str()); |
| } else if (hdrEmpty) { |
| /* This seems unnecessary, as the data size is encoded in the file_hdr_t::_data_size_kib, even for reset file headers |
| // Read symlink, find efp directory name which is efp size in KiB |
| // TODO: place this bit into a common function as it is also used in EmptyFilePool.cpp::deleteSymlink() |
| char buff[1024]; |
| ssize_t len = ::readlink((*i).c_str(), buff, 1023); |
| if (len < 0) { |
| std::ostringstream oss; |
| oss << "symlink=\"" << (*i) << "\"" << FORMAT_SYSERR(errno); |
| throw jexception(jerrno::JERR__SYMLINK, oss.str(), "RecoveryManager", "analyzeJournalFileHeaders"); |
| } |
| buff[len] = '\0'; // ::readlink() does not terminate the string |
| // Find second and third '/' from back of string, which contains the EFP directory name |
| *(::strrchr(buff, '/')) = '\0'; |
| *(::strrchr(buff, '/')) = '\0'; |
| int efpDataSize_kib = atoi(::strrchr(buff, '/') + 1); |
| uninitFileList_.push_back(*i); |
| efpIdentity.pn_ = fileHeader._efp_partition; |
| efpIdentity.ds_ = efpDataSize_kib; |
| */ |
| efpIdentity.pn_ = fileHeader._efp_partition; |
| efpIdentity.ds_ = fileHeader._data_size_kib; |
| 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, 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; |
| } |
| // 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; |
| } |
| } |
| |
| //std::cerr << "*** RecoveryManager::analyzeJournalFileHeaders() fileNumberMap_.size()=" << fileNumberMap_.size() << std::endl; // DEBUG |
| if (fileNumberMap_.empty()) { |
| journalEmptyFlag_ = true; |
| } else { |
| currentJournalFileItr_ = fileNumberMap_.begin(); |
| } |
| } |
| |
| void RecoveryManager::checkFileStreamOk(bool checkEof) { |
| if (inFileStream_.fail() || inFileStream_.bad() || checkEof ? inFileStream_.eof() : false) { |
| std::ostringstream oss; |
| oss << "Stream status: fail=" << (inFileStream_.fail()?"T":"F") << " bad=" << (inFileStream_.bad()?"T":"F"); |
| if (checkEof) { |
| oss << " eof=" << (inFileStream_.eof()?"T":"F"); |
| } |
| throw jexception(jerrno::JERR_RCVM_STREAMBAD, oss.str(), "RecoveryManager", "checkFileStreamOk"); |
| } |
| } |
| |
| 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; |
| oss << " (dblk alignment offset = 0x" << (recordPosition % QLS_DBLK_SIZE_BYTES); |
| throw jexception(jerrno::JERR_RCVM_NOTDBLKALIGNED, oss.str(), "RecoveryManager", "checkJournalAlignment"); |
| } |
| std::streampos currentPosn = recordPosition; |
| unsigned sblkOffset = currentPosn % QLS_SBLK_SIZE_BYTES; |
| if (sblkOffset) |
| { |
| std::ostringstream oss1; |
| 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()); |
| |
| 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"); |
| } |
| outFileStream.seekp(currentPosn); |
| |
| // Prepare write buffer containing a single empty record (1 dblk) |
| void* writeBuffer = std::malloc(QLS_DBLK_SIZE_BYTES); |
| if (writeBuffer == 0) { |
| throw jexception(jerrno::JERR__MALLOC, "RecoveryManager", "checkJournalAlignment"); |
| } |
| const uint32_t xmagic = QLS_EMPTY_MAGIC; |
| ::memcpy(writeBuffer, (const void*)&xmagic, sizeof(xmagic)); |
| ::memset((char*)writeBuffer + sizeof(xmagic), QLS_CLEAN_CHAR, QLS_DBLK_SIZE_BYTES - sizeof(xmagic)); |
| |
| // Write as many empty records as are needed to get to sblk boundary |
| while (currentPosn % QLS_SBLK_SIZE_BYTES) { |
| outFileStream.write((const char*)writeBuffer, QLS_DBLK_SIZE_BYTES); |
| if (outFileStream.fail()) { |
| throw jexception(jerrno::JERR_RCVM_WRITE, "RecoveryManager", "checkJournalAlignment"); |
| } |
| std::ostringstream oss2; |
| 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(); |
| } |
| outFileStream.close(); |
| std::free(writeBuffer); |
| journalLogRef_.log(JournalLog::LOG_INFO, queueName_, "Bad record alignment fixed."); |
| } |
| lastRecord(start_fid, currentPosn); |
| } |
| |
| bool RecoveryManager::decodeRecord(jrec& record, |
| std::size_t& cumulativeSizeRead, |
| ::rec_hdr_t& headerRecord, |
| const uint64_t start_fid, |
| const std::streampos recordOffset) |
| { |
| if (highestRecordId_ == 0) { |
| highestRecordId_ = headerRecord._rid; |
| } else if (headerRecord._rid - highestRecordId_ < 0x8000000000000000ULL) { // RFC 1982 comparison for unsigned 64-bit |
| highestRecordId_ = headerRecord._rid; |
| } |
| |
| bool done = false; |
| while (!done) { |
| try { |
| done = record.decode(headerRecord, &inFileStream_, cumulativeSizeRead, recordOffset); |
| } |
| catch (const jexception& e) { |
| if (e.err_code() == jerrno::JERR_JREC_BADRECTAIL) { |
| std::ostringstream oss; |
| oss << jerrno::err_msg(e.err_code()) << e.additional_info(); |
| journalLogRef_.log(JournalLog::LOG_INFO, queueName_, oss.str()); |
| } else { |
| journalLogRef_.log(JournalLog::LOG_INFO, queueName_, e.what()); |
| } |
| checkJournalAlignment(start_fid, recordOffset); |
| return false; |
| } |
| if (!done && needNextFile()) { |
| if (!getNextFile(false)) { |
| checkJournalAlignment(start_fid, recordOffset); |
| return false; |
| } |
| } |
| } |
| return true; |
| } |
| |
| std::string RecoveryManager::getCurrentFileName() const { |
| return currentJournalFileItr_->second->journalFilePtr_->getFqFileName(); |
| } |
| |
| uint64_t RecoveryManager::getCurrentFileNumber() const { |
| return currentJournalFileItr_->first; |
| } |
| |
| bool RecoveryManager::getFile(const uint64_t fileNumber, bool jumpToFirstRecordOffsetFlag) { |
| if (inFileStream_.is_open()) { |
| inFileStream_.close(); |
| //std::cout << " f=" << getCurrentFileName() << "]" << std::flush; // DEBUG |
| inFileStream_.clear(); // clear eof flag, req'd for older versions of c++ |
| } |
| currentJournalFileItr_ = fileNumberMap_.find(fileNumber); |
| if (currentJournalFileItr_ == fileNumberMap_.end()) { |
| return false; |
| } |
| inFileStream_.open(getCurrentFileName().c_str(), std::ios_base::in | std::ios_base::binary); |
| if (!inFileStream_.good()) { |
| throw jexception(jerrno::JERR__FILEIO, getCurrentFileName(), "RecoveryManager", "getFile"); |
| } |
| //std::cout << " [F=" << getCurrentFileName() << std::flush; // DEBUG |
| |
| if (!readFileHeader()) { |
| return false; |
| } |
| std::streamoff foffs = jumpToFirstRecordOffsetFlag ? firstRecordOffset_ : QLS_JRNL_FHDR_RES_SIZE_SBLKS * QLS_SBLK_SIZE_BYTES; |
| inFileStream_.seekg(foffs); |
| return true; |
| } |
| |
| bool RecoveryManager::getNextFile(bool jumpToFirstRecordOffsetFlag) { |
| if (fileNumberMap_.empty()) { |
| return false; |
| } |
| if (inFileStream_.is_open()) { |
| inFileStream_.close(); |
| //std::cout << " .f=" << getCurrentFileName() << "]" << std::flush; // DEBUG |
| 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++ |
| } |
| inFileStream_.open(getCurrentFileName().c_str(), std::ios_base::in | std::ios_base::binary); |
| if (!inFileStream_.good()) { |
| throw jexception(jerrno::JERR__FILEIO, getCurrentFileName(), "RecoveryManager", "getNextFile"); |
| } |
| //std::cout << " [.F=" << getCurrentFileName() << std::flush; // DEBUG |
| |
| if (!readFileHeader()) { |
| return false; |
| } |
| std::streamoff foffs = jumpToFirstRecordOffsetFlag ? firstRecordOffset_ : QLS_JRNL_FHDR_RES_SIZE_SBLKS * QLS_SBLK_SIZE_BYTES; |
| inFileStream_.seekg(foffs); |
| return true; |
| } |
| |
| bool RecoveryManager::getNextRecordHeader() |
| { |
| std::size_t cum_size_read = 0; |
| void* xidp = 0; |
| rec_hdr_t h; |
| |
| bool hdr_ok = false; |
| uint64_t file_id = currentJournalFileItr_->second->journalFilePtr_->getFileSeqNum(); |
| std::streampos file_pos = 0; |
| if (inFileStream_.is_open()) { |
| inFileStream_.clear(); |
| file_pos = inFileStream_.tellg(); |
| } |
| if (file_pos == std::streampos(-1)) { |
| std::ostringstream oss; |
| oss << "tellg() failure: fail=" << (inFileStream_.fail()?"T":"F") << " bad=" << (inFileStream_.bad()?"T":"F"); |
| oss << " eof=" << (inFileStream_.eof()?"T":"F") << " good=" << (inFileStream_.good()?"T":"F"); |
| oss << " rdstate=0x" << std::hex << inFileStream_.rdstate() << std::dec; |
| throw jexception(jerrno::JERR_RCVM_STREAMBAD, oss.str(), "RecoveryManager", "getNextRecordHeader"); |
| } |
| while (!hdr_ok) { |
| if (needNextFile()) { |
| if (!getNextFile(true)) { |
| lastRecord(file_id, file_pos); |
| return false; |
| } |
| } |
| file_id = currentJournalFileItr_->second->journalFilePtr_->getFileSeqNum(); |
| file_pos = inFileStream_.tellg(); |
| if (file_pos == std::streampos(-1)) { |
| std::ostringstream oss; |
| oss << "tellg() failure: fail=" << (inFileStream_.fail()?"T":"F") << " bad=" << (inFileStream_.bad()?"T":"F"); |
| oss << " eof=" << (inFileStream_.eof()?"T":"F") << " good=" << (inFileStream_.good()?"T":"F"); |
| oss << " rdstate=0x" << std::hex << inFileStream_.rdstate() << std::dec; |
| throw jexception(jerrno::JERR_RCVM_STREAMBAD, oss.str(), "RecoveryManager", "getNextRecordHeader"); |
| } |
| inFileStream_.read((char*)&h, sizeof(rec_hdr_t)); |
| if (inFileStream_.gcount() == sizeof(rec_hdr_t)) { |
| hdr_ok = true; |
| } else { |
| if (needNextFile()) { |
| if (!getNextFile(true)) { |
| lastRecord(file_id, file_pos); |
| return false; |
| } |
| } |
| } |
| } |
| |
| 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) { |
| checkJournalAlignment(file_id, file_pos); |
| return false; |
| } |
| enq_rec er; |
| if (!decodeRecord(er, cum_size_read, h, start_fid, file_pos)) { |
| return false; |
| } |
| if (!er.is_transient()) { // Ignore transient msgs |
| fileNumberMap_[start_fid]->journalFilePtr_->incrEnqueuedRecordCount(); |
| if (er.xid_size()) { |
| er.get_xid(&xidp); |
| if (xidp == 0) { |
| throw jexception(jerrno::JERR_RCVM_NULLXID, "ENQ", "RecoveryManager", "getNextRecordHeader"); |
| } |
| std::string xid((char*)xidp, er.xid_size()); |
| transactionMapRef_.insert_txn_data(xid, txn_data_t(h._rid, 0, start_fid, file_pos, true, false, false)); |
| if (transactionMapRef_.set_aio_compl(xid, h._rid) < txn_map::TMAP_OK) { // fail - xid or rid not found |
| std::ostringstream oss; |
| oss << std::hex << "_tmap.set_aio_compl: txn_enq xid=\"" << xid << "\" rid=0x" << h._rid; |
| throw jexception(jerrno::JERR_MAP_NOTFOUND, oss.str(), "RecoveryManager", "getNextRecordHeader"); |
| } |
| } else { |
| if (enqueueMapRef_.insert_pfid(h._rid, start_fid, file_pos) < enq_map::EMAP_OK) { // fail |
| // The only error code emap::insert_pfid() returns is enq_map::EMAP_DUP_RID. |
| std::ostringstream oss; |
| oss << std::hex << "rid=0x" << h._rid << " _pfid=0x" << start_fid; |
| throw jexception(jerrno::JERR_MAP_DUPLICATE, oss.str(), "RecoveryManager", "getNextRecordHeader"); |
| } |
| } |
| } |
| } |
| break; |
| case QLS_DEQ_MAGIC: |
| { |
| //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) { |
| checkJournalAlignment(file_id, file_pos); |
| return false; |
| } |
| deq_rec dr; |
| if (!decodeRecord(dr, cum_size_read, h, start_fid, file_pos)) { |
| return false; |
| } |
| if (dr.xid_size()) { |
| // If the enqueue is part of a pending txn, it will not yet be in emap |
| enqueueMapRef_.lock(dr.deq_rid()); // ignore not found error |
| dr.get_xid(&xidp); |
| if (xidp == 0) { |
| throw jexception(jerrno::JERR_RCVM_NULLXID, "DEQ", "RecoveryManager", "getNextRecordHeader"); |
| } |
| std::string xid((char*)xidp, dr.xid_size()); |
| transactionMapRef_.insert_txn_data(xid, txn_data_t(dr.rid(), dr.deq_rid(), start_fid, file_pos, |
| false, false, dr.is_txn_coml_commit())); |
| if (transactionMapRef_.set_aio_compl(xid, dr.rid()) < txn_map::TMAP_OK) { // fail - xid or rid not found |
| std::ostringstream oss; |
| oss << std::hex << "_tmap.set_aio_compl: txn_deq xid=\"" << xid << "\" rid=0x" << dr.rid(); |
| throw jexception(jerrno::JERR_MAP_NOTFOUND, oss.str(), "RecoveryManager", "getNextRecordHeader"); |
| } |
| } 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]->journalFilePtr_->decrEnqueuedRecordCount(); |
| } |
| } |
| } |
| break; |
| case QLS_TXA_MAGIC: |
| { |
| //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) { |
| checkJournalAlignment(file_id, file_pos); |
| return false; |
| } |
| txn_rec ar; |
| if (!decodeRecord(ar, cum_size_read, h, start_fid, file_pos)) { |
| return false; |
| } |
| // Delete this txn from tmap, unlock any locked records in emap |
| ar.get_xid(&xidp); |
| if (xidp == 0) { |
| throw jexception(jerrno::JERR_RCVM_NULLXID, "ABT", "RecoveryManager", "getNextRecordHeader"); |
| } |
| std::string xid((char*)xidp, ar.xid_size()); |
| 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->fid_]->journalFilePtr_->decrEnqueuedRecordCount(); |
| } else { |
| enqueueMapRef_.unlock(itr->drid_); // ignore not found error |
| } |
| } |
| } |
| break; |
| case QLS_TXC_MAGIC: |
| { |
| //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) { |
| checkJournalAlignment(file_id, file_pos); |
| return false; |
| } |
| txn_rec cr; |
| if (!decodeRecord(cr, cum_size_read, h, start_fid, file_pos)) { |
| return false; |
| } |
| // Delete this txn from tmap, process records into emap |
| cr.get_xid(&xidp); |
| if (xidp == 0) { |
| throw jexception(jerrno::JERR_RCVM_NULLXID, "CMT", "RecoveryManager", "getNextRecordHeader"); |
| } |
| std::string xid((char*)xidp, cr.xid_size()); |
| 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_) { // txn enqueue |
| //std::cout << "[rid=0x" << std::hex << itr->rid_ << std::dec << " fid=" << itr->fid_ << " fpos=0x" << std::hex << itr->foffs_ << "]" << std::dec << std::flush; // DEBUG |
| if (enqueueMapRef_.insert_pfid(itr->rid_, itr->fid_, itr->foffs_) < enq_map::EMAP_OK) { // fail |
| // The only error code emap::insert_pfid() returns is enq_map::EMAP_DUP_RID. |
| std::ostringstream oss; |
| oss << std::hex << "rid=0x" << itr->rid_ << " _pfid=0x" << itr->fid_; |
| throw jexception(jerrno::JERR_MAP_DUPLICATE, oss.str(), "RecoveryManager", "getNextRecordHeader"); |
| } |
| } 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]->journalFilePtr_->decrEnqueuedRecordCount(); |
| } |
| } |
| } |
| break; |
| case QLS_EMPTY_MAGIC: |
| { |
| //std::cout << ".x" << std::flush; // DEBUG |
| uint32_t rec_dblks = jrec::size_dblks(sizeof(::rec_hdr_t)); |
| 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; |
| } |
| } |
| } |
| break; |
| case 0: |
| //std::cout << " 0x" << std::hex << file_pos << ".0" << std::dec << std::endl << std::flush; // DEBUG |
| 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(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); |
| } |
| return true; |
| } |
| |
| void RecoveryManager::prepareRecordList() { |
| // Set up recordIdList_ from enqueue map and transaction map |
| recordIdList_.clear(); |
| |
| // Extract records from enqueue list |
| std::vector<uint64_t> ridList; |
| enqueueMapRef_.rid_list(ridList); |
| qpid::linearstore::journal::enq_map::emap_data_struct_t eds; |
| for (std::vector<uint64_t>::const_iterator i=ridList.begin(); i!=ridList.end(); ++i) { |
| enqueueMapRef_.get_data(*i, eds); |
| recordIdList_.push_back(RecoveredRecordData_t(*i, eds._pfid, eds._file_posn, false)); |
| } |
| |
| // Extract records from pending transaction enqueues |
| std::vector<std::string> xidList; |
| transactionMapRef_.xid_list(xidList); |
| for (std::vector<std::string>::const_iterator j=xidList.begin(); j!=xidList.end(); ++j) { |
| qpid::linearstore::journal::txn_data_list_t tdsl = transactionMapRef_.get_tdata_list(*j); |
| for (qpid::linearstore::journal::tdl_itr_t k=tdsl.begin(); k!=tdsl.end(); ++k) { |
| if (k->enq_flag_) { |
| recordIdList_.push_back(RecoveredRecordData_t(k->rid_, k->fid_, k->foffs_, true)); |
| } |
| } |
| } |
| |
| std::sort(recordIdList_.begin(), recordIdList_.end(), recordIdListCompare); |
| recordIdListConstItr_ = recordIdList_.begin(); |
| } |
| |
| void RecoveryManager::readJournalData(char* target, |
| const std::streamsize readSize) { |
| std::streamoff bytesRead = 0; |
| while (bytesRead < readSize) { |
| std::streampos file_pos = inFileStream_.tellg(); |
| if (file_pos == std::streampos(-1)) { |
| std::ostringstream oss; |
| oss << "tellg() failure: fail=" << (inFileStream_.fail()?"T":"F") << " bad=" << (inFileStream_.bad()?"T":"F"); |
| throw jexception(jerrno::JERR_RCVM_STREAMBAD, oss.str(), "RecoveryManager", "readJournalData"); |
| } |
| inFileStream_.read(target + bytesRead, readSize - bytesRead); |
| std::streamoff thisReadSize = inFileStream_.gcount(); |
| if (thisReadSize < readSize) { |
| if (needNextFile()) { |
| getNextFile(false); |
| } |
| file_pos = inFileStream_.tellg(); |
| if (file_pos == std::streampos(-1)) { |
| std::ostringstream oss; |
| oss << "tellg() failure: fail=" << (inFileStream_.fail()?"T":"F") << " bad=" << (inFileStream_.bad()?"T":"F"); |
| throw jexception(jerrno::JERR_RCVM_STREAMBAD, oss.str(), "RecoveryManager", "readJournalData"); |
| } |
| } |
| bytesRead += thisReadSize; |
| } |
| } |
| |
| bool RecoveryManager::readFileHeader() { |
| file_hdr_t fhdr; |
| inFileStream_.read((char*)&fhdr, sizeof(fhdr)); |
| checkFileStreamOk(true); |
| if (::file_hdr_check(&fhdr, QLS_FILE_MAGIC, QLS_JRNL_VERSION, efpFileSize_kib_, QLS_MAX_QUEUE_NAME_LEN) != 0) { |
| firstRecordOffset_ = fhdr._fro; |
| currentSerial_ = fhdr._rhdr._serial; |
| } else { |
| inFileStream_.close(); |
| if (currentJournalFileItr_ == fileNumberMap_.begin()) { |
| journalEmptyFlag_ = true; |
| } |
| return false; |
| } |
| return true; |
| } |
| |
| // static private |
| bool RecoveryManager::readJournalFileHeader(const std::string& journalFileName, |
| ::file_hdr_t& fileHeaderRef, |
| std::string& queueName) { |
| const std::size_t headerBlockSize = QLS_JRNL_FHDR_RES_SIZE_SBLKS * QLS_SBLK_SIZE_KIB * 1024; |
| char buffer[headerBlockSize]; |
| std::ifstream ifs(journalFileName.c_str(), std::ifstream::in | std::ifstream::binary); |
| if (!ifs.good()) { |
| std::ostringstream oss; |
| oss << "File=" << journalFileName; |
| throw jexception(jerrno::JERR_RCVM_OPENRD, oss.str(), "RecoveryManager", "readJournalFileHeader"); |
| } |
| ifs.read(buffer, headerBlockSize); |
| if (!ifs) { |
| std::streamsize s = ifs.gcount(); |
| ifs.close(); |
| std::ostringstream oss; |
| oss << "File=" << journalFileName << "; attempted_read_size=" << headerBlockSize << "; actual_read_size=" << s; |
| throw jexception(jerrno::JERR_RCVM_READ, oss.str(), "RecoveryManager", "readJournalFileHeader"); |
| } |
| ifs.close(); |
| ::memcpy(&fileHeaderRef, buffer, sizeof(::file_hdr_t)); |
| if (::file_hdr_check(&fileHeaderRef, QLS_FILE_MAGIC, QLS_JRNL_VERSION, 0, QLS_MAX_QUEUE_NAME_LEN)) { |
| return false; |
| } |
| queueName.assign(buffer + sizeof(::file_hdr_t), fileHeaderRef._queue_name_len); |
| return true; |
| } |
| |
| void RecoveryManager::removeEmptyFiles(EmptyFilePool* emptyFilePoolPtr) { |
| while (fileNumberMap_.begin()->second->journalFilePtr_->getEnqueuedRecordCount() == 0 && fileNumberMap_.size() > 1) { |
| RecoveredFileData_t* rfdp = fileNumberMap_.begin()->second; |
| emptyFilePoolPtr->returnEmptyFileSymlink(rfdp->journalFilePtr_->getFqFileName()); |
| delete rfdp->journalFilePtr_; |
| delete rfdp; |
| fileNumberMap_.erase(fileNumberMap_.begin()->first); |
| } |
| } |
| |
| void RecoveryManager::removeUninitFiles(EmptyFilePool* emptyFilePoolPtr) { |
| while (uninitFileList_.size() > 0) { |
| emptyFilePoolPtr->returnEmptyFileSymlink(uninitFileList_.back()); |
| uninitFileList_.pop_back(); |
| }; |
| } |
| |
| }}} |
