diff --git a/src/stream/stream_media_buffer.c b/src/stream/stream_media_buffer.c index e15df015..99ba911b 100644 --- a/src/stream/stream_media_buffer.c +++ b/src/stream/stream_media_buffer.c @@ -159,7 +159,7 @@ int stream_media_store_feed(struct stream_media_store *s, uint64_t seq, const ui int result = stream_media_parse(data, length, &b); if (result != STREAM_OK) { s->failed = 1; return result; } if (s->configured && (b.format.audio_codec != s->format.audio_codec || b.format.revision < s->format.revision - || (b.format.revision == s->format.revision && (b.format.width != s->format.width + || (b.format.revision == s->format.revision && (b.format.video_codec != s->format.video_codec || b.format.width != s->format.width || b.format.height != s->format.height || b.format.rotation != s->format.rotation)))) { DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "media index conflicting format source=%llx revision=%u/%u audio=%u/%u", (unsigned long long)s->descriptor.source_node_id, s->format.revision, b.format.revision, diff --git a/src/stream/stream_network.c b/src/stream/stream_network.c index 7a030462..c86b1214 100644 --- a/src/stream/stream_network.c +++ b/src/stream/stream_network.c @@ -1806,6 +1806,15 @@ static void pressure_tick(struct stream_network *net, uint64_t now) { } } +uint64_t stream_network_pending_bytes(struct UTUN_INSTANCE *inst, const struct stream_descriptor *d) { + struct stream_network *net = stream_network_get(inst); uint64_t bytes = 0; + if (!net || !d) { DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "pending stream snapshot unavailable"); return 0; } + for (struct stream_peer *p = net->peers; p; p = p->next) + for (struct STREAM_RECEIVE *f = p->flows; f; f = f->next) + if (!f->outgoing && same_stream(&f->descriptor, d)) bytes += f->load.queued.bytes; + return bytes; +} + int stream_peer_get_load(struct UTUN_INSTANCE *inst, uint64_t node, struct stream_peer_load_info *info) { struct stream_network *net = stream_network_get(inst); struct stream_peer *p = net ? peer_find(net, node) : NULL; if (!p || !info) { DEBUG_WARN(DEBUG_CATEGORY_STREAM, "stream load unavailable peer=%llx", (unsigned long long)node); return STREAM_ERR_STATE; } @@ -1819,10 +1828,11 @@ int stream_peer_get_load(struct UTUN_INSTANCE *inst, uint64_t node, struct strea result.load.removed = p->load.removed; result.load.rejected = p->load.rejected; struct ETCP_CONN *conn = node_conn_direct_get_conn(p->ncd); if (conn) { - struct ll_queue *queues[] = {conn->send_input_q, conn->input_queue, conn->input_send_q}; - for (unsigned i = 0; i < 3; i++) if (queues[i]) { + struct ll_queue *queues[] = {conn->send_input_q, conn->input_queue}; + for (unsigned i = 0; i < 2; i++) if (queues[i]) { result.transport_queued_packets += queues[i]->count; result.transport_queued_bytes += queues[i]->total_bytes; } + result.transport_queued_bytes = etcp_tx_queued(conn); result.inflight_bytes = conn->unacked_bytes; } *info = result; return STREAM_OK; diff --git a/src/stream/stream_network.h b/src/stream/stream_network.h index 00528529..e0ea113d 100644 --- a/src/stream/stream_network.h +++ b/src/stream/stream_network.h @@ -55,6 +55,8 @@ struct stream_peer_load_info { int global_pressure, request_active, accepting; }; int stream_peer_get_load(struct UTUN_INSTANCE *inst, uint64_t peer, struct stream_peer_load_info *info); +/* Оставшиеся RS-части всех локальных выдач точной публикации; принятые ETCP части сюда не входят. */ +uint64_t stream_network_pending_bytes(struct UTUN_INSTANCE *inst, const struct stream_descriptor *descriptor); int stream_network_init(struct UTUN_INSTANCE *inst, struct stream_network **out); void stream_network_destroy(struct stream_network *network);