Browse Source

Enforce distinct manual and automatic upstream ownership

master
evgeny 2 days ago
parent
commit
6749d09025
  1. 7
      src/stream/stream.c
  2. 3
      src/stream/stream_network.h
  3. 3
      tests/test_stream_service.c

7
src/stream/stream.c

@ -38,6 +38,7 @@ struct STREAM_HANDLE {
struct stream_service *service;
struct stream_session *session;
int publisher;
int selection_owned;
int reason;
};
@ -634,8 +635,8 @@ int stream_local_get(struct UTUN_INSTANCE *inst, const struct stream_descriptor
/* Выданный локальный handle проверяется до создания сетевых ресурсов. */
int stream_handle_context(struct STREAM_HANDLE *h, struct UTUN_INSTANCE **inst, struct stream_descriptor *d) {
if (!inst || !d || !stream_handle_active(h) || h->session->source) {
DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "network receive requires active reader handle"); return STREAM_ERR_STATE;
if (!inst || !d || !stream_handle_active(h) || h->session->source || (h->session->optimizer && !h->selection_owned)) {
DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "network receive requires active manual reader or internal selector handle"); return STREAM_ERR_STATE;
}
*inst = h->service->inst; *d = h->session->descriptor; return STREAM_OK;
}
@ -696,7 +697,7 @@ void stream_optimizers_tick(struct UTUN_INSTANCE *inst, uint64_t now_ms) {
if (!service || service->stopping) return;
for (struct stream_session *s = service->sessions; s; s = s->next) {
if (!s->optimizer || s->closing || s->invalidated) continue;
struct STREAM_HANDLE subscription = {.service = service, .session = s};
struct STREAM_HANDLE subscription = {.service = service, .session = s, .selection_owned = 1};
stream_optimizer_tick(s->optimizer, &subscription, now_ms);
}
}

3
src/stream/stream_network.h

@ -1,7 +1,8 @@
/* Сетевые выдачи raw ETCP с собственным NCD-владением пира; reverse/терминальные уведомления через router.
* Одна управляющая FIFO и waiter на пира для всех групп/трансляций. DATA читается из кольца в момент отправки.
* STREAM_RECEIVE принадлежит вызывающему даже после отмены/stop; close снимает только данную выдачу.
* API выполняется в uasync-потоке. Автоматический выбор/перенос поставщиков — отдельный планировщик. */
* API выполняется в uasync-потоке. Внешние ручные операции требуют stream_subscribe_manual_open;
* автоматический выбор/перенос поставщиков владеет своими handles через отдельный планировщик. */
#ifndef STREAM_NETWORK_H
#define STREAM_NETWORK_H

3
tests/test_stream_service.c

@ -637,6 +637,9 @@ static void test_automatic(void) {
CHECK(stream_subscribe_open(nodes[1], GROUP2, nodes[3]->node_id, &subscribers[1]) == STREAM_OK);
CHECK(stream_subscribe_open(nodes[1], GROUP2, nodes[3]->node_id, &second) == STREAM_OK);
CHECK(stream_subscribe_manual_open(nodes[1], GROUP2, nodes[3]->node_id, &manual) == STREAM_ERR_STATE && !manual);
struct STREAM_RECEIVE *foreign = NULL;
CHECK(stream_receive_open(subscribers[1], nodes[3]->node_id, 0, UINT64_MAX, &foreign) == STREAM_ERR_ARGUMENT && !foreign);
CHECK(stream_receive_restart(subscribers[1]) == STREAM_ERR_STATE);
deadline = get_time_tb() + 150000;
int relay_announced = 0;
do {

Loading…
Cancel
Save