Add audio node processing and connection functionality

include/AudioNode.h

@@ -25,6 +25,13 @@
 #ifndef LMMS_AUDIO_NODE_H
 #define LMMS_AUDIO_NODE_H
 
+#include <atomic>
+#include <condition_variable>
+#include <list>
+#include <mutex>
+#include <thread>
+#include <vector>
+
 #include "lmms_basics.h"
 
 namespace lmms {
@@ -33,18 +40,55 @@ class AudioNode
 public:
 	struct Buffer
 	{
-		const sampleFrame* dest;
+		const sampleFrame* buffer;
 		size_t size;
 	};
 
-	virtual ~AudioNode() = default;
+	class Processor
+	{
+		Processor(unsigned int numWorkers = std::thread::hardware_concurrency());
+		~Processor();
+		void process(AudioNode& target);
+
+	private:
+		void populateQueue(AudioNode& target);
+		void run();
+
+		AudioNode* m_target;
+		std::list<AudioNode*> m_queue;
+		std::vector<std::thread> m_workers;
+
+		std::condition_variable m_runCond;
+		std::mutex m_runMutex;
+
+		std::atomic<bool> m_done = false;
+		std::atomic<bool> m_targetComplete = false;
+	};
+
+	AudioNode(size_t size);
+	~AudioNode();
+
+	//! Render audio for an audio period and store results in `dest` of size `size`.
+	virtual void render(const sampleFrame* dest, size_t size) = 0;
+
+	//! Mix in `src` of size `size` as input to this node's buffer.
+	//! Mixes only up to the size of the node.
+	void push(const sampleFrame* src, size_t size);
+
+	//! Connect output from this node to the input of `dest`.
+	void connect(AudioNode* dest);
 
-	//! Push a buffer as input to this node.
-	virtual void push(Buffer buffer) = 0;
+	//! Disconnect output from this node from the input of `dest`. 
+	void disconnect(AudioNode* dest);
 
-	//! Run audio processing logic for the current audio period.
-	//! Returns the result of the audio processing as a buffer.
-	virtual auto pull() -> Buffer = 0;
+private:
+	auto process() -> Buffer;
+	std::vector<sampleFrame> m_buffer;
+	std::vector<AudioNode*> m_dependencies;
+	std::vector<AudioNode*> m_destinations;
+	std::atomic<int> m_numInputs = 0;
+	std::atomic<int> m_numDependencies = 0;
+	std::mutex m_connectionMutex, m_processMutex;
 };
 } // namespace lmms
 

src/core/CMakeLists.txt

@@ -96,6 +96,7 @@ set(LMMS_SRCS
 	core/audio/AudioFileFlac.cpp
 	core/audio/AudioFileWave.cpp
 	core/audio/AudioJack.cpp
+	core/audio/AudioNode.cpp
 	core/audio/AudioOss.cpp
 	core/audio/AudioSndio.cpp
 	core/audio/AudioPort.cpp

src/core/audio/AudioNode.cpp

@@ -0,0 +1,177 @@
+/*
+ * AudioNode.cpp
+ *
+ * Copyright (c) 2024 saker <[email protected]>
+ *
+ * This file is part of LMMS - https://lmms.io
+ *
+ * This program is free software; you can redistribute it and/or
+ * modify it under the terms of the GNU General Public
+ * License as published by the Free Software Foundation; either
+ * version 2 of the License, or (at your option) any later version.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
+ * General Public License for more details.
+ *
+ * You should have received a copy of the GNU General Public
+ * License along with this program (see COPYING); if not, write to the
+ * Free Software Foundation, Inc., 51 Franklin Street, Fifth Floor,
+ * Boston, MA 02110-1301 USA.
+ *
+ */
+
+#include "AudioNode.h"
+
+#include <algorithm>
+#include <cassert>
+#include <emmintrin.h>
+#include <functional>
+#include <unordered_map>
+
+#include "MixHelpers.h"
+
+namespace lmms {
+
+AudioNode::AudioNode(size_t size)
+	: m_buffer(size)
+{
+}
+
+AudioNode::~AudioNode()
+{
+	for (const auto& dest : m_destinations)
+	{
+		disconnect(dest);
+	}
+
+	for (const auto& dependency : m_dependencies)
+	{
+		dependency->disconnect(this);
+	}
+}
+
+auto AudioNode::process() -> Buffer
+{
+	const auto lock = std::lock_guard{m_processMutex};
+	render(m_buffer.data(), m_buffer.size());
+	return {m_buffer.data(), m_buffer.size()};
+}
+
+void AudioNode::push(const sampleFrame* src, size_t size)
+{
+	const auto lock = std::lock_guard{m_processMutex};
+	MixHelpers::add(m_buffer.data(), src, m_buffer.size());
+	m_numInputs.fetch_add(1, std::memory_order_relaxed);
+}
+
+void AudioNode::connect(AudioNode* dest)
+{
+	const auto lock = std::scoped_lock{m_connectionMutex, dest->m_connectionMutex};
+	m_destinations.push_back(dest);
+	dest->m_dependencies.push_back(this);
+	dest->m_numDependencies.fetch_add(1, std::memory_order_relaxed);
+}
+
+void AudioNode::disconnect(AudioNode* dest)
+{
+	const auto lock = std::scoped_lock{m_connectionMutex, dest->m_connectionMutex};
+	m_destinations.erase(std::find(m_destinations.begin(), m_destinations.end(), dest));
+	dest->m_dependencies.erase(std::find(dest->m_dependencies.begin(), dest->m_dependencies.end(), this));
+	dest->m_numDependencies.fetch_sub(1, std::memory_order_relaxed);
+}
+
+AudioNode::Processor::Processor(unsigned int numWorkers)
+{
+	for (unsigned int i = 0; i < numWorkers; ++i)
+	{
+		m_workers.emplace_back([this] { run(); });
+	}
+}
+
+AudioNode::Processor::~Processor()
+{
+	m_done = true;
+	for (auto& worker : m_workers)
+	{
+		worker.join();
+	}
+}
+
+void AudioNode::Processor::process(AudioNode& target)
+{
+	m_target = &target;
+
+	{
+		auto lock = std::lock_guard{m_runMutex};
+		populateQueue(target);
+	}
+
+	m_runCond.notify_all();
+
+	// TODO C++20: Use std::atomic::wait
+	while (!m_targetComplete)
+	{
+		_mm_pause();
+	}
+
+	m_target = nullptr;
+	m_targetComplete.store(false, std::memory_order_relaxed);
+    m_queue.clear();
+}
+
+void AudioNode::Processor::populateQueue(AudioNode& target)
+{
+	auto temporaryMarks = std::unordered_map<AudioNode*, bool>{};
+	auto permanentMarks = std::unordered_map<AudioNode*, bool>{};
+
+	std::function<void(AudioNode*)> visit = [&](AudioNode* node) {
+		if (permanentMarks.find(node) != permanentMarks.end()) { return; }
+		assert(temporaryMarks.find(node) == temporaryMarks.end());
+
+		temporaryMarks.emplace(node, true);
+		for (auto& dependency : node->m_dependencies)
+		{
+			visit(dependency);
+		}
+
+		temporaryMarks.erase(node);
+		permanentMarks.emplace(node, true);
+		m_queue.push_front(node);
+	};
+
+	visit(&target);
+}
+
+void AudioNode::Processor::run()
+{
+	while (!m_done)
+	{
+		auto nodeToProcess = static_cast<AudioNode*>(nullptr);
+		{
+			auto lock = std::unique_lock{m_runMutex};
+			m_runCond.wait(lock, [this] { return !m_queue.empty() || m_done; });
+			if (m_done) { break; }
+
+			nodeToProcess = m_queue.front();
+			m_queue.pop_front();
+		}
+
+		while (nodeToProcess->m_numInputs < nodeToProcess->m_numDependencies)
+		{
+			_mm_pause();
+		}
+
+		const auto buffer = nodeToProcess->process();
+		for (const auto& dest : nodeToProcess->m_destinations)
+		{
+			dest->push(buffer.buffer, buffer.size);
+		}
+
+		nodeToProcess->m_numInputs = 0;
+		if (nodeToProcess == m_target) { m_targetComplete.store(true, std::memory_order_relaxed); }
+	}
+}
+
+} // namespace lmms