async_lock_lf

Author: esufaev

include/pot/sync/async_lock.h

@@ -2,15 +2,20 @@
 
 #include <atomic>
 #include <coroutine>
-#include <mutex>
-#include <queue>
+#include <optional>
+#include <thread>
+#include <utility>
+
+#include "pot/executors/executor.h"
+#include "pot/algorithms/lfqueue.h" 
 
 namespace pot::sync
 {
+  class async_lock
+  {
+    public:
+        using queue_type = std::pair<std::coroutine_handle<>, pot::executor*>;
 
-class async_lock
-{
-  public:
     class scoped_lock_guard
     {
       public:
@@ -48,62 +53,59 @@ class async_lock
 
     struct lock_awaiter
     {
-        async_lock &m_lock;
-
-        bool await_ready() const noexcept
-        {
-            bool expected = false;
-            return m_lock.m_is_locked.compare_exchange_strong(expected, true,
-                                                              std::memory_order_acquire);
-        }
+      async_lock& m_lock;
+      pot::executor* m_executor;
 
-        bool await_suspend(std::coroutine_handle<> h)
-        {
-            std::lock_guard guard(m_lock.m_queue_mutex);
+      bool await_ready() const noexcept { return false; }
 
-            bool expected = false;
-            if (m_lock.m_is_locked.compare_exchange_strong(expected, true,
-                                                           std::memory_order_acquire))
-            {
-                return false;
-            }
+      bool await_suspend(std::coroutine_handle<> handle)
+      {
+          int prev_state = m_lock.m_state.fetch_add(1, std::memory_order_acq_rel);
+          if (prev_state == 0)
+          {
+              return false; 
+          }
 
-            m_lock.m_waiters.push(h);
-            return true;
-        }
+          while (!m_lock.m_queue.push({handle, m_executor})) std::this_thread::yield();
+          return true; 
+      }
 
-        scoped_lock_guard await_resume() noexcept { return scoped_lock_guard{m_lock}; }
+      scoped_lock_guard await_resume() noexcept { return scoped_lock_guard(m_lock); } 
     };
 
-    [[nodiscard]] lock_awaiter lock() { return lock_awaiter{*this}; }
+    [[nodiscard]] lock_awaiter lock() 
+    {
+      return lock_awaiter{*this, nullptr};
+    }
+
+    [[nodiscard]] lock_awaiter lock(pot::executor* executor)
+    {
+      return lock_awaiter{*this, executor};
+    }
+
+    private:
+      std::atomic<int> m_state{0};      
+      pot::algorithms::lfqueue<queue_type> m_queue;
 
-  private:
     void unlock()
     {
-        std::coroutine_handle<> next_handle = nullptr;
+        int prev_state = m_state.fetch_sub(1, std::memory_order_acq_rel);
+        if (prev_state == 1) return;
 
+        std::optional<queue_type> next_task;
+        
+        while (!(next_task = m_queue.pop())) std::this_thread::yield();
+
+        auto [handle, executor] = *next_task;
+
+        if (executor)
         {
-            std::lock_guard guard(m_queue_mutex);
-            if (m_waiters.empty())
-            {
-                m_is_locked.store(false, std::memory_order_release);
-            }
-            else
-            {
-                next_handle = m_waiters.front();
-                m_waiters.pop();
-            }
+            executor->run_detached([h = handle]() mutable { h.resume(); });
         }
-
-        if (next_handle)
+        else
         {
-            next_handle.resume();
+            handle.resume();
         }
     }
-
-    std::atomic<bool> m_is_locked{false};
-    std::mutex m_queue_mutex;
-    std::queue<std::coroutine_handle<>> m_waiters;
-};
-
-} // namespace pot::sync
+  };
+}

test/CMakeLists.txt

@@ -4,20 +4,24 @@ include(${CMAKE_SOURCE_DIR}/cmake/Catch2.cmake)
 # Create test executable
 set(TEST_SOURCES
   # test_async_lock.cpp
+  test_async_lock_lf.cpp
   # test_task.cpp
   # test_parfor.cpp
   # test_executor.cpp
   # test_LU.cpp
-  test_fill.cpp
+  # test_fill.cpp
   # test_thread.cpp
   # test_resume_on.cpp
+  # test_exp.cpp
+  # test_mandelbrot.cpp
+  # test_disb.cpp
 )
 
-find_package(Eigen3 3.3 REQUIRED NO_MODULE)
+find_package(Eigen3 REQUIRED)
 
 add_executable(tests ${TEST_SOURCES})
 target_link_libraries(tests PRIVATE pot Catch2::Catch2WithMain)
-target_link_libraries(tests PRIVATE pot Eigen3::Eigen)
+target_link_libraries(tests PRIVATE Eigen3::Eigen)
 target_link_libraries(tests PRIVATE pot concurrencpp::concurrencpp)
 target_link_libraries(tests PRIVATE pot fmt::fmt)
 target_link_libraries(tests PRIVATE pot OpenMP::OpenMP_CXX)

test/test_async_lock_lf.cpp

@@ -0,0 +1,100 @@
+#define CATCH_CONFIG_MAIN
+#include <catch2/catch_test_macros.hpp>
+
+#include <Eigen/Sparse>
+#include <fmt/core.h>
+#include <memory>
+#include <vector>
+#include <chrono>
+
+
+#include "pot/algorithms/parfor.h"
+#include "pot/executors/exp/thread_pool_executor_exp.h" 
+#include "pot/sync/async_lock.h"
+#include "pot/sync/async_lock_m.h"
+#include "pot/utils/time_it.h"
+
+
+std::tuple<double, double, double> calculation(const int64_t row_idx)
+{
+    double vl = 0.0, vc = 0.0, vr = 0.0;
+    for (auto i = 0; i < 100; ++i)
+    {
+        double val = static_cast<double>(row_idx) * 10.0;
+        vl += -std::exp(std::sin(std::cos(val))) - std::sin(10);
+        vc += std::exp(std::sin(std::cos(val)));
+        vr += -std::exp(std::sin(std::cos(val))) + std::sin(10);
+    }
+    return {vl, vc, vr};
+}
+
+
+template <typename Executor, typename LockType> 
+void buildA_pot(const int64_t n, std::shared_ptr<Executor> executor)
+{
+    std::pmr::synchronized_pool_resource pool(std::pmr::new_delete_resource());
+    pot::memory::set_memory_resource(&pool);
+
+    Eigen::SparseMatrix<double> A(n, n);
+    A.resize(n, n);
+
+    LockType locker; 
+
+    pot::algorithms::parfor(*executor, (int64_t)0, n,
+                            [&](int64_t i) -> pot::coroutines::task<void>
+                            {
+                                auto [vl, vc, vr] = calculation(i);
+                                auto _ = co_await locker.lock(&*executor);
+
+                                if (i > 0)
+                                    A.coeffRef(i, i - 1) = vl;
+                                A.coeffRef(i, i) = vc;
+                                if (i < n - 1)
+                                    A.coeffRef(i, i + 1) = vr;
+                            })
+        .get();
+}
+
+TEST_CASE("Async Lock Performance: Std Mutex vs Lock-Free", "[benchmark]")
+{
+    
+    const std::vector<int64_t> thread_counts = {8};
+    const std::vector<int64_t> sizes = {10'000, 20'000, 30'000, 40'000, 50'000, 60'000};
+    const size_t test_count = 3;
+
+    SECTION("Compare locks on GQ Executor")
+    {
+        
+        fmt::print("{:>7} {:>7} | {:>12} {:>12}\n", 
+                   "Threads", "Size", "Std Mutex", "Lock-Free");
+        fmt::print("{:-<45}\n", "");
+
+        for (auto thread_count : thread_counts)
+        {
+            
+            auto exec_gq = std::make_shared<pot::executors::thread_pool_executor_gq>("GQ", thread_count);
+
+            for (auto n : sizes)
+            {
+                
+                auto time_lock_std = pot::utils::time_it<std::chrono::duration<double>>(
+                    test_count, 
+                    []() {}, 
+                    buildA_pot<pot::executors::thread_pool_executor_gq, pot::sync::async_lock_opt>, 
+                    n, exec_gq);
+
+                
+                auto time_lock_lf = pot::utils::time_it<std::chrono::duration<double>>(
+                    test_count, 
+                    []() {}, 
+                    buildA_pot<pot::executors::thread_pool_executor_gq, pot::sync::async_lock_lf>, 
+                    n, exec_gq);
+
+                
+                fmt::print("{:7} {:7} | {:12.5f} {:12.5f}\n",
+                           thread_count, n, 
+                           time_lock_std.count(), time_lock_lf.count());
+            }
+        }
+    }
+}