jeka/utun2
1
0 Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity

Preserve complete audio frames and reject invalid ring buffers

Browse Source
This commit is contained in:
evgeny committed 2026-10-27 03:42:03 +01:00
1 parent 7fa20ceb79
commit efe879bfd9
3 files changed
+152 -25
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

tools/chatgui/CMakeLists.txt
+5
View File
@@ -597,6 +597,11 @@ target_include_directories(test_config_updater PRIVATE ${CMAKE_SOURCE_DIR})
target_link_libraries(test_config_updater PRIVATE ${QT_CORE} ${QT_TEST_LIB} utun pthread)
add_test(NAME test_config_updater COMMAND test_config_updater)
add_executable(test_audio_frame_ring tests/test_audio_frame_ring.cpp)
target_link_libraries(test_audio_frame_ring PRIVATE ${QT_CORE} ${QT_TEST_LIB} utun pthread)
add_test(NAME test_audio_frame_ring COMMAND test_audio_frame_ring)
set_tests_properties(test_audio_frame_ring PROPERTIES TIMEOUT 30)
add_executable(test_node_config tests/test_node_config.cpp transport/node_config.cpp transport/config_updater.cpp)
target_link_libraries(test_node_config PRIVATE ${QT_LIBS} ${QT_TEST_LIB} utun pthread)
if(CMAKE_SYSTEM_NAME STREQUAL "Linux")
tools/chatgui/src/audio_frame_ring.h
+57 -25
View File
@@ -1,67 +1,99 @@
#pragma once
#include <cstdint>
#include <cstddef>
#include <cstring>
#include <mutex>
#include <stdexcept>
#include <vector>
#include "../../../lib/debug_config.h"
/* Кольцевой буфер кадров переменной длины (SPSC, mutex-protected).
/* Кольцевой буфер непрозрачных кадров переменной длины под mutex.
*
* Producer (uasync-поток) и consumer (аудио-поток). Память предвыделена одним
* блоком — ноль аллокаций на кадр. При переполнении дропается старейший кадр
* (семантика джиттер-буфера).
* push/pop/clear/size допускаются из разных потоков, включая нескольких producers/consumers.
* Данные копируются: буферы вызывающего нужны только до возврата вызова.
* Два вектора (данные и длины) выделяются конструктором, без аллокаций на кадр.
* Переполнение штатно заменяет старейший кадр. Инициализация/clear логируются в MEDIA/DEBUG,
* неверные аргументы — MEDIA/ERROR. Перед уничтожением остановить всех пользователей.
*/
class AudioFrameRing {
public:
AudioFrameRing(int capacity, int slotSize)
// capacity/slotSize > 0; неверные размеры дают invalid_argument, переполнение — length_error.
// Ошибка выделения памяти даёт стандартное исключение vector; все отказы логируются.
AudioFrameRing(int capacity, int slotSize) try
: cap_(capacity), slotSize_(slotSize),
buf_((size_t)capacity * (size_t)slotSize, 0),
lens_((size_t)capacity, 0) {}
buf_(buffer_size(capacity, slotSize), 0), lens_((size_t)capacity, 0) {
DEBUG_DEBUG(DEBUG_CATEGORY_MEDIA, "AudioFrameRing: initialized capacity=%d slot_size=%d bytes=%zu", capacity, slotSize, buf_.size());
} catch (const std::exception& error) {
DEBUG_ERROR(DEBUG_CATEGORY_MEDIA, "AudioFrameRing: initialization failed capacity=%d slot_size=%d: %s",
capacity, slotSize, error.what());
throw;
}
/* Producer: положить кадр. Возвращает false если len > slotSize.
* При переполнении дропает старейший кадр (успех). */
/* Скопировать data[0..len), 1<=len<=slotSize; false при NULL/неверной длине, очередь не меняется.
* При переполнении дропает старейший кадр (true). */
bool push(const uint8_t* data, int len) {
if (len <= 0 || len > slotSize_) return false;
if (!data || len <= 0 || len > slotSize_) {
DEBUG_ERROR(DEBUG_CATEGORY_MEDIA, "AudioFrameRing: push rejected data_present=%d len=%d slot_size=%d", data != nullptr, len, slotSize_);
return false;
}
std::lock_guard<std::mutex> lk(mtx_);
if (count_ == cap_) {
head_ = (head_ + 1) % cap_; /* drop oldest */
count_--;
}
std::memcpy(buf_.data() + (size_t)tail_ * slotSize_, data, (size_t)len);
lens_[(size_t)tail_] = (uint16_t)len;
lens_[(size_t)tail_] = len;
tail_ = (tail_ + 1) % cap_;
count_++;
return true;
}
/* Consumer: взять старейший кадр. Возвращает длину (0 = пусто). */
/* Скопировать старейший кадр и удалить его, вернуть положительную длину; 0 — пусто.
* NULL out/недостаточный outCap дают -1, кадр остаётся для повторной попытки. */
int pop(uint8_t* out, int outCap) {
std::lock_guard<std::mutex> lk(mtx_);
std::unique_lock<std::mutex> lk(mtx_);
if (count_ == 0) return 0;
int len = (int)lens_[(size_t)head_];
if (out && len > 0 && len <= outCap)
std::memcpy(out, buf_.data() + (size_t)head_ * slotSize_, (size_t)len);
if (!out || outCap < len) {
lk.unlock();
DEBUG_ERROR(DEBUG_CATEGORY_MEDIA, "AudioFrameRing: pop rejected output_present=%d capacity=%d required=%d", out != nullptr, outCap, len);
return -1;
}
std::memcpy(out, buf_.data() + (size_t)head_ * slotSize_, (size_t)len);
head_ = (head_ + 1) % cap_;
count_--;
return len;
}
void clear() {
std::lock_guard<std::mutex> lk(mtx_);
head_ = tail_ = count_ = 0;
void clear() { // удалить все кадры; память сохраняется, следующий push начинает новую очередь
int removed;
{
std::lock_guard<std::mutex> lk(mtx_);
removed = count_;
head_ = tail_ = count_ = 0;
}
DEBUG_DEBUG(DEBUG_CATEGORY_MEDIA, "AudioFrameRing: cleared frames=%d capacity=%d", removed, cap_);
}
int size() const {
int size() const { // моментальный снимок количества; между size и pop возможны другие операции
std::lock_guard<std::mutex> lk(mtx_);
return count_;
}
int capacity() const { return cap_; }
int capacity() const { return cap_; } // неизменная ёмкость в кадрах
private:
int cap_, slotSize_;
int head_ = 0, tail_ = 0, count_ = 0;
std::vector<uint8_t> buf_;
std::vector<uint16_t> lens_;
mutable std::mutex mtx_;
static size_t buffer_size(int capacity, int slotSize) {
if (capacity <= 0 || slotSize <= 0) throw std::invalid_argument("capacity and slot size must be positive");
if ((size_t)capacity > std::vector<uint8_t>().max_size() / (size_t)slotSize)
throw std::length_error("frame buffer size exceeds vector limit");
return (size_t)capacity * (size_t)slotSize;
}
const int cap_, slotSize_; // ёмкость в кадрах, максимум байтов на кадр
int head_ = 0, tail_ = 0, count_ = 0; // индексы чтения/записи и заполнение, под mtx_
std::vector<uint8_t> buf_; // cap_*slotSize_ байтов, слот выбирается индексом кадра
std::vector<int> lens_; // фактическая длина каждого занятого слота без сужения uint16_t
mutable std::mutex mtx_; // сериализует копирование и изменение индексов
};
tools/chatgui/tests/test_audio_frame_ring.cpp
+90
View File
@@ -0,0 +1,90 @@
/* Целые кадры, переполнение, ошибочные аргументы и конкурентные producers/consumer. */
#include "../src/audio_frame_ring.h"
#include <QtTest>
#include <QScopeGuard>
#include <atomic>
#include <thread>
class AudioFrameRingTest : public QObject {
Q_OBJECT
private slots:
void initTestCase() { debug_config_init(); debug_enable_console(1); debug_set_level(DEBUG_LEVEL_WARN); }
void invalidGeometry() {
for (int capacity : {-1, 0, 1}) for (int slot : {-1, 0, 1}) {
if (capacity > 0 && slot > 0) continue;
QVERIFY_EXCEPTION_THROWN((void)AudioFrameRing(capacity, slot), std::invalid_argument);
}
}
void overflowDropsOnlyOldest() {
AudioFrameRing ring(2, 1);
for (uint8_t value : {1, 2, 3}) QVERIFY(ring.push(&value, 1));
QCOMPARE(ring.size(), 2); QCOMPARE(ring.capacity(), 2);
uint8_t value = 0;
QCOMPARE(ring.pop(&value, 1), 1); QCOMPARE(value, uint8_t(2));
QCOMPARE(ring.pop(&value, 1), 1); QCOMPARE(value, uint8_t(3));
QCOMPARE(ring.pop(nullptr, 0), 0);
}
void rejectsArgumentsWithoutConsumingFrame() {
AudioFrameRing ring(3, 4);
uint8_t data[] = {1, 2, 3, 4};
QVERIFY(ring.push(data, 4));
QVERIFY(!ring.push(nullptr, 1)); QVERIFY(!ring.push(data, 0));
QVERIFY(!ring.push(data, -1)); QVERIFY(!ring.push(data, 5));
QCOMPARE(ring.pop(nullptr, 4), -1);
uint8_t output[] = {9, 9, 9, 9};
QCOMPARE(ring.pop(output, 3), -1); QCOMPARE(output[0], uint8_t(9));
QCOMPARE(ring.size(), 1);
QCOMPARE(ring.pop(output, 4), 4); QCOMPARE(memcmp(output, data, 4), 0);
}
void preservesLengthBeyondUint16() {
AudioFrameRing ring(2, 70000);
const std::vector<uint8_t> data(65536, 0x5a);
std::vector<uint8_t> output(data.size() + 1, 0xa5);
QVERIFY(ring.push(data.data(), int(data.size())));
QCOMPARE(ring.pop(output.data(), int(data.size())), int(data.size()));
QCOMPARE(memcmp(data.data(), output.data(), data.size()), 0);
QCOMPARE(output.back(), uint8_t(0xa5));
}
void clearAfterWrappedQueue() {
AudioFrameRing ring(3, 1);
uint8_t value = 7;
for (int i = 0; i < 8; ++i) QVERIFY(ring.push(&value, 1));
ring.clear(); QCOMPARE(ring.size(), 0); QCOMPARE(ring.pop(&value, 1), 0);
value = 9; QVERIFY(ring.push(&value, 1)); value = 0;
QCOMPARE(ring.pop(&value, 1), 1); QCOMPARE(value, uint8_t(9));
}
void concurrentFrameIntegrity() {
constexpr int producers = 4, perProducer = 1000, total = producers * perProducer;
AudioFrameRing ring(total, sizeof(uint32_t));
std::atomic<int> done{0};
std::atomic<bool> failed{false};
std::vector<std::thread> threads;
const auto join = qScopeGuard([&] { for (auto& thread : threads) if (thread.joinable()) thread.join(); });
for (int producer = 0; producer < producers; ++producer) threads.emplace_back([&, producer] {
for (uint32_t i = 0; i < perProducer; ++i) {
const uint32_t value = producer * perProducer + i;
if (!ring.push(reinterpret_cast<const uint8_t*>(&value), sizeof(value))) failed = true;
}
++done;
});
std::vector<bool> seen(total, false);
int received = 0;
while (done.load() < producers || ring.size()) {
uint32_t value = 0;
const int length = ring.pop(reinterpret_cast<uint8_t*>(&value), sizeof(value));
if (!length) { std::this_thread::yield(); continue; }
QCOMPARE(length, int(sizeof(value))); QVERIFY(value < total); QVERIFY(!seen[value]);
seen[value] = true; ++received;
}
QVERIFY(!failed.load()); QCOMPARE(received, total);
}
};
QTEST_GUILESS_MAIN(AudioFrameRingTest)
#include "test_audio_frame_ring.moc"