Manage HeaderPage - PageManager interaction + automatic DB extension

src/config/README.md

@@ -74,6 +74,9 @@ The configuration parameters can be of different uses:
 |`hashDB64`|test|boolean|Use HashDB64 new database (experimental)|false|HASHDB64|
 |`kvDBMaxVersions`|production|u64|Maximum number of KV versionn in Database|131072|HASHDB64_MAX_VERSIONS|
 |`dbCacheSynchURL`|test|string|URL of the HashDB service to synchronize the Database cache (experimental)|""|DB_CACHE_SYNCH_URL|
+|`hashDBFileName`|test|string|core name used for the hashDB files (path,numbering and extension not included). If hashDBFileName is empty in-memory version of the hashDB is used (only for DEBUG purposes). |""|HASHDB_FILE_NAME|
+|`hashDBFileSize`|test|u64| hashDB files size in GB|128|HASHDB_FILE_SIZE|
+|`hashDBFolder`|test|string|folder containing the hashDB files|hashdb|HASHDB_FOLDER|
 |`aggregatorServerPort`|test|u16|Aggregator server GRPC port|50081|AGGREGATOR_SERVER_PORT|
 |**`aggregatorClientPort`**|production|u16|Aggregator client GRPC port to connect to|50081|AGGREGATOR_SERVER_PORT|
 |**`aggregatorClientHost`**|production|string|Aggregator client GRPC host name to connect to, i.e. Aggregator server host name|"127.0.0.1"|AGGREGATOR_CLIENT_HOST|

src/config/config.cpp

@@ -197,6 +197,9 @@ void Config::load(json &config)
     ParseBool(config, "hashDB64", "HASHDB64", hashDB64, false);
     ParseU64(config, "kvDBMaxVersions", "HASHDB64_MAX_VERSIONS", kvDBMaxVersions, 131072);
     ParseString(config, "dbCacheSynchURL", "DB_CACHE_SYNCH_URL", dbCacheSynchURL, "");
+    ParseString(config, "hashDBFileName", "HASHDB_FILE_NAME", hashDBFileName, "");
+    ParseU64(config, "hashDBFileSize", "HASHDB_FILE_SIZE", hashDBFileSize, 128);    
+    ParseString(config, "hashDBFolder", "HASHDB_FOLDER", hashDBFolder, "hashdb");
     ParseU16(config, "aggregatorServerPort", "AGGREGATOR_SERVER_PORT", aggregatorServerPort, 50081);
     ParseU16(config, "aggregatorClientPort", "AGGREGATOR_CLIENT_PORT", aggregatorClientPort, 50081);
     ParseString(config, "aggregatorClientHost", "AGGREGATOR_CLIENT_HOST", aggregatorClientHost, "127.0.0.1");
@@ -433,6 +436,9 @@ void Config::print(void)
     zklog.info("    hashDB64=" + to_string(hashDB64));
     zklog.info("    kvDBMaxVersions=" + to_string(kvDBMaxVersions));
     zklog.info("    dbCacheSynchURL=" + dbCacheSynchURL);
+    zklog.info("    hashDBFileName=" + hashDBFileName);
+    zklog.info("    hashDBFileSize=" + to_string(hashDBFileSize));
+    zklog.info("    hastDBFolder=" + hashDBFolder);
     zklog.info("    aggregatorServerPort=" + to_string(aggregatorServerPort));
     zklog.info("    aggregatorClientPort=" + to_string(aggregatorClientPort));
     zklog.info("    aggregatorClientHost=" + aggregatorClientHost);

src/config/config.hpp

@@ -87,6 +87,9 @@ class Config
     bool hashDB64;
     uint64_t kvDBMaxVersions;
     string dbCacheSynchURL;
+    string hashDBFileName;
+    uint64_t hashDBFileSize;
+    string hashDBFolder;
 
     // Aggregator service (client)
     uint16_t aggregatorServerPort;

src/hashdb64/database_64.cpp

@@ -27,17 +27,19 @@
 // Helper functions
 string removeBSXIfExists64(string s) {return ((s.at(0) == '\\') && (s.at(1) == 'x')) ? s.substr(2) : s;}
 
-Database64::Database64 (Goldilocks &fr, const Config &config) : headerPageNumber(0), currentFlushId(0), ctx(pageManager, config)
+Database64::Database64 (Goldilocks &fr, const Config &config) : headerPageNumber(0), currentFlushId(0), pageManager(), ctx(pageManager, config)
 {
     // Init mutex
     pthread_mutex_init(&mutex, NULL);
 
     zkresult zkr;
     headerPageNumber = 0;
-    zkr = HeaderPage::InitEmptyPage(ctx, headerPageNumber);
+
+    // Init page manager
+    zkr = pageManager.init(ctx);
     if (zkr != ZKR_SUCCESS)
     {
-        zklog.error("Database64::Database64() failed calling HeaderPage::InitEmptyPage() result=" + zkresult2string(zkr));
+        zklog.error("Database64::Database64() failed to init page manager result=" + zkresult2string(zkr));
         exitProcess();
     }
 };

src/hashdb64/database_64.hpp

@@ -70,9 +70,9 @@ class Database64
     uint64_t headerPageNumber;
     pthread_mutex_t mutex;
     uint64_t currentFlushId;
+    PageManager pageManager;
     PageContext ctx;
 
-
 public:
 
     // Constructor and destructor

src/hashdb64/page/header_page.cpp

@@ -9,6 +9,24 @@
 #include "page_list_page.hpp"
 #include "root_version_page.hpp"
 
+
+zkresult HeaderPage::Check (PageContext &ctx, const uint64_t headerPageNumber)
+{
+    // Get the header page
+    HeaderStruct * page = (HeaderStruct *)ctx.pageManager.getPageAddress(headerPageNumber);
+
+    // Check the uuid
+    for(int i=0; i<32; ++i){
+        if (page->uuid[i] != ctx.uuid[i])
+        {
+            zklog.error("HeaderPage::Check() found uuid[" + to_string(i) + "]=" + to_string(page->uuid[i]) + " != config.hashDBUUID[" + to_string(i) + "]=" + to_string(ctx.uuid[i]));
+            exitProcess();
+            return ZKR_DB_ERROR;
+        }
+    }
+    return ZKR_SUCCESS;
+
+}
 zkresult HeaderPage::InitEmptyPage (PageContext &ctx, const uint64_t headerPageNumber)
 {
     zkresult zkr;
@@ -64,13 +82,18 @@ zkresult HeaderPage::InitEmptyPage (PageContext &ctx, const uint64_t headerPageN
     }
 
     // Create the free pages page, and init it
+    page->firstUnusedPage = 2;
     page->freePages = ctx.pageManager.getFreePage();
     zkr = PageListPage::InitEmptyPage(ctx, page->freePages);
     if (zkr != ZKR_SUCCESS)
     {
         zklog.error("HeaderPage::InitEmptyPage() failed calling PageListPage::InitEmptyPage(freePages) result=" + zkresult2string(zkr));
         return zkr;
     }
+    // UUID
+    for(int i=0; i<32; ++i){
+        page->uuid[i] = ctx.uuid[i];
+    }
 
     return ZKR_SUCCESS;
 }
@@ -137,7 +160,18 @@ zkresult HeaderPage::CreateFreePages (PageContext &ctx, uint64_t &headerPageNumb
 
 }
 
-zkresult HeaderPage::setFirstUnusedPage (PageContext &ctx, uint64_t &headerPageNumber, const uint64_t firstUnusedPage)
+zkresult HeaderPage::GetFirstUnusedPage (PageContext &ctx,const uint64_t headerPageNumber, uint64_t &firstUnusedPage)
+{
+    // Get header page
+    HeaderStruct * page = (HeaderStruct *)ctx.pageManager.getPageAddress(headerPageNumber);
+
+    // Call the specific method
+    firstUnusedPage = page->firstUnusedPage;
+
+    return ZKR_SUCCESS;
+}
+
+zkresult HeaderPage::SetFirstUnusedPage (PageContext &ctx,uint64_t &headerPageNumber, const uint64_t firstUnusedPage)
 {
     // Get an editable page
     headerPageNumber = ctx.pageManager.editPage(headerPageNumber);

src/hashdb64/page/header_page.hpp

@@ -14,6 +14,9 @@
 
 struct HeaderStruct
 {
+    // UUID
+    uint8_t uuid[32];
+
     // Raw data
     uint64_t firstRawDataPage;
     uint64_t rawDataPage;
@@ -39,6 +42,10 @@ struct HeaderStruct
 class HeaderPage
 {
 public:
+
+    // Header uuid
+    static zkresult Check (PageContext &ctx, const uint64_t headerPageNumber);
+
     // Header-only methods
     static zkresult InitEmptyPage  (PageContext &ctx, const uint64_t  headerPageNumber);
     static uint64_t GetLastVersion (PageContext &ctx, const uint64_t  headerPageNumber);
@@ -48,7 +55,8 @@ class HeaderPage
     static zkresult GetFreePagesContainer (PageContext &ctx, const uint64_t  headerPageNumber, vector<uint64_t> (&containerPages));
     static zkresult GetFreePages          (PageContext &ctx, const uint64_t  headerPageNumber, vector<uint64_t> (&freePages));
     static zkresult CreateFreePages       (PageContext &ctx,       uint64_t &headerPageNumber, vector<uint64_t> (&freePages), vector<uint64_t> (&containerPages));
-    static zkresult setFirstUnusedPage    (PageContext &ctx,       uint64_t &headerPageNumber, const uint64_t firstUnusedPage);
+    static zkresult GetFirstUnusedPage    (PageContext &ctx,  const uint64_t  headerPageNumber, uint64_t &firstUnusedPage);
+    static zkresult SetFirstUnusedPage    (PageContext &ctx,       uint64_t &headerPageNumber, const uint64_t firstUnusedPage);
 
     // Root version methods
     static zkresult ReadRootVersion  (PageContext &ctx, const uint64_t  headerPageNumber, const string &root,       uint64_t &version);

src/hashdb64/page/page_list_page.cpp

@@ -178,8 +178,9 @@ zkresult PageListPage::GetContainerPages (PageContext &ctx, const uint64_t  page
 zkresult PageListPage::CreatePages (PageContext &ctx, uint64_t &pageNumber_, vector<uint64_t> (&freePages), vector<uint64_t> (&containerPages))
 {
     if(containerPages.size() == 0){
-        pageNumber_ = 0;
-        return ZKR_SUCCESS;
+        zklog.error("PageListPage::CreatePages() containerPages.size() == 0");
+        exitProcess();
+        return ZKR_DB_ERROR;
     }
     pageNumber_ = containerPages[0];
 
@@ -219,7 +220,7 @@ zkresult PageListPage::CreatePages (PageContext &ctx, uint64_t &pageNumber_, vec
         }
 
         // Update the page offset
-        page->nextPageNumberAndOffset = (offset + 8) << 48;
+        page->nextPageNumberAndOffset = offset << 48;
         page->previousPageNumber = (i == containerPages.size() - 1) ? 0 : containerPages[i + 1];
     }