Changes
diff --git a/CMakeLists.txt b/CMakeLists.txt
index 63b515a..8af2b29 100644
--- a/CMakeLists.txt
+++ b/CMakeLists.txt
@@ -15,28 +15,45 @@ if(NOT TARGET SQLite3::SQLite3)
add_library(SQLite3::SQLite3 ALIAS SQLite::SQLite3)
endif()
-add_library(status_tracker_data src/data_source_sqlite.cpp)
-target_compile_features(status_tracker_data PUBLIC cxx_std_23)
-set_target_properties(status_tracker_data PROPERTIES CXX_EXTENSIONS OFF)
-target_include_directories(status_tracker_data PUBLIC
- ${CMAKE_CURRENT_SOURCE_DIR}/src
+set(SOURCE_FILES
+ src/data_source_sqlite.cpp
+ src/thread_pool.cpp
)
-target_include_directories(status_tracker_data SYSTEM PUBLIC
- ${libmw_SOURCE_DIR}/includes
+set(LIBS
+ mw::sqlite
+ SQLite3::SQLite3
+ Threads::Threads
)
-target_link_libraries(status_tracker_data
- PRIVATE mw::sqlite
- PUBLIC SQLite3::SQLite3 Threads::Threads
+set(INCLUDES
+ ${CMAKE_CURRENT_SOURCE_DIR}/src
+ ${libmw_SOURCE_DIR}/includes
)
-add_executable(status_tracker src/main.cpp)
-target_link_libraries(status_tracker PRIVATE status_tracker_data)
+add_executable(status_tracker ${SOURCE_FILES} src/main.cpp)
target_compile_features(status_tracker PRIVATE cxx_std_23)
set_target_properties(status_tracker PROPERTIES CXX_EXTENSIONS OFF)
+target_include_directories(status_tracker PRIVATE ${INCLUDES})
+target_link_libraries(status_tracker PRIVATE ${LIBS})
include(CTest)
if(BUILD_TESTING)
- add_executable(data_source_sqlite_test tests/data_source_sqlite_test.cpp)
- target_link_libraries(data_source_sqlite_test PRIVATE status_tracker_data)
- add_test(NAME DataSourceSqlite COMMAND data_source_sqlite_test)
+ FetchContent_Declare(googletest
+ GIT_REPOSITORY https://github.com/google/googletest.git
+ GIT_TAG HEAD
+ )
+ FetchContent_MakeAvailable(googletest)
+ include(GoogleTest)
+
+ set(TEST_FILES
+ tests/data_source_sqlite_test.cpp
+ tests/thread_pool_test.cpp
+ )
+ add_executable(status_tracker_test ${SOURCE_FILES} ${TEST_FILES})
+ target_compile_features(status_tracker_test PRIVATE cxx_std_23)
+ set_target_properties(status_tracker_test PROPERTIES CXX_EXTENSIONS OFF)
+ target_include_directories(status_tracker_test PRIVATE ${INCLUDES})
+ target_link_libraries(status_tracker_test PRIVATE
+ ${LIBS} GTest::gtest_main
+ )
+ gtest_discover_tests(status_tracker_test PROPERTIES TIMEOUT 30)
endif()
diff --git a/src/thread_pool.cpp b/src/thread_pool.cpp
new file mode 100644
index 0000000..8621519
--- /dev/null
+++ b/src/thread_pool.cpp
@@ -0,0 +1,136 @@
+#include "thread_pool.h"
+
+#include <exception>
+#include <string>
+#include <utility>
+
+ThreadPool::ThreadPool(std::size_t worker_count)
+ : workers(worker_count)
+{}
+
+mw::E<std::unique_ptr<ThreadPool>> ThreadPool::create(std::size_t worker_count)
+{
+ if(worker_count == 0)
+ {
+ return std::unexpected(mw::runtimeError(
+ "Thread pool capacity must be positive"));
+ }
+ try
+ {
+ auto pool = std::unique_ptr<ThreadPool>(new ThreadPool(worker_count));
+ for(auto& worker : pool->workers)
+ {
+ worker.thread = std::thread(&ThreadPool::run, pool.get(),
+ std::ref(worker));
+ }
+ return pool;
+ }
+ catch(const std::exception& error)
+ {
+ return std::unexpected(mw::runtimeError(
+ std::string("Failed to create thread pool: ") + error.what()));
+ }
+}
+
+ThreadPool::~ThreadPool()
+{
+ stop();
+}
+
+void ThreadPool::stop()
+{
+ {
+ std::lock_guard lock(mutex);
+ stopping = true;
+ }
+ for(auto& worker : workers)
+ {
+ worker.work_ready.notify_one();
+ }
+ for(auto& worker : workers)
+ {
+ if(worker.thread.joinable())
+ {
+ worker.thread.join();
+ }
+ }
+}
+
+mw::E<std::optional<std::future<void>>> ThreadPool::trySubmit(
+ std::move_only_function<void()> task)
+{
+ if(!task)
+ {
+ return std::unexpected(mw::runtimeError(
+ "Thread pool task must not be empty"));
+ }
+ try
+ {
+ std::unique_lock lock(mutex);
+ if(stopping)
+ {
+ return std::nullopt;
+ }
+ for(auto& worker : workers)
+ {
+ if(!worker.busy)
+ {
+ std::packaged_task<void()> packaged(std::move(task));
+ auto result = packaged.get_future();
+ worker.task = std::move(packaged);
+ worker.busy = true;
+ ++outstanding;
+ lock.unlock();
+ worker.work_ready.notify_one();
+ return std::optional<std::future<void>>(std::move(result));
+ }
+ }
+ return std::nullopt;
+ }
+ catch(const std::exception& error)
+ {
+ return std::unexpected(mw::runtimeError(
+ std::string("Failed to submit thread pool task: ") + error.what()));
+ }
+}
+
+void ThreadPool::waitIdle()
+{
+ std::unique_lock lock(mutex);
+ while(outstanding != 0)
+ {
+ idle.wait(lock);
+ }
+}
+
+void ThreadPool::run(Worker& worker)
+{
+ while(true)
+ {
+ std::packaged_task<void()> task;
+ {
+ std::unique_lock lock(mutex);
+ while(!worker.task.valid() && !stopping)
+ {
+ worker.work_ready.wait(lock);
+ }
+ if(!worker.task.valid())
+ {
+ return;
+ }
+ task = std::move(worker.task);
+ }
+ // packaged_task delivers exceptions to its future.
+ task();
+ task = {};
+ {
+ std::lock_guard lock(mutex);
+ worker.busy = false;
+ --outstanding;
+ if(outstanding == 0)
+ {
+ idle.notify_all();
+ }
+ }
+ }
+}
diff --git a/src/thread_pool.h b/src/thread_pool.h
new file mode 100644
index 0000000..850d442
--- /dev/null
+++ b/src/thread_pool.h
@@ -0,0 +1,60 @@
+#pragma once
+
+#include <condition_variable>
+#include <cstddef>
+#include <functional>
+#include <future>
+#include <memory>
+#include <mutex>
+#include <optional>
+#include <thread>
+#include <vector>
+
+#include <mw/error.hpp>
+
+/// Fixed worker pool that rejects tasks when all workers are reserved.
+class ThreadPool
+{
+public:
+ /// Start workers, returning validation or initialization errors.
+ static mw::E<std::unique_ptr<ThreadPool>> create(std::size_t worker_count);
+
+ /// Finish accepted tasks and join workers. Must not run on a worker.
+ /// Callers must stop submitting before destroying the pool.
+ ~ThreadPool();
+
+ /// Pools own their workers and cannot be copied.
+ ThreadPool(const ThreadPool&) = delete;
+ /// Pools cannot transfer ownership through copy assignment.
+ ThreadPool& operator=(const ThreadPool&) = delete;
+
+ /// Reserve a worker or return nullopt immediately if full.
+ /// The future reports completion or the task's exception.
+ /// Invalid tasks and submission failures return errors.
+ mw::E<std::optional<std::future<void>>> trySubmit(
+ std::move_only_function<void()> task);
+
+ /// Wait until all accepted tasks finish. Must not run on a worker.
+ /// Concurrent submissions may extend the wait.
+ void waitIdle();
+
+private:
+ explicit ThreadPool(std::size_t worker_count);
+
+ struct Worker
+ {
+ std::condition_variable work_ready;
+ std::packaged_task<void()> task;
+ bool busy = false;
+ std::thread thread;
+ };
+
+ void run(Worker& worker);
+ void stop();
+
+ std::mutex mutex;
+ std::condition_variable idle;
+ std::size_t outstanding = 0;
+ bool stopping = false;
+ std::vector<Worker> workers;
+};
diff --git a/tests/data_source_sqlite_test.cpp b/tests/data_source_sqlite_test.cpp
index 3a748f1..b1b0bdf 100644
--- a/tests/data_source_sqlite_test.cpp
+++ b/tests/data_source_sqlite_test.cpp
@@ -3,58 +3,51 @@
#include <atomic>
#include <cstdlib>
#include <filesystem>
-#include <iostream>
-#include <stdexcept>
+#include <gtest/gtest.h>
#include <thread>
#include <unistd.h>
namespace
{
-void require(bool condition, const char* message)
-{
- if(!condition)
- {
- throw std::runtime_error(message);
- }
-}
-
-void History()
+TEST(DataSourceSqlite, History)
{
auto opened = DataSourceSqlite::open(":memory:");
- require(opened.has_value(), "Open memory database");
+ ASSERT_TRUE(opened.has_value()) << "Open memory database";
DataSourceInterface& source = **opened;
- require(!source.latest("missing").value(), "Missing latest result");
- require(source.history("missing", 0, 100).value().empty(),
- "Missing history");
+ ASSERT_TRUE(!source.latest("missing").value()) << "Missing latest result";
+ ASSERT_TRUE(source.history("missing", 0, 100).value().empty())
+ << "Missing history";
const std::string service_id = "blog'; DROP TABLE status_history; --";
const StatusRecord newer{service_id, 2200000000, 5000000000,
ProbeStatus::GOOD};
const StatusRecord older{service_id, 2199999999, 0, ProbeStatus::BAD};
const StatusRecord tied{service_id, 2200000000, 42, ProbeStatus::OTHER};
- require(source.save(newer).has_value(), "Save 64-bit fields");
- require(source.save(older).has_value(), "Save out of order");
- require(source.save(tied).has_value(), "Retain same-second results");
- require(source.save({"other", 2200000001, 1, ProbeStatus::GOOD})
- .has_value(), "Save separate service");
- require(source.latest(service_id).value().value() == tied,
- "Latest uses timestamp and insertion order");
+ ASSERT_TRUE(source.save(newer).has_value()) << "Save 64-bit fields";
+ ASSERT_TRUE(source.save(older).has_value()) << "Save out of order";
+ ASSERT_TRUE(source.save(tied).has_value()) << "Retain same-second results";
+ ASSERT_TRUE(source.save({"other", 2200000001, 1, ProbeStatus::GOOD})
+ .has_value()) << "Save separate service";
+ ASSERT_TRUE(source.latest(service_id).value().value() == tied)
+ << "Latest uses timestamp and insertion order";
const auto rows = source.history(service_id, 2199999999, 2200000001)
.value();
- require(rows == std::vector<StatusRecord>{older, newer, tied},
- "Ordered history and bound service ID");
- require(source.history(service_id, 2199999999, 2200000000).value() ==
- std::vector<StatusRecord>{older}, "Exclusive upper bound");
- require(source.history(service_id, 0, 0).value().empty(), "Empty range");
- require(!source.history(service_id, 1, 0), "Reject reversed bounds");
- require(!source.save({"", 0, 0, ProbeStatus::GOOD}), "Reject empty ID");
- require(!source.save({"x", 0, -1, ProbeStatus::GOOD}),
- "Reject negative duration");
- require(!source.save({"x", 0, 0, static_cast<ProbeStatus>(3)}),
- "Reject former NA value");
- require(!source.save({"x", 0, 0, static_cast<ProbeStatus>(99)}),
- "Reject invalid status");
- require(!DataSourceSqlite::open(""), "Reject empty path");
+ ASSERT_TRUE((rows == std::vector<StatusRecord>{older, newer, tied}))
+ << "Ordered history and bound service ID";
+ ASSERT_TRUE((source.history(service_id, 2199999999, 2200000000).value() ==
+ std::vector<StatusRecord>{older})) << "Exclusive upper bound";
+ ASSERT_TRUE(source.history(service_id, 0, 0).value().empty())
+ << "Empty range";
+ ASSERT_TRUE(!source.history(service_id, 1, 0)) << "Reject reversed bounds";
+ ASSERT_TRUE(!source.save({"", 0, 0, ProbeStatus::GOOD}))
+ << "Reject empty ID";
+ ASSERT_TRUE(!source.save({"x", 0, -1, ProbeStatus::GOOD}))
+ << "Reject negative duration";
+ ASSERT_TRUE(!source.save({"x", 0, 0, static_cast<ProbeStatus>(3)}))
+ << "Reject former NA value";
+ ASSERT_TRUE(!source.save({"x", 0, 0, static_cast<ProbeStatus>(99)}))
+ << "Reject invalid status";
+ ASSERT_TRUE(!DataSourceSqlite::open("")) << "Reject empty path";
}
void writeResults(DataSourceInterface& source, std::atomic<bool>& success,
@@ -70,7 +63,7 @@ void writeResults(DataSourceInterface& source, std::atomic<bool>& success,
}
}
-void ConcurrentAccess()
+TEST(DataSourceSqlite, ConcurrentAccess)
{
auto source = DataSourceSqlite::open(":memory:").value();
std::atomic<bool> success{true};
@@ -84,60 +77,58 @@ void ConcurrentAccess()
{
worker.join();
}
- require(success, "Concurrent reads and writes");
- require(source->history("service", 0, 800).value().size() == 800,
- "Concurrent writes retain all results");
+ EXPECT_TRUE(success.load()) << "Concurrent reads and writes";
+ ASSERT_TRUE(source->history("service", 0, 800).value().size() == 800)
+ << "Concurrent writes retain all results";
}
-void Persistence()
+class DataSourceSqliteFile : public testing::Test
{
- std::string pattern =
- (std::filesystem::temp_directory_path() / "status_tracker_XXXXXX")
- .string();
- const int descriptor = mkstemp(pattern.data());
- require(descriptor >= 0, "Create temporary database file");
- close(descriptor);
- const StatusRecord record{"persistent", 123, 456, ProbeStatus::GOOD};
- try
+protected:
+ void SetUp() override
{
- {
- auto source = DataSourceSqlite::open(pattern).value();
- require(source->save(record).has_value(), "Save to disk");
- auto second = DataSourceSqlite::open(pattern).value();
- require(second->latest("persistent").value().value() == record,
- "Read committed data from another connection");
- }
- {
- auto source = DataSourceSqlite::open(pattern).value();
- require(source->latest("persistent").value().value() == record,
- "Reopen existing schema and history");
- }
+ path = (std::filesystem::temp_directory_path() /
+ "status_tracker_XXXXXX").string();
+ const int descriptor = mkstemp(path.data());
+ ASSERT_GE(descriptor, 0);
+ close(descriptor);
}
- catch(...)
+
+ void TearDown() override
{
- std::filesystem::remove(pattern);
- throw;
+ std::error_code error;
+ std::filesystem::remove(path, error);
+ std::filesystem::remove(path + "-wal", error);
+ std::filesystem::remove(path + "-shm", error);
}
- std::filesystem::remove(pattern);
- require(!DataSourceSqlite::open(pattern + "/missing.sqlite"),
- "Report database open errors");
-}
-}
+ std::string path;
+};
-/// Run storage integration tests without relying on debug-only assertions.
-int main()
+TEST_F(DataSourceSqliteFile, Persistence)
{
- try
+ const StatusRecord record{"persistent", 123, 456, ProbeStatus::GOOD};
{
- History();
- ConcurrentAccess();
- Persistence();
+ auto opened = DataSourceSqlite::open(path);
+ ASSERT_TRUE(opened) << opened.error().msg();
+ auto& source = *opened;
+ ASSERT_TRUE(source->save(record));
+ auto second = DataSourceSqlite::open(path);
+ ASSERT_TRUE(second) << second.error().msg();
+ auto latest = (*second)->latest("persistent");
+ ASSERT_TRUE(latest);
+ ASSERT_TRUE(*latest);
+ EXPECT_EQ(**latest, record);
}
- catch(const std::exception& error)
{
- std::cerr << error.what() << '\n';
- return 1;
+ auto opened = DataSourceSqlite::open(path);
+ ASSERT_TRUE(opened) << opened.error().msg();
+ auto latest = (*opened)->latest("persistent");
+ ASSERT_TRUE(latest);
+ ASSERT_TRUE(*latest);
+ EXPECT_EQ(**latest, record);
}
- return 0;
+ EXPECT_FALSE(DataSourceSqlite::open(path + "/missing.sqlite"));
+}
+
}
diff --git a/tests/thread_pool_test.cpp b/tests/thread_pool_test.cpp
new file mode 100644
index 0000000..8a6ac32
--- /dev/null
+++ b/tests/thread_pool_test.cpp
@@ -0,0 +1,175 @@
+#include "thread_pool.h"
+
+#include <atomic>
+#include <gtest/gtest.h>
+#include <latch>
+#include <memory>
+#include <limits>
+#include <stdexcept>
+#include <string>
+
+namespace
+{
+
+void increment(std::atomic<int>& count)
+{
+ ++count;
+}
+
+void block(std::shared_future<void> release, std::latch& started,
+ std::atomic<int>& completed)
+{
+ started.count_down();
+ release.wait();
+ ++completed;
+}
+
+void fail()
+{
+ throw std::runtime_error("Task failed");
+}
+
+void assignValue(const std::unique_ptr<int>& value, std::atomic<int>& result)
+{
+ result = *value;
+}
+
+TEST(ThreadPool, CapacityAndReuse)
+{
+ std::atomic<int> completed{0};
+ std::latch started(2);
+ auto pool = ThreadPool::create(2).value();
+ // Promise destruction releases waiters after a failed assertion.
+ std::promise<void> release;
+ const auto ready = release.get_future().share();
+ auto first = pool->trySubmit(
+ std::bind_front(block, ready, std::ref(started), std::ref(completed)))
+ .value();
+ auto second = pool->trySubmit(
+ std::bind_front(block, ready, std::ref(started), std::ref(completed)))
+ .value();
+ ASSERT_TRUE(first.has_value() && second.has_value())
+ << "Accept up to capacity";
+ started.wait();
+ auto full = pool->trySubmit(
+ std::bind_front(increment, std::ref(completed)));
+ ASSERT_TRUE(full.has_value() && !*full) << "Full pool is not an error";
+ release.set_value();
+ first->get();
+ second->get();
+ pool->waitIdle();
+ EXPECT_EQ(completed.load(), 2) << "Rejected task never executes";
+ auto next = pool->trySubmit(
+ std::bind_front(increment, std::ref(completed))).value();
+ ASSERT_TRUE(next.has_value()) << "Reuse a worker";
+ next->get();
+ pool->waitIdle();
+ EXPECT_EQ(completed.load(), 3) << "Execute reused worker task";
+}
+
+TEST(ThreadPool, ExceptionsAndMoveOnlyTasks)
+{
+ std::atomic<int> result{0};
+ auto pool = ThreadPool::create(1).value();
+ auto failed = pool->trySubmit(fail).value();
+ ASSERT_TRUE(failed.has_value()) << "Accept throwing task";
+ bool caught = false;
+ try
+ {
+ failed->get();
+ }
+ catch(const std::runtime_error& error)
+ {
+ caught = std::string(error.what()) == "Task failed";
+ }
+ ASSERT_TRUE(caught) << "Propagate task exception through future";
+ pool->waitIdle();
+ auto next = pool->trySubmit(std::bind_front(
+ assignValue, std::make_unique<int>(42), std::ref(result))).value();
+ ASSERT_TRUE(next.has_value()) << "Worker survives task exception";
+ next->get();
+ pool->waitIdle();
+ EXPECT_EQ(result.load(), 42) << "Execute move-only task";
+}
+
+void submitBlocked(ThreadPool& pool, std::shared_future<void> release,
+ std::latch& started, std::atomic<int>& completed,
+ std::atomic<int>& accepted)
+{
+ auto submitted = pool.trySubmit(std::bind_front(
+ block, release, std::ref(started), std::ref(completed)));
+ if(submitted && *submitted)
+ {
+ ++accepted;
+ }
+}
+
+TEST(ThreadPool, ConcurrentSubmission)
+{
+ std::atomic<int> completed{0};
+ std::atomic<int> accepted{0};
+ std::latch started(4);
+ auto pool = ThreadPool::create(4).value();
+ std::promise<void> release;
+ const auto ready = release.get_future().share();
+ std::vector<std::jthread> producers;
+ for(int i = 0; i < 16; ++i)
+ {
+ producers.emplace_back(submitBlocked, std::ref(*pool), ready,
+ std::ref(started), std::ref(completed),
+ std::ref(accepted));
+ }
+ producers.clear();
+ EXPECT_EQ(accepted.load(), 4) << "Concurrent submissions respect capacity";
+ started.wait();
+ release.set_value();
+ pool->waitIdle();
+ EXPECT_EQ(completed.load(), 4) << "Execute each accepted task once";
+}
+
+TEST(ThreadPool, Destruction)
+{
+ std::atomic<int> completed{0};
+ std::latch started(2);
+ std::optional<std::future<void>> first;
+ std::optional<std::future<void>> second;
+ {
+ auto pool = ThreadPool::create(2).value();
+ std::promise<void> release;
+ const auto ready = release.get_future().share();
+ first = pool->trySubmit(std::bind_front(
+ block, ready, std::ref(started), std::ref(completed))).value();
+ second = pool->trySubmit(std::bind_front(
+ block, ready, std::ref(started), std::ref(completed))).value();
+ ASSERT_TRUE(first.has_value() && second.has_value()) << "Accept tasks";
+ // Promise destruction releases work immediately before pool teardown.
+ }
+ first->get();
+ second->get();
+ EXPECT_EQ(completed.load(), 2) << "Destruction finishes all accepted work";
+}
+
+TEST(ThreadPool, InvalidArguments)
+{
+ auto invalid = ThreadPool::create(0);
+ ASSERT_TRUE(!invalid) << "Reject zero capacity";
+ ASSERT_TRUE(!invalid.error().msg().empty()) << "Explain invalid capacity";
+ auto oversized = ThreadPool::create(
+ std::numeric_limits<std::size_t>::max());
+ ASSERT_TRUE(!oversized) << "Return worker allocation failure as an error";
+ ASSERT_TRUE(!oversized.error().msg().empty())
+ << "Explain initialization failure";
+ auto pool = ThreadPool::create(1).value();
+ auto empty = pool->trySubmit({});
+ ASSERT_TRUE(!empty) << "Reject empty task with an error";
+ ASSERT_TRUE(!empty.error().msg().empty()) << "Explain invalid task";
+ std::atomic<int> completed{0};
+ auto valid = pool->trySubmit(
+ std::bind_front(increment, std::ref(completed)));
+ ASSERT_TRUE(valid && *valid)
+ << "Invalid submission leaves capacity available";
+ (*valid)->get();
+ pool->waitIdle();
+}
+
+}