Move threadpool under xss:tp:: namespace

Author: Raghuveer Devulapalli

src/avx512-16bit-qsort.hpp

@@ -564,7 +564,7 @@ avx512_qsort_fp16_helper(uint16_t *arr, arrsize_t arrsize)
         arrsize_t task_threshold = std::max((arrsize_t)100000, arrsize / 100);
 
         // Create a thread pool
-        ThreadPool pool(thread_count);
+        xss::tp::ThreadPool pool(thread_count);
 
         // Initial sort task
         qsort_threads<vtype, comparator, T>(arr,

src/xss-common-qsort.h

@@ -588,7 +588,7 @@ static void qsort_threads(type_t *arr,
                           arrsize_t right,
                           arrsize_t max_iters,
                           arrsize_t task_threshold,
-                          ThreadPool &thread_pool)
+                          xss::tp::ThreadPool &thread_pool)
 {
     /*
      * Resort to std::sort if quicksort isn't making any progress
@@ -629,20 +629,21 @@ static void qsort_threads(type_t *arr,
     if (pivot != leftmostValue) {
         bool parallel_left = (pivot_index - left) > task_threshold;
         if (parallel_left) {
-            submit_task(thread_pool,
-                        [arr,
-                         left,
-                         pivot_index,
-                         max_iters,
-                         task_threshold,
-                         &thread_pool]() {
-                            qsort_threads<vtype, comparator>(arr,
-                                                             left,
-                                                             pivot_index - 1,
-                                                             max_iters - 1,
-                                                             task_threshold,
-                                                             thread_pool);
-                        });
+            xss::tp::submit_task(thread_pool,
+                                 [arr,
+                                  left,
+                                  pivot_index,
+                                  max_iters,
+                                  task_threshold,
+                                  &thread_pool]() {
+                                     qsort_threads<vtype, comparator>(
+                                             arr,
+                                             left,
+                                             pivot_index - 1,
+                                             max_iters - 1,
+                                             task_threshold,
+                                             thread_pool);
+                                 });
         }
         else {
             qsort_threads<vtype, comparator>(arr,
@@ -658,20 +659,21 @@ static void qsort_threads(type_t *arr,
     if (pivot != rightmostValue) {
         bool parallel_right = (right - pivot_index) > task_threshold;
         if (parallel_right) {
-            submit_task(thread_pool,
-                        [arr,
-                         pivot_index,
-                         right,
-                         max_iters,
-                         task_threshold,
-                         &thread_pool]() {
-                            qsort_threads<vtype, comparator>(arr,
-                                                             pivot_index,
-                                                             right,
-                                                             max_iters - 1,
-                                                             task_threshold,
-                                                             thread_pool);
-                        });
+            xss::tp::submit_task(thread_pool,
+                                 [arr,
+                                  pivot_index,
+                                  right,
+                                  max_iters,
+                                  task_threshold,
+                                  &thread_pool]() {
+                                     qsort_threads<vtype, comparator>(
+                                             arr,
+                                             pivot_index,
+                                             right,
+                                             max_iters - 1,
+                                             task_threshold,
+                                             thread_pool);
+                                 });
         }
         else {
             qsort_threads<vtype, comparator>(arr,
@@ -764,7 +766,7 @@ X86_SIMD_SORT_INLINE void xss_qsort(T *arr, arrsize_t arrsize, bool hasnan)
                     = std::max((arrsize_t)100000, arrsize / 100);
 
             // Create a thread pool
-            ThreadPool pool(thread_count);
+            xss::tp::ThreadPool pool(thread_count);
 
             // Initial sort task
             qsort_threads<vtype, comparator, T>(arr,

src/xss-thread-pool.hpp

@@ -16,7 +16,10 @@
 #include <vector>
 #include <atomic>
 
-/*
+namespace xss {
+namespace tp {
+
+    /*
  * ThreadPool class and doc: Generated by copilot
  * This thread pool implementation is a simple and efficient way to manage a
  * pool of threads for executing tasks concurrently. It uses a std::queue to store
@@ -25,102 +28,116 @@
  * execute it. The thread pool can be stopped gracefully, and it also provides
  * a way to wait for all tasks to complete before stopping.
  * */
-class ThreadPool {
-private:
-    std::vector<std::thread> workers;
-    std::queue<std::function<void()>> tasks;
-    std::mutex queue_mutex;
-    std::condition_variable condition; // Condition variable for task queue
-    std::condition_variable done_condition; // Condition variable for waiting
-    int active_tasks {0};
-    bool stop;
+    class ThreadPool {
+    private:
+        std::vector<std::thread> workers;
+        std::queue<std::function<void()>> tasks;
+        std::mutex queue_mutex;
+        std::condition_variable condition; // Condition variable for task queue
+        std::condition_variable
+                done_condition; // Condition variable for waiting
+        int active_tasks {0};
+        bool stop;
 
-public:
-    ThreadPool(size_t num_threads) : stop(false)
-    {
-        for (size_t i = 0; i < num_threads; ++i) {
-            // Create a worker thread and add it to the pool
-            // Each thread will run a lambda function that waits for tasks
-            workers.emplace_back([this] {
-                while (true) {
-                    // Lock the queue mutex and wait for a task to be available
-                    std::unique_lock<std::mutex> lock(queue_mutex);
-                    // Wait until there is a task or the pool is stopped
-                    condition.wait(lock,
-                                   [this] { return stop || !tasks.empty(); });
+    public:
+        ThreadPool(size_t num_threads) : stop(false)
+        {
+            for (size_t i = 0; i < num_threads; ++i) {
+                // Create a worker thread and add it to the pool
+                // Each thread will run a lambda function that waits for tasks
+                workers.emplace_back([this] {
+                    while (true) {
+                        // Lock the queue mutex and wait for a task to be available
+                        std::unique_lock<std::mutex> lock(queue_mutex);
+                        // Wait until there is a task or the pool is stopped
+                        condition.wait(lock, [this] {
+                            return stop || !tasks.empty();
+                        });
 
-                    // Check if we need to terminate the thread
-                    if (stop && tasks.empty()) { return; }
+                        // Check if we need to terminate the thread
+                        if (stop && tasks.empty()) { return; }
 
-                    // Extract the next task from the queue
-                    auto task = std::move(tasks.front());
-                    tasks.pop();
-                    // Unlock the mutex before executing the task
-                    lock.unlock();
-                    // Execute the task:
-                    task();
-                }
-            });
+                        // Extract the next task from the queue
+                        auto task = std::move(tasks.front());
+                        tasks.pop();
+                        // Unlock the mutex before executing the task
+                        lock.unlock();
+                        // Execute the task:
+                        task();
+                    }
+                });
+            }
         }
-    }
 
-    template <class F>
-    void enqueue(F &&func)
-    {
-        // Add a new task to the queue and notify one of the worker threads
-        std::unique_lock<std::mutex> lock(queue_mutex);
-        tasks.emplace(std::forward<F>(func));
-        condition.notify_one();
-    }
+        template <class F>
+        void enqueue(F &&func)
+        {
+            // Add a new task to the queue and notify one of the worker threads
+            std::unique_lock<std::mutex> lock(queue_mutex);
+            tasks.emplace(std::forward<F>(func));
+            condition.notify_one();
+        }
 
-    ~ThreadPool()
-    {
-        // Stop the thread pool and join all threads
-        std::unique_lock<std::mutex> lock(queue_mutex);
-        stop = true;
-        lock.unlock();
-        condition.notify_all();
-        for (std::thread &worker : workers) {
-            worker.join();
+        ~ThreadPool()
+        {
+            // Stop the thread pool and join all threads
+            std::unique_lock<std::mutex> lock(queue_mutex);
+            stop = true;
+            lock.unlock();
+            condition.notify_all();
+            for (std::thread &worker : workers) {
+                worker.join();
+            }
         }
-    }
 
-    // Wait for all tasks to complete before stopping the pool
-    void wait_all()
-    {
-        std::unique_lock<std::mutex> lock(queue_mutex);
-        done_condition.wait(
-                lock, [this] { return tasks.empty() && (active_tasks == 0); });
-        // lock is automatically released here
-    }
+        // Wait for all tasks to complete before stopping the pool
+        void wait_all()
+        {
+            std::unique_lock<std::mutex> lock(queue_mutex);
+            done_condition.wait(lock, [this] {
+                return tasks.empty() && (active_tasks == 0);
+            });
+            // lock is automatically released here
+        }
 
-    // Track the number of active tasks
-    void task_start()
-    {
-        std::unique_lock<std::mutex> lock(queue_mutex);
-        active_tasks++;
-        // lock is automatically released here
-    }
+        // Track the number of active tasks
+        void task_start()
+        {
+            std::unique_lock<std::mutex> lock(queue_mutex);
+            active_tasks++;
+            // lock is automatically released here
+        }
+
+        // Decrement the active task count and notify if all tasks are done
+        void task_end()
+        {
+            std::unique_lock<std::mutex> lock(queue_mutex);
+            active_tasks--;
+            if (tasks.empty() && active_tasks == 0) {
+                done_condition.notify_all();
+            }
+            // lock is automatically released here
+        }
+    };
 
-    // Decrement the active task count and notify if all tasks are done
-    void task_end()
+    // Wrapper for submitting tasks to the thread pool with automatic tracking
+    template <typename F>
+    void submit_task(ThreadPool &pool, F &&f)
     {
-        std::unique_lock<std::mutex> lock(queue_mutex);
-        active_tasks--;
-        if (tasks.empty() && active_tasks == 0) { done_condition.notify_all(); }
-        // lock is automatically released here
+        pool.task_start();
+        pool.enqueue([f = std::forward<F>(f), &pool]() {
+            try {
+                f();
+            } catch (...) {
+                // Ensure task_end is called even if the task throws an exception
+                pool.task_end();
+                throw; // Re-throw the exception
+            }
+            pool.task_end();
+        });
     }
-};
 
-// Wrapper for submitting tasks to the thread pool with automatic tracking
-template <typename F>
-void submit_task(ThreadPool &pool, F &&f)
-{
-    pool.task_start();
-    pool.enqueue([f = std::forward<F>(f), &pool]() {
-        f();
-        pool.task_end();
-    });
-}
+} // namespace tp
+} // namespace xss
 
 #endif // XSS_THREAD_POOL