|
|
|
|
@ -12,6 +12,7 @@
|
|
|
|
|
#define TX_LIMIT (2u * 1024u * 1024u) |
|
|
|
|
#define TX_RECORDS 128 |
|
|
|
|
#define MEDIA_AGE_US 250000u |
|
|
|
|
#define PUBLISHER_DRAIN_US 2000000u |
|
|
|
|
|
|
|
|
|
struct publisher_record { |
|
|
|
|
struct publisher_record *next; |
|
|
|
|
@ -66,6 +67,15 @@ static int publisher_schedule(struct stream_publisher *p) {
|
|
|
|
|
|
|
|
|
|
/* Только uasync под lock; терминальный снимок остаётся доступен до следующего open. */ |
|
|
|
|
static void publisher_close(struct stream_publisher *p, int reason) { |
|
|
|
|
struct stream_info info; unsigned readers = 0; uint64_t pending = 0, drained = 0; |
|
|
|
|
if (p->handle && stream_handle_active(p->handle) && stream_get_info(p->handle, &info) == STREAM_OK) { |
|
|
|
|
readers = info.owners - 1; pending = stream_network_pending_bytes(p->inst, &info.descriptor); |
|
|
|
|
} |
|
|
|
|
if (p->status.state == STREAM_PUBLISHER_DRAINING) drained = get_time_us() - (p->drain_until - PUBLISHER_DRAIN_US); |
|
|
|
|
DEBUG_INFO(DEBUG_CATEGORY_STREAM, |
|
|
|
|
"publisher closing id=%llu epoch=%llx reason=%d state=%d drain_ms=%llu pending_RS=%llu readers=%u video=%zu audio=%zu", |
|
|
|
|
(unsigned long long)p->status.id, (unsigned long long)p->status.epoch, reason, p->status.state, |
|
|
|
|
(unsigned long long)(drained / 1000), (unsigned long long)pending, readers, p->video.bytes, p->audio.bytes); |
|
|
|
|
if (p->timer) uasync_cancel_timeout(p->inst->ua, p->timer); |
|
|
|
|
if (p->stats_timer) uasync_cancel_timeout(p->inst->ua, p->stats_timer); |
|
|
|
|
p->timer = p->stats_timer = NULL; |
|
|
|
|
@ -74,8 +84,6 @@ static void publisher_close(struct stream_publisher *p, int reason) {
|
|
|
|
|
p->status.state = reason ? STREAM_PUBLISHER_FAILED : STREAM_PUBLISHER_ENDED; p->status.reason = reason; |
|
|
|
|
p->status.video_bytes = p->status.audio_bytes = p->status.total_bytes = 0; |
|
|
|
|
p->status.stream_bytes = p->status.transport_bytes = p->status.inflight_bytes = p->status.acked_1s = 0; |
|
|
|
|
DEBUG_INFO(DEBUG_CATEGORY_STREAM, "publisher closed id=%llu epoch=%llx reason=%d", |
|
|
|
|
(unsigned long long)p->status.id, (unsigned long long)p->status.epoch, reason); |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
/* Один снимок всех соединений, без повторного учёта input_send_q внутри inflight. */ |
|
|
|
|
@ -193,7 +201,11 @@ static void publisher_process(void *arg) {
|
|
|
|
|
} |
|
|
|
|
if (p->handle && p->stopping && !p->video.head && !p->audio.head) { |
|
|
|
|
if (stream_publish_finish(p->handle) != STREAM_OK) publisher_close(p, STREAM_CLOSE_FAILED); |
|
|
|
|
else { p->status.state = STREAM_PUBLISHER_DRAINING; p->drain_until = now + 2000000; } |
|
|
|
|
else { |
|
|
|
|
p->status.state = STREAM_PUBLISHER_DRAINING; p->drain_until = now + PUBLISHER_DRAIN_US; |
|
|
|
|
DEBUG_INFO(DEBUG_CATEGORY_STREAM, "publisher FIN id=%llu epoch=%llx drain_ms=%u", |
|
|
|
|
(unsigned long long)p->status.id, (unsigned long long)p->status.epoch, PUBLISHER_DRAIN_US / 1000); |
|
|
|
|
} |
|
|
|
|
} |
|
|
|
|
} |
|
|
|
|
if (p->handle && p->status.state == STREAM_PUBLISHER_DRAINING && now >= p->drain_until) publisher_close(p, 0); |
|
|
|
|
|