NO-JIRA: create branch for qpid-cpp 0.34 RC process

git-svn-id: https://svn.apache.org/repos/asf/qpid/branches/qpid-cpp-0.34-rc@1687469 13f79535-47bb-0310-9956-ffa450edef68

1 files changed, 349 insertions, 0 deletions

diff --git a/qpid/cpp/src/qpid/linearstore/journal/JournalFile.cpp b/qpid/cpp/src/qpid/linearstore/journal/JournalFile.cpp
new file mode 100644
index 0000000000..ed03a8413f
--- /dev/null
+++ b/qpid/cpp/src/qpid/linearstore/journal/JournalFile.cpp
@@ -0,0 +1,349 @@
+/*
+ *
+ * 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_;
+}
+
+
+}}}