Feature/thread utils

examples/CMakeLists.txt

@@ -10,3 +10,9 @@ FetchContent_Declare(
 FetchContent_MakeAvailable(spdlog)
 
 add_subdirectory(shared_memory)
+
+if (WIN32)
+
+else()
+add_subdirectory(linux_deadline)
+endif()

examples/linux_deadline/CMakeLists.txt

@@ -0,0 +1,22 @@
+cmake_minimum_required(VERSION 3.14)
+project(example_linux_deadline  VERSION 1.0.0 LANGUAGES CXX)
+
+
+if(NOT TARGET CPPUTILS2::cpputils2)
+    message("Not target!")
+    find_package(CPPUTILS2 REQUIRED)
+
+    Include(FetchContent)
+FetchContent_Declare(
+  spdlog
+  GIT_REPOSITORY https://github.com/gabime/spdlog.git
+  GIT_TAG        v1.x
+)
+
+FetchContent_MakeAvailable(spdlog)
+endif()
+
+
+add_executable(linux_deadline src/main.cpp)
+
+target_link_libraries(linux_deadline PRIVATE CPPUTILS2::cpputils2 spdlog::spdlog)

examples/linux_deadline/src/main.cpp

@@ -0,0 +1,84 @@
+
+#include <cpputils2/cpputils2.hpp>
+#include <cpputils2/linux/sched/sched.hpp>
+
+#include <atomic>
+#include <chrono>
+#include <csignal>
+#include <iostream>
+#include <spdlog/spdlog.h>
+
+#include <linux/sched.h>
+
+const int32_t expected = 42;
+const std::string file_name("test_shm");
+
+std::atomic_flag run_thread = ATOMIC_FLAG_INIT;
+
+void signalHandler(int signum)
+{
+  if (signum == SIGXCPU)
+  {
+    SPDLOG_INFO("Caught signal SIGXCPU. We are overruning the deadline");
+    run_thread.clear();
+  }
+  else if (signum == SIGINT)
+  {
+    run_thread.clear();
+  }
+}
+
+int main(int argc, char **argv)
+{
+  SPDLOG_INFO("cpputils2 version {}", CppUtils2::VERSION);
+  SPDLOG_INFO("Test deadline Linux scheduler. Ctrl-C to finish test.");
+
+  CppUtils2::sched_attr attr;
+  attr.size = sizeof(CppUtils2::sched_attr);
+  attr.sched_policy = SCHED_DEADLINE;
+  attr.sched_priority = 0;
+
+  attr.sched_runtime = 200'000'000;
+  attr.sched_deadline = 500'000'000;
+  attr.sched_period = 500'000'000; // 1s
+  attr.sched_flags = SCHED_FLAG_RESET_ON_FORK | SCHED_FLAG_RECLAIM | SCHED_FLAG_DL_OVERRUN;
+  signal(SIGXCPU, signalHandler);
+  signal(SIGINT, signalHandler);
+
+  run_thread.test_and_set();
+
+  auto ret = CppUtils2::set_self_attributes(&attr);
+
+  if (!ret.has_value())
+  {
+    auto error = ret.error();
+    SPDLOG_ERROR("Error setting attributes. Error code: {}, errno: {}", error, errno);
+    return -1;
+  }
+
+  auto start = std::chrono::high_resolution_clock::now();
+
+  bool first = true;
+  uint64_t mean_time_ns = 0;
+  uint64_t count = 0;
+
+  while (run_thread.test())
+  {
+    if (first)
+    {
+      first = false;
+    }
+    else
+    {
+      auto elapsed = std::chrono::duration_cast<std::chrono::nanoseconds>(std::chrono::high_resolution_clock::now() - start).count();
+      std::cout << "Elapsed: " << elapsed << "ns\n";
+      start = std::chrono::high_resolution_clock::now();
+      mean_time_ns += elapsed;
+      count++;
+    }
+    sched_yield();
+  }
+
+  SPDLOG_INFO("Exiting. Mean activation time {}ns", mean_time_ns / count);
+  return 0;
+}

src/CMakeLists.txt

@@ -48,6 +48,8 @@ else()
     include/cpputils2/linux/futex/futex.hpp
     include/cpputils2/linux/futex/shared_futex.hpp
     include/cpputils2/linux/thread/thread.hpp
+    include/cpputils2/linux/sched/sched.hpp
+    include/cpputils2/linux/priomutex/priomutex.hpp
     )
 
 endif()

src/include/cpputils2/linux/priomutex/priomutex.hpp

@@ -0,0 +1,89 @@
+#pragma once
+
+#include "cpputils2/common/types.hpp"
+
+#include <pthread.h>
+
+namespace CppUtils2
+{
+
+    class PrioMutex
+    {
+
+    public:
+        using native_handle_type = pthread_mutex_t *;
+
+        PrioMutex()
+        {
+            initialized = false;
+            pthread_mutexattr_t attr;
+
+            int ret = pthread_mutexattr_init(&attr);
+            if (ret != 0)
+            {
+                return;
+            }
+
+            ret = pthread_mutexattr_setprotocol(&attr, PTHREAD_PRIO_INHERIT);
+            if (ret != 0)
+            {
+                return;
+            }
+
+            ret = pthread_mutex_init(&pt_mutex, &attr);
+            if (ret != 0)
+            {
+                return;
+            }
+
+            initialized = true;
+        }
+
+        virtual ~PrioMutex()
+        {
+            pthread_mutex_destroy(&pt_mutex);
+        }
+
+        PrioMutex(const PrioMutex &) = delete;
+        PrioMutex &operator=(const PrioMutex &) = delete;
+
+        void lock()
+        {
+            auto ret = pthread_mutex_lock(&pt_mutex);
+            if (ret != 0)
+            {
+                // (yangosoft) TODO handle error
+                initialized = false;
+            }
+        }
+
+        void unlock() noexcept
+        {
+            pthread_mutex_unlock(&pt_mutex);
+        }
+
+        bool try_lock() noexcept
+        {
+            return pthread_mutex_trylock(&pt_mutex) == 0;
+        }
+
+        native_handle_type native_handle() noexcept
+        {
+            return &pt_mutex;
+        }
+
+        bool is_lock_free() const noexcept
+        {
+            return false;
+        }
+
+        bool is_initialized() const noexcept
+        {
+            return initialized;
+        }
+
+    private:
+        pthread_mutex_t pt_mutex;
+        bool initialized;
+    };
+} // namespace CppUtils2

src/include/cpputils2/linux/sched/sched.hpp

@@ -0,0 +1,94 @@
+#pragma once
+
+#include "cpputils2/common/types.hpp"
+
+#include <cassert>
+#include <cstdint>
+#include <expected>
+#include <pthread.h>
+#include <sched.h>
+#include <sys/syscall.h>
+#include <unistd.h>
+
+#include <span>
+
+namespace CppUtils2
+{
+
+    // (yangosoft) As man 2 page says, this is a generic structure for scheduling attributes
+    struct sched_attr
+    {
+        uint32_t size;               /* Size of this structure */
+        uint32_t sched_policy = 0;   /* Policy (SCHED_*) */
+        uint64_t sched_flags = 0;    /* Flags */
+        int32_t sched_nice = 0;      /* Nice value (SCHED_OTHER,
+                                  SCHED_BATCH) */
+        uint32_t sched_priority = 0; /* Static priority (SCHED_FIFO,
+                               SCHED_RR) */
+        /* Remaining fields are for SCHED_DEADLINE */
+        uint64_t sched_runtime = 0;
+        uint64_t sched_deadline = 0;
+        uint64_t sched_period = 0;
+
+        sched_attr() : size(sizeof(sched_attr)) {}
+    };
+
+    int32_t sched_getattr(pid_t pid, sched_attr *attr, unsigned int size, unsigned int flags = 0)
+    {
+        return syscall(SYS_sched_getattr, pid, attr, size, flags);
+    }
+
+    int32_t sched_setattr(pid_t pid, const sched_attr *attr, uint32_t flags = 0)
+    {
+        return syscall(SYS_sched_setattr, pid, attr, flags);
+    }
+
+    std::expected<Result, int32_t> set_self_attributes(const sched_attr *attr)
+    {
+        pid_t me = getpid();
+        int flags = 0;
+
+        int32_t ret = sched_setattr(me, attr, flags);
+
+        if (ret != 0)
+        {
+            return std::unexpected(ret);
+        }
+
+        return Result::RET_OK;
+    }
+
+    std::expected<Result, int32_t> set_process_core_affinity(pid_t pid, const cpu_set_t *mask)
+    {
+        int ret = sched_setaffinity(pid, sizeof(cpu_set_t), mask);
+        if (ret != 0)
+        {
+            return std::unexpected(errno);
+        }
+        return Result::RET_OK;
+    }
+
+    std::expected<Result, int32_t> set_self_core_affinity(const cpu_set_t *mask)
+    {
+        return set_process_core_affinity(getpid(), mask);
+    }
+
+    std::expected<Result, int32_t> set_process_core_affinity(pid_t pid, const std::span<uint32_t, std::dynamic_extent> &mask)
+    {
+
+        cpu_set_t cpuset;
+        CPU_ZERO(&cpuset);
+        for (auto core : mask)
+        {
+            CPU_SET(core, &cpuset);
+        }
+
+        return set_process_core_affinity(pid, &cpuset);
+    }
+
+    std::expected<Result, int32_t> set_self_core_affinity(const std::span<uint32_t, std::dynamic_extent> &mask)
+    {
+        return set_process_core_affinity(getpid(), mask);
+    }
+
+} // namespace CppUtils2

src/include/cpputils2/linux/thread/thread.hpp

@@ -94,4 +94,16 @@ namespace CppUtils2
         return config;
     }
 
+    int32_t pin_thread_to_core(std::thread &thread, const int core_id)
+    {
+        cpu_set_t cpuset;
+        CPU_ZERO(&cpuset);
+        CPU_SET(core_id, &cpuset);
+
+        // ensure that native_handler() is a pthread_t
+        assert(typeid(thread.native_handle()) == typeid(pthread_t));
+
+        return pthread_setaffinity_np(thread.native_handle(), sizeof(cpu_set_t), &cpuset);
+    }
+
 }

src/test/cpputils2_tests.cpp

@@ -14,6 +14,8 @@
 #include "cpputils2/linux/net/socket/udpsocketserver.hpp"
 #include "cpputils2/linux/net/socket/udsclient.hpp"
 #include "cpputils2/linux/net/socket/udsserver.hpp"
+#include "cpputils2/linux/priomutex/priomutex.hpp"
+#include "cpputils2/linux/sched/sched.hpp"
 #include "cpputils2/linux/shm/shm.hpp"
 #include "cpputils2/linux/thread/thread.hpp"
 #endif
@@ -162,6 +164,20 @@ namespace
     t.join();
   }
 
+  TEST(Scheduler, Scheduler)
+  {
+    EXPECT_EQ(CppUtils2::Result::RET_OK, CppUtils2::Result::RET_OK);
+  }
+
+  TEST(PriorityMutex, PriorityMutex)
+  {
+    CppUtils2::PrioMutex mutex;
+    bool is_initialized = mutex.is_initialized();
+    EXPECT_TRUE(is_initialized);
+    mutex.lock();
+    mutex.unlock();
+  }
+
 #endif
 
 #ifdef _WIN32