| /* |
| * |
| * 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/JournalFile.h" |
| |
| #include <fcntl.h> |
| #include "qpid/linearstore/journal/jcfg.h" |
| #include "qpid/linearstore/journal/pmgr.h" |
| #include "qpid/linearstore/journal/utils/file_hdr.h" |
| #include <unistd.h> |
| |
| namespace qpid { |
| namespace linearstore { |
| namespace journal { |
| |
| JournalFile::JournalFile(const std::string& fqFileName, |
| const efpIdentity_t& efpIdentity, |
| const uint64_t fileSeqNum, |
| const std::string queueName) : |
| efpIdentity_(efpIdentity), |
| fqFileName_(fqFileName), |
| fileSeqNum_(fileSeqNum), |
| queueName_(queueName), |
| serial_(getRandom64()), |
| firstRecordOffset_(0ULL), |
| fileHandle_(-1), |
| fileCloseFlag_(false), |
| fileHeaderBasePtr_ (0), |
| fileHeaderPtr_(0), |
| aioControlBlockPtr_(0), |
| fileSize_dblks_(((efpIdentity.ds_ * 1024) + (QLS_JRNL_FHDR_RES_SIZE_SBLKS * QLS_SBLK_SIZE_BYTES)) / QLS_DBLK_SIZE_BYTES), |
| initializedFlag_(false), |
| enqueuedRecordCount_("JournalFile::enqueuedRecordCount", 0), |
| submittedDblkCount_("JournalFile::submittedDblkCount", 0), |
| completedDblkCount_("JournalFile::completedDblkCount", 0), |
| outstandingAioOpsCount_("JournalFile::outstandingAioOpsCount", 0) |
| {} |
| |
| JournalFile::JournalFile(const std::string& fqFileName, |
| const ::file_hdr_t& fileHeader, |
| const std::string queueName) : |
| efpIdentity_(fileHeader._efp_partition, fileHeader._data_size_kib), |
| fqFileName_(fqFileName), |
| fileSeqNum_(fileHeader._file_number), |
| queueName_(queueName), |
| serial_(fileHeader._rhdr._serial), |
| firstRecordOffset_(fileHeader._fro), |
| fileHandle_(-1), |
| fileCloseFlag_(false), |
| fileHeaderBasePtr_ (0), |
| fileHeaderPtr_(0), |
| aioControlBlockPtr_(0), |
| fileSize_dblks_(((fileHeader._data_size_kib * 1024) + (QLS_JRNL_FHDR_RES_SIZE_SBLKS * QLS_SBLK_SIZE_BYTES)) / QLS_DBLK_SIZE_BYTES), |
| initializedFlag_(false), |
| enqueuedRecordCount_("JournalFile::enqueuedRecordCount", 0), |
| submittedDblkCount_("JournalFile::submittedDblkCount", 0), |
| completedDblkCount_("JournalFile::completedDblkCount", 0), |
| outstandingAioOpsCount_("JournalFile::outstandingAioOpsCount", 0) |
| {} |
| |
| JournalFile::~JournalFile() { |
| finalize(); |
| } |
| |
| void |
| JournalFile::initialize(const uint32_t completedDblkCount) { |
| if (!initializedFlag_) { |
| if (::posix_memalign(&fileHeaderBasePtr_, QLS_AIO_ALIGN_BOUNDARY_BYTES, QLS_JRNL_FHDR_RES_SIZE_SBLKS * QLS_SBLK_SIZE_KIB * 1024)) |
| { |
| std::ostringstream oss; |
| oss << "posix_memalign(): blksize=" << QLS_AIO_ALIGN_BOUNDARY_BYTES << " size=" << (QLS_JRNL_FHDR_RES_SIZE_SBLKS * QLS_SBLK_SIZE_KIB * 1024); |
| oss << FORMAT_SYSERR(errno); |
| throw jexception(jerrno::JERR__MALLOC, oss.str(), "JournalFile", "initialize"); |
| } |
| fileHeaderPtr_ = (::file_hdr_t*)fileHeaderBasePtr_; |
| aioControlBlockPtr_ = new aio_cb; |
| initializedFlag_ = true; |
| } |
| if (completedDblkCount > 0UL) { |
| submittedDblkCount_.set(completedDblkCount); |
| completedDblkCount_.set(completedDblkCount); |
| } |
| } |
| |
| void |
| JournalFile::finalize() { |
| if (fileHeaderBasePtr_ != 0) { |
| std::free(fileHeaderBasePtr_); |
| fileHeaderBasePtr_ = 0; |
| fileHeaderPtr_ = 0; |
| } |
| if (aioControlBlockPtr_ != 0) { |
| delete(aioControlBlockPtr_); |
| aioControlBlockPtr_ = 0; |
| } |
| } |
| |
| const std::string JournalFile::getFqFileName() const { |
| return fqFileName_; |
| } |
| |
| uint64_t JournalFile::getFileSeqNum() const { |
| return fileSeqNum_; |
| } |
| |
| uint64_t JournalFile::getSerial() const { |
| return serial_; |
| } |
| |
| int JournalFile::open() { |
| fileHandle_ = ::open(fqFileName_.c_str(), O_WRONLY | O_DIRECT, S_IRUSR | S_IWUSR | S_IRGRP | S_IROTH); // 0644 -rw-r--r-- |
| if (fileHandle_ < 0) { |
| std::ostringstream oss; |
| oss << "file=\"" << fqFileName_ << "\"" << FORMAT_SYSERR(errno); |
| throw jexception(jerrno::JERR_JNLF_OPEN, oss.str(), "JournalFile", "open"); |
| } |
| return fileHandle_; |
| } |
| |
| void JournalFile::close() { |
| if (fileHandle_ >= 0) { |
| if (getOutstandingAioDblks()) { |
| fileCloseFlag_ = true; // Close later when all outstanding AIOs have returned |
| } else { |
| int res = ::close(fileHandle_); |
| fileHandle_ = -1; |
| if (res != 0) { |
| std::ostringstream oss; |
| oss << "file=\"" << fqFileName_ << "\"" << FORMAT_SYSERR(errno); |
| throw jexception(jerrno::JERR_JNLF_CLOSE, oss.str(), "JournalFile", "open"); |
| } |
| } |
| } |
| } |
| |
| void JournalFile::asyncFileHeaderWrite(io_context_t ioContextPtr, |
| const efpPartitionNumber_t efpPartitionNumber, |
| const efpDataSize_kib_t efpDataSize_kib, |
| const uint16_t userFlags, |
| const uint64_t recordId, |
| const uint64_t firstRecordOffset) { |
| firstRecordOffset_ = firstRecordOffset; |
| ::file_hdr_create(fileHeaderPtr_, QLS_FILE_MAGIC, QLS_JRNL_VERSION, QLS_JRNL_FHDR_RES_SIZE_SBLKS, efpPartitionNumber, efpDataSize_kib); |
| ::file_hdr_init(fileHeaderBasePtr_, |
| QLS_JRNL_FHDR_RES_SIZE_SBLKS * QLS_SBLK_SIZE_KIB * 1024, |
| userFlags, |
| serial_, |
| recordId, |
| firstRecordOffset, |
| fileSeqNum_, |
| queueName_.size(), |
| queueName_.data()); |
| const std::size_t wr_size = QLS_JRNL_FHDR_RES_SIZE_SBLKS * QLS_SBLK_SIZE_KIB * 1024; |
| if (!isOpen()) open(); |
| aio::prep_pwrite(aioControlBlockPtr_, fileHandle_, (void*)fileHeaderBasePtr_, wr_size, 0UL); |
| if (!aio::is_aligned(aioControlBlockPtr_->u.c.buf, QLS_AIO_ALIGN_BOUNDARY_BYTES)) { |
| std::ostringstream oss; |
| oss << "AIO operation on misaligned buffer: iocb->u.c.buf=" << aioControlBlockPtr_->u.c.buf << std::endl; |
| throw jexception(jerrno::JERR__AIO, oss.str(), "JournalFile", "asyncFileHeaderWrite"); |
| } |
| if (aio::submit(ioContextPtr, 1, &aioControlBlockPtr_) < 0) { |
| std::ostringstream oss; |
| oss << "queue=\"" << queueName_ << "\" fid=0x" << std::hex << fileSeqNum_ << " wr_size=0x" << wr_size << " foffs=0x0"; |
| throw jexception(jerrno::JERR__AIO, oss.str(), "JournalFile", "asyncFileHeaderWrite"); |
| } |
| addSubmittedDblkCount(QLS_JRNL_FHDR_RES_SIZE_SBLKS * QLS_SBLK_SIZE_DBLKS); |
| incrOutstandingAioOperationCount(); |
| } |
| |
| void JournalFile::asyncPageWrite(io_context_t ioContextPtr, |
| aio_cb* aioControlBlockPtr, |
| void* data, |
| uint32_t dataSize_dblks) { |
| const std::size_t wr_size = dataSize_dblks * QLS_DBLK_SIZE_BYTES; |
| const uint64_t foffs = submittedDblkCount_.get() * QLS_DBLK_SIZE_BYTES; |
| if (!isOpen()) open(); |
| aio::prep_pwrite_2(aioControlBlockPtr, fileHandle_, data, wr_size, foffs); |
| if (!aio::is_aligned(aioControlBlockPtr->u.c.buf, QLS_AIO_ALIGN_BOUNDARY_BYTES)) { |
| std::ostringstream oss; |
| oss << "AIO operation on misaligned buffer: iocb->u.c.buf=" << aioControlBlockPtr->u.c.buf << std::endl; |
| throw jexception(jerrno::JERR__AIO, oss.str(), "JournalFile", "asyncPageWrite"); |
| } |
| pmgr::page_cb* pcbp = (pmgr::page_cb*)(aioControlBlockPtr->data); // This page's control block (pcb) |
| pcbp->_wdblks = dataSize_dblks; |
| pcbp->_jfp = this; |
| if (aio::submit(ioContextPtr, 1, &aioControlBlockPtr) < 0) { |
| std::ostringstream oss; |
| oss << "queue=\"" << queueName_ << "\" fid=0x" << std::hex << fileSeqNum_ << " wr_size=0x" << wr_size << " foffs=0x" << foffs; |
| throw jexception(jerrno::JERR__AIO, oss.str(), "JournalFile", "asyncPageWrite"); |
| } |
| addSubmittedDblkCount(dataSize_dblks); |
| incrOutstandingAioOperationCount(); |
| } |
| |
| uint32_t JournalFile::getEnqueuedRecordCount() const { |
| return enqueuedRecordCount_.get(); |
| } |
| |
| uint32_t JournalFile::incrEnqueuedRecordCount() { |
| return enqueuedRecordCount_.increment(); |
| } |
| |
| uint32_t JournalFile::decrEnqueuedRecordCount() { |
| return enqueuedRecordCount_.decrementLimit(); |
| } |
| |
| uint32_t JournalFile::addCompletedDblkCount(const uint32_t a) { |
| return completedDblkCount_.addLimit(a, submittedDblkCount_.get(), jerrno::JERR_JNLF_CMPLOFFSOVFL); |
| } |
| |
| uint16_t JournalFile::getOutstandingAioOperationCount() const { |
| return outstandingAioOpsCount_.get(); |
| } |
| |
| uint16_t JournalFile::decrOutstandingAioOperationCount() { |
| uint16_t r = outstandingAioOpsCount_.decrementLimit(); |
| if (fileCloseFlag_ && outstandingAioOpsCount_ == 0) { // Delayed close |
| close(); |
| } |
| return r; |
| } |
| |
| efpIdentity_t JournalFile::getEfpIdentity() const { |
| return efpIdentity_; |
| } |
| |
| uint64_t JournalFile::getFirstRecordOffset() const { |
| return firstRecordOffset_; |
| } |
| |
| void JournalFile::setFirstRecordOffset(const uint64_t firstRecordOffset) { |
| firstRecordOffset_ = firstRecordOffset; |
| } |
| |
| // --- Status helper functions --- |
| |
| bool JournalFile::isEmpty() const { |
| return submittedDblkCount_ == 0; |
| } |
| |
| bool JournalFile::isNoEnqueuedRecordsRemaining() const { |
| return /*!enqueueStarted_ &&*/ // Not part-way through encoding an enqueue |
| isFullAndComplete() && // Full with all AIO returned |
| enqueuedRecordCount_ == 0; // No remaining enqueued records |
| } |
| |
| // debug aid |
| const std::string JournalFile::status_str(const uint8_t indentDepth) const { |
| std::string indent((size_t)indentDepth, '.'); |
| std::ostringstream oss; |
| oss << indent << "JournalFile: fileName=" << getFileName() << std::endl; |
| oss << indent << " directory=" << getDirectory() << std::endl; |
| oss << indent << " fileSizeDblks=" << fileSize_dblks_ << std::endl; |
| oss << indent << " open=" << (isOpen() ? "T" : "F") << std::endl; |
| oss << indent << " fileHandle=" << fileHandle_ << std::endl; |
| oss << indent << " enqueuedRecordCount=" << getEnqueuedRecordCount() << std::endl; |
| oss << indent << " submittedDblkCount=" << getSubmittedDblkCount() << std::endl; |
| oss << indent << " completedDblkCount=" << getCompletedDblkCount() << std::endl; |
| oss << indent << " outstandingAioOpsCount=" << getOutstandingAioOperationCount() << std::endl; |
| oss << indent << " isEmpty()=" << (isEmpty() ? "T" : "F") << std::endl; |
| oss << indent << " isDataEmpty()=" << (isDataEmpty() ? "T" : "F") << std::endl; |
| oss << indent << " dblksRemaining()=" << dblksRemaining() << std::endl; |
| oss << indent << " isFull()=" << (isFull() ? "T" : "F") << std::endl; |
| oss << indent << " isFullAndComplete()=" << (isFullAndComplete() ? "T" : "F") << std::endl; |
| oss << indent << " getOutstandingAioDblks()=" << getOutstandingAioDblks() << std::endl; |
| oss << indent << " getNextFile()=" << (getNextFile() ? "T" : "F") << std::endl; |
| return oss.str(); |
| } |
| |
| // --- protected functions --- |
| |
| const std::string JournalFile::getDirectory() const { |
| return fqFileName_.substr(0, fqFileName_.rfind('/')); |
| } |
| |
| const std::string JournalFile::getFileName() const { |
| return fqFileName_.substr(fqFileName_.rfind('/')+1); |
| } |
| |
| //static |
| uint64_t JournalFile::getRandom64() { |
| // TODO: ::rand() is not thread safe, either lock or use rand_r(seed) with a thread-local seed. |
| return ((uint64_t)::rand() << QLS_RAND_SHIFT1) | ((uint64_t)::rand() << QLS_RAND_SHIFT2) | (::rand() & QLS_RAND_MASK); |
| } |
| |
| bool JournalFile::isOpen() const { |
| return fileHandle_ >= 0; |
| } |
| |
| uint32_t JournalFile::getSubmittedDblkCount() const { |
| return submittedDblkCount_.get(); |
| } |
| |
| uint32_t JournalFile::addSubmittedDblkCount(const uint32_t a) { |
| return submittedDblkCount_.addLimit(a, fileSize_dblks_, jerrno::JERR_JNLF_FILEOFFSOVFL); |
| } |
| |
| uint32_t JournalFile::getCompletedDblkCount() const { |
| return completedDblkCount_.get(); |
| } |
| |
| uint16_t JournalFile::incrOutstandingAioOperationCount() { |
| return outstandingAioOpsCount_.increment(); |
| } |
| |
| u_int32_t JournalFile::dblksRemaining() const { |
| return fileSize_dblks_ - submittedDblkCount_; |
| } |
| |
| bool JournalFile::getNextFile() const { |
| return isFull(); |
| } |
| |
| u_int32_t JournalFile::getOutstandingAioDblks() const { |
| return submittedDblkCount_ - completedDblkCount_; |
| } |
| |
| bool JournalFile::isDataEmpty() const { |
| return submittedDblkCount_ <= QLS_JRNL_FHDR_RES_SIZE_SBLKS * QLS_SBLK_SIZE_DBLKS; |
| } |
| |
| bool JournalFile::isFull() const { |
| return submittedDblkCount_ == fileSize_dblks_; |
| } |
| |
| bool JournalFile::isFullAndComplete() const { |
| return completedDblkCount_ == fileSize_dblks_; |
| } |
| |
| |
| }}} |
