Back on track. Fix locking/database sync issues. Remove unneeded class variables. More sensible sqlite usage.

dsroberts · dsroberts · commit 2d32542a6d5a · 2025-01-03T13:56:35.000+11:00
diff --git a/jitter.cpp b/jitter.cpp
@@ -5,10 +5,10 @@
 
 namespace tsp {
 Jitter::Jitter(std::chrono::milliseconds limit)
-    : rng(std::mt19937(std::random_device{}())),
-      dist(std::uniform_int_distribution<int64_t>(-abs(limit).count(),
+    : rng_(std::mt19937(std::random_device{}())),
+      dist_(std::uniform_int_distribution<int64_t>(-abs(limit).count(),
                                                   abs(limit).count())) {}
 std::chrono::milliseconds Jitter::get() {
-  return std::chrono::milliseconds(dist(rng));
+  return std::chrono::milliseconds(dist_(rng_));
 }
 } // namespace tsp
diff --git a/jitter.hpp b/jitter.hpp
@@ -11,7 +11,7 @@ class Jitter {
   std::chrono::milliseconds get();
 
 private:
-  std::mt19937 rng;
-  std::uniform_int_distribution<int64_t> dist;
+  std::mt19937 rng_;
+  std::uniform_int_distribution<int64_t> dist_;
 };
 } // namespace tsp
diff --git a/locker.cpp b/locker.cpp
@@ -12,16 +12,20 @@
 #include "functions.hpp"
 
 namespace tsp {
+
+std::array<sighandler_t, NSIG> prev_sigs;
+
 int lockfile_fd = -1;
 void handle_signal(int sig) {
   if (lockfile_fd != -1) {
     flock(lockfile_fd, LOCK_UN);
     close(lockfile_fd);
   }
+  prev_sigs[sig](sig);
 }
 
 Locker::Locker() {
-  lockfile_fd = open(lock_file_path.c_str(), O_WRONLY | O_CREAT, 0600);
+  lockfile_fd = open(lock_file_path_.c_str(), O_RDONLY | O_CREAT, 0600);
   if (lockfile_fd == -1) {
     die_with_err_errno("Unable to open lockfile", lockfile_fd);
   }
diff --git a/locker.hpp b/locker.hpp
@@ -1,24 +1,18 @@
 #pragma once
 
 #include <string>
-#include <array>
-#include <unistd.h>
-#include <signal.h>
 
 #include "functions.hpp"
 
-namespace tsp
-{
-    class Locker
-    {
-    public:
-        Locker();
-        ~Locker();
-        void lock();
-        void unlock();
+namespace tsp {
+class Locker {
+public:
+  Locker();
+  ~Locker();
+  void lock();
+  void unlock();
 
-    private:
-        const std::string lock_file_path{get_tmp() / ".affinity_lock_file.lock"};
-        std::array<sighandler_t, NSIG> prev_sigs;
-    };
-}
+private:
+  const std::string lock_file_path_{get_tmp() / ".affinity_lock_file.lock"};
+};
+} // namespace tsp
diff --git a/output_manager.cpp b/output_manager.cpp
@@ -16,49 +16,49 @@
 namespace tsp {
 Output_handler::Output_handler(bool disappear, bool separate_stderr,
                                std::string jobid, bool rw)
-    : stdout_fn{get_tmp() / (out_file_template + jobid)},
-      stderr_fn{get_tmp() / (err_file_template + jobid)}, disappear_{disappear},
+    : stdout_fn_{get_tmp() / (out_file_template + jobid)},
+      stderr_fn_{get_tmp() / (err_file_template + jobid)}, disappear_{disappear},
       separate_stderr_{separate_stderr} {}
 void Output_handler::init_pipes() {
   if (disappear_) {
-    stdout_fd = open("/dev/null", O_WRONLY | O_CREAT, 0666);
-    stderr_fd = open("/dev/null", O_WRONLY | O_CREAT, 0666);
+    stdout_fd_ = open("/dev/null", O_WRONLY | O_CREAT, 0666);
+    stderr_fd_ = open("/dev/null", O_WRONLY | O_CREAT, 0666);
   } else {
-    stdout_fd = open(stdout_fn.c_str(), O_WRONLY | O_CREAT, 0600);
+    stdout_fd_ = open(stdout_fn_.c_str(), O_WRONLY | O_CREAT, 0600);
 
     if (separate_stderr_) {
-      stderr_fd = open(stderr_fn.c_str(), O_WRONLY | O_CREAT, 0600);
+      stderr_fd_ = open(stderr_fn_.c_str(), O_WRONLY | O_CREAT, 0600);
     } else {
-      stderr_fd = stdout_fd;
+      stderr_fd_ = stdout_fd_;
     }
   }
-  dup2(stdout_fd, 1);
-  dup2(stderr_fd, 2);
+  dup2(stdout_fd_, 1);
+  dup2(stderr_fd_, 2);
 }
 
 std::pair<std::string, std::string> Output_handler::get_output() {
-  std::ifstream stdout_stream(stdout_fn);
+  std::ifstream stdout_stream(stdout_fn_);
   std::stringstream ss_out{};
   ss_out << stdout_stream.rdbuf();
   stdout_stream.close();
-  std::filesystem::remove(stdout_fn);
-  out_bufs.first = ss_out.str();
+  //std::filesystem::remove(stdout_fn);
+  out_bufs_.first = ss_out.str();
 
-  std::ifstream stderr_stream(stderr_fn);
+  std::ifstream stderr_stream(stderr_fn_);
   std::stringstream ss_err{};
   ss_err << stdout_stream.rdbuf();
   stderr_stream.close();
-  std::filesystem::remove(stderr_fn);
-  out_bufs.second = ss_err.str();
+  std::filesystem::remove(stderr_fn_);
+  out_bufs_.second = ss_err.str();
 
-  return out_bufs;
+  return out_bufs_;
 }
 Output_handler::~Output_handler() {
-  if (stdout_fd != -1) {
-    close(stdout_fd);
+  if (stdout_fd_ != -1) {
+    close(stdout_fd_);
   }
-  if (stderr_fd != -1) {
-    close(stderr_fd);
+  if (stderr_fd_ != -1) {
+    close(stderr_fd_);
   }
 }
 } // namespace tsp
diff --git a/output_manager.hpp b/output_manager.hpp
@@ -7,19 +7,19 @@ constexpr std::string err_file_template{"tsp.e"};
 namespace tsp {
 class Output_handler {
 public:
-  int stdout_fd;
-  int stderr_fd;
   Output_handler(bool disappear, bool separate_stderr, std::string jobid,
                  bool rw);
   void init_pipes();
   std::pair<std::string, std::string> get_output();
   ~Output_handler();
 
 private:
-  std::string stdout_fn{};
-  std::string stderr_fn{};
+  std::string stdout_fn_{};
+  std::string stderr_fn_{};
+  int stdout_fd_;
+  int stderr_fd_;
   bool disappear_;
   bool separate_stderr_;
-  std::pair<std::string, std::string> out_bufs{};
+  std::pair<std::string, std::string> out_bufs_{};
 };
 } // namespace tsp
diff --git a/proc_affinity.cpp b/proc_affinity.cpp
@@ -5,20 +5,21 @@
 #include <numeric>
 
 #include "functions.hpp"
+#include "status_manager.hpp"
 
 namespace tsp {
 
-Proc_affinity::Proc_affinity(uint32_t nslots, pid_t pid)
-    : nslots(nslots), pid(pid),
-      my_path(std::filesystem::read_symlink("/proc/self/exe")),
-      cpuset_from_cgroup(get_cgroup()) {
+Proc_affinity::Proc_affinity(Status_Manager &sm, uint32_t nslots, pid_t pid)
+    : sm_(sm), nslots_(nslots), pid_(pid),
+      my_path_(std::filesystem::read_symlink("/proc/self/exe")),
+      cpuset_from_cgroup_(get_cgroup()) {
   // Open cgroups file
-  if (nslots > cpuset_from_cgroup.size()) {
+  if (nslots > cpuset_from_cgroup_.size()) {
     die_with_err("More slots requested than available on the system, this "
                  "process can never run.",
                  -1);
   }
-  CPU_ZERO(&mask);
+  CPU_ZERO(&mask_);
 }
 
 std::vector<uint32_t> Proc_affinity::bind() {
@@ -31,52 +32,42 @@ std::vector<uint32_t> Proc_affinity::bind() {
   }
   std::sort(siblings_affinity.begin(), siblings_affinity.end());
   std::vector<uint32_t> allowed_cores;
-  std::set_difference(cpuset_from_cgroup.begin(), cpuset_from_cgroup.end(),
+  std::set_difference(cpuset_from_cgroup_.begin(), cpuset_from_cgroup_.end(),
                       siblings_affinity.begin(), siblings_affinity.end(),
                       std::inserter(allowed_cores, allowed_cores.begin()));
 
-  for (auto i = 0ul; i < nslots; i++) {
-    CPU_SET(allowed_cores[i], &mask);
+  for (auto i = 0ul; i < nslots_; i++) {
+    CPU_SET(allowed_cores[i], &mask_);
     out.push_back(allowed_cores[i]);
   }
-  if (sched_setaffinity(0, sizeof(cpu_set_t), &mask) == -1) {
+  if (sched_setaffinity(0, sizeof(cpu_set_t), &mask_) == -1) {
     die_with_err_errno("Unable to set CPU affinity", -1);
   }
   return out;
 }
 
 std::vector<pid_t> Proc_affinity::get_siblings() {
-  std::vector<pid_t> out;
-  // Find all the other versions of this application running
-  for (const auto &entry : std::filesystem::directory_iterator("/proc")) {
-    if (std::find(skip_paths.begin(), skip_paths.end(),
-                  entry.path().filename()) != skip_paths.end()) {
-      continue;
-    }
-    if (std::filesystem::exists(entry.path() / "exe")) {
-      try {
-        if (std::filesystem::read_symlink(entry.path() / "exe") == my_path) {
-          out.push_back(std::stoul(entry.path().filename()));
-        }
-      } catch (std::filesystem::filesystem_error &e) {
-        // process went away
-        continue;
-      }
+  auto all_tsp_pids = sm_.get_running_job_pids();
+  for (std::vector<pid_t>::iterator it = all_tsp_pids.begin();
+       it != all_tsp_pids.end();) {
+    if (*it == pid_) {
+      all_tsp_pids.erase(it);
+    } else {
+      it++;
     }
   }
-  return out;
+  return all_tsp_pids;
 };
 
 std::vector<uint32_t> Proc_affinity::get_sibling_affinity(pid_t pid) {
   std::vector<uint32_t> out;
   cpu_set_t mask;
-  // Just return an empty vector if the semaphore file is present
   if (sched_getaffinity(pid, sizeof(mask), &mask) == -1) {
     // Process may have been killed - so it isn't taking
     // resources any more
     return out;
   }
-  for (const auto &i : cpuset_from_cgroup) {
+  for (const auto &i : cpuset_from_cgroup_) {
     if (CPU_ISSET(i, &mask)) {
       out.push_back(i);
     }
diff --git a/proc_affinity.hpp b/proc_affinity.hpp
@@ -8,21 +8,22 @@
 #include <unistd.h>
 #include <vector>
 
+#include "status_manager.hpp"
+
 namespace tsp {
 class Proc_affinity {
 public:
-  Proc_affinity(uint32_t nslots, pid_t pid);
+  Proc_affinity(Status_Manager &sm, uint32_t nslots, pid_t pid);
   std::vector<uint32_t> bind();
 
 private:
-  const uint32_t nslots;
-  const pid_t pid;
-  const std::filesystem::path my_path;
-  const std::vector<uint32_t> cpuset_from_cgroup;
-  cpu_set_t mask;
+  Status_Manager &sm_;
+  const uint32_t nslots_;
+  const pid_t pid_;
+  const std::filesystem::path my_path_;
+  const std::vector<uint32_t> cpuset_from_cgroup_;
+  cpu_set_t mask_;
   std::vector<pid_t> get_siblings();
   std::vector<uint32_t> get_sibling_affinity(pid_t pid);
-  std::vector<std::string> skip_paths = {std::to_string(pid), "self",
-                                         "thread-self"};
 };
 } // namespace tsp
diff --git a/run_cmd.cpp b/run_cmd.cpp
@@ -4,6 +4,7 @@
 #include <cstring>
 #include <filesystem>
 #include <fstream>
+#include <ranges>
 #include <sys/types.h>
 #include <sys/wait.h>
 #include <unistd.h>
@@ -12,38 +13,57 @@
 
 namespace tsp {
 Run_cmd::Run_cmd(char *cmdline[], int start, int end)
-    : is_openmpi(check_mpi(cmdline[0])), rf_name(get_tmp()) {
+    : is_openmpi(check_mpi(cmdline[start])) {
   for (int i = start; i < end; i++) {
-    proc_to_run.push_back(cmdline[i]);
+    proc_to_run_.emplace_back(cmdline[i]);
   }
-  proc_to_run.push_back(nullptr);
 }
+
 Run_cmd::~Run_cmd() {
   if (is_openmpi) {
-    if (!rf_name.empty()) {
-      std::filesystem::remove(rf_name);
+    if (!rf_name_.empty()) {
+      std::filesystem::remove(rf_name_);
+    }
+    if (argv_holder_ != nullptr) {
+      free(argv_holder_);
     }
   }
 }
 
 std::string Run_cmd::print() {
   std::stringstream out;
-  for (const auto &i : proc_to_run) {
+  for (const auto &i : proc_to_run_) {
     out << i << " ";
   }
   return out.str();
 }
 
+const char *Run_cmd::get_argv_0() { return proc_to_run_[0].c_str(); }
+
+char **Run_cmd::get_argv() {
+  // Do it the old fashioned way
+  if (argv_holder_ == nullptr) {
+    if (nullptr == (argv_holder_ = static_cast<char **>(
+                        malloc((proc_to_run_.size() + 1) * sizeof(char *))))) {
+      die_with_err_errno("Malloc failed", -1);
+    }
+    for (const auto &[i, p] : std::views::enumerate(proc_to_run_)) {
+      argv_holder_[i] = p.data();
+    }
+    argv_holder_[proc_to_run_.size()] = nullptr;
+  }
+  return argv_holder_;
+}
+
 void Run_cmd::add_rankfile(std::vector<uint32_t> procs, uint32_t nslots) {
   make_rankfile(procs, nslots);
-  proc_to_run.insert(proc_to_run.begin() + 1, rf_name.string().data());
-  proc_to_run.insert(proc_to_run.begin() + 1, std::string("rf").data());
-  proc_to_run.push_back(nullptr);
+  proc_to_run_.emplace(proc_to_run_.begin() + 1, rf_name_);
+  proc_to_run_.emplace(proc_to_run_.begin() + 1, "-rf");
 }
 
 void Run_cmd::make_rankfile(std::vector<uint32_t> procs, uint32_t nslots) {
-  rf_name /= std::to_string(getpid()) + "_rankfile.txt";
-  std::ofstream rf_stream(rf_name);
+  rf_name_ = get_tmp() / (std::to_string(getpid()) + "_rankfile.txt");
+  std::ofstream rf_stream(rf_name_);
   if (rf_stream.is_open()) {
     for (uint32_t i = 0; i < nslots; i++) {
       rf_stream << "rank " + std::to_string(i) +
@@ -55,7 +75,11 @@ void Run_cmd::make_rankfile(std::vector<uint32_t> procs, uint32_t nslots) {
 }
 bool Run_cmd::check_mpi(const char *exe_name) {
   std::string prog_name(exe_name);
-  if (prog_name == "mpirun" || prog_name == "mpiexec") {
+  std::string prog_test(prog_name);
+  if (prog_name.starts_with('/')) {
+    prog_test = std::filesystem::path(prog_name).filename().string();
+  }
+  if (prog_test == "mpirun" || prog_test == "mpiexec") {
     // OpenMPI does not respect parent process binding,
     // so we need to check if we're attempting to run
     // OpenMPI, and if we are, we need to construct a
diff --git a/run_cmd.hpp b/run_cmd.hpp
diff --git a/status_manager.cpp b/status_manager.cpp
diff --git a/status_manager.hpp b/status_manager.hpp
diff --git a/tsp.cpp b/tsp.cpp
