fix: eliminate dataraces in thread_pool, queues and kqueue

Author: fclivaz42

include/dpp/queues.h

@@ -465,6 +465,11 @@ class DPP_EXPORT request_concurrency_queue {
 	 */
 	std::shared_mutex in_mutex;
 
+	/**
+	 * @brief Removals queue mutex thread safety.
+	 */
+	std::mutex	rem_mutex;
+
 	/**
 	 * @brief Inbound queue timer. The timer is called every second,
 	 * and when it wakes up it checks for requests pending to be sent in the queue.

src/dpp/queues.cpp

@@ -153,6 +153,8 @@ void populate_result(const std::string &url, cluster* owner, http_request_comple
 /* Returns true if the request has been made */
 bool http_request::is_completed()
 {
+	// For some odd reason, BSD requires this to have a lock to prevent data races even though `completed` is std::atomic.
+	std::unique_lock<std::mutex> lock(this_captured_mutex);
 	return completed;
 }
 
@@ -281,8 +283,8 @@ http_request_completion_t http_request::run(request_concurrency_queue* processor
 					{
 						std::lock_guard<std::mutex> lock(this_captured_mutex);
 						this_captured = false;
+						this_captured_signal.notify_all();
 					}
-					this_captured_signal.notify_all();
 				});
 			}
 		);
@@ -317,7 +319,7 @@ request_concurrency_queue::request_concurrency_queue(class cluster* owner, class
 		tick_and_deliver_requests(in_index);
 		/* Clear pending removals in the removals queue */
 		if (time(nullptr) % 90 == 0) {
-			std::scoped_lock lock1{in_mutex};
+			std::scoped_lock lock1{rem_mutex};
 			for (auto it = removals.cbegin(); it != removals.cend();) {
 				if ((*it)->is_completed()) {
 					it = removals.erase(it);
@@ -409,6 +411,7 @@ void request_concurrency_queue::tick_and_deliver_requests(uint32_t index)
 			{
 				/* Find the owned pointer in requests_in */
 				std::scoped_lock lock1{in_mutex};
+				std::scoped_lock lock2{rem_mutex};
 
 				const std::string &key = request_view->endpoint;
 				auto [begin, end] = std::equal_range(requests_in.begin(), requests_in.end(), key, compare_request{});

src/dpp/socketengines/kqueue.cpp

@@ -108,6 +108,7 @@ struct DPP_EXPORT socket_engine_kqueue : public socket_engine_base {
 
 			if ((eh->flags & WANT_DELETION) != 0L) {
 				remove_socket(kev.ident);
+				std::lock_guard<std::shared_mutex>	lg(this->fds_mutex);
 				fds.erase(kev.ident);
 			}
 		}

src/dpp/thread_pool.cpp

@@ -66,20 +66,18 @@ thread_pool::~thread_pool() {
 	{
 		std::unique_lock<std::mutex> lock(queue_mutex);
 		stop = true;
+		cv.notify_all();
 	}
 
-	cv.notify_all();
 	for (auto &thread: threads) {
 		thread.join();
 	}
 }
 
 void thread_pool::enqueue(thread_pool_task task) {
-	{
-		std::unique_lock<std::mutex> lock(queue_mutex);
-		tasks.emplace(std::move(task));
-	}
+	std::unique_lock<std::mutex> lock(queue_mutex);
+	tasks.emplace(std::move(task));
 	cv.notify_one();
 }
 
-}
+}