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

Add bounded call video transport and capability exchange

master
evgeny 5 days ago
parent
dffde9691d
commit
14a56d9b33
2 changed files with 269 additions and 0 deletions
Whitespace
Show all changes Ignore whitespace when comparing lines Ignore changes in amount of whitespace Ignore changes in whitespace at EOL
Split View
Diff Options
Show Stats Download Patch File Download Diff File
  1. 230
      src/call/call_video_transport.c
  2. 39
      src/call/call_video_transport.h

230
src/call/call_video_transport.c
Unescape View File

@ -0,0 +1,230 @@
#include "call_video_transport.h"
#include "call_priv.h"
#include "../utun_instance.h"
#include "../routing_layer/etcp_router.h"
#include "../routing_layer/route_crypto.h"
#include "../../lib/mem.h"
#include "../../lib/u_async.h"
#include "../../lib/debug_config.h"
#include <string.h>
enum { VIDEO_CAPS = 1, VIDEO_KEY_REQUEST = 2 };
struct call_video_transport {
struct call_session *session;
struct call_video_caps local, peer;
struct call_video_ops ops;
void *arg, *timer;
struct call_video_receiver *receiver;
struct call_video_frame tx;
uint8_t *data;
uint64_t tx_started_tb, tx_last_id, key_request_tb;
int64_t audio_end_pts, tx_last_pts;
int have_peer, need_key;
};
static struct call_session *active(struct UTUN_INSTANCE *inst, uint64_t id) {
struct call_session *s = inst ? call_session_find(inst->call, id) : NULL;
return s && s->state == CALL_ST_ACTIVE ? s : NULL;
}
static void tick(void *arg);
static struct call_video_transport *ensure(struct call_session *s) {
if (s->video) return s->video;
struct call_video_transport *v = u_calloc(1, sizeof(*v));
if (!v) { DEBUG_ERROR(DEBUG_CATEGORY_CALL, "video transport allocation failed"); return NULL; }
v->receiver = call_video_receiver_create();
if (!v->receiver) { u_free(v); return NULL; }
v->session = s; v->need_key = 1; v->audio_end_pts = v->tx_last_pts = -1;
v->timer = uasync_set_timeout(s->ctx->inst->ua, 500, v, tick, "call_video");
if (!v->timer) {
DEBUG_ERROR(DEBUG_CATEGORY_CALL, "video timer allocation failed");
call_video_receiver_destroy(v->receiver); u_free(v); return NULL;
}
s->video = v; return v;
}
static int control_send(struct call_video_transport *v, uint8_t type) {
struct call_session *s = v->session;
uint8_t data[9 + CALL_VIDEO_CAPS_BYTES]; data[0] = type; memcpy(data + 1, &s->call_id, 8);
size_t length = 9;
if (type == VIDEO_CAPS) {
if (call_video_caps_encode(&v->local, data + 9)) return -1;
length += CALL_VIDEO_CAPS_BYTES;
}
struct ETCP_ROUTER_CONN *conn = etcp_router_conn_get(s->ctx->inst, s->group_id, s->peer_node_id, ETCP_RT_ID_CALL_VIDEO_CONTROL);
if (!conn || etcp_router_conn_send(conn, data, length, ROUTE_CRYPTO_SIGN | ROUTE_CRYPTO_ENCRYPT)) {
DEBUG_WARN(DEBUG_CATEGORY_CALL, "video control send failed call=%llu type=%u", (unsigned long long)s->call_id, type); return -1;
}
return 0;
}
static void discard_tx(struct call_video_transport *v, const char *reason) {
DEBUG_WARN(DEBUG_CATEGORY_CALL, "video TX discard call=%llu frame=%llu offset=%u total=%u reason=%s",
(unsigned long long)v->session->call_id, (unsigned long long)v->tx.frame_id, v->tx.offset, v->tx.size, reason);
u_free(v->data); v->data = NULL; v->need_key = 1;
if (v->ops.need_key) v->ops.need_key(v->arg);
}
static void flush(struct call_video_transport *v) {
if (!v->data) return;
struct call_session *s = v->session;
if (get_time_tb() - v->tx_started_tb > 20000) { discard_tx(v, "deadline"); return; }
// Видео не обгоняет соответствующий аудиокадр. Отказ камеры не задерживает звук.
if (v->tx.pts_us > v->audio_end_pts) return;
struct ETCP_ROUTER_CONN *conn = etcp_router_conn_get(s->ctx->inst, s->group_id, s->peer_node_id, ETCP_RT_ID_CALL_VIDEO);
if (!conn) { discard_tx(v, "router_allocation"); return; }
uint8_t wire[CALL_VIDEO_HEADER_BYTES + CALL_VIDEO_FRAGMENT_BYTES];
// Ограниченный burst одного вызова; повторяется по audio send и таймеру 50мс.
for (unsigned budget = 0; v->data && budget < 16; budget++) {
if (!etcp_router_send_q_has_room(s->ctx->inst, s->group_id, s->peer_node_id, ETCP_RT_ID_CALL_VIDEO)) return;
uint32_t remaining = v->tx.size - v->tx.offset;
v->tx.length = remaining < CALL_VIDEO_FRAGMENT_BYTES ? remaining : CALL_VIDEO_FRAGMENT_BYTES;
v->tx.data = v->data + v->tx.offset;
size_t length;
if (call_video_fragment_encode(&v->tx, wire, sizeof(wire), &length)
|| etcp_router_conn_send(conn, wire, length, 0)) { discard_tx(v, "router_send"); return; }
v->tx.offset += v->tx.length;
if (v->tx.offset == v->tx.size) { u_free(v->data); v->data = NULL; }
}
}
static void tick(void *arg) {
struct call_video_transport *v = arg;
v->timer = NULL;
flush(v);
v->timer = uasync_set_timeout(v->session->ctx->inst->ua, 500, v, tick, "call_video");
if (!v->timer) DEBUG_ERROR(DEBUG_CATEGORY_CALL, "video timer rearm failed call=%llu", (unsigned long long)v->session->call_id);
}
void call_video_audio_sent(struct call_session *s, int64_t end_pts_us) {
if (!s->video || end_pts_us < 0) return;
s->video->audio_end_pts = end_pts_us;
flush(s->video);
}
void call_video_transport_stop(struct call_session *s) {
struct call_video_transport *v = s->video;
if (!v) return;
s->video = NULL;
if (v->timer) uasync_cancel_timeout(s->ctx->inst->ua, v->timer);
call_video_receiver_destroy(v->receiver); u_free(v->data); u_free(v);
// Уже принятые роутером пакеты содержат call_id и не принадлежат следующему звонку.
etcp_router_resume_receive(s->ctx->inst, s->group_id, s->peer_node_id, ETCP_RT_ID_CALL_VIDEO);
}
int call_video_configure(struct UTUN_INSTANCE *inst, uint64_t id, const struct call_video_caps *caps,
const struct call_video_ops *ops, void *arg) {
struct call_session *s = active(inst, id);
if (!s || !call_video_caps_valid(caps) || (ops && (!ops->frame || !ops->receive_ready))) {
DEBUG_ERROR(DEBUG_CATEGORY_CALL, "video configure invalid call/capabilities/callbacks id=%llu", (unsigned long long)id); return -1;
}
struct call_video_transport *v = ensure(s);
if (!v) return -1;
v->local = *caps; v->ops = ops ? *ops : (struct call_video_ops){0}; v->arg = arg;
if (!ops) {
memset(&v->local, 0, sizeof(v->local));
if (v->data) discard_tx(v, "platform_detached");
call_video_receiver_reset(v->receiver);
}
if (v->have_peer && v->ops.capabilities) v->ops.capabilities(v->arg, &v->peer);
call_video_resume_receive(inst, id);
DEBUG_INFO(DEBUG_CATEGORY_CALL, "video configured call=%llu encode=%u/%u decode=%u/%u", (unsigned long long)id,
v->local.encode[0].width, v->local.encode[1].width, v->local.decode[0].width, v->local.decode[1].width);
return control_send(v, VIDEO_CAPS);
}
int call_video_send_frame(struct UTUN_INSTANCE *inst, const struct call_video_frame *f) {
struct call_session *s = f ? active(inst, f->call_id) : NULL;
struct call_video_transport *v = s ? s->video : NULL;
if (!v || !call_video_format_valid(&f->format) || !f->data || !f->size || f->size > CALL_VIDEO_MAX_FRAME
|| f->offset || f->length != f->size || !f->frame_id || f->pts_us < 0 || f->pts_us > INT64_MAX / 2
|| (f->flags & ~CALL_VIDEO_KEY) || f->frame_id <= v->tx_last_id || f->pts_us <= v->tx_last_pts) {
DEBUG_WARN(DEBUG_CATEGORY_CALL, "video TX invalid frame/session"); return -1;
}
if (!v->have_peer || call_video_choose(&v->local, &v->peer, f->format.codec, f->format.width, f->format.height, f->format.fps)
!= f->format.codec) { DEBUG_WARN(DEBUG_CATEGORY_CALL, "video TX unsupported peer format"); return -1; }
if (v->data) { v->need_key = 1; return 0; }
int changed = !call_video_same_format(&f->format, &v->tx.format);
if (f->format.generation < v->tx.format.generation || (changed && f->format.generation == v->tx.format.generation)) {
DEBUG_ERROR(DEBUG_CATEGORY_CALL, "video TX format changed without new generation"); return -1;
}
if (changed || (v->tx_last_id && f->frame_id != v->tx_last_id + 1)) v->need_key = 1;
if (v->need_key && !(f->flags & CALL_VIDEO_KEY)) return 0;
v->data = u_malloc(f->size);
if (!v->data) { DEBUG_ERROR(DEBUG_CATEGORY_CALL, "video TX allocation failed bytes=%u", f->size); return -1; }
memcpy(v->data, f->data, f->size); v->tx = *f; v->tx.data = v->data;
v->tx_started_tb = get_time_tb(); v->tx_last_id = f->frame_id; v->tx_last_pts = f->pts_us; v->need_key = 0;
flush(v); return 1;
}
void call_video_resume_receive(struct UTUN_INSTANCE *inst, uint64_t id) {
struct call_session *s = active(inst, id);
if (s) etcp_router_resume_receive(inst, s->group_id, s->peer_node_id, ETCP_RT_ID_CALL_VIDEO);
}
void call_video_request_key(struct UTUN_INSTANCE *inst, uint64_t id) {
struct call_session *s = active(inst, id);
struct call_video_transport *v = s ? s->video : NULL;
if (!v) return;
uint64_t now = get_time_tb();
if (v->key_request_tb && now - v->key_request_tb < 5000) return;
if (!control_send(v, VIDEO_KEY_REQUEST)) v->key_request_tb = now;
}
static int ready(struct UTUN_INSTANCE *inst, uint64_t group, uint64_t peer, const uint8_t *data, size_t length) {
struct call_video_frame f;
if (call_video_fragment_parse(data, length, &f)) return 1;
struct call_session *s = active(inst, f.call_id);
struct call_video_transport *v = s ? s->video : NULL;
if (!v || s->group_id != group || s->peer_node_id != peer || !v->ops.frame || f.offset) return 1;
return v->ops.receive_ready(v->arg, f.size);
}
static void receive(struct ETCP_CONN *conn, struct ll_entry *entry) {
if (!entry || !entry->dgram || entry->len < ROUTER_SVC_PAYLOAD_OFF || !conn) {
DEBUG_WARN(DEBUG_CATEGORY_CALL, "video RX invalid router envelope");
if (entry) { queue_dgram_free(entry); queue_entry_free(entry); } return;
}
const uint8_t *data = entry->dgram + ROUTER_SVC_PAYLOAD_OFF;
size_t length = entry->len - ROUTER_SVC_PAYLOAD_OFF;
uint64_t group, peer, id = 0;
memcpy(&group, entry->dgram + ROUTER_SVC_GROUP_OFF, 8); memcpy(&peer, entry->dgram + ROUTER_SVC_SRC_OFF, 8);
int control = entry->dgram[0] == ETCP_RT_ID_CALL_VIDEO_CONTROL;
struct call_video_frame f;
if (control) {
if (length != 9 && length != 9 + CALL_VIDEO_CAPS_BYTES) goto invalid;
memcpy(&id, data + 1, 8);
} else {
if (call_video_fragment_parse(data, length, &f)) goto done;
id = f.call_id;
}
struct call_session *s = active(conn->instance, id);
if (!s) goto done; // Уже завершённый звонок не держит router receive_q.
if (s->group_id != group || s->peer_node_id != peer) goto invalid;
struct call_video_transport *v = ensure(s);
if (!v) goto done;
if (control) {
if (data[0] == VIDEO_CAPS && length == 9 + CALL_VIDEO_CAPS_BYTES) {
struct call_video_caps caps;
if (call_video_caps_parse(data + 9, CALL_VIDEO_CAPS_BYTES, &caps)) goto done;
v->peer = caps; v->have_peer = 1;
DEBUG_INFO(DEBUG_CATEGORY_CALL, "video peer capabilities call=%llu encode=%u/%u decode=%u/%u", (unsigned long long)id,
caps.encode[0].width, caps.encode[1].width, caps.decode[0].width, caps.decode[1].width);
if (v->ops.capabilities) v->ops.capabilities(v->arg, &v->peer);
} else if (data[0] == VIDEO_KEY_REQUEST && length == 9) {
v->need_key = 1;
if (v->ops.need_key) v->ops.need_key(v->arg);
} else goto invalid;
} else if (v->ops.frame) {
if (!v->have_peer || call_video_choose(&v->peer, &v->local, f.format.codec, f.format.width, f.format.height, f.format.fps)
!= f.format.codec) goto invalid;
struct call_video_frame complete;
int result = call_video_receiver_push(v->receiver, &f, get_time_us(), &complete);
if (result == 1) v->ops.frame(v->arg, &complete);
else if (call_video_receiver_needs_key(v->receiver)) call_video_request_key(conn->instance, id);
}
goto done;
invalid:
DEBUG_WARN(DEBUG_CATEGORY_CALL, "video RX invalid peer/group/control/format call=%llu peer=%llu bytes=%zu",
(unsigned long long)id, (unsigned long long)peer, length);
done:
queue_dgram_free(entry); queue_entry_free(entry);
}
int call_video_transport_bind(struct UTUN_INSTANCE *inst) {
if (etcp_router_bind(inst, ETCP_RT_ID_CALL_VIDEO_CONTROL, receive)) return -1;
if (etcp_router_bind(inst, ETCP_RT_ID_CALL_VIDEO, receive)) {
etcp_router_unbind(inst, ETCP_RT_ID_CALL_VIDEO_CONTROL); return -1;
}
etcp_router_set_receive_ready(inst, ETCP_RT_ID_CALL_VIDEO, ready); return 0;
}
void call_video_transport_unbind(struct UTUN_INSTANCE *inst) {
etcp_router_set_receive_ready(inst, ETCP_RT_ID_CALL_VIDEO, NULL);
etcp_router_unbind(inst, ETCP_RT_ID_CALL_VIDEO); etcp_router_unbind(inst, ETCP_RT_ID_CALL_VIDEO_CONTROL);
}

39
src/call/call_video_transport.h
Unescape View File

@ -0,0 +1,39 @@
/* Видеотранспорт звонка: uasync-поток, отдельные control/audio/video router channels.
* Callback не вызывает изменяющий API звонка реентерабельно; команды worker возвращает через post.
* Callback копирует нужные данные до возврата. При stop callbacks больше не вызываются. */
#ifndef CALL_VIDEO_TRANSPORT_H
#define CALL_VIDEO_TRANSPORT_H
#include "call_video.h"
#include "call_video_rate.h"
struct UTUN_INSTANCE;
struct call_session;
#ifdef __cplusplus
extern "C" {
#endif
struct call_video_ops {
void (*capabilities)(void *arg, const struct call_video_caps *peer);
int (*receive_ready)(void *arg, uint32_t bytes);
void (*frame)(void *arg, const struct call_video_frame *frame);
void (*need_key)(void *arg);
};
/* Настройка активного звонка, включая receive-only. NULL ops снимает платформенные callbacks.
* capabilities отражают только реально проверенные профили. 0 — успешно, -1 — ошибка. */
int call_video_configure(struct UTUN_INSTANCE *inst, uint64_t call_id, const struct call_video_caps *caps,
const struct call_video_ops *ops, void *arg);
/* Копирует один полный кадр. 1 принят, 0 backpressure/ожидание KEY, -1 неверный формат/сессия.
* После 0 producer обязан запросить KEY: пропуск P-кадра разрывает цепочку references.
* Неполная отправка ограничена 2с; следующие кадры не увеличивают очередь. */
int call_video_send_frame(struct UTUN_INSTANCE *inst, const struct call_video_frame *frame);
void call_video_resume_receive(struct UTUN_INSTANCE *inst, uint64_t call_id);
/* Decoder потерял references. Запрос идемпотентный, отправляется не чаще раза в 500мс. */
void call_video_request_key(struct UTUN_INSTANCE *inst, uint64_t call_id);
/* Только call.c: управление общей сессией и граница уже отправленного аудио. */
int call_video_transport_bind(struct UTUN_INSTANCE *inst);
void call_video_transport_unbind(struct UTUN_INSTANCE *inst);
void call_video_transport_stop(struct call_session *session);
void call_video_audio_sent(struct call_session *session, int64_t end_pts_us);
#ifdef __cplusplus
}
#endif
#endif
Write Preview
Loading…
Cancel
Save