QPID-5671: [linearstore] Add ability to use disk partitions and select per-queue EFPs. Modified previous code and added new code to allow a seamless updrade of the modified EFP and journal directory structure from the old to the new. The "efp" directory is removed from the directory tree containing the EFPs in a partition and all directories are moved up a level. The journal files are replaced with symlinks as they are used and replaced into the EFP.

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

5 files changed, 82 insertions, 5 deletions

diff --git a/qpid/cpp/src/qpid/linearstore/journal/EmptyFilePool.cpp b/qpid/cpp/src/qpid/linearstore/journal/EmptyFilePool.cpp
index 371443f46b..cc54191c98 100644
--- a/qpid/cpp/src/qpid/linearstore/journal/EmptyFilePool.cpp
+++ b/qpid/cpp/src/qpid/linearstore/journal/EmptyFilePool.cpp
@@ -144,7 +144,15 @@ std::string EmptyFilePool::takeEmptyFile(const std::string& destDirectory) {
 }
 
 void EmptyFilePool::returnEmptyFileSymlink(const std::string& emptyFileSymlink) {
-    returnEmptyFile(deleteSymlink(emptyFileSymlink));
+    if (isFile(emptyFileSymlink)) {
+        returnEmptyFile(emptyFileSymlink);
+    } else if(isSymlink(emptyFileSymlink)) {
+        returnEmptyFile(deleteSymlink(emptyFileSymlink));
+    } else {
+        std::ostringstream oss;
+        oss << "File \"" << emptyFileSymlink << "\" is neither a file nor a symlink";
+        throw jexception(jerrno::JERR_EFP_BADFILETYPE, oss.str(), "EmptyFilePool", "returnEmptyFileSymlink");
+    }
 }
 
 //static
@@ -422,4 +430,27 @@ std::string EmptyFilePool::deleteSymlink(const std::string& fqLinkName) {
     return std::string(buff, len);
 }
 
+//static
+bool EmptyFilePool::isFile(const std::string& fqName) {
+    struct stat buff;
+    if (::lstat(fqName.c_str(), &buff)) {
+        std::ostringstream oss;
+        oss << "lstat file=\"" << fqName << "\"" << FORMAT_SYSERR(errno);
+        throw jexception(jerrno::JERR_EFP_LSTAT, oss.str(), "EmptyFilePool", "isFile");
+    }
+    return S_ISREG(buff.st_mode);
+}
+
+//static
+bool EmptyFilePool::isSymlink(const std::string& fqName) {
+    struct stat buff;
+    if (::lstat(fqName.c_str(), &buff)) {
+        std::ostringstream oss;
+        oss << "lstat file=\"" << fqName << "\"" << FORMAT_SYSERR(errno);
+        throw jexception(jerrno::JERR_EFP_LSTAT, oss.str(), "EmptyFilePool", "isSymlink");
+    }
+    return S_ISLNK(buff.st_mode);
+
+}
+
 }}}
diff --git a/qpid/cpp/src/qpid/linearstore/journal/EmptyFilePool.h b/qpid/cpp/src/qpid/linearstore/journal/EmptyFilePool.h
index dbc3992f46..1a1264fa26 100644
--- a/qpid/cpp/src/qpid/linearstore/journal/EmptyFilePool.h
+++ b/qpid/cpp/src/qpid/linearstore/journal/EmptyFilePool.h
@@ -102,6 +102,8 @@ protected:
     static int createSymLink(const std::string& fqFileName,
                              const std::string& fqLinkName);
     static std::string deleteSymlink(const std::string& fqLinkName);
+    static bool isFile(const std::string& fqName);
+    static bool isSymlink(const std::string& fqName);
 };
 
 }}}
diff --git a/qpid/cpp/src/qpid/linearstore/journal/EmptyFilePoolPartition.cpp b/qpid/cpp/src/qpid/linearstore/journal/EmptyFilePoolPartition.cpp
index c10caebc6f..a31855e0d8 100644
--- a/qpid/cpp/src/qpid/linearstore/journal/EmptyFilePoolPartition.cpp
+++ b/qpid/cpp/src/qpid/linearstore/journal/EmptyFilePoolPartition.cpp
@@ -24,7 +24,9 @@
 #include <iomanip>
 #include "qpid/linearstore/journal/EmptyFilePool.h"
 #include "qpid/linearstore/journal/jdir.h"
+#include "qpid/linearstore/journal/JournalLog.h"
 #include "qpid/linearstore/journal/slock.h"
+#include <unistd.h>
 
 namespace qpid {
 namespace linearstore {
@@ -54,13 +56,38 @@ EmptyFilePoolPartition::~EmptyFilePoolPartition() {
 
 void
 EmptyFilePoolPartition::findEmptyFilePools() {
-//std::cout << "*** Reading " << partitionDir_ << std::endl; // DEBUG
+//std::cout << "*** EmptyFilePoolPartition::findEmptyFilePools(): Reading " << partitionDir_ << std::endl; // DEBUG
     std::vector<std::string> dirList;
-        jdir::read_dir(partitionDir_, dirList, true, false, false, true);
+    bool upgradeDirStructureFlag = false;
+    std::string oldPartitionDir;
+        jdir::read_dir(partitionDir_, dirList, true, false, false, false);
+//std::cout << "*** dirList.size()=" << dirList.size() << "; dirList.front()=" << dirList.front() << std::endl; // DEBUG
+        if (dirList.size() == 1 && dirList.front().compare("efp") == 0) {
+            upgradeDirStructureFlag = true;
+            oldPartitionDir = partitionDir_ + "/efp";
+//std::cout << "*** oldPartitionDir=" << oldPartitionDir << std::endl; // DEBUG
+            dirList.clear();
+            jdir::read_dir(oldPartitionDir, dirList, true, false, false, false);
+        }
         for (std::vector<std::string>::iterator i = dirList.begin(); i != dirList.end(); ++i) {
+            std::string fqFileName(partitionDir_ + "/" + *i);
+            if (upgradeDirStructureFlag) {
+                std::string fqOldFileName(partitionDir_ + "/efp/" + *i);
+                if (::rename(fqOldFileName.c_str(), fqFileName.c_str())) {
+                    // File move failed
+                    std::ostringstream oss;
+                    oss << "File \'" << fqOldFileName << "\' could not be renamed to \'" << fqFileName << "\' (" << FORMAT_SYSERR(errno) << "); file deleted";
+                    journalLogRef_.log(JournalLog::LOG_WARN, oss.str());
+                    if (::unlink(fqOldFileName.c_str())) {
+                        std::ostringstream oss;
+                        oss << "File \'" << fqOldFileName << "\' could not be deleted (" << FORMAT_SYSERR(errno) << "\'; file orphaned";
+                        journalLogRef_.log(JournalLog::LOG_WARN, oss.str());
+                    }
+                }
+            }
             EmptyFilePool* efpp = 0;
             try {
-                efpp = new EmptyFilePool(*i, this, overwriteBeforeReturnFlag_, truncateFlag_, journalLogRef_);
+                efpp = new EmptyFilePool(fqFileName, this, overwriteBeforeReturnFlag_, truncateFlag_, journalLogRef_);
                 {
                     slock l(efpMapMutex_);
                     efpMap_[efpp->dataSize_kib()] = efpp;
@@ -71,12 +98,23 @@ EmptyFilePoolPartition::findEmptyFilePools() {
                     delete efpp;
                     efpp = 0;
                 }
-                //std::cerr << "WARNING: " << e.what() << std::endl;
+                std::ostringstream oss;
+                oss << "EmptyFilePool create failed: " << e.what();
+                journalLogRef_.log(JournalLog::LOG_WARN, oss.str());
             }
             if (efpp != 0) {
                 efpp->initialize();
             }
         }
+        if (upgradeDirStructureFlag) {
+            std::string oldEfpDir(partitionDir_ + "/efp");
+            if (::rmdir(oldEfpDir.c_str())) {
+                // Unable to delete old "efp" dir
+                std::ostringstream oss;
+                oss << "Unable to delete old EFP directory \'" << oldEfpDir << "\' (" << FORMAT_SYSERR(errno) << "\'; directory orphaned";
+                journalLogRef_.log(JournalLog::LOG_WARN, oss.str());
+            }
+        }
 }
 
 EmptyFilePool* EmptyFilePoolPartition::getEmptyFilePool(const efpDataSize_kib_t efpDataSize_kib) {
diff --git a/qpid/cpp/src/qpid/linearstore/journal/jerrno.cpp b/qpid/cpp/src/qpid/linearstore/journal/jerrno.cpp
index e1602cf961..2d0f0c2f94 100644
--- a/qpid/cpp/src/qpid/linearstore/journal/jerrno.cpp
+++ b/qpid/cpp/src/qpid/linearstore/journal/jerrno.cpp
@@ -113,6 +113,8 @@ const uint32_t jerrno::JERR_EFP_BADEFPDIRNAME    = 0x0d03;
 const uint32_t jerrno::JERR_EFP_NOEFP            = 0x0d04;
 const uint32_t jerrno::JERR_EFP_EMPTY            = 0x0d05;
 const uint32_t jerrno::JERR_EFP_SYMLINK          = 0x0d06;
+const uint32_t jerrno::JERR_EFP_LSTAT            = 0x0d07;
+const uint32_t jerrno::JERR_EFP_BADFILETYPE      = 0x0d08;
 
 // Negative returns for some functions
 const int32_t jerrno::AIO_TIMEOUT                = -1;
@@ -208,6 +210,8 @@ jerrno::__init()
     _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[JERR_EFP_SYMLINK] = "JERR_EFP_SYMLINK: Symbolic link operation failed";
+    _err_map[JERR_EFP_LSTAT] = "JERR_EFP_LSTAT: lstat() operation failed";
+    _err_map[JERR_EFP_BADFILETYPE] = "JERR_EFP_BADFILETYPE: File type incorrect for operation";
 
     //_err_map[] = "";
 
diff --git a/qpid/cpp/src/qpid/linearstore/journal/jerrno.h b/qpid/cpp/src/qpid/linearstore/journal/jerrno.h
index 36410f9844..104618efaa 100644
--- a/qpid/cpp/src/qpid/linearstore/journal/jerrno.h
+++ b/qpid/cpp/src/qpid/linearstore/journal/jerrno.h
@@ -131,6 +131,8 @@ namespace journal {
         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
         static const uint32_t JERR_EFP_SYMLINK;         ///< Symbolic Link operation failed
+        static const uint32_t JERR_EFP_LSTAT;           ///< lstat operation failed
+        static const uint32_t JERR_EFP_BADFILETYPE;     ///< Bad file type
 
         // Negative returns for some functions
         static const int32_t AIO_TIMEOUT;               ///< Timeout waiting for AIO return