correct(?) audio "capture-then-playback" code (pulseaudio-simple)

README.asciidoc

so, finally, I fixed the bug.

highlights

auto playback = pa_simple_new(nullptr, "jopa-ng",
		PA_STREAM_PLAYBACK, nullptr, "playback", &ss,
		nullptr,
		nullptr,  // (1)
		nullptr);

pa_buffer_attr ba {
	.maxlength = uint32_t(-1),
	.fragsize  = buf_nbyte,
};
auto capture = pa_simple_new(nullptr, "jopa-ng",
		PA_STREAM_RECORD, nullptr, "capture", &ss,
		nullptr,
		&ba,      // (2)
		nullptr);

1. use default buffer attributes for playback.

2. use custom buffer attributes for capturing. you should only set maxlength and fragsize in the buffer attributes.

pa.cc

#include <thread>
#include <array>
#include <chrono>
#include <pulse/simple.h>
#include <atomic>
#include <algorithm>
#include <mutex>
#include <condition_variable>

#include <string>
#include <iostream>
using std::cin;
using std::cout;
using std::cerr;
using std::endl;

//////////////////////////////////////////////////////////////////

auto mux(std::thread&& a, std::thread&& b)
{
	return std::thread{
		[](std::thread&& a, std::thread&& b) {
			if (a.joinable()) a.join();
			if (b.joinable()) b.join();
		},
		std::move(a), std::move(b)
	};
}

auto mux(std::thread&& a, std::thread&& b, std::thread&& c)
{
	return mux(mux(std::move(a), std::move(b)), std::move(c));
}


//////////////////////////////////////////////////////////////////


struct sync_once
{
	using mutex_type = std::mutex;
	using condv_type = std::condition_variable;
	using lock_type = std::unique_lock<mutex_type>;

	void fire()
	{
		{
			lock_type _(m);
			wake = true;
		}
		cv.notify_all();
	}

	void wait()
	{
		lock_type _(m);
		cv.wait(_, [this] { return wake; });
	}

private:
	bool wake{false};
	mutex_type m;
	condv_type cv;
};


//////////////////////////////////////////////////////////////////

// single reader single writer wait-free ring buffer
template <int Capacity>
struct ringbuffer
{
	static constexpr auto capacity = Capacity;
	using value_type = std::array<float, capacity>;

	bool write(float const buf[], int len) noexcept
	{
		if (free() < len) return true;

		auto bend = &buf[len];
		auto pend = end();
		auto p = tbegin();
		auto p0 = begin();
		for (; buf != bend; buf++) {
			*p++ = *buf;
			if (p == pend) p = p0;
		}
		tail  = p - p0;

		return false;
	}

	bool read(float buf[], int len) noexcept
	{
		if (size() < len) return true;

		auto bend = &buf[len];
		auto pend = end();
		auto p = hbegin();
		auto p0 = begin();
		for (; buf != bend; buf++) {
			*buf = *p++;
			if (p == pend) p = p0;
		}
		head  = p - p0;

		return false;
	}

	int size() const noexcept
	{
		auto s = tail-head;
		if (s < 0) s += capacity;
		return s;
	}
	int free() const noexcept { return capacity-size(); }

private:
	value_type value;
	volatile int head{};
	volatile int tail{};

	float const*    end() const noexcept { return &value[capacity]; }
	float const* hbegin() const noexcept { return &value[head    ]; }
	float      * tbegin()       noexcept { return &value[tail    ]; }
	float      *  begin()       noexcept { return &value[       0]; }
	float const*  begin() const noexcept { return &value[       0]; }
};


//////////////////////////////////////////////////////////////////


struct pulseaudio
{
	using clock = std::chrono::high_resolution_clock;

	pulseaudio(int sample_rate) :
		ss {
			.format   = PA_SAMPLE_FLOAT32,
			.rate     = (unsigned int)sample_rate,
			.channels = 2,
		}
	{
		//*
		// playback
		std::thread playback([this]() {
			auto playback = pa_simple_new(nullptr, "jopa-ng",
					PA_STREAM_PLAYBACK, nullptr, "playback", &ss,
					nullptr, nullptr, nullptr);
			float buf[buf_len] = {0};
			sync.wait();

			// wait until there are 10 seconds data recorded
			while (running && ring.size() < 48000*10*2) {
				if (pa_simple_write(playback, buf, buf_nbyte, nullptr))
					cerr << "write failed" << endl;
				counter_playback++;
			}

			counter_playback = counter_capture;	// catch up
			while (running) {
				auto start = clock::now();
				ring.read(buf, buf_len);
				if (pa_simple_write(playback, buf, buf_nbyte, nullptr))
					cerr << "write failed" << endl;
				counter_playback++;

				auto dura = clock::now()-start;
				auto ns = std::chrono::duration_cast<std::chrono::nanoseconds>(dura);
				benchmark_playback = ns.count() * 1e-6;
			}
			pa_simple_free(playback);
		});

		// capture
		std::thread capture([this]() {
			pa_buffer_attr ba {
				.maxlength = uint32_t(-1),
				.fragsize  = buf_nbyte,
			};
			auto capture = pa_simple_new(nullptr, "jopa-ng",
					PA_STREAM_RECORD, nullptr, "capture", &ss,
					nullptr, &ba, nullptr);
			float buf[buf_len] = {0};
			sync.wait();
			while (running) {
				auto start = clock::now();
				if (pa_simple_read(capture, buf, buf_nbyte, nullptr))
					cerr << "read failed" << endl;
				counter_capture++;

				auto dura = clock::now()-start;
				auto ns = std::chrono::duration_cast<std::chrono::nanoseconds>(dura);
				benchmark_capture = ns.count() * 1e-6;
				ring.write(buf, buf_len);
			}
			pa_simple_free(capture);
		});
		// */

		// counter
		std::thread counter([this]() {
			using namespace std::literals;
			sync.wait();
			while (running) {
				auto c = counter_capture;
				auto p = counter_playback;
				auto d = c-p;

				cout
					<< "\r\e[1;31m" << c << "\e[0m"
					<< "  \e[1;32m" << p << "\e[0m"
					<< "  \e[1;3" << (d > 0 ? "1m+" : d < 0 ? "1m-" : "2m*") << "\e[0m"
					<< "  \e[1;33m" << d << "\e[0m"
					<< "  \e[1;31m" << benchmark_capture  << "\e[0m"
					<< "  \e[1;32m" << benchmark_playback << "\e[0m"
					<< "  \e[1;34m" << ring.size() << "\e[0m"
					<< "                ";
				cout.flush();
				std::this_thread::sleep_for(10ms);
			}
		});
		th = mux(std::move(playback), std::move(capture), std::move(counter));
	}

	~pulseaudio()
	{
		running = false;
		if (th.joinable()) th.join();
	}

	void start()
	{
		running = true;
		sync.fire();
	}

private:
	sync_once sync;
	std::thread th;
	volatile bool running;
	volatile int counter_playback{0};
	volatile int counter_capture {0};
	volatile double benchmark_playback{0};
	volatile double benchmark_capture {0};

	pa_sample_spec ss;
	static constexpr auto buf_len_mono = 128;						// how many floats in a mono track
	static constexpr auto buf_len      = buf_len_mono*2;			// how many floats in a stereo track
	static constexpr auto buf_nbyte    = buf_len*sizeof(float);		// how many bytes  in a stereo track
	ringbuffer<48000*10*4> ring;
};


//////////////////////////////////////////////////////////////////
//////////////////////////////////////////////////////////////////


int main()
{
	cout << "starting..." << endl;
	pulseaudio pa{48000};
	pa.start();

	cout << "ready." << endl;
	while (true) {
		std::string cmd;
		std::getline(cin, cmd);
		if (cmd == "q" || cmd == "quit" || !cin)
			break;
	}
	cout << "exiting..." << endl;
}