Skip to content

C
car-controlserver
Commit 3585cc8c authored by 957dd's avatar 957dd
Browse files
Options

c语音推流使用cpu软件解码，硬件h264编码，提升流畅度

parent 4687e6c8
Show whitespace changes
Inline Side-by-side
Showing with 2026 additions and 439 deletions
build/main
View file @ 3585cc8c
No preview for this file type
drivers/sensors/audioplay.c
View file @ 3585cc8c
#include "common.h" #include "common.h"
#include "audioplay.h" #include "audioplay.h"
#include "audio_sink.h" /* USB声卡排队锁 */
#include "device_identity.h" #include "device_identity.h"
#include "mqtt_init.h" #include "mqtt_init.h"
#include "http_config_mqtt.h" #include "http_config_mqtt.h"
#include "audiotts_play.h" #include "audiotts_play.h"
#include "wifi_autoconfig.h" #include "wifi_autoconfig.h"
#include <stdio.h> #include <stdio.h>
#include <pthread.h>
#include <strings.h> #include <strings.h>
#include <unistd.h> #include <unistd.h>
#include <sys/wait.h> #include <sys/wait.h>
#define AUDIO_USB_ALSA_DEVICE "hw:2,0" #define AUDIO_USB_ALSA_DEVICE "hw:2,0"
#define AUDIO_LOCAL_ALSA_DEVICE "plughw:2,0"
#define AUDIO_LOCAL_PLAY_TIMEOUT_SEC 8
static int s_audio_status=7; static int s_audio_status=7;
static char s_urlbuf[512]; static char s_urlbuf[512];
...@@ -21,6 +25,7 @@ static double s_audio_volume=0.8; ...@@ -21,6 +25,7 @@ static double s_audio_volume=0.8;
static int s_local_play_pending = 0; static int s_local_play_pending = 0;
static char s_local_filepath[512]; static char s_local_filepath[512];
static pthread_mutex_t s_local_play_mutex = PTHREAD_MUTEX_INITIALIZER;
static int local_is_cn_lang(const char *lang) { static int local_is_cn_lang(const char *lang) {
return lang != NULL && (strcmp(lang, AUDIO_LANG_ZH) == 0 || strcmp(lang, "cn") == 0); return lang != NULL && (strcmp(lang, AUDIO_LANG_ZH) == 0 || strcmp(lang, "cn") == 0);
...@@ -74,12 +79,16 @@ static int local_resolve_filepath(const char *filename, const char *language_ove ...@@ -74,12 +79,16 @@ static int local_resolve_filepath(const char *filename, const char *language_ove
} }
static void local_queue_play(const char *filename, const char *language) { static void local_queue_play(const char *filename, const char *language) {
if (!local_resolve_filepath(filename, language, s_local_filepath, sizeof(s_local_filepath))) { char resolved[512];
if (!local_resolve_filepath(filename, language, resolved, sizeof(resolved))) {
my_zlog_warn("2017 本地音频不存在: %s", filename); my_zlog_warn("2017 本地音频不存在: %s", filename);
return; return;
} }
pthread_mutex_lock(&s_local_play_mutex);
snprintf(s_local_filepath, sizeof(s_local_filepath), "%s", resolved);
s_local_play_pending = 1; s_local_play_pending = 1;
my_zlog_info("2017 已排队本地音频: %s", s_local_filepath); pthread_mutex_unlock(&s_local_play_mutex);
my_zlog_info("2017 已排队本地音频: %s", resolved);
} }
void audioplay_local_mqtt_receive(cJSON *body) { void audioplay_local_mqtt_receive(cJSON *body) {
...@@ -143,6 +152,76 @@ static double audioplay_volume_clamp(double v) { ...@@ -143,6 +152,76 @@ static double audioplay_volume_clamp(double v) {
int audio_wheat_init(); int audio_wheat_init();
static int audio_system_exit_code(int status) {
if (status == -1) {
return -1;
}
if (WIFEXITED(status)) {
return WEXITSTATUS(status);
}
if (WIFSIGNALED(status)) {
return 128 + WTERMSIG(status);
}
return -1;
}
static void shell_single_quote(char *out, size_t size, const char *in) {
size_t pos = 0;
if (!out || size == 0) {
return;
}
out[pos++] = '\'';
if (in) {
for (const char *p = in; *p && pos + 5 < size; p++) {
if (*p == '\'') {
const char *esc = "'\\''";
for (const char *e = esc; *e && pos + 1 < size; e++) {
out[pos++] = *e;
}
} else {
out[pos++] = *p;
}
}
}
if (pos + 1 < size) {
out[pos++] = '\'';
}
out[pos] = '\0';
}
static int play_local_audio_file(const char *filepath) {
char quoted_path[1024];
char command[2048];
int ret;
int exit_code;
shell_single_quote(quoted_path, sizeof(quoted_path), filepath);
/* 排队等待USB声卡: 如果audio_sink正在播放手机音频, 等它释放 */
audio_sink_lock_alsa();
snprintf(command, sizeof(command),
/* sync=true 防止 EOS 提前关 ALSA 丢尾部(只播前半段); channels=2 上混规避 USB 声卡单声道 ring_buffer CRITICAL */
"timeout %ds gst-launch-1.0 -q filesrc location=%s ! mpegaudioparse ! mpg123audiodec ! audioconvert ! audioresample ! audio/x-raw,channels=2 ! alsasink device=%s sync=true >/dev/null 2>&1",
AUDIO_LOCAL_PLAY_TIMEOUT_SEC, quoted_path, AUDIO_LOCAL_ALSA_DEVICE);
ret = system(command);
exit_code = audio_system_exit_code(ret);
if (exit_code != 0) {
my_zlog_warn("本地音频 GStreamer 播放失败 exit=%d，尝试 ffplay: %s",
exit_code, filepath);
snprintf(command, sizeof(command),
"timeout %ds ffplay -nodisp -autoexit -loglevel warning %s",
AUDIO_LOCAL_PLAY_TIMEOUT_SEC, quoted_path);
ret = system(command);
exit_code = audio_system_exit_code(ret);
}
audio_sink_unlock_alsa();
return exit_code;
}
//接收音频播放 //接收音频播放
void audioplay_mqtt_receive(cJSON *json) { void audioplay_mqtt_receive(cJSON *json) {
// 解析"audioLink"字段（修正了原始JSON中的拼写错误） // 解析"audioLink"字段（修正了原始JSON中的拼写错误）
...@@ -212,7 +291,7 @@ void audioplay_send_mqtt() { ...@@ -212,7 +291,7 @@ void audioplay_send_mqtt() {
//音频播放 //音频播放
void audioplay_cycle(){ void audioplay_cycle(){
char command[1024]; char command[2048];
int ret; int ret;
while(1){ while(1){
if(s_audio_status==0){ if(s_audio_status==0){
...@@ -241,16 +320,25 @@ void audioplay_cycle(){ ...@@ -241,16 +320,25 @@ void audioplay_cycle(){
audioplay_send_mqtt(); audioplay_send_mqtt();
} }
char local_filepath[sizeof(s_local_filepath)];
int local_play_pending = 0;
pthread_mutex_lock(&s_local_play_mutex);
if (s_local_play_pending) { if (s_local_play_pending) {
s_local_play_pending = 0; s_local_play_pending = 0;
snprintf(command, sizeof(command), snprintf(local_filepath, sizeof(local_filepath), "%s", s_local_filepath);
"ffplay -nodisp -autoexit -loglevel quiet \"%s\"", s_local_filepath); local_play_pending = 1;
my_zlog_debug("播放本地音频: %s", s_local_filepath); }
ret = system(command); pthread_mutex_unlock(&s_local_play_mutex);
if (WIFEXITED(ret) && WEXITSTATUS(ret) == 0) {
my_zlog_debug("本地音频播放完成: %s", s_local_filepath); if (local_play_pending) {
int exit_code;
my_zlog_info("播放本地音频: %s", local_filepath);
exit_code = play_local_audio_file(local_filepath);
if (exit_code == 0) {
my_zlog_debug("本地音频播放完成: %s", local_filepath);
} else { } else {
my_zlog_warn("本地音频播放失败: %s", s_local_filepath); my_zlog_warn("本地音频播放失败 exit=%d: %s", exit_code, local_filepath);
} }
} }
...@@ -374,7 +462,7 @@ int audio_speaker_init() { ...@@ -374,7 +462,7 @@ int audio_speaker_init() {
int audio_init(){ int audio_init(){
delay_s(5); delay_s(1);
audio_wheat_init(); audio_wheat_init();
delay_s(1); delay_s(1);
audio_speaker_init(); audio_speaker_init();
......
drivers/sensors/audioplay.c.bak.20260710_111731 0 → 100644
View file @ 3585cc8c
#include "common.h"
#include "audioplay.h"
#include "audio_sink.h" /* USB声卡排队锁 */
#include "device_identity.h"
#include "mqtt_init.h"
#include "http_config_mqtt.h"
#include "audiotts_play.h"
#include "wifi_autoconfig.h"
#include <stdio.h>
#include <pthread.h>
#include <strings.h>
#include <unistd.h>
#include <sys/wait.h>
#define AUDIO_USB_ALSA_DEVICE "hw:2,0"
#define AUDIO_LOCAL_ALSA_DEVICE "plughw:2,0"
#define AUDIO_LOCAL_PLAY_TIMEOUT_SEC 8
static int s_audio_status=7;
static char s_urlbuf[512];
static char s_keybuf[256];
static double s_audio_volume=0.8;
#define AUDIO_DRIVING_LEVEL_MAX 30
static int s_local_play_pending = 0;
static char s_local_filepath[512];
static pthread_mutex_t s_local_play_mutex = PTHREAD_MUTEX_INITIALIZER;
static int local_is_cn_lang(const char *lang) {
return lang != NULL && (strcmp(lang, AUDIO_LANG_ZH) == 0 || strcmp(lang, "cn") == 0);
}
static int local_is_en_lang(const char *lang) {
return lang != NULL &&
(strcmp(lang, AUDIO_LANG_EN) == 0 || strcmp(lang, "en") == 0 || strcmp(lang, "EN") == 0);
}
static void local_resolve_lang(const char *language_override, char *out, size_t size) {
if (language_override == NULL || language_override[0] == '\0' ||
strcasecmp(language_override, "default") == 0) {
snprintf(out, size, "%s", audio_get_language());
return;
}
if (local_is_cn_lang(language_override)) {
snprintf(out, size, "%s", AUDIO_LANG_ZH);
return;
}
if (local_is_en_lang(language_override)) {
snprintf(out, size, "%s", AUDIO_LANG_EN);
return;
}
snprintf(out, size, "%s", audio_get_language());
my_zlog_warn("2017 未知 language: %s，使用当前语言: %s", language_override, out);
}
static int local_resolve_filepath(const char *filename, const char *language_override,
char *out, size_t size) {
char lang[8];
int use_default_lang = (language_override == NULL || language_override[0] == '\0' ||
strcasecmp(language_override, "default") == 0);
local_resolve_lang(language_override, lang, sizeof(lang));
if (local_is_cn_lang(lang)) {
snprintf(out, size, "%s/%s", AUDIO_LOCAL_BASE, filename);
return access(out, F_OK) == 0;
}
snprintf(out, size, "%s/%s/%s", AUDIO_LOCAL_BASE, lang, filename);
if (access(out, F_OK) == 0) {
return 1;
}
if (use_default_lang) {
snprintf(out, size, "%s/%s", AUDIO_LOCAL_BASE, filename);
return access(out, F_OK) == 0;
}
return 0;
}
static void local_queue_play(const char *filename, const char *language) {
char resolved[512];
if (!local_resolve_filepath(filename, language, resolved, sizeof(resolved))) {
my_zlog_warn("2017 本地音频不存在: %s", filename);
return;
}
pthread_mutex_lock(&s_local_play_mutex);
snprintf(s_local_filepath, sizeof(s_local_filepath), "%s", resolved);
s_local_play_pending = 1;
pthread_mutex_unlock(&s_local_play_mutex);
my_zlog_info("2017 已排队本地音频: %s", resolved);
}
void audioplay_local_mqtt_receive(cJSON *body) {
if (!cJSON_IsObject(body)) {
my_zlog_warn("2017 body 无效");
return;
}
cJSON *status = cJSON_GetObjectItemCaseSensitive(body, "status");
cJSON *level_item = cJSON_GetObjectItemCaseSensitive(body, "level");
cJSON *language = cJSON_GetObjectItemCaseSensitive(body, "language");
if (!cJSON_IsString(status) || status->valuestring == NULL) {
my_zlog_warn("2017 缺少 status");
return;
}
const char *lang = "default";
if (cJSON_IsString(language) && language->valuestring != NULL) {
lang = language->valuestring;
}
if (strcmp(status->valuestring, "speed") == 0) {
local_queue_play("Speedupforoneminute.mp3", lang);
return;
}
if (strcmp(status->valuestring, "drivinglevel") == 0) {
int level = 0;
if (cJSON_IsString(level_item) && level_item->valuestring != NULL) {
level = atoi(level_item->valuestring);
} else if (cJSON_IsNumber(level_item)) {
level = level_item->valueint;
} else {
my_zlog_warn("2017 drivinglevel 缺少 level");
return;
}
if (level < 1 || level > AUDIO_DRIVING_LEVEL_MAX) {
my_zlog_warn("2017 drivinglevel level 无效: %d (1-%d)",
level, AUDIO_DRIVING_LEVEL_MAX);
return;
}
char filename[48];
snprintf(filename, sizeof(filename), "DrivingLevel%d.mp3", level);
local_queue_play(filename, lang);
return;
}
my_zlog_warn("2017 未知 status: %s", status->valuestring);
}
static double audioplay_volume_clamp(double v) {
if (v < 0.0) {
return 0.0;
}
if (v > 2.0) {
return 2.0;
}
return v;
}
int audio_wheat_init();
static int audio_system_exit_code(int status) {
if (status == -1) {
return -1;
}
if (WIFEXITED(status)) {
return WEXITSTATUS(status);
}
if (WIFSIGNALED(status)) {
return 128 + WTERMSIG(status);
}
return -1;
}
static void shell_single_quote(char *out, size_t size, const char *in) {
size_t pos = 0;
if (!out || size == 0) {
return;
}
out[pos++] = '\'';
if (in) {
for (const char *p = in; *p && pos + 5 < size; p++) {
if (*p == '\'') {
const char *esc = "'\\''";
for (const char *e = esc; *e && pos + 1 < size; e++) {
out[pos++] = *e;
}
} else {
out[pos++] = *p;
}
}
}
if (pos + 1 < size) {
out[pos++] = '\'';
}
out[pos] = '\0';
}
static int play_local_audio_file(const char *filepath) {
char quoted_path[1024];
char command[2048];
int ret;
int exit_code;
shell_single_quote(quoted_path, sizeof(quoted_path), filepath);
/* 排队等待USB声卡: 如果audio_sink正在播放手机音频, 等它释放 */
audio_sink_lock_alsa();
snprintf(command, sizeof(command),
"timeout %ds gst-launch-1.0 -q filesrc location=%s ! mpegaudioparse ! mpg123audiodec ! audioconvert ! audioresample ! alsasink device=%s sync=false async=false",
AUDIO_LOCAL_PLAY_TIMEOUT_SEC, quoted_path, AUDIO_LOCAL_ALSA_DEVICE);
ret = system(command);
exit_code = audio_system_exit_code(ret);
if (exit_code != 0) {
my_zlog_warn("本地音频 GStreamer 播放失败 exit=%d，尝试 ffplay: %s",
exit_code, filepath);
snprintf(command, sizeof(command),
"timeout %ds ffplay -nodisp -autoexit -loglevel warning %s",
AUDIO_LOCAL_PLAY_TIMEOUT_SEC, quoted_path);
ret = system(command);
exit_code = audio_system_exit_code(ret);
}
audio_sink_unlock_alsa();
return exit_code;
}
//接收音频播放
void audioplay_mqtt_receive(cJSON *json) {
// 解析"audioLink"字段（修正了原始JSON中的拼写错误）
cJSON *audio_link = cJSON_GetObjectItemCaseSensitive(json, "audioLink");
if (cJSON_IsString(audio_link) && (audio_link->valuestring != NULL)) {
my_zlog_debug("音频链接: %s", audio_link->valuestring);
char *url=audio_link->valuestring;
snprintf(s_urlbuf, sizeof(s_urlbuf), "%s", url);
} else {
my_zlog_warn("错误：无法解析音频链接字段");
}
// 解析"key"字段
cJSON *key_char = cJSON_GetObjectItemCaseSensitive(json, "key");
if (cJSON_IsString(key_char) && (key_char->valuestring != NULL)) {
my_zlog_debug("音频链接KEY: %s", key_char->valuestring);
char *key=key_char->valuestring;
snprintf(s_keybuf, sizeof(s_keybuf), "%s", key);
} else {
my_zlog_warn("错误：无法解析音频链接");
s_audio_status=5;
}
// 解析"status"字段
cJSON *s_audio_status_val = cJSON_GetObjectItemCaseSensitive(json, "status");
if (cJSON_IsNumber(s_audio_status_val)) {
my_zlog_debug("标志: %d", s_audio_status_val->valueint);
s_audio_status=s_audio_status_val->valueint;
} else {
my_zlog_warn("错误：无法解析标志字段");
}
// 解析"volume"字段
cJSON *volume = cJSON_GetObjectItemCaseSensitive(json, "volume");
if (cJSON_IsNumber(volume)) {
my_zlog_debug("声量大小: %.3f", volume->valuedouble);
s_audio_volume = audioplay_volume_clamp(volume->valuedouble);
} else {
my_zlog_warn("错误：无法解析声量字段");
}
}
//发送音频播放是否完毕
void audioplay_send_mqtt() {
cJSON *root = cJSON_CreateObject();
cJSON *body = cJSON_CreateObject();
cJSON *head = cJSON_CreateObject();
// 添加各个字段到 JSON 对象
cJSON_AddStringToObject(body, "type", "audio");
cJSON_AddStringToObject(body, "audioLink", s_urlbuf);
cJSON_AddStringToObject(body, "key", s_keybuf);
cJSON_AddNumberToObject(body, "status", s_audio_status);
cJSON_AddNumberToObject(body, "volume", s_audio_volume);
cJSON_AddNumberToObject(head, "message_type",3001);
cJSON_AddItemToObject(root, "body", body);
cJSON_AddItemToObject(root, "head",head);
// 将 JSON 对象转换为字符串
char* json_string = cJSON_PrintUnformatted(root);
my_zlog_debug("%s",json_string);
mqtt_publish_to_all(mqtt_topic_pure_number(), json_string, 0);
free(json_string);
cJSON_Delete(root);
}
//音频播放
void audioplay_cycle(){
char command[2048];
int ret;
while(1){
if(s_audio_status==0){
char *urlmoddle=s_urlbuf;
s_audio_volume = audioplay_volume_clamp(s_audio_volume);
snprintf(command, sizeof(command),
"sudo ffplay -nodisp -autoexit -loglevel quiet -af \"volume=%.3f\" \"%s\"",
s_audio_volume,urlmoddle);
my_zlog_debug("播放地址: %s", s_urlbuf);
my_zlog_debug("执行播放命令: %s", command);
ret = system(command);
if (ret != 0) {
my_zlog_error("播放失败");
s_audio_status=2;
}
if (WIFEXITED(ret) && WEXITSTATUS(ret) == 0) {
my_zlog_debug("播放已成功完成 : %s ", s_urlbuf);
s_audio_status=1;
} else {
my_zlog_warn("播放失败或中断: %s ", s_urlbuf);
s_audio_status=2;
}
audioplay_send_mqtt();
}
char local_filepath[sizeof(s_local_filepath)];
int local_play_pending = 0;
pthread_mutex_lock(&s_local_play_mutex);
if (s_local_play_pending) {
s_local_play_pending = 0;
snprintf(local_filepath, sizeof(local_filepath), "%s", s_local_filepath);
local_play_pending = 1;
}
pthread_mutex_unlock(&s_local_play_mutex);
if (local_play_pending) {
int exit_code;
my_zlog_info("播放本地音频: %s", local_filepath);
exit_code = play_local_audio_file(local_filepath);
if (exit_code == 0) {
my_zlog_debug("本地音频播放完成: %s", local_filepath);
} else {
my_zlog_warn("本地音频播放失败 exit=%d: %s", exit_code, local_filepath);
}
}
video_tts_play();
delay_ms(100);
}
}
static void trim_line(char *s)
{
if (!s)
return;
s[strcspn(s, "\r\n")] = '\0';
}
static int pulse_pactl_exit_code(int status)
{
if (status == -1)
return -1;
if (WIFEXITED(status))
return WEXITSTATUS(status);
return -1;
}
static int pulse_output_means_already_loaded(const char *output)
{
if (!output || !output[0])
return 0;
return strstr(output, "already loaded") != NULL ||
strstr(output, "Already loaded") != NULL ||
strstr(output, "Module exists") != NULL;
}
static int pulse_module_exists(const char *module_name, const char *device)
{
FILE *fp = popen("pactl list modules short 2>/dev/null", "r");
char line[512];
char needle[96];
if (!fp)
return 0;
snprintf(needle, sizeof(needle), "device=%s", device);
while (fgets(line, sizeof(line), fp) != NULL) {
if (strstr(line, module_name) != NULL && strstr(line, needle) != NULL) {
pclose(fp);
return 1;
}
}
pclose(fp);
return 0;
}
static int pulse_load_alsa_module(const char *module_name, const char *label)
{
char cmd[256];
char output[512];
char line[256];
if (pulse_module_exists(module_name, AUDIO_USB_ALSA_DEVICE)) {
my_zlog_info("%s 已注册，跳过 (%s device=%s)",
label, module_name, AUDIO_USB_ALSA_DEVICE);
return 0;
}
snprintf(cmd, sizeof(cmd), "pactl load-module %s device=%s 2>&1",
module_name, AUDIO_USB_ALSA_DEVICE);
FILE *fp = popen(cmd, "r");
if (!fp) {
my_zlog_warn("%s 注册失败: 无法执行 pactl (%s)", label, strerror(errno));
return -1;
}
output[0] = '\0';
while (fgets(line, sizeof(line), fp) != NULL) {
trim_line(line);
if (!line[0])
continue;
if (output[0]) {
strncat(output, "; ", sizeof(output) - strlen(output) - 1);
}
strncat(output, line, sizeof(output) - strlen(output) - 1);
}
int status = pclose(fp);
int exit_code = pulse_pactl_exit_code(status);
if (exit_code == 0) {
my_zlog_info("%s 注册成功 (%s device=%s)%s%s",
label, module_name, AUDIO_USB_ALSA_DEVICE,
output[0] ? ", pactl: " : "", output[0] ? output : "");
return 0;
}
if (pulse_output_means_already_loaded(output) ||
pulse_module_exists(module_name, AUDIO_USB_ALSA_DEVICE)) {
my_zlog_info("%s 已注册，跳过 (%s device=%s)%s%s",
label, module_name, AUDIO_USB_ALSA_DEVICE,
output[0] ? ": " : "", output[0] ? output : "");
return 0;
}
my_zlog_warn("%s 注册失败 (%s device=%s): exit=%d%s%s",
label, module_name, AUDIO_USB_ALSA_DEVICE, exit_code,
output[0] ? ", pactl: " : ", pactl 无输出",
output[0] ? output : "");
return -1;
}
int audio_wheat_init(){
return pulse_load_alsa_module("module-alsa-sink", "喇叭");
}
int audio_speaker_init() {
return pulse_load_alsa_module("module-alsa-source", "麦");
}
int audio_init(){
delay_s(1);
audio_wheat_init();
delay_s(1);
audio_speaker_init();
return 0;
}
#define CONFIG_FILE "/usr/share/pulseaudio/alsa-mixer/profile-sets/default.conf"
static const char *s_yundea_block =
"# 为 Yundea 1076 USB 二合一设备添加配置\n"
"[Mapping yundea-input]\n"
"device-strings = hw:%f,0\n"
"direction = input\n"
"priority = 100\n"
"channel-map = mono\n"
"\n"
"[Mapping yundea-output]\n"
"device-strings = hw:%f,0\n"
"direction = output\n"
"priority = 100\n"
"channel-map = left,right\n"
"\n"
"[Profile yundea-duplex]\n"
"input-mappings = yundea-input\n"
"output-mappings = yundea-output\n"
"priority = 200\n";
/*
*功能：加入usb声卡配置
*
*/
int audio_config_init() {
delay_s(12);
FILE *fp = fopen(CONFIG_FILE, "r");
if (!fp) {
perror("无法打开配置文件进行读取");
return 1;
}
// 获取文件大小
fseek(fp, 0, SEEK_END);
long size = ftell(fp);
fseek(fp, 0, SEEK_SET);
char *buffer = malloc(size + 1);
if (!buffer) {
my_zlog_error("内存分配失败");
fclose(fp);
return 1;
}
size_t result =fread(buffer, 1, size, fp);
if (result != size) {
my_zlog_warn("Error: Failed to read audio data");
}
buffer[size] = '\0';
fclose(fp);
// 检查是否包含完整配置块
if (strstr(buffer, s_yundea_block) != NULL) {
my_zlog_debug("配置文件中已包含指定配置，跳过添加。");
free(buffer);
return 0;
}
free(buffer);
my_zlog_info("未找到指定配置，正在追加到文件末尾...");
// 打开文件追加配置
fp = fopen(CONFIG_FILE, "a");
if (!fp) {
my_zlog_error("无法打开配置文件进行写入");
return 1;
}
fprintf(fp, "\n%s", s_yundea_block);
fclose(fp);
my_zlog_info("配置已成功追加。");
return 0;
}
modules/CMakeLists.txt
View file @ 3585cc8c
...@@ -8,6 +8,7 @@ pkg_check_modules(WEBRTCPUSH_GST REQUIRED ...@@ -8,6 +8,7 @@ pkg_check_modules(WEBRTCPUSH_GST REQUIRED
) )
pkg_check_modules(WEBRTCPUSH_JSON REQUIRED json-glib-1.0) pkg_check_modules(WEBRTCPUSH_JSON REQUIRED json-glib-1.0)
pkg_check_modules(WEBRTCPUSH_SOUP REQUIRED libsoup-2.4) pkg_check_modules(WEBRTCPUSH_SOUP REQUIRED libsoup-2.4)
pkg_check_modules(WEBRTCPUSH_JPEG REQUIRED libjpeg)
file(GLOB_RECURSE MODULES_SOURCES file(GLOB_RECURSE MODULES_SOURCES
${CMAKE_CURRENT_SOURCE_DIR}/logger/*.c ${CMAKE_CURRENT_SOURCE_DIR}/logger/*.c
...@@ -47,6 +48,7 @@ set(MODULES_INCLUDE_DIRS ...@@ -47,6 +48,7 @@ set(MODULES_INCLUDE_DIRS
${WEBRTCPUSH_GST_INCLUDE_DIRS} ${WEBRTCPUSH_GST_INCLUDE_DIRS}
${WEBRTCPUSH_JSON_INCLUDE_DIRS} ${WEBRTCPUSH_JSON_INCLUDE_DIRS}
${WEBRTCPUSH_SOUP_INCLUDE_DIRS} ${WEBRTCPUSH_SOUP_INCLUDE_DIRS}
${WEBRTCPUSH_JPEG_INCLUDE_DIRS}
PARENT_SCOPE PARENT_SCOPE
) )
...@@ -54,6 +56,7 @@ set(WEBRTCPUSH_LIBRARIES ...@@ -54,6 +56,7 @@ set(WEBRTCPUSH_LIBRARIES
${WEBRTCPUSH_GST_LIBRARIES} ${WEBRTCPUSH_GST_LIBRARIES}
${WEBRTCPUSH_JSON_LIBRARIES} ${WEBRTCPUSH_JSON_LIBRARIES}
${WEBRTCPUSH_SOUP_LIBRARIES} ${WEBRTCPUSH_SOUP_LIBRARIES}
${WEBRTCPUSH_JPEG_LIBRARIES}
PARENT_SCOPE PARENT_SCOPE
) )
...@@ -61,5 +64,6 @@ set(WEBRTCPUSH_CFLAGS ...@@ -61,5 +64,6 @@ set(WEBRTCPUSH_CFLAGS
${WEBRTCPUSH_GST_CFLAGS_OTHER} ${WEBRTCPUSH_GST_CFLAGS_OTHER}
${WEBRTCPUSH_JSON_CFLAGS_OTHER} ${WEBRTCPUSH_JSON_CFLAGS_OTHER}
${WEBRTCPUSH_SOUP_CFLAGS_OTHER} ${WEBRTCPUSH_SOUP_CFLAGS_OTHER}
${WEBRTCPUSH_JPEG_CFLAGS_OTHER}
PARENT_SCOPE PARENT_SCOPE
) )
modules/webrtcpush/audio_sink.c
View file @ 3585cc8c
#include "audio_sink.h" #include "audio_sink.h"
#include "webrtcpush_config.h"
#include "webrtcpush_log.h" #include "webrtcpush_log.h"
#include <gst/gst.h> /* USB声卡排队锁: audio_sink 和本地音频串行访问 plughw:2,0 */
#include <gst/app/gstappsrc.h> static GMutex g_alsa_device_lock;
static AudioSink *g_audio_sink_singleton = NULL;
/* /*
* 手机→设备方向的音频播放管道（按键模式）： * 手机→设备方向的音频不再走 RTP：手机端 audio 为 recvonly（只收不发），
* appsrc(opus payload) → opusdec → audioconvert → audioresample → volume → alsasink * 手机→设备的音频通过 DataChannel 发送 MP4，由 decodebin 播放。
* * 因此本模块不再创建 GStreamer 接收管道，仅保留 USB 声卡全局锁，
* 手机端是"按住说话"模式（最长 15s），不是持续推流，因此： * 供 DataChannel 播放线程与本地提示音串行访问 ALSA 设备使用。
* - on_audio_message 收到包后入队，立即返回（不阻塞 libdatachannel 线程）
* - 独立线程消费队列，维护"按键会话"
* - 会话开始（首包到达）：pipeline 切到 PLAYING
* - 会话结束（500ms 无包 或 15s 超时）：flush appsrc + pipeline 切到 READY
* 切到 READY 释放 ALSA 设备，避免 alsasink 持续占用/空转
*/ */
#define AUDIO_SINK_SESSION_TIMEOUT_MS 500 /* 500ms 无包认为按键结束 */
#define AUDIO_SINK_SESSION_MAX_MS 15000 /* 单次按键最长 15s */
typedef struct {
uint8_t *data;
size_t size;
} SinkPacket;
struct AudioSink { struct AudioSink {
GstElement *pipeline; GMutex lock;
GstElement *appsrc;
GstElement *vol;
GAsyncQueue *queue; /* 待处理 Opus 包队列 */
GMutex lock; /* 保护 pipeline 状态切换 */
GThread *thread; /* 消费线程 */
gboolean quit; /* 退出标志 */
gint64 session_start_us; /* 当前按键会话开始时间（0=无会话） */
gint64 last_packet_us; /* 最后一个包到达时间 */
gboolean pipeline_playing; /* pipeline 当前是否 PLAYING */
}; };
static void flush_queue(AudioSink *src) void audio_sink_lock_alsa(void) { g_mutex_lock(&g_alsa_device_lock); }
{ void audio_sink_unlock_alsa(void) { g_mutex_unlock(&g_alsa_device_lock); }
SinkPacket *pkt;
while ((pkt = g_async_queue_try_pop(src->queue)) != NULL) {
g_free(pkt->data);
g_free(pkt);
}
}
static void set_pipeline_state_locked(AudioSink *src, GstState state) void audio_sink_interrupt(AudioSink *src)
{ {
if (!src->pipeline) (void)src;
return; /* RTP 接收管道已移除，无会话需要中断；DataChannel 播放通过 lock/unlock 串行访问 ALSA。 */
/* 切到 READY 时 flush appsrc，避免旧数据残留导致下次会话首帧异常 */
if (state == GST_STATE_READY && src->appsrc) {
gst_element_send_event(src->pipeline,
gst_event_new_flush_start());
gst_element_send_event(src->pipeline,
gst_event_new_flush_stop(FALSE));
}
gst_element_set_state(src->pipeline, state);
src->pipeline_playing = (state == GST_STATE_PLAYING);
}
static gpointer audio_sink_thread(gpointer data)
{
AudioSink *src = data;
gint64 now;
GstBuffer *buf;
while (1) {
/* 等待包，超时 100ms 用于检查会话超时 */
SinkPacket *pkt = g_async_queue_timeout_pop(src->queue, 100 * 1000);
now = g_get_monotonic_time();
g_mutex_lock(&src->lock);
if (src->quit) {
g_mutex_unlock(&src->lock);
if (pkt) { g_free(pkt->data); g_free(pkt); }
break;
}
/* 会话超时检查 */
if (src->session_start_us > 0) {
gint64 idle_ms = (now - src->last_packet_us) / 1000;
gint64 sess_ms = (now - src->session_start_us) / 1000;
if (idle_ms >= AUDIO_SINK_SESSION_TIMEOUT_MS ||
sess_ms >= AUDIO_SINK_SESSION_MAX_MS) {
if (src->pipeline_playing) {
set_pipeline_state_locked(src, GST_STATE_READY);
my_zlog_info("audio_sink: session ended (idle=%lldms sess=%lldms)",
(long long)idle_ms, (long long)sess_ms);
}
src->session_start_us = 0;
flush_queue(src);
/* 丢弃超时后到达的旧包 */
if (pkt) {
g_free(pkt->data);
g_free(pkt);
pkt = NULL;
}
g_mutex_unlock(&src->lock);
continue;
}
}
if (pkt) {
/* 新会话开始 */
if (src->session_start_us == 0) {
src->session_start_us = now;
my_zlog_info("audio_sink: session start");
if (!src->pipeline_playing)
set_pipeline_state_locked(src, GST_STATE_PLAYING);
}
src->last_packet_us = now;
/* push 到 appsrc（pipeline PLAYING 状态） */
if (src->appsrc && src->pipeline_playing) {
buf = gst_buffer_new_wrapped(g_memdup2(pkt->data, pkt->size), pkt->size);
GST_BUFFER_DTS(buf) = GST_CLOCK_TIME_NONE;
GST_BUFFER_PTS(buf) = GST_CLOCK_TIME_NONE;
if (gst_app_src_push_buffer(GST_APP_SRC(src->appsrc), buf) != GST_FLOW_OK) {
my_zlog_warn("audio_sink: push_buffer failed");
}
}
g_free(pkt->data);
g_free(pkt);
}
g_mutex_unlock(&src->lock);
}
return NULL;
} }
AudioSink *audio_sink_start(const char *alsa_device, char **error_message) AudioSink *audio_sink_start(const char *alsa_device, char **error_message)
{ {
AudioSink *src; AudioSink *src;
GstElement *pipe, *asrc, *dec, *conv, *resample, *vol, *sink; (void)alsa_device;
GstCaps *caps;
GstStateChangeReturn ret;
if (!alsa_device || !alsa_device[0]) {
if (error_message) if (error_message)
*error_message = g_strdup("no ALSA device"); *error_message = NULL;
return NULL;
}
src = g_new0(AudioSink, 1); src = g_new0(AudioSink, 1);
pipe = gst_pipeline_new("audio-sink-pipe");
asrc = gst_element_factory_make("appsrc", "asrc");
dec = gst_element_factory_make("opusdec", "dec");
conv = gst_element_factory_make("audioconvert", "conv");
resample = gst_element_factory_make("audioresample", "resample");
vol = gst_element_factory_make("volume", "vol");
sink = gst_element_factory_make("alsasink", "sink");
if (!pipe || !asrc || !dec || !conv || !resample || !vol || !sink) {
if (error_message)
*error_message = g_strdup("failed to create audio sink GStreamer elements");
if (pipe)
gst_object_unref(pipe);
g_free(src);
return NULL;
}
caps = gst_caps_new_empty_simple("audio/x-opus");
g_object_set(asrc,
"caps", caps,
"format", GST_FORMAT_BYTES,
"is-live", TRUE,
"emit-signals", FALSE,
"min-latency", (gint64)0,
"max-bytes", (guint64)(1 * 1024 * 1024),
NULL);
gst_caps_unref(caps);
g_object_set(sink,
"device", alsa_device,
"buffer-time", (gint64)20000,
"latency-time", (gint64)10000,
"sync", FALSE,
NULL);
g_object_set(vol, "volume", 0.5, NULL);
gst_bin_add_many(GST_BIN(pipe), asrc, dec, conv, resample, vol, sink, NULL);
if (!gst_element_link_many(asrc, dec, conv, resample, vol, sink, NULL)) {
if (error_message)
*error_message = g_strdup("failed to link audio sink chain");
gst_object_unref(pipe);
g_free(src);
return NULL;
}
src->pipeline = pipe;
src->appsrc = asrc;
src->vol = vol;
src->queue = g_async_queue_new();
g_mutex_init(&src->lock); g_mutex_init(&src->lock);
src->session_start_us = 0; g_audio_sink_singleton = src;
src->pipeline_playing = FALSE; my_zlog_info("audio_sink: started (RTP receive removed, ALSA lock only)");
/* 初始状态 READY（不占 ALSA 设备，等首包到来再 PLAYING） */
ret = gst_element_set_state(pipe, GST_STATE_READY);
if (ret == GST_STATE_CHANGE_FAILURE) {
if (error_message)
*error_message = g_strdup("audio sink pipeline failed to reach READY");
gst_element_set_state(pipe, GST_STATE_NULL);
gst_object_unref(pipe);
g_async_queue_unref(src->queue);
g_mutex_clear(&src->lock);
g_free(src);
return NULL;
}
src->thread = g_thread_new("audio-sink", audio_sink_thread, src);
if (!src->thread) {
if (error_message)
*error_message = g_strdup("failed to create audio sink thread");
audio_sink_stop(src);
return NULL;
}
my_zlog_info("audio_sink: started device=%s opus=%uch %uHz (push-to-talk)",
alsa_device, WEBRTCPUSH_OPUS_CHANNELS, WEBRTCPUSH_OPUS_CLOCKRATE);
return src; return src;
} }
...@@ -233,45 +43,15 @@ void audio_sink_stop(AudioSink *src) ...@@ -233,45 +43,15 @@ void audio_sink_stop(AudioSink *src)
{ {
if (!src) if (!src)
return; return;
if (src->thread) { if (g_audio_sink_singleton == src)
g_mutex_lock(&src->lock); g_audio_sink_singleton = NULL;
src->quit = TRUE;
g_mutex_unlock(&src->lock);
g_thread_join(src->thread);
src->thread = NULL;
}
if (src->pipeline) {
gst_element_set_state(src->pipeline, GST_STATE_NULL);
gst_object_unref(src->pipeline);
}
if (src->queue) {
flush_queue(src);
g_async_queue_unref(src->queue);
}
g_mutex_clear(&src->lock); g_mutex_clear(&src->lock);
g_free(src); g_free(src);
} }
gboolean audio_sink_push_opus(AudioSink *src, const uint8_t *data, size_t size)
{
SinkPacket *pkt;
if (!src || !src->queue || !data || size == 0)
return FALSE;
/* 入队，由消费线程处理（不阻塞 libdatachannel 回调线程） */
pkt = g_new0(SinkPacket, 1);
pkt->data = (uint8_t *)g_memdup2(data, size);
pkt->size = size;
g_async_queue_push(src->queue, pkt);
return TRUE;
}
void audio_sink_set_volume(AudioSink *src, double volume) void audio_sink_set_volume(AudioSink *src, double volume)
{ {
if (!src || !src->vol) (void)src;
return; (void)volume;
if (volume < 0.0) /* 无 GStreamer 管道，音量设置为空操作（保留接口供 volume_control 调用）。 */
volume = 0.0;
if (volume > 1.0)
volume = 1.0;
g_object_set(src->vol, "volume", volume, NULL);
} }
modules/webrtcpush/audio_sink.h
View file @ 3585cc8c
...@@ -17,4 +17,11 @@ gboolean audio_sink_push_opus(AudioSink *src, const uint8_t *data, size_t size); ...@@ -17,4 +17,11 @@ gboolean audio_sink_push_opus(AudioSink *src, const uint8_t *data, size_t size);
/* 设置播放音量 0.0~1.0 */ /* 设置播放音量 0.0~1.0 */
void audio_sink_set_volume(AudioSink *src, double volume); void audio_sink_set_volume(AudioSink *src, double volume);
/* USB声卡排队锁: 本地音频和audio_sink串行访问同一个USB声卡 */
void audio_sink_lock_alsa(void);
void audio_sink_unlock_alsa(void);
/* 中断当前按键会话, 释放ALSA设备让DataChannel音频能立即播放 */
void audio_sink_interrupt(AudioSink *src);
#endif #endif
\ No newline at end of file
modules/webrtcpush/mpp_h264_source.c
View file @ 3585cc8c
...@@ -4,9 +4,13 @@ ...@@ -4,9 +4,13 @@
#include "webrtcpush_log.h" #include "webrtcpush_log.h"
#include <gst/app/gstappsink.h> #include <gst/app/gstappsink.h>
#include <gst/app/gstappsrc.h>
#include <gst/video/video-event.h> #include <gst/video/video-event.h>
#include <fcntl.h> #include <fcntl.h>
#include <stdio.h>
#include <jpeglib.h>
#include <linux/videodev2.h> #include <linux/videodev2.h>
#include <setjmp.h>
#include <string.h> #include <string.h>
#include <sys/ioctl.h> #include <sys/ioctl.h>
#include <unistd.h> #include <unistd.h>
...@@ -19,6 +23,10 @@ ...@@ -19,6 +23,10 @@
#define MPP_VIDEO_MAX_BPS WEBRTCPUSH_MAX_BITRATE #define MPP_VIDEO_MAX_BPS WEBRTCPUSH_MAX_BITRATE
#define MPP_H264_PROFILE 66 /* Baseline: 与 SDP profile-level-id=42e01f 一致，低延迟优先 */ #define MPP_H264_PROFILE 66 /* Baseline: 与 SDP profile-level-id=42e01f 一致，低延迟优先 */
#define MPP_IDR_EVENT_RETRY_US 500000 #define MPP_IDR_EVENT_RETRY_US 500000
#define MJPEG_DECODE_WARN_US 15000
#define MJPEG_DECODE_WARN_INTERVAL_US 5000000
#define PERF_LOG_FRAMES (WEBRTCPUSH_H264_FPS * 5)
#define TIMING_RING_SIZE 32
#define WRTC_V4L2_CAP_VIDEO_CAPTURE (1u << 0) #define WRTC_V4L2_CAP_VIDEO_CAPTURE (1u << 0)
#define WRTC_V4L2_CAP_VIDEO_CAPTURE_MPLANE (1u << 12) #define WRTC_V4L2_CAP_VIDEO_CAPTURE_MPLANE (1u << 12)
...@@ -30,17 +38,40 @@ typedef enum { ...@@ -30,17 +38,40 @@ typedef enum {
CAPTURE_MJPEG_720P30, CAPTURE_MJPEG_720P30,
} CaptureMode; } CaptureMode;
typedef struct {
guint64 pts_ns;
gint64 capture_us;
gint64 push_us;
gint64 enc_in_us;
guint decode_us;
guint convert_us;
gboolean valid;
} FrameTiming;
struct MppH264Source { struct MppH264Source {
GstElement *pipeline; GstElement *pipeline;
GstElement *capture_pipeline;
GstElement *enc; GstElement *enc;
GstElement *parse; GstElement *parse;
GstElement *eq; GstElement *eq;
GstElement *appsink; GstElement *appsink;
GstElement *mjpeg_sink;
GstElement *raw_appsrc;
GstElement *vsrc; /* v4l2src 指针，用于 recover 时更新设备 */ GstElement *vsrc; /* v4l2src 指针，用于 recover 时更新设备 */
gboolean use_mpp; gboolean use_mpp;
gboolean manual_mjpeg;
volatile gboolean stop_decode;
GThread *decode_thread;
guint8 *frame_buf; guint8 *frame_buf;
size_t frame_size; size_t frame_size;
size_t frame_capacity; size_t frame_capacity;
guint8 *i420_buf;
size_t i420_capacity;
guint8 *nv12_buf;
size_t nv12_capacity;
guint8 *jpeg_tmp_u;
guint8 *jpeg_tmp_v;
size_t jpeg_tmp_capacity;
guint keyframe_req; guint keyframe_req;
volatile gboolean pending_idr; volatile gboolean pending_idr;
gboolean pending_idr_event_sent; gboolean pending_idr_event_sent;
...@@ -48,7 +79,24 @@ struct MppH264Source { ...@@ -48,7 +79,24 @@ struct MppH264Source {
volatile gboolean pending_recover; volatile gboolean pending_recover;
gulong enc_sink_probe_id; gulong enc_sink_probe_id;
guint bus_watch_id; guint bus_watch_id;
guint capture_bus_watch_id;
gchar *sprop_parameter_sets; gchar *sprop_parameter_sets;
GMutex timing_lock;
FrameTiming timings[TIMING_RING_SIZE];
guint timing_next;
guint perf_frames;
guint64 perf_decode_sum_us;
guint64 perf_convert_sum_us;
guint64 perf_encode_sum_us;
guint64 perf_total_sum_us;
guint perf_decode_max_us;
guint perf_convert_max_us;
guint perf_encode_max_us;
guint perf_total_max_us;
guint jpeg_slow_count;
guint jpeg_slow_last_us;
guint jpeg_slow_max_us;
gint64 jpeg_slow_last_warn_us;
}; };
static gboolean pull_sample(MppH264Source *src, gboolean require_idr, static gboolean pull_sample(MppH264Source *src, gboolean require_idr,
...@@ -56,6 +104,9 @@ static gboolean pull_sample(MppH264Source *src, gboolean require_idr, ...@@ -56,6 +104,9 @@ static gboolean pull_sample(MppH264Source *src, gboolean require_idr,
size_t *size, gboolean *is_idr, size_t *size, gboolean *is_idr,
guint64 *pts_ns); guint64 *pts_ns);
static gboolean send_force_key_unit(MppH264Source *src); static gboolean send_force_key_unit(MppH264Source *src);
static gpointer mjpeg_decode_thread(gpointer user_data);
static void timing_mark_encoder_in(MppH264Source *src, guint64 pts_ns,
gint64 enc_in_us);
static gboolean bus_watch_cb(GstBus *bus, GstMessage *msg, gpointer user_data) static gboolean bus_watch_cb(GstBus *bus, GstMessage *msg, gpointer user_data)
{ {
...@@ -243,7 +294,7 @@ static CaptureMode probe_720p_capture_mode(const char *device) ...@@ -243,7 +294,7 @@ static CaptureMode probe_720p_capture_mode(const char *device)
return CAPTURE_MJPEG_720P24; return CAPTURE_MJPEG_720P24;
} }
if (mjpg30_ok) { if (mjpg30_ok) {
my_zlog_info("mpp_h264_source: %s 720p MJPEG@30fps -> mppjpegdec -> videorate@%dfps", my_zlog_info("mpp_h264_source: %s 720p MJPEG@30fps -> libjpeg-turbo -> videorate@%dfps",
device, WEBRTCPUSH_H264_FPS); device, WEBRTCPUSH_H264_FPS);
return CAPTURE_MJPEG_720P30; return CAPTURE_MJPEG_720P30;
} }
...@@ -276,6 +327,194 @@ static void configure_post_enc_queue(GstElement *q) ...@@ -276,6 +327,194 @@ static void configure_post_enc_queue(GstElement *q)
NULL); NULL);
} }
static void object_set_uint_if_exists(GObject *obj, const gchar *name, guint value)
{
if (obj && g_object_class_find_property(G_OBJECT_GET_CLASS(obj), name))
g_object_set(obj, name, value, NULL);
}
static void object_set_int_if_exists(GObject *obj, const gchar *name, gint value)
{
if (obj && g_object_class_find_property(G_OBJECT_GET_CLASS(obj), name))
g_object_set(obj, name, value, NULL);
}
static gboolean ensure_byte_buffer(guint8 **buf, size_t *capacity, size_t need)
{
if (!buf || !capacity || need == 0)
return FALSE;
if (*capacity >= need && *buf)
return TRUE;
*buf = g_realloc(*buf, need);
*capacity = need;
return *buf != NULL;
}
static void configure_raw_appsrc(GstElement *appsrc, gint fps)
{
if (!appsrc)
return;
g_object_set(appsrc,
"is-live", TRUE,
"format", GST_FORMAT_TIME,
"block", FALSE,
"do-timestamp", FALSE,
NULL);
object_set_uint_if_exists(G_OBJECT(appsrc), "max-buffers", 2);
/* appsrc leaky-type: 2 = downstream/drop oldest on recent GStreamer. */
object_set_int_if_exists(G_OBJECT(appsrc), "leaky-type", 2);
{
gchar *cap_str = g_strdup_printf(
"video/x-raw,format=NV12,width=%d,height=%d,framerate=%d/1",
MPP_VIDEO_WIDTH, MPP_VIDEO_HEIGHT, fps);
GstCaps *caps = gst_caps_from_string(cap_str);
g_object_set(appsrc, "caps", caps, NULL);
gst_caps_unref(caps);
g_free(cap_str);
}
}
static void timing_store(MppH264Source *src, guint64 pts_ns, gint64 capture_us,
gint64 push_us, guint decode_us, guint convert_us)
{
FrameTiming *t;
if (!src || pts_ns == G_MAXUINT64)
return;
g_mutex_lock(&src->timing_lock);
t = &src->timings[src->timing_next++ % TIMING_RING_SIZE];
t->pts_ns = pts_ns;
t->capture_us = capture_us;
t->push_us = push_us;
t->enc_in_us = 0;
t->decode_us = decode_us;
t->convert_us = convert_us;
t->valid = TRUE;
g_mutex_unlock(&src->timing_lock);
}
static void timing_mark_encoder_in(MppH264Source *src, guint64 pts_ns,
gint64 enc_in_us)
{
guint i;
if (!src || pts_ns == G_MAXUINT64)
return;
g_mutex_lock(&src->timing_lock);
for (i = 0; i < TIMING_RING_SIZE; i++) {
FrameTiming *t = &src->timings[i];
if (t->valid && t->pts_ns == pts_ns) {
t->enc_in_us = enc_in_us;
break;
}
}
g_mutex_unlock(&src->timing_lock);
}
static gboolean timing_take(MppH264Source *src, guint64 pts_ns, FrameTiming *out)
{
guint i;
gboolean found = FALSE;
if (!src || !out || pts_ns == G_MAXUINT64)
return FALSE;
g_mutex_lock(&src->timing_lock);
for (i = 0; i < TIMING_RING_SIZE; i++) {
FrameTiming *t = &src->timings[i];
if (t->valid && t->pts_ns == pts_ns) {
*out = *t;
t->valid = FALSE;
found = TRUE;
break;
}
}
g_mutex_unlock(&src->timing_lock);
return found;
}
static void perf_record(MppH264Source *src, const FrameTiming *t, gint64 out_us)
{
guint encode_us;
guint total_us;
if (!src || !t)
return;
encode_us = (t->enc_in_us > 0 && out_us > t->enc_in_us)
? (guint)(out_us - t->enc_in_us)
: (t->push_us > 0 && out_us > t->push_us)
? (guint)(out_us - t->push_us)
: 0;
total_us = (t->capture_us > 0 && out_us > t->capture_us)
? (guint)(out_us - t->capture_us)
: 0;
src->perf_frames++;
src->perf_decode_sum_us += t->decode_us;
src->perf_convert_sum_us += t->convert_us;
src->perf_encode_sum_us += encode_us;
src->perf_total_sum_us += total_us;
if (t->decode_us > src->perf_decode_max_us)
src->perf_decode_max_us = t->decode_us;
if (t->convert_us > src->perf_convert_max_us)
src->perf_convert_max_us = t->convert_us;
if (encode_us > src->perf_encode_max_us)
src->perf_encode_max_us = encode_us;
if (total_us > src->perf_total_max_us)
src->perf_total_max_us = total_us;
if (src->perf_frames >= PERF_LOG_FRAMES) {
guint n = src->perf_frames;
my_zlog_info("mpp_h264_source: perf avg/max jpeg_decode=%u/%u ms i420_nv12=%u/%u ms mpp_encode=%u/%u ms total=%u/%u ms",
(guint)(src->perf_decode_sum_us / n / 1000U),
src->perf_decode_max_us / 1000U,
(guint)(src->perf_convert_sum_us / n / 1000U),
src->perf_convert_max_us / 1000U,
(guint)(src->perf_encode_sum_us / n / 1000U),
src->perf_encode_max_us / 1000U,
(guint)(src->perf_total_sum_us / n / 1000U),
src->perf_total_max_us / 1000U);
src->perf_frames = 0;
src->perf_decode_sum_us = 0;
src->perf_convert_sum_us = 0;
src->perf_encode_sum_us = 0;
src->perf_total_sum_us = 0;
src->perf_decode_max_us = 0;
src->perf_convert_max_us = 0;
src->perf_encode_max_us = 0;
src->perf_total_max_us = 0;
}
}
static void note_slow_jpeg_decode(MppH264Source *src, guint decode_us)
{
gint64 now_us;
if (!src || decode_us <= MJPEG_DECODE_WARN_US)
return;
now_us = g_get_monotonic_time();
src->jpeg_slow_count++;
src->jpeg_slow_last_us = decode_us;
if (decode_us > src->jpeg_slow_max_us)
src->jpeg_slow_max_us = decode_us;
if (src->jpeg_slow_last_warn_us > 0 &&
now_us - src->jpeg_slow_last_warn_us < MJPEG_DECODE_WARN_INTERVAL_US)
return;
my_zlog_warn("mpp_h264_source: JPEG decode slow count=%u last=%u us max=%u us (>15ms, rate-limited 5s)",
src->jpeg_slow_count, src->jpeg_slow_last_us,
src->jpeg_slow_max_us);
src->jpeg_slow_count = 0;
src->jpeg_slow_last_us = 0;
src->jpeg_slow_max_us = 0;
src->jpeg_slow_last_warn_us = now_us;
}
static void configure_v4l2_src(GstElement *src, gboolean compressed_mjpeg) static void configure_v4l2_src(GstElement *src, gboolean compressed_mjpeg)
{ {
if (!src) if (!src)
...@@ -295,6 +534,316 @@ static void configure_v4l2_src(GstElement *src, gboolean compressed_mjpeg) ...@@ -295,6 +534,316 @@ static void configure_v4l2_src(GstElement *src, gboolean compressed_mjpeg)
} }
} }
typedef struct {
struct jpeg_error_mgr pub;
jmp_buf jump;
char message[JMSG_LENGTH_MAX];
} MjpegErrorMgr;
static void mjpeg_error_exit(j_common_ptr cinfo)
{
MjpegErrorMgr *err = (MjpegErrorMgr *)cinfo->err;
(*cinfo->err->format_message)(cinfo, err->message);
longjmp(err->jump, 1);
}
static gboolean decode_mjpeg_to_i420(MppH264Source *src,
const guint8 *jpeg_data,
size_t jpeg_size,
guint *decode_us_out)
{
struct jpeg_decompress_struct cinfo;
MjpegErrorMgr jerr;
gint64 start_us;
gboolean ok = FALSE;
guint width;
guint height;
guint uv_width;
guint uv_height;
guint i420_size;
gint h0, v0, h1, v1, h2, v2;
gboolean is_420;
gboolean is_422;
if (decode_us_out)
*decode_us_out = 0;
if (!src || !jpeg_data || jpeg_size == 0)
return FALSE;
memset(&cinfo, 0, sizeof(cinfo));
memset(&jerr, 0, sizeof(jerr));
cinfo.err = jpeg_std_error(&jerr.pub);
jerr.pub.error_exit = mjpeg_error_exit;
if (setjmp(jerr.jump)) {
my_zlog_warn("mpp_h264_source: libjpeg-turbo decode failed: %s",
jerr.message[0] ? jerr.message : "unknown");
jpeg_destroy_decompress(&cinfo);
return FALSE;
}
start_us = g_get_monotonic_time();
jpeg_create_decompress(&cinfo);
jpeg_mem_src(&cinfo, (unsigned char *)jpeg_data, (unsigned long)jpeg_size);
jpeg_read_header(&cinfo, TRUE);
width = cinfo.image_width;
height = cinfo.image_height;
if (width != MPP_VIDEO_WIDTH || height != MPP_VIDEO_HEIGHT ||
cinfo.num_components != 3) {
my_zlog_warn("mpp_h264_source: unsupported MJPEG %ux%u components=%d",
width, height, cinfo.num_components);
jpeg_destroy_decompress(&cinfo);
return FALSE;
}
h0 = cinfo.comp_info[0].h_samp_factor;
v0 = cinfo.comp_info[0].v_samp_factor;
h1 = cinfo.comp_info[1].h_samp_factor;
v1 = cinfo.comp_info[1].v_samp_factor;
h2 = cinfo.comp_info[2].h_samp_factor;
v2 = cinfo.comp_info[2].v_samp_factor;
is_420 = h0 == 2 && v0 == 2 && h1 == 1 && v1 == 1 && h2 == 1 && v2 == 1;
is_422 = h0 == 2 && v0 == 1 && h1 == 1 && v1 == 1 && h2 == 1 && v2 == 1;
if (!is_420 && !is_422) {
my_zlog_warn("mpp_h264_source: unsupported MJPEG subsampling Y=%dx%d Cb=%dx%d Cr=%dx%d",
h0, v0, h1, v1, h2, v2);
jpeg_destroy_decompress(&cinfo);
return FALSE;
}
uv_width = width / 2;
uv_height = height / 2;
i420_size = width * height + uv_width * uv_height * 2;
if (!ensure_byte_buffer(&src->i420_buf, &src->i420_capacity, i420_size)) {
jpeg_destroy_decompress(&cinfo);
return FALSE;
}
if (is_422 &&
(!ensure_byte_buffer(&src->jpeg_tmp_u, &src->jpeg_tmp_capacity,
uv_width * DCTSIZE) ||
!ensure_byte_buffer(&src->jpeg_tmp_v, &src->jpeg_tmp_capacity,
uv_width * DCTSIZE))) {
jpeg_destroy_decompress(&cinfo);
return FALSE;
}
cinfo.raw_data_out = TRUE;
cinfo.out_color_space = JCS_YCbCr;
cinfo.do_fancy_upsampling = FALSE;
cinfo.dct_method = JDCT_IFAST;
jpeg_start_decompress(&cinfo);
{
guint8 *y_plane = src->i420_buf;
guint8 *u_plane = y_plane + width * height;
guint8 *v_plane = u_plane + uv_width * uv_height;
const guint raw_rows = cinfo.max_v_samp_factor * DCTSIZE;
JSAMPROW y_rows[DCTSIZE * 4];
JSAMPROW u_rows[DCTSIZE * 4];
JSAMPROW v_rows[DCTSIZE * 4];
JSAMPARRAY planes[3] = {y_rows, u_rows, v_rows};
while (cinfo.output_scanline < cinfo.output_height) {
guint base_y = cinfo.output_scanline;
guint i;
JDIMENSION got;
for (i = 0; i < raw_rows; i++) {
guint y = base_y + i;
if (y >= height)
y = height - 1;
y_rows[i] = y_plane + y * width;
}
if (is_420) {
guint chroma_base = base_y / 2;
for (i = 0; i < DCTSIZE; i++) {
guint cy = chroma_base + i;
if (cy >= uv_height)
cy = uv_height - 1;
u_rows[i] = u_plane + cy * uv_width;
v_rows[i] = v_plane + cy * uv_width;
}
} else {
for (i = 0; i < DCTSIZE; i++) {
u_rows[i] = src->jpeg_tmp_u + i * uv_width;
v_rows[i] = src->jpeg_tmp_v + i * uv_width;
}
}
got = jpeg_read_raw_data(&cinfo, planes, raw_rows);
if (got == 0)
break;
if (is_422) {
guint chroma_base = base_y / 2;
guint pairs = got / 2;
for (i = 0; i < pairs && chroma_base + i < uv_height; i++) {
guint x;
guint8 *dst_u = u_plane + (chroma_base + i) * uv_width;
guint8 *dst_v = v_plane + (chroma_base + i) * uv_width;
guint8 *u0 = src->jpeg_tmp_u + (i * 2) * uv_width;
guint8 *u1 = src->jpeg_tmp_u + (i * 2 + 1) * uv_width;
guint8 *v0p = src->jpeg_tmp_v + (i * 2) * uv_width;
guint8 *v1p = src->jpeg_tmp_v + (i * 2 + 1) * uv_width;
for (x = 0; x < uv_width; x++) {
dst_u[x] = (guint8)(((guint)u0[x] + (guint)u1[x] + 1U) >> 1);
dst_v[x] = (guint8)(((guint)v0p[x] + (guint)v1p[x] + 1U) >> 1);
}
}
}
}
}
jpeg_finish_decompress(&cinfo);
jpeg_destroy_decompress(&cinfo);
ok = TRUE;
if (decode_us_out)
*decode_us_out = (guint)(g_get_monotonic_time() - start_us);
return ok;
}
static gboolean i420_to_nv12(MppH264Source *src, guint *convert_us_out)
{
const guint width = MPP_VIDEO_WIDTH;
const guint height = MPP_VIDEO_HEIGHT;
const guint y_size = width * height;
const guint uv_width = width / 2;
const guint uv_height = height / 2;
const guint frame_size = y_size + y_size / 2;
const guint8 *u_plane;
const guint8 *v_plane;
guint8 *uv_dst;
gint64 start_us;
guint row;
if (convert_us_out)
*convert_us_out = 0;
if (!src || !src->i420_buf)
return FALSE;
if (!ensure_byte_buffer(&src->nv12_buf, &src->nv12_capacity, frame_size))
return FALSE;
start_us = g_get_monotonic_time();
memcpy(src->nv12_buf, src->i420_buf, y_size);
u_plane = src->i420_buf + y_size;
v_plane = u_plane + uv_width * uv_height;
uv_dst = src->nv12_buf + y_size;
for (row = 0; row < uv_height; row++) {
guint col;
const guint8 *u = u_plane + row * uv_width;
const guint8 *v = v_plane + row * uv_width;
guint8 *dst = uv_dst + row * width;
for (col = 0; col < uv_width; col++) {
dst[col * 2] = u[col];
dst[col * 2 + 1] = v[col];
}
}
if (convert_us_out)
*convert_us_out = (guint)(g_get_monotonic_time() - start_us);
return TRUE;
}
static gpointer mjpeg_decode_thread(gpointer user_data)
{
MppH264Source *src = user_data;
guint64 fallback_pts = 0;
const GstClockTime frame_duration =
gst_util_uint64_scale_int(1, GST_SECOND, WEBRTCPUSH_H264_FPS);
while (src && !src->stop_decode) {
GstSample *sample;
GstBuffer *inbuf;
GstMapInfo inmap;
GstBuffer *outbuf;
GstMapInfo outmap;
GstFlowReturn flow;
GstClockTime pts;
GstClockTime duration;
gint64 capture_us;
gint64 push_us;
guint decode_us = 0;
guint convert_us = 0;
const guint nv12_size = MPP_VIDEO_WIDTH * MPP_VIDEO_HEIGHT * 3 / 2;
if (!src->mjpeg_sink || !src->raw_appsrc) {
g_usleep(10000);
continue;
}
sample = gst_app_sink_try_pull_sample(
GST_APP_SINK(src->mjpeg_sink), 200 * GST_MSECOND);
if (!sample)
continue;
inbuf = gst_sample_get_buffer(sample);
if (!inbuf) {
gst_sample_unref(sample);
continue;
}
pts = GST_BUFFER_PTS(inbuf);
duration = GST_BUFFER_DURATION(inbuf);
if (!GST_CLOCK_TIME_IS_VALID(pts)) {
pts = fallback_pts;
fallback_pts += frame_duration;
}
if (!GST_CLOCK_TIME_IS_VALID(duration))
duration = frame_duration;
else
duration = frame_duration;
capture_us = g_get_monotonic_time();
if (!gst_buffer_map(inbuf, &inmap, GST_MAP_READ)) {
gst_sample_unref(sample);
continue;
}
if (!decode_mjpeg_to_i420(src, inmap.data, inmap.size, &decode_us)) {
gst_buffer_unmap(inbuf, &inmap);
gst_sample_unref(sample);
continue;
}
gst_buffer_unmap(inbuf, &inmap);
gst_sample_unref(sample);
note_slow_jpeg_decode(src, decode_us);
if (!i420_to_nv12(src, &convert_us))
continue;
outbuf = gst_buffer_new_allocate(NULL, nv12_size, NULL);
if (!outbuf)
continue;
if (!gst_buffer_map(outbuf, &outmap, GST_MAP_WRITE)) {
gst_buffer_unref(outbuf);
continue;
}
memcpy(outmap.data, src->nv12_buf, nv12_size);
gst_buffer_unmap(outbuf, &outmap);
GST_BUFFER_PTS(outbuf) = pts;
GST_BUFFER_DTS(outbuf) = GST_CLOCK_TIME_NONE;
GST_BUFFER_DURATION(outbuf) = duration;
push_us = g_get_monotonic_time();
timing_store(src, (guint64)pts, capture_us, push_us,
decode_us, convert_us);
flow = gst_app_src_push_buffer(GST_APP_SRC(src->raw_appsrc), outbuf);
if (flow != GST_FLOW_OK && !src->stop_decode)
my_zlog_warn("mpp_h264_source: appsrc push failed flow=%d", flow);
}
if (src && src->raw_appsrc)
gst_app_src_end_of_stream(GST_APP_SRC(src->raw_appsrc));
return NULL;
}
static GstElement *make_h264_encoder(gboolean prefer_mpp, gboolean use_test, static GstElement *make_h264_encoder(gboolean prefer_mpp, gboolean use_test,
gboolean *use_mpp_out) gboolean *use_mpp_out)
{ {
...@@ -383,11 +932,20 @@ static GstPadProbeReturn enc_sink_idr_probe(GstPad *pad, GstPadProbeInfo *info, ...@@ -383,11 +932,20 @@ static GstPadProbeReturn enc_sink_idr_probe(GstPad *pad, GstPadProbeInfo *info,
gpointer user_data) gpointer user_data)
{ {
MppH264Source *src = user_data; MppH264Source *src = user_data;
GstBuffer *buffer;
(void)pad; (void)pad;
if (!(GST_PAD_PROBE_INFO_TYPE(info) & GST_PAD_PROBE_TYPE_BUFFER)) if (!(GST_PAD_PROBE_INFO_TYPE(info) & GST_PAD_PROBE_TYPE_BUFFER))
return GST_PAD_PROBE_OK; return GST_PAD_PROBE_OK;
if (!src || !src->pending_idr) if (!src)
return GST_PAD_PROBE_OK;
buffer = GST_PAD_PROBE_INFO_BUFFER(info);
if (buffer)
timing_mark_encoder_in(src, (guint64)GST_BUFFER_PTS(buffer),
g_get_monotonic_time());
if (!src->pending_idr)
return GST_PAD_PROBE_OK; return GST_PAD_PROBE_OK;
{ {
...@@ -564,18 +1122,255 @@ static gboolean link_pipeline(GstElement *pipe, GstElement **elements, guint cou ...@@ -564,18 +1122,255 @@ static gboolean link_pipeline(GstElement *pipe, GstElement **elements, guint cou
return TRUE; return TRUE;
} }
static gboolean start_manual_mjpeg_pipeline(MppH264Source *src,
const char *vdev,
CaptureMode capture_mode,
char **error_message)
{
GstElement *cap_pipe = NULL;
GstElement *enc_pipe = NULL;
GstElement *vsrc = NULL;
GstElement *caps_in = NULL;
GstElement *jpegparse = NULL;
GstElement *mjpeg_sink = NULL;
GstElement *rawsrc = NULL;
GstElement *vrate = NULL;
GstElement *caps_nv12 = NULL;
GstElement *q = NULL;
GstElement *eq = NULL;
GstElement *enc = NULL;
GstElement *parse = NULL;
GstElement *caps_h264 = NULL;
GstElement *sink = NULL;
gboolean use_mpp = FALSE;
gint mjpeg_fps = (capture_mode == CAPTURE_MJPEG_720P24)
? WEBRTCPUSH_H264_FPS
: 30;
GstStateChangeReturn ret;
cap_pipe = gst_pipeline_new("mjpeg-capture-pipe");
enc_pipe = gst_pipeline_new("mpp-h264-pipe");
vsrc = gst_element_factory_make("v4l2src", "vsrc");
caps_in = gst_element_factory_make("capsfilter", "caps_in");
jpegparse = gst_element_factory_make("jpegparse", "jpegparse");
mjpeg_sink = gst_element_factory_make("appsink", "mjpeg_sink");
rawsrc = gst_element_factory_make("appsrc", "rawsrc");
vrate = gst_element_factory_make("videorate", "vrate");
caps_nv12 = gst_element_factory_make("capsfilter", "caps_nv12");
q = gst_element_factory_make("queue", "vq");
eq = gst_element_factory_make("queue", "eq");
enc = make_h264_encoder(TRUE, FALSE, &use_mpp);
parse = gst_element_factory_make("h264parse", "parse");
caps_h264 = gst_element_factory_make("capsfilter", "caps_h264");
sink = gst_element_factory_make("appsink", "sink");
if (!cap_pipe || !enc_pipe || !vsrc || !caps_in || !jpegparse ||
!mjpeg_sink || !rawsrc || !vrate || !caps_nv12 || !q || !eq ||
!enc || !parse || !caps_h264 || !sink) {
if (error_message)
*error_message = g_strdup("manual MJPEG element create failed");
goto fail;
}
g_object_set(vsrc, "device", vdev, NULL);
configure_v4l2_src(vsrc, TRUE);
{
gchar *cap_str = g_strdup_printf(
"image/jpeg,width=%d,height=%d,framerate=%d/1",
MPP_VIDEO_WIDTH, MPP_VIDEO_HEIGHT, mjpeg_fps);
GstCaps *caps = gst_caps_from_string(cap_str);
g_object_set(caps_in, "caps", caps, NULL);
gst_caps_unref(caps);
g_free(cap_str);
}
g_object_set(mjpeg_sink,
"emit-signals", FALSE,
"sync", FALSE,
"max-buffers", 2,
"drop", TRUE,
NULL);
configure_raw_appsrc(rawsrc, mjpeg_fps);
g_object_set(vrate, "max-rate", WEBRTCPUSH_H264_FPS, "drop-only", TRUE, NULL);
configure_pre_enc_queue(q);
configure_post_enc_queue(eq);
g_object_set(parse, "config-interval", -1, NULL);
{
gchar *cap_str = g_strdup_printf(
"video/x-raw,format=NV12,width=%d,height=%d,framerate=%d/1",
MPP_VIDEO_WIDTH, MPP_VIDEO_HEIGHT, WEBRTCPUSH_H264_FPS);
GstCaps *caps = gst_caps_from_string(cap_str);
g_object_set(caps_nv12, "caps", caps, NULL);
gst_caps_unref(caps);
g_free(cap_str);
}
{
GstCaps *hc = gst_caps_from_string(
"video/x-h264,stream-format=byte-stream,alignment=au");
g_object_set(caps_h264, "caps", hc, NULL);
gst_caps_unref(hc);
}
g_object_set(sink,
"emit-signals", FALSE,
"sync", FALSE,
"max-buffers", WEBRTCPUSH_APPSINK_MAX_BUFFERS,
"drop", TRUE,
NULL);
{
GstElement *cap_elems[] = {vsrc, caps_in, jpegparse, mjpeg_sink};
GstElement *enc_elems[] = {rawsrc, vrate, caps_nv12, q, enc, parse,
caps_h264, eq, sink};
if (!link_pipeline(cap_pipe, cap_elems, G_N_ELEMENTS(cap_elems))) {
if (error_message)
*error_message = g_strdup("failed to link MJPEG capture chain");
goto fail;
}
if (!link_pipeline(enc_pipe, enc_elems, G_N_ELEMENTS(enc_elems))) {
if (error_message)
*error_message = g_strdup("failed to link appsrc H264 encode chain");
goto fail;
}
}
src->manual_mjpeg = TRUE;
src->capture_pipeline = cap_pipe;
src->pipeline = enc_pipe;
src->vsrc = vsrc;
src->mjpeg_sink = mjpeg_sink;
src->raw_appsrc = rawsrc;
src->enc = enc;
src->parse = parse;
src->eq = eq;
src->appsink = sink;
src->use_mpp = use_mpp;
{
GstPad *enc_sink = gst_element_get_static_pad(enc, "sink");
if (enc_sink) {
src->enc_sink_probe_id = gst_pad_add_probe(
enc_sink, GST_PAD_PROBE_TYPE_BUFFER, enc_sink_idr_probe, src, NULL);
gst_object_unref(enc_sink);
}
}
{
GstBus *bus = gst_element_get_bus(enc_pipe);
if (bus) {
src->bus_watch_id = gst_bus_add_watch(bus, bus_watch_cb, src);
gst_object_unref(bus);
}
}
{
GstBus *bus = gst_element_get_bus(cap_pipe);
if (bus) {
src->capture_bus_watch_id = gst_bus_add_watch(bus, bus_watch_cb, src);
gst_object_unref(bus);
}
}
gst_pipeline_set_latency(GST_PIPELINE(enc_pipe), 0);
gst_pipeline_set_latency(GST_PIPELINE(cap_pipe), 0);
ret = gst_element_set_state(enc_pipe, GST_STATE_PLAYING);
if (ret == GST_STATE_CHANGE_FAILURE) {
if (error_message)
*error_message = g_strdup("manual encoder pipeline failed to PLAYING");
goto fail_started;
}
ret = gst_element_set_state(cap_pipe, GST_STATE_PLAYING);
if (ret == GST_STATE_CHANGE_FAILURE) {
if (error_message)
*error_message = g_strdup("manual capture pipeline failed to PLAYING");
goto fail_started;
}
src->decode_thread = g_thread_new("mjpeg-decode", mjpeg_decode_thread, src);
{
const guint8 *warm_data;
size_t warm_size;
gboolean warm_idr;
if (!pull_sample(src, FALSE, 8000, &warm_data, &warm_size, &warm_idr, NULL)) {
if (error_message)
*error_message = g_strdup("manual encoder warmup timeout (no frame in 8s)");
my_zlog_error("mpp_h264_source: manual MJPEG warmup failed");
goto fail_started;
}
my_zlog_info("mpp_h264_source: warmup OK first_au=%zu bytes idr=%d",
warm_size, warm_idr ? 1 : 0);
}
my_zlog_info("mpp_h264_source: 720p MJPEG@%dfps -> libjpeg-turbo(I420) -> NV12@%dfps -> mpph264enc",
mjpeg_fps, WEBRTCPUSH_H264_FPS);
my_zlog_info("mpp_h264_source: live encoder started 720p (mpp=%d pre_buf=%d post_buf=%d)",
use_mpp ? 1 : 0, WEBRTCPUSH_MPP_PRE_ENC_BUFFERS,
WEBRTCPUSH_MPP_POST_ENC_BUFFERS);
return TRUE;
fail_started:
if (src) {
src->stop_decode = TRUE;
if (src->raw_appsrc)
gst_app_src_end_of_stream(GST_APP_SRC(src->raw_appsrc));
if (src->capture_pipeline)
gst_element_set_state(src->capture_pipeline, GST_STATE_NULL);
if (src->pipeline)
gst_element_set_state(src->pipeline, GST_STATE_NULL);
if (src->decode_thread) {
g_thread_join(src->decode_thread);
src->decode_thread = NULL;
}
if (src->bus_watch_id) {
g_source_remove(src->bus_watch_id);
src->bus_watch_id = 0;
}
if (src->capture_bus_watch_id) {
g_source_remove(src->capture_bus_watch_id);
src->capture_bus_watch_id = 0;
}
if (src->capture_pipeline) {
gst_object_unref(src->capture_pipeline);
src->capture_pipeline = NULL;
}
if (src->pipeline) {
gst_object_unref(src->pipeline);
src->pipeline = NULL;
}
src->vsrc = NULL;
src->mjpeg_sink = NULL;
src->raw_appsrc = NULL;
src->enc = NULL;
src->parse = NULL;
src->eq = NULL;
src->appsink = NULL;
}
return FALSE;
fail:
if (cap_pipe) {
gst_element_set_state(cap_pipe, GST_STATE_NULL);
gst_object_unref(cap_pipe);
}
if (enc_pipe) {
gst_element_set_state(enc_pipe, GST_STATE_NULL);
gst_object_unref(enc_pipe);
}
return FALSE;
}
MppH264Source *mpp_h264_source_start(const char *video_device, char **error_message) MppH264Source *mpp_h264_source_start(const char *video_device, char **error_message)
{ {
MppH264Source *src; MppH264Source *src;
GstElement *pipe, *vsrc, *caps_in = NULL, *jpegparse = NULL, *jpegdec = NULL; GstElement *pipe, *vsrc, *caps_in = NULL;
GstElement *conv = NULL, *vrate, *caps_nv12, *enc, *parse, *q, *eq, *caps_h264, *sink; GstElement *conv = NULL, *vrate, *caps_nv12, *enc, *parse, *q, *eq, *caps_h264, *sink;
gchar *auto_v4l = NULL; gchar *auto_v4l = NULL;
const gchar *vdev = video_device; const gchar *vdev = video_device;
gboolean use_test = FALSE; gboolean use_test = FALSE;
gboolean use_auto = FALSE; gboolean use_auto = FALSE;
gboolean use_mpp = FALSE; gboolean use_mpp = FALSE;
gboolean use_mpp_jpeg = FALSE;
gint mjpeg_fps = WEBRTCPUSH_H264_FPS;
CaptureMode capture_mode = CAPTURE_NONE; CaptureMode capture_mode = CAPTURE_NONE;
GstStateChangeReturn ret; GstStateChangeReturn ret;
...@@ -583,6 +1378,7 @@ MppH264Source *mpp_h264_source_start(const char *video_device, char **error_mess ...@@ -583,6 +1378,7 @@ MppH264Source *mpp_h264_source_start(const char *video_device, char **error_mess
gst_init(NULL, NULL); gst_init(NULL, NULL);
src = g_new0(MppH264Source, 1); src = g_new0(MppH264Source, 1);
g_mutex_init(&src->timing_lock);
use_test = (vdev && g_strcmp0(vdev, "test") == 0); use_test = (vdev && g_strcmp0(vdev, "test") == 0);
use_auto = (!vdev || g_strcmp0(vdev, "auto") == 0); use_auto = (!vdev || g_strcmp0(vdev, "auto") == 0);
...@@ -609,6 +1405,16 @@ MppH264Source *mpp_h264_source_start(const char *video_device, char **error_mess ...@@ -609,6 +1405,16 @@ MppH264Source *mpp_h264_source_start(const char *video_device, char **error_mess
return NULL; return NULL;
} }
if (!use_test && capture_mode != CAPTURE_RAW_NV12_720P) {
if (start_manual_mjpeg_pipeline(src, vdev, capture_mode, error_message)) {
g_free(auto_v4l);
return src;
}
g_free(auto_v4l);
g_free(src);
return NULL;
}
pipe = gst_pipeline_new("mpp-h264-pipe"); pipe = gst_pipeline_new("mpp-h264-pipe");
vrate = gst_element_factory_make("videorate", "vrate"); vrate = gst_element_factory_make("videorate", "vrate");
caps_nv12 = gst_element_factory_make("capsfilter", "caps_nv12"); caps_nv12 = gst_element_factory_make("capsfilter", "caps_nv12");
...@@ -638,62 +1444,6 @@ MppH264Source *mpp_h264_source_start(const char *video_device, char **error_mess ...@@ -638,62 +1444,6 @@ MppH264Source *mpp_h264_source_start(const char *video_device, char **error_mess
g_free(cap_str); g_free(cap_str);
} }
my_zlog_info("mpp_h264_source: 720p NV12 direct capture (no JPEG decode)"); my_zlog_info("mpp_h264_source: 720p NV12 direct capture (no JPEG decode)");
} else {
vsrc = gst_element_factory_make("v4l2src", "vsrc");
caps_in = gst_element_factory_make("capsfilter", "caps_in");
jpegparse = gst_element_factory_make("jpegparse", "jpegparse");
if (WEBRTCPUSH_MJPEG_CPU_DECODE) {
jpegdec = gst_element_factory_make("jpegdec", "jpegdec");
use_mpp_jpeg = FALSE;
if (jpegdec) {
conv = gst_element_factory_make("videoconvert", "jpegconv");
g_object_set(jpegdec,
"idct-method", 1, /* ifast */
"discard-corrupted-frames", TRUE,
"qos", TRUE,
NULL);
} else {
jpegdec = gst_element_factory_make("mppjpegdec", "jpegdec");
if (jpegdec) {
use_mpp_jpeg = TRUE;
g_object_set(jpegdec, "format", 23, "fast-mode", TRUE, NULL);
}
}
} else {
jpegdec = gst_element_factory_make("mppjpegdec", "jpegdec");
if (jpegdec) {
use_mpp_jpeg = TRUE;
g_object_set(jpegdec, "format", 23, "fast-mode", TRUE, NULL);
} else {
jpegdec = gst_element_factory_make("jpegdec", "jpegdec");
use_mpp_jpeg = FALSE;
if (jpegdec) {
conv = gst_element_factory_make("videoconvert", "jpegconv");
g_object_set(jpegdec,
"idct-method", 1,
"discard-corrupted-frames", TRUE,
"qos", TRUE,
NULL);
}
}
}
mjpeg_fps = (capture_mode == CAPTURE_MJPEG_720P24) ? WEBRTCPUSH_H264_FPS : 30;
g_object_set(vsrc, "device", vdev, NULL);
configure_v4l2_src(vsrc, TRUE);
{
gchar *cap_str = g_strdup_printf(
"image/jpeg,width=%d,height=%d,framerate=%d/1",
MPP_VIDEO_WIDTH, MPP_VIDEO_HEIGHT, mjpeg_fps);
GstCaps *caps = gst_caps_from_string(cap_str);
g_object_set(caps_in, "caps", caps, NULL);
gst_caps_unref(caps);
g_free(cap_str);
}
my_zlog_info("mpp_h264_source: 720p MJPEG@%dfps -> %s%s -> NV12@%dfps",
mjpeg_fps,
use_mpp_jpeg ? "mppjpegdec" : "jpegdec",
use_mpp_jpeg ? "" : "+videoconvert",
WEBRTCPUSH_H264_FPS);
} }
if (!pipe || !vsrc || !vrate || !caps_nv12 || !q || !eq || !enc || !parse || if (!pipe || !vsrc || !vrate || !caps_nv12 || !q || !eq || !enc || !parse ||
...@@ -756,27 +1506,6 @@ MppH264Source *mpp_h264_source_start(const char *video_device, char **error_mess ...@@ -756,27 +1506,6 @@ MppH264Source *mpp_h264_source_start(const char *video_device, char **error_mess
g_free(src); g_free(src);
return NULL; return NULL;
} }
} else {
gboolean linked;
if (use_mpp_jpeg) {
GstElement *elems[] = {vsrc, caps_in, jpegparse, jpegdec, vrate,
caps_nv12, q, enc, parse, caps_h264, eq, sink};
linked = jpegparse && jpegdec &&
link_pipeline(pipe, elems, G_N_ELEMENTS(elems));
} else {
GstElement *elems[] = {vsrc, caps_in, jpegparse, jpegdec, conv, vrate,
caps_nv12, q, enc, parse, caps_h264, eq, sink};
linked = jpegparse && jpegdec && conv &&
link_pipeline(pipe, elems, G_N_ELEMENTS(elems));
}
if (!linked) {
if (error_message)
*error_message = g_strdup("failed to link MJPEG 720p chain");
gst_object_unref(pipe);
g_free(auto_v4l);
g_free(src);
return NULL;
}
} }
src->vsrc = vsrc; src->vsrc = vsrc;
...@@ -878,6 +1607,11 @@ static gboolean pull_sample(MppH264Source *src, gboolean require_idr, ...@@ -878,6 +1607,11 @@ static gboolean pull_sample(MppH264Source *src, gboolean require_idr,
} }
pts = GST_BUFFER_PTS(buffer); pts = GST_BUFFER_PTS(buffer);
if (src->manual_mjpeg && GST_CLOCK_TIME_IS_VALID(pts)) {
FrameTiming timing;
if (timing_take(src, (guint64)pts, &timing))
perf_record(src, &timing, g_get_monotonic_time());
}
idr = buffer_is_idr(src->frame_buf, src->frame_size); idr = buffer_is_idr(src->frame_buf, src->frame_size);
if (idr && src->pending_idr) { if (idr && src->pending_idr) {
src->pending_idr = FALSE; src->pending_idr = FALSE;
...@@ -991,6 +1725,8 @@ gboolean mpp_h264_source_recover(MppH264Source *src) ...@@ -991,6 +1725,8 @@ gboolean mpp_h264_source_recover(MppH264Source *src)
my_zlog_warn("mpp_h264_source: recovering pipeline (NULL -> PLAYING)"); my_zlog_warn("mpp_h264_source: recovering pipeline (NULL -> PLAYING)");
src->pending_idr = FALSE; src->pending_idr = FALSE;
src->pending_recover = FALSE; src->pending_recover = FALSE;
if (src->capture_pipeline)
gst_element_set_state(src->capture_pipeline, GST_STATE_NULL);
gst_element_set_state(src->pipeline, GST_STATE_NULL); gst_element_set_state(src->pipeline, GST_STATE_NULL);
/* 重新探测摄像头设备（USB 重枚举后可能从 video0 变成 video1/2） */ /* 重新探测摄像头设备（USB 重枚举后可能从 video0 变成 video1/2） */
if (src->vsrc) { if (src->vsrc) {
...@@ -1013,6 +1749,13 @@ gboolean mpp_h264_source_recover(MppH264Source *src) ...@@ -1013,6 +1749,13 @@ gboolean mpp_h264_source_recover(MppH264Source *src)
my_zlog_error("mpp_h264_source: recover failed to reach PLAYING"); my_zlog_error("mpp_h264_source: recover failed to reach PLAYING");
return FALSE; return FALSE;
} }
if (src->capture_pipeline) {
ret = gst_element_set_state(src->capture_pipeline, GST_STATE_PLAYING);
if (ret == GST_STATE_CHANGE_FAILURE) {
my_zlog_error("mpp_h264_source: capture recover failed to reach PLAYING");
return FALSE;
}
}
if (!pull_sample(src, FALSE, 5000, &data, &size, &idr, NULL)) { if (!pull_sample(src, FALSE, 5000, &data, &size, &idr, NULL)) {
my_zlog_error("mpp_h264_source: recover warmup failed"); my_zlog_error("mpp_h264_source: recover warmup failed");
return FALSE; return FALSE;
...@@ -1036,12 +1779,31 @@ void mpp_h264_source_stop(MppH264Source *src) ...@@ -1036,12 +1779,31 @@ void mpp_h264_source_stop(MppH264Source *src)
if (!src) if (!src)
return; return;
src->stop_decode = TRUE;
if (src->raw_appsrc)
gst_app_src_end_of_stream(GST_APP_SRC(src->raw_appsrc));
if (src->capture_pipeline)
gst_element_set_state(src->capture_pipeline, GST_STATE_NULL);
if (src->pipeline)
gst_element_set_state(src->pipeline, GST_STATE_NULL);
if (src->decode_thread) {
g_thread_join(src->decode_thread);
src->decode_thread = NULL;
}
if (src->bus_watch_id) { if (src->bus_watch_id) {
g_source_remove(src->bus_watch_id); g_source_remove(src->bus_watch_id);
src->bus_watch_id = 0; src->bus_watch_id = 0;
} }
if (src->capture_bus_watch_id) {
g_source_remove(src->capture_bus_watch_id);
src->capture_bus_watch_id = 0;
}
if (src->capture_pipeline) {
gst_object_unref(src->capture_pipeline);
src->capture_pipeline = NULL;
}
if (src->pipeline) { if (src->pipeline) {
gst_element_set_state(src->pipeline, GST_STATE_NULL);
gst_object_unref(src->pipeline); gst_object_unref(src->pipeline);
src->pipeline = NULL; src->pipeline = NULL;
} }
...@@ -1049,7 +1811,14 @@ void mpp_h264_source_stop(MppH264Source *src) ...@@ -1049,7 +1811,14 @@ void mpp_h264_source_stop(MppH264Source *src)
src->parse = NULL; src->parse = NULL;
src->eq = NULL; src->eq = NULL;
src->appsink = NULL; src->appsink = NULL;
src->mjpeg_sink = NULL;
src->raw_appsrc = NULL;
g_free(src->sprop_parameter_sets); g_free(src->sprop_parameter_sets);
g_free(src->frame_buf); g_free(src->frame_buf);
g_free(src->i420_buf);
g_free(src->nv12_buf);
g_free(src->jpeg_tmp_u);
g_free(src->jpeg_tmp_v);
g_mutex_clear(&src->timing_lock);
g_free(src); g_free(src);
} }
modules/webrtcpush/rtc_client.c
View file @ 3585cc8c
...@@ -10,9 +10,14 @@ ...@@ -10,9 +10,14 @@
#include <rtc/rtc.h> #include <rtc/rtc.h>
#include <glib/gstdio.h>
#include <stdio.h> #include <stdio.h>
#include <stdlib.h>
#include <string.h> #include <string.h>
#define WEBRTCPUSH_DC_AUDIO_MAX_BYTES (2U * 1024U * 1024U)
#define WEBRTCPUSH_DC_AUDIO_DIR "/tmp/webrtcpush_audio"
typedef struct { typedef struct {
gchar *candidate; gchar *candidate;
gchar *mid; gchar *mid;
...@@ -49,6 +54,13 @@ typedef struct { ...@@ -49,6 +54,13 @@ typedef struct {
} RtcFrameQueue; } RtcFrameQueue;
typedef struct { typedef struct {
guint8 *data;
gsize size;
guint seq;
void *client_ptr; /* RtcClient*, 用void*避免前向声明问题 */
} DcMp3PlayTask;
typedef struct {
AppState *app; AppState *app;
GMutex lock; GMutex lock;
GMutex send_lock; GMutex send_lock;
...@@ -71,13 +83,16 @@ typedef struct { ...@@ -71,13 +83,16 @@ typedef struct {
guint64 last_source_pts; guint64 last_source_pts;
gboolean source_pts_valid; gboolean source_pts_valid;
guint target_bitrate; guint target_bitrate;
guint applied_mpp_bitrate;
guint remb_ceiling; guint remb_ceiling;
guint last_remb_bitrate; guint last_remb_bitrate;
guint remb_filtered; guint remb_filtered;
guint remb_down_samples; guint remb_down_samples;
gint64 last_bitrate_ramp_us; gint64 last_bitrate_ramp_us;
gint64 last_mpp_bitrate_apply_us;
gint64 last_idr_request_us; gint64 last_idr_request_us;
gint64 last_stats_log_us; gint64 last_stats_log_us;
gint64 last_remb_sane_log_us;
gchar *remote_ice_ufrag; gchar *remote_ice_ufrag;
gchar *video_mid; gchar *video_mid;
gint video_mline_index; gint video_mline_index;
...@@ -87,15 +102,24 @@ typedef struct { ...@@ -87,15 +102,24 @@ typedef struct {
AudioSink *audio_sink; AudioSink *audio_sink;
VolumeControl *volume_ctrl; VolumeControl *volume_ctrl;
int audio_track; int audio_track;
int audio_payload_type;
gboolean audio_track_open; gboolean audio_track_open;
gboolean audio_send_enabled; gboolean audio_send_enabled;
gboolean audio_recv_enabled;
GThread *audio_send_thread; GThread *audio_send_thread;
guint32 audio_rtp_timestamp; guint32 audio_rtp_timestamp;
guint64 audio_pts_base; guint64 audio_pts_base;
guint64 audio_last_pts; guint64 audio_last_pts;
gboolean audio_pts_valid; gboolean audio_pts_valid;
/* DataChannel: 前端 MP3 分片 + 右上角 Mbps */
int data_channel;
GByteArray *dc_audio_buf;
guint dc_audio_seq;
/* 排队播放: 多片段连续播放不间隔 */
GAsyncQueue *dc_play_queue; /* DcMp3PlayTask队列 */
GThread *dc_play_thread; /* 常驻播放线程 */
volatile gboolean dc_play_quit;
GQueue *pending_local_ice; GQueue *pending_local_ice;
GQueue *pending_remote_ice; GQueue *pending_remote_ice;
} RtcClient; } RtcClient;
...@@ -527,6 +551,10 @@ static gboolean request_idr(RtcClient *client, gboolean bypass_throttle) ...@@ -527,6 +551,10 @@ static gboolean request_idr(RtcClient *client, gboolean bypass_throttle)
return FALSE; return FALSE;
now = g_get_monotonic_time(); now = g_get_monotonic_time();
g_mutex_lock(&client->lock); g_mutex_lock(&client->lock);
if (client->need_idr) {
g_mutex_unlock(&client->lock);
return FALSE;
}
if (!bypass_throttle && client->last_idr_request_us > 0 && if (!bypass_throttle && client->last_idr_request_us > 0 &&
now - client->last_idr_request_us < now - client->last_idr_request_us <
(gint64)WEBRTCPUSH_PLI_IDR_THROTTLE_MS * 1000) { (gint64)WEBRTCPUSH_PLI_IDR_THROTTLE_MS * 1000) {
...@@ -582,29 +610,241 @@ static void RTC_API on_state_change(int pc, rtcState state, void *ptr) ...@@ -582,29 +610,241 @@ static void RTC_API on_state_change(int pc, rtcState state, void *ptr)
} }
} }
/* 常驻播放线程: 从队列取任务连续播放, 队列空时释放ALSA锁 */
static gpointer dc_play_queue_thread(gpointer data)
{
RtcClient *client = data;
gboolean holds_lock = FALSE;
int idle_count = 0;
while (!client->dc_play_quit) {
/* 等300ms看有没有新任务, 有就播, 连续3次空就释放锁 */
DcMp3PlayTask *task = g_async_queue_timeout_pop(client->dc_play_queue, 300000);
if (!task) {
/* 队列空 */
if (holds_lock) {
idle_count++;
if (idle_count >= 3) {
/* 约1s无新任务, 释放ALSA锁让本地音频能播 */
audio_sink_unlock_alsa();
holds_lock = FALSE;
idle_count = 0;
}
}
continue;
}
idle_count = 0;
if (!task->data || task->size == 0) {
g_free(task->data);
g_free(task);
continue;
}
/* 首次或有任务时获取锁, 播放期间不释放 */
if (!holds_lock) {
RtcClient *rc = (RtcClient*)task->client_ptr;
audio_sink_interrupt(rc ? rc->audio_sink : NULL);
audio_sink_lock_alsa();
holds_lock = TRUE;
}
/* 写文件 + playbin播放 */
gchar *path = g_strdup_printf("%s/phone_%u.bin", WEBRTCPUSH_DC_AUDIO_DIR, task->seq);
GError *err = NULL;
if (g_mkdir_with_parents(WEBRTCPUSH_DC_AUDIO_DIR, 0700) != 0 ||
!g_file_set_contents(path, (const gchar*)task->data, (gssize)task->size, &err)) {
my_zlog_warn("libdatachannel: save phone audio failed: %s", err ? err->message : "?");
if (err) g_error_free(err);
g_free(path);
g_free(task->data);
g_free(task);
continue;
}
gchar *quoted = g_shell_quote(path);
gchar *cmd = g_strdup_printf(
/*
* 修复两个问题:
* 1) sync=false 会让 playbin 在 EOS 时立即关闭 ALSA 设备, ring buffer
* 尾部尚未播完就被 snd_pcm_drop 丢弃, 表现为"只播放前面部分"。
* 改 sync=true 让 sink 按时钟节拍播放并在 EOS 时 drain 完整音频。
* 2) USB 声卡(plughw:2,0)播单声道会触发
* gst_audio_ring_buffer_set_channel_positions 的 "should not be reached"
* CRITICAL; 用 audioconvert 上混到 2 声道规避。
*/
"timeout 20s gst-launch-1.0 -q playbin uri=file://%s audio-sink=\"audioconvert ! audioresample ! audio/x-raw,channels=2 ! alsasink device=%s sync=true\" video-sink=fakesink",
quoted, WEBRTCPUSH_AUDIO_PLAYBACK_DEVICE);
int status = -1;
gchar *child_out = NULL, *child_err = NULL;
my_zlog_info("libdatachannel: playing %u bytes (seq=%u)", (guint)task->size, task->seq);
/* capture & discard child stdout/stderr: suppress rga_api version + gst progress noise */
if (!g_spawn_command_line_sync(cmd, &child_out, &child_err, &status, &err)) {
my_zlog_warn("libdatachannel: play phone audio failed: %s", err ? err->message : "?");
if (err) g_error_free(err);
} else if (status == 0) {
my_zlog_info("libdatachannel: phone audio played (%u bytes)", (guint)task->size);
} else {
my_zlog_warn("libdatachannel: phone audio playback exit=%d (%u bytes)",
status, (guint)task->size);
}
g_remove(path);
g_free(child_out);
g_free(child_err);
g_free(cmd);
g_free(quoted);
g_free(path);
g_free(task->data);
g_free(task);
}
/* 退出前释放锁 */
if (holds_lock)
audio_sink_unlock_alsa();
return NULL;
}
static void start_dc_mp3_playback(RtcClient *client, guint8 *data, gsize size)
{
DcMp3PlayTask *task;
if (!client || !data || size == 0)
return;
/* 首次播放时创建队列+线程 */
if (!client->dc_play_queue) {
client->dc_play_queue = g_async_queue_new();
client->dc_play_quit = FALSE;
client->dc_play_thread = g_thread_new("dc-play-q", dc_play_queue_thread, client);
}
task = g_new0(DcMp3PlayTask, 1);
task->data = data;
task->size = size;
task->seq = ++client->dc_audio_seq;
task->client_ptr = client;
g_async_queue_push(client->dc_play_queue, task);
}
static gboolean dc_message_is_eof(const char *message, int size)
{
if (!message)
return FALSE;
if (size < 0)
return strcmp(message, "EOF") == 0;
return size == 3 && memcmp(message, "EOF", 3) == 0;
}
/* 接收手机发来的 Opus RTP，depayload 后送入播放管道 */ static void RTC_API on_data_channel_message(int dc, const char *message, int size, void *ptr)
static void RTC_API on_audio_message(int track, const char *message, int size, void *ptr)
{ {
RtcClient *client = ptr; RtcClient *client = ptr;
const guint8 *rtp = (const guint8 *)message; guint8 *play_data = NULL;
const guint8 *payload; gsize play_size = 0;
int payload_len, header_len;
(void)track; (void)dc;
if (!client || !client->audio_sink || !client->audio_recv_enabled || if (!client || !message)
!message || size < 12) return;
if (dc_message_is_eof(message, size)) {
g_mutex_lock(&client->lock);
if (client->dc_audio_buf && client->dc_audio_buf->len > 0) {
play_size = client->dc_audio_buf->len;
play_data = g_memdup2(client->dc_audio_buf->data, play_size);
g_byte_array_set_size(client->dc_audio_buf, 0);
}
g_mutex_unlock(&client->lock);
if (play_data) {
my_zlog_info("libdatachannel: DataChannel MP3 EOF, playback %u bytes", (guint)play_size);
start_dc_mp3_playback(client, play_data, play_size);
} else {
my_zlog_warn("libdatachannel: DataChannel MP3 EOF with empty buffer");
}
return; return;
header_len = 12 + ((rtp[0] & 0x0F) * 4);
if (size > header_len && (rtp[0] & 0x10)) {
int ext_len = ((rtp[header_len + 2] << 8) | rtp[header_len + 3]) * 4 + 4;
header_len += ext_len;
} }
if (size <= header_len)
if (size <= 0)
return;
g_mutex_lock(&client->lock);
if (!client->dc_audio_buf)
client->dc_audio_buf = g_byte_array_new();
if (client->dc_audio_buf->len + (guint)size > WEBRTCPUSH_DC_AUDIO_MAX_BYTES) {
my_zlog_warn("libdatachannel: DataChannel MP3 too large (%u + %d), drop buffer",
client->dc_audio_buf->len, size);
g_byte_array_set_size(client->dc_audio_buf, 0);
} else {
g_byte_array_append(client->dc_audio_buf, (const guint8 *)message, (guint)size);
}
g_mutex_unlock(&client->lock);
}
static void send_datachannel_mbps(RtcClient *client, guint actual_kbps)
{
int dc;
guint target_bitrate;
char msg[32];
double mbps;
if (!client)
return;
g_mutex_lock(&client->lock);
dc = client->data_channel;
target_bitrate = client->target_bitrate;
g_mutex_unlock(&client->lock);
if (dc < 0 || !rtcIsOpen(dc))
return; return;
payload = rtp + header_len; mbps = actual_kbps > 0 ? ((double)actual_kbps / 1000.0) :
payload_len = size - header_len; ((double)target_bitrate / 1000000.0);
audio_sink_push_opus(client->audio_sink, payload, (size_t)payload_len); snprintf(msg, sizeof(msg), "%.2f", mbps);
rtcSendMessage(dc, msg, -1);
}
static void RTC_API on_data_channel_open(int dc, void *ptr)
{
RtcClient *client = ptr;
char label[64] = {0};
rtcGetDataChannelLabel(dc, label, (int)sizeof(label));
my_zlog_info("libdatachannel: DataChannel open label=%s", label[0] ? label : "?");
send_datachannel_mbps(client, 0);
}
static void RTC_API on_data_channel_closed(int dc, void *ptr)
{
RtcClient *client = ptr;
if (!client)
return;
g_mutex_lock(&client->lock);
if (client->data_channel == dc)
client->data_channel = -1;
g_mutex_unlock(&client->lock);
my_zlog_info("libdatachannel: DataChannel closed");
}
static void RTC_API on_data_channel_error(int dc, const char *error, void *ptr)
{
(void)dc;
(void)ptr;
my_zlog_warn("libdatachannel: DataChannel error: %s", error ? error : "?");
}
static void setup_data_channel_callbacks(RtcClient *client, int dc)
{
if (!client || dc < 0)
return;
rtcSetUserPointer(dc, client);
rtcSetOpenCallback(dc, on_data_channel_open);
rtcSetClosedCallback(dc, on_data_channel_closed);
rtcSetErrorCallback(dc, on_data_channel_error);
rtcSetMessageCallback(dc, on_data_channel_message);
}
static void RTC_API on_incoming_data_channel(int pc, int dc, void *ptr)
{
RtcClient *client = ptr;
char label[64] = {0};
(void)pc;
rtcGetDataChannelLabel(dc, label, (int)sizeof(label));
my_zlog_info("libdatachannel: incoming DataChannel label=%s", label[0] ? label : "?");
setup_data_channel_callbacks(client, dc);
} }
static void RTC_API on_track_open(int track, void *ptr) static void RTC_API on_track_open(int track, void *ptr)
...@@ -684,7 +924,7 @@ static void RTC_API on_pli(int track, void *ptr) ...@@ -684,7 +924,7 @@ static void RTC_API on_pli(int track, void *ptr)
return; return;
requested = request_idr(client, FALSE); requested = request_idr(client, FALSE);
if (!requested) { if (!requested) {
my_zlog_debug("libdatachannel: PLI throttled, skip duplicate IDR request"); my_zlog_debug("libdatachannel: PLI ignored, IDR already pending or throttled");
return; return;
} }
g_mutex_lock(&client->lock); g_mutex_lock(&client->lock);
...@@ -771,18 +1011,41 @@ static int clear_stale_pacing(RtcClient *client, int track, ...@@ -771,18 +1011,41 @@ static int clear_stale_pacing(RtcClient *client, int track,
return dropped; return dropped;
} }
static guint bitrate_delta(guint a, guint b)
{
return a > b ? a - b : b - a;
}
static void apply_bitrate(RtcClient *client, int track, guint bitrate) static void apply_bitrate(RtcClient *client, int track, guint bitrate)
{ {
guint b; guint b;
guint old_b; guint old_b;
guint old_mpp_b;
guint mpp_delta;
gint64 now_us;
gint64 last_mpp_apply_us;
gboolean apply_mpp = FALSE;
if (!client) if (!client)
return; return;
b = clamp_bitrate(bitrate); b = clamp_bitrate(bitrate);
now_us = g_get_monotonic_time();
g_mutex_lock(&client->lock); g_mutex_lock(&client->lock);
old_b = client->target_bitrate; old_b = client->target_bitrate;
old_mpp_b = client->applied_mpp_bitrate;
last_mpp_apply_us = client->last_mpp_bitrate_apply_us;
if (old_mpp_b == 0)
old_mpp_b = old_b ? old_b : b;
client->target_bitrate = b; client->target_bitrate = b;
mpp_delta = bitrate_delta(old_mpp_b, b);
if (mpp_delta >= WEBRTCPUSH_MPP_RECONFIG_MIN_DELTA_BPS &&
now_us - last_mpp_apply_us >=
(gint64)WEBRTCPUSH_MPP_RECONFIG_MIN_INTERVAL_MS * 1000) {
apply_mpp = TRUE;
client->applied_mpp_bitrate = b;
client->last_mpp_bitrate_apply_us = now_us;
}
g_mutex_unlock(&client->lock); g_mutex_unlock(&client->lock);
if (old_b == b) if (old_b == b)
...@@ -794,10 +1057,16 @@ static void apply_bitrate(RtcClient *client, int track, guint bitrate) ...@@ -794,10 +1057,16 @@ static void apply_bitrate(RtcClient *client, int track, guint bitrate)
rtcSetPacingBitrate(track, pacing_bitrate_for_encoder(b)); rtcSetPacingBitrate(track, pacing_bitrate_for_encoder(b));
g_mutex_unlock(&client->send_lock); g_mutex_unlock(&client->send_lock);
} }
if (apply_mpp) {
mpp_h264_source_set_bitrate(client->h264_source, b); mpp_h264_source_set_bitrate(client->h264_source, b);
my_zlog_info("libdatachannel: RTCP bitrate control -> MPP encoder=%u->%u kbps pacing=%u kbps", my_zlog_info("libdatachannel: RTCP bitrate control -> target=%u->%u kbps, MPP=%u->%u kbps pacing=%u kbps",
old_b / 1000, b / 1000, old_b / 1000, b / 1000, old_mpp_b / 1000, b / 1000,
pacing_bitrate_for_encoder(b) / 1000);
} else {
my_zlog_info("libdatachannel: RTCP bitrate control -> target=%u->%u kbps, MPP hold=%u kbps pacing=%u kbps",
old_b / 1000, b / 1000, old_mpp_b / 1000,
pacing_bitrate_for_encoder(b) / 1000); pacing_bitrate_for_encoder(b) / 1000);
}
} }
/* Public Internet: climb slowly so a transient REMB spike cannot build latency. */ /* Public Internet: climb slowly so a transient REMB spike cannot build latency. */
...@@ -865,44 +1134,77 @@ static void RTC_API on_remb(int track, unsigned int bitrate, void *ptr) ...@@ -865,44 +1134,77 @@ static void RTC_API on_remb(int track, unsigned int bitrate, void *ptr)
RtcClient *client = ptr; RtcClient *client = ptr;
guint current; guint current;
guint usable; guint usable;
guint estimate_bitrate;
guint filtered; guint filtered;
guint ceiling; guint ceiling;
guint next = 0; guint next = 0;
guint down_samples = 0; guint down_samples = 0;
guint required_down_samples = WEBRTCPUSH_REMB_DOWN_CONFIRMATIONS;
gboolean severe_down = FALSE; gboolean severe_down = FALSE;
gboolean can_step_down = FALSE; gboolean can_step_down = FALSE;
gboolean have_pacing;
gboolean first_frame_sent;
gboolean remb_sane_floor = FALSE;
gboolean log_remb_sane_floor = FALSE;
guint pacing_packets = 0;
guint pacing_bytes = 0;
guint pacing_delay_ms = 0;
gint64 now_us; gint64 now_us;
if (!client || bitrate == 0) if (!client || bitrate == 0)
return; return;
now_us = g_get_monotonic_time(); now_us = g_get_monotonic_time();
usable = clamp_bitrate( have_pacing = pacing_queue_snapshot(client, track, &pacing_packets,
(guint)(((guint64)bitrate * WEBRTCPUSH_REMB_UTIL_PERCENT) / 100U)); &pacing_bytes, &pacing_delay_ms);
g_mutex_lock(&client->lock); g_mutex_lock(&client->lock);
current = client->target_bitrate; current = client->target_bitrate;
first_frame_sent = client->first_frame_sent;
client->last_remb_bitrate = bitrate; client->last_remb_bitrate = bitrate;
can_step_down = now_us - client->last_bitrate_ramp_us >= can_step_down = now_us - client->last_bitrate_ramp_us >=
(gint64)WEBRTCPUSH_BITRATE_RAMP_DOWN_MS * 1000; (gint64)WEBRTCPUSH_BITRATE_RAMP_DOWN_MS * 1000;
estimate_bitrate = bitrate;
if (first_frame_sent && have_pacing &&
bitrate < WEBRTCPUSH_REMB_SANE_RAW_MAX_BPS &&
pacing_delay_ms <= WEBRTCPUSH_REMB_SANE_PACING_MAX_MS) {
estimate_bitrate = WEBRTCPUSH_REMB_SANE_FLOOR_BPS;
remb_sane_floor = TRUE;
if (client->last_remb_sane_log_us == 0 ||
now_us - client->last_remb_sane_log_us >= 5000000) {
client->last_remb_sane_log_us = now_us;
log_remb_sane_floor = TRUE;
}
}
usable = clamp_bitrate(
(guint)(((guint64)estimate_bitrate * WEBRTCPUSH_REMB_UTIL_PERCENT) / 100U));
if (client->remb_filtered == 0) { if (client->remb_filtered == 0) {
filtered = bitrate; filtered = estimate_bitrate;
} else if (bitrate < client->remb_filtered) { } else if (estimate_bitrate < client->remb_filtered) {
/* React to a real fall, but do not mirror every 200ms estimator sample. */ /* React to a real fall, but do not mirror every 200ms estimator sample. */
filtered = (guint)(((guint64)client->remb_filtered * 2U + bitrate) / 3U); filtered = (guint)(((guint64)client->remb_filtered * 2U + estimate_bitrate) / 3U);
} else { } else {
/* Rising bandwidth needs sustained evidence. */ /* Rising bandwidth: 3:1 权重，跌落后能较快爬回（原 7:1 太慢） */
filtered = (guint)(((guint64)client->remb_filtered * 7U + bitrate) / 8U); filtered = (guint)(((guint64)client->remb_filtered * 3U + estimate_bitrate) / 4U);
} }
/*
* 浏览器/手机端的 REMB 有时会在 pacing 几乎为空时给出 500~900kbps
* 的低估值。这个场景不是网络堵塞，若继续拿旧 filtered 参与 ceiling，
* 会把 720p 长时间压回 800kbps。保护命中时把 filtered 也抬到 sane floor，
* 让发送端按“可疑但不堵”的网络慢慢爬升，而不是突然降清晰度。
*/
if (remb_sane_floor && filtered < estimate_bitrate)
filtered = estimate_bitrate;
severe_down = ((guint64)usable * 100U) < severe_down = ((guint64)usable * 100U) <
((guint64)current * WEBRTCPUSH_REMB_SEVERE_PERCENT); ((guint64)current * WEBRTCPUSH_REMB_SEVERE_PERCENT);
if (severe_down) { if (severe_down) {
/* Reset the filter too, otherwise its old high value immediately rebounds. */ /* Reset the filter too, otherwise its old high value immediately rebounds. */
filtered = bitrate; filtered = estimate_bitrate;
ceiling = usable; ceiling = usable;
client->remb_down_samples = 0; required_down_samples = WEBRTCPUSH_REMB_SEVERE_CONFIRMATIONS;
} else { } else {
ceiling = clamp_bitrate( ceiling = clamp_bitrate(
(guint)(((guint64)filtered * WEBRTCPUSH_REMB_UTIL_PERCENT) / 100U)); (guint)(((guint64)filtered * WEBRTCPUSH_REMB_UTIL_PERCENT) / 100U));
...@@ -911,14 +1213,13 @@ static void RTC_API on_remb(int track, unsigned int bitrate, void *ptr) ...@@ -911,14 +1213,13 @@ static void RTC_API on_remb(int track, unsigned int bitrate, void *ptr)
client->remb_ceiling = ceiling; client->remb_ceiling = ceiling;
if (ceiling + WEBRTCPUSH_REMB_DOWN_MIN_STEP <= current) { if (ceiling + WEBRTCPUSH_REMB_DOWN_MIN_STEP <= current) {
if (!severe_down)
client->remb_down_samples++; client->remb_down_samples++;
down_samples = client->remb_down_samples; down_samples = client->remb_down_samples;
if (severe_down || if (client->remb_down_samples >= required_down_samples) {
client->remb_down_samples >= WEBRTCPUSH_REMB_DOWN_CONFIRMATIONS) {
if (!can_step_down) { if (!can_step_down) {
g_mutex_unlock(&client->lock); g_mutex_unlock(&client->lock);
my_zlog_debug("libdatachannel: REMB down held raw=%u filtered=%u ceiling=%u current=%u kbps", my_zlog_debug("libdatachannel: REMB down held %u/%u raw=%u filtered=%u ceiling=%u current=%u kbps",
down_samples, required_down_samples,
bitrate / 1000, filtered / 1000, bitrate / 1000, filtered / 1000,
ceiling / 1000, current / 1000); ceiling / 1000, current / 1000);
return; return;
...@@ -933,24 +1234,33 @@ static void RTC_API on_remb(int track, unsigned int bitrate, void *ptr) ...@@ -933,24 +1234,33 @@ static void RTC_API on_remb(int track, unsigned int bitrate, void *ptr)
} }
g_mutex_unlock(&client->lock); g_mutex_unlock(&client->lock);
if (log_remb_sane_floor) {
my_zlog_info("libdatachannel: RTCP REMB suspicious low raw=%u kbps, use=%u kbps (pacing=%u pkt/%u bytes/%ums)",
bitrate / 1000, estimate_bitrate / 1000,
pacing_packets, pacing_bytes, pacing_delay_ms);
}
if (next) { if (next) {
apply_bitrate(client, track, next); apply_bitrate(client, track, next);
if (severe_down) { if (severe_down) {
my_zlog_info("libdatachannel: RTCP REMB raw=%u filtered=%u kbps, encoder=%u kbps ceiling=%u kbps, severe smoothed", my_zlog_info("libdatachannel: RTCP REMB raw=%u%s filtered=%u kbps, encoder=%u kbps ceiling=%u kbps, severe smoothed",
bitrate / 1000, filtered / 1000, next / 1000, bitrate / 1000, remb_sane_floor ? " sane-floor" : "",
ceiling / 1000); filtered / 1000, next / 1000, ceiling / 1000);
} else { } else {
my_zlog_info("libdatachannel: RTCP REMB raw=%u filtered=%u kbps, encoder=%u kbps ceiling=%u kbps after %u samples", my_zlog_info("libdatachannel: RTCP REMB raw=%u%s filtered=%u kbps, encoder=%u kbps ceiling=%u kbps after %u samples",
bitrate / 1000, filtered / 1000, next / 1000, bitrate / 1000, remb_sane_floor ? " sane-floor" : "",
ceiling / 1000, down_samples); filtered / 1000, next / 1000, ceiling / 1000,
down_samples);
} }
} else if (ceiling > current) { } else if (ceiling > current) {
my_zlog_debug("libdatachannel: REMB raw=%u filtered=%u ceiling=%u kbps (slow ramp)", my_zlog_debug("libdatachannel: REMB raw=%u%s filtered=%u ceiling=%u kbps (slow ramp)",
bitrate / 1000, filtered / 1000, ceiling / 1000); bitrate / 1000, remb_sane_floor ? " sane-floor" : "",
filtered / 1000, ceiling / 1000);
} else if (down_samples > 0) { } else if (down_samples > 0) {
my_zlog_debug("libdatachannel: REMB down candidate %u/%u raw=%u filtered=%u kbps", my_zlog_debug("libdatachannel: REMB down candidate %u/%u raw=%u%s filtered=%u kbps",
down_samples, WEBRTCPUSH_REMB_DOWN_CONFIRMATIONS, down_samples, required_down_samples,
bitrate / 1000, filtered / 1000); bitrate / 1000, remb_sane_floor ? " sane-floor" : "",
filtered / 1000);
} }
} }
...@@ -1117,29 +1427,46 @@ static gpointer send_thread_main(gpointer data) ...@@ -1117,29 +1427,46 @@ static gpointer send_thread_main(gpointer data)
now - last_pacing_flush_us >= now - last_pacing_flush_us >=
(gint64)WEBRTCPUSH_PACING_GUARD_COOLDOWN_MS * 1000) { (gint64)WEBRTCPUSH_PACING_GUARD_COOLDOWN_MS * 1000) {
guint dropped_frames = 0; guint dropped_frames = 0;
int dropped = clear_stale_pacing(client, track, &dropped_frames); int dropped;
gboolean in_idr_grace;
gboolean should_request_idr; gboolean should_request_idr;
gint64 last_idr_us; gint64 last_idr_us;
gint64 last_idr_request_us;
gboolean need_idr_pending; gboolean need_idr_pending;
last_pacing_flush_us = now;
g_mutex_lock(&client->lock); g_mutex_lock(&client->lock);
last_idr_us = client->last_idr_sent_us; last_idr_us = client->last_idr_sent_us;
last_idr_request_us = client->last_idr_request_us;
need_idr_pending = client->need_idr; need_idr_pending = client->need_idr;
in_idr_grace =
last_idr_us > 0 &&
now - last_idr_us <
(gint64)WEBRTCPUSH_PACING_IDR_GRACE_MS * 1000;
g_mutex_unlock(&client->lock);
if (in_idr_grace) {
my_zlog_warn("libdatachannel: pacing backlog %u packets/%u bytes ~= %ums; IDR grace, hold queue/bitrate",
pacing_packets, pacing_bytes, pacing_delay_ms);
continue;
}
last_pacing_flush_us = now;
dropped = clear_stale_pacing(client, track, &dropped_frames);
should_request_idr = should_request_idr =
!need_idr_pending && !need_idr_pending && dropped > 0 &&
(last_idr_us == 0 || (last_idr_us == 0 ||
now - last_idr_us >= now - last_idr_us >=
(gint64)WEBRTCPUSH_PACING_RESYNC_IDR_MS * 1000) &&
(last_idr_request_us == 0 ||
now - last_idr_request_us >=
(gint64)WEBRTCPUSH_PACING_RESYNC_IDR_MS * 1000); (gint64)WEBRTCPUSH_PACING_RESYNC_IDR_MS * 1000);
g_mutex_unlock(&client->lock);
if (should_request_idr) { if (should_request_idr) {
request_idr(client, TRUE); request_idr(client, TRUE);
my_zlog_warn("libdatachannel: pacing backlog %u packets/%u bytes ~= %ums; dropped=%d RTP/%u encoded and request IDR", my_zlog_warn("libdatachannel: pacing backlog %u packets/%u bytes ~= %ums; dropped=%d RTP/%u encoded and request IDR",
pacing_packets, pacing_bytes, pacing_delay_ms, pacing_packets, pacing_bytes, pacing_delay_ms,
dropped > 0 ? dropped : 0, dropped_frames); dropped > 0 ? dropped : 0, dropped_frames);
} else { } else {
my_zlog_warn("libdatachannel: pacing backlog %u packets/%u bytes ~= %ums; dropped=%d RTP/%u encoded without IDR (recent/pending)", my_zlog_warn("libdatachannel: pacing backlog %u packets/%u bytes ~= %ums; dropped=%d RTP/%u encoded without IDR (recent/pending/no-drop)",
pacing_packets, pacing_bytes, pacing_delay_ms, pacing_packets, pacing_bytes, pacing_delay_ms,
dropped > 0 ? dropped : 0, dropped_frames); dropped > 0 ? dropped : 0, dropped_frames);
} }
...@@ -1172,6 +1499,19 @@ static gpointer send_thread_main(gpointer data) ...@@ -1172,6 +1499,19 @@ static gpointer send_thread_main(gpointer data)
remb_bps / 1000, remb_bps / 1000,
pacing_packets, pacing_bytes, pacing_delay_ms, pacing_packets, pacing_bytes, pacing_delay_ms,
queue_dropped); queue_dropped);
send_datachannel_mbps(client, actual_kbps);
/* 修复: MPP CBR失控时actual>>target, pacing必须跟actual否则永远堆积 */
if (actual_kbps > 0) {
guint actual_bps = actual_kbps * 1000U;
guint new_pacing = (guint)((guint64)actual_bps *
WEBRTCPUSH_PACING_HEADROOM_PERCENT / 100U);
if (new_pacing > WEBRTCPUSH_PACING_MAX_BITRATE)
new_pacing = WEBRTCPUSH_PACING_MAX_BITRATE;
g_mutex_lock(&client->send_lock);
if (client->track == track)
rtcSetPacingBitrate(track, new_pacing);
g_mutex_unlock(&client->send_lock);
}
sent_window_bytes = 0; sent_window_bytes = 0;
sent_window_frames = 0; sent_window_frames = 0;
sent_window_start_us = now; sent_window_start_us = now;
...@@ -1385,17 +1725,11 @@ static gboolean parse_offer(const char *sdp, OfferVideo *video, OfferAudio *audi ...@@ -1385,17 +1725,11 @@ static gboolean parse_offer(const char *sdp, OfferVideo *video, OfferAudio *audi
} }
} }
if (in_audio) { if (in_audio) {
/* 手机 audio 固定 recvonly（只收不发），设备 answer 固定 sendonly，
* 不再解析 offer 方向；仅取 mid 和 opus rtpmap。 */
if (g_str_has_prefix(line, "a=mid:")) { if (g_str_has_prefix(line, "a=mid:")) {
g_free(audio->mid); g_free(audio->mid);
audio->mid = g_strdup(line + strlen("a=mid:")); audio->mid = g_strdup(line + strlen("a=mid:"));
} else if (g_str_has_prefix(line, "a=sendonly")) {
audio->direction = 1;
} else if (g_str_has_prefix(line, "a=recvonly")) {
audio->direction = 2;
} else if (g_str_has_prefix(line, "a=sendrecv")) {
audio->direction = 3;
} else if (g_str_has_prefix(line, "a=inactive")) {
audio->direction = 4;
} else if (g_str_has_prefix(line, "a=rtpmap:")) { } else if (g_str_has_prefix(line, "a=rtpmap:")) {
gint pt = -1; gint pt = -1;
gchar codec[32] = {0}; gchar codec[32] = {0};
...@@ -1425,20 +1759,25 @@ static void destroy_peer(RtcClient *client) ...@@ -1425,20 +1759,25 @@ static void destroy_peer(RtcClient *client)
int pc; int pc;
int track; int track;
int audio_track; int audio_track;
int data_channel;
g_mutex_lock(&client->lock); g_mutex_lock(&client->lock);
pc = client->pc; pc = client->pc;
track = client->track; track = client->track;
audio_track = client->audio_track; audio_track = client->audio_track;
data_channel = client->data_channel;
client->pc = -1; client->pc = -1;
client->track = -1; client->track = -1;
client->data_channel = -1;
client->track_open = FALSE; client->track_open = FALSE;
client->first_frame_sent = FALSE; client->first_frame_sent = FALSE;
client->last_idr_sent_us = 0; client->last_idr_sent_us = 0;
client->audio_track = -1; client->audio_track = -1;
client->audio_payload_type = -1;
client->audio_track_open = FALSE; client->audio_track_open = FALSE;
client->audio_send_enabled = FALSE; client->audio_send_enabled = FALSE;
client->audio_recv_enabled = FALSE; if (client->dc_audio_buf)
g_byte_array_set_size(client->dc_audio_buf, 0);
client->answer_sent = FALSE; client->answer_sent = FALSE;
client->need_idr = TRUE; client->need_idr = TRUE;
client->source_pts_valid = FALSE; client->source_pts_valid = FALSE;
...@@ -1458,6 +1797,10 @@ static void destroy_peer(RtcClient *client) ...@@ -1458,6 +1797,10 @@ static void destroy_peer(RtcClient *client)
rtcClose(audio_track); rtcClose(audio_track);
rtcDeleteTrack(audio_track); rtcDeleteTrack(audio_track);
} }
if (data_channel >= 0) {
rtcClose(data_channel);
rtcDeleteDataChannel(data_channel);
}
if (pc >= 0) { if (pc >= 0) {
rtcClosePeerConnection(pc); rtcClosePeerConnection(pc);
rtcDeletePeerConnection(pc); rtcDeletePeerConnection(pc);
...@@ -1535,34 +1878,22 @@ static gboolean setup_audio_track(RtcClient *client, int pc, ...@@ -1535,34 +1878,22 @@ static gboolean setup_audio_track(RtcClient *client, int pc,
rtcTrackInit track_init; rtcTrackInit track_init;
rtcPacketizerInit packetizer; rtcPacketizerInit packetizer;
guint32 ssrc = g_random_int(); guint32 ssrc = g_random_int();
gboolean offer_send;
gboolean offer_recv;
gboolean send_audio; gboolean send_audio;
gboolean recv_audio;
int track; int track;
if (ssrc == 0) if (ssrc == 0)
ssrc = 1; ssrc = 1;
memset(&track_init, 0, sizeof(track_init)); memset(&track_init, 0, sizeof(track_init));
/* SDP 未声明 direction 时按 WebRTC 默认 sendrecv 处理。 */ /* 手机端 audio 为 recvonly（只收不发），设备固定 sendonly 推 Opus 给手机。
offer_send = (audio->direction == 0 || audio->direction == 1 || * 手机→设备的音频走 DataChannel 发 MP4，不再通过 RTP 接收。 */
audio->direction == 3); send_audio = client->audio_source != NULL;
offer_recv = (audio->direction == 0 || audio->direction == 2 || if (send_audio)
audio->direction == 3);
send_audio = offer_recv && client->audio_source;
recv_audio = offer_send && client->audio_sink;
if (send_audio && recv_audio)
track_init.direction = RTC_DIRECTION_SENDRECV;
else if (send_audio)
track_init.direction = RTC_DIRECTION_SENDONLY; track_init.direction = RTC_DIRECTION_SENDONLY;
else if (recv_audio)
track_init.direction = RTC_DIRECTION_RECVONLY;
else else
track_init.direction = RTC_DIRECTION_INACTIVE; track_init.direction = RTC_DIRECTION_INACTIVE;
my_zlog_info("libdatachannel: audio track direction: offer=%d answer=%d", my_zlog_info("libdatachannel: audio track direction: answer=%d (sendonly)",
audio->direction, (int)track_init.direction); (int)track_init.direction);
track_init.codec = RTC_CODEC_OPUS; track_init.codec = RTC_CODEC_OPUS;
track_init.payloadType = audio->payload_type; track_init.payloadType = audio->payload_type;
track_init.ssrc = ssrc; track_init.ssrc = ssrc;
...@@ -1578,8 +1909,6 @@ static gboolean setup_audio_track(RtcClient *client, int pc, ...@@ -1578,8 +1909,6 @@ static gboolean setup_audio_track(RtcClient *client, int pc,
rtcSetOpenCallback(track, on_track_open); rtcSetOpenCallback(track, on_track_open);
rtcSetClosedCallback(track, on_track_closed); rtcSetClosedCallback(track, on_track_closed);
rtcSetErrorCallback(track, on_track_error); rtcSetErrorCallback(track, on_track_error);
if (recv_audio)
rtcSetMessageCallback(track, on_audio_message);
memset(&packetizer, 0, sizeof(packetizer)); memset(&packetizer, 0, sizeof(packetizer));
packetizer.ssrc = ssrc; packetizer.ssrc = ssrc;
...@@ -1597,13 +1926,12 @@ static gboolean setup_audio_track(RtcClient *client, int pc, ...@@ -1597,13 +1926,12 @@ static gboolean setup_audio_track(RtcClient *client, int pc,
return FALSE; return FALSE;
} }
} }
/* recv_audio 的 RTP payload 由 on_audio_message 送入 audio_sink。 */
g_mutex_lock(&client->lock); g_mutex_lock(&client->lock);
client->audio_rtp_timestamp = packetizer.timestamp; client->audio_rtp_timestamp = packetizer.timestamp;
client->audio_payload_type = audio->payload_type;
client->audio_pts_valid = FALSE; client->audio_pts_valid = FALSE;
client->audio_send_enabled = send_audio; client->audio_send_enabled = send_audio;
client->audio_recv_enabled = recv_audio;
g_mutex_unlock(&client->lock); g_mutex_unlock(&client->lock);
*track_out = track; *track_out = track;
return TRUE; return TRUE;
...@@ -1641,12 +1969,17 @@ int rtc_client_start(AppState *app) ...@@ -1641,12 +1969,17 @@ int rtc_client_start(AppState *app)
client->app = app; client->app = app;
client->pc = -1; client->pc = -1;
client->track = -1; client->track = -1;
client->audio_track = -1;
client->audio_payload_type = -1;
client->data_channel = -1;
client->target_bitrate = WEBRTCPUSH_INITIAL_BITRATE; client->target_bitrate = WEBRTCPUSH_INITIAL_BITRATE;
client->applied_mpp_bitrate = WEBRTCPUSH_INITIAL_BITRATE;
client->remb_ceiling = WEBRTCPUSH_INITIAL_BITRATE; client->remb_ceiling = WEBRTCPUSH_INITIAL_BITRATE;
client->last_remb_bitrate = 0; client->last_remb_bitrate = 0;
client->remb_filtered = 0; client->remb_filtered = 0;
client->remb_down_samples = 0; client->remb_down_samples = 0;
client->last_bitrate_ramp_us = g_get_monotonic_time(); client->last_bitrate_ramp_us = g_get_monotonic_time();
client->last_mpp_bitrate_apply_us = client->last_bitrate_ramp_us;
client->pending_local_ice = g_queue_new(); client->pending_local_ice = g_queue_new();
client->pending_remote_ice = g_queue_new(); client->pending_remote_ice = g_queue_new();
g_mutex_init(&client->lock); g_mutex_init(&client->lock);
...@@ -1746,6 +2079,25 @@ void rtc_client_stop(AppState *app) ...@@ -1746,6 +2079,25 @@ void rtc_client_stop(AppState *app)
clear_local_ice(client); clear_local_ice(client);
g_mutex_unlock(&client->lock); g_mutex_unlock(&client->lock);
g_queue_free(client->pending_local_ice); g_queue_free(client->pending_local_ice);
if (client->dc_audio_buf)
g_byte_array_unref(client->dc_audio_buf);
/* 停止播放队列线程 */
if (client->dc_play_queue) {
client->dc_play_quit = TRUE;
g_async_queue_push(client->dc_play_queue, GINT_TO_POINTER(-1)); /* 唤醒 */
if (client->dc_play_thread)
g_thread_join(client->dc_play_thread);
/* 清空队列剩余任务 */
while (1) {
gpointer p = g_async_queue_try_pop(client->dc_play_queue);
if (!p || p == GINT_TO_POINTER(-1)) break;
DcMp3PlayTask *t = p;
if (t) { g_free(t->data); g_free(t); }
}
g_async_queue_unref(client->dc_play_queue);
client->dc_play_queue = NULL;
client->dc_play_thread = NULL;
}
mpp_h264_source_stop(client->h264_source); mpp_h264_source_stop(client->h264_source);
frame_queue_fini(&client->frame_queue); frame_queue_fini(&client->frame_queue);
g_cond_clear(&client->track_cond); g_cond_clear(&client->track_cond);
...@@ -1764,6 +2116,7 @@ int rtc_client_handle_offer(AppState *app, const char *sdp) ...@@ -1764,6 +2116,7 @@ int rtc_client_handle_offer(AppState *app, const char *sdp)
int pc; int pc;
int track = -1; int track = -1;
int audio_track = -1; int audio_track = -1;
int data_channel = -1;
gboolean duplicate; gboolean duplicate;
if (!client || !sdp) if (!client || !sdp)
...@@ -1791,7 +2144,7 @@ int rtc_client_handle_offer(AppState *app, const char *sdp) ...@@ -1791,7 +2144,7 @@ int rtc_client_handle_offer(AppState *app, const char *sdp)
config.iceServers = s_ice_servers; config.iceServers = s_ice_servers;
config.iceServersCount = (int)G_N_ELEMENTS(s_ice_servers); config.iceServersCount = (int)G_N_ELEMENTS(s_ice_servers);
config.certificateType = RTC_CERTIFICATE_ECDSA; config.certificateType = RTC_CERTIFICATE_ECDSA;
config.forceMediaTransport = TRUE; /* 预分配 SRTP 媒体通道，与 ENABLE_DATACHANNEL=0 一致 */ config.forceMediaTransport = TRUE; /* 预分配 SRTP 媒体通道；DataChannel 使用同一个 PC 的 SCTP m-line */
config.mtu = 1200; config.mtu = 1200;
pc = rtcCreatePeerConnection(&config); pc = rtcCreatePeerConnection(&config);
if (pc < 0) { if (pc < 0) {
...@@ -1803,7 +2156,21 @@ int rtc_client_handle_offer(AppState *app, const char *sdp) ...@@ -1803,7 +2156,21 @@ int rtc_client_handle_offer(AppState *app, const char *sdp)
rtcSetLocalDescriptionCallback(pc, on_local_description); rtcSetLocalDescriptionCallback(pc, on_local_description);
rtcSetLocalCandidateCallback(pc, on_local_candidate); rtcSetLocalCandidateCallback(pc, on_local_candidate);
rtcSetStateChangeCallback(pc, on_state_change); rtcSetStateChangeCallback(pc, on_state_change);
rtcSetDataChannelCallback(pc, on_incoming_data_channel);
#if WEBRTCPUSH_ENABLE_DATACHANNEL
data_channel = rtcCreateDataChannel(pc, "myDataChannel");
if (data_channel >= 0) {
setup_data_channel_callbacks(client, data_channel);
my_zlog_info("libdatachannel: DataChannel created label=myDataChannel");
} else {
my_zlog_warn("libdatachannel: failed to create DataChannel");
}
#endif
if (!setup_track(client, pc, &video, &track)) { if (!setup_track(client, pc, &video, &track)) {
if (data_channel >= 0)
rtcDeleteDataChannel(data_channel);
rtcDeletePeerConnection(pc); rtcDeletePeerConnection(pc);
offer_video_clear(&video); offer_video_clear(&video);
offer_audio_clear(&audio); offer_audio_clear(&audio);
...@@ -1827,15 +2194,18 @@ int rtc_client_handle_offer(AppState *app, const char *sdp) ...@@ -1827,15 +2194,18 @@ int rtc_client_handle_offer(AppState *app, const char *sdp)
client->pc = pc; client->pc = pc;
client->track = track; client->track = track;
client->audio_track = audio_track; client->audio_track = audio_track;
client->data_channel = data_channel;
client->video_mline_index = video.mline_index; client->video_mline_index = video.mline_index;
client->video_mid = g_strdup(video.mid); client->video_mid = g_strdup(video.mid);
client->remote_ice_ufrag = g_strdup(video.ice_ufrag); client->remote_ice_ufrag = g_strdup(video.ice_ufrag);
client->target_bitrate = WEBRTCPUSH_INITIAL_BITRATE; client->target_bitrate = WEBRTCPUSH_INITIAL_BITRATE;
client->applied_mpp_bitrate = WEBRTCPUSH_INITIAL_BITRATE;
client->remb_ceiling = WEBRTCPUSH_INITIAL_BITRATE; client->remb_ceiling = WEBRTCPUSH_INITIAL_BITRATE;
client->last_remb_bitrate = 0; client->last_remb_bitrate = 0;
client->remb_filtered = 0; client->remb_filtered = 0;
client->remb_down_samples = 0; client->remb_down_samples = 0;
client->last_bitrate_ramp_us = g_get_monotonic_time(); client->last_bitrate_ramp_us = g_get_monotonic_time();
client->last_mpp_bitrate_apply_us = client->last_bitrate_ramp_us;
client->need_idr = TRUE; /* 内联，避免 request_idr 二次加锁死锁 */ client->need_idr = TRUE; /* 内联，避免 request_idr 二次加锁死锁 */
client->last_idr_request_us = g_get_monotonic_time(); client->last_idr_request_us = g_get_monotonic_time();
client->source_pts_valid = FALSE; client->source_pts_valid = FALSE;
......
modules/webrtcpush/webrtcpush_config.h
View file @ 3585cc8c
...@@ -18,36 +18,45 @@ ...@@ -18,36 +18,45 @@
/* /*
* 与 gst_webrtc_pipeline / jywy 浏览器推流对齐的码率策略。 * 与 gst_webrtc_pipeline / jywy 浏览器推流对齐的码率策略。
* 首屏先用 900kbps，不像 1.4Mbps 那样猛冲，也不要低到一进来就糊。 * 首屏先用 900kbps，不像 1.4Mbps 那样猛冲，也不要低到一进来就糊。
* RTCP REMB 只作为码率趋势，下降也做阶梯平滑，避免 MPP 动态切码率时卡顿。 * RTCP REMB 只作为码率趋势：小步慢降、慢升、带滞回，尽量接近浏览器的无感自适应。
*/ */
#define WEBRTCPUSH_INITIAL_BITRATE 900000U #define WEBRTCPUSH_INITIAL_BITRATE 900000U
#define WEBRTCPUSH_MIN_BITRATE 500000U #define WEBRTCPUSH_MIN_BITRATE 800000U
#define WEBRTCPUSH_MAX_BITRATE 2800000U #define WEBRTCPUSH_MAX_BITRATE 2600000U
#define WEBRTCPUSH_REMB_UTIL_PERCENT 80U #define WEBRTCPUSH_REMB_UTIL_PERCENT 80U
#define WEBRTCPUSH_REMB_DOWN_MIN_STEP 100000U #define WEBRTCPUSH_REMB_DOWN_MIN_STEP 100000U
#define WEBRTCPUSH_REMB_DOWN_CONFIRMATIONS 4U #define WEBRTCPUSH_REMB_DOWN_CONFIRMATIONS 8U
#define WEBRTCPUSH_REMB_SEVERE_CONFIRMATIONS 2U
#define WEBRTCPUSH_REMB_SEVERE_PERCENT 65U #define WEBRTCPUSH_REMB_SEVERE_PERCENT 65U
#define WEBRTCPUSH_BITRATE_RAMP_UP_MS 2500U #define WEBRTCPUSH_BITRATE_RAMP_UP_MS 1800U
#define WEBRTCPUSH_BITRATE_RAMP_DOWN_MS 1500U #define WEBRTCPUSH_BITRATE_RAMP_DOWN_MS 2000U
#define WEBRTCPUSH_REMB_DOWN_STEP_PERCENT 15U #define WEBRTCPUSH_REMB_DOWN_STEP_PERCENT 10U
#define WEBRTCPUSH_REMB_DOWN_STEP_MIN_BPS 80000U #define WEBRTCPUSH_REMB_DOWN_STEP_MIN_BPS 60000U
#define WEBRTCPUSH_REMB_DOWN_STEP_MAX_BPS 180000U #define WEBRTCPUSH_REMB_DOWN_STEP_MAX_BPS 120000U
#define WEBRTCPUSH_PACING_HEADROOM_PERCENT 140U /* REMB 低估保护：pacing 不堵时，过低浏览器估计不直接压糊 720p。 */
#define WEBRTCPUSH_REMB_SANE_FLOOR_BPS 2200000U
#define WEBRTCPUSH_REMB_SANE_RAW_MAX_BPS 1200000U
#define WEBRTCPUSH_REMB_SANE_PACING_MAX_MS 100U
#define WEBRTCPUSH_PACING_HEADROOM_PERCENT 180U
#define WEBRTCPUSH_PACING_INTERVAL_MS 5U #define WEBRTCPUSH_PACING_INTERVAL_MS 5U
#define WEBRTCPUSH_PACING_MAX_BITRATE 4000000U #define WEBRTCPUSH_PACING_MAX_BITRATE 4000000U
#define WEBRTCPUSH_PACING_MAX_QUEUE_MS 250U #define WEBRTCPUSH_PACING_MAX_QUEUE_MS 260U
#define WEBRTCPUSH_PACING_GUARD_COOLDOWN_MS 2000U #define WEBRTCPUSH_PACING_GUARD_COOLDOWN_MS 1000U
/* 首个/刚恢复的 IDR 允许短暂排队，避免首屏关键帧刚发出就被清队列 */ /* 首个/刚恢复的 IDR 允许短暂排队，避免首屏关键帧刚发出就被清队列 */
#define WEBRTCPUSH_PACING_IDR_GRACE_MS 800U #define WEBRTCPUSH_PACING_IDR_GRACE_MS 250U
/* pacing 清队列后，若近期已有 IDR，不要反复强制 IDR 造成 I 帧风暴 */ /* pacing 清队列后，若近期已有 IDR，不要反复强制 IDR 造成 I 帧风暴 */
#define WEBRTCPUSH_PACING_RESYNC_IDR_MS 5000U #define WEBRTCPUSH_PACING_RESYNC_IDR_MS 5000U
/* RK MPP 运行中小幅改 bps 容易顿一下；小变化只调 pacing，少重配硬编。 */
#define WEBRTCPUSH_MPP_RECONFIG_MIN_DELTA_BPS 200000U
#define WEBRTCPUSH_MPP_RECONFIG_MIN_INTERVAL_MS 5000U
/* RTP 分片与 NACK（MTU=1200，留 SRTP/DTLS/FU 余量） */ /* RTP 分片与 NACK（MTU=1200，留 SRTP/DTLS/FU 余量） */
#define WEBRTCPUSH_RTP_MAX_FRAGMENT 1050U #define WEBRTCPUSH_RTP_MAX_FRAGMENT 1050U
#define WEBRTCPUSH_NACK_PACKETS 512U #define WEBRTCPUSH_NACK_PACKETS 512U
/* PLI/IDR 节流：避免频繁 force-key-unit 拉高瞬时码率 */ /* PLI/IDR 节流：避免频繁 force-key-unit 拉高瞬时码率 */
#define WEBRTCPUSH_PLI_IDR_THROTTLE_MS 500U #define WEBRTCPUSH_PLI_IDR_THROTTLE_MS 2000U /* 500->2000: 避免IDR风暴, 每个IDR都加大pacing堆积 */
/* 周期性推流指标日志间隔 */ /* 周期性推流指标日志间隔 */
#define WEBRTCPUSH_STATS_LOG_INTERVAL_MS 8000U #define WEBRTCPUSH_STATS_LOG_INTERVAL_MS 8000U
...@@ -66,8 +75,7 @@ ...@@ -66,8 +75,7 @@
#define WEBRTCPUSH_MPP_POST_ENC_BUFFERS 1 /* 编码后保留 AU */ #define WEBRTCPUSH_MPP_POST_ENC_BUFFERS 1 /* 编码后保留 AU */
#define WEBRTCPUSH_APPSINK_MAX_BUFFERS 1 #define WEBRTCPUSH_APPSINK_MAX_BUFFERS 1
/* MJPEG 解压：1=GStreamer CPU jpegdec 优先；0=mppjpegdec 优先 */ /* native MJPEG 解压固定使用 libjpeg-turbo(jpeglib) 手动解码；旧 GStreamer 解码分支已删除 */
#define WEBRTCPUSH_MJPEG_CPU_DECODE 1
/* MPP 码控收紧：低 REMB 时实际 H264 不能长期高于目标太多 */ /* MPP 码控收紧：低 REMB 时实际 H264 不能长期高于目标太多 */
#define WEBRTCPUSH_MPP_BPS_MIN_PERCENT 60U #define WEBRTCPUSH_MPP_BPS_MIN_PERCENT 60U
...@@ -96,6 +104,15 @@ ...@@ -96,6 +104,15 @@
/* 音频播放（手机->设备）：ALSA 喇叭设备 */ /* 音频播放（手机->设备）：ALSA 喇叭设备 */
#define WEBRTCPUSH_AUDIO_PLAYBACK_DEVICE "plughw:2,0" #define WEBRTCPUSH_AUDIO_PLAYBACK_DEVICE "plughw:2,0"
/*
* 手机->设备喊话音量策略：
* 0：后端返回 0 就按 0 播放（静音）
* 1：后端返回 0 时按 WEBRTCPUSH_AUDIO_ZERO_VOLUME_MIN 播放，便于测试喊话链路
*/
#define WEBRTCPUSH_AUDIO_ZERO_VOLUME_AS_MIN 0
#define WEBRTCPUSH_AUDIO_ZERO_VOLUME_MIN 0.5
/* 手机喊话结束后短暂保温，避免每个短包都冷启动声卡；到时仍会释放喇叭。 */
#define WEBRTCPUSH_AUDIO_SINK_IDLE_TIMEOUT_MS 3000U
/* 后端音量控制接口 */ /* 后端音量控制接口 */
#define WEBRTCPUSH_VOLUME_API_BASE "https://fcrs-api.yd-ss.com/api/drive/use/status/" #define WEBRTCPUSH_VOLUME_API_BASE "https://fcrs-api.yd-ss.com/api/drive/use/status/"
...@@ -104,10 +121,11 @@ ...@@ -104,10 +121,11 @@
/* /*
* 设备侧 DataChannel（myDataChannel）与手机 createDataChannel('init') 争用 SCTP， * 设备侧 DataChannel（myDataChannel）：
* 易触发 sctpenc association error 并导致管道闪断/进程崩溃。仅推流可不建。 * - 前端 ondatachannel 后用 remoteChannel 发送 MP3 分片 + "EOF" 给设备播放；
* - 设备也通过该通道发送 Mbps 字符串，更新右上角网络显示。
*/ */
#define WEBRTCPUSH_ENABLE_DATACHANNEL 0 #define WEBRTCPUSH_ENABLE_DATACHANNEL 1
/* WebRTC 信令 WebSocket 主机（路径 /websocket?dev=设备号） */ /* WebRTC 信令 WebSocket 主机（路径 /websocket?dev=设备号） */
#define WEBRTCPUSH_SIGNAL_HOST "signal.yd-ss.com" #define WEBRTCPUSH_SIGNAL_HOST "signal.yd-ss.com"
......
zlog.conf
View file @ 3585cc8c
...@@ -9,4 +9,4 @@ file perms = 600 ...@@ -9,4 +9,4 @@ file perms = 600
millisecond = "%d(%Y-%m-%d %H:%M:%S).%ms [%V] %m%n" millisecond = "%d(%Y-%m-%d %H:%M:%S).%ms [%V] %m%n"
[rules] [rules]
my_log.* "/home/orangepi/car/master/log/log_2026-07-08.log"; millisecond my_log.* "/home/orangepi/car/master/log/log_2026-07-10.log"; millisecond
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment