jeka
/
utun2
1
0
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity
Browse Source

Activate supernode stream relays on demand and release idle owners

master
evgeny 7 days ago
parent
661cf032fe
commit
29d91d4b19
2 changed files with 406 additions and 23 deletions
Whitespace
Show all changes Ignore whitespace when comparing lines Ignore changes in amount of whitespace Ignore changes in whitespace at EOL
Split View
Diff Options
Show Stats Download Patch File Download Diff File
  1. 383
      src/stream/stream_network.c
  2. 46
      src/stream/stream_optimizer.c

383
src/stream/stream_network.c
Unescape View File

@ -7,6 +7,7 @@
#include "../transport_layer/etcp.h"
#include "../transport_layer/node_conn_direct.h"
#include "../routing_layer/route_crypto.h"
#include "../routing_layer/topo_node_sqlite.h"
#include "stream_member.h"
#include "stream_priv.h"
@ -19,6 +20,27 @@
#define STREAM_LEASE_TB 60000
#define STREAM_CONNECT_TB 50000
#define STREAM_TICK_TB 1000
#define STREAM_RELAY_LIMIT 8
#define STREAM_RELAY_REQUEST_LIMIT 128
#define STREAM_RELAY_START_MS 15000
#define STREAM_RELAY_HANDOFF_MS 5000
#define STREAM_RELAY_IDLE_MS 3000
#define STREAM_RELAY_RETAIN_MS 60000
/* Запросы не владеют буфером: один внутренний handle обслуживает всех получателей точной эпохи. */
struct stream_relay_request {
struct stream_relay_request *next;
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;
uint8_t end[STREAM_END_SIZE];
};
struct stream_relay {
struct stream_relay *next;
struct stream_descriptor descriptor;
struct STREAM_HANDLE *owner;
uint64_t idle_ms;
};
enum stream_peer_reset_flags { STREAM_RESET_ROTATE = 1, STREAM_RESET_KEEP_PENDING = 2 };
@ -71,6 +93,10 @@ struct stream_network {
int global_pressure;
uint64_t local_busy_until_ms, global_since_ms;
int global_candidate;
struct stream_relay_request *relay_requests;
struct stream_relay *relays;
uint64_t relay_epoch;
unsigned relay_count, relay_request_count;
};
static void peer_pump(void *arg);
@ -79,6 +105,8 @@ static void receive_raw(struct ETCP_CONN *conn, struct ll_entry *entry);
static void receive_routed(struct ETCP_CONN *conn, struct ll_entry *entry);
static void pressure_tick(struct stream_network *net, uint64_t now);
static void pressure_receive(struct stream_peer *peer, const struct stream_message *message);
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);
/* Ключи очереди — cursor+кольцо, её возраст учитывает реальное время вставки и момент появления выдачи. */
static uint64_t flow_age(struct STREAM_RECEIVE *f, uint64_t now) {
@ -169,22 +197,22 @@ static int control_put(struct stream_peer *p, struct stream_message m, const uin
}
/* Routed-управление обязательно подписано end-to-end, DATA этим путём не передаётся. */
static int route_control(struct stream_peer *p, struct stream_message m, const uint8_t *payload, size_t length) {
struct UTUN_INSTANCE *inst = p->network->inst;
if (!etcp_router_has_route(inst, m.descriptor.group_id, p->node_id)
|| !etcp_router_send_q_has_room(inst, m.descriptor.group_id, p->node_id, ETCP_RT_ID_STREAM)) {
static int route_control(struct stream_network *net, uint64_t peer, struct stream_message m, const uint8_t *payload, size_t length) {
struct UTUN_INSTANCE *inst = net->inst;
if (!etcp_router_has_route(inst, m.descriptor.group_id, peer)
|| !etcp_router_send_q_has_room(inst, m.descriptor.group_id, peer, ETCP_RT_ID_STREAM)) {
DEBUG_DEBUG(DEBUG_CATEGORY_STREAM, "stream routed control waits for route/room peer=%llx command=%u",
(unsigned long long)p->node_id, m.command); return STREAM_WAIT;
(unsigned long long)peer, m.command); return STREAM_WAIT;
}
struct ll_entry *e = ll_alloc_lldgram((uint16_t)(1 + STREAM_WIRE_HEADER + length));
if (!e) { DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "stream routed control allocation failed"); return STREAM_ERR_MEMORY; }
e->dgram[0] = ETCP_RT_ID_STREAM; m.sender_epoch = p->own_epoch; m.receiver_epoch = p->other_epoch;
e->dgram[0] = ETCP_RT_ID_STREAM;
stream_wire_header(e->dgram + 1, &m);
if (length) memcpy(e->dgram + 1 + STREAM_WIRE_HEADER, payload, length);
e->len = 1 + STREAM_WIRE_HEADER + length;
int result = etcp_route_send(inst, m.descriptor.group_id, p->node_id, e, 0, ROUTE_CRYPTO_SIGN | ROUTE_CRYPTO_ENCRYPT);
int result = etcp_route_send(inst, m.descriptor.group_id, peer, e, 0, ROUTE_CRYPTO_SIGN | ROUTE_CRYPTO_ENCRYPT);
if (result < 0) DEBUG_WARN(DEBUG_CATEGORY_STREAM, "stream routed control failed peer=%llx command=%u result=%d",
(unsigned long long)p->node_id, m.command, result);
(unsigned long long)peer, m.command, result);
return result < 0 ? STREAM_ERR_STATE : STREAM_OK;
}
@ -207,9 +235,10 @@ static void flow_close(struct STREAM_RECEIVE *f, int reason, int notify) {
if (notify && p && f->info.handle) {
uint8_t payload = (uint8_t)reason;
struct stream_message m = {.descriptor = f->descriptor, .handle = f->info.handle, .subblock = f->subblock,
.command = f->outgoing ? STREAM_CMD_UNSUBSCRIBE : STREAM_CMD_CLOSED};
.command = f->outgoing ? STREAM_CMD_UNSUBSCRIBE : STREAM_CMD_CLOSED,
.sender_epoch = p->own_epoch, .receiver_epoch = p->other_epoch};
if (p->up) control_put(p, m, &payload, 1);
else if (!f->outgoing) route_control(p, m, &payload, 1);
else if (!f->outgoing) route_control(p->network, p->node_id, m, &payload, 1);
}
f->info.state = STREAM_RECEIVE_CLOSED; f->info.reason = reason;
if (reason != STREAM_REASON_STOPPED)
@ -236,7 +265,10 @@ static void peer_sweep(struct stream_peer *p) {
p->network->flow_count--; f->peer = NULL; f->next = NULL;
if (!f->owned) u_free(f);
}
if (p->flows) p->idle = 0;
int relay_pending = 0;
for (struct stream_relay_request *r = p->network->relay_requests; r; r = r->next)
if (r->peer == p->node_id && r->status <= STREAM_RELAY_READY) relay_pending = 1;
if (p->flows || relay_pending) p->idle = 0;
else if (!p->idle) p->idle = get_time_tb();
}
@ -282,8 +314,9 @@ static void ncd_changed(struct NODE_CONN_DIRECT *handle, enum ncd_event event, v
if (!f->outgoing && f->info.handle) {
uint8_t reason = STREAM_REASON_TRANSPORT;
struct stream_message m = {.descriptor = f->descriptor, .handle = f->info.handle,
.subblock = f->subblock, .command = STREAM_CMD_CLOSED};
route_control(p, m, &reason, 1);
.subblock = f->subblock, .command = STREAM_CMD_CLOSED,
.sender_epoch = p->own_epoch, .receiver_epoch = p->other_epoch};
route_control(p->network, p->node_id, m, &reason, 1);
}
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);
@ -701,6 +734,12 @@ static void subscribe_request(struct stream_peer *p, struct stream_message m, st
static void dispatch(struct stream_network *net, uint64_t sender, const uint8_t *wire, size_t length, int routed) {
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);
else DEBUG_WARN(DEBUG_CATEGORY_STREAM, "relay command requires signed routed control peer=%llx command=%u",
(unsigned long long)sender, m.command);
return;
}
if (m.command == STREAM_CMD_OVERLOAD || m.command == STREAM_CMD_OVERLOAD_SELECT) {
struct stream_peer *p = peer_find(net, sender);
if (!routed && p && p->up && m.receiver_epoch == p->own_epoch && m.sender_epoch == p->other_epoch && !m.handle) {
@ -921,7 +960,7 @@ static void network_tick(void *arg) {
&& f->info.last_available != UINT64_MAX))
&& now >= f->deadline)
flow_close(f, f->outgoing ? STREAM_REASON_TRANSPORT : STREAM_REASON_LEASE, 1);
else if (f->outgoing && f->info.state >= STREAM_RECEIVE_ACTIVE && now - f->renewed >= STREAM_RENEW_TB) {
else if (f->outgoing && f->info.state == STREAM_RECEIVE_ACTIVE && now - f->renewed >= STREAM_RENEW_TB) {
struct stream_message m = {.descriptor = f->descriptor, .handle = f->info.handle,
.command = STREAM_CMD_RENEW, .subblock = f->subblock};
if (control_put(p, m, NULL, 0) == STREAM_OK) f->renewed = now;
@ -931,10 +970,9 @@ static void network_tick(void *arg) {
for (struct STREAM_RECEIVE *f = p->flows; f; f = f->next)
if (f->outgoing && f->info.state < STREAM_RECEIVE_ACTIVE && now < f->deadline) { pending = f; break; }
if (!p->up && pending && (!p->reverse_at || now - p->reverse_at >= 10000)) {
struct stream_message m = {.descriptor = pending->descriptor, .command = STREAM_CMD_REVERSE_REQ};
uint64_t saved = p->other_epoch; p->other_epoch = 0;
if (route_control(p, m, NULL, 0) == STREAM_OK) p->reverse_at = now;
p->other_epoch = saved;
struct stream_message m = {.descriptor = pending->descriptor, .command = STREAM_CMD_REVERSE_REQ,
.sender_epoch = p->own_epoch};
if (route_control(net, p->node_id, m, NULL, 0) == STREAM_OK) p->reverse_at = now;
}
p->busy--; peer_sweep(p);
if (p->idle && now - p->idle >= STREAM_LEASE_TB && !p->control->count) peer_free(p);
@ -942,6 +980,7 @@ static void network_tick(void *arg) {
}
pressure_tick(net, now_ms);
stream_optimizers_tick(net->inst, now_ms);
relay_tick(net, now_ms);
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");
@ -954,7 +993,8 @@ int stream_network_init(struct UTUN_INSTANCE *inst, struct stream_network **out)
}
struct stream_network *net = u_calloc(1, sizeof(*net));
if (!net) { DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "stream network allocation failed"); return STREAM_ERR_MEMORY; }
net->inst = inst;
net->inst = inst; net->relay_epoch = new_epoch();
if (!net->relay_epoch) { u_free(net); return STREAM_ERR_CRYPTO; }
stream_load_init(&net->load, get_time_tb() / 10); net->next_tick_ms = get_time_tb() / 10 + STREAM_TICK_TB / 10;
if (etcp_bind(inst, ETCP_ID_STREAM, receive_raw) < 0 || etcp_router_bind(inst, ETCP_RT_ID_STREAM, receive_routed) < 0) {
DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "stream bindings registration failed"); stream_network_destroy(net); return STREAM_ERR_STATE;
@ -976,6 +1016,13 @@ 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);
while (net->relays) {
struct stream_relay *relay = net->relays; net->relays = relay->next;
stream_close(relay->owner); u_free(relay);
}
while (net->relay_requests) {
struct stream_relay_request *r = net->relay_requests; net->relay_requests = r->next; u_free(r);
}
etcp_remove_conn_status_cbk(net->inst, transport_changed, net);
etcp_unbind(net->inst, ETCP_ID_STREAM); etcp_router_unbind(net->inst, ETCP_RT_ID_STREAM);
while (net->peers) peer_free(net->peers);
@ -989,6 +1036,11 @@ void stream_network_changed(struct stream_network *net) {
/* Кольцо освобождается только после отмены сетевых курсоров данной точной эпохи. */
void stream_network_cancel(struct stream_network *net, const struct stream_descriptor *d) {
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->retain_ms = get_time_tb() / 10 + STREAM_RELAY_RETAIN_MS;
}
for (struct stream_peer *p = net->peers; p; p = p->next) {
p->busy++;
for (struct STREAM_RECEIVE *f = p->flows; f; f = f->next)
@ -1026,7 +1078,8 @@ static int receive_open(struct STREAM_HANDLE *subscription, uint64_t node, uint8
struct stream_network *net = stream_network_get(inst);
struct stream_member_state member;
if (!net || !stream_member_online(inst, d.group_id, node) || stream_member_load(inst, d.group_id, node, &member, NULL) != STREAM_OK
|| !stream_member_has_subscription(&member, d.source_node_id, d.source_epoch)) {
|| (!stream_member_has_subscription(&member, d.source_node_id, d.source_epoch)
&& !stream_relay_confirmed(inst, &d, node))) {
DEBUG_WARN(DEBUG_CATEGORY_STREAM, "stream receive supplier unavailable peer=%llx", (unsigned long long)node); return STREAM_WAIT;
}
struct stream_peer *p = peer_open(net, node);
@ -1105,6 +1158,296 @@ int stream_receive_get_info(struct STREAM_RECEIVE *f, struct stream_receive_info
return STREAM_OK;
}
/* Терминальные записи остаются на минуту: повтор REQUEST/CANCEL не создаёт новое владение. */
static struct stream_relay_request *relay_request_find(struct stream_network *net, const struct stream_descriptor *key,
uint64_t peer, uint64_t id, int outgoing) {
for (struct stream_relay_request *r = net->relay_requests; r; r = r->next)
if (r->outgoing == outgoing && r->peer == peer && same_stream(&r->descriptor, key) && (!id || r->id == id)) return r;
return NULL;
}
/* Лимит включает tombstones и проверяется до выделения; исчерпание не вытесняет защиту от повторов. */
static struct stream_relay_request *relay_request_new(struct stream_network *net, const struct stream_descriptor *key,
uint64_t peer, uint64_t id, uint64_t epoch, int outgoing) {
if (net->relay_request_count >= STREAM_RELAY_REQUEST_LIMIT) {
DEBUG_WARN(DEBUG_CATEGORY_STREAM, "relay request limit peer=%llx count=%u", (unsigned long long)peer,
net->relay_request_count); return NULL;
}
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->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,
(unsigned long long)peer, (unsigned long long)id, outgoing, (unsigned long long)r->deadline_ms);
return r;
}
/* Наличие транспортных handles/INFO/ENDED не является действующей раздачей. */
static unsigned relay_active(struct stream_network *net, const struct stream_descriptor *key, uint64_t peer, int outgoing) {
unsigned count = 0;
for (struct stream_peer *p = net->peers; p; p = p->next) {
if (peer && p->node_id != peer) continue;
for (struct STREAM_RECEIVE *f = p->flows; f; f = f->next)
if (f->outgoing == outgoing && !f->info.probe_only && f->info.state == STREAM_RECEIVE_ACTIVE
&& same_stream(&f->descriptor, key)) count++;
}
return count;
}
/* Неуспешная помощь закрывает только выдачи своего пира и трансляции, не общий ETCP. */
static void relay_finish(struct stream_network *net, struct stream_relay_request *r, int status, int reason, uint64_t now) {
if (r->status >= STREAM_RELAY_REJECTED) return;
r->status = status; r->reason = reason; r->retain_ms = now + STREAM_RELAY_RETAIN_MS; r->dirty = 1;
if (r->outgoing) {
struct stream_peer *p = peer_find(net, r->peer);
if (p) {
p->busy++;
for (struct STREAM_RECEIVE *f = p->flows; f; f = f->next)
if (f->outgoing && same_stream(&f->descriptor, &r->descriptor)) flow_close(f, reason, 1);
p->busy--; peer_sweep(p);
}
}
DEBUG_INFO(DEBUG_CATEGORY_STREAM, "relay request finished source=%llx epoch=%llx peer=%llx id=%llx out=%d status=%d reason=%d",
(unsigned long long)r->descriptor.source_node_id, (unsigned long long)r->descriptor.source_epoch,
(unsigned long long)r->peer, (unsigned long long)r->id, r->outgoing, status, reason);
}
/* Одна подписка/кольцо на точную эпоху, независимо от числа ожидающих и действующих получателей. */
static int relay_open(struct stream_network *net, const struct stream_descriptor *key) {
for (struct stream_relay *relay = net->relays; relay; relay = relay->next)
if (same_stream(&relay->descriptor, key)) return stream_handle_active(relay->owner) ? STREAM_OK : STREAM_ERR_STATE;
if (net->relay_count >= STREAM_RELAY_LIMIT) {
DEBUG_WARN(DEBUG_CATEGORY_STREAM, "relay stream limit count=%u limit=%u", net->relay_count, STREAM_RELAY_LIMIT);
return STREAM_ERR_STATE;
}
struct stream_relay *relay = u_calloc(1, sizeof(*relay));
if (!relay) { DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "relay context allocation failed"); return STREAM_ERR_MEMORY; }
int result = stream_relay_open(net->inst, key, &relay->owner);
if (result != STREAM_OK) { u_free(relay); return result; }
relay->descriptor = *key; relay->next = net->relays; net->relays = relay; net->relay_count++;
DEBUG_INFO(DEBUG_CATEGORY_STREAM, "relay receiving started group=%llx source=%llx epoch=%llx streams=%u limit=%u",
(unsigned long long)key->group_id, (unsigned long long)key->source_node_id,
(unsigned long long)key->source_epoch, net->relay_count, STREAM_RELAY_LIMIT);
return STREAM_OK;
}
/* READY ускоряет поиск только в подтверждённом запросе; membership и DATA-подпись остаются обязательными. */
int stream_relay_confirmed(struct UTUN_INSTANCE *inst, const struct stream_descriptor *key, uint64_t peer) {
struct stream_network *net = stream_network_get(inst);
if (!net || net->stopping) return 0;
struct stream_relay_request *r = relay_request_find(net, key, peer, 0, 1);
return r && (r->status == STREAM_RELAY_BOUND
|| (r->status == STREAM_RELAY_READY && get_time_tb() / 10 < r->deadline_ms));
}
/* Автоматическая помощь выбирает один доступный канальный суперузел; неудачные попытки имеют cooldown. */
void stream_relay_demand(struct UTUN_INSTANCE *inst, const struct stream_descriptor *key, int needed, uint64_t now) {
struct stream_network *net = stream_network_get(inst); struct stream_local local;
if (!net || net->stopping || stream_local_get(inst, key, &local) != STREAM_OK) return;
for (struct stream_relay_request *r = net->relay_requests; r; r = r->next) {
if (!r->outgoing || !same_stream(&r->descriptor, key) || r->status >= STREAM_RELAY_REJECTED) continue;
unsigned active = relay_active(net, key, r->peer, 1);
int drained = local.finished && (local.last == UINT64_MAX || (local.anchored && local.next > local.last));
if (local.relay_only || drained || (!needed && !active)
|| (needed && r->status == STREAM_RELAY_BOUND && now - r->progress_ms >= STREAM_RELAY_START_MS))
relay_finish(net, r, STREAM_RELAY_CANCELLED, STREAM_REASON_STOPPED, now);
else return;
}
if (!needed || local.relay_only || local.finished) return;
struct TOPO_GROUP *group = topo_groups_find(inst->topo_groups, key->group_id);
struct topo_connect_peer *peers = NULL; size_t count = 0;
if (!group || topo_node_sqlite_get_connect_peers(inst->topo_sqlite_db, group->channel_id, &peers, &count) < 0) return;
uint64_t chosen = 0;
for (size_t i = 0; i < count; i++) {
uint64_t peer = peers[i].node_id;
if (!peers[i].supernode || peer == inst->node_id || peer == key->source_node_id
|| !stream_member_online(inst, key->group_id, peer) || !node_conn_direct_compatible(inst, peer)) continue;
struct stream_relay_request *previous = relay_request_find(net, key, peer, 0, 1);
if (previous) continue;
struct stream_member_state member;
if (stream_member_load(inst, key->group_id, peer, &member, NULL) != STREAM_OK) continue;
chosen = peer; break;
}
u_free(peers);
if (!chosen) return;
uint64_t id = new_epoch(), epoch = new_epoch();
if (!id || !epoch) return;
struct stream_relay_request *r = relay_request_new(net, key, chosen, id, epoch, 1);
if (r && !peer_open(net, chosen)) relay_finish(net, r, STREAM_RELAY_CANCELLED, STREAM_REASON_TRANSPORT, now);
}
/* Только end-to-end signed router-вход; команды до local_get авторизуются отдельно по канальному членству. */
static void relay_dispatch(struct stream_network *net, uint64_t sender, const struct stream_message *m) {
uint64_t now = get_time_tb() / 10;
if (!m->handle || m->subblock || (m->command != STREAM_CMD_RELAY_STATUS && m->length)) {
DEBUG_WARN(DEBUG_CATEGORY_STREAM, "relay malformed control peer=%llx command=%u bytes=%zu handle=%llx",
(unsigned long long)sender, m->command, m->length, (unsigned long long)m->handle); return;
}
struct stream_member_state member;
if (stream_member_load(net->inst, m->descriptor.group_id, sender, &member, NULL) != STREAM_OK) {
DEBUG_WARN(DEBUG_CATEGORY_STREAM, "relay control unauthorized peer=%llx group=%llx", (unsigned long long)sender,
(unsigned long long)m->descriptor.group_id); return;
}
if (m->command == STREAM_CMD_RELAY_STATUS) {
struct stream_relay_request *r = relay_request_find(net, &m->descriptor, sender, m->handle, 1);
if (!r || r->status >= STREAM_RELAY_REJECTED || m->receiver_epoch != r->epoch
|| (r->server_epoch && r->server_epoch != m->sender_epoch)) {
DEBUG_DEBUG(DEBUG_CATEGORY_STREAM, "relay stale status peer=%llx id=%llx", (unsigned long long)sender,
(unsigned long long)m->handle); return;
}
if (m->length < 6 || m->payload[0] < STREAM_RELAY_STARTING || m->payload[0] > STREAM_RELAY_CANCELLED
|| m->length != (size_t)(6 + (m->payload[0] == STREAM_RELAY_ENDED ? STREAM_END_SIZE : 0))) {
DEBUG_WARN(DEBUG_CATEGORY_STREAM, "relay invalid status peer=%llx bytes=%zu", (unsigned long long)sender, m->length); return;
}
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;
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;
if (remaining > STREAM_RELAY_HANDOFF_MS) remaining = STREAM_RELAY_HANDOFF_MS;
if (r->status == STREAM_RELAY_STARTING) {
r->status = STREAM_RELAY_READY; r->deadline_ms = now + remaining;
DEBUG_INFO(DEBUG_CATEGORY_STREAM, "relay ready source=%llx peer=%llx id=%llx handoff_ms=%llu",
(unsigned long long)r->descriptor.source_node_id, (unsigned long long)sender,
(unsigned long long)r->id, (unsigned long long)remaining);
stream_relay_ready(net->inst, &r->descriptor, sender);
}
} else if (status == STREAM_RELAY_ENDED) {
if (stream_read_be(m->payload + 6 + 34, 8) != UINT64_MAX
|| stream_ingest_end(net->inst, &r->descriptor, m->payload + 6, STREAM_END_SIZE) != STREAM_OK) {
DEBUG_WARN(DEBUG_CATEGORY_STREAM, "relay empty END verification failed peer=%llx", (unsigned long long)sender);
relay_finish(net, r, STREAM_RELAY_CANCELLED, STREAM_REASON_PROTOCOL, now);
} else relay_finish(net, r, STREAM_RELAY_CANCELLED, STREAM_REASON_STOPPED, now);
} else if (status >= STREAM_RELAY_REJECTED) {
relay_finish(net, r, STREAM_RELAY_CANCELLED, m->payload[1], now);
}
return;
}
if (m->receiver_epoch && m->receiver_epoch != net->relay_epoch) {
DEBUG_DEBUG(DEBUG_CATEGORY_STREAM, "relay stale server generation peer=%llx", (unsigned long long)sender); return;
}
struct stream_relay_request *r = relay_request_find(net, &m->descriptor, sender, m->handle, 0);
if (r && r->epoch != m->sender_epoch) {
DEBUG_WARN(DEBUG_CATEGORY_STREAM, "relay request nonce conflicts peer=%llx id=%llx", (unsigned long long)sender,
(unsigned long long)m->handle); return;
}
if (!r) r = relay_request_new(net, &m->descriptor, sender, m->handle, m->sender_epoch, 0);
if (!r) return;
if (m->command == STREAM_CMD_RELAY_CANCEL) relay_finish(net, r, STREAM_RELAY_CANCELLED, STREAM_REASON_STOPPED, now);
else {
r->dirty = 1;
if (r->status >= STREAM_RELAY_REJECTED) return;
if (!stream_member_online(net->inst, m->descriptor.group_id, sender)
|| stream_member_supernode(net->inst, m->descriptor.group_id, net->inst->node_id) != 1) {
relay_finish(net, r, STREAM_RELAY_REJECTED, STREAM_REASON_MEMBER, now); return;
}
if (net->global_pressure) { relay_finish(net, r, STREAM_RELAY_REJECTED, STREAM_REASON_OVERLOAD, now); return; }
for (size_t i = 0; i < member.count; i++)
if (member.subscriptions[i].source_node_id == m->descriptor.source_node_id
&& member.subscriptions[i].source_epoch == m->descriptor.source_epoch && member.subscriptions[i].relay_only) {
relay_finish(net, r, STREAM_RELAY_REJECTED, STREAM_REASON_LOOP, now); 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 && result != STREAM_WAIT)
relay_finish(net, r, STREAM_RELAY_REJECTED, result == STREAM_ERR_STATE ? STREAM_REASON_STOPPED : STREAM_REASON_LIMIT, now);
}
}
/* Один network tick обслуживает bootstrap, передачу владения выдачам, отмену и остановку пустых ретрансляторов. */
static void relay_tick(struct stream_network *net, uint64_t now) {
for (struct stream_relay_request *r = net->relay_requests; r; r = r->next) {
if (r->status < STREAM_RELAY_REJECTED) {
if (now >= r->checked_ms) {
r->checked_ms = now + 1000;
struct stream_member_state member;
int valid = stream_member_online(net->inst, r->descriptor.group_id, r->peer)
&& stream_member_load(net->inst, r->descriptor.group_id, r->peer, &member, NULL) == STREAM_OK
&& stream_member_supernode(net->inst, r->descriptor.group_id, r->outgoing ? r->peer : net->inst->node_id) == 1;
if (!valid) relay_finish(net, r, r->outgoing ? STREAM_RELAY_CANCELLED : STREAM_RELAY_REJECTED, STREAM_REASON_MEMBER, 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;
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);
}
struct stream_local local;
if (r->outgoing && stream_local_get(net->inst, &r->descriptor, &local) == STREAM_OK
&& r->progress_seq != local.next) { r->progress_seq = local.next; r->progress_ms = now; }
} else if (r->status == STREAM_RELAY_BOUND) {
relay_finish(net, r, STREAM_RELAY_CANCELLED, STREAM_REASON_STOPPED, now);
} else if (r->status < STREAM_RELAY_REJECTED && now >= r->deadline_ms) {
DEBUG_WARN(DEBUG_CATEGORY_STREAM, "relay attempt expired source=%llx peer=%llx id=%llx status=%d",
(unsigned long long)r->descriptor.source_node_id, (unsigned long long)r->peer,
(unsigned long long)r->id, r->status);
relay_finish(net, r, r->outgoing ? STREAM_RELAY_CANCELLED : STREAM_RELAY_REJECTED, STREAM_REASON_TRANSPORT, now);
}
if (!r->outgoing && r->status == STREAM_RELAY_STARTING) {
struct stream_local local;
if (stream_local_get(net->inst, &r->descriptor, &local) == STREAM_OK) {
if (local.finished && local.last == UINT64_MAX) {
memcpy(r->end, local.end, STREAM_END_SIZE);
relay_finish(net, r, STREAM_RELAY_ENDED, STREAM_REASON_STOPPED, now);
} else {
uint8_t info[STREAM_INFO_SIZE + STREAM_PATH_MAX * 8] = {0};
int available = 0;
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;
}
}
}
}
}
if (!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;
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};
uint8_t payload[6 + STREAM_END_SIZE]; size_t length = 0;
if (r->outgoing) m.command = r->status >= STREAM_RELAY_REJECTED ? STREAM_CMD_RELAY_CANCEL : STREAM_CMD_RELAY_REQUEST;
else {
m.command = STREAM_CMD_RELAY_STATUS; payload[0] = (uint8_t)r->status; payload[1] = (uint8_t)r->reason;
uint64_t remaining = r->deadline_ms > now ? r->deadline_ms - now : 0;
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;
if (route_control(net, r->peer, m, payload, length) == STREAM_OK) r->dirty = 0;
}
struct stream_relay **link = &net->relays;
while (*link) {
struct stream_relay *relay = *link;
unsigned pending = 0;
for (struct stream_relay_request *r = net->relay_requests; r; r = r->next)
if (!r->outgoing && same_stream(&r->descriptor, &relay->descriptor) && r->status <= STREAM_RELAY_READY) pending++;
unsigned active = relay_active(net, &relay->descriptor, 0, 0);
int valid = stream_handle_active(relay->owner)
&& stream_member_supernode(net->inst, relay->descriptor.group_id, net->inst->node_id) == 1;
if (valid && (pending || active)) { relay->idle_ms = 0; link = &relay->next; continue; }
if (valid && !relay->idle_ms) relay->idle_ms = now;
if (valid && now - relay->idle_ms < STREAM_RELAY_IDLE_MS) { link = &relay->next; continue; }
*link = relay->next; net->relay_count--;
DEBUG_INFO(DEBUG_CATEGORY_STREAM, "relay receiving stopped source=%llx epoch=%llx pending=%u downstream=%u valid=%d idle_ms=%llu",
(unsigned long long)relay->descriptor.source_node_id, (unsigned long long)relay->descriptor.source_epoch,
pending, active, valid, (unsigned long long)(now - relay->idle_ms));
stream_close(relay->owner); u_free(relay);
}
struct stream_relay_request **request_link = &net->relay_requests;
while (*request_link) {
struct stream_relay_request *r = *request_link;
if (r->status < STREAM_RELAY_REJECTED || now < r->retain_ms) { request_link = &r->next; continue; }
*request_link = r->next; net->relay_request_count--; u_free(r);
}
}
/* Добавление/вытеснение учитывает только ещё не переданные блоки действующей downstream-выдачи. */
void stream_network_cache_event(struct stream_network *net, const struct stream_descriptor *d, const uint8_t *wire, size_t length, int added) {
if (!net || net->stopping) return;

46
src/stream/stream_optimizer.c
Unescape View File

@ -29,7 +29,7 @@ struct stream_supplier {
struct STREAM_RECEIVE *probe;
struct stream_offer *offers;
unsigned probe_sub, failures;
int online, compatible, source;
int online, compatible, source, relay;
};
struct stream_selection {
struct STREAM_RECEIVE *current, *replacement;
@ -44,7 +44,7 @@ struct stream_optimizer {
struct stream_selection_settings settings;
struct stream_selection_info info;
struct stream_probe_schedule schedule;
uint64_t roster_ms, progress_ms, next_seq, diagnostic_ms;
uint64_t roster_ms, progress_ms, next_seq, diagnostic_ms, deficient_ms, relay_poll_ms;
int progress_valid, draining;
};
@ -78,6 +78,7 @@ static void roster_update(struct stream_optimizer *o, struct STREAM_HANDLE *h, u
s->node = candidates[i].node_id; s->next = o->suppliers; o->suppliers = s;
}
s->online = 1; s->source = candidates[i].source;
s->relay = stream_relay_confirmed(o->inst, &o->descriptor, s->node);
s->compatible = node_conn_direct_compatible(o->inst, s->node);
}
u_free(candidates);
@ -165,7 +166,8 @@ static void probes_schedule(struct stream_optimizer *o, struct STREAM_HANDLE *h,
struct stream_supplier *chosen = NULL;
for (struct stream_supplier *s = o->suppliers; s; s = s->next) {
if (!s->compatible || s->cycle == o->schedule.cycle || s->probe || now < s->retry_ms) continue;
if (!chosen || s->source > chosen->source || (s->source == chosen->source && s->node < chosen->node)) chosen = s;
if (!chosen || s->relay > chosen->relay || (s->relay == chosen->relay && (s->source > chosen->source
|| (s->source == chosen->source && s->node < chosen->node)))) chosen = s;
}
if (!chosen) break;
probe_start(o, h, chosen, now);
@ -299,12 +301,25 @@ int stream_optimizer_configure(struct stream_optimizer *o, const struct stream_s
void stream_optimizer_info(struct stream_optimizer *o, struct stream_selection_info *info) { *info = o->info; }
/* READY отменяет ожидание старого цикла/ошибки только для подтверждённого поставщика. */
void stream_optimizer_refresh(struct stream_optimizer *o, uint64_t peer) {
o->roster_ms = o->schedule.next_ms = 0;
struct stream_supplier *s = supplier_find(o, peer);
if (s) {
s->cycle = s->retry_ms = 0; s->failures = 0;
if (s->offers) for (unsigned sub = 0; sub < o->descriptor.n; sub++) s->offers[sub].retry_ms = 0;
}
DEBUG_DEBUG(DEBUG_CATEGORY_STREAM, "stream optimizer relay wake source=%llx peer=%llx",
(unsigned long long)o->descriptor.source_node_id, (unsigned long long)peer);
}
void stream_optimizer_tick(struct stream_optimizer *o, struct STREAM_HANDLE *h, uint64_t now) {
struct stream_local local;
if (stream_local_get(o->inst, &o->descriptor, &local) != STREAM_OK) return;
if (local.finished) {
/* END одного поставщика задаёт границу источника; остальные выдачи ещё могут требовать восстановления. */
if (local.last == UINT64_MAX || (local.anchored && local.next > local.last)) {
stream_relay_demand(o->inst, &o->descriptor, 0, now);
for (struct stream_supplier *s = o->suppliers; s; s = s->next) { stream_receive_close(s->probe); s->probe = NULL; }
if (o->draining) DEBUG_INFO(DEBUG_CATEGORY_STREAM, "stream optimizer END drained source=%llx next=%llu last=%llu",
(unsigned long long)o->descriptor.source_node_id, (unsigned long long)local.next, (unsigned long long)local.last);
@ -441,6 +456,31 @@ void stream_optimizer_tick(struct stream_optimizer *o, struct STREAM_HANDLE *h,
if (o->selected[sub].replacement) o->info.overlaps++;
}
if (o->info.subflows != o->descriptor.n) o->info.protected_by_fec = 0;
/* Помощь нужна для получения k субпотоков, а не для идеальной защиты n/r; пауза источника не является отказом пути. */
unsigned active = 0, viable = 0; int newer = 0;
for (unsigned sub = 0; sub < o->descriptor.n; sub++) {
struct STREAM_RECEIVE *current = o->selected[sub].current;
if (!current) continue;
struct stream_receive_info info; stream_receive_get_info(current, &info);
if (info.state != STREAM_RECEIVE_ACTIVE || info.close_requested) continue;
active++;
if ((info.blocks && info.metrics_valid && !info.metrics.stopped) || (!info.blocks && info.last_available == UINT64_MAX)) viable++;
}
for (struct stream_supplier *s = o->suppliers; s; s = s->next)
if (s->offers) for (unsigned sub = 0; sub < o->descriptor.n; sub++) {
struct stream_receive_info *offer = &s->offers[sub].info;
if (offer->info_ms && now - offer->info_ms < STREAM_OFFER_TTL_MS
&& offer->last_available != UINT64_MAX && offer->last_available > local.next) newer = 1;
}
unsigned required = (unsigned)(o->descriptor.n - o->descriptor.r);
int deficient = active < required || (viable < required && newer);
if (!deficient) o->deficient_ms = 0;
else if (!o->deficient_ms) o->deficient_ms = now;
if (now >= o->relay_poll_ms) {
o->relay_poll_ms = now + STREAM_PROBE_PERIOD_MS;
int needed = deficient && now - o->deficient_ms >= STREAM_MIGRATION_TIMEOUT_MS;
stream_relay_demand(o->inst, &o->descriptor, needed, now);
}
/* Пока нет первого блока, раз в секунду показываем причину ожидания каждого поставщика. */
if (!local.anchored && now >= o->diagnostic_ms) {
o->diagnostic_ms = now + STREAM_PROBE_PERIOD_MS;

Write Preview
Loading…
Cancel
Save