Use std::future over std::thread for background jobsHEADmain

Easier checking if a job has completed [successfully] and reseting state
for the next time.

3 files changed, 16 insertions, 7 deletions

diff --git a/src/ingestor.cpp b/src/ingestor.cpp
index ba89991..692ddd2 100644
--- a/src/ingestor.cpp
+++ b/src/ingestor.cpp
@@ -223,17 +223,21 @@ namespace WebStat {
 	Ingestor::runJobsAsNeeded()
 	{
 		auto runJobAsNeeded = [this, now = Job::LastRunTime::clock::now()](Job & job, auto freq) {
-			if (!job.currentRun && expired(job.lastRun, freq, now)) {
-				job.currentRun.emplace([this, now, freq, &job]() {
+			if (job.currentRun) {
+				if (job.currentRun->valid()) {
 					try {
-						(this->*job.impl)();
+						job.currentRun->get();
 						job.lastRun = now;
 					}
 					catch (const std::exception &) {
 						// Error, retry in half the frequency
 						job.lastRun = now - (freq / 2);
 					}
-				});
+					job.currentRun.reset();
+				}
+			}
+			else if (expired(job.lastRun, freq, now)) {
+				job.currentRun.emplace(std::async(job.impl, this));
 			}
 		};
 		runJobAsNeeded(ingestParkedLines, settings.freqIngestParkedLines);
diff --git a/src/ingestor.hpp b/src/ingestor.hpp
index c5628d6..0bf2297 100644
--- a/src/ingestor.hpp
+++ b/src/ingestor.hpp
@@ -8,6 +8,7 @@
 #include <connection_fwd.h>
 #include <cstdio>
 #include <flat_set>
+#include <future>
 #include <scn/scan.h>
 #include <span>
 #include <sys/utsname.h>
@@ -78,7 +79,7 @@ namespace WebStat {
 
 			const Impl impl;
 			LastRunTime lastRun {LastRunTime::clock::now()};
-			std::optional<std::thread> currentRun {std::nullopt};
+			std::optional<std::future<unsigned int>> currentRun;
 		};
 
 		Job::LastRunTime lastCheckedJobs {Job::LastRunTime::clock::now()};
diff --git a/test/test-ingest.cpp b/test/test-ingest.cpp
index 558444d..41b3690 100644
--- a/test/test-ingest.cpp
+++ b/test/test-ingest.cpp
@@ -311,7 +311,9 @@ BOOST_AUTO_TEST_CASE(IngestParkedJob,
 	ingestParkedLines.lastRun = now - settings.freqIngestParkedLines - 1s;
 	runJobsAsNeeded();
 	BOOST_REQUIRE(ingestParkedLines.currentRun);
-	ingestParkedLines.currentRun->join();
+	ingestParkedLines.currentRun->wait();
+	runJobsAsNeeded();
+	BOOST_REQUIRE(!ingestParkedLines.currentRun);
 	BOOST_CHECK_EQUAL(linesParsed, 1);
 	BOOST_CHECK_EQUAL(linesDiscarded, 0);
 	BOOST_CHECK_GE(ingestParkedLines.lastRun, now);
@@ -326,7 +328,9 @@ BOOST_AUTO_TEST_CASE(JobErrorRescheduler, *boost::unit_test::depends_on("I/Inges
 	std::filesystem::permissions(settings.fallbackDir / LOGLINE1_PARKED, std::filesystem::perms::owner_write);
 	runJobsAsNeeded();
 	BOOST_REQUIRE(ingestParkedLines.currentRun);
-	ingestParkedLines.currentRun->join();
+	ingestParkedLines.currentRun->wait();
+	runJobsAsNeeded();
+	BOOST_REQUIRE(!ingestParkedLines.currentRun);
 	BOOST_CHECK(std::filesystem::exists(settings.fallbackDir / LOGLINE1_PARKED));
 	BOOST_CHECK_GE(ingestParkedLines.lastRun, now - (settings.freqIngestParkedLines / 2) - 1s);
 	BOOST_CHECK_LE(ingestParkedLines.lastRun, now - (settings.freqIngestParkedLines / 2) + 1s);