diff --git a/doc/stream.txt b/doc/stream.txt index 9c8e43a2..93f340e6 100644 --- a/doc/stream.txt +++ b/doc/stream.txt @@ -8,7 +8,7 @@ Принятые решения: - Транспорт — существующий ETCP с быстрыми переповторами, мультилинком, шифрованием и управлением нагрузкой. -- Параметры FEC задаёт источник; 6+2 — значение по умолчанию, а не ограничение протокола. +- Параметры FEC задаёт источник; 8+0 — значение по умолчанию, а не ограничение протокола. - Источник подписывает всю пачку один раз до RS; подпись восстанавливается вместе с содержимым и проверяется после decode. - В источник добавляется циклический timestamp uint16_t в миллисекундах; он сохраняется при всей ретрансляции. Задержка оценивается относительно предыдущих данных по дельтам timestamp и времени прихода, включая полную остановку. @@ -68,7 +68,7 @@ INFO/SUBSCRIBE/REVERSE_REQ к неподписанному узлу получа Источник формирует пачки и кодирует их lib/reed_solomon: - n — полное число субпотоков, r — число проверочных, k = n-r — число исходных. -- По умолчанию n=8, r=2, k=6. Любые k кодограмм одной пачки восстанавливают её содержимое. +- По умолчанию n=8, r=0, k=8. Любые k кодограмм одной пачки восстанавливают её содержимое. - subblock от 0 до n-1 определяет субпоток; первые k блоков систематические, остальные проверочные. - batch_seq — общий 64-битный индекс пачки; блок субпотока X определяется парой (batch_seq, X). Позиция N в запросе — batch_seq, а не байтовое смещение или отдельный счётчик субпотока. @@ -78,7 +78,7 @@ INFO/SUBSCRIBE/REVERSE_REQ к неподписанному узлу получа Дескриптор содержит ключ трансляции, n, r и максимальный размер RS-кодограммы block_size. Параметры задаёт только источник и не меняет в пределах source_epoch; изменение создаёт новый запуск. В настройках desktop/Android раздел «Трансляция» сохраняет общее число субблоков и число избыточных RS. -В [chatserver] это stream_subblocks (1..255, по умолчанию 8) и stream_rs_redundant (0..n-1, по умолчанию 2). +В [chatserver] это stream_subblocks (1..255, по умолчанию 8) и stream_rs_redundant (0..n-1, по умолчанию 0). Медиапубликатор читает параметры при запуске и один раз выделяет рабочий буфер по ёмкости пачки, но не больше предела медиаблока 65536 байт. Изменение настроек действует только со следующего запуска. Дескриптор подтверждается подписанной опцией источника; поля FEC также входят в подписанную пачку. @@ -87,9 +87,9 @@ INFO/SUBSCRIBE/REVERSE_REQ к неподписанному узлу получа Максимальный размер части субпотока по умолчанию — 1300 байт: внешние метаданные 47 и RS до 1253 байт. Внутри RS находятся подписанные метаданные 55 байт, содержимое и одна 64-байтная подпись. -block_size включает 9-байтный заголовок RS seq/subblock. Максимальная длина содержимого при 6+2 — 7345 байт. +block_size включает 9-байтный заголовок RS seq/subblock. Максимальная длина содержимого при 8+0 — 9833 байт. При длине содержимого L фактический RS-size = 9 + ceil((55 + L + 64)/k), полный размер части = 47 + RS-size. -Например, L=1000 даёт восемь частей по 243 байта; L=7345 даёт восемь частей по 1300 байт. +Например, при 8+0 L=1000 даёт восемь частей по 196 байт; L=9833 даёт восемь частей по 1300 байт. Один контекст RS принимает пачки разной длины без пересоздания матрицы; все части одной пачки равны по размеру. Handle получателя, поколение stream-сессии, изменяемый путь и служебные заголовки ETCP идут сверх этого бюджета. Не обещаем соответствия одной кодограммы одному UDP-пакету: необходимую фрагментацию выполняет ETCP normalizer. @@ -743,7 +743,7 @@ Restart в той же source_epoch сохраняет consumer cursor: bootstra смешивание отвергается до изменения данных. stream_publish_finish не освобождает буфер источника. Медиаблок uAV2 находится целиком внутри входа RS, поэтому header восстанавливается вместе с данными. -При штатных 6+2 полный медиаблок <=7345 байт. Раскладка big-endian, без packed-структур: +При штатных 8+0 полный медиаблок <=9833 байт. Раскладка big-endian, без packed-структур: fixed header (32 байта): magic="uAV2":4, video_codec:1, audio_codec:1, video_count:1, audio_count:1, diff --git a/src/chat/chat_setting.c b/src/chat/chat_setting.c index 19da7b74..05bd637c 100644 --- a/src/chat/chat_setting.c +++ b/src/chat/chat_setting.c @@ -36,7 +36,7 @@ struct chat_setting_def { static const struct chat_setting_def g_setting_defs[] = { {"stream_subblocks", CHAT_SETTING_INT, 8, 1, 255}, - {"stream_rs_redundant", CHAT_SETTING_INT, 2, 0, 254}, + {"stream_rs_redundant", CHAT_SETTING_INT, 0, 0, 254}, {"storage_autoload", CHAT_SETTING_BOOL, 1, 0, 1}, {"opus_codec_preset", CHAT_SETTING_INT, 1, 0, 4}, {"radio_opus_mode", CHAT_SETTING_INT, 0, 0, 1}, diff --git a/src/stream/stream_proto.h b/src/stream/stream_proto.h index 0e4fc1ff..69399ce5 100644 --- a/src/stream/stream_proto.h +++ b/src/stream/stream_proto.h @@ -32,7 +32,7 @@ #define STREAM_BLOCK_DEFAULT_SIZE 1300 #define STREAM_RS_DEFAULT_SIZE (STREAM_BLOCK_DEFAULT_SIZE - STREAM_BLOCK_OVERHEAD) #define STREAM_DEFAULT_N 8 -#define STREAM_DEFAULT_R 2 +#define STREAM_DEFAULT_R 0 #define STREAM_DEFAULT_INPUT_CAPACITY \ ((STREAM_DEFAULT_N - STREAM_DEFAULT_R) * (STREAM_RS_DEFAULT_SIZE - RS_HEADER_SIZE) - STREAM_BATCH_OVERHEAD) #define STREAM_BLOCK_MAX_SIZE (RS_MAX_BLOCK_SIZE + STREAM_BLOCK_OVERHEAD) diff --git a/tests/test_android_stream.c b/tests/test_android_stream.c index 79f9dfa6..f574c229 100644 --- a/tests/test_android_stream.c +++ b/tests/test_android_stream.c @@ -69,7 +69,7 @@ int main(void) { debug_config_init(); debug_set_level(DEBUG_LEVEL_WARN); debug_set_category_level(DEBUG_CATEGORY_STREAM, DEBUG_LEVEL_INFO); size_t before = u_get_allocated_count(); test_clock_init(100); ua = uasync_create(); CHECK(ua); struct UTUN_INSTANCE *inst = utun_instance_create_from_str(ua, - "[global]\nauto_sockets=no\n[chatserver]\ngroup_autoconnect=0\nstream_subblocks=8\nstream_rs_redundant=0\nstorage_autoload=0\nmember_files_enabled=0\n"); + "[global]\nauto_sockets=no\n[chatserver]\ngroup_autoconnect=0\nstorage_autoload=0\nmember_files_enabled=0\n"); CHECK(inst && !utun_core_start(inst) && !chat_service_start(inst)); chat_core_create_channel_auto(inst, "Stream \"test\" 😀"); sqlite3_stmt *statement = NULL; diff --git a/tests/test_stream_core.c b/tests/test_stream_core.c index 54c5e82d..56272f35 100644 --- a/tests/test_stream_core.c +++ b/tests/test_stream_core.c @@ -20,6 +20,8 @@ static const struct stream_descriptor descriptor = { .group_id = UINT64_C(0x0123456789abcdef), .source_node_id = UINT64_C(0xfedcba9876543210), .source_epoch = UINT64_C(0x0102030405060708), .block_size = STREAM_RS_DEFAULT_SIZE, .n = 8, .r = 2 }; +/* Стенд потерь использует 6+2 независимо от настроек публикации по умолчанию. */ +#define TEST_INPUT_CAPACITY (6 * (STREAM_RS_DEFAULT_SIZE - RS_HEADER_SIZE) - STREAM_BATCH_OVERHEAD) static uint8_t private_key[32]; static unsigned cache_added, cache_removed; static void cache_observer(void *arg, const uint8_t *wire, size_t length, int added) { @@ -75,8 +77,8 @@ static void pattern(uint8_t *data, size_t length) { static void test_signed_blocks(void) { struct stream_encoder *encoder = NULL; CHECK(stream_encoder_create(&encoder, &descriptor, private_key) == STREAM_OK); - CHECK(stream_block_capacity(&descriptor) == 1300 && stream_input_size(&descriptor) == STREAM_DEFAULT_INPUT_CAPACITY); - uint8_t input[STREAM_DEFAULT_INPUT_CAPACITY]; pattern(input, sizeof(input)); + CHECK(stream_block_capacity(&descriptor) == 1300 && stream_input_size(&descriptor) == TEST_INPUT_CAPACITY); + uint8_t input[TEST_INPUT_CAPACITY]; pattern(input, sizeof(input)); const uint8_t *blocks[RS_MAX_BLOCKS]; uint64_t seq = UINT64_C(0x1122334455667788); size_t size = 0; @@ -97,7 +99,7 @@ static void test_signed_blocks(void) { } struct rs_codec *rs = NULL; CHECK(rs_init(&rs, descriptor.n, descriptor.r, descriptor.block_size) == RS_OK); - uint8_t decoded[STREAM_DEFAULT_INPUT_CAPACITY + STREAM_BATCH_OVERHEAD]; + uint8_t decoded[TEST_INPUT_CAPACITY + STREAM_BATCH_OVERHEAD]; CHECK(rs_decode(rs, received, descriptor.block_size, decoded, sizeof(decoded)) == RS_OK); CHECK(!memcmp(decoded, STREAM_BATCH_DOMAIN, 12)); CHECK(!memcmp(decoded + STREAM_BATCH_HEADER, input, sizeof(input))); @@ -184,7 +186,7 @@ static void test_wire(void) { static void test_fec_and_padding(void) { struct stream_encoder *encoder = NULL; CHECK(stream_encoder_create(&encoder, &descriptor, private_key) == STREAM_OK); - uint8_t input[STREAM_DEFAULT_INPUT_CAPACITY], output[STREAM_DEFAULT_INPUT_CAPACITY]; pattern(input, sizeof(input)); + uint8_t input[TEST_INPUT_CAPACITY], output[TEST_INPUT_CAPACITY]; pattern(input, sizeof(input)); const uint8_t *blocks[RS_MAX_BLOCKS]; size_t size = 0; CHECK(stream_encoder_encode(encoder, 0, 15, 0, input, sizeof(input), blocks, &size) == STREAM_OK && size == 1300); @@ -241,7 +243,7 @@ static void test_variable_batches(void) { struct stream_encoder *encoder = NULL; struct stream_buffer *buffer = NULL; CHECK(stream_encoder_create(&encoder, &descriptor, private_key) == STREAM_OK); CHECK(stream_buffer_create(&buffer, &descriptor, stream_encoder_public_key(encoder), 64 * 1300) == STREAM_OK); - uint8_t input[STREAM_DEFAULT_INPUT_CAPACITY], output[STREAM_DEFAULT_INPUT_CAPACITY]; pattern(input, sizeof(input)); + uint8_t input[TEST_INPUT_CAPACITY], output[TEST_INPUT_CAPACITY]; pattern(input, sizeof(input)); const size_t lengths[] = {1, 7, sizeof(input), 17, 1000, 6, 2}; size_t sizes[sizeof(lengths) / sizeof(lengths[0])], stored = 0; const uint8_t *blocks[RS_MAX_BLOCKS]; @@ -313,7 +315,7 @@ static void test_rejected_batches(void) { CHECK(stream_encoder_encode(encoder, 20, 42, 0, input, 17, blocks, &size) == STREAM_OK); const uint8_t *received[RS_MAX_BLOCKS] = {0}; for (unsigned sub = 0; sub < 8; sub++) received[sub] = blocks[sub] + STREAM_BLOCK_RS_OFF; - uint8_t decoded[STREAM_DEFAULT_INPUT_CAPACITY + STREAM_BATCH_OVERHEAD]; + uint8_t decoded[TEST_INPUT_CAPACITY + STREAM_BATCH_OVERHEAD]; size_t rs_size = size - STREAM_BLOCK_OVERHEAD, decoded_size = 6 * (rs_size - RS_HEADER_SIZE); CHECK(rs_decode(rs, received, rs_size, decoded, sizeof(decoded)) == RS_OK); struct stream_block_view view; CHECK(stream_block_parse(&descriptor, blocks[0], size, &view) == STREAM_OK); @@ -463,7 +465,7 @@ static void test_source_timer(void) { struct UASYNC *ua = uasync_create(); CHECK(ua != NULL); struct stream_source *source = NULL; struct source_events events = {0}; CHECK(stream_source_create(&source, ua, &descriptor, private_key, source_ready, &events) == STREAM_OK); - uint8_t input[STREAM_DEFAULT_INPUT_CAPACITY], output[STREAM_DEFAULT_INPUT_CAPACITY]; pattern(input, sizeof(input)); + uint8_t input[TEST_INPUT_CAPACITY], output[TEST_INPUT_CAPACITY]; pattern(input, sizeof(input)); CHECK(stream_source_input_capacity(source) == sizeof(input)); size_t accepted = 999; CHECK(stream_source_write(source, NULL, 0, &accepted) == STREAM_OK && accepted == 0); diff --git a/tests/test_stream_service.c b/tests/test_stream_service.c index c3ec49ef..82905275 100644 --- a/tests/test_stream_service.c +++ b/tests/test_stream_service.c @@ -27,6 +27,7 @@ #define GROUP1 UINT64_C(710001) #define GROUP2 UINT64_C(710002) #define SINGLE_INPUT_CAPACITY (STREAM_RS_DEFAULT_SIZE - RS_HEADER_SIZE - STREAM_BATCH_OVERHEAD) +#define PRESSURE_INPUT_CAPACITY (6 * (STREAM_RS_DEFAULT_SIZE - RS_HEADER_SIZE) - STREAM_BATCH_OVERHEAD) #define CHECK(expr) do { if (!(expr)) { \ DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "test_stream_service failed line=%d: %s", __LINE__, #expr); exit(1); \ } } while (0) @@ -46,7 +47,7 @@ static etcp_recv_fn saved_raw[4], saved_route; static int drop_renew, gap_batch; static unsigned pressure_data, overload_seen, selects_seen; static int ignore_select; -static uint8_t pressure_payload[STREAM_DEFAULT_INPUT_CAPACITY], stale_select[9 + STREAM_OVERLOAD_MAX * 8 + STREAM_WIRE_HEADER]; +static uint8_t pressure_payload[PRESSURE_INPUT_CAPACITY], stale_select[9 + STREAM_OVERLOAD_MAX * 8 + STREAM_WIRE_HEADER]; static size_t stale_select_length; static etcp_recv_fn pressure_handlers[4]; static int simulated_cpu; @@ -501,7 +502,8 @@ static void test_network(void) { CHECK(member_sync_start(nodes[i], nodes[3]->node_id, "710002", NULL, NULL) == 0); } struct STREAM_HANDLE *publisher = NULL, *subscribers[3] = {0}; - CHECK(stream_publish_open(nodes[3], GROUP2, NULL, &publisher) == STREAM_OK); + struct stream_parameters fec_parameters = {.n = 8, .r = 2, .block_size = STREAM_RS_DEFAULT_SIZE}; + CHECK(stream_publish_open(nodes[3], GROUP2, &fec_parameters, &publisher) == STREAM_OK); struct stream_info origin; CHECK(stream_get_info(publisher, &origin) == STREAM_OK); /* Подписанный END отдельно отвергает подмену подписи, индекса и группы. */ uint8_t end[STREAM_END_SIZE]; uint64_t last; @@ -675,7 +677,8 @@ static void test_network(void) { static void test_pressure(void) { memset(pressure_payload, 42, sizeof(pressure_payload)); struct STREAM_HANDLE *publisher = NULL, *subscriber = NULL; - CHECK(stream_publish_open(nodes[3], GROUP2, NULL, &publisher) == STREAM_OK); + struct stream_parameters parameters = {.n = 8, .r = 2, .block_size = STREAM_RS_DEFAULT_SIZE}; + CHECK(stream_publish_open(nodes[3], GROUP2, ¶meters, &publisher) == STREAM_OK); struct stream_info origin; CHECK(stream_get_info(publisher, &origin) == STREAM_OK); wait_state(1, 3, GROUP2, origin.descriptor.source_epoch, 0); CHECK(stream_subscribe_manual_open(nodes[1], GROUP2, nodes[3]->node_id, &subscriber) == STREAM_OK); @@ -1560,7 +1563,7 @@ static void lifecycle_event(struct UTUN_INSTANCE *inst, const struct stream_even else CHECK(event->type == STREAM_EVENT_CLOSED); } -/* Первый поставщик имеет лишь 3 из 6 нужных частей; пауза источника не оправдывает зависший bootstrap. */ +/* Первый поставщик имеет лишь 3 части пачки; пауза источника не оправдывает зависший bootstrap. */ static void test_partial_bootstrap_and_end(void) { struct STREAM_HANDLE *publisher = NULL, *relay = NULL, *viewer = NULL; CHECK(stream_publish_open(nodes[0], GROUP1, NULL, &publisher) == STREAM_OK); @@ -1755,7 +1758,8 @@ int main(int argc, char **argv) { test_profile_atomicity(); stream_set_event_callback(nodes[0], on_stream, NULL); stream_set_event_callback(nodes[1], on_stream, NULL); struct STREAM_HANDLE *publisher = NULL, *publisher2 = NULL, *other_group = NULL, *mobile = NULL, *desktop = NULL; - CHECK(stream_publish_open(nodes[0], GROUP1, NULL, &publisher) == STREAM_OK); + struct stream_parameters fec_parameters = {.n = 8, .r = 2, .block_size = STREAM_RS_DEFAULT_SIZE}; + CHECK(stream_publish_open(nodes[0], GROUP1, &fec_parameters, &publisher) == STREAM_OK); CHECK(stream_publish_open(nodes[0], GROUP2, NULL, &publisher2) == STREAM_OK); struct stream_info info; CHECK(stream_get_info(publisher, &info) == STREAM_OK); uint64_t epoch = info.descriptor.source_epoch; diff --git a/tools/chatgui-android/app/src/main/java/com/utun/chat/data/ConfigProvider.kt b/tools/chatgui-android/app/src/main/java/com/utun/chat/data/ConfigProvider.kt index ed9074c1..1ac03faf 100644 --- a/tools/chatgui-android/app/src/main/java/com/utun/chat/data/ConfigProvider.kt +++ b/tools/chatgui-android/app/src/main/java/com/utun/chat/data/ConfigProvider.kt @@ -233,7 +233,7 @@ class ConfigProvider(private val context: Context) { key == "chat.storage_maxsize_gb" -> context.dataStore.data.first()[chatStorageMaxsizeGb] ?: 1 key == "chat.opus_codec_preset" -> context.dataStore.data.first()[chatOpusCodecPreset] ?: 1 key == "chat.stream_subblocks" -> context.dataStore.data.first()[streamSubblocks] ?: 8 - key == "chat.stream_rs_redundant" -> context.dataStore.data.first()[streamRsRedundant] ?: 2 + key == "chat.stream_rs_redundant" -> context.dataStore.data.first()[streamRsRedundant] ?: 0 key == "chat.call_jitter_max_reserve_ms" -> context.dataStore.data.first()[callJitterMaxReserveMs] ?: 1000 key == "chat.compressor_enabled" -> context.dataStore.data.first()[chatCompressorEnabled] ?: 1 key == "chat.compressor_max_gain_db" -> context.dataStore.data.first()[chatCompressorMaxGainDb] ?: 15 diff --git a/tools/chatgui/src/settingsdialog.cpp b/tools/chatgui/src/settingsdialog.cpp index 5c26dc18..07fc1cda 100644 --- a/tools/chatgui/src/settingsdialog.cpp +++ b/tools/chatgui/src/settingsdialog.cpp @@ -245,7 +245,7 @@ SettingsDialog::SettingsDialog(const QString& configPath, DbManager* db, UtunNod QSettings streamConfig(m_configPath, QSettings::IniFormat); m_streamSubblocks->setValue(streamConfig.value("chatserver/stream_subblocks", 8).toInt()); m_streamRedundant->setRange(0, m_streamSubblocks->value() - 1); - m_streamRedundant->setValue(streamConfig.value("chatserver/stream_rs_redundant", 2).toInt()); + m_streamRedundant->setValue(streamConfig.value("chatserver/stream_rs_redundant", 0).toInt()); connect(m_streamSubblocks, QOverload::of(&QSpinBox::valueChanged), this, [this](int n) { m_streamRedundant->setMaximum(n - 1); });