initial queue

Author: ConorWilliams

.clang-format

@@ -0,0 +1,16 @@
+BasedOnStyle: Google
+AccessModifierOffset: '-2'
+AlignTrailingComments: 'true'
+AllowAllParametersOfDeclarationOnNextLine: 'false'
+AlwaysBreakTemplateDeclarations: 'No'
+BreakBeforeBraces: Attach
+ColumnLimit: '100'
+ConstructorInitializerAllOnOneLineOrOnePerLine: 'true'
+IncludeBlocks: Regroup
+IndentPPDirectives: AfterHash
+IndentWidth: '4'
+BreakBeforeBinaryOperators: All
+BreakBeforeTernaryOperators: 'true'
+
+BinPackArguments: false
+BinPackParameters: false

.cmake-format

@@ -0,0 +1,57 @@
+format:
+  tab_size: 4
+  line_width: 100
+  dangle_parens: true
+
+parse:
+  additional_commands:
+    cpmaddpackage:
+      pargs:
+        nargs: '*'
+        flags: []
+      spelling: CPMAddPackage
+      kwargs: &cpmaddpackagekwargs
+        NAME: 1
+        FORCE: 1
+        VERSION: 1
+        GIT_TAG: 1
+        DOWNLOAD_ONLY: 1
+        GITHUB_REPOSITORY: 1
+        GITLAB_REPOSITORY: 1
+        GIT_REPOSITORY: 1
+        SVN_REPOSITORY: 1
+        SVN_REVISION: 1
+        SOURCE_DIR: 1
+        DOWNLOAD_COMMAND: 1
+        FIND_PACKAGE_ARGUMENTS: 1
+        NO_CACHE: 1
+        GIT_SHALLOW: 1
+        URL: 1
+        URL_HASH: 1
+        URL_MD5: 1
+        DOWNLOAD_NAME: 1
+        DOWNLOAD_NO_EXTRACT: 1
+        HTTP_USERNAME: 1
+        HTTP_PASSWORD: 1
+        OPTIONS: +
+    cpmfindpackage:
+      pargs:
+        nargs: '*'
+        flags: []
+      spelling: CPMFindPackage
+      kwargs: *cpmaddpackagekwargs
+    packageproject:
+      pargs:
+        nargs: '*'
+        flags: []
+      spelling: packageProject
+      kwargs:
+        NAME: 1
+        VERSION: 1
+        NAMESPACE: 1
+        INCLUDE_DIR: 1
+        INCLUDE_DESTINATION: 1
+        BINARY_DIR: 1
+        COMPATIBILITY: 1
+        VERSION_HEADER: 1
+        DEPENDENCIES: +

CMakeLists.txt

@@ -0,0 +1,83 @@
+cmake_minimum_required(VERSION 3.14 FATAL_ERROR)
+
+# ---- Project ----
+
+# Note: update this to your new project's name and version
+project(
+  Threadpool
+  VERSION 0.1
+  LANGUAGES CXX
+)
+
+# ---- Include guards ----
+
+if(PROJECT_SOURCE_DIR STREQUAL PROJECT_BINARY_DIR)
+  message(
+    FATAL_ERROR
+      "In-source builds not allowed. Please make a new directory (called a build directory) and run CMake from there."
+  )
+endif()
+
+# ---- Add dependencies ----
+
+find_package( Threads )
+
+include(cmake/CPM.cmake)
+
+CPMAddPackage("gh:TheLartians/PackageProject.cmake@1.4.1")
+CPMAddPackage("gh:Naios/function2#4.1.0")
+
+
+
+
+# ---- Create library ----
+
+set(sources "include/singlequeue.cpp")
+
+# Note: for header-only libraries change all PUBLIC flags to INTERFACE and create an interface
+# target: add_library(Threadpool INTERFACE)
+add_library(Threadpool ${sources})
+
+target_compile_features(Threadpool PUBLIC cxx_std_20)
+
+target_compile_options(Threadpool PRIVATE -Wall -Wextra -Wpedantic)
+
+# Being a cross-platform target, we enforce standards conformance on MSVC
+target_compile_options(Threadpool PUBLIC "$<$<COMPILE_LANG_AND_ID:CXX,MSVC>:/permissive>")
+
+# Link dependencies
+target_link_libraries(Threadpool PUBLIC function2::function2 ${CMAKE_THREAD_LIBS_INIT})
+
+target_include_directories(
+  Threadpool PUBLIC $<BUILD_INTERFACE:${PROJECT_SOURCE_DIR}/include>
+                    $<INSTALL_INTERFACE:include/${PROJECT_NAME}-${PROJECT_VERSION}>
+)
+
+# ---- Create an installable target ----
+# this allows users to install and find the library via `find_package()`.
+
+# the location where the project's version header will be placed should match the project's regular
+# header paths
+string(TOLOWER ${PROJECT_NAME}/version.h VERSION_HEADER_LOCATION)
+
+packageProject(
+  NAME ${PROJECT_NAME}
+  VERSION ${PROJECT_VERSION}
+  NAMESPACE ${PROJECT_NAME}
+  BINARY_DIR ${PROJECT_BINARY_DIR}
+  INCLUDE_DIR ${PROJECT_SOURCE_DIR}/include
+  INCLUDE_DESTINATION include/${PROJECT_NAME}-${PROJECT_VERSION}
+  VERSION_HEADER "${VERSION_HEADER_LOCATION}"
+  COMPATIBILITY SameMajorVersion
+  DEPENDENCIES ""
+)
+
+
+# ---- Conditionally build examples ----
+
+option(BUILD_EXAMPLES "Build the examples" OFF)
+
+if(BUILD_EXAMPLES)
+ add_executable(example "include/main.cpp")
+ target_link_libraries(example PUBLIC Threadpool::Threadpool)
+endif()

cmake/CPM.cmake

@@ -0,0 +1,21 @@
+set(CPM_DOWNLOAD_VERSION 0.31.1)
+
+if(CPM_SOURCE_CACHE)
+  # Expand relative path. This is important if the provided path contains a tilde (~)
+  get_filename_component(CPM_SOURCE_CACHE ${CPM_SOURCE_CACHE} ABSOLUTE)
+  set(CPM_DOWNLOAD_LOCATION "${CPM_SOURCE_CACHE}/cpm/CPM_${CPM_DOWNLOAD_VERSION}.cmake")
+elseif(DEFINED ENV{CPM_SOURCE_CACHE})
+  set(CPM_DOWNLOAD_LOCATION "$ENV{CPM_SOURCE_CACHE}/cpm/CPM_${CPM_DOWNLOAD_VERSION}.cmake")
+else()
+  set(CPM_DOWNLOAD_LOCATION "${CMAKE_BINARY_DIR}/cmake/CPM_${CPM_DOWNLOAD_VERSION}.cmake")
+endif()
+
+if(NOT (EXISTS ${CPM_DOWNLOAD_LOCATION}))
+  message(STATUS "Downloading CPM.cmake to ${CPM_DOWNLOAD_LOCATION}")
+  file(DOWNLOAD
+       https://github.com/cpm-cmake/CPM.cmake/releases/download/v${CPM_DOWNLOAD_VERSION}/CPM.cmake
+       ${CPM_DOWNLOAD_LOCATION}
+  )
+endif()
+
+include(${CPM_DOWNLOAD_LOCATION})

include/lockfree.hpp

@@ -0,0 +1,26 @@
+
+
+#include <chrono>
+#include <future>
+#include <iostream>
+#include <ratio>
+#include <thread>
+
+#include "singlequeue.hpp"
+
+int main() {
+    ThreadPool pool(12);
+
+    std::vector<std::future<int>> _futures;
+
+    for (size_t i = 0; i < 10; i++) {
+        pool.execute([time = 100]() {
+            std::this_thread::sleep_for(std::chrono::milliseconds(time));
+            std::cout << "Waited :" << time << std::endl;
+            return 3;
+        });
+    }
+
+    std::cout << "Working\n";
+    return 0;
+}

include/main.cpp

@@ -0,0 +1,36 @@
+#include "singlequeue.hpp"
+
+ThreadPool::ThreadPool(size_t threads) {
+    for (size_t i = 0; i < threads; ++i) {
+        _threads.emplace_back(std::thread([&]() {
+            std::unique_lock<std::mutex> queue_lock(_task_mutex, std::defer_lock);
+
+            while (true) {
+                queue_lock.lock();
+
+                _task_cv.wait(queue_lock, [&]() { return !_tasks.empty() || _stop_threads; });
+
+                if (_stop_threads && _tasks.empty()) {
+                    return;
+                }
+
+                auto one_shot = std::move(_tasks.front());
+
+                _tasks.pop();
+
+                queue_lock.unlock();
+
+                std::invoke(std::move(one_shot));
+            }
+        }));
+    }
+}
+
+ThreadPool::~ThreadPool() {
+    _stop_threads = true;
+    _task_cv.notify_all();
+
+    for (auto &thread : _threads) {
+        thread.join();
+    }
+}

include/singlequeue.cpp

@@ -0,0 +1,78 @@
+#pragma once
+
+#include <concepts>
+#include <condition_variable>
+#include <future>
+#include <mutex>
+#include <queue>
+#include <thread>
+#include <type_traits>  // invoke_result
+#include <utility>
+#include <vector>
+
+#include "function2/function2.hpp"
+
+// Bind F and args... into nullary lambda
+template <typename F, typename... Args> auto bind(F &&f, Args &&...arg) {
+    return [f = std::forward<F>(f), ... arg = std::forward<Args>(arg)]() mutable -> decltype(auto) {
+        return std::invoke(std::forward<F>(f), std::forward<Args>(arg)...);
+    };
+}
+
+// Like std::packaged_task<R() &&>, but garantees no type erasure.
+template <std::invocable F> class NullaryOneShot {
+  public:
+    using invoke_result_t = std::invoke_result_t<F>;
+
+    NullaryOneShot(F &&fn) : _fn(std::forward<F>(fn)) {}
+
+    void operator()() && {
+        if constexpr (!std::is_same_v<void, invoke_result_t>) {
+            _promise.set_value(std::invoke(std::forward<F>(_fn)));
+        } else {
+            std::invoke(std::forward<F>(_fn));
+            _promise.set_value();
+        }
+    }
+
+    std::future<invoke_result_t> get_future() { return _promise.get_future(); }
+
+  private:
+    F _fn;
+    std::promise<invoke_result_t> _promise;
+};
+
+template <typename F> NullaryOneShot(F &&) -> NullaryOneShot<F>;
+
+class ThreadPool {
+  public:
+    explicit ThreadPool(size_t threads = std::thread::hardware_concurrency());
+
+    ~ThreadPool();
+
+    template <typename F, typename... Args> auto execute(F &&f, Args &&...args) {
+        //
+        auto task = NullaryOneShot(bind(std::forward<F>(f), std::forward<Args>(args)...));
+
+        auto future = task.get_future();
+        //
+        std::unique_lock<std::mutex> queue_lock(_task_mutex, std::defer_lock);
+
+        queue_lock.lock();
+
+        _tasks.emplace(std::move(task));
+
+        queue_lock.unlock();
+
+        _task_cv.notify_one();
+
+        return future;
+    }
+
+  private:
+    std::vector<std::thread> _threads;
+    std::queue<fu2::unique_function<void() &&>> _tasks;
+    std::mutex _task_mutex;
+    std::condition_variable _task_cv;
+    bool _stop_threads = false;
+};