Changes
diff --git a/CMakeLists.txt b/CMakeLists.txt
index eceb15d..9726858 100644
--- a/CMakeLists.txt
+++ b/CMakeLists.txt
@@ -69,6 +69,7 @@ set(LIBS
mw::http-server
mw::sqlite
mw::url
+ spdlog::spdlog
ryml::ryml
CURL::libcurl
c-ares::cares
diff --git a/prd.md b/prd.md
index 51d69da..0dedc67 100644
--- a/prd.md
+++ b/prd.md
@@ -77,13 +77,12 @@ A simple self-hosted service status tracker.
- Probe workflow:
1. A priority queue of tasks, ordered by next due time.
2. Each endpoint has an atomic bool `in_flight`.
- 3. The scheduler dequeue a task, If no worker is available, ditch
- the task. If there is a worker available, if the endpoint is
- already inflight, ditch the task; otherwise mark the endpoint as
- inflight, and dispatch it to a worker.
+ 3. The scheduler dequeues a task. If the endpoint is already in flight,
+ skip this occurrence; otherwise mark it in flight and wait for a
+ worker. Shutdown or an execution error cancels this wait.
4. The scheduler enqueue the next task of that endpoint. Note that
if the schedued time of the next task is in the past, we should
- find the next one that’s in the future.
+ find the next one that’s in the future after dispatch.
5. The scheduler peeks the queue and wait for the next task to
trigger.
6. Meanwhile the worker does the probing, and save the result to db
diff --git a/src/scheduler.cpp b/src/scheduler.cpp
index 5077a9d..aa97957 100644
--- a/src/scheduler.cpp
+++ b/src/scheduler.cpp
@@ -1,7 +1,29 @@
#include "scheduler.h"
+#include <string_view>
#include <utility>
+#include <spdlog/spdlog.h>
+
+namespace
+{
+
+std::string_view probeStatusName(ProbeStatus status)
+{
+ switch(status)
+ {
+ case ProbeStatus::GOOD:
+ return "GOOD";
+ case ProbeStatus::BAD:
+ return "BAD";
+ case ProbeStatus::OTHER:
+ return "OTHER";
+ }
+ return "UNKNOWN";
+}
+
+}
+
bool Scheduler::Later::operator()(const ScheduledTask& left,
const ScheduledTask& right) const
{
@@ -104,7 +126,9 @@ mw::E<void> Scheduler::run(std::stop_token stop_token)
}
started = true;
}
- auto result = runLoop(stop_token);
+ std::stop_callback on_stop(
+ stop_token, std::bind_front(&Scheduler::requestStop, this));
+ auto result = runLoop(submission_stop.get_token());
pool->waitIdle();
std::lock_guard lock(mutex);
if(result && error)
@@ -133,21 +157,25 @@ mw::E<void> Scheduler::runLoop(std::stop_token stop_token)
{
auto scheduled = queue.top();
auto& service = *scheduled.service;
- auto next = nextDueTime(
- scheduled.due, service.config.interval, now);
- if(!next)
- {
- return std::unexpected(next.error());
- }
queue.pop();
- scheduled.due = *next;
- queue.push(scheduled);
lock.unlock();
auto dispatched = dispatch(service);
if(!dispatched)
{
return std::unexpected(dispatched.error());
}
+ if(stop_token.stop_requested())
+ {
+ break;
+ }
+ auto next = nextDueTime(
+ scheduled.due, service.config.interval, Clock::now());
+ if(!next)
+ {
+ return std::unexpected(next.error());
+ }
+ scheduled.due = *next;
+ queue.push(scheduled);
continue;
}
if(!queue.empty())
@@ -191,7 +219,8 @@ mw::E<void> Scheduler::dispatch(Service& service)
return {};
}
FlightGuard guard{service.in_flight};
- auto submitted = pool->trySubmit(Task{*this, service, nullptr});
+ auto submitted = pool->submit(Task{*this, service, nullptr},
+ submission_stop.get_token());
if(!submitted)
{
return std::unexpected(submitted.error());
@@ -229,9 +258,13 @@ void Scheduler::execute(Task& task)
auto result = task.probe->probe();
if(!result)
{
+ spdlog::error("Probe for service {} failed: {}", service_id,
+ result.error().msg());
reportError(service_id, std::move(result.error()));
return;
}
+ spdlog::info("Probe for service {} returned {}", service_id,
+ probeStatusName(result->status));
auto saved = data_source.save({service_id, result->timestamp,
result->duration_microsecond, result->status});
if(!saved)
@@ -249,7 +282,12 @@ void Scheduler::reportError(const std::string& service_id, mw::Error failure)
error = mw::runtimeError(service_id + ": " + failure.msg());
}
}
- wake.notify_one();
+ requestStop();
+}
+
+void Scheduler::requestStop()
+{
+ submission_stop.request_stop();
}
bool Scheduler::hasError() const
diff --git a/src/scheduler.h b/src/scheduler.h
index 3f38c1f..55f2f7d 100644
--- a/src/scheduler.h
+++ b/src/scheduler.h
@@ -44,7 +44,8 @@ public:
~Scheduler() = default;
/// Run once on the calling thread, initially making every task due.
- /// Skip occurrences while busy and preserve each task's interval cadence.
+ /// Wait for workers; skip occurrences of services already in flight.
+ /// Preserve cadence without replaying intervals missed while waiting.
/// Stop on cancellation or execution error, then finish accepted work.
mw::E<void> run(std::stop_token stop_token);
@@ -81,9 +82,11 @@ private:
void execute(Task& task);
void reportError(const std::string& service_id, mw::Error error);
bool hasError() const;
+ void requestStop();
std::mutex mutex;
std::condition_variable_any wake;
+ std::stop_source submission_stop;
bool started = false;
std::optional<mw::Error> error;
std::priority_queue<ScheduledTask, std::vector<ScheduledTask>, Later> queue;
diff --git a/src/thread_pool.cpp b/src/thread_pool.cpp
index 84a7e6b..81ca4ea 100644
--- a/src/thread_pool.cpp
+++ b/src/thread_pool.cpp
@@ -44,6 +44,7 @@ void ThreadPool::stop()
std::lock_guard lock(mutex);
stopping = true;
}
+ available.notify_all();
for(auto& worker : workers)
{
worker.work_ready.notify_one();
@@ -59,6 +60,24 @@ void ThreadPool::stop()
mw::E<std::optional<std::future<void>>> ThreadPool::trySubmit(
std::move_only_function<void()> task)
+{
+ return submitImpl(std::move(task), {}, false);
+}
+
+mw::E<std::optional<std::future<void>>> ThreadPool::submit(
+ std::move_only_function<void()> task, std::stop_token stop_token)
+{
+ return submitImpl(std::move(task), stop_token, true);
+}
+
+bool ThreadPool::hasCapacity() const
+{
+ return stopping || outstanding < workers.size();
+}
+
+mw::E<std::optional<std::future<void>>> ThreadPool::submitImpl(
+ std::move_only_function<void()> task, std::stop_token stop_token,
+ bool wait)
{
if(!task)
{
@@ -66,7 +85,12 @@ mw::E<std::optional<std::future<void>>> ThreadPool::trySubmit(
"Thread pool task must not be empty"));
}
std::unique_lock lock(mutex);
- if(stopping)
+ if(wait)
+ {
+ available.wait(lock, stop_token,
+ std::bind_front(&ThreadPool::hasCapacity, this));
+ }
+ if(stopping || stop_token.stop_requested())
{
return std::nullopt;
}
@@ -125,5 +149,6 @@ void ThreadPool::run(Worker& worker)
idle.notify_all();
}
}
+ available.notify_one();
}
}
diff --git a/src/thread_pool.h b/src/thread_pool.h
index ef21253..61f7ced 100644
--- a/src/thread_pool.h
+++ b/src/thread_pool.h
@@ -7,12 +7,13 @@
#include <memory>
#include <mutex>
#include <optional>
+#include <stop_token>
#include <thread>
#include <vector>
#include <mw/error.hpp>
-/// Fixed worker pool that rejects tasks when all workers are reserved.
+/// Fixed worker pool with immediate and cancellation-aware submission.
class ThreadPool
{
public:
@@ -34,6 +35,12 @@ public:
mw::E<std::optional<std::future<void>>> trySubmit(
std::move_only_function<void()> task);
+ /// Wait for a free worker; return nullopt if cancelled or stopping.
+ /// Accepted tasks finish normally. Must not run on a worker.
+ /// Invalid tasks return errors; task exceptions terminate the process.
+ mw::E<std::optional<std::future<void>>> submit(
+ std::move_only_function<void()> task, std::stop_token stop_token);
+
/// Wait until all accepted tasks finish. Must not run on a worker.
/// Concurrent submissions may extend the wait.
void waitIdle();
@@ -51,9 +58,14 @@ private:
void run(Worker& worker);
void stop();
+ bool hasCapacity() const;
+ mw::E<std::optional<std::future<void>>> submitImpl(
+ std::move_only_function<void()> task, std::stop_token stop_token,
+ bool wait);
std::mutex mutex;
std::condition_variable idle;
+ std::condition_variable_any available;
std::size_t outstanding = 0;
bool stopping = false;
std::vector<Worker> workers;
diff --git a/tests/scheduler_test.cpp b/tests/scheduler_test.cpp
index a07da20..1c1f6e3 100644
--- a/tests/scheduler_test.cpp
+++ b/tests/scheduler_test.cpp
@@ -51,9 +51,11 @@ public:
const Scheduler* scheduler = nullptr;
SaveMode mode = SaveMode::SUCCESS;
std::atomic<bool> flight_during_save{true};
+ std::atomic<int> save_attempts{0};
mw::E<void> save(const StatusRecord& record) override
{
+ ++save_attempts;
if(scheduler)
{
const auto* service = scheduler->findService(record.service_id);
@@ -399,7 +401,7 @@ TEST(Scheduler, SkipsInFlightAndFinishesAcceptedWorkOnStop)
EXPECT_TRUE(source.latest("a_slow").value());
}
-TEST(Scheduler, SkipsWhenFullAndRetriesAtNextInterval)
+TEST(Scheduler, WaitsForCapacityWithoutWaitingForNextInterval)
{
WatchingSource source;
std::atomic<int> calls{0};
@@ -408,7 +410,8 @@ TEST(Scheduler, SkipsWhenFullAndRetriesAtNextInterval)
std::stop_source stop;
auto release = std::make_unique<std::promise<void>>();
Configuration configuration{1, ":memory:", {{"Group", {
- serviceConfig("a_slow"), serviceConfig("retry", 1ms)
+ serviceConfig("a_slow"), serviceConfig("retry"),
+ serviceConfig("third")
}}}};
auto scheduler = Scheduler::create(configuration, source, BlockingFactory{
release->get_future().share(), started, calls});
@@ -421,6 +424,7 @@ TEST(Scheduler, SkipsWhenFullAndRetriesAtNextInterval)
EXPECT_FALSE(source.waitForRecords("retry", 1, 20ms));
release_on_exit->set_value();
ASSERT_TRUE(source.waitForRecords("retry", 1));
+ ASSERT_TRUE(source.waitForRecords("third", 1));
stop.request_stop();
ASSERT_EQ(running.wait_for(1s), std::future_status::ready);
EXPECT_TRUE(running.get());
@@ -428,6 +432,60 @@ TEST(Scheduler, SkipsWhenFullAndRetriesAtNextInterval)
EXPECT_EQ(calls.load(), 1);
}
+TEST(Scheduler, CancelsPendingSubmissionOnStopOrError)
+{
+ for(bool fail_save : {false, true})
+ {
+ WatchingSource source;
+ if(fail_save)
+ {
+ source.mode = WatchingSource::SaveMode::ERROR;
+ }
+ std::atomic<int> calls{0};
+ std::promise<void> started;
+ auto ready = started.get_future();
+ std::stop_source stop;
+ auto release = std::make_unique<std::promise<void>>();
+ Configuration configuration{1, ":memory:", {{"Group", {
+ serviceConfig("a_slow"), serviceConfig("pending")
+ }}}};
+ auto scheduler = Scheduler::create(
+ configuration, source, BlockingFactory{
+ release->get_future().share(), started, calls});
+ ASSERT_TRUE(scheduler);
+ auto running = std::async(std::launch::async, &Scheduler::run,
+ scheduler->get(), stop.get_token());
+ auto release_on_exit = std::move(release);
+ StopOnExit cleanup{stop};
+ ASSERT_EQ(ready.wait_for(1s), std::future_status::ready);
+ const auto* pending = (*scheduler)->findService("pending");
+ const auto deadline = Scheduler::Clock::now() + 1s;
+ while(!pending->in_flight && Scheduler::Clock::now() < deadline)
+ {
+ std::this_thread::yield();
+ }
+ ASSERT_TRUE(pending->in_flight);
+ if(!fail_save)
+ {
+ stop.request_stop();
+ const auto cancelled_by = Scheduler::Clock::now() + 1s;
+ while(pending->in_flight &&
+ Scheduler::Clock::now() < cancelled_by)
+ {
+ std::this_thread::yield();
+ }
+ EXPECT_FALSE(pending->in_flight);
+ EXPECT_EQ(running.wait_for(20ms), std::future_status::timeout);
+ }
+ release_on_exit->set_value();
+ ASSERT_EQ(running.wait_for(1s), std::future_status::ready);
+ EXPECT_EQ(running.get().has_value(), !fail_save);
+ EXPECT_FALSE(pending->in_flight);
+ EXPECT_FALSE(source.latest("pending").value());
+ EXPECT_EQ(source.save_attempts.load(), 1);
+ }
+}
+
TEST(Scheduler, InvalidConfiguration)
{
WatchingSource source;