Rethink things a bit. Store everything in a database and use that to serialise operations

Author: dsroberts

Diff for: CMakeLists.txt

@@ -4,6 +4,8 @@ project(
     tsp_for_hpc
     LANGUAGES CXX)
 
+find_package(SQLite3 REQUIRED)
+
 set(CMAKE_CXX_STANDARD 20)
 set(CMAKE_CXX_FLAGS "${CMAKE_CXX_FLAGS} -Wall")
 
@@ -14,9 +16,10 @@ set(sources
     run_cmd.cpp
     status_manager.cpp
     output_manager.cpp
-    proc_manager.cpp
+    proc_affinity.cpp
     tsp.cpp)
 
 add_executable(tsp-hpc ${sources})
 
-target_include_directories(tsp-hpc PUBLIC .)
+target_link_libraries(tsp-hpc PUBLIC ${SQLite3_LIBRARIES})
+include_directories(. ${SQLite3_INCLUDE_DIRS})

Diff for: functions.cpp

@@ -1,7 +1,19 @@
-#include <cstdlib>
+
 #include <string>
-#include <stdexcept>
+#include <vector>
+#include <cstdint>
 #include <cstring>
+#include <sys/types.h>
+#include <unistd.h>
+#include <sstream>
+#include <numeric>
+#include <fstream>
+#include <filesystem>
+
+#include "functions.hpp"
+
+#define CGROUP_CPUSET_PATH_PREFIX "/sys/fs/cgroup/cpuset"
+#define CPUSET_FILE "/cpuset.cpus"
 
 const char *get_tmp()
 {
@@ -11,12 +23,86 @@ const char *get_tmp()
         out = "/tmp";
     }
     return out;
-}
-
+};
 void die_with_err(std::string msg, int status)
 {
     std::string out(msg);
     out.append("\nstat=" + std::to_string(status) + ", errno=" + std::to_string(errno));
     out.append(std::string("\n") + strerror(errno));
     throw std::runtime_error(out);
-}
+};
+std::vector<uint32_t> parse_cpuset_range(std::string in)
+{
+    std::stringstream ss1(in);
+    std::string token;
+    std::vector<std::uint32_t> out;
+    while (std::getline(ss1, token, ','))
+    {
+        if (token.find('-') == std::string::npos)
+        {
+            out.push_back(std::stoul(token));
+        }
+        else
+        {
+            std::stringstream ss2(token);
+            std::string starts, ends;
+            std::getline(ss2, starts, '-');
+            std::getline(ss2, ends, '-');
+            std::vector<std::uint32_t> tmp(std::stoul(ends) - std::stoul(starts) + 1);
+            std::iota(tmp.begin(), tmp.end(), std::stoul(starts));
+            out.insert(out.end(), tmp.begin(), tmp.end());
+        }
+    }
+    return out;
+};
+std::vector<std::uint32_t> get_cgroup()
+{
+    std::filesystem::path cgroup_fn(std::string("/proc/" + std::to_string(getpid()) + "/cgroup"));
+    if (!std::filesystem::exists(cgroup_fn))
+    {
+        throw std::runtime_error("Cgroup file for process " + std::to_string(getpid()) + " not found");
+    }
+    std::string line;
+    std::filesystem::path cpuset_path;
+    // get cpuset path
+    std::ifstream cgroup_file(cgroup_fn);
+    if (cgroup_file.is_open())
+    {
+        while (std::getline(cgroup_file, line))
+        {
+            std::vector<std::string> seglist;
+            std::string segment;
+            std::stringstream ss(line);
+            while (std::getline(ss, segment, ':'))
+            {
+                seglist.push_back(segment);
+            };
+            if (seglist[1] == "cpuset")
+            {
+                cpuset_path = CGROUP_CPUSET_PATH_PREFIX;
+                cpuset_path += seglist[2];
+                cpuset_path += CPUSET_FILE;
+            }
+            if (!cpuset_path.empty())
+            {
+                break;
+            }
+        }
+        cgroup_file.close();
+    }
+    else
+    {
+        throw std::runtime_error("Unable to open cgroup file " + cgroup_fn.string());
+    }
+    // read cpuset file
+    std::ifstream cpuset_file(cpuset_path);
+    if (cpuset_file.is_open())
+    {
+        std::getline(cpuset_file, line);
+        return parse_cpuset_range(line);
+    }
+    else
+    {
+        throw std::runtime_error("Unable to open cpuset file " + cpuset_path.string());
+    }
+};

Diff for: functions.hpp

@@ -1,5 +1,11 @@
 #pragma once
+
 #include <string>
+#include <vector>
+#include <cstdint>
+#include <numeric>
 
-const char* get_tmp();
-void die_with_err(std::string msg, int status);
+const char *get_tmp();
+void die_with_err(std::string msg, int status);
+std::vector<uint32_t> parse_cpuset_range(std::string in);
+std::vector<std::uint32_t> get_cgroup();

Diff for: proc_affinity.cpp

@@ -0,0 +1,121 @@
+#include <fstream>
+#include <algorithm>
+#include <numeric>
+
+#include "proc_affinity.hpp"
+#include "functions.hpp"
+#include "semaphore.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())
+    {
+        // Open cgroups file
+        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);
+
+    }
+
+    std::vector<uint32_t> Proc_affinity::bind()
+    {
+        auto sibling_pids = get_siblings();
+        std::vector<uint32_t> siblings_affinity;
+        std::vector<uint32_t> out;
+        for (auto i : sibling_pids)
+        {
+            auto tmp = get_sibling_affinity(i);
+            siblings_affinity.insert(siblings_affinity.end(), tmp.begin(), tmp.end());
+        }
+        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(),
+                            siblings_affinity.begin(), siblings_affinity.end(),
+                            std::inserter(allowed_cores, allowed_cores.begin()));
+
+        if (allowed_cores.size() < nslots)
+        {
+            // Something has beaten us to the punch, report failed and
+            // try again the next time
+            return out;
+        }
+        for (uint32_t i = 0; 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)
+        {
+            die_with_err("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;
+                }
+            }
+        }
+        return out;
+    };
+
+    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
+        try
+        {
+            for (const auto &entry : std::filesystem::directory_iterator("/proc/" + std::to_string(pid) + "/fd"))
+                if (std::filesystem::read_symlink(entry).string().find(SEMAPHORE_FILE_TEMPLATE) != std::string::npos)
+                {
+                    // Semaphore present, ignore
+                    return out;
+                }
+        }
+
+        catch (std::filesystem::filesystem_error &e)
+        {
+            // Process went away
+            return out;
+        }
+        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)
+        {
+            if (CPU_ISSET(i, &mask))
+            {
+                out.push_back(i);
+            }
+        }
+        return out;
+    };
+};

Diff for: proc_manager.hpp renamed to proc_affinity.hpp

@@ -8,28 +8,23 @@
 #include <filesystem>
 #include <sys/types.h>
 
-#define CGROUP_CPUSET_PATH_PREFIX "/sys/fs/cgroup/cpuset"
-#define CPUSET_FILE "/cpuset.cpus"
-
 namespace tsp
 {
-    class Tsp_Proc
+    class Proc_affinity
     {
     public:
-        const pid_t pid;
-        std::vector<uint32_t> allowed_cores;
-        Tsp_Proc(uint32_t nslots);
-        bool allowed_to_run();
-        void refresh_allowed_cores();
-        ~Tsp_Proc() {};
+        Proc_affinity(uint32_t nslots, pid_t pid);
+        ~Proc_affinity() {};
+        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;
         std::vector<pid_t> get_siblings();
-        std::vector<std::uint32_t> parse_cpuset_range(std::string in);
         std::vector<uint32_t> get_sibling_affinity(pid_t pid);
-        std::vector<uint32_t> get_cgroup();
+        std::vector<std::string> skip_paths = {std::to_string(pid), "self", "thread-self"};
     };
 }