You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@qpid.apache.org by kp...@apache.org on 2013/09/20 18:43:11 UTC
svn commit: r1525050 [3/4] - in /qpid/branches/linearstore/qpid: cpp/src/
cpp/src/qpid/legacystore/jrnl/ cpp/src/qpid/linearstore/
cpp/src/qpid/linearstore/jrnl/ cpp/src/qpid/linearstore/jrnl/utils/
tools/src/py/linearstore/
Added: qpid/branches/linearstore/qpid/cpp/src/qpid/linearstore/jrnl/EmptyFilePoolManager.cpp
URL: http://svn.apache.org/viewvc/qpid/branches/linearstore/qpid/cpp/src/qpid/linearstore/jrnl/EmptyFilePoolManager.cpp?rev=1525050&view=auto
==============================================================================
--- qpid/branches/linearstore/qpid/cpp/src/qpid/linearstore/jrnl/EmptyFilePoolManager.cpp (added)
+++ qpid/branches/linearstore/qpid/cpp/src/qpid/linearstore/jrnl/EmptyFilePoolManager.cpp Fri Sep 20 16:43:10 2013
@@ -0,0 +1,175 @@
+/*
+ *
+ * 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 "EmptyFilePoolManager.h"
+
+#include <dirent.h>
+#include "qpid/linearstore/jrnl/EmptyFilePoolPartition.h"
+#include "qpid/linearstore/jrnl/jdir.h"
+#include "qpid/linearstore/jrnl/slock.h"
+#include <vector>
+
+// DEBUG
+//#include <iostream>
+
+namespace qpid {
+namespace qls_jrnl {
+
+EmptyFilePoolManager::EmptyFilePoolManager(const std::string& qlsStorePath_) :
+ qlsStorePath(qlsStorePath_)
+{}
+
+EmptyFilePoolManager::~EmptyFilePoolManager() {
+ slock l(partitionMapMutex);
+ for (partitionMapItr_t i = partitionMap.begin(); i != partitionMap.end(); ++i) {
+ delete i->second;
+ }
+ partitionMap.clear();
+}
+
+void
+EmptyFilePoolManager::findEfpPartitions() {
+ //std::cout << "*** Reading " << qlsStorePath << std::endl; // DEBUG
+ std::vector<std::string> dirList;
+ jdir::read_dir(qlsStorePath, dirList, true, false, true);
+ for (std::vector<std::string>::iterator i = dirList.begin(); i != dirList.end(); ++i) {
+ if ((*i)[0] == 'p' && i->length() == 4) { // Filter: look only at names pNNN
+ efpPartitionNumber_t pn = ::atoi(i->c_str() + 1);
+ std::string fullDirPath(qlsStorePath + "/" + (*i));
+ EmptyFilePoolPartition* efppp = 0;
+ try {
+ efppp = new EmptyFilePoolPartition(pn, fullDirPath);
+ {
+ slock l(partitionMapMutex);
+ partitionMap[pn] = efppp;
+ }
+ } catch (const std::exception& e) {
+ if (efppp != 0) {
+ delete efppp;
+ efppp = 0;
+ }
+ //std::cerr << "Unable to initialize partition " << pn << " (\'" << fullDirPath << "\'): " << e.what() << std::endl;
+ }
+ if (efppp != 0)
+ efppp->findEmptyFilePools();
+ }
+ }
+}
+
+uint16_t
+EmptyFilePoolManager::getNumEfpPartitions() const {
+ return partitionMap.size();
+}
+
+EmptyFilePoolPartition*
+EmptyFilePoolManager::getEfpPartition(const efpPartitionNumber_t partitionNumber) {
+ partitionMapItr_t i = partitionMap.find(partitionNumber);
+ if (i == partitionMap.end())
+ return 0;
+ else
+ return i->second;
+}
+
+void
+EmptyFilePoolManager::getEfpPartitionNumbers(std::vector<efpPartitionNumber_t>& partitionNumberList,
+ const efpFileSizeKib_t efpFileSizeKb) const {
+ slock l(partitionMapMutex);
+ for (partitionMapConstItr_t i=partitionMap.begin(); i!=partitionMap.end(); ++i) {
+ if (efpFileSizeKb == 0) {
+ partitionNumberList.push_back(i->first);
+ } else {
+ std::vector<efpFileSizeKib_t> efpFileSizeList;
+ i->second->getEmptyFilePoolSizesKb(efpFileSizeList);
+ for (std::vector<efpFileSizeKib_t>::iterator j=efpFileSizeList.begin(); j!=efpFileSizeList.end(); ++j) {
+ if (*j == efpFileSizeKb) {
+ partitionNumberList.push_back(i->first);
+ break;
+ }
+ }
+ }
+ }
+}
+
+void
+EmptyFilePoolManager::getEfpPartitions(std::vector<EmptyFilePoolPartition*>& partitionList,
+ const efpFileSizeKib_t efpFileSizeKb) {
+ slock l(partitionMapMutex);
+ for (partitionMapConstItr_t i=partitionMap.begin(); i!=partitionMap.end(); ++i) {
+ if (efpFileSizeKb == 0) {
+ partitionList.push_back(i->second);
+ } else {
+ std::vector<efpFileSizeKib_t> efpFileSizeList;
+ i->second->getEmptyFilePoolSizesKb(efpFileSizeList);
+ for (std::vector<efpFileSizeKib_t>::iterator j=efpFileSizeList.begin(); j!=efpFileSizeList.end(); ++j) {
+ if (*j == efpFileSizeKb) {
+ partitionList.push_back(i->second);
+ break;
+ }
+ }
+ }
+ }
+}
+
+void
+EmptyFilePoolManager::getEfpFileSizes(std::vector<efpFileSizeKib_t>& efpFileSizeList,
+ const efpPartitionNumber_t efpPartitionNumber) const {
+ if (efpPartitionNumber == 0) {
+ for (partitionMapConstItr_t i=partitionMap.begin(); i!=partitionMap.end(); ++i) {
+ i->second->getEmptyFilePoolSizesKb(efpFileSizeList);
+ }
+ } else {
+ partitionMapConstItr_t i = partitionMap.find(efpPartitionNumber);
+ if (i != partitionMap.end()) {
+ i->second->getEmptyFilePoolSizesKb(efpFileSizeList);
+ }
+ }
+}
+
+void
+EmptyFilePoolManager::getEmptyFilePools(std::vector<EmptyFilePool*>& emptyFilePoolList,
+ const efpPartitionNumber_t efpPartitionNumber) {
+ if (efpPartitionNumber == 0) {
+ for (partitionMapConstItr_t i=partitionMap.begin(); i!=partitionMap.end(); ++i) {
+ i->second->getEmptyFilePools(emptyFilePoolList);
+ }
+ } else {
+ partitionMapConstItr_t i = partitionMap.find(efpPartitionNumber);
+ if (i != partitionMap.end()) {
+ i->second->getEmptyFilePools(emptyFilePoolList);
+ }
+ }
+}
+
+EmptyFilePool*
+EmptyFilePoolManager::getEmptyFilePool(const efpPartitionNumber_t partitionNumber,
+ const efpFileSizeKib_t efpFileSizeKib) {
+ EmptyFilePoolPartition* efppp = getEfpPartition(partitionNumber);
+ if (efppp != 0)
+ return efppp->getEmptyFilePool(efpFileSizeKib);
+ return 0;
+}
+
+EmptyFilePool*
+EmptyFilePoolManager::getEmptyFilePool(const efpIdentity_t efpIdentity) {
+ return getEmptyFilePool(efpIdentity.first, efpIdentity.second);
+}
+
+}} // namespace qpid::qls_jrnl
Added: qpid/branches/linearstore/qpid/cpp/src/qpid/linearstore/jrnl/EmptyFilePoolManager.h
URL: http://svn.apache.org/viewvc/qpid/branches/linearstore/qpid/cpp/src/qpid/linearstore/jrnl/EmptyFilePoolManager.h?rev=1525050&view=auto
==============================================================================
--- qpid/branches/linearstore/qpid/cpp/src/qpid/linearstore/jrnl/EmptyFilePoolManager.h (added)
+++ qpid/branches/linearstore/qpid/cpp/src/qpid/linearstore/jrnl/EmptyFilePoolManager.h Fri Sep 20 16:43:10 2013
@@ -0,0 +1,63 @@
+/*
+ *
+ * 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_QLS_JRNL_EMPTYFILEPOOLMANAGER_H_
+#define QPID_QLS_JRNL_EMPTYFILEPOOLMANAGER_H_
+
+#include <map>
+#include "qpid/linearstore/jrnl/EmptyFilePoolPartition.h"
+#include "qpid/linearstore/jrnl/smutex.h"
+#include <string>
+
+namespace qpid {
+namespace qls_jrnl {
+
+class EmptyFilePoolManager
+{
+protected:
+ typedef std::map<efpPartitionNumber_t, EmptyFilePoolPartition*> partitionMap_t;
+ typedef partitionMap_t::iterator partitionMapItr_t;
+ typedef partitionMap_t::const_iterator partitionMapConstItr_t;
+
+ std::string qlsStorePath;
+ partitionMap_t partitionMap;
+ smutex partitionMapMutex;
+
+public:
+ EmptyFilePoolManager(const std::string& qlsStorePath_);
+ virtual ~EmptyFilePoolManager();
+ void findEfpPartitions();
+
+ uint16_t getNumEfpPartitions() const;
+ EmptyFilePoolPartition* getEfpPartition(const efpPartitionNumber_t partitionNumber);
+ void getEfpPartitionNumbers(std::vector<efpPartitionNumber_t>& partitionNumberList, const efpFileSizeKib_t efpFileSizeKb = 0) const;
+ void getEfpPartitions(std::vector<EmptyFilePoolPartition*>& partitionList, const efpFileSizeKib_t efpFileSizeKb = 0);
+
+ void getEfpFileSizes(std::vector<efpFileSizeKib_t>& efpFileSizeList, const efpPartitionNumber_t efpPartitionNumber = 0) const;
+ void getEmptyFilePools(std::vector<EmptyFilePool*>& emptyFilePoolList, const efpPartitionNumber_t efpPartitionNumber = 0);
+
+ EmptyFilePool* getEmptyFilePool(const efpPartitionNumber_t partitionNumber, const efpFileSizeKib_t efpFileSizeKb);
+ EmptyFilePool* getEmptyFilePool(const efpIdentity_t efpIdentity);
+};
+
+}} // namespace qpid::qls_jrnl
+
+#endif /* QPID_QLS_JRNL_EMPTYFILEPOOLMANAGER_H_ */
Added: qpid/branches/linearstore/qpid/cpp/src/qpid/linearstore/jrnl/EmptyFilePoolPartition.cpp
URL: http://svn.apache.org/viewvc/qpid/branches/linearstore/qpid/cpp/src/qpid/linearstore/jrnl/EmptyFilePoolPartition.cpp?rev=1525050&view=auto
==============================================================================
--- qpid/branches/linearstore/qpid/cpp/src/qpid/linearstore/jrnl/EmptyFilePoolPartition.cpp (added)
+++ qpid/branches/linearstore/qpid/cpp/src/qpid/linearstore/jrnl/EmptyFilePoolPartition.cpp Fri Sep 20 16:43:10 2013
@@ -0,0 +1,134 @@
+/*
+ *
+ * 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/jrnl/EmptyFilePoolPartition.h"
+
+#include <dirent.h>
+#include "qpid/linearstore/jrnl/jdir.h"
+#include "qpid/linearstore/jrnl/jerrno.h"
+#include "qpid/linearstore/jrnl/jexception.h"
+#include "qpid/linearstore/jrnl/slock.h"
+
+//#include <iostream> // DEBUG
+
+namespace qpid {
+namespace qls_jrnl {
+
+const std::string EmptyFilePoolPartition::efpTopLevelDir("efp"); // Sets the top-level efp dir within a partition
+
+EmptyFilePoolPartition::EmptyFilePoolPartition(const efpPartitionNumber_t partitionNum_, const std::string& partitionDir_) :
+ partitionNum(partitionNum_),
+ partitionDir(partitionDir_)
+{
+ validatePartitionDir();
+}
+
+EmptyFilePoolPartition::~EmptyFilePoolPartition() {
+ slock l(efpMapMutex);
+ for (efpMapItr_t i = efpMap.begin(); i != efpMap.end(); ++i) {
+ delete i->second;
+ }
+ efpMap.clear();
+}
+
+void
+EmptyFilePoolPartition::validatePartitionDir() {
+ if (!jdir::is_dir(partitionDir)) {
+ std::ostringstream ss;
+ ss << "Invalid partition directory: \'" << partitionDir << "\' is not a directory";
+ throw jexception(jerrno::JERR_EFP_BADPARTITIONDIR, ss.str(), "EmptyFilePoolPartition", "validatePartitionDir");
+ }
+ // TODO: other validity checks here
+}
+
+void
+EmptyFilePoolPartition::findEmptyFilePools() {
+ //std::cout << "Reading " << partitionDir << std::endl; // DEBUG
+ std::vector<std::string> dirList;
+ jdir::read_dir(partitionDir, dirList, true, false, false);
+ bool foundEfpDir = false;
+ for (std::vector<std::string>::iterator i = dirList.begin(); i != dirList.end(); ++i) {
+ if (i->compare(efpTopLevelDir) == 0) {
+ foundEfpDir = true;
+ break;
+ }
+ }
+ if (foundEfpDir) {
+ std::string efpDir(partitionDir + "/" + efpTopLevelDir);
+ //std::cout << "Reading " << efpDir << std::endl; // DEBUG
+ dirList.clear();
+ jdir::read_dir(efpDir, dirList, true, false, false);
+ for (std::vector<std::string>::iterator i = dirList.begin(); i != dirList.end(); ++i) {
+ std::string efpSizeDir(efpDir + "/" + (*i));
+ EmptyFilePool* efpp = 0;
+ try {
+ efpp = new EmptyFilePool(efpSizeDir, this);
+ {
+ slock l(efpMapMutex);
+ efpMap[efpp->fileSizeKib()] = efpp;
+ }
+ }
+ catch (const std::exception& e) {
+ if (efpp != 0) {
+ delete efpp;
+ efpp = 0;
+ }
+ //std::cerr << "WARNING: " << e.what() << std::endl;
+ }
+ if (efpp != 0)
+ efpp->initialize();
+ }
+ }
+}
+
+efpPartitionNumber_t
+EmptyFilePoolPartition::partitionNumber() const {
+ return partitionNum;
+}
+
+std::string
+EmptyFilePoolPartition::partitionDirectory() const {
+ return partitionDir;
+}
+
+EmptyFilePool*
+EmptyFilePoolPartition::getEmptyFilePool(const efpFileSizeKib_t efpFileSizeKb) {
+ efpMapItr_t i = efpMap.find(efpFileSizeKb);
+ if (i == efpMap.end())
+ return 0;
+ return i->second;
+}
+
+void
+EmptyFilePoolPartition::getEmptyFilePoolSizesKb(std::vector<efpFileSizeKib_t>& efpFileSizesKbList) const {
+ for (efpMapConstItr_t i=efpMap.begin(); i!=efpMap.end(); ++i) {
+ efpFileSizesKbList.push_back(i->first);
+ }
+}
+
+void
+EmptyFilePoolPartition::getEmptyFilePools(std::vector<EmptyFilePool*>& efpList) {
+ for (efpMapItr_t i=efpMap.begin(); i!=efpMap.end(); ++i) {
+ efpList.push_back(i->second);
+ }
+}
+
+}} // namespace qpid::qls_jrnl
Added: qpid/branches/linearstore/qpid/cpp/src/qpid/linearstore/jrnl/EmptyFilePoolPartition.h
URL: http://svn.apache.org/viewvc/qpid/branches/linearstore/qpid/cpp/src/qpid/linearstore/jrnl/EmptyFilePoolPartition.h?rev=1525050&view=auto
==============================================================================
--- qpid/branches/linearstore/qpid/cpp/src/qpid/linearstore/jrnl/EmptyFilePoolPartition.h (added)
+++ qpid/branches/linearstore/qpid/cpp/src/qpid/linearstore/jrnl/EmptyFilePoolPartition.h Fri Sep 20 16:43:10 2013
@@ -0,0 +1,72 @@
+/*
+ *
+ * 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_QLS_JRNL_EMPTYFILEPOOLPARTITION_H_
+#define QPID_QLS_JRNL_EMPTYFILEPOOLPARTITION_H_
+
+namespace qpid {
+namespace qls_jrnl {
+
+ class EmptyFilePoolPartition;
+
+}} // namespace qpid::qls_jrnl
+
+#include "qpid/linearstore/jrnl/EmptyFilePool.h"
+#include "qpid/linearstore/jrnl/EmptyFilePoolTypes.h"
+#include "qpid/linearstore/jrnl/smutex.h"
+#include <string>
+#include <map>
+#include <vector>
+
+namespace qpid {
+namespace qls_jrnl {
+
+class EmptyFilePoolPartition
+{
+public:
+ static const std::string efpTopLevelDir;
+protected:
+ typedef std::map<efpFileSizeKib_t, EmptyFilePool*> efpMap_t;
+ typedef efpMap_t::iterator efpMapItr_t;
+ typedef efpMap_t::const_iterator efpMapConstItr_t;
+
+ const efpPartitionNumber_t partitionNum;
+ const std::string partitionDir;
+ efpMap_t efpMap;
+ smutex efpMapMutex;
+
+ void validatePartitionDir();
+
+public:
+ EmptyFilePoolPartition(const efpPartitionNumber_t partitionNum_, const std::string& partitionDir_);
+ virtual ~EmptyFilePoolPartition();
+ void findEmptyFilePools();
+ efpPartitionNumber_t partitionNumber() const;
+ std::string partitionDirectory() const;
+
+ EmptyFilePool* getEmptyFilePool(const efpFileSizeKib_t efpFileSizeKb);
+ void getEmptyFilePoolSizesKb(std::vector<efpFileSizeKib_t>& efpFileSizesKbList) const;
+ void getEmptyFilePools(std::vector<EmptyFilePool*>& efpList);
+};
+
+}} // namespace qpid::qls_jrnl
+
+#endif /* QPID_QLS_JRNL_EMPTYFILEPOOLPARTITION_H_ */
Added: qpid/branches/linearstore/qpid/cpp/src/qpid/linearstore/jrnl/EmptyFilePoolTypes.h
URL: http://svn.apache.org/viewvc/qpid/branches/linearstore/qpid/cpp/src/qpid/linearstore/jrnl/EmptyFilePoolTypes.h?rev=1525050&view=auto
==============================================================================
--- qpid/branches/linearstore/qpid/cpp/src/qpid/linearstore/jrnl/EmptyFilePoolTypes.h (added)
+++ qpid/branches/linearstore/qpid/cpp/src/qpid/linearstore/jrnl/EmptyFilePoolTypes.h Fri Sep 20 16:43:10 2013
@@ -0,0 +1,37 @@
+ /*
+ *
+ * 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_QLS_JRNL_EMPTYFILEPOOLTYPES_H_
+#define QPID_QLS_JRNL_EMPTYFILEPOOLTYPES_H_
+
+#include <stdint.h>
+
+namespace qpid {
+namespace qls_jrnl {
+
+ typedef uint64_t efpFileSizeKib_t;
+ typedef uint32_t efpFileCount_t;
+ typedef uint16_t efpPartitionNumber_t;
+ typedef std::pair<efpPartitionNumber_t, efpFileSizeKib_t> efpIdentity_t;
+
+}} // namespace qpid::qls_jrnl
+
+#endif /* QPID_QLS_JRNL_EMPTYFILEPOOLTYPES_H_ */
Added: qpid/branches/linearstore/qpid/cpp/src/qpid/linearstore/jrnl/JournalFile.cpp
URL: http://svn.apache.org/viewvc/qpid/branches/linearstore/qpid/cpp/src/qpid/linearstore/jrnl/JournalFile.cpp?rev=1525050&view=auto
==============================================================================
--- qpid/branches/linearstore/qpid/cpp/src/qpid/linearstore/jrnl/JournalFile.cpp (added)
+++ qpid/branches/linearstore/qpid/cpp/src/qpid/linearstore/jrnl/JournalFile.cpp Fri Sep 20 16:43:10 2013
@@ -0,0 +1,54 @@
+/*
+ *
+ * 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/jrnl/JournalFile.h"
+
+namespace qpid {
+namespace qls_jrnl {
+
+JournalFile::JournalFile(const std::string& fqFileName_) :
+ fqfn(fqFileName_)
+{}
+
+JournalFile::~JournalFile() {}
+
+const std::string
+JournalFile::directory() const {
+ return fqfn.substr(0, fqfn.rfind('/'));
+}
+
+const std::string
+JournalFile::fileName() const {
+ return fqfn.substr(fqfn.rfind('/'));
+}
+
+const std::string
+JournalFile::fqFileName() const {
+ return fqfn;
+}
+
+bool
+JournalFile::empty() const {
+ // TODO: return true if no still-enqueued records (or parts of records) exist in this file
+ return true;
+}
+
+}} // namespace qpid::qls_jrnl
Added: qpid/branches/linearstore/qpid/cpp/src/qpid/linearstore/jrnl/JournalFile.h
URL: http://svn.apache.org/viewvc/qpid/branches/linearstore/qpid/cpp/src/qpid/linearstore/jrnl/JournalFile.h?rev=1525050&view=auto
==============================================================================
--- qpid/branches/linearstore/qpid/cpp/src/qpid/linearstore/jrnl/JournalFile.h (added)
+++ qpid/branches/linearstore/qpid/cpp/src/qpid/linearstore/jrnl/JournalFile.h Fri Sep 20 16:43:10 2013
@@ -0,0 +1,46 @@
+/*
+ *
+ * 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_JOURNALFILE_H_
+#define QPID_LINEARSTORE_JOURNALFILE_H_
+
+#include <string>
+
+namespace qpid {
+namespace qls_jrnl {
+
+class JournalFile
+{
+protected:
+ const std::string fqfn;
+public:
+ JournalFile(const std::string& fqFileName_);
+ virtual ~JournalFile();
+
+ const std::string directory() const;
+ const std::string fileName() const;
+ const std::string fqFileName() const;
+ bool empty() const;
+};
+
+}} // namespace qpid::qls_jrnl
+
+#endif // QPID_LINEARSTORE_JOURNALFILE_H_
Added: qpid/branches/linearstore/qpid/cpp/src/qpid/linearstore/jrnl/JournalFileController.cpp
URL: http://svn.apache.org/viewvc/qpid/branches/linearstore/qpid/cpp/src/qpid/linearstore/jrnl/JournalFileController.cpp?rev=1525050&view=auto
==============================================================================
--- qpid/branches/linearstore/qpid/cpp/src/qpid/linearstore/jrnl/JournalFileController.cpp (added)
+++ qpid/branches/linearstore/qpid/cpp/src/qpid/linearstore/jrnl/JournalFileController.cpp Fri Sep 20 16:43:10 2013
@@ -0,0 +1,142 @@
+/*
+ *
+ * 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/jrnl/JournalFileController.h"
+
+#include <fstream>
+#include "qpid/linearstore/jrnl/EmptyFilePool.h"
+#include "qpid/linearstore/jrnl/jcfg.h"
+#include "qpid/linearstore/jrnl/JournalFile.h"
+#include "qpid/linearstore/jrnl/slock.h"
+#include "qpid/linearstore/jrnl/utils/file_hdr.h"
+
+#include <iostream> // DEBUG
+
+namespace qpid {
+namespace qls_jrnl {
+
+JournalFileController::JournalFileController(const std::string& dir_,
+ EmptyFilePool* efpp_) :
+ dir(dir_),
+ efpp(efpp_),
+ fileSeqCounter(0)
+{
+ //std::cout << "*** JournalFileController::JournalFileController() dir=" << dir << std::endl;
+}
+
+JournalFileController::~JournalFileController() {}
+
+void
+JournalFileController::pullEmptyFileFromEfp(const uint64_t recId_,
+ const uint64_t firstRecOffs_,
+ const std::string& queueName_) {
+ std::string ef = efpp->takeEmptyFile(dir);
+ //std::cout << "*** JournalFileController::pullEmptyFileFromEfp() qn=" << queueName_ << " ef=" << ef << std::endl;
+ const JournalFile* jfp = new JournalFile(ef/*efpp->takeEmptyFile(dir)*/);
+ initialzeFileHeader(jfp->fqFileName(), recId_, firstRecOffs_, getNextFileSeqNum(), queueName_);
+ {
+ slock l(journalFileListMutex);
+ journalFileList.push_back(jfp);
+ }
+}
+
+void
+JournalFileController::purgeFilesToEfp() {
+ slock l(journalFileListMutex);
+ while (journalFileList.front()->empty()) {
+
+ efpp->returnEmptyFile(journalFileList.front());
+ delete journalFileList.front();
+ journalFileList.pop_front();
+ }
+}
+
+void
+JournalFileController::finalize() {}
+
+void
+JournalFileController::setFileSeqNum(const uint64_t fileSeqNum) {
+ fileSeqCounter = fileSeqNum;
+}
+
+// protected
+
+std::string
+JournalFileController::readFileHeader(file_hdr_t* fhdr_,
+ const std::string& fileName_) {
+ //std::cout << "*** JournalFileController::readFileHeader() fn=" << fileName_ << std::endl;
+ char buff[JRNL_SBLK_SIZE];
+ std::ifstream ifs(fileName_.c_str(), std::ifstream::in | std::ifstream::binary);
+ if (ifs.good()) {
+ ifs.read(buff, JRNL_SBLK_SIZE);
+ ifs.close();
+ std::memcpy(fhdr_, buff, sizeof(file_hdr_t));
+ return std::string(buff + sizeof(file_hdr_t), fhdr_->_queue_name_len);
+ } else {
+ std::cerr << "ERROR: Could not open file \"" << fileName_ << "\" for reading." << std::endl;
+ }
+ return std::string("");
+}
+
+void
+JournalFileController::writeFileHeader(const file_hdr_t* fhdr_,
+ const std::string& queueName_,
+ const std::string& fileName_) {
+ //std::cout << "*** JournalFileController::writeFileHeader() qn=" << queueName_ << " fn=" << fileName_ << std::endl;
+ std::fstream fs(fileName_.c_str(), std::fstream::in | std::fstream::out | std::fstream::binary);
+ if (fs.good()) {
+ fs.seekp(0);
+ fs.write((const char*)fhdr_, sizeof(file_hdr_t));
+ fs.write(queueName_.data(), fhdr_->_queue_name_len);
+ fs.close();
+ } else {
+ std::cerr << "ERROR: Could not open file \"" << fileName_ << "\" for writing." << std::endl;
+ }
+}
+
+void
+JournalFileController::resetFileHeader(const std::string& fileName_) {
+ //std::cout << "*** JournalFileController::resetFileHeader() fn=" << fileName_ << std::endl;
+ file_hdr_t fhdr;
+ readFileHeader(&fhdr, fileName_);
+ ::file_hdr_reset(&fhdr);
+ writeFileHeader(&fhdr, std::string(""), fileName_);
+}
+
+void
+JournalFileController::initialzeFileHeader(const std::string& fileName_,
+ const uint64_t recId_,
+ const uint64_t firstRecOffs_,
+ const uint64_t fileSeqNum_,
+ const std::string& queueName_) {
+ //std::cout << "*** JournalFileController::initialzeFileHeader() fn=" << fileName_ << " sn=" << fileSeqNum_ << " qn=" << queueName_ << std::endl;
+ file_hdr_t fhdr;
+ readFileHeader(&fhdr, fileName_);
+ ::file_hdr_init(&fhdr, 0, recId_, firstRecOffs_, fileSeqNum_, queueName_.length(), queueName_.data());
+ writeFileHeader(&fhdr, queueName_, fileName_);
+}
+
+uint64_t
+JournalFileController::getNextFileSeqNum() {
+ return __sync_add_and_fetch(&fileSeqCounter, 1); // GCC atomic increment, not portable
+}
+
+}} // namespace qpid::qls_jrnl
Added: qpid/branches/linearstore/qpid/cpp/src/qpid/linearstore/jrnl/JournalFileController.h
URL: http://svn.apache.org/viewvc/qpid/branches/linearstore/qpid/cpp/src/qpid/linearstore/jrnl/JournalFileController.h?rev=1525050&view=auto
==============================================================================
--- qpid/branches/linearstore/qpid/cpp/src/qpid/linearstore/jrnl/JournalFileController.h (added)
+++ qpid/branches/linearstore/qpid/cpp/src/qpid/linearstore/jrnl/JournalFileController.h Fri Sep 20 16:43:10 2013
@@ -0,0 +1,68 @@
+/*
+ *
+ * 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_JOURNALFILECONTROLLER_H_
+#define QPID_LINEARSTORE_JOURNALFILECONTROLLER_H_
+
+#include <deque>
+#include "qpid/linearstore/jrnl/smutex.h"
+
+struct file_hdr_t;
+namespace qpid {
+namespace qls_jrnl {
+class EmptyFilePool;
+class JournalFile;
+//typedef struct file_hdr_t file_hdr_t;
+
+class JournalFileController
+{
+protected:
+ typedef std::deque<const JournalFile*> JournalFileList_t;
+ typedef JournalFileList_t::iterator JournalFileListItr_t;
+
+ const std::string dir;
+ EmptyFilePool* efpp;
+ uint64_t fileSeqCounter;
+ JournalFileList_t journalFileList;
+ smutex journalFileListMutex;
+
+public:
+ JournalFileController(const std::string& dir,
+ EmptyFilePool* efpp);
+ virtual ~JournalFileController();
+
+ void pullEmptyFileFromEfp(const uint64_t recId, const uint64_t firstRecOffs, const std::string& queueName);
+ void purgeFilesToEfp();
+ void finalize();
+ void setFileSeqNum(const uint64_t fileSeqNum);
+
+protected:
+ std::string readFileHeader(file_hdr_t* fhdr, const std::string& fileName);
+ void writeFileHeader(const file_hdr_t* fhdr, const std::string& queueName, const std::string& fileName);
+ void resetFileHeader(const std::string& fileName);
+ void initialzeFileHeader(const std::string& fileName, const uint64_t recId, const uint64_t firstRecOffs,
+ const uint64_t fileSeqNum, const std::string& queueName);
+ uint64_t getNextFileSeqNum();
+};
+
+}} // namespace qpid::qls_jrnl
+
+#endif // QPID_LINEARSTORE_JOURNALFILECONTROLLER_H_
Added: qpid/branches/linearstore/qpid/cpp/src/qpid/linearstore/jrnl/JournalLog.cpp
URL: http://svn.apache.org/viewvc/qpid/branches/linearstore/qpid/cpp/src/qpid/linearstore/jrnl/JournalLog.cpp?rev=1525050&view=auto
==============================================================================
--- qpid/branches/linearstore/qpid/cpp/src/qpid/linearstore/jrnl/JournalLog.cpp (added)
+++ qpid/branches/linearstore/qpid/cpp/src/qpid/linearstore/jrnl/JournalLog.cpp Fri Sep 20 16:43:10 2013
@@ -0,0 +1,62 @@
+/*
+ *
+ * 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 "JournalLog.h"
+#include <iostream>
+
+namespace qpid {
+namespace qls_jrnl {
+
+JournalLog::JournalLog() {}
+
+JournalLog::~JournalLog() {}
+
+void
+JournalLog::log(log_level_t ll, const std::string& jid, const std::string& log_stmt) const {
+ log(ll, jid.c_str(), log_stmt.c_str());
+}
+
+void
+JournalLog::log(log_level_t ll, const char* jid, const char* const log_stmt) const {
+ if (ll > LOG_ERROR) {
+ std::cerr << log_level_str(ll) << ": Journal \"" << jid << "\": " << log_stmt << std::endl;
+ } else if (ll > LOG_INFO) {
+ std::cout << log_level_str(ll) << ": Journal \"" << jid << "\": " << log_stmt << std::endl;
+ }
+
+}
+
+const char*
+JournalLog::log_level_str(log_level_t ll) {
+ switch (ll)
+ {
+ case LOG_TRACE: return "TRACE";
+ case LOG_DEBUG: return "DEBUG";
+ case LOG_INFO: return "INFO";
+ case LOG_NOTICE: return "NOTICE";
+ case LOG_WARN: return "WARN";
+ case LOG_ERROR: return "ERROR";
+ case LOG_CRITICAL: return "CRITICAL";
+ }
+ return "<log level unknown>";
+}
+
+}} // namespace qpid::qls_jrnl
Added: qpid/branches/linearstore/qpid/cpp/src/qpid/linearstore/jrnl/JournalLog.h
URL: http://svn.apache.org/viewvc/qpid/branches/linearstore/qpid/cpp/src/qpid/linearstore/jrnl/JournalLog.h?rev=1525050&view=auto
==============================================================================
--- qpid/branches/linearstore/qpid/cpp/src/qpid/linearstore/jrnl/JournalLog.h (added)
+++ qpid/branches/linearstore/qpid/cpp/src/qpid/linearstore/jrnl/JournalLog.h Fri Sep 20 16:43:10 2013
@@ -0,0 +1,56 @@
+/*
+ *
+ * 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_JOURNALLOG_H_
+#define QPID_LINEARSTORE_JOURNALLOG_H_
+
+#include <string>
+
+namespace qpid {
+namespace qls_jrnl {
+
+class JournalLog
+{
+public:
+ typedef enum _log_level
+ {
+ LOG_TRACE = 0,
+ LOG_DEBUG,
+ LOG_INFO,
+ LOG_NOTICE,
+ LOG_WARN,
+ LOG_ERROR,
+ LOG_CRITICAL
+ } log_level_t;
+
+protected:
+ JournalLog();
+ virtual ~JournalLog();
+
+public:
+ virtual void log(log_level_t level, const std::string& jid, const std::string& log_stmt) const;
+ virtual void log(log_level_t level, const char* jid, const char* const log_stmt) const;
+ static const char* log_level_str(log_level_t ll);
+};
+
+}} // namespace qpid::qls_jrnl
+
+#endif // QPID_LINEARSTORE_JOURNALLOG_H_
Modified: qpid/branches/linearstore/qpid/cpp/src/qpid/linearstore/jrnl/enums.h
URL: http://svn.apache.org/viewvc/qpid/branches/linearstore/qpid/cpp/src/qpid/linearstore/jrnl/enums.h?rev=1525050&r1=1525049&r2=1525050&view=diff
==============================================================================
--- qpid/branches/linearstore/qpid/cpp/src/qpid/linearstore/jrnl/enums.h (original)
+++ qpid/branches/linearstore/qpid/cpp/src/qpid/linearstore/jrnl/enums.h Fri Sep 20 16:43:10 2013
@@ -19,8 +19,8 @@
*
*/
-#ifndef QPID_LEGACYSTORE_JRNL_ENUMS_H
-#define QPID_LEGACYSTORE_JRNL_ENUMS_H
+#ifndef QPID_LINEARSTORE_JRNL_ENUMS_H
+#define QPID_LINEARSTORE_JRNL_ENUMS_H
namespace qpid
{
@@ -64,6 +64,7 @@ namespace qls_jrnl
return "<iores unknown>";
}
+/*
enum _log_level
{
LOG_TRACE = 0,
@@ -74,9 +75,9 @@ namespace qls_jrnl
LOG_ERROR,
LOG_CRITICAL
};
- typedef _log_level log_level;
+ typedef _log_level log_level_t;
- static inline const char* log_level_str(log_level ll)
+ static inline const char* log_level_str(log_level_t ll)
{
switch (ll)
{
@@ -90,8 +91,9 @@ namespace qls_jrnl
}
return "<log level unknown>";
}
+*/
}}
-#endif // ifndef QPID_LEGACYSTORE_JRNL_ENUMS_H
+#endif // ifndef QPID_LINEARSTORE_JRNL_ENUMS_H
Modified: qpid/branches/linearstore/qpid/cpp/src/qpid/linearstore/jrnl/jcfg.h
URL: http://svn.apache.org/viewvc/qpid/branches/linearstore/qpid/cpp/src/qpid/linearstore/jrnl/jcfg.h?rev=1525050&r1=1525049&r2=1525050&view=diff
==============================================================================
--- qpid/branches/linearstore/qpid/cpp/src/qpid/linearstore/jrnl/jcfg.h (original)
+++ qpid/branches/linearstore/qpid/cpp/src/qpid/linearstore/jrnl/jcfg.h Fri Sep 20 16:43:10 2013
@@ -40,7 +40,6 @@
#endif
*/
-
/**
* <b>Rule:</b> Data block size (JRNL_DBLK_SIZE) MUST be a power of 2 such that
* <pre>
@@ -51,6 +50,7 @@
#define JRNL_DBLK_SIZE 128 /**< Data block size in bytes (CANNOT BE LESS THAN 32!) */
#define JRNL_SBLK_SIZE_DBLKS 32 /**< Disk softblock size in multiples of JRNL_DBLK_SIZE */
#define JRNL_SBLK_SIZE JRNL_SBLK_SIZE_DBLKS * JRNL_DBLK_SIZE /**< Disk softblock size in bytes */
+#define JRNL_SBLK_SIZE_KIB JRNL_SBLK_SIZE / 1024 /**< Disk softblock size in KiB */
//#define JRNL_MIN_FILE_SIZE 128 ///< Min. jrnl file size in sblks (excl. file_hdr)
//#define JRNL_MAX_FILE_SIZE 4194176 ///< Max. jrnl file size in sblks (excl. file_hdr)
//#define JRNL_MIN_NUM_FILES 4 ///< Min. number of journal files
@@ -68,7 +68,7 @@
//
//#define JRNL_INFO_EXTENSION "jinf" ///< Extension for journal info files
//#define JRNL_DATA_EXTENSION "jdat" ///< Extension for journal data files
-#define QLS_JRNL_FILE_EXTENSION "jdat" /**< Extension for journal data files */
+#define QLS_JRNL_FILE_EXTENSION ".jrnl" /**< Extension for journal data files */
//#define RHM_JDAT_TXA_MAGIC 0x614d4852 ///< ("RHMa" in little endian) Magic for dtx abort hdrs
#define QLS_TXA_MAGIC 0x61534c51 /**< ("RHMa" in little endian) Magic for dtx abort hdrs */
//#define RHM_JDAT_TXC_MAGIC 0x634d4852 ///< ("RHMc" in little endian) Magic for dtx commit hdrs
@@ -83,6 +83,7 @@
#define QLS_EMPTY_MAGIC 0x78534c51 /**< ("QLSx" in little endian) Magic for empty dblk */
//#define RHM_JDAT_VERSION 0x01 ///< Version (of file layout)
#define QLS_JRNL_VERSION 0x0002 /**< Version (of file layout) */
+#define QLS_JRNL_FHDRSIZESBLKS 0x0001 /**< Journal file header size in sblks (as defined by JRNL_SBLK_SIZE) */
//#define RHM_CLEAN_CHAR 0xff ///< Char used to clear empty space on disk
#define QLS_CLEAN_CHAR 0xff ///< Char used to clear empty space on disk
//
Modified: qpid/branches/linearstore/qpid/cpp/src/qpid/linearstore/jrnl/jcntl.cpp
URL: http://svn.apache.org/viewvc/qpid/branches/linearstore/qpid/cpp/src/qpid/linearstore/jrnl/jcntl.cpp?rev=1525050&r1=1525049&r2=1525050&view=diff
==============================================================================
--- qpid/branches/linearstore/qpid/cpp/src/qpid/linearstore/jrnl/jcntl.cpp (original)
+++ qpid/branches/linearstore/qpid/cpp/src/qpid/linearstore/jrnl/jcntl.cpp Fri Sep 20 16:43:10 2013
@@ -29,9 +29,11 @@
#include <fstream>
#include <iomanip>
#include <iostream>
+#include <qpid/linearstore/jrnl/EmptyFilePool.h>
//#include "qpid/linearstore/jrnl/file_hdr.h"
#include "qpid/linearstore/jrnl/jerrno.h"
//#include "qpid/linearstore/jrnl/jinf.h"
+#include "qpid/linearstore/jrnl/JournalFileController.h"
#include <limits>
#include <sstream>
#include <unistd.h>
@@ -62,15 +64,16 @@ bool jcntl::init_statics()
// Functions
-jcntl::jcntl(const std::string& jid, const std::string& jdir, const std::string& base_filename):
+jcntl::jcntl(const std::string& jid, const std::string& jdir/*, const std::string& base_filename*/):
_jid(jid),
- _jdir(jdir, base_filename),
- _base_filename(base_filename),
+ _jdir(jdir/*, base_filename*/),
+// _base_filename(base_filename),
_init_flag(false),
_stop_flag(false),
_readonly_flag(false),
- _autostop(true),
- _jfsize_sblks(0),
+// _autostop(true),
+ _jfcp(0),
+// _jfsize_sblks(0),
// _lpmgr(),
_emap(),
_tmap(),
@@ -87,11 +90,16 @@ jcntl::~jcntl()
try { stop(true); }
catch (const jexception& e) { std::cerr << e << std::endl; }
// _lpmgr.finalize();
+ if (_jfcp) {
+ _jfcp->finalize();
+ delete _jfcp;
+ _jfcp = 0;
+ }
}
void
jcntl::initialize(/*const uint16_t num_jfiles, const bool ae, const uint16_t ae_max_jfiles,
- const uint32_t jfsize_sblks,*/ const uint16_t wcache_num_pages, const uint32_t wcache_pgsize_sblks,
+ const uint32_t jfsize_sblks,*/ EmptyFilePool* efpp, const uint16_t wcache_num_pages, const uint32_t wcache_pgsize_sblks,
aio_callback* const cbp)
{
_init_flag = false;
@@ -101,6 +109,12 @@ jcntl::initialize(/*const uint16_t num_j
_emap.clear();
_tmap.clear();
+ if (_jfcp) {
+ _jfcp->finalize();
+ delete _jfcp;
+ _jfcp = 0;
+ }
+
// _lpmgr.finalize();
// Set new file geometry parameters
@@ -117,6 +131,8 @@ jcntl::initialize(/*const uint16_t num_j
_jdir.clear_dir();
// _lpmgr.initialize(num_jfiles, ae, ae_max_jfiles, this, &new_fcntl);
+ _jfcp = new JournalFileController(_jdir.dirname(), efpp);
+ _jfcp->pullEmptyFileFromEfp(1, 4096, _jid);
// _wrfc.initialize(_jfsize_sblks);
// _rrfc.initialize();
// _rrfc.set_findex(0);
@@ -159,7 +175,7 @@ jcntl::recover(/*const uint16_t num_jfil
highest_rid = _rcvdat._h_rid;
if (_rcvdat._jfull)
throw jexception(jerrno::JERR_JCNTL_RECOVERJFULL, "jcntl", "recover");
- this->log(LOG_DEBUG, _rcvdat.to_log(_jid));
+ this->log(LOG_DEBUG, _jid, _rcvdat.to_log(_jid));
// _lpmgr.recover(_rcvdat, this, &new_fcntl);
@@ -192,6 +208,7 @@ void
jcntl::delete_jrnl_files()
{
stop(true); // wait for AIO to complete
+ _jfcp->purgeFilesToEfp();
_jdir.delete_dir();
}
@@ -407,20 +424,22 @@ jcntl::flush(const bool block_till_aio_c
return res;
}
+/*
void
-jcntl::log(log_level ll, const std::string& log_stmt) const
+jcntl::log(log_level_t ll, const std::string& log_stmt) const
{
log(ll, log_stmt.c_str());
}
void
-jcntl::log(log_level ll, const char* const log_stmt) const
+jcntl::log(log_level_t ll, const char* const log_stmt) const
{
if (ll > LOG_INFO)
{
std::cout << log_level_str(ll) << ": Journal \"" << _jid << "\": " << log_stmt << std::endl;
}
}
+*/
/*
void
@@ -523,7 +542,7 @@ jcntl::handle_aio_wait(const iores res,
{
std::ostringstream oss;
oss << "get_events() returned JERR_JCNTL_AIOCMPLWAIT; wmgr_status: " << _wmgr.status_str();
- this->log(LOG_CRITICAL, oss.str());
+ this->log(LOG_CRITICAL, _jid, oss.str());
throw jexception(jerrno::JERR_JCNTL_AIOCMPLWAIT, "jcntl", "handle_aio_wait");
}
}
@@ -877,8 +896,8 @@ jcntl::jfile_cycle(uint16_t& fid, std::i
if (!ifsp->is_open())
{
std::ostringstream oss;
- oss << _jdir.dirname() << "/" << _base_filename << ".";
- oss << std::hex << std::setfill('0') << std::setw(4) << fid << "." << QLS_JRNL_FILE_EXTENSION;
+ oss << _jdir.dirname() << "/" /*<< _base_filename*/ << "."; // TODO - linear journal name
+ oss << std::hex << std::setfill('0') << std::setw(4) << fid << QLS_JRNL_FILE_EXTENSION;
ifsp->clear(); // clear eof flag, req'd for older versions of c++
ifsp->open(oss.str().c_str(), std::ios_base::in | std::ios_base::binary);
if (!ifsp->good())
@@ -946,12 +965,12 @@ jcntl::check_journal_alignment(const uin
oss << std::hex << "Bad record alignment found at fid=0x" << fid;
oss << " offs=0x" << file_pos << " (likely journal overwrite boundary); " << std::dec;
oss << (JRNL_SBLK_SIZE_DBLKS - (sblk_offs/JRNL_DBLK_SIZE)) << " filler record(s) required.";
- this->log(LOG_WARN, oss.str());
+ this->log(LOG_WARN, _jid, oss.str());
}
const uint32_t xmagic = QLS_EMPTY_MAGIC;
std::ostringstream oss;
- oss << _jdir.dirname() << "/" << _base_filename << ".";
- oss << std::hex << std::setfill('0') << std::setw(4) << fid << "." << QLS_JRNL_FILE_EXTENSION;
+ oss << _jdir.dirname() << "/" /*<< _base_filename*/ << "."; // TODO linear journal name
+ oss << std::hex << std::setfill('0') << std::setw(4) << fid << QLS_JRNL_FILE_EXTENSION;
std::ofstream ofsp(oss.str().c_str(),
std::ios_base::in | std::ios_base::out | std::ios_base::binary);
if (!ofsp.good())
@@ -971,7 +990,7 @@ jcntl::check_journal_alignment(const uin
assert(!ofsp.fail());
std::ostringstream oss;
oss << std::hex << "Recover phase write: Wrote filler record: fid=0x" << fid << " offs=0x" << file_pos;
- this->log(LOG_NOTICE, oss.str());
+ this->log(LOG_NOTICE, _jid, oss.str());
file_pos = ofsp.tellp();
}
ofsp.close();
@@ -979,7 +998,7 @@ jcntl::check_journal_alignment(const uin
rd._lfid = fid;
// if (!rd._frot)
// rd._ffid = (fid + 1) % rd._njf;
- this->log(LOG_INFO, "Bad record alignment fixed.");
+ this->log(LOG_INFO, _jid, "Bad record alignment fixed.");
}
rd._eo = file_pos;
}
Modified: qpid/branches/linearstore/qpid/cpp/src/qpid/linearstore/jrnl/jcntl.h
URL: http://svn.apache.org/viewvc/qpid/branches/linearstore/qpid/cpp/src/qpid/linearstore/jrnl/jcntl.h?rev=1525050&r1=1525049&r2=1525050&view=diff
==============================================================================
--- qpid/branches/linearstore/qpid/cpp/src/qpid/linearstore/jrnl/jcntl.h (original)
+++ qpid/branches/linearstore/qpid/cpp/src/qpid/linearstore/jrnl/jcntl.h Fri Sep 20 16:43:10 2013
@@ -31,6 +31,7 @@ namespace qls_jrnl
#include <cstddef>
#include <deque>
+#include <qpid/linearstore/jrnl/JournalLog.h>
#include "qpid/linearstore/jrnl/jdir.h"
//#include "qpid/linearstore/jrnl/fcntl.h"
//#include "qpid/linearstore/jrnl/lpmgr.h"
@@ -45,6 +46,8 @@ namespace qpid
{
namespace qls_jrnl
{
+class EmptyFilePool;
+class JournalFileController;
/**
* \brief Access and control interface for the journal. This is the top-level class for the
@@ -56,7 +59,7 @@ namespace qls_jrnl
* which is used per data block written to the journal, and is used to track its status through
* the AIO enqueue, read and dequeue process.
*/
- class jcntl
+ class jcntl : public JournalLog
{
protected:
/**
@@ -85,7 +88,7 @@ namespace qls_jrnl
* that will be written to disk. No file separator characters should be included here, but
* all other legal filename characters are valid.
*/
- std::string _base_filename;
+// std::string _base_filename;
/**
* \brief Initialized flag
@@ -121,10 +124,11 @@ namespace qls_jrnl
* marker. If not set, then attempts to write will throw exceptions until the journal
* file low water marker moves to the next journal file.
*/
- bool _autostop; ///< Autostop flag - stops journal when overrun occurs
+ //bool _autostop; ///< Autostop flag - stops journal when overrun occurs
// Journal control structures
- uint32_t _jfsize_sblks; ///< Journal file size in sblks
+ JournalFileController* _jfcp;///< Journal File Controller
+ //uint32_t _jfsize_sblks; ///< Journal file size in sblks
//lpmgr _lpmgr; ///< LFID-PFID manager tracks inserted journal files
enq_map _emap; ///< Enqueue map for low water mark management
txn_map _tmap; ///< Transaction map open transactions
@@ -148,7 +152,7 @@ namespace qls_jrnl
* \param jdir The directory which will contain the journal files.
* \param base_filename The string which will be used to start all journal filenames.
*/
- jcntl(const std::string& jid, const std::string& jdir, const std::string& base_filename);
+ jcntl(const std::string& jid, const std::string& jdir/*, const std::string& base_filename*/);
/**
* \brief Destructor.
@@ -190,7 +194,7 @@ namespace qls_jrnl
* \exception TODO
*/
void initialize(/*const uint16_t num_jfiles, const bool auto_expand, const uint16_t ae_max_jfiles,
- const uint32_t jfsize_sblks,*/ const uint16_t wcache_num_pages, const uint32_t wcache_pgsize_sblks,
+ const uint32_t jfsize_sblks,*/EmptyFilePool* efpp, const uint16_t wcache_num_pages, const uint32_t wcache_pgsize_sblks,
aio_callback* const cbp);
/**
@@ -628,17 +632,17 @@ namespace qls_jrnl
* the same directory, their base filenames <b>MUST</b> be different or else the instances
* will overwrite one another.
*/
- inline const std::string& base_filename() const { return _base_filename; }
+// inline const std::string& base_filename() const { return _base_filename; }
// inline uint16_t num_jfiles() const; { return _lpmgr.num_jfiles(); }
// inline fcntl* get_fcntlp(const uint16_t lfid) const { return _lpmgr.get_fcntlp(lfid); }
- inline uint32_t jfsize_sblks() const { return _jfsize_sblks; }
+// inline uint32_t jfsize_sblks() const { return _jfsize_sblks; }
// Logging
- virtual void log(log_level level, const std::string& log_stmt) const;
- virtual void log(log_level level, const char* const log_stmt) const;
+// virtual void log(log_level_t level, const std::string& log_stmt) const;
+// virtual void log(log_level_t level, const char* const log_stmt) const;
// FIXME these are _rmgr to _wmgr interactions, remove when _rmgr contains ref to _wmgr:
//void chk_wr_frot();
Modified: qpid/branches/linearstore/qpid/cpp/src/qpid/linearstore/jrnl/jdir.cpp
URL: http://svn.apache.org/viewvc/qpid/branches/linearstore/qpid/cpp/src/qpid/linearstore/jrnl/jdir.cpp?rev=1525050&r1=1525049&r2=1525050&view=diff
==============================================================================
--- qpid/branches/linearstore/qpid/cpp/src/qpid/linearstore/jrnl/jdir.cpp (original)
+++ qpid/branches/linearstore/qpid/cpp/src/qpid/linearstore/jrnl/jdir.cpp Fri Sep 20 16:43:10 2013
@@ -37,9 +37,9 @@ namespace qpid
namespace qls_jrnl
{
-jdir::jdir(const std::string& dirname, const std::string& _base_filename):
- _dirname(dirname),
- _base_filename(_base_filename)
+jdir::jdir(const std::string& dirname/*, const std::string& _base_filename*/):
+ _dirname(dirname)/*,
+ _base_filename(_base_filename)*/
{}
jdir::~jdir()
@@ -88,21 +88,22 @@ jdir::create_dir(const std::string& dirn
void
jdir::clear_dir(const bool create_flag)
{
- clear_dir(_dirname, _base_filename, create_flag);
+ clear_dir(_dirname/*, _base_filename*/, create_flag);
}
void
-jdir::clear_dir(const char* dirname, const char* base_filename, const bool create_flag)
+jdir::clear_dir(const char* dirname/*, const char* base_filename*/, const bool create_flag)
{
- clear_dir(std::string(dirname), std::string(base_filename), create_flag);
+ clear_dir(std::string(dirname)/*, std::string(base_filename)*/, create_flag);
}
void
-jdir::clear_dir(const std::string& dirname, const std::string&
+jdir::clear_dir(const std::string& dirname/*, const std::string&
#ifndef RHM_JOWRITE
base_filename
#endif
+*/
, const bool create_flag)
{
DIR* dir = ::opendir(dirname.c_str());
@@ -117,7 +118,7 @@ jdir::clear_dir(const std::string& dirna
oss << "dir=\"" << dirname << "\"" << FORMAT_SYSERR(errno);
throw jexception(jerrno::JERR_JDIR_OPENDIR, oss.str(), "jdir", "clear_dir");
}
-#ifndef RHM_JOWRITE
+//#ifndef RHM_JOWRITE
struct dirent* entry;
bool found = false;
std::string bak_dir;
@@ -126,13 +127,13 @@ jdir::clear_dir(const std::string& dirna
// Ignore . and ..
if (std::strcmp(entry->d_name, ".") != 0 && std::strcmp(entry->d_name, "..") != 0)
{
- if (std::strlen(entry->d_name) > base_filename.size())
+ if (std::strlen(entry->d_name) >= 3) // 'bak'
{
- if (std::strncmp(entry->d_name, base_filename.c_str(), base_filename.size()) == 0)
+ if (std::strncmp(entry->d_name, "bak", 3) == 0)
{
if (!found)
{
- bak_dir = create_bak_dir(dirname, base_filename);
+ bak_dir = create_bak_dir(dirname/*, base_filename*/);
found = true;
}
std::ostringstream oldname;
@@ -154,16 +155,16 @@ jdir::clear_dir(const std::string& dirna
// FIXME: Find out why this fails with false alarms/errors from time to time...
// While commented out, there is no error capture from reading dir entries.
// check_err(errno, dir, dirname, "clear_dir");
-#endif
+//#endif
close_dir(dir, dirname, "clear_dir");
}
// === push_down ===
std::string
-jdir::push_down(const std::string& dirname, const std::string& target_dir, const std::string& bak_dir_base)
+jdir::push_down(const std::string& dirname, const std::string& target_dir/*, const std::string& bak_dir_base*/)
{
- std::string bak_dir_name = create_bak_dir(dirname, bak_dir_base);
+ std::string bak_dir_name = create_bak_dir(dirname/*, bak_dir_base*/);
DIR* dir = ::opendir(dirname.c_str());
if (!dir)
@@ -202,18 +203,18 @@ jdir::push_down(const std::string& dirna
void
jdir::verify_dir()
{
- verify_dir(_dirname, _base_filename);
+ verify_dir(_dirname/*, _base_filename*/);
}
void
-jdir::verify_dir(const char* dirname, const char* base_filename)
+jdir::verify_dir(const char* dirname/*, const char* base_filename*/)
{
- verify_dir(std::string(dirname), std::string(base_filename));
+ verify_dir(std::string(dirname)/*, std::string(base_filename)*/);
}
void
-jdir::verify_dir(const std::string& dirname, const std::string& /*base_filename*/)
+jdir::verify_dir(const std::string& dirname/*, const std::string& base_filename*/)
{
if (!is_dir(dirname))
{
@@ -323,7 +324,7 @@ jdir::delete_dir(const std::string& dirn
std::string
-jdir::create_bak_dir(const std::string& dirname, const std::string& base_filename)
+jdir::create_bak_dir(const std::string& dirname)
{
DIR* dir = ::opendir(dirname.c_str());
long dir_num = 0L;
@@ -339,13 +340,11 @@ jdir::create_bak_dir(const std::string&
// Ignore . and ..
if (std::strcmp(entry->d_name, ".") != 0 && std::strcmp(entry->d_name, "..") != 0)
{
- if (std::strlen(entry->d_name) == base_filename.size() + 10) // Format: basename.bak.XXXX
+ if (std::strlen(entry->d_name) == 9) // Format: _bak.XXXX
{
- std::ostringstream oss;
- oss << "_" << base_filename << ".bak.";
- if (std::strncmp(entry->d_name, oss.str().c_str(), base_filename.size() + 6) == 0)
+ if (std::strncmp(entry->d_name, "_bak.", 5) == 0)
{
- long this_dir_num = std::strtol(entry->d_name + base_filename.size() + 6, 0, 16);
+ long this_dir_num = std::strtol(entry->d_name + 5, 0, 16);
if (this_dir_num > dir_num)
dir_num = this_dir_num;
}
@@ -358,8 +357,7 @@ jdir::create_bak_dir(const std::string&
close_dir(dir, dirname, "create_bak_dir");
std::ostringstream dn;
- dn << dirname << "/_" << base_filename << ".bak." << std::hex << std::setw(4) <<
- std::setfill('0') << ++dir_num;
+ dn << dirname << "/_bak." << std::hex << std::setw(4) << std::setfill('0') << ++dir_num;
if (::mkdir(dn.str().c_str(), S_IRWXU | S_IRWXG | S_IROTH))
{
std::ostringstream oss;
@@ -411,6 +409,32 @@ jdir::exists(const std::string& name)
}
void
+jdir::read_dir(const std::string& name, std::vector<std::string>& dir_list, const bool incl_dirs, const bool incl_files, const bool incl_links) {
+ struct stat s;
+ if (is_dir(name)) {
+ DIR* dir = ::opendir(name.c_str());
+ if (dir != 0) {
+ struct dirent* entry;
+ while ((entry = ::readdir(dir)) != 0) {
+ if (std::strcmp(entry->d_name, ".") != 0 && std::strcmp(entry->d_name, "..") != 0) { // Ignore . and ..
+ std::string full_name(name + "/" + entry->d_name);
+ if (::stat(full_name.c_str(), &s))
+ {
+ ::closedir(dir);
+ std::ostringstream oss;
+ oss << "stat: file=\"" << full_name << "\"" << FORMAT_SYSERR(errno);
+ throw jexception(jerrno::JERR_JDIR_STAT, oss.str(), "jdir", "delete_dir");
+ }
+ if ((S_ISREG(s.st_mode) && incl_files) || (S_ISDIR(s.st_mode) && incl_dirs) || (S_ISLNK(s.st_mode) && incl_links))
+ dir_list.push_back(entry->d_name);
+ }
+ }
+ }
+ close_dir(dir, name, "read_dir");
+ }
+}
+
+void
jdir::check_err(const int err_num, DIR* dir, const std::string& dir_name, const std::string& fn_name)
{
if (err_num)
Modified: qpid/branches/linearstore/qpid/cpp/src/qpid/linearstore/jrnl/jdir.h
URL: http://svn.apache.org/viewvc/qpid/branches/linearstore/qpid/cpp/src/qpid/linearstore/jrnl/jdir.h?rev=1525050&r1=1525049&r2=1525050&view=diff
==============================================================================
--- qpid/branches/linearstore/qpid/cpp/src/qpid/linearstore/jrnl/jdir.h (original)
+++ qpid/branches/linearstore/qpid/cpp/src/qpid/linearstore/jrnl/jdir.h Fri Sep 20 16:43:10 2013
@@ -19,8 +19,8 @@
*
*/
-#ifndef QPID_LEGACYSTORE_JRNL_JDIR_H
-#define QPID_LEGACYSTORE_JRNL_JDIR_H
+#ifndef QPID_LINEARSTORE_JRNL_JDIR_H
+#define QPID_LINEARSTORE_JRNL_JDIR_H
namespace qpid
{
@@ -32,6 +32,7 @@ class jdir;
//#include "qpid/linearstore/jrnl/jinf.h"
#include <dirent.h>
#include <string>
+#include <vector>
namespace qpid
{
@@ -46,7 +47,7 @@ namespace qls_jrnl
{
private:
std::string _dirname;
- std::string _base_filename;
+ //std::string _base_filename;
public:
@@ -57,7 +58,7 @@ namespace qls_jrnl
* \param base_filename Filename root used in the creation of %journal files
* and sub-directories.
*/
- jdir(const std::string& dirname, const std::string& base_filename);
+ jdir(const std::string& dirname/*, const std::string& base_filename*/);
virtual ~jdir();
@@ -119,7 +120,7 @@ namespace qls_jrnl
* directory failed.
* \exception jerrno::JERR_JDIR_CLOSEDIR The directory handle could not be closed.
*/
- static void clear_dir(const char* dirname, const char* base_filename,
+ static void clear_dir(const char* dirname/*, const char* base_filename*/,
const bool create_flag = true);
/**
@@ -137,7 +138,7 @@ namespace qls_jrnl
* directory failed.
* \exception jerrno::JERR_JDIR_CLOSEDIR The directory handle could not be closed.
*/
- static void clear_dir(const std::string& dirname, const std::string& base_filename,
+ static void clear_dir(const std::string& dirname/*, const std::string& base_filename*/,
const bool create_flag = true);
@@ -152,7 +153,7 @@ namespace qls_jrnl
* \param bak_dir_base Base name for backup directory to be created in dirname, into which target_dir will be moved.
* \return Name of backup dir into which target_dir was pushed.
*/
- static std::string push_down(const std::string& dirname, const std::string& target_dir, const std::string& bak_dir_base);
+ static std::string push_down(const std::string& dirname, const std::string& target_dir/*, const std::string& bak_dir_base*/);
/**
@@ -184,7 +185,7 @@ namespace qls_jrnl
* \exception jerrno::JERR_JINF_CVALIDFAIL Error validating %jinf file
* \exception jerrno::JERR_JDIR_NOSUCHFILE Expected jdat file is missing
*/
- static void verify_dir(const char* dirname, const char* base_filename);
+ static void verify_dir(const char* dirname/*, const char* base_filename*/);
/**
* \brief Verify that dirname is a valid %journal directory.
@@ -201,7 +202,7 @@ namespace qls_jrnl
* \exception jerrno::JERR_JINF_CVALIDFAIL Error validating %jinf file
* \exception jerrno::JERR_JDIR_NOSUCHFILE Expected jdat file is missing
*/
- static void verify_dir(const std::string& dirname, const std::string& base_filename);
+ static void verify_dir(const std::string& dirname/*, const std::string& base_filename*/);
/**
* \brief Delete the %journal directory and all files and sub--directories that it may
@@ -273,8 +274,8 @@ namespace qls_jrnl
* \exception jerrno::JERR_JDIR_CLOSEDIR The directory handle could not be closed.
* \exception jerrno::JERR_JDIR_MKDIR The backup directory could not be deleted.
*/
- static std::string create_bak_dir(const std::string& dirname,
- const std::string& base_filename);
+ static std::string create_bak_dir(const std::string& dirname/*,
+ const std::string& base_filename*/);
/**
* \brief Return the directory name as a string.
@@ -284,7 +285,7 @@ namespace qls_jrnl
/**
* \brief Return the %journal base filename name as a string.
*/
- inline const std::string& base_filename() const { return _base_filename; }
+// inline const std::string& base_filename() const { return _base_filename; }
/**
* \brief Test whether the named file is a directory.
@@ -335,6 +336,8 @@ namespace qls_jrnl
*/
static bool exists(const std::string& name);
+ static void read_dir(const std::string& name, std::vector<std::string>& dir_list, const bool incl_dirs, const bool incl_files, const bool incl_links);
+
/**
* \brief Stream operator
*/
@@ -363,4 +366,4 @@ namespace qls_jrnl
}}
-#endif // ifndef QPID_LEGACYSTORE_JRNL_JDIR_H
+#endif // ifndef QPID_LINEARSTORE_JRNL_JDIR_H
Modified: qpid/branches/linearstore/qpid/cpp/src/qpid/linearstore/jrnl/jerrno.cpp
URL: http://svn.apache.org/viewvc/qpid/branches/linearstore/qpid/cpp/src/qpid/linearstore/jrnl/jerrno.cpp?rev=1525050&r1=1525049&r2=1525050&view=diff
==============================================================================
--- qpid/branches/linearstore/qpid/cpp/src/qpid/linearstore/jrnl/jerrno.cpp (original)
+++ qpid/branches/linearstore/qpid/cpp/src/qpid/linearstore/jrnl/jerrno.cpp Fri Sep 20 16:43:10 2013
@@ -31,39 +31,39 @@ std::map<uint32_t, const char*>::iterato
bool jerrno::_initialized = jerrno::__init();
// generic errors
-const uint32_t jerrno::JERR__MALLOC = 0x0100;
-const uint32_t jerrno::JERR__UNDERFLOW = 0x0101;
-const uint32_t jerrno::JERR__NINIT = 0x0102;
-const uint32_t jerrno::JERR__AIO = 0x0103;
-const uint32_t jerrno::JERR__FILEIO = 0x0104;
-const uint32_t jerrno::JERR__RTCLOCK = 0x0105;
-const uint32_t jerrno::JERR__PTHREAD = 0x0106;
-const uint32_t jerrno::JERR__TIMEOUT = 0x0107;
-const uint32_t jerrno::JERR__UNEXPRESPONSE = 0x0108;
-const uint32_t jerrno::JERR__RECNFOUND = 0x0109;
-const uint32_t jerrno::JERR__NOTIMPL = 0x010a;
+const uint32_t jerrno::JERR__MALLOC = 0x0100;
+const uint32_t jerrno::JERR__UNDERFLOW = 0x0101;
+const uint32_t jerrno::JERR__NINIT = 0x0102;
+const uint32_t jerrno::JERR__AIO = 0x0103;
+const uint32_t jerrno::JERR__FILEIO = 0x0104;
+const uint32_t jerrno::JERR__RTCLOCK = 0x0105;
+const uint32_t jerrno::JERR__PTHREAD = 0x0106;
+const uint32_t jerrno::JERR__TIMEOUT = 0x0107;
+const uint32_t jerrno::JERR__UNEXPRESPONSE = 0x0108;
+const uint32_t jerrno::JERR__RECNFOUND = 0x0109;
+const uint32_t jerrno::JERR__NOTIMPL = 0x010a;
// class jcntl
-const uint32_t jerrno::JERR_JCNTL_STOPPED = 0x0200;
-const uint32_t jerrno::JERR_JCNTL_READONLY = 0x0201;
-const uint32_t jerrno::JERR_JCNTL_AIOCMPLWAIT = 0x0202;
-const uint32_t jerrno::JERR_JCNTL_UNKNOWNMAGIC = 0x0203;
-const uint32_t jerrno::JERR_JCNTL_NOTRECOVERED = 0x0204;
-const uint32_t jerrno::JERR_JCNTL_RECOVERJFULL = 0x0205;
-const uint32_t jerrno::JERR_JCNTL_OWIMISMATCH = 0x0206;
+const uint32_t jerrno::JERR_JCNTL_STOPPED = 0x0200;
+const uint32_t jerrno::JERR_JCNTL_READONLY = 0x0201;
+const uint32_t jerrno::JERR_JCNTL_AIOCMPLWAIT = 0x0202;
+const uint32_t jerrno::JERR_JCNTL_UNKNOWNMAGIC = 0x0203;
+const uint32_t jerrno::JERR_JCNTL_NOTRECOVERED = 0x0204;
+const uint32_t jerrno::JERR_JCNTL_RECOVERJFULL = 0x0205;
+const uint32_t jerrno::JERR_JCNTL_OWIMISMATCH = 0x0206;
// class jdir
-const uint32_t jerrno::JERR_JDIR_NOTDIR = 0x0300;
-const uint32_t jerrno::JERR_JDIR_MKDIR = 0x0301;
-const uint32_t jerrno::JERR_JDIR_OPENDIR = 0x0302;
-const uint32_t jerrno::JERR_JDIR_READDIR = 0x0303;
-const uint32_t jerrno::JERR_JDIR_CLOSEDIR = 0x0304;
-const uint32_t jerrno::JERR_JDIR_RMDIR = 0x0305;
-const uint32_t jerrno::JERR_JDIR_NOSUCHFILE = 0x0306;
-const uint32_t jerrno::JERR_JDIR_FMOVE = 0x0307;
-const uint32_t jerrno::JERR_JDIR_STAT = 0x0308;
-const uint32_t jerrno::JERR_JDIR_UNLINK = 0x0309;
-const uint32_t jerrno::JERR_JDIR_BADFTYPE = 0x030a;
+const uint32_t jerrno::JERR_JDIR_NOTDIR = 0x0300;
+const uint32_t jerrno::JERR_JDIR_MKDIR = 0x0301;
+const uint32_t jerrno::JERR_JDIR_OPENDIR = 0x0302;
+const uint32_t jerrno::JERR_JDIR_READDIR = 0x0303;
+const uint32_t jerrno::JERR_JDIR_CLOSEDIR = 0x0304;
+const uint32_t jerrno::JERR_JDIR_RMDIR = 0x0305;
+const uint32_t jerrno::JERR_JDIR_NOSUCHFILE = 0x0306;
+const uint32_t jerrno::JERR_JDIR_FMOVE = 0x0307;
+const uint32_t jerrno::JERR_JDIR_STAT = 0x0308;
+const uint32_t jerrno::JERR_JDIR_UNLINK = 0x0309;
+const uint32_t jerrno::JERR_JDIR_BADFTYPE = 0x030a;
// class fcntl
//const uint32_t jerrno::JERR_FCNTL_OPENWR = 0x0400;
@@ -82,31 +82,31 @@ const uint32_t jerrno::JERR_JDIR_BADFTYP
//const uint32_t jerrno::JERR_RRFC_OPENRD = 0x0600;
// class jrec, enq_rec, deq_rec, txn_rec
-const uint32_t jerrno::JERR_JREC_BADRECHDR = 0x0700;
-const uint32_t jerrno::JERR_JREC_BADRECTAIL = 0x0701;
+const uint32_t jerrno::JERR_JREC_BADRECHDR = 0x0700;
+const uint32_t jerrno::JERR_JREC_BADRECTAIL = 0x0701;
// class wmgr
-const uint32_t jerrno::JERR_WMGR_BADPGSTATE = 0x0801;
-const uint32_t jerrno::JERR_WMGR_BADDTOKSTATE = 0x0802;
-const uint32_t jerrno::JERR_WMGR_ENQDISCONT = 0x0803;
-const uint32_t jerrno::JERR_WMGR_DEQDISCONT = 0x0804;
-const uint32_t jerrno::JERR_WMGR_DEQRIDNOTENQ = 0x0805;
+const uint32_t jerrno::JERR_WMGR_BADPGSTATE = 0x0801;
+const uint32_t jerrno::JERR_WMGR_BADDTOKSTATE = 0x0802;
+const uint32_t jerrno::JERR_WMGR_ENQDISCONT = 0x0803;
+const uint32_t jerrno::JERR_WMGR_DEQDISCONT = 0x0804;
+const uint32_t jerrno::JERR_WMGR_DEQRIDNOTENQ = 0x0805;
// class rmgr
-const uint32_t jerrno::JERR_RMGR_UNKNOWNMAGIC = 0x0900;
-const uint32_t jerrno::JERR_RMGR_RIDMISMATCH = 0x0901;
+const uint32_t jerrno::JERR_RMGR_UNKNOWNMAGIC = 0x0900;
+const uint32_t jerrno::JERR_RMGR_RIDMISMATCH = 0x0901;
//const uint32_t jerrno::JERR_RMGR_FIDMISMATCH = 0x0902;
-const uint32_t jerrno::JERR_RMGR_ENQSTATE = 0x0903;
-const uint32_t jerrno::JERR_RMGR_BADRECTYPE = 0x0904;
+const uint32_t jerrno::JERR_RMGR_ENQSTATE = 0x0903;
+const uint32_t jerrno::JERR_RMGR_BADRECTYPE = 0x0904;
// class data_tok
-const uint32_t jerrno::JERR_DTOK_ILLEGALSTATE = 0x0a00;
+const uint32_t jerrno::JERR_DTOK_ILLEGALSTATE = 0x0a00;
// const uint32_t jerrno::JERR_DTOK_RIDNOTSET = 0x0a01;
// class enq_map, txn_map
-const uint32_t jerrno::JERR_MAP_DUPLICATE = 0x0b00;
-const uint32_t jerrno::JERR_MAP_NOTFOUND = 0x0b01;
-const uint32_t jerrno::JERR_MAP_LOCKED = 0x0b02;
+const uint32_t jerrno::JERR_MAP_DUPLICATE = 0x0b00;
+const uint32_t jerrno::JERR_MAP_NOTFOUND = 0x0b01;
+const uint32_t jerrno::JERR_MAP_LOCKED = 0x0b02;
// class jinf
//const uint32_t jerrno::JERR_JINF_CVALIDFAIL = 0x0c00;
@@ -121,6 +121,13 @@ const uint32_t jerrno::JERR_MAP_LOCKED
//const uint32_t jerrno::JERR_JINF_OWIBAD = 0x0c09;
//const uint32_t jerrno::JERR_JINF_ZEROLENFILE = 0x0c0a;
+// EFP errors
+const uint32_t jerrno::JERR_EFP_BADPARTITIONNAME = 0x0d01;
+const uint32_t jerrno::JERR_EFP_BADPARTITIONDIR = 0x0d02;
+const uint32_t jerrno::JERR_EFP_BADEFPDIRNAME = 0x0d03;
+const uint32_t jerrno::JERR_EFP_NOEFP = 0x0d04;
+const uint32_t jerrno::JERR_EFP_EMPTY = 0x0d05;
+
// Negative returns for some functions
const int32_t jerrno::AIO_TIMEOUT = -1;
const int32_t jerrno::LOCK_TAKEN = -2;
@@ -222,6 +229,13 @@ jerrno::__init()
// _err_map[JERR_JINF_OWIBAD] = "JERR_JINF_OWIBAD: Journal data files have inconsistent OWI flags; >1 transition found in non-auto-expand or min-size journal";
// _err_map[JERR_JINF_ZEROLENFILE] = "JERR_JINF_ZEROLENFILE: Journal info file zero length";
+ // EFP errors
+ _err_map[JERR_EFP_BADPARTITIONNAME] = "JERR_EFP_BADPARTITIONNAME: Invalid partition name (must be \'pNNN\' where NNN is a non-zero number)";
+ _err_map[JERR_EFP_BADEFPDIRNAME] = "JERR_EFP_BADEFPDIRNAME: Bad Empty File Pool directory name (must be \'NNNk\', where NNN is a number which is a multiple of 4)";
+ _err_map[JERR_EFP_BADPARTITIONDIR] = "JERR_EFP_BADPARTITIONDIR: Invalid partition directory";
+ _err_map[JERR_EFP_NOEFP] = "JERR_EFP_NOEFP: No Empty File Pool found for given partition and empty file size";
+ _err_map[JERR_EFP_EMPTY] = "JERR_EFP_EMPTY: Empty File Pool is empty";
+
//_err_map[] = "";
return true;
Modified: qpid/branches/linearstore/qpid/cpp/src/qpid/linearstore/jrnl/jerrno.h
URL: http://svn.apache.org/viewvc/qpid/branches/linearstore/qpid/cpp/src/qpid/linearstore/jrnl/jerrno.h?rev=1525050&r1=1525049&r2=1525050&view=diff
==============================================================================
--- qpid/branches/linearstore/qpid/cpp/src/qpid/linearstore/jrnl/jerrno.h (original)
+++ qpid/branches/linearstore/qpid/cpp/src/qpid/linearstore/jrnl/jerrno.h Fri Sep 20 16:43:10 2013
@@ -140,6 +140,13 @@ namespace qls_jrnl
// static const uint32_t JERR_JINF_OWIBAD; ///< OWI inconsistent (>1 transition in non-ae journal)
// static const uint32_t JERR_JINF_ZEROLENFILE; ///< Journal info file is zero length (empty).
+ // EFP errors
+ static const uint32_t JERR_EFP_BADPARTITIONNAME; ///< Partition name invalid or of value 0
+ static const uint32_t JERR_EFP_BADEFPDIRNAME; ///< Empty File Pool directory name invalid
+ static const uint32_t JERR_EFP_BADPARTITIONDIR; ///< Invalid partition directory
+ static const uint32_t JERR_EFP_NOEFP; ///< No EFP found for given partition and file size
+ static const uint32_t JERR_EFP_EMPTY; ///< EFP empty
+
// Negative returns for some functions
static const int32_t AIO_TIMEOUT; ///< Timeout waiting for AIO return
static const int32_t LOCK_TAKEN; ///< Attempted to take lock, but it was taken by another thread
Modified: qpid/branches/linearstore/qpid/cpp/src/qpid/linearstore/jrnl/rmgr.cpp
URL: http://svn.apache.org/viewvc/qpid/branches/linearstore/qpid/cpp/src/qpid/linearstore/jrnl/rmgr.cpp?rev=1525050&r1=1525049&r2=1525050&view=diff
==============================================================================
--- qpid/branches/linearstore/qpid/cpp/src/qpid/linearstore/jrnl/rmgr.cpp (original)
+++ qpid/branches/linearstore/qpid/cpp/src/qpid/linearstore/jrnl/rmgr.cpp Fri Sep 20 16:43:10 2013
@@ -61,7 +61,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
@@ -638,6 +638,7 @@ rmgr::init_aio_reads(const int16_t /*fir
void
rmgr::rotate_page()
{
+/*
_page_cb_arr[_pg_index]._rdblks = 0;
_page_cb_arr[_pg_index]._state = UNUSED;
if (_pg_offset_dblks >= JRNL_RMGR_PAGE_SIZE * JRNL_SBLK_SIZE_DBLKS)
@@ -654,6 +655,7 @@ rmgr::rotate_page()
// Need to move reset into if (_rrfc.file_rotate()) above.
if (_pg_cntr >= (_jc->jfsize_sblks() / JRNL_RMGR_PAGE_SIZE))
_pg_cntr = 0;
+*/
}
uint32_t
Modified: qpid/branches/linearstore/qpid/cpp/src/qpid/linearstore/jrnl/utils/file_hdr.c
URL: http://svn.apache.org/viewvc/qpid/branches/linearstore/qpid/cpp/src/qpid/linearstore/jrnl/utils/file_hdr.c?rev=1525050&r1=1525049&r2=1525050&view=diff
==============================================================================
--- qpid/branches/linearstore/qpid/cpp/src/qpid/linearstore/jrnl/utils/file_hdr.c (original)
+++ qpid/branches/linearstore/qpid/cpp/src/qpid/linearstore/jrnl/utils/file_hdr.c Fri Sep 20 16:43:10 2013
@@ -20,28 +20,69 @@
*/
#include "file_hdr.h"
+#include <string.h>
-int file_hdr_init(file_hdr_t* dest, const uint32_t magic, const uint16_t version, const uint16_t uflag,
- const uint64_t rid, const uint64_t fro, const uint32_t file_count, const uint64_t file_size,
- const uint64_t file_number) {
- rec_hdr_init(&dest->_rhdr, magic, version, uflag, rid);
+void file_hdr_create(file_hdr_t* dest, const uint32_t magic, const uint16_t version, const uint16_t fhdr_size_sblks,
+ const uint16_t efp_partition, const uint64_t file_size) {
+ rec_hdr_init(&dest->_rhdr, magic, version, 0, 0);
+ dest->_fhdr_size_sblks = fhdr_size_sblks;
+ dest->_efp_partition = efp_partition;
+ dest->_reserved = 0;
+ dest->_file_size_kib = file_size;
+ dest->_fro = 0;
+ dest->_ts_nsec = 0;
+ dest->_ts_sec = 0;
+ dest->_file_number = 0;
+ dest->_queue_name_len = 0;
+}
+
+int file_hdr_init(file_hdr_t* dest, const uint16_t uflag, const uint64_t rid, const uint64_t fro,
+ const uint64_t file_number, const uint16_t queue_name_len, const char* queue_name) {
+ dest->_rhdr._uflag = uflag;
+ dest->_rhdr._rid = rid;
dest->_fro = fro;
- dest->_file_count = file_count;
- dest->_file_size = file_size;
dest->_file_number = file_number;
+ if (sizeof(file_hdr_t) + queue_name_len < MAX_FILE_HDR_LEN) {
+ dest->_queue_name_len = queue_name_len;
+ } else {
+ dest->_queue_name_len = MAX_FILE_HDR_LEN - sizeof(file_hdr_t);
+ }
+ dest->_queue_name_len = queue_name_len;
+ memcpy(dest + sizeof(file_hdr_t), queue_name, queue_name_len);
return set_time_now(dest);
}
void file_hdr_copy(file_hdr_t* dest, const file_hdr_t* src) {
rec_hdr_copy(&dest->_rhdr, &src->_rhdr);
+ dest->_fhdr_size_sblks = src->_fhdr_size_sblks; // Should this be copied?
+ dest->_efp_partition = src->_efp_partition; // Should this be copied?
+ dest->_file_size_kib = src->_file_size_kib;
dest->_fro = src->_fro;
dest->_ts_sec = src->_ts_sec;
dest->_ts_nsec = src->_ts_nsec;
- dest->_file_count = src->_file_count;
- dest->_file_size = src->_file_size;
dest->_file_number = src->_file_number;
}
+void file_hdr_reset(file_hdr_t* target) {
+ target->_rhdr._uflag = 0;
+ target->_rhdr._rid = 0;
+ target->_fro = 0;
+ target->_ts_sec = 0;
+ target->_ts_nsec = 0;
+ target->_file_number = 0;
+ target->_queue_name_len = 0;
+ memset(target + sizeof(file_hdr_t), 0, MAX_FILE_HDR_LEN - sizeof(file_hdr_t));
+}
+
+int is_file_hdr_reset(file_hdr_t* target) {
+ return target->_rhdr._uflag == 0 &&
+ target->_rhdr._rid == 0 &&
+ target->_ts_sec == 0 &&
+ target->_ts_nsec == 0 &&
+ target->_file_number == 0 &&
+ target->_queue_name_len == 0;
+}
+
int set_time_now(file_hdr_t *fh)
{
struct timespec ts;
Modified: qpid/branches/linearstore/qpid/cpp/src/qpid/linearstore/jrnl/utils/file_hdr.h
URL: http://svn.apache.org/viewvc/qpid/branches/linearstore/qpid/cpp/src/qpid/linearstore/jrnl/utils/file_hdr.h?rev=1525050&r1=1525049&r2=1525050&view=diff
==============================================================================
--- qpid/branches/linearstore/qpid/cpp/src/qpid/linearstore/jrnl/utils/file_hdr.h (original)
+++ qpid/branches/linearstore/qpid/cpp/src/qpid/linearstore/jrnl/utils/file_hdr.h Fri Sep 20 16:43:10 2013
@@ -28,6 +28,8 @@
extern "C"{
#endif
+#define MAX_FILE_HDR_LEN 4096 // Set to 1 sblk
+
#pragma pack(1)
/**
@@ -47,44 +49,50 @@ extern "C"{
* +---+---+---+---+---+---+---+---+ | struct rec_hdr_t
* | first rid in file | |
* +---+---+---+---+---+---+---+---+ -+
+ * | fs | partn | reserved |
+ * +---+---+---+---+---+---+---+---+
+ * | file-size |
+ * +---+---+---+---+---+---+---+---+
* | fro |
* +---+---+---+---+---+---+---+---+
* | timestamp (sec) |
* +---+---+---+---+---+---+---+---+
* | timestamp (ns) |
* +---+---+---+---+---+---+---+---+
- * | file-count | reserved |
- * +---+---+---+---+---+---+---+---+
- * | file-size |
- * +---+---+---+---+---+---+---+---+
* | file-number |
* +---+---+---+---+---+---+---+---+
- * | n-len | Queue Name... |
+ * | qnl | Queue Name... |
* +-------+ |
* | |
* +---+---+---+---+---+---+---+---+
*
- * ver = file version (If the format or encoding of this file changes, then this
- * number should be incremented)
+ * ver = Journal version
+ * rid = Record ID
+ * fs = File header size in sblks (defined by JRNL_SBLK_SIZE)
+ * partn = EFP partition from which this file came
* fro = First Record Offset
- * n-len = Length of the queue name in octets.
+ * qnl = Length of the queue name in octets.
* </pre>
*/
typedef struct file_hdr_t {
- rec_hdr_t _rhdr; /**< Common record header struct, but rid field is used for rid of first compete record in file */
- uint64_t _fro; /**< First Record Offset (FRO) */
- uint64_t _ts_sec; /**< Time stamp (seconds part) */
- uint64_t _ts_nsec; /**< Time stamp (nanoseconds part) */
- uint32_t _file_count; /**< Total number of files in linear sequence at time of writing this file */
+ rec_hdr_t _rhdr; /**< Common record header struct, but rid field is used for rid of first compete record in file */
+ uint16_t _fhdr_size_sblks; /**< File header size in sblks (defined by JRNL_SBLK_SIZE) */
+ uint16_t _efp_partition; /**< EFP Partition number from which this file was obtained */
uint32_t _reserved;
- uint64_t _file_size; /**< Size of this file in octets, including header sblk */
- uint64_t _file_number; /**< The logical number of this file in a monotonically increasing sequence */
- uint16_t _name_length; /**< Length of the queue name in octets, which follows this struct in the header */
+ uint64_t _file_size_kib; /**< Size of this file in KiB, excluding header sblk */
+ uint64_t _fro; /**< First Record Offset (FRO) */
+ uint64_t _ts_sec; /**< Time stamp (seconds part) */
+ uint64_t _ts_nsec; /**< Time stamp (nanoseconds part) */
+ uint64_t _file_number; /**< The logical number of this file in a monotonically increasing sequence */
+ uint16_t _queue_name_len; /**< Length of the queue name in octets, which follows this struct in the header */
} file_hdr_t;
-int file_hdr_init(file_hdr_t* dest, const uint32_t magic, const uint16_t version, const uint16_t uflag,
- const uint64_t rid, const uint64_t fro, const uint32_t file_count, const uint64_t file_size,
- const uint64_t file_number);
+void file_hdr_create(file_hdr_t* dest, const uint32_t magic, const uint16_t version, const uint16_t fhdr_size_sblks,
+ const uint16_t efp_partition, const uint64_t file_size);
+int file_hdr_init(file_hdr_t* dest, const uint16_t uflag, const uint64_t rid, const uint64_t fro,
+ const uint64_t file_number, const uint16_t queue_name_len, const char* queue_name);
+void file_hdr_reset(file_hdr_t* target);
+int is_file_hdr_reset(file_hdr_t* target);
void file_hdr_copy(file_hdr_t* dest, const file_hdr_t* src);
int set_time_now(file_hdr_t *fh);
void set_time(file_hdr_t *fh, struct timespec *ts);
Modified: qpid/branches/linearstore/qpid/cpp/src/qpid/linearstore/jrnl/wmgr.cpp
URL: http://svn.apache.org/viewvc/qpid/branches/linearstore/qpid/cpp/src/qpid/linearstore/jrnl/wmgr.cpp?rev=1525050&r1=1525049&r2=1525050&view=diff
==============================================================================
--- qpid/branches/linearstore/qpid/cpp/src/qpid/linearstore/jrnl/wmgr.cpp (original)
+++ qpid/branches/linearstore/qpid/cpp/src/qpid/linearstore/jrnl/wmgr.cpp Fri Sep 20 16:43:10 2013
@@ -45,9 +45,9 @@ wmgr::wmgr(jcntl* jc, enq_map& emap, txn
_fhdr_ptr_arr(0),
_fhdr_aio_cb_arr(0),
_cached_offset_dblks(0),
- _jfsize_dblks(0),
- _jfsize_pgs(0),
- _num_jfiles(0),
+// _jfsize_dblks(0),
+// _jfsize_pgs(0),
+// _num_jfiles(0),
_enq_busy(false),
_deq_busy(false),
_abort_busy(false),
@@ -65,9 +65,9 @@ wmgr::wmgr(jcntl* jc, enq_map& emap, txn
_fhdr_ptr_arr(0),
_fhdr_aio_cb_arr(0),
_cached_offset_dblks(0),
- _jfsize_dblks(0),
- _jfsize_pgs(0),
- _num_jfiles(0),
+// _jfsize_dblks(0),
+// _jfsize_pgs(0),
+// _num_jfiles(0),
_enq_busy(false),
_deq_busy(false),
_abort_busy(false),
@@ -94,9 +94,9 @@ wmgr::initialize(aio_callback* const cbp
initialize(cbp, wcache_pgsize_sblks, wcache_num_pages);
- _jfsize_dblks = _jc->jfsize_sblks() * JRNL_SBLK_SIZE_DBLKS;
- _jfsize_pgs = _jc->jfsize_sblks() / _cache_pgsize_sblks;
- assert(_jc->jfsize_sblks() % JRNL_RMGR_PAGE_SIZE == 0);
+// _jfsize_dblks = _jc->jfsize_sblks() * JRNL_SBLK_SIZE_DBLKS;
+// _jfsize_pgs = _jc->jfsize_sblks() / _cache_pgsize_sblks;
+// assert(_jc->jfsize_sblks() % JRNL_RMGR_PAGE_SIZE == 0);
if (eo)
{
@@ -555,7 +555,7 @@ wmgr::file_header_check(const uint64_t /
}
void
-wmgr::flush_check(iores& res, bool& cont, bool& done)
+wmgr::flush_check(iores& res, bool& /*cont*/, bool& done)
{
// Is page is full, flush
if (_pg_offset_dblks >= _cache_pgsize_sblks * JRNL_SBLK_SIZE_DBLKS)
@@ -569,6 +569,7 @@ wmgr::flush_check(iores& res, bool& cont
done = true;
}
+/*
// If file is full, rotate to next file
if (_pg_cntr >= _jfsize_pgs)
{
@@ -583,6 +584,7 @@ wmgr::flush_check(iores& res, bool& cont
done = true;
}
}
+*/
}
}
@@ -590,12 +592,14 @@ iores
wmgr::flush()
{
iores res = write_flush();
+/*
if (_pg_cntr >= _jfsize_pgs)
{
iores rfres = rotate_file();
if (rfres != RHM_IORES_SUCCESS)
res = rfres;
}
+*/
return res;
}
@@ -836,12 +840,13 @@ wmgr::is_txn_synced(const std::string& x
}
void
-wmgr::initialize(aio_callback* const cbp, const uint32_t wcache_pgsize_sblks, const uint16_t wcache_num_pages)
+wmgr::initialize(aio_callback* const /*cbp*/, const uint32_t /*wcache_pgsize_sblks*/, const uint16_t /*wcache_num_pages*/)
{
+/*
pmgr::initialize(cbp, wcache_pgsize_sblks, wcache_num_pages);
wmgr::clean();
- /*_num_jfiles = _jc->num_jfiles();*/ // TODO: replace for linearstore: _jc->num_jfiles()
- if (::posix_memalign(&_fhdr_base_ptr, _sblksize, _sblksize /** _num_jfiles*/))
+ _num_jfiles = _jc->num_jfiles(); // TODO: replace for linearstore: _jc->num_jfiles()
+ if (::posix_memalign(&_fhdr_base_ptr, _sblksize, _sblksize * _num_jfiles))
{
wmgr::clean();
std::ostringstream oss;
@@ -863,6 +868,7 @@ wmgr::initialize(aio_callback* const cbp
_ddtokl.clear();
_cached_offset_dblks = 0;
_enq_busy = false;
+*/
}
iores
@@ -988,7 +994,7 @@ void
wmgr::write_fhdr(uint64_t rid, uint16_t fid, uint16_t /*lid*/, std::size_t fro)
{
file_hdr_t fhdr/*(QLS_FILE_MAGIC, QLS_JRNL_VERSION, rid, fid, lid, fro, _wrfc.owi(), true)*/;
- /*int err =*/ ::file_hdr_init(&fhdr, QLS_FILE_MAGIC, QLS_JRNL_VERSION, 0, rid, fro, 0, 0, 0);
+ /*int err =*/ ::file_hdr_init(&fhdr, 0, rid, fro, 0, _jc->id().length(), _jc->id().c_str());
std::memcpy(_fhdr_ptr_arr[fid], &fhdr, sizeof(fhdr));
#ifdef RHM_CLEAN
std::memset((char*)_fhdr_ptr_arr[fid] + sizeof(fhdr), RHM_CLEAN_CHAR, _sblksize - sizeof(fhdr));
@@ -1030,8 +1036,8 @@ wmgr::clean()
if (_fhdr_aio_cb_arr)
{
- for (uint32_t i=0; i<_num_jfiles; i++)
- delete _fhdr_aio_cb_arr[i];
+// for (uint32_t i=0; i<_num_jfiles; i++)
+// delete _fhdr_aio_cb_arr[i];
std::free(_fhdr_aio_cb_arr);
_fhdr_aio_cb_arr = 0;
}
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@qpid.apache.org
For additional commands, e-mail: commits-help@qpid.apache.org