Integrated hpfs hash map cache. (#107)

src/comm/comm_server.cpp

@@ -314,6 +314,8 @@ namespace comm
             std::vector<std::string> args_vec;
             args_vec.reserve(16);
 
+            const std::string max_msg_size_str = (max_msg_size > 0) ? std::to_string(max_msg_size) : "134217728"; //128MB
+
             // Fill process args.
             args_vec.push_back(conf::ctx.websocketd_exe_path);
             args_vec.push_back("--port");
@@ -325,10 +327,7 @@ namespace comm
             args_vec.push_back(conf::ctx.tls_key_file);
             args_vec.push_back(is_binary ? "--binary=true" : "--binary=false");
             args_vec.push_back(use_size_header ? "--sizeheader=true" : "--sizeheader=false");
-
-            if (max_msg_size > 0)
-                args_vec.push_back(std::string("--maxframe=").append(std::to_string(max_msg_size)));
-
+            args_vec.push_back(std::string("--maxframe=").append(max_msg_size_str));
             args_vec.push_back("--loglevel=error");
             args_vec.push_back("nc"); // netcat (OpenBSD) is used for domain socket redirection.
             args_vec.push_back("-U"); // Use UNIX domain socket

src/cons/cons.cpp

@@ -941,7 +941,7 @@ namespace cons
     {
         pid_t pid;
         std::string mount_dir;
-        if (hpfs::start_fs_session(pid, mount_dir, "ro", true) == -1)
+        if (hpfs::start_fs_session(pid, mount_dir, "ro", true, 60000) == -1)
             return -1;
 
         int res = get_hash(hash, mount_dir, "/");

src/hpfs/hpfs.cpp

@@ -6,8 +6,10 @@
 
 namespace hpfs
 {
-    pid_t merge_pid = 0;
+    constexpr const char *HPFS_TRACE_ARG = "trace=none";
+    constexpr uint16_t INIT_CHECK_INTERVAL = 20;
 
+    pid_t merge_pid = 0;
     bool init_success = false;
 
     int init()
@@ -55,6 +57,7 @@ namespace hpfs
                 conf::ctx.hpfs_exe_path.data(),
                 (char *)"merge",
                 conf::ctx.state_dir.data(),
+                (char *)HPFS_TRACE_ARG,
                 NULL};
 
             const int ret = execv(execv_args[0], execv_args);
@@ -71,13 +74,14 @@ namespace hpfs
     }
 
     int start_fs_session(pid_t &session_pid, std::string &mount_dir,
-                         const char *mode, const bool hash_map_enabled)
+                         const char *mode, const bool hash_map_enabled, const uint16_t timeout)
     {
         const pid_t pid = fork();
 
         if (pid > 0)
         {
             // HotPocket process.
+            LOG_DBG << "Starting hpfs " << mode << " session...";
 
             // If the mount dir is not specified, assign a mount dir based on hpfs process id.
             if (mount_dir.empty())
@@ -91,11 +95,12 @@ namespace hpfs
                                                : mount_dir;
 
             // Wait until hpfs is initialized properly.
+            const uint16_t max_retries = timeout / INIT_CHECK_INTERVAL;
             bool hpfs_initialized = false;
-            uint8_t retry_count = 0;
+            uint16_t retry_count = 0;
             do
             {
-                util::sleep(20);
+                util::sleep(INIT_CHECK_INTERVAL);
 
                 // Check if process is still running.
                 if (kill(pid, 0) == -1)
@@ -110,7 +115,15 @@ namespace hpfs
                 hpfs_initialized = (stat(check_path.c_str(), &st) == 0 &&
                                     (hash_map_enabled || st.st_ino == 1));
 
-            } while (!hpfs_initialized && ++retry_count < 200);
+                // The only error that warrants a retry is ENOENT (no entry).
+                // When hpfs is fully initialized we should receive some file from check_path.
+                if (!hpfs_initialized && errno != ENOENT)
+                {
+                    LOG_ERR << errno << ": Error in checking hpfs status.";
+                    break;
+                }
+
+            } while (!hpfs_initialized && ++retry_count <= max_retries);
 
             // Kill the process if hpfs couldn't be initialized after the wait period.
             if (!hpfs_initialized)
@@ -141,6 +154,7 @@ namespace hpfs
                 conf::ctx.state_dir.data(),
                 mount_dir.data(),
                 (char *)(hash_map_enabled ? "hmap=true" : "hmap=false"),
+                (char *)HPFS_TRACE_ARG,
                 NULL};
 
             const int ret = execv(execv_args[0], execv_args);

src/hpfs/hpfs.hpp

@@ -17,7 +17,7 @@ namespace hpfs
     void deinit();
     int start_merge_process();
     int start_fs_session(pid_t &session_pid, std::string &mount_dir,
-                         const char *mode, const bool hash_map_enabled);
+                         const char *mode, const bool hash_map_enabled, const uint16_t timeout = 4000);
     int get_hash(h32 &hash, const std::string_view mount_dir, const std::string_view vpath);
     int get_file_block_hashes(std::vector<h32> &hashes, const std::string_view mount_dir, const std::string_view vpath);
     int get_dir_children_hashes(std::vector<child_hash_node> &hash_nodes, const std::string_view mount_dir, const std::string_view dir_vpath);

src/state/state_sync.cpp

@@ -88,7 +88,6 @@ namespace state_sync
                 LOG_INFO << "State sync: Starting sync for target state: " << ctx.target_state;
             }
 
-            LOG_DBG << "State sync: Starting hpfs rw session...";
             pid_t hpfs_pid = 0;
             if (hpfs::start_fs_session(hpfs_pid, ctx.hpfs_mount_dir, "rw", true) != -1)
             {