From e70811725901d3f6589f15375ea0f1fb84ef9911 Mon Sep 17 00:00:00 2001 From: evgeny Date: Tue, 13 Oct 2026 16:09:13 +0200 Subject: [PATCH] Send relay state transitions without retry delays --- src/stream/stream.c | 14 +++++++++++--- src/stream/stream_network.c | 30 ++++++++++++++++++++---------- src/stream/stream_priv.h | 2 +- 3 files changed, 32 insertions(+), 14 deletions(-) diff --git a/src/stream/stream.c b/src/stream/stream.c index d139505d..43981bec 100644 --- a/src/stream/stream.c +++ b/src/stream/stream.c @@ -33,7 +33,7 @@ struct stream_session { uint8_t end[STREAM_END_SIZE]; int end_valid; unsigned owners, relay_owners, holds; - int anchored, closing, invalidated, finished, end_delivered, reason; + int anchored, decoded, closing, invalidated, finished, end_delivered, reason; int restarting, restart_gap, notify_close; }; @@ -296,6 +296,14 @@ static void session_deliver(void *arg) { int result = s->next_seq == UINT64_MAX ? STREAM_WAIT : stream_buffer_decode(s->buffer, s->next_seq, s->output, stream_input_size(&s->descriptor), &length); if (result == STREAM_OK) { + if (!s->decoded) { + s->decoded = 1; + DEBUG_INFO(DEBUG_CATEGORY_STREAM, "stream first batch reconstructed group=%llx source=%llx epoch=%llx seq=%llu k=%u", + (unsigned long long)s->descriptor.group_id, (unsigned long long)s->descriptor.source_node_id, + (unsigned long long)s->descriptor.source_epoch, (unsigned long long)s->next_seq, + (unsigned)(s->descriptor.n - s->descriptor.r)); + stream_network_wake(service->network); + } event.type = STREAM_EVENT_DATA; event.first_seq = event.last_seq = s->next_seq++; event.data = s->output; event.length = length; emit = 1; again = 1; } else if (result == STREAM_EXPIRED) { @@ -802,7 +810,7 @@ int stream_local_get(struct UTUN_INSTANCE *inst, const struct stream_descriptor struct stream_session *s = session_find(inst->stream, key->group_id, key->source_node_id); if (!s || s->descriptor.source_epoch != key->source_epoch) return STREAM_WAIT; *local = (struct stream_local){.descriptor = s->descriptor, .buffer = s->buffer, .start = s->start_seq, .last = s->last_seq, .next = s->next_seq, - .anchored = s->anchored, .publisher = s->source != NULL, .finished = s->end_valid, + .anchored = s->anchored, .decoded = s->decoded, .publisher = s->source != NULL, .finished = s->end_valid, .relay_only = !s->source && s->owners == s->relay_owners, .end = s->end_valid ? s->end : NULL}; return STREAM_OK; } @@ -863,7 +871,7 @@ int stream_local_restart(struct UTUN_INSTANCE *inst, const struct stream_descrip s->restarting = 1; } s->restart_gap = 0; - s->anchored = s->end_delivered = 0; s->passed_before = s->next_seq = s->start_seq = 0; + s->anchored = s->decoded = s->end_delivered = 0; s->passed_before = s->next_seq = s->start_seq = 0; DEBUG_INFO(DEBUG_CATEGORY_STREAM, "stream receiver restart from end group=%llx source=%llx epoch=%llx retained=%u floor=%llu known=%d", (unsigned long long)key->group_id, (unsigned long long)key->source_node_id, (unsigned long long)key->source_epoch, local.buffer != NULL, (unsigned long long)s->restart_from, s->restarting); return STREAM_OK; diff --git a/src/stream/stream_network.c b/src/stream/stream_network.c index a02b1bb6..d83a9d9d 100644 --- a/src/stream/stream_network.c +++ b/src/stream/stream_network.c @@ -30,9 +30,10 @@ /* Запросы не владеют буфером: один внутренний handle обслуживает всех получателей точной эпохи. */ struct stream_relay_request { struct stream_relay_request *next; + struct stream_network *network; struct stream_descriptor descriptor; uint64_t peer, id, epoch, server_epoch, deadline_ms, retain_ms, sent_ms, checked_ms, progress_ms, progress_seq; - int outgoing, status, reason, dirty, granted, blocked; + int outgoing, status, reason, dirty, urgent, granted, blocked; struct queue_waiter_handle waiter; uint8_t end[STREAM_END_SIZE]; }; @@ -185,6 +186,7 @@ static void send_ready(struct ll_queue *queue, void *arg) { static void relay_send_ready(struct ll_queue *queue, void *arg) { (void)queue; struct stream_relay_request *r = arg; r->blocked = 0; r->dirty = 1; r->sent_ms = 0; + stream_network_wake(r->network); } /* Сохранённый указатель очереди используется только пока waiter действительно зарегистрирован. */ @@ -1090,7 +1092,7 @@ void stream_network_cancel(struct stream_network *net, const struct stream_descr if (!net) return; for (struct stream_relay_request *r = net->relay_requests; r; r = r->next) if (same_stream(&r->descriptor, d) && r->status < STREAM_RELAY_REJECTED) { - r->status = STREAM_RELAY_CANCELLED; r->reason = STREAM_REASON_STOPPED; r->dirty = 1; + r->status = STREAM_RELAY_CANCELLED; r->reason = STREAM_REASON_STOPPED; r->dirty = r->urgent = 1; r->retain_ms = get_time_tb() / 10 + STREAM_RELAY_RETAIN_MS; } for (struct stream_peer *p = net->peers; p; p = p->next) { @@ -1099,6 +1101,7 @@ void stream_network_cancel(struct stream_network *net, const struct stream_descr if (same_stream(&f->descriptor, d)) flow_close(f, STREAM_REASON_STOPPED, 1); p->busy--; peer_sweep(p); } + stream_network_wake(net); } /* Исключение поставщика/получателя или self закрывает выдачи, сохраняя локальное кольцо остальных трансляций. */ @@ -1228,7 +1231,9 @@ static struct stream_relay_request *relay_request_new(struct stream_network *net struct stream_relay_request *r = u_calloc(1, sizeof(*r)); if (!r) { DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "relay request allocation failed"); return NULL; } r->descriptor = *key; r->peer = peer; r->id = id; r->epoch = epoch; r->outgoing = outgoing; - r->status = STREAM_RELAY_STARTING; r->dirty = 1; r->deadline_ms = get_time_tb() / 10 + STREAM_RELAY_START_MS; + r->network = net; + r->status = STREAM_RELAY_STARTING; r->dirty = r->urgent = 1; + r->deadline_ms = get_time_tb() / 10 + STREAM_RELAY_START_MS; r->next = net->relay_requests; net->relay_requests = r; net->relay_request_count++; DEBUG_INFO(DEBUG_CATEGORY_STREAM, "relay request opened group=%llx source=%llx epoch=%llx peer=%llx id=%llx out=%d deadline=%llu", (unsigned long long)key->group_id, (unsigned long long)key->source_node_id, (unsigned long long)key->source_epoch, @@ -1253,7 +1258,7 @@ static void relay_finish(struct stream_network *net, struct stream_relay_request if (r->status >= STREAM_RELAY_REJECTED) return; etcp_router_cancel_send_ready(net->inst, r->descriptor.group_id, r->peer, ETCP_RT_ID_STREAM, &r->waiter); memset(&r->waiter, 0, sizeof(r->waiter)); r->blocked = 0; - r->status = status; r->reason = reason; r->retain_ms = now + STREAM_RELAY_RETAIN_MS; r->dirty = 1; + r->status = status; r->reason = reason; r->retain_ms = now + STREAM_RELAY_RETAIN_MS; r->dirty = r->urgent = 1; if (r->outgoing) { struct stream_peer *p = peer_find(net, r->peer); if (p) { @@ -1360,7 +1365,11 @@ static void relay_dispatch(struct stream_network *net, uint64_t sender, const st if (stream_member_supernode(net->inst, m->descriptor.group_id, sender) != 1) { relay_finish(net, r, STREAM_RELAY_CANCELLED, STREAM_REASON_MEMBER, now); return; } - r->server_epoch = m->sender_epoch; + if (!r->server_epoch) { + r->server_epoch = m->sender_epoch; r->dirty = r->urgent = 1; + DEBUG_DEBUG(DEBUG_CATEGORY_STREAM, "relay server generation confirmed peer=%llx id=%llx generation=%llx", + (unsigned long long)sender, (unsigned long long)r->id, (unsigned long long)r->server_epoch); + } unsigned status = m->payload[0]; uint64_t remaining = stream_read_be(m->payload + 2, 4); if (status == STREAM_RELAY_READY || status == STREAM_RELAY_BOUND) { if (!remaining && status == STREAM_RELAY_READY) return; @@ -1415,7 +1424,7 @@ static void relay_dispatch(struct stream_network *net, uint64_t sender, const st if (!m->receiver_epoch) return; if (!peer_open(net, sender)) { relay_finish(net, r, STREAM_RELAY_REJECTED, STREAM_REASON_TRANSPORT, now); return; } int result = relay_open(net, &m->descriptor); - if (result == STREAM_OK) r->granted = 1; + if (result == STREAM_OK && !r->granted) { r->granted = r->urgent = 1; } if (result != STREAM_OK && result != STREAM_WAIT) relay_finish(net, r, STREAM_RELAY_REJECTED, result == STREAM_ERR_STATE ? STREAM_REASON_STOPPED : STREAM_REASON_LIMIT, now); } @@ -1436,7 +1445,7 @@ static void relay_tick(struct stream_network *net, uint64_t now) { unsigned active = relay_active(net, &r->descriptor, r->peer, r->outgoing); if (r->status < STREAM_RELAY_REJECTED && active) { if (r->status != STREAM_RELAY_BOUND) { - r->status = STREAM_RELAY_BOUND; r->dirty = 1; r->progress_ms = now; + r->status = STREAM_RELAY_BOUND; r->dirty = r->urgent = 1; r->progress_ms = now; DEBUG_INFO(DEBUG_CATEGORY_STREAM, "relay request bound to issuance source=%llx peer=%llx out=%d flows=%u", (unsigned long long)r->descriptor.source_node_id, (unsigned long long)r->peer, r->outgoing, active); } @@ -1463,14 +1472,15 @@ static void relay_tick(struct stream_network *net, uint64_t now) { for (unsigned sub = 0; sub < local.descriptor.n && !available; sub++) available = info_payload(&local, (uint8_t)sub, UINT64_MAX, r->peer, net->inst->node_id, info) == 0; if (available) { - r->status = STREAM_RELAY_READY; r->deadline_ms = now + STREAM_RELAY_HANDOFF_MS; r->dirty = 1; + r->status = STREAM_RELAY_READY; r->deadline_ms = now + STREAM_RELAY_HANDOFF_MS; + r->dirty = r->urgent = 1; } } } } } if (r->blocked || (!r->dirty && !(r->outgoing && r->status <= STREAM_RELAY_READY && now - r->sent_ms >= 1000))) continue; - if (r->sent_ms && now - r->sent_ms < 1000) continue; + if (!r->urgent && r->sent_ms && now - r->sent_ms < 1000) continue; struct stream_message m = {.descriptor = r->descriptor, .handle = r->id, .sender_epoch = r->outgoing ? r->epoch : net->relay_epoch, .receiver_epoch = r->outgoing ? r->server_epoch : r->epoch}; @@ -1482,7 +1492,7 @@ static void relay_tick(struct stream_network *net, uint64_t now) { stream_write_be(payload + 2, remaining, 4); length = 6; if (r->status == STREAM_RELAY_ENDED) { memcpy(payload + 6, r->end, STREAM_END_SIZE); length += STREAM_END_SIZE; } } - r->sent_ms = now; + r->sent_ms = now; r->urgent = 0; if (route_control(net, r->peer, m, payload, length, r) == STREAM_OK) r->dirty = 0; } struct stream_relay **link = &net->relays; diff --git a/src/stream/stream_priv.h b/src/stream/stream_priv.h index f77b38a1..f035962f 100644 --- a/src/stream/stream_priv.h +++ b/src/stream/stream_priv.h @@ -11,7 +11,7 @@ struct stream_local { struct stream_descriptor descriptor; struct stream_buffer *buffer; uint64_t start, last, next; - int anchored, publisher, finished, relay_only; + int anchored, decoded, publisher, finished, relay_only; const uint8_t *end; }; /* Заимствованные буфер/END действуют только до следующего callback/мутации сервиса. */