Changes
diff --git a/src/probe.cpp b/src/probe.cpp
index 33b99e0..5003d97 100644
--- a/src/probe.cpp
+++ b/src/probe.cpp
@@ -2,7 +2,6 @@
#include "socket_probe.h"
#include <climits>
-#include <exception>
#include <span>
#include <utility>
@@ -173,30 +172,23 @@ mw::E<ProbeStatus> check(const Config& config,
template<typename Config>
mw::E<ProbeResult> measure(const Config& config, std::chrono::seconds timeout)
{
- try
+ const auto start = Clock::now();
+ auto valid_timeout = validateTimeout(timeout, start);
+ if(!valid_timeout)
{
- const auto start = Clock::now();
- auto valid_timeout = validateTimeout(timeout, start);
- if(!valid_timeout)
- {
- return std::unexpected(valid_timeout.error());
- }
- const auto timestamp = std::chrono::duration_cast<std::chrono::seconds>(
- std::chrono::system_clock::now().time_since_epoch()).count();
- auto status = check(config, timeout, start + timeout);
- if(!status)
- {
- return std::unexpected(status.error());
- }
- const auto duration =
- std::chrono::duration_cast<std::chrono::microseconds>(
- Clock::now() - start).count();
- return ProbeResult{*status, timestamp, duration};
+ return std::unexpected(valid_timeout.error());
}
- catch(const std::exception& error)
+ const auto timestamp = std::chrono::duration_cast<std::chrono::seconds>(
+ std::chrono::system_clock::now().time_since_epoch()).count();
+ auto status = check(config, timeout, start + timeout);
+ if(!status)
{
- return std::unexpected(mw::runtimeError(error.what()));
+ return std::unexpected(status.error());
}
+ const auto duration =
+ std::chrono::duration_cast<std::chrono::microseconds>(
+ Clock::now() - start).count();
+ return ProbeResult{*status, timestamp, duration};
}
}
@@ -213,19 +205,12 @@ mw::E<ProbeResult> HttpProbe::probe()
mw::E<std::unique_ptr<ProbeInterface>> createProbe(
const HttpEndpoint& config, std::chrono::seconds timeout)
{
- try
+ auto valid = prepare(config, timeout);
+ if(!valid)
{
- auto valid = prepare(config, timeout);
- if(!valid)
- {
- return std::unexpected(valid.error());
- }
- return std::unique_ptr<ProbeInterface>(new HttpProbe(config, timeout));
- }
- catch(const std::exception& error)
- {
- return std::unexpected(mw::runtimeError(error.what()));
+ return std::unexpected(valid.error());
}
+ return std::unique_ptr<ProbeInterface>(new HttpProbe(config, timeout));
}
TcpProbe::TcpProbe(TcpEndpoint config, std::chrono::seconds timeout)
@@ -240,19 +225,12 @@ mw::E<ProbeResult> TcpProbe::probe()
mw::E<std::unique_ptr<ProbeInterface>> createProbe(
const TcpEndpoint& config, std::chrono::seconds timeout)
{
- try
+ auto valid = prepare(config, timeout);
+ if(!valid)
{
- auto valid = prepare(config, timeout);
- if(!valid)
- {
- return std::unexpected(valid.error());
- }
- return std::unique_ptr<ProbeInterface>(new TcpProbe(config, timeout));
- }
- catch(const std::exception& error)
- {
- return std::unexpected(mw::runtimeError(error.what()));
+ return std::unexpected(valid.error());
}
+ return std::unique_ptr<ProbeInterface>(new TcpProbe(config, timeout));
}
UdpProbe::UdpProbe(UdpEndpoint config, std::chrono::seconds timeout)
@@ -267,19 +245,12 @@ mw::E<ProbeResult> UdpProbe::probe()
mw::E<std::unique_ptr<ProbeInterface>> createProbe(
const UdpEndpoint& config, std::chrono::seconds timeout)
{
- try
- {
- auto valid = prepare(config, timeout);
- if(!valid)
- {
- return std::unexpected(valid.error());
- }
- return std::unique_ptr<ProbeInterface>(new UdpProbe(config, timeout));
- }
- catch(const std::exception& error)
+ auto valid = prepare(config, timeout);
+ if(!valid)
{
- return std::unexpected(mw::runtimeError(error.what()));
+ return std::unexpected(valid.error());
}
+ return std::unique_ptr<ProbeInterface>(new UdpProbe(config, timeout));
}
IcmpProbe::IcmpProbe(IcmpEndpoint config, std::chrono::seconds timeout)
@@ -294,17 +265,10 @@ mw::E<ProbeResult> IcmpProbe::probe()
mw::E<std::unique_ptr<ProbeInterface>> createProbe(
const IcmpEndpoint& config, std::chrono::seconds timeout)
{
- try
- {
- auto valid = prepare(config, timeout);
- if(!valid)
- {
- return std::unexpected(valid.error());
- }
- return std::unique_ptr<ProbeInterface>(new IcmpProbe(config, timeout));
- }
- catch(const std::exception& error)
+ auto valid = prepare(config, timeout);
+ if(!valid)
{
- return std::unexpected(mw::runtimeError(error.what()));
+ return std::unexpected(valid.error());
}
+ return std::unique_ptr<ProbeInterface>(new IcmpProbe(config, timeout));
}
diff --git a/src/scheduler.cpp b/src/scheduler.cpp
index 6608f20..5077a9d 100644
--- a/src/scheduler.cpp
+++ b/src/scheduler.cpp
@@ -1,6 +1,5 @@
#include "scheduler.h"
-#include <exception>
#include <utility>
bool Scheduler::Later::operator()(const ScheduledTask& left,
@@ -43,56 +42,49 @@ mw::E<std::unique_ptr<Scheduler>> Scheduler::create(
return std::unexpected(mw::runtimeError(
"Scheduler requires workers and a probe factory"));
}
- try
+ auto scheduler = std::unique_ptr<Scheduler>(
+ new Scheduler(data_source, std::move(probe_factory)));
+ for(const auto& group : configuration.groups)
{
- auto scheduler = std::unique_ptr<Scheduler>(
- new Scheduler(data_source, std::move(probe_factory)));
- for(const auto& group : configuration.groups)
+ for(const auto& config : group.services)
{
- for(const auto& config : group.services)
+ if(config.id.empty() ||
+ config.id.find('\0') != std::string::npos)
{
- if(config.id.empty() ||
- config.id.find('\0') != std::string::npos)
- {
- return std::unexpected(mw::runtimeError(
- "Invalid service ID"));
- }
- if(scheduler->services.contains(config.id))
- {
- return std::unexpected(mw::runtimeError(
- "Duplicate service ID: " + config.id));
- }
- const auto now = Clock::now();
- auto next = nextDueTime(now, config.interval, now);
- if(!next)
- {
- return std::unexpected(mw::runtimeError(
- config.id + ": " + next.error().msg()));
- }
- auto service = std::make_unique<Service>();
- service->config = config;
- // Validate without running the probe or the injected factory.
- auto probe = service->createProbe();
- if(!probe)
- {
- return std::unexpected(mw::runtimeError(
- config.id + ": " + probe.error().msg()));
- }
- scheduler->services.emplace(config.id, std::move(service));
+ return std::unexpected(mw::runtimeError(
+ "Invalid service ID"));
}
+ if(scheduler->services.contains(config.id))
+ {
+ return std::unexpected(mw::runtimeError(
+ "Duplicate service ID: " + config.id));
+ }
+ const auto now = Clock::now();
+ auto next = nextDueTime(now, config.interval, now);
+ if(!next)
+ {
+ return std::unexpected(mw::runtimeError(
+ config.id + ": " + next.error().msg()));
+ }
+ auto service = std::make_unique<Service>();
+ service->config = config;
+ // Validate without running the probe or the injected factory.
+ auto probe = service->createProbe();
+ if(!probe)
+ {
+ return std::unexpected(mw::runtimeError(
+ config.id + ": " + probe.error().msg()));
+ }
+ scheduler->services.emplace(config.id, std::move(service));
}
- auto pool = ThreadPool::create(configuration.worker_count);
- if(!pool)
- {
- return std::unexpected(pool.error());
- }
- scheduler->pool = std::move(*pool);
- return scheduler;
}
- catch(const std::exception& error)
+ auto pool = ThreadPool::create(configuration.worker_count);
+ if(!pool)
{
- return std::unexpected(mw::runtimeError(error.what()));
+ return std::unexpected(pool.error());
}
+ scheduler->pool = std::move(*pool);
+ return scheduler;
}
const Service* Scheduler::findService(const std::string& service_id) const
@@ -112,15 +104,7 @@ mw::E<void> Scheduler::run(std::stop_token stop_token)
}
started = true;
}
- mw::E<void> result;
- try
- {
- result = runLoop(stop_token);
- }
- catch(const std::exception& exception)
- {
- result = std::unexpected(mw::runtimeError(exception.what()));
- }
+ auto result = runLoop(stop_token);
pool->waitIdle();
std::lock_guard lock(mutex);
if(result && error)
@@ -229,41 +213,30 @@ void Scheduler::execute(Task& task)
{
FlightGuard guard{task.service.in_flight};
const auto& service_id = task.service.config.id;
- try
+ auto probe = probe_factory(task.service);
+ if(!probe)
{
- auto probe = probe_factory(task.service);
- if(!probe)
- {
- reportError(service_id, std::move(probe.error()));
- return;
- }
- if(!*probe)
- {
- reportError(service_id, mw::runtimeError(
- "Probe factory returned a null probe"));
- return;
- }
- task.probe = std::move(*probe);
- auto result = task.probe->probe();
- if(!result)
- {
- reportError(service_id, std::move(result.error()));
- return;
- }
- auto saved = data_source.save({service_id, result->timestamp,
- result->duration_microsecond, result->status});
- if(!saved)
- {
- reportError(service_id, std::move(saved.error()));
- }
+ reportError(service_id, std::move(probe.error()));
+ return;
+ }
+ if(!*probe)
+ {
+ reportError(service_id, mw::runtimeError(
+ "Probe factory returned a null probe"));
+ return;
}
- catch(const std::exception& exception)
+ task.probe = std::move(*probe);
+ auto result = task.probe->probe();
+ if(!result)
{
- reportError(service_id, mw::runtimeError(exception.what()));
+ reportError(service_id, std::move(result.error()));
+ return;
}
- catch(...)
+ auto saved = data_source.save({service_id, result->timestamp,
+ result->duration_microsecond, result->status});
+ if(!saved)
{
- reportError(service_id, mw::runtimeError("Unknown task exception"));
+ reportError(service_id, std::move(saved.error()));
}
}
diff --git a/src/service.cpp b/src/service.cpp
index bb91825..9673047 100644
--- a/src/service.cpp
+++ b/src/service.cpp
@@ -1,7 +1,5 @@
#include "service.h"
-#include <exception>
-
namespace
{
@@ -32,13 +30,6 @@ struct ProbeFactory
mw::E<std::unique_ptr<ProbeInterface>> Service::createProbe() const
{
- try
- {
- return std::visit(ProbeFactory{config.timeout, config.url},
- config.endpoint);
- }
- catch(const std::exception& error)
- {
- return std::unexpected(mw::runtimeError(error.what()));
- }
+ return std::visit(ProbeFactory{config.timeout, config.url},
+ config.endpoint);
}
diff --git a/src/thread_pool.cpp b/src/thread_pool.cpp
index 8621519..84a7e6b 100644
--- a/src/thread_pool.cpp
+++ b/src/thread_pool.cpp
@@ -1,9 +1,18 @@
#include "thread_pool.h"
-#include <exception>
-#include <string>
#include <utility>
+namespace
+{
+
+// Terminate before packaged_task can capture an unexpected exception.
+void invokeTask(std::move_only_function<void()>& task) noexcept
+{
+ task();
+}
+
+}
+
ThreadPool::ThreadPool(std::size_t worker_count)
: workers(worker_count)
{}
@@ -15,21 +24,13 @@ mw::E<std::unique_ptr<ThreadPool>> ThreadPool::create(std::size_t worker_count)
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)
+ auto pool = std::unique_ptr<ThreadPool>(new ThreadPool(worker_count));
+ for(auto& worker : pool->workers)
{
- return std::unexpected(mw::runtimeError(
- std::string("Failed to create thread pool: ") + error.what()));
+ worker.thread = std::thread(&ThreadPool::run, pool.get(),
+ std::ref(worker));
}
+ return pool;
}
ThreadPool::~ThreadPool()
@@ -64,34 +65,27 @@ mw::E<std::optional<std::future<void>>> ThreadPool::trySubmit(
return std::unexpected(mw::runtimeError(
"Thread pool task must not be empty"));
}
- try
+ std::unique_lock lock(mutex);
+ if(stopping)
{
- 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)
+ for(auto& worker : workers)
{
- return std::unexpected(mw::runtimeError(
- std::string("Failed to submit thread pool task: ") + error.what()));
+ if(!worker.busy)
+ {
+ std::packaged_task<void()> packaged(
+ std::bind_front(invokeTask, 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;
}
void ThreadPool::waitIdle()
@@ -120,7 +114,6 @@ void ThreadPool::run(Worker& worker)
}
task = std::move(worker.task);
}
- // packaged_task delivers exceptions to its future.
task();
task = {};
{
diff --git a/src/thread_pool.h b/src/thread_pool.h
index 850d442..ef21253 100644
--- a/src/thread_pool.h
+++ b/src/thread_pool.h
@@ -16,7 +16,7 @@
class ThreadPool
{
public:
- /// Start workers, returning validation or initialization errors.
+ /// Start workers, returning validation 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.
@@ -29,8 +29,8 @@ public:
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.
+ /// The future reports completion. Task exceptions terminate the process.
+ /// Invalid tasks return errors.
mw::E<std::optional<std::future<void>>> trySubmit(
std::move_only_function<void()> task);
diff --git a/tests/scheduler_test.cpp b/tests/scheduler_test.cpp
index 1e81ef3..6b91986 100644
--- a/tests/scheduler_test.cpp
+++ b/tests/scheduler_test.cpp
@@ -283,8 +283,7 @@ mw::E<std::unique_ptr<ProbeInterface>> throwingFactory(
TEST(Scheduler, FailuresClearServiceFlightState)
{
- for(auto factory : {factoryError, failedProbe, throwingProbe,
- nullProbe, throwingFactory})
+ for(auto factory : {factoryError, failedProbe, nullProbe})
{
WatchingSource source;
auto scheduler = Scheduler::create(oneService(), source, factory);
@@ -304,25 +303,21 @@ TEST(Scheduler, FailuresClearServiceFlightState)
TEST(Scheduler, PersistenceFailuresClearServiceFlightState)
{
- for(auto mode : {WatchingSource::SaveMode::ERROR,
- WatchingSource::SaveMode::EXCEPTION})
- {
- WatchingSource source;
- source.mode = mode;
- auto scheduler = Scheduler::create(oneService(), source, makeFake);
- ASSERT_TRUE(scheduler);
- source.scheduler = scheduler->get();
- std::stop_source stop;
- auto running = std::async(std::launch::async, &Scheduler::run,
- scheduler->get(), stop.get_token());
- StopOnExit cleanup{stop};
- ASSERT_EQ(running.wait_for(1s), std::future_status::ready);
- auto result = running.get();
- ASSERT_FALSE(result);
- EXPECT_NE(result.error().msg().find("service:"), std::string::npos);
- EXPECT_FALSE((*scheduler)->findService("service")->in_flight.load());
- EXPECT_TRUE(source.flight_during_save.load());
- }
+ WatchingSource source;
+ source.mode = WatchingSource::SaveMode::ERROR;
+ auto scheduler = Scheduler::create(oneService(), source, makeFake);
+ ASSERT_TRUE(scheduler);
+ source.scheduler = scheduler->get();
+ std::stop_source stop;
+ auto running = std::async(std::launch::async, &Scheduler::run,
+ scheduler->get(), stop.get_token());
+ StopOnExit cleanup{stop};
+ ASSERT_EQ(running.wait_for(1s), std::future_status::ready);
+ auto result = running.get();
+ ASSERT_FALSE(result);
+ EXPECT_NE(result.error().msg().find("service:"), std::string::npos);
+ EXPECT_FALSE((*scheduler)->findService("service")->in_flight.load());
+ EXPECT_TRUE(source.flight_during_save.load());
}
class BlockingProbe final : public ProbeInterface
@@ -461,3 +456,26 @@ TEST(Scheduler, InvalidConfiguration)
}
}
+
+TEST(SchedulerDeathTest, UnexpectedProbeExceptionsTerminate)
+{
+ for(auto factory : {throwingProbe, throwingFactory})
+ {
+ ASSERT_DEATH({
+ WatchingSource source;
+ auto scheduler = Scheduler::create(oneService(), source, factory);
+ (*scheduler)->run(std::stop_token{});
+ }, "");
+ }
+}
+
+TEST(SchedulerDeathTest, UnexpectedSaveExceptionTerminates)
+{
+ ASSERT_DEATH({
+ WatchingSource source;
+ source.mode = WatchingSource::SaveMode::EXCEPTION;
+ auto scheduler = Scheduler::create(oneService(), source, makeFake);
+ source.scheduler = scheduler->get();
+ (*scheduler)->run(std::stop_token{});
+ }, "");
+}
diff --git a/tests/thread_pool_test.cpp b/tests/thread_pool_test.cpp
index 8a6ac32..3c86cf4 100644
--- a/tests/thread_pool_test.cpp
+++ b/tests/thread_pool_test.cpp
@@ -67,26 +67,13 @@ TEST(ThreadPool, CapacityAndReuse)
EXPECT_EQ(completed.load(), 3) << "Execute reused worker task";
}
-TEST(ThreadPool, ExceptionsAndMoveOnlyTasks)
+TEST(ThreadPool, MoveOnlyTasks)
{
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";
+ ASSERT_TRUE(next.has_value()) << "Accept move-only task";
next->get();
pool->waitIdle();
EXPECT_EQ(result.load(), 42) << "Execute move-only task";
@@ -154,11 +141,8 @@ 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";
+ EXPECT_THROW(ThreadPool::create(
+ std::numeric_limits<std::size_t>::max()), std::length_error);
auto pool = ThreadPool::create(1).value();
auto empty = pool->trySubmit({});
ASSERT_TRUE(!empty) << "Reject empty task with an error";
@@ -173,3 +157,12 @@ TEST(ThreadPool, InvalidArguments)
}
}
+
+TEST(ThreadPoolDeathTest, UnexpectedTaskExceptionTerminates)
+{
+ ASSERT_DEATH({
+ auto pool = ThreadPool::create(1).value();
+ auto submitted = pool->trySubmit(fail).value();
+ submitted->get();
+ }, "");
+}