Cleanup: use asio thread pool.

2 files changed, 51 insertions, 90 deletions

diff --git a/src/leqm_nrt.cc b/src/leqm_nrt.cc
index 6756a8d..469890c 100644
--- a/src/leqm_nrt.cc
+++ b/src/leqm_nrt.cc
@@ -23,6 +23,8 @@
 
 #include "leqm_nrt.h"
 
+#include <boost/bind.hpp>
+
 #include <stdio.h>
 #include <math.h>
 #ifdef LEQM_NRT_WITH_LIBSNDFILE
@@ -54,89 +56,62 @@ using namespace leqm_nrt;
 
 namespace leqm_nrt {
 
-class Worker
+void worker_thread(std::vector<double>& buffer, int nsamples, int nch, std::vector<double> const& ir, Sum* sum, std::vector<double> chconf)
 {
-public:
-	Worker(std::vector<double> buffer, int nsamples, int nch, std::vector<double> const& ir, Sum* sum, std::vector<double> chconf)
-	{
-		_thread = std::thread(std::bind(&Worker::process, this, buffer, nsamples, nch, ir, sum, chconf));
-	}
-
-	Worker(Worker& other) = delete;
-	bool operator=(Worker&) = delete;
-	Worker(Worker&& other) = delete;
-	bool operator=(Worker&&) = delete;
-
-	~Worker()
-	{
-		try {
-			_thread.join();
-		} catch (...)
-		{
+	auto accumulate_ch = [](std::vector<double>& ch_accumulator, std::vector<double> const& input_channel, int nsamples) {
+		for (auto i = 0; i < nsamples; i++) {
+			ch_accumulator[i] += input_channel[i];
 		}
-	}
-
-
-private:
-	void process(std::vector<double>& buffer, int nsamples, int nch, std::vector<double> const& ir, Sum* sum, std::vector<double> chconf)
-	{
-		auto accumulate_ch = [](std::vector<double>& ch_accumulator, std::vector<double> const& input_channel, int nsamples) {
-			for (auto i = 0; i < nsamples; i++) {
-				ch_accumulator[i] += input_channel[i];
-			}
-		};
-
-		auto convolve = [](std::vector<double> const& signal, std::vector<double> const& ir) {
-			std::vector<double> result(signal.size());
-			double sum = 0.0;
-			for (auto i = 0U; i < signal.size(); i++) {
-				auto m = i;
-				for (int l = ir.size()- 1; l >= 0; l--, m++) {
-					if (m >= signal.size()) {
-						m -= signal.size();
-					}
-					sum += signal[m] * ir[l];
+	};
+
+	auto convolve = [](std::vector<double> const& signal, std::vector<double> const& ir) {
+		std::vector<double> result(signal.size());
+		double sum = 0.0;
+		for (auto i = 0U; i < signal.size(); i++) {
+			auto m = i;
+			for (int l = ir.size()- 1; l >= 0; l--, m++) {
+				if (m >= signal.size()) {
+					m -= signal.size();
 				}
-				result[i] = sum;
-				sum = 0.0;
+				sum += signal[m] * ir[l];
 			}
-			return result;
-		};
+			result[i] = sum;
+			sum = 0.0;
+		}
+		return result;
+	};
 
-		auto rectify = [](std::vector<double> const& input) {
-			std::vector<double> squared;
-			for (auto i: input) {
-				squared.push_back(pow(i, 2));
-			}
-			return squared;
-		};
+	auto rectify = [](std::vector<double> const& input) {
+		std::vector<double> squared;
+		for (auto i: input) {
+			squared.push_back(pow(i, 2));
+		}
+		return squared;
+	};
 
-		/* Round-up in case nsamples is not a multiple of nch */
-		int const frames = (nsamples + nch - 1) / nch;
+	/* Round-up in case nsamples is not a multiple of nch */
+	int const frames = (nsamples + nch - 1) / nch;
 
-		std::vector<double> ch_sum_accumulator_norm(frames);
-		std::vector<double> ch_sum_accumulator_conv(frames);
+	std::vector<double> ch_sum_accumulator_norm(frames);
+	std::vector<double> ch_sum_accumulator_conv(frames);
 
-		for (int ch = 0; ch < nch; ch++) {
+	for (int ch = 0; ch < nch; ch++) {
 
-			std::vector<double> normalized_buffer(frames);
+		std::vector<double> normalized_buffer(frames);
 
-			for (int n = ch, m = 0; n < nsamples; n += nch, m++) {
-				// use this for calibration depending on channel config for ex. chconf[6] = {1.0, 1.0, 1.0, 1.0, 0.707945784, 0.707945784} could be the default for 5.1 soundtracks
-				//so not normalized but calibrated
-				normalized_buffer[m] = buffer[n] * chconf[ch]; //this scale amplitude according to specified calibration
-			}
-
-			auto convolved_buffer = convolve(normalized_buffer, ir);
-			accumulate_ch(ch_sum_accumulator_norm, rectify(normalized_buffer), frames);
-			accumulate_ch(ch_sum_accumulator_conv, rectify(convolved_buffer), frames);
+		for (int n = ch, m = 0; n < nsamples; n += nch, m++) {
+			// use this for calibration depending on channel config for ex. chconf[6] = {1.0, 1.0, 1.0, 1.0, 0.707945784, 0.707945784} could be the default for 5.1 soundtracks
+			//so not normalized but calibrated
+			normalized_buffer[m] = buffer[n] * chconf[ch]; //this scale amplitude according to specified calibration
 		}
 
-		sum->sum_samples(ch_sum_accumulator_norm, ch_sum_accumulator_conv, frames);
+		auto convolved_buffer = convolve(normalized_buffer, ir);
+		accumulate_ch(ch_sum_accumulator_norm, rectify(normalized_buffer), frames);
+		accumulate_ch(ch_sum_accumulator_conv, rectify(convolved_buffer), frames);
 	}
 
-	std::thread _thread;
-};
+	sum->sum_samples(ch_sum_accumulator_norm, ch_sum_accumulator_conv, frames);
+}
 
 }
 
@@ -327,6 +302,7 @@ Calculator::Calculator(
 	: _channels(channels)
 	, _channel_corrections(channel_corrections)
 	, _num_cpu(num_cpu)
+	, _pool(num_cpu)
 {
 	if ((sample_rate * buffer_size_ms) % 1000) {
 		throw BadBufferSizeError();
@@ -350,20 +326,7 @@ void Calculator::process_buffer()
 		return;
 	}
 
-	_workers.push_back(
-			std::make_shared<Worker>(
-				_buffer,
-				_buffer_free_offset,
-				_channels,
-				_ir,
-				&_sum,
-				_channel_corrections
-				)
-			);
-
-	if (static_cast<int>(_workers.size()) == _num_cpu) {
-		_workers.clear();
-	}
+	boost::asio::post(_pool, boost::bind(&worker_thread, _buffer, _buffer_free_offset, _channels, _ir, &_sum, _channel_corrections));
 
 	_buffer_free_offset = 0;
 }
@@ -391,7 +354,7 @@ void Calculator::add(std::vector<double> samples)
 double Calculator::leq_m()
 {
 	process_buffer();
-	_workers.clear();
+	_pool.join();
 
 	return _sum.leqm();
 }
@@ -400,7 +363,7 @@ double Calculator::leq_m()
 double Calculator::leq_nw()
 {
 	process_buffer();
-	_workers.clear();
+	_pool.join();
 
 	return _sum.rms();
 }
diff --git a/src/leqm_nrt.h b/src/leqm_nrt.h
index 7a0f3f8..b3ade0e 100644
--- a/src/leqm_nrt.h
+++ b/src/leqm_nrt.h
@@ -23,6 +23,7 @@
 
 #pragma once
 
+#include <boost/asio.hpp>
 #include <string>
 #include <vector>
 #include <functional>
@@ -152,9 +153,6 @@ public:
 };
 
 
-class Worker;
-
-
 class Calculator
 {
 public:
@@ -170,7 +168,7 @@ public:
 
 	~Calculator()
 	{
-		_workers.clear();
+		_pool.join();
 	}
 
 	Calculator(Calculator&) = delete;
@@ -189,7 +187,7 @@ private:
 	int _channels;
 	std::vector<double> _channel_corrections;
 	int _num_cpu;
-	std::vector<std::shared_ptr<Worker>> _workers;
+	boost::asio::thread_pool _pool;
 	Sum _sum;
 	std::vector<double> _ir;
 	std::vector<double> _buffer;