Add support for bounding CPU usage during CLP compression.

components/core/CMakeLists.txt

@@ -342,6 +342,8 @@ set(SOURCE_FILES_unitTest
         src/clp/clp/CommandLineArguments.hpp
         src/clp/clp/compression.cpp
         src/clp/clp/compression.hpp
+        src/clp/clp/CpuUsageController.cpp
+        src/clp/clp/CpuUsageController.hpp
         src/clp/clp/decompression.cpp
         src/clp/clp/decompression.hpp
         src/clp/clp/FileCompressor.cpp

components/core/src/clp/clp/CMakeLists.txt

@@ -154,6 +154,8 @@ set(
         CommandLineArguments.hpp
         compression.cpp
         compression.hpp
+        CpuUsageController.cpp
+        CpuUsageController.hpp
         decompression.cpp
         decompression.hpp
         FileCompressor.cpp

components/core/src/clp/clp/CommandLineArguments.cpp

@@ -328,6 +328,8 @@ CommandLineArguments::parse_arguments(int argc, char const* argv[]) {
             compression_positional_options_description.add("input-paths", -1);
 
             // Define compression-specific options
+            constexpr double cCpuUsageBoundMax{1.0};
+            double cpu_usage_bound{};
             po::options_description options_compression("Compression Options");
             // boost::program_options doesn't support boolean flags which can be set to false, so
             // we use a string argument to set the flag manually.
@@ -384,6 +386,10 @@ CommandLineArguments::parse_arguments(int argc, char const* argv[]) {
                             ->default_value(m_schema_file_path),
                     "Path to a schema file. If not specified, heuristics are used to determine "
                     "dictionary variables. See README-Schema.md for details."
+            )(
+                    "cpu-usage-bound",
+                    po::value<double>(&cpu_usage_bound)->value_name("FLOAT")->default_value(cCpuUsageBoundMax),
+                    "CPU usage bound for compression. Valid range: (0, 1), inclusive."
             );
 
             po::options_description all_compression_options;
@@ -463,6 +469,15 @@ CommandLineArguments::parse_arguments(int argc, char const* argv[]) {
                     );
                 }
             }
+
+            if (cCpuUsageBoundMax != cpu_usage_bound) {
+                if (0 >= cpu_usage_bound || cCpuUsageBoundMax < cpu_usage_bound) {
+                    throw invalid_argument(
+                            "Invalid CPU usage bound: " + std::to_string(cpu_usage_bound)
+                    );
+                }
+                m_cpu_usage_bound.emplace(cpu_usage_bound);
+            }
         }
 
         // Validate an output directory was specified

components/core/src/clp/clp/CommandLineArguments.hpp

@@ -1,6 +1,7 @@
 #ifndef CLP_CLP_COMMANDLINEARGUMENTS_HPP
 #define CLP_CLP_COMMANDLINEARGUMENTS_HPP
 
+#include <optional>
 #include <string>
 #include <vector>
 
@@ -77,6 +78,10 @@ class CommandLineArguments : public CommandLineArgumentsBase {
 
     GlobalMetadataDBConfig const& get_metadata_db_config() const { return m_metadata_db_config; }
 
+    [[nodiscard]] auto get_cpu_usage_bound() const -> std::optional<double> {
+        return m_cpu_usage_bound;
+    }
+
 private:
     // Methods
     void print_basic_usage() const override;
@@ -104,6 +109,7 @@ class CommandLineArguments : public CommandLineArgumentsBase {
     std::string m_archives_dir;
     std::vector<std::string> m_input_paths;
     GlobalMetadataDBConfig m_metadata_db_config;
+    std::optional<double> m_cpu_usage_bound;
 };
 }  // namespace clp::clp
 

components/core/src/clp/clp/CpuUsageController.cpp

@@ -0,0 +1,38 @@
+#include "CpuUsageController.hpp"
+
+#include <csignal>
+#include <iostream>
+#include <thread>
+
+namespace clp {
+    CpuUsageController::CpuUsageController(pid_t pid, double ratio) : m_pid{pid} {
+        if (0 >= ratio || 1 <= ratio) {
+            std::cerr << "Invalid ratio: " << ratio << "\n";
+            exit(-1);
+        }
+
+        constexpr std::chrono::duration<double, std::chrono::milliseconds::period> cTimeWindow{50};
+        auto const execution_time{cTimeWindow * ratio};
+        m_execution_time = std::chrono::milliseconds{
+                std::chrono::duration_cast<std::chrono::milliseconds>(execution_time)
+        };
+        m_sleep_time = std::chrono::milliseconds{
+                std::chrono::duration_cast<std::chrono::milliseconds>(cTimeWindow - execution_time)
+        };
+    }
+
+    auto CpuUsageController::start() -> void {
+        while (true) {
+            std::this_thread::sleep_for(m_execution_time);
+            if (auto const err{kill(m_pid, SIGSTOP)}; 0 != err) {
+                std::cerr << "Failed to send `SIGSTOP`. Error code: " << err << "\n";
+                exit(-1);
+            }
+            std::this_thread::sleep_for(m_sleep_time);
+            if (auto const err{kill(m_pid, SIGCONT)}; 0 != err) {
+                std::cerr << "Failed to send `SIGCONT`. Error code: " << err << "\n";
+                exit(-1);
+            }
+        }
+    }
+}  // namespace clp

components/core/src/clp/clp/CpuUsageController.hpp

@@ -0,0 +1,39 @@
+#ifndef CLP_CPUUSAGECONTROLLER_HPP
+#define CLP_CPUUSAGECONTROLLER_HPP
+
+#include <sys/types.h>
+
+#include <chrono>
+
+namespace clp {
+/**
+ * Class for controlling the CPU usage of a process. It should be running in a
+ * separate process.
+ * The current implementation has the following limitations:
+ * 1. The controller can only define a upper bound for the target process. It
+ * does not monitor the CPU usage of the target process.
+ * 2. The controller exits if it fails to signal the target process. However,
+ * the signal is sent using `pid`, which means if the target process terminates
+ * but the original target pid has been reused by other process, the new process
+ * with the same pid will be controlled unexpected.
+ * 3. The controller cannot control the thread-level behaviour. It is assumed
+ * the target process is single-threaded. Otherwise, all threads will be
+ * controlled with the upper bound.
+ * 4. There is no guarantee that whether the controller process will run before
+ * the target monitored process. There is no CPU usage bound guaranteed before
+ * controller process gets scheduled to run.
+ */
+    class CpuUsageController {
+    public:
+        CpuUsageController(pid_t pid, double ratio);
+
+        [[noreturn]] auto start() -> void;
+
+    private:
+        pid_t m_pid;
+        std::chrono::milliseconds m_execution_time{0};
+        std::chrono::microseconds m_sleep_time{0};
+    };
+}  // namespace clp
+
+#endif

components/core/src/clp/clp/run.cpp

@@ -1,7 +1,10 @@
 #include "run.hpp"
 
+#include <memory>
 #include <unordered_set>
 
+#include <unistd.h>
+
 #include <log_surgeon/LogParser.hpp>
 #include <spdlog/sinks/stdout_sinks.h>
 
@@ -10,6 +13,7 @@
 #include "../Utils.hpp"
 #include "CommandLineArguments.hpp"
 #include "compression.hpp"
+#include "CpuUsageController.hpp"
 #include "decompression.hpp"
 #include "utils.hpp"
 
@@ -18,6 +22,18 @@ using std::unordered_set;
 using std::vector;
 
 namespace clp::clp {
+namespace {
+    class ChildProcessCleaner {
+    public:
+        ChildProcessCleaner(int pid) : m_pid{pid} {};
+
+        ~ChildProcessCleaner() { kill(m_pid, SIGTERM); }
+
+    private:
+        int m_pid;
+    };
+}
+
 int run(int argc, char const* argv[]) {
     // Program-wide initialization
     try {
@@ -55,7 +71,18 @@ int run(int argc, char const* argv[]) {
     }
 
     auto command = command_line_args.get_command();
+    std::unique_ptr<ChildProcessCleaner> child_process_cleaner;
     if (CommandLineArguments::Command::Compress == command) {
+        auto const cpu_usage_bound{command_line_args.get_cpu_usage_bound()};
+        if (cpu_usage_bound.has_value()) {
+            auto const child_pid{fork()};
+            if (0 == child_pid) {
+                CpuUsageController(getppid(), cpu_usage_bound.value()).start();
+                // Never return
+            }
+            child_process_cleaner = std::make_unique<ChildProcessCleaner>(child_pid);
+        }
+
         /// TODO: make this not a unique_ptr and test performance difference
         std::unique_ptr<log_surgeon::ReaderParser> reader_parser;
         if (!command_line_args.get_use_heuristic()) {