<commit_before>#pragma once

#include <array>
#include <atomic>
#include <condition_variable>
#include <list>
#include <mutex>
#include <thread>

template <typename F>
struct ThreadPool {
private:
	std::vector<std::thread> threads;
	std::vector<F> vector;

	std::atomic_bool joined;
	std::atomic_int running;
	std::condition_variable go_signal;
	std::condition_variable idle_signal;
	std::mutex vector_mutex;

	void run() {
		++this->running;

		while (!this->joined) {
			F job;

			{
				std::unique_lock<std::mutex> vector_lock(this->vector_mutex);

				if (this->vector.empty()) {
					--this->running;

					this->idle_signal.notify_one();

					// releases vector_lock, waits for job or join event
					this->go_signal.wait(vector_lock, [this] {
						// aquires vector_lock
						return !this->vector.empty() || this->joined;
					});
					// vector_lock is re-aquired

					if (this->joined) break;
					++this->running;
				}

				job = vector.back();
				this->vector.pop_back();
			}

			// execute
			job();
		}
	}

public:
	ThreadPool(size_t n) : joined(false), running(0) {
		this->threads.reserve(n);

		for (size_t i = 0; i < n; ++i) {
			this->threads.emplace_back(std::thread([this] { this->run(); }));
		}
	}

	~ThreadPool() {
		if (this->joined) return;
		this->join();
	}

	void push(const F f) {
		assert(!this->joined);
		std::lock_guard<std::mutex> vector_guard(this->vector_mutex);

		this->vector.push_back(f);
		this->go_signal.notify_one();
	}

	void join() {
		assert(!this->joined);
		this->wait();
		this->joined = true;

		// wake any waiting threads
		this->go_signal.notify_all();

		// join joinable threads
		for (auto &x : this->threads) {
			if (!x.joinable()) continue;

			x.join();
		}
	}

	void wait() {
		assert(!this->joined);
		if (this->running == 0) return;

		std::unique_lock<std::mutex> vector_lock(this->vector_mutex);

		this->idle_signal.wait(vector_lock, [this]() {
			return this->running == 0;
		});
	}
};<commit_msg>threadpool: wait for job pool to empty when waiting, not just running<commit_after>#pragma once

#include <array>
#include <atomic>
#include <condition_variable>
#include <list>
#include <mutex>
#include <thread>

template <typename F>
struct ThreadPool {
private:
	std::vector<std::thread> threads;
	std::vector<F> vector;

	std::atomic_bool joined;
	std::atomic_int running;
	std::condition_variable go_signal;
	std::condition_variable idle_signal;
	std::mutex vector_mutex;

	void run() {
		++this->running;

		while (!this->joined) {
			F job;

			{
				std::unique_lock<std::mutex> vector_lock(this->vector_mutex);

				if (this->vector.empty()) {
					--this->running;

					this->idle_signal.notify_one();

					// releases vector_lock, waits for job or join event
					this->go_signal.wait(vector_lock, [this] {
						// aquires vector_lock
						return !this->vector.empty() || this->joined;
					});
					// vector_lock is re-aquired

					if (this->joined) break;
					++this->running;
				}

				job = vector.back();
				this->vector.pop_back();
			}

			// execute
			job();
		}
	}

public:
	ThreadPool(size_t n) : joined(false), running(0) {
		this->threads.reserve(n);

		for (size_t i = 0; i < n; ++i) {
			this->threads.emplace_back(std::thread([this] { this->run(); }));
		}
	}

	~ThreadPool() {
		if (this->joined) return;
		this->join();
	}

	void push(const F f) {
		assert(!this->joined);
		std::lock_guard<std::mutex> vector_guard(this->vector_mutex);

		this->vector.push_back(f);
		this->go_signal.notify_one();
	}

	void join() {
		assert(!this->joined);
		this->wait();
		this->joined = true;

		// wake any waiting threads
		this->go_signal.notify_all();

		// join joinable threads
		for (auto &x : this->threads) {
			if (!x.joinable()) continue;

			x.join();
		}
	}

	void wait() {
		assert(!this->joined);

		std::unique_lock<std::mutex> vector_lock(this->vector_mutex);
		if (this->vector.empty() && this->running == 0) return;

		this->idle_signal.wait(vector_lock, [this]() {
			return this->vector.empty() && this->running == 0;
		});
	}
};
{'fraction_non_alphanumeric': 0.17629629629629628, 'fraction_numerical': 0.0019753086419753087, 'mean_word_length': 3.551685393258427, 'pattern_counts': {'":': 0, '<': 29, '<?xml version=': 0, '>': 91, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '3477078', 'n_tokens_mistral': 1679, 'n_tokens_neox': 1434, 'n_words': 291}