QPID-4984: WIP - compiles, but not functional.

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

Author: Kim van der Riet

qpid/cpp/src/linearstore.cmake

@@ -80,12 +80,18 @@ if (BUILD_LINEARSTORE)
         qpid/linearstore/jrnl/deq_rec.cpp
         qpid/linearstore/jrnl/enq_map.cpp
         qpid/linearstore/jrnl/enq_rec.cpp
+        qpid/linearstore/jrnl/EmptyFilePool.cpp
+        qpid/linearstore/jrnl/EmptyFilePoolManager.cpp
+        qpid/linearstore/jrnl/EmptyFilePoolPartition.cpp
         #qpid/linearstore/jrnl/fcntl.cpp
         qpid/linearstore/jrnl/jcntl.cpp
         qpid/linearstore/jrnl/jdir.cpp
         qpid/linearstore/jrnl/jerrno.cpp
         qpid/linearstore/jrnl/jexception.cpp
         #qpid/linearstore/jrnl/jinf.cpp
+		qpid/linearstore/jrnl/JournalFile.cpp
+		qpid/linearstore/jrnl/JournalFileController.cpp
+		qpid/linearstore/jrnl/JournalLog.cpp
         qpid/linearstore/jrnl/jrec.cpp
         #qpid/linearstore/jrnl/lp_map.cpp
         #qpid/linearstore/jrnl/lpmgr.cpp
@@ -106,6 +112,7 @@ if (BUILD_LINEARSTORE)
         qpid/linearstore/BindingDbt.cpp
         qpid/linearstore/BufferValue.cpp
         qpid/linearstore/DataTokenImpl.cpp
+        qpid/linearstore/EmptyFilePoolManagerImpl.cpp
         qpid/linearstore/IdDbt.cpp
         qpid/linearstore/IdSequence.cpp
         qpid/linearstore/JournalImpl.cpp
@@ -118,9 +125,9 @@ if (BUILD_LINEARSTORE)
         qpid/linearstore/jrnl/utils/deq_hdr.c
         qpid/linearstore/jrnl/utils/enq_hdr.c
         qpid/linearstore/jrnl/utils/file_hdr.c
-	qpid/linearstore/jrnl/utils/rec_hdr.c
-	qpid/linearstore/jrnl/utils/rec_tail.c
-	qpid/linearstore/jrnl/utils/txn_hdr.c
+        qpid/linearstore/jrnl/utils/rec_hdr.c
+        qpid/linearstore/jrnl/utils/rec_tail.c
+        qpid/linearstore/jrnl/utils/txn_hdr.c
     )
 
     # linearstore include directories

qpid/cpp/src/qpid/legacystore/jrnl/rmgr.cpp

@@ -72,7 +72,7 @@ rmgr::initialize(aio_callback* const cbp)
         throw jexception(jerrno::JERR__MALLOC, oss.str(), "rmgr", "initialize");
     }
     _fhdr_aio_cb_ptr = new aio_cb;
-    std::memset(_fhdr_aio_cb_ptr, 0, sizeof(aio_cb*));
+    std::memset(_fhdr_aio_cb_ptr, 0, sizeof(aio_cb));
 }
 
 void

qpid/cpp/src/qpid/linearstore/EmptyFilePoolManagerImpl.cpp

@@ -0,0 +1,59 @@
+/*
+ *
+ * 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 "EmptyFilePoolManagerImpl.h"
+
+#include "QpidLog.h"
+#include "qpid/linearstore/jrnl/EmptyFilePoolTypes.h"
+
+namespace qpid {
+namespace linearstore {
+
+EmptyFilePoolManagerImpl::EmptyFilePoolManagerImpl(const std::string& qlsStorePath) :
+            qpid::qls_jrnl::EmptyFilePoolManager(qlsStorePath)
+{}
+
+EmptyFilePoolManagerImpl::~EmptyFilePoolManagerImpl() {}
+
+void EmptyFilePoolManagerImpl::findEfpPartitions() {
+    qpid::qls_jrnl::EmptyFilePoolManager::findEfpPartitions();
+    QLS_LOG(info, "EFP Manager initialization complete");
+    std::vector<qpid::qls_jrnl::EmptyFilePoolPartition*> partitionList;
+    std::vector<qpid::qls_jrnl::EmptyFilePool*> filePoolList;
+    getEfpPartitions(partitionList);
+    if (partitionList.size() == 0) {
+        QLS_LOG(error, "NO EFP PARTITIONS FOUND! No queue creation is possible.")
+    } else {
+        QLS_LOG(info, "> EFP Partitions found: " << partitionList.size());
+        for (std::vector<qpid::qls_jrnl::EmptyFilePoolPartition*>::const_iterator i=partitionList.begin(); i!= partitionList.end(); ++i) {
+            filePoolList.clear();
+            (*i)->getEmptyFilePools(filePoolList);
+            QLS_LOG(info, "  * Partition " << (*i)->partitionNumber() << " containing " << filePoolList.size() << " pool" <<
+                          (filePoolList.size()>1 ? "s" : "") << " at \'" << (*i)->partitionDirectory() << "\'");
+            for (std::vector<qpid::qls_jrnl::EmptyFilePool*>::const_iterator j=filePoolList.begin(); j!=filePoolList.end(); ++j) {
+                QLS_LOG(info, "    - EFP \'" << (*j)->fileSizeKib() << "k\' containing " << (*j)->numEmptyFiles() <<
+                              " files of size " << (*j)->fileSizeKib() << " KiB totaling " << (*j)->cumFileSizeKib() << " KiB");
+            }
+        }
+    }
+}
+
+}} /* namespace qpid::linearstore */

qpid/cpp/src/qpid/linearstore/EmptyFilePoolManagerImpl.h

@@ -0,0 +1,40 @@
+/*
+ *
+ * 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.
+ *
+ */
+
+#ifndef QPID_LINEARSTORE_EMPTYFILEPOOLMANAGERIMPL_H_
+#define QPID_LINEARSTORE_EMPTYFILEPOOLMANAGERIMPL_H_
+
+#include "qpid/linearstore/jrnl/EmptyFilePoolManager.h"
+
+namespace qpid {
+namespace linearstore {
+
+class EmptyFilePoolManagerImpl: public qpid::qls_jrnl::EmptyFilePoolManager
+{
+public:
+    EmptyFilePoolManagerImpl(const std::string& qlsStorePath);
+    virtual ~EmptyFilePoolManagerImpl();
+    void findEfpPartitions();
+};
+
+}} /* namespace qpid::linearstore */
+
+#endif /* QPID_LINEARSTORE_EMPTYFILEPOOLMANAGERIMPL_H_ */

qpid/cpp/src/qpid/linearstore/JournalImpl.cpp

@@ -23,16 +23,17 @@
 
 #include "qpid/linearstore/jrnl/jerrno.h"
 #include "qpid/linearstore/jrnl/jexception.h"
+#include "qpid/linearstore/jrnl/EmptyFilePool.h"
 #include "qpid/log/Statement.h"
 #include "qpid/management/ManagementAgent.h"
-#include "qmf/org/apache/qpid/linearstore/ArgsJournalExpand.h"
+//#include "qmf/org/apache/qpid/linearstore/ArgsJournalExpand.h"
 #include "qmf/org/apache/qpid/linearstore/EventCreated.h"
 #include "qmf/org/apache/qpid/linearstore/EventEnqThresholdExceeded.h"
 #include "qmf/org/apache/qpid/linearstore/EventFull.h"
 #include "qmf/org/apache/qpid/linearstore/EventRecovered.h"
 #include "qpid/sys/Monitor.h"
 #include "qpid/sys/Timer.h"
-#include "qpid/linearstore/Log.h"
+#include "qpid/linearstore/QpidLog.h"
 #include "qpid/linearstore/StoreException.h"
 
 using namespace qpid::qls_jrnl;
@@ -53,22 +54,23 @@ void GetEventsFireEvent::fire() { qpid::sys::Mutex::ScopedLock sl(_gefe_lock); i
 JournalImpl::JournalImpl(qpid::sys::Timer& timer_,
                          const std::string& journalId,
                          const std::string& journalDirectory,
-                         const std::string& journalBaseFilename,
+//                         const std::string& journalBaseFilename,
                          const qpid::sys::Duration getEventsTimeout,
                          const qpid::sys::Duration flushTimeout,
                          qpid::management::ManagementAgent* a,
                          DeleteCallback onDelete):
-                         jcntl(journalId, journalDirectory, journalBaseFilename),
+                         jcntl(journalId, journalDirectory/*, journalBaseFilename*/),
                          timer(timer_),
                          getEventsTimerSetFlag(false),
-                         lastReadRid(0),
+                         efpp(0),
+//                         lastReadRid(0),
                          writeActivityFlag(false),
                          flushTriggeredFlag(true),
-                         _xidp(0),
-                         _datap(0),
-                         _dlen(0),
-                         _dtok(),
-                         _external(false),
+//                         _xidp(0),
+//                         _datap(0),
+//                         _dlen(0),
+//                         _dtok(),
+//                         _external(false),
                          deleteCallback(onDelete)
 {
     getEventsFireEventsPtr = new GetEventsFireEvent(this, getEventsTimeout);
@@ -82,7 +84,7 @@ JournalImpl::JournalImpl(qpid::sys::Timer& timer_,
 
     QLS_LOG2(notice, _jid, "Created");
     std::ostringstream oss;
-    oss << "Journal directory = \"" << journalDirectory << "\"; Base file name = \"" << journalBaseFilename << "\"";
+    oss << "Journal directory = \"" << journalDirectory << "\"";
     QLS_LOG2(debug, _jid, oss.str());
 }
 
@@ -95,7 +97,7 @@ JournalImpl::~JournalImpl()
 	}
     getEventsFireEventsPtr->cancel();
     inactivityFireEventPtr->cancel();
-    free_read_buffers();
+//    free_read_buffers();
 
     if (_mgmtObject.get() != 0) {
         _mgmtObject->resourceDestroy();
@@ -116,14 +118,14 @@ JournalImpl::initManagement(qpid::management::ManagementAgent* a)
 
         _mgmtObject->set_name(_jid);
         _mgmtObject->set_directory(_jdir.dirname());
-        _mgmtObject->set_baseFileName(_base_filename);
+//        _mgmtObject->set_baseFileName(_base_filename);
         _mgmtObject->set_readPageSize(JRNL_RMGR_PAGE_SIZE * JRNL_SBLK_SIZE);
         _mgmtObject->set_readPages(JRNL_RMGR_PAGES);
 
         // The following will be set on initialize(), but being properties, these must be set to 0 in the meantime
-        _mgmtObject->set_initialFileCount(0);
-        _mgmtObject->set_dataFileSize(0);
-        _mgmtObject->set_currentFileCount(0);
+        //_mgmtObject->set_initialFileCount(0);
+        //_mgmtObject->set_dataFileSize(0);
+        //_mgmtObject->set_currentFileCount(0);
         _mgmtObject->set_writePageSize(0);
         _mgmtObject->set_writePages(0);
 
@@ -133,22 +135,23 @@ JournalImpl::initManagement(qpid::management::ManagementAgent* a)
 
 
 void
-JournalImpl::initialize(/*const uint16_t num_jfiles,
-                        const bool auto_expand,
-                        const uint16_t ae_max_jfiles,
-                        const uint32_t jfsize_sblks,*/
+JournalImpl::initialize(qpid::qls_jrnl::EmptyFilePool* efpp_,
                         const uint16_t wcache_num_pages,
                         const uint32_t wcache_pgsize_sblks,
                         qpid::qls_jrnl::aio_callback* const cbp)
 {
-    std::ostringstream oss;
-//    oss << "Initialize; num_jfiles=" << num_jfiles << " jfsize_sblks=" << jfsize_sblks;
-    oss << "Initialize;";
-    oss << " wcache_pgsize_sblks=" << wcache_pgsize_sblks;
-    oss << " wcache_num_pages=" << wcache_num_pages;
-    QLS_LOG2(debug, _jid, oss.str());
-    jcntl::initialize(/*num_jfiles, auto_expand, ae_max_jfiles, jfsize_sblks,*/ wcache_num_pages, wcache_pgsize_sblks, cbp);
-    QLS_LOG2(debug, _jid, "Initialization complete");
+    efpp = efpp_;
+//    efpp->createJournal(_jdir);
+//    QLS_LOG2(notice, _jid, "Initialized");
+//    std::ostringstream oss;
+////    oss << "Initialize; num_jfiles=" << num_jfiles << " jfsize_sblks=" << jfsize_sblks;
+//    oss << "Initialize; efpPartitionNumber=" << efpp_->getPartitionNumber();
+//    oss << " efpFileSizeKb=" << efpp_->fileSizeKib();
+//    oss << " wcache_pgsize_sblks=" << wcache_pgsize_sblks;
+//    oss << " wcache_num_pages=" << wcache_num_pages;
+//    QLS_LOG2(debug, _jid, oss.str());
+    jcntl::initialize(/*num_jfiles, auto_expand, ae_max_jfiles, jfsize_sblks,*/ efpp, wcache_num_pages, wcache_pgsize_sblks, cbp);
+//    QLS_LOG2(debug, _jid, "Initialization complete");
     // TODO: replace for linearstore: _lpmgr
 /*
     if (_mgmtObject.get() != 0)
@@ -261,6 +264,7 @@ JournalImpl::recover_complete()
 //#define AIO_SLEEP_TIME_US   10 // 0.01 ms
 // Return true if content is recovered from store; false if content is external and must be recovered from an external store.
 // Throw exception for all errors.
+/*
 bool
 JournalImpl::loadMsgContent(uint64_t rid, std::string& data, size_t length, size_t offset)
 {
@@ -351,6 +355,7 @@ JournalImpl::loadMsgContent(uint64_t rid, std::string& data, size_t length, size
     }
     return true;
 }
+*/
 
 void
 JournalImpl::enqueue_data_record(const void* const data_buff, const size_t tot_data_len,
@@ -574,6 +579,7 @@ void
 JournalImpl::rd_aio_cb(std::vector<uint16_t>& /*pil*/)
 {}
 
+/*
 void
 JournalImpl::free_read_buffers()
 {
@@ -586,6 +592,12 @@ JournalImpl::free_read_buffers()
         _datap = 0;
     }
 }
+*/
+
+void
+JournalImpl::createStore() {
+
+}
 
 void
 JournalImpl::handleIoResult(const iores r)
@@ -624,12 +636,13 @@ JournalImpl::handleIoResult(const iores r)
     }
 }
 
-qpid::management::Manageable::status_t JournalImpl::ManagementMethod (uint32_t methodId,
+qpid::management::Manageable::status_t JournalImpl::ManagementMethod (uint32_t /*methodId*/,
                                                                       qpid::management::Args& /*args*/,
                                                                       std::string& /*text*/)
 {
     Manageable::status_t status = Manageable::STATUS_UNKNOWN_METHOD;
 
+/*
     switch (methodId)
     {
     case _qmf::Journal::METHOD_EXPAND :
@@ -640,6 +653,7 @@ qpid::management::Manageable::status_t JournalImpl::ManagementMethod (uint32_t m
         status = Manageable::STATUS_NOT_IMPLEMENTED;
         break;
     }
+*/
 
     return status;
 }