diff --git a/src/stream/stream.c b/src/stream/stream.c index b257a846..d139505d 100644 --- a/src/stream/stream.c +++ b/src/stream/stream.c @@ -557,6 +557,7 @@ static int session_open(struct UTUN_INSTANCE *inst, const struct stream_descript DEBUG_INFO(DEBUG_CATEGORY_STREAM, "local stream handle opened group=%llx source=%llx epoch=%llx publisher=%d owners=%u", (unsigned long long)d->group_id, (unsigned long long)d->source_node_id, (unsigned long long)d->source_epoch, publisher, s->owners); + if (created && s->optimizer) stream_network_wake(service->network); return STREAM_OK; } diff --git a/src/stream/stream_network.c b/src/stream/stream_network.c index 3634180f..a02b1bb6 100644 --- a/src/stream/stream_network.c +++ b/src/stream/stream_network.c @@ -86,7 +86,7 @@ struct stream_peer { struct stream_network { struct UTUN_INSTANCE *inst; struct stream_peer *peers; - void *timer; + void *timer, *work; unsigned peer_count, flow_count; int stopping; struct stream_load load; @@ -97,7 +97,7 @@ struct stream_network { int global_candidate; struct stream_relay_request *relay_requests; struct stream_relay *relays; - uint64_t relay_epoch; + uint64_t relay_epoch, info_order; unsigned relay_count, relay_request_count; }; @@ -110,6 +110,21 @@ static void pressure_receive(struct stream_peer *peer, const struct stream_messa static void relay_tick(struct stream_network *net, uint64_t now); static void relay_dispatch(struct stream_network *net, uint64_t sender, const struct stream_message *message); +/* Отложенное управление читает живые сессии; callbacks протокола уже закончили работу с handles. */ +static void network_work(void *arg) { + struct stream_network *net = arg; net->work = NULL; + uint64_t now = get_time_tb() / 10; + stream_optimizers_tick(net->inst, now); + relay_tick(net, now); +} + +/* Один call_soon объединяет INFO/ACCEPT/READY; DATA не запускает дополнительный обход на каждом блоке. */ +void stream_network_wake(struct stream_network *net) { + if (!net || net->stopping || net->work) return; + net->work = uasync_call_soon(net->inst->ua, net, network_work); + if (!net->work) DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "stream control scheduling failed"); +} + /* Ключи очереди — cursor+кольцо, её возраст учитывает реальное время вставки и момент появления выдачи. */ static uint64_t flow_age(struct STREAM_RECEIVE *f, uint64_t now) { if (f->outgoing || !f->load.queued.packets) return 0; @@ -348,6 +363,7 @@ static void ncd_changed(struct NODE_CONN_DIRECT *handle, enum ncd_event event, v peer_reset(p, STREAM_REASON_TRANSPORT, STREAM_RESET_ROTATE, NULL); } else DEBUG_WARN(DEBUG_CATEGORY_STREAM, "stream NCD initial timeout peer=%llx", (unsigned long long)p->node_id); DEBUG_DEBUG(DEBUG_CATEGORY_STREAM, "stream NCD event peer=%llx event=%d up=%d", (unsigned long long)p->node_id, event, p->up); + stream_network_wake(p->network); } /* На пира ровно один stream NCD owner, общий для всех его выдач и групп. */ @@ -762,7 +778,7 @@ static void dispatch(struct stream_network *net, uint64_t sender, const uint8_t struct stream_message m; struct stream_local local; if (stream_wire_parse(wire, length, &m) != STREAM_OK || !sender || sender == net->inst->node_id) return; if (m.command >= STREAM_CMD_RELAY_REQUEST) { - if (routed) relay_dispatch(net, sender, &m); + if (routed) { relay_dispatch(net, sender, &m); stream_network_wake(net); } else DEBUG_WARN(DEBUG_CATEGORY_STREAM, "relay command requires signed routed control peer=%llx command=%u", (unsigned long long)sender, m.command); return; @@ -836,6 +852,10 @@ static void dispatch(struct stream_network *net, uint64_t sender, const uint8_t if (p->other_epoch && p->other_epoch != m.sender_epoch) peer_reset(p, STREAM_REASON_TRANSPORT, STREAM_RESET_KEEP_PENDING, f); p->other_epoch = m.sender_epoch; f->info_ready = f->info.info_ready = 1; + if (net->info_order == UINT64_MAX) { + DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "stream INFO response order exhausted"); + flow_close(f, STREAM_REASON_LIMIT, 0); + } else f->info.info_order = ++net->info_order; f->info.local_epoch = p->own_epoch; f->info.peer_epoch = p->other_epoch; peer_schedule(p); uint64_t rtt = get_time_tb() / 10 - f->probe_sent_ms; f->info.rtt_valid = f->probe_sent_ms != 0; f->info.rtt_ms = rtt > UINT32_MAX ? UINT32_MAX : (uint32_t)rtt; @@ -910,6 +930,8 @@ static void dispatch(struct stream_network *net, uint64_t sender, const uint8_t (unsigned long long)sender, m.command, (unsigned long long)m.handle, m.subblock, m.length); receive_progress(net, &local.descriptor); p->busy--; peer_sweep(p); + if (m.command == STREAM_CMD_INFO_RESP || m.command == STREAM_CMD_ACCEPT || m.command == STREAM_CMD_REJECT + || m.command == STREAM_CMD_CLOSED || (m.command == STREAM_CMD_DATA && !local.anchored)) stream_network_wake(net); } /* Entry/dgram принадлежат обработчику во всех ветках. */ @@ -1006,8 +1028,7 @@ static void network_tick(void *arg) { p = next; } pressure_tick(net, now_ms); - stream_optimizers_tick(net->inst, now_ms); - relay_tick(net, now_ms); + stream_network_wake(net); net->next_tick_ms = get_time_tb() / 10 + STREAM_TICK_TB / 10; net->timer = uasync_set_timeout(net->inst->ua, STREAM_TICK_TB, net, network_tick, "stream_network"); if (!net->timer) DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "stream network timer scheduling failed"); @@ -1045,6 +1066,7 @@ int stream_network_init(struct UTUN_INSTANCE *inst, struct stream_network **out) void stream_network_destroy(struct stream_network *net) { if (!net) return; net->stopping = 1; if (net->timer) uasync_cancel_timeout(net->inst->ua, net->timer); + if (net->work) uasync_call_soon_cancel(net->inst->ua, net->work); while (net->relays) { struct stream_relay *relay = net->relays; net->relays = relay->next; stream_close(relay->owner); u_free(relay); diff --git a/src/stream/stream_network.h b/src/stream/stream_network.h index 755f3453..00528529 100644 --- a/src/stream/stream_network.h +++ b/src/stream/stream_network.h @@ -26,7 +26,7 @@ struct stream_receive_info { uint32_t rtt_ms; int metrics_valid, supplier_metrics_valid, rtt_valid, close_requested; struct stream_path path; - uint64_t first_available, last_available, info_ms, supplier_queue_bytes, progress_age_ms; + uint64_t first_available, last_available, info_ms, info_order, supplier_queue_bytes, progress_age_ms; uint32_t path_rtt_ms; uint16_t free_slots, cpu_basis_points; uint8_t subblock; @@ -59,6 +59,8 @@ int stream_peer_get_load(struct UTUN_INSTANCE *inst, uint64_t peer, struct strea int stream_network_init(struct UTUN_INSTANCE *inst, struct stream_network **out); void stream_network_destroy(struct stream_network *network); void stream_network_changed(struct stream_network *network); +/* Объединённый отложенный проход управления; не запускает selector внутри packet/NCD callbacks. */ +void stream_network_wake(struct stream_network *network); void stream_network_cancel(struct stream_network *network, const struct stream_descriptor *descriptor); void stream_network_membership_changed(struct stream_network *network, uint64_t node_id, const char *channel); void stream_network_cache_event(struct stream_network *network, const struct stream_descriptor *descriptor,