jeka
/
utun2
1
0
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity
You can not select more than 25 topics Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
2515 Commits
15 Branches
2 Tags
150 MiB
 
 
 
 
 
 
Tree: 109936f846
Branches Tags
${ item.name }
Create tag ${ searchTerm }
Create branch ${ searchTerm }
from '109936f846'
${ noResults }
utun2/src/stream/stream_network.c

739 lines
44 KiB
Raw Blame History

#include <string.h>
#include "../../lib/debug_config.h"
#include "../../lib/mem.h"
#include "../../lib/u_async.h"
#include "../utun_instance.h"
#include "../transport_layer/etcp.h"
#include "../transport_layer/node_conn_direct.h"
#include "../routing_layer/route_crypto.h"
#include "stream_member.h"
#include "stream_priv.h"
#define STREAM_PEER_LIMIT 64
#define STREAM_FLOW_LIMIT 256
#define STREAM_PEER_FLOW_LIMIT 64
#define STREAM_CONTROL_LIMIT 128
#define STREAM_CONTROL_MAX (STREAM_WIRE_HEADER + STREAM_INFO_SIZE + 1 + STREAM_PATH_MAX * 8)
#define STREAM_RENEW_TB 20000
#define STREAM_LEASE_TB 60000
#define STREAM_CONNECT_TB 50000
#define STREAM_TICK_TB 1000
struct stream_peer;
struct STREAM_RECEIVE {
struct STREAM_RECEIVE *next;
struct stream_peer *peer;
struct stream_descriptor descriptor;
struct stream_receive_info info;
uint64_t requested, deadline, renewed;
uint8_t subblock;
int outgoing, owned, bootstrap, info_sent, info_ready;
};
struct stream_peer {
struct stream_peer *next;
struct stream_network *network;
struct NODE_CONN_DIRECT *ncd;
struct STREAM_RECEIVE *flows, *last_sent;
struct ll_queue *control, *wait_q;
struct queue_waiter_handle waiter;
void *pump;
uint64_t node_id, own_epoch, other_epoch, next_handle, highwater, born, idle, reverse_at;
int up, busy;
};
struct stream_network {
struct UTUN_INSTANCE *inst;
struct stream_peer *peers;
void *timer;
unsigned peer_count, flow_count;
int stopping;
};
static void peer_pump(void *arg);
static void network_tick(void *arg);
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 int same_stream(const struct stream_descriptor *a, const struct stream_descriptor *b) {
return a->group_id == b->group_id && a->source_node_id == b->source_node_id && a->source_epoch == b->source_epoch;
}
/* Водяной знак и управляющая очередь принадлежат сессии пира, а не отдельной группе. */
static struct stream_peer *peer_find(struct stream_network *net, uint64_t node) {
for (struct stream_peer *p = net->peers; p; p = p->next) if (p->node_id == node) return p;
return NULL;
}
static uint64_t new_epoch(void) {
uint64_t value = 0;
do {
if (random_bytes((uint8_t *)&value, sizeof(value)) != 0) {
DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "stream peer epoch generation failed"); return 0;
}
} while (!value);
return value;
}
/* Waiter callback только планирует pump; рекурсивная отправка из ETCP-очереди исключена. */
static void peer_schedule(struct stream_peer *p) {
if (p->network->stopping || p->pump) return;
p->pump = uasync_call_soon(p->network->inst->ua, p, peer_pump);
if (!p->pump) DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "stream pump scheduling failed peer=%llx", (unsigned long long)p->node_id);
}
static void send_ready(struct ll_queue *queue, void *arg) {
(void)queue; struct stream_peer *p = arg; p->wait_q = NULL; peer_schedule(p);
}
/* Сохранённый указатель очереди используется только пока waiter действительно зарегистрирован. */
static void peer_cancel_wait(struct stream_peer *p) {
if (p->waiter.internal || p->waiter.call_soon_id)
queue_waiter_cancel(p->waiter.call_soon_id ? p->waiter.defer_q : p->wait_q, &p->waiter);
memset(&p->waiter, 0, sizeof(p->waiter)); p->wait_q = NULL;
}
/* Управление мало и ограничено; FIFO фиксирует порядок выделенных handles. В ней никогда нет DATA. */
static int control_put(struct stream_peer *p, struct stream_message m, const uint8_t *payload, size_t length) {
if (length + STREAM_WIRE_HEADER > STREAM_CONTROL_MAX || p->control->count >= STREAM_CONTROL_LIMIT) {
DEBUG_WARN(DEBUG_CATEGORY_STREAM, "stream control limit peer=%llx count=%d command=%u bytes=%zu",
(unsigned long long)p->node_id, p->control->count, m.command, length); return STREAM_ERR_STATE;
}
struct ll_entry *e = ll_alloc_lldgram((uint16_t)(STREAM_WIRE_HEADER + length));
if (!e) { DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "stream control allocation failed command=%u", m.command); return STREAM_ERR_MEMORY; }
m.sender_epoch = p->own_epoch; m.receiver_epoch = p->other_epoch;
stream_wire_header(e->dgram, &m);
if (length) memcpy(e->dgram + STREAM_WIRE_HEADER, payload, length);
e->len = STREAM_WIRE_HEADER + length;
if (queue_data_put(p->control, e) != 0) {
DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "stream control enqueue failed peer=%llx", (unsigned long long)p->node_id);
return STREAM_ERR_STATE;
}
peer_schedule(p); return STREAM_OK;
}
/* 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)) {
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;
}
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;
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);
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);
return result < 0 ? STREAM_ERR_STATE : STREAM_OK;
}
/* Закрытие не ждёт доставки уведомления. Память flow освобождается после выхода из pump/receiver. */
static void flow_close(struct STREAM_RECEIVE *f, int reason, int notify) {
if (f->info.state == STREAM_RECEIVE_CLOSED) return;
struct stream_peer *p = f->peer;
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};
if (p->up) control_put(p, m, &payload, 1);
else if (!f->outgoing) route_control(p, m, &payload, 1);
}
f->info.state = STREAM_RECEIVE_CLOSED; f->info.reason = reason;
DEBUG_INFO(DEBUG_CATEGORY_STREAM, "stream issuance closed peer=%llx group=%llx source=%llx handle=%llu sub=%u out=%d reason=%d blocks=%llu",
(unsigned long long)f->info.peer, (unsigned long long)f->descriptor.group_id,
(unsigned long long)f->descriptor.source_node_id, (unsigned long long)f->info.handle, f->subblock,
f->outgoing, reason, (unsigned long long)f->info.blocks);
}
/* Caller-owned receive после detach живёт до close; внутренние downstream принадлежат модулю. */
static void peer_sweep(struct stream_peer *p) {
if (p->busy) return;
struct STREAM_RECEIVE **link = &p->flows;
while (*link) {
struct STREAM_RECEIVE *f = *link;
if (f->info.state != STREAM_RECEIVE_CLOSED) { link = &f->next; continue; }
*link = f->next; if (p->last_sent == f) p->last_sent = NULL;
p->network->flow_count--; f->peer = NULL; f->next = NULL;
if (!f->owned) u_free(f);
}
if (p->flows) p->idle = 0;
else if (!p->idle) p->idle = get_time_tb();
}
/* Invalidation сохраняет транспорт чужих владельцев; старые DATA больше не имеют живого handle. */
static void peer_reset(struct stream_peer *p, int reason) {
peer_cancel_wait(p);
struct ll_entry *e;
while ((e = queue_data_get(p->control))) { queue_dgram_free(e); queue_entry_free(e); }
for (struct STREAM_RECEIVE *f = p->flows; f; f = f->next) flow_close(f, reason, 0);
p->other_epoch = p->highwater = 0; p->next_handle = 1; p->last_sent = NULL;
p->own_epoch = new_epoch(); peer_sweep(p);
DEBUG_INFO(DEBUG_CATEGORY_STREAM, "stream peer session reset peer=%llx epoch=%llx reason=%d",
(unsigned long long)p->node_id, (unsigned long long)p->own_epoch, reason);
}
/* NCD подтверждает UP собственного владельца, даже если ETCP уже использовался BGP/другим стримом. */
static void ncd_changed(struct NODE_CONN_DIRECT *handle, enum ncd_event event, void *arg) {
(void)handle; struct stream_peer *p = arg;
if (p->network->stopping) return;
if (event == NCD_EVENT_UP) { p->up = 1; peer_schedule(p); }
else if (event == NCD_EVENT_DOWN || event == NCD_EVENT_CLOSED) {
p->up = 0;
for (struct STREAM_RECEIVE *f = p->flows; f; f = f->next)
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);
}
peer_reset(p, STREAM_REASON_TRANSPORT);
} 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 NCD owner, общий для всех его выдач и групп. */
static struct stream_peer *peer_open(struct stream_network *net, uint64_t node) {
struct stream_peer *p = peer_find(net, node);
if (p) return p;
if (net->peer_count == STREAM_PEER_LIMIT) { DEBUG_WARN(DEBUG_CATEGORY_STREAM, "stream peer limit reached"); return NULL; }
p = u_calloc(1, sizeof(*p));
if (!p) { DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "stream peer allocation failed"); return NULL; }
p->network = net; p->node_id = node; p->own_epoch = new_epoch(); p->next_handle = 1; p->born = get_time_tb();
p->control = queue_new(net->inst->ua, 0, 0, 0, "stream_control");
if (!p->control || !p->own_epoch || node_conn_direct_open(net->inst, node, ncd_changed, p, &p->ncd, NULL) < 0) {
DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "stream peer NCD/init failed peer=%llx", (unsigned long long)node);
if (p->ncd) node_conn_direct_close(p->ncd);
if (p->control) queue_free(p->control); u_free(p); return NULL;
}
p->next = net->peers; net->peers = p; net->peer_count++;
DEBUG_INFO(DEBUG_CATEGORY_STREAM, "stream peer opened peer=%llx epoch=%llx", (unsigned long long)node, (unsigned long long)p->own_epoch);
return p;
}
/* Все callbacks/waiters отменяются до NCD close и освобождения контекста. */
static void peer_free(struct stream_peer *p) {
struct stream_network *net = p->network;
if (p->pump) uasync_call_soon_cancel(net->inst->ua, p->pump);
peer_cancel_wait(p);
for (struct STREAM_RECEIVE *f = p->flows; f; f = f->next) flow_close(f, STREAM_REASON_STOPPED, 0);
peer_sweep(p);
struct ll_entry *e;
while ((e = queue_data_get(p->control))) { queue_dgram_free(e); queue_entry_free(e); }
queue_free(p->control);
node_conn_direct_set_callback(p->ncd, NULL, NULL); node_conn_direct_close(p->ncd);
struct stream_peer **link = &net->peers;
while (*link != p) link = &(*link)->next;
*link = p->next; net->peer_count--;
DEBUG_INFO(DEBUG_CATEGORY_STREAM, "stream peer released peer=%llx", (unsigned long long)p->node_id); u_free(p);
}
static struct STREAM_RECEIVE *flow_find(struct stream_peer *p, const struct stream_message *m, int outgoing) {
for (struct STREAM_RECEIVE *f = p->flows; f; f = f->next)
if (f->outgoing == outgoing && f->info.state != STREAM_RECEIVE_CLOSED && same_stream(&f->descriptor, &m->descriptor)
&& f->subblock == m->subblock && (m->handle ? f->info.handle == m->handle : !f->info.handle)) return f;
return NULL;
}
/* Лимиты проверяются до выделения памяти, NCD не определяет право доступа. */
static struct STREAM_RECEIVE *flow_new(struct stream_peer *p, const struct stream_descriptor *d, uint8_t sub, int outgoing) {
unsigned count = 0; for (struct STREAM_RECEIVE *f = p->flows; f; f = f->next) count++;
if (p->network->flow_count == STREAM_FLOW_LIMIT || count == STREAM_PEER_FLOW_LIMIT) {
DEBUG_WARN(DEBUG_CATEGORY_STREAM, "stream issuance limit peer=%llx local=%u global=%u",
(unsigned long long)p->node_id, count, p->network->flow_count); return NULL;
}
struct STREAM_RECEIVE *f = u_calloc(1, sizeof(*f));
if (!f) { DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "stream issuance allocation failed"); return NULL; }
f->peer = p; f->descriptor = *d; f->subblock = sub; f->outgoing = outgoing;
f->info.peer = p->node_id; f->deadline = get_time_tb() + STREAM_CONNECT_TB;
f->next = p->flows; p->flows = f; p->network->flow_count++; p->idle = 0; return f;
}
/* Точная локальная эпоха проверяется до чтения подписанного профиля пира (unsubscribed => нет crypto). */
static int authorized(struct stream_network *net, uint64_t peer, const struct stream_message *m, struct stream_local *local) {
if (stream_local_get(net->inst, &m->descriptor, local) != STREAM_OK) return 0;
struct stream_member_state state;
int result = stream_member_load(net->inst, m->descriptor.group_id, peer, &state, NULL);
if (result != STREAM_OK) {
DEBUG_WARN(DEBUG_CATEGORY_STREAM, "stream peer authorization failed peer=%llx group=%llx result=%d",
(unsigned long long)peer, (unsigned long long)m->descriptor.group_id, result); return 0;
}
return 1;
}
/* INFO и ACCEPT описывают реальный последний/начальный блок с его историей, а не BGP-метрику. */
static int info_payload(struct stream_local *local, uint8_t sub, uint64_t seq, uint64_t peer, uint64_t self, uint8_t *payload) {
const uint8_t *wire; size_t length; uint64_t actual, first, last; struct stream_path path;
int result = stream_buffer_read(local->buffer, seq, sub, &wire, &length, &actual);
if (result == STREAM_EXPIRED) return STREAM_REASON_EXPIRED;
if (result != STREAM_OK
|| stream_buffer_range(local->buffer, sub, &first, &last) != STREAM_OK
|| stream_buffer_path(local->buffer, actual, sub, &path) != STREAM_OK) return STREAM_REASON_POSITION;
if (stream_path_validate(&path, local->descriptor.source_node_id, self, peer) != STREAM_OK || path.count == STREAM_PATH_MAX)
return STREAM_REASON_LOOP;
stream_write_be(payload, actual, 8); stream_write_be(payload + 8, first, 8); stream_write_be(payload + 16, last, 8);
payload[24] = local->descriptor.n; payload[25] = local->descriptor.r;
stream_write_be(payload + 26, local->descriptor.block_size, 2);
stream_path_write(payload + 28, &path); return 0;
}
static int parse_info(const struct stream_message *m, const struct stream_descriptor *d, uint64_t peer, uint64_t self, uint64_t *start) {
struct stream_path path; size_t used;
if (m->length < STREAM_INFO_SIZE || m->payload[24] != d->n || m->payload[25] != d->r
|| stream_read_be(m->payload + 26, 2) != d->block_size
|| stream_path_parse(m->payload + 28, m->length - 28, &path, &used) != STREAM_OK || m->length != 28 + used
|| path.count == STREAM_PATH_MAX || stream_path_validate(&path, d->source_node_id, peer, self) != STREAM_OK
|| stream_read_be(m->payload, 8) == UINT64_MAX || stream_read_be(m->payload, 8) > stream_read_be(m->payload + 16, 8)
|| stream_read_be(m->payload + 8, 8) > stream_read_be(m->payload + 16, 8)) {
DEBUG_WARN(DEBUG_CATEGORY_STREAM, "stream INFO/ACCEPT invalid command=%u bytes=%zu", m->command, m->length);
return STREAM_ERR_FORMAT;
}
*start = stream_read_be(m->payload, 8); return STREAM_OK;
}
/* Один пакет на свободный порог; собственный буфер переживает синхронное освобождение entry в etcp_send. */
static int raw_send(struct stream_peer *p, struct stream_message m, const uint8_t *payload, size_t length) {
struct ETCP_CONN *conn = node_conn_direct_get_conn(p->ncd);
struct ll_entry *e = ll_alloc_lldgram((uint16_t)(STREAM_WIRE_HEADER + length));
if (!e) { DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "stream raw allocation failed bytes=%zu", length); return STREAM_ERR_MEMORY; }
m.sender_epoch = p->own_epoch; m.receiver_epoch = p->other_epoch;
stream_wire_header(e->dgram, &m); if (length) memcpy(e->dgram + STREAM_WIRE_HEADER, payload, length);
e->len = STREAM_WIRE_HEADER + length;
if (etcp_send(conn, e) != 0) {
queue_dgram_free(e); queue_entry_free(e);
DEBUG_WARN(DEBUG_CATEGORY_STREAM, "stream raw send failed peer=%llx command=%u", (unsigned long long)p->node_id, m.command);
return STREAM_ERR_STATE;
}
DEBUG_TRACE(DEBUG_CATEGORY_STREAM, "stream raw sent peer=%llx command=%u handle=%llu bytes=%zu",
(unsigned long long)p->node_id, m.command, (unsigned long long)m.handle, length);
return STREAM_OK;
}
/* Производитель не хранит готовые DATA: cursor проверяется непосредственно перед копированием в ETCP. */
static int downstream_send(struct STREAM_RECEIVE *f) {
struct stream_peer *p = f->peer; struct stream_local local;
if (stream_local_get(p->network->inst, &f->descriptor, &local) != STREAM_OK) { flow_close(f, STREAM_REASON_STOPPED, 1); return 0; }
struct stream_message m = {.descriptor = f->descriptor, .handle = f->info.handle, .subblock = f->subblock};
const uint8_t *wire; size_t length; uint64_t actual;
int result = f->info.next == UINT64_MAX ? STREAM_WAIT
: stream_buffer_read(local.buffer, f->info.next, f->subblock, &wire, &length, &actual);
if (result == STREAM_OK) {
struct stream_path path;
if (stream_buffer_path(local.buffer, actual, f->subblock, &path) != STREAM_OK
|| stream_path_validate(&path, f->descriptor.source_node_id, p->network->inst->node_id, p->node_id) != STREAM_OK) {
flow_close(f, STREAM_REASON_LOOP, 1); return 0;
}
uint8_t payload[1 + STREAM_PATH_MAX * 8 + STREAM_BLOCK_MAX_SIZE];
size_t path_bytes = stream_path_write(payload, &path); memcpy(payload + path_bytes, wire, length);
m.command = STREAM_CMD_DATA;
if (raw_send(p, m, payload, path_bytes + length) != STREAM_OK) { flow_close(f, STREAM_REASON_TRANSPORT, 0); return 0; }
if (f->info.state == STREAM_RECEIVE_ACTIVE) { f->info.next = actual + 1; f->info.blocks++; }
return 1;
}
if (result == STREAM_EXPIRED || result == STREAM_GAP) {
uint64_t next;
int available = stream_buffer_next_seq(local.buffer, f->info.next, f->subblock, &next) == STREAM_OK;
if (!available && local.finished && local.last != UINT64_MAX && f->info.next <= local.last) {
next = local.last + 1; available = 1;
}
if (available && next > f->info.next) {
uint8_t payload[16]; stream_write_be(payload, f->info.next, 8); stream_write_be(payload + 8, next, 8);
m.command = STREAM_CMD_GAP;
if (raw_send(p, m, payload, sizeof(payload)) == STREAM_OK) {
f->info.next = next; f->info.gaps++; return 1;
}
flow_close(f, STREAM_REASON_TRANSPORT, 0); return 0;
}
}
if (local.finished && local.last != UINT64_MAX && f->info.next > local.last) {
m.command = STREAM_CMD_END;
if (raw_send(p, m, local.end, STREAM_END_SIZE) == STREAM_OK) {
f->info.state = STREAM_RECEIVE_ENDED; return 1;
}
flow_close(f, STREAM_REASON_TRANSPORT, 0);
}
return 0;
}
/* INFO предшествует выделению handle; остальные latest-запросы ждут подтверждённый DATA старт первой выдачи. */
static void prepare_requests(struct stream_peer *p) {
for (struct STREAM_RECEIVE *f = p->flows; f; f = f->next) {
if (!f->outgoing || f->info.state >= STREAM_RECEIVE_REQUESTED) continue;
struct stream_message m = {.descriptor = f->descriptor, .subblock = f->subblock};
if (!f->info_sent) {
m.command = STREAM_CMD_INFO_REQ;
if (control_put(p, m, NULL, 0) == STREAM_OK) { f->info_sent = 1; f->info.state = STREAM_RECEIVE_INFO; }
}
if (!f->info_ready || !p->other_epoch) continue;
uint64_t requested = f->requested;
struct stream_local local;
if (stream_local_get(p->network->inst, &f->descriptor, &local) != STREAM_OK) { flow_close(f, STREAM_REASON_STOPPED, 0); continue; }
if (requested == UINT64_MAX) {
if (local.anchored) requested = local.start;
else if (!f->bootstrap) continue;
}
if (p->next_handle == UINT64_MAX) { flow_close(f, STREAM_REASON_LIMIT, 0); continue; }
uint8_t payload[8]; stream_write_be(payload, requested, 8);
m.command = STREAM_CMD_SUBSCRIBE; m.handle = p->next_handle;
if (control_put(p, m, payload, sizeof(payload)) == STREAM_OK) {
f->info.handle = p->next_handle++; f->requested = requested; f->info.state = STREAM_RECEIVE_REQUESTED;
DEBUG_INFO(DEBUG_CATEGORY_STREAM, "stream SUBSCRIBE queued peer=%llx handle=%llu sub=%u index=%llu bootstrap=%d",
(unsigned long long)p->node_id, (unsigned long long)f->info.handle, f->subblock,
(unsigned long long)requested, f->bootstrap);
}
}
}
/* Управление FIFO прежде DATA, затем round-robin курсоров; один waiter на общий send_input_q. */
static void peer_pump(void *arg) {
struct stream_peer *p = arg; p->pump = NULL;
if (!p->up || !p->own_epoch || p->network->stopping) return;
struct ETCP_CONN *conn = node_conn_direct_get_conn(p->ncd);
if (!conn || !conn->send_input_q || conn->state == 2 || conn->close_requested || conn->reinit_pending || !conn->initialized) return;
if (conn->send_input_q->count) {
p->wait_q = conn->send_input_q;
if (queue_waiter_wait(p->wait_q, &p->waiter, send_ready, p) < 0)
DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "stream transport waiter failed peer=%llx", (unsigned long long)p->node_id);
return;
}
p->busy++; prepare_requests(p);
int sent = 0;
struct ll_entry *e = queue_data_get(p->control);
if (e) {
if (etcp_send(conn, e) != 0) {
queue_dgram_free(e); queue_entry_free(e);
DEBUG_WARN(DEBUG_CATEGORY_STREAM, "stream control send failed peer=%llx", (unsigned long long)p->node_id);
peer_reset(p, STREAM_REASON_TRANSPORT);
} else sent = 1;
queue_resume_callback(p->control);
} else {
struct STREAM_RECEIVE *first = p->last_sent && p->last_sent->next ? p->last_sent->next : p->flows;
struct STREAM_RECEIVE *f = first;
while (f) {
if (!f->outgoing && f->info.state == STREAM_RECEIVE_ACTIVE && downstream_send(f)) { p->last_sent = f; sent = 1; break; }
f = f->next ? f->next : p->flows;
if (f == first) break;
}
}
p->busy--; peer_sweep(p);
if (sent && p->up) {
p->wait_q = conn->send_input_q;
if (queue_waiter_wait(p->wait_q, &p->waiter, send_ready, p) < 0)
DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "stream post-send waiter failed peer=%llx", (unsigned long long)p->node_id);
}
}
/* Известный повтор продлевает аренду без смены cursor; любой новый handle обязан превысить watermark. */
static void subscribe_request(struct stream_peer *p, struct stream_message m, struct stream_local *local) {
uint8_t payload[STREAM_INFO_SIZE + STREAM_PATH_MAX * 8];
int reason = 0;
if (m.length != 8 || !m.handle || m.subblock >= local->descriptor.n) reason = STREAM_REASON_PROTOCOL;
struct STREAM_RECEIVE *f = reason ? NULL : flow_find(p, &m, 0);
uint64_t requested = m.length == 8 ? stream_read_be(m.payload, 8) : 0;
if (f && f->requested != requested) reason = STREAM_REASON_CONFLICT;
if (!f && !reason && m.handle <= p->highwater) reason = STREAM_REASON_STALE;
if (!f && !reason) {
p->highwater = m.handle;
reason = info_payload(local, m.subblock, requested, p->node_id, p->network->inst->node_id, payload);
if (!reason) {
f = flow_new(p, &local->descriptor, m.subblock, 0);
if (!f) reason = STREAM_REASON_LIMIT;
else {
f->info.handle = m.handle; f->requested = requested; f->info.start = f->info.next = stream_read_be(payload, 8);
f->info.state = STREAM_RECEIVE_ACTIVE;
}
}
} else if (f && !reason) {
reason = info_payload(local, m.subblock, f->info.start, p->node_id, p->network->inst->node_id, payload);
/* Идемпотентный ACCEPT возвращает первоначальный индекс даже если тот уже вытеснен. */
if (reason == STREAM_REASON_POSITION || reason == STREAM_REASON_EXPIRED) {
reason = info_payload(local, m.subblock, UINT64_MAX, p->node_id, p->network->inst->node_id, payload);
if (!reason) stream_write_be(payload, f->info.start, 8);
}
}
if (reason) {
uint8_t rejection[17] = {(uint8_t)reason}; uint64_t first = UINT64_MAX, last = UINT64_MAX;
if (m.subblock < local->descriptor.n) stream_buffer_range(local->buffer, m.subblock, &first, &last);
stream_write_be(rejection + 1, first, 8); stream_write_be(rejection + 9, last, 8);
m.command = STREAM_CMD_REJECT; control_put(p, m, rejection, sizeof(rejection));
DEBUG_WARN(DEBUG_CATEGORY_STREAM, "stream SUBSCRIBE rejected peer=%llx handle=%llu sub=%u reason=%d watermark=%llu",
(unsigned long long)p->node_id, (unsigned long long)m.handle, m.subblock, reason, (unsigned long long)p->highwater);
return;
}
f->deadline = get_time_tb() + STREAM_LEASE_TB;
m.command = STREAM_CMD_ACCEPT; control_put(p, m, payload, STREAM_INFO_SIZE + payload[28] * 8);
DEBUG_INFO(DEBUG_CATEGORY_STREAM, "stream ACCEPT queued peer=%llx handle=%llu sub=%u start=%llu lease=6000ms",
(unsigned long long)p->node_id, (unsigned long long)m.handle, m.subblock, (unsigned long long)f->info.start);
}
/* Raw transport authenticates непосредственного отправителя; сначала local key, затем членство и peer epoch. */
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 (stream_local_get(net->inst, &m.descriptor, &local) != STREAM_OK) {
DEBUG_TRACE(DEBUG_CATEGORY_STREAM, "stream packet ignored without local subscription command=%u", m.command); return;
}
struct stream_peer *p = peer_find(net, sender);
if (routed && m.command == STREAM_CMD_REVERSE_REQ) {
if (m.length || m.handle || m.receiver_epoch || !authorized(net, sender, &m, &local)) return;
if (!p) p = peer_open(net, sender);
if (p) { p->idle = get_time_tb(); DEBUG_INFO(DEBUG_CATEGORY_STREAM, "stream REVERSE_REQ peer=%llx attempt=%llx NCD=%p",
(unsigned long long)sender, (unsigned long long)m.sender_epoch, (void *)p->ncd); }
return;
}
if (routed && m.command != STREAM_CMD_CLOSED) {
DEBUG_WARN(DEBUG_CATEGORY_STREAM, "stream routed DATA/control forbidden command=%u", m.command); return;
}
if (m.command == STREAM_CMD_INFO_REQ && !routed) {
if (m.handle || m.length || m.subblock >= local.descriptor.n || !authorized(net, sender, &m, &local)) return;
if (!p) p = peer_open(net, sender);
if (!p || (m.receiver_epoch && m.receiver_epoch != p->own_epoch)) return;
if (p->other_epoch && p->other_epoch != m.sender_epoch) {
peer_reset(p, STREAM_REASON_TRANSPORT);
/* Remote INFO начинает новое поколение; его следующий ответ подтвердит свежий local epoch. */
}
p->other_epoch = m.sender_epoch;
uint8_t payload[STREAM_INFO_SIZE + STREAM_PATH_MAX * 8];
int reason = info_payload(&local, m.subblock, UINT64_MAX, sender, net->inst->node_id, payload);
m.command = reason ? STREAM_CMD_REJECT : STREAM_CMD_INFO_RESP;
if (reason) { uint8_t value = (uint8_t)reason; control_put(p, m, &value, 1); }
else control_put(p, m, payload, STREAM_INFO_SIZE + payload[28] * 8);
return;
}
if (!p || (!p->up && !routed) || m.receiver_epoch != p->own_epoch
|| (p->other_epoch && m.sender_epoch != p->other_epoch)) {
DEBUG_DEBUG(DEBUG_CATEGORY_STREAM, "stream stale/session packet peer=%llx command=%u handle=%llu",
(unsigned long long)sender, m.command, (unsigned long long)m.handle); return;
}
p->busy++;
struct STREAM_RECEIVE *f = flow_find(p, &m, m.command != STREAM_CMD_SUBSCRIBE && m.command != STREAM_CMD_RENEW
&& m.command != STREAM_CMD_UNSUBSCRIBE);
if (m.command == STREAM_CMD_INFO_RESP && f && !m.handle && f->info_sent) {
uint64_t start;
if (parse_info(&m, &f->descriptor, sender, net->inst->node_id, &start) == STREAM_OK) {
p->other_epoch = m.sender_epoch; f->info_ready = 1; peer_schedule(p);
} else flow_close(f, STREAM_REASON_PROTOCOL, 0);
} else if (m.command == STREAM_CMD_SUBSCRIBE) {
if (!authorized(net, sender, &m, &local)) {
uint8_t reason = STREAM_REASON_MEMBER; m.command = STREAM_CMD_REJECT; control_put(p, m, &reason, 1);
} else subscribe_request(p, m, &local);
} else if (m.command == STREAM_CMD_ACCEPT && f) {
uint64_t start;
if (parse_info(&m, &f->descriptor, sender, net->inst->node_id, &start) != STREAM_OK
|| (f->requested != UINT64_MAX && start != f->requested)) flow_close(f, STREAM_REASON_PROTOCOL, 1);
else if (f->info.state == STREAM_RECEIVE_REQUESTED) {
f->info.start = f->info.next = start; f->info.state = STREAM_RECEIVE_ACTIVE; f->renewed = get_time_tb();
DEBUG_INFO(DEBUG_CATEGORY_STREAM, "stream receive ACCEPT peer=%llx handle=%llu sub=%u start=%llu",
(unsigned long long)sender, (unsigned long long)m.handle, m.subblock, (unsigned long long)start);
}
} else if (m.command == STREAM_CMD_DATA && f && f->info.state == STREAM_RECEIVE_ACTIVE) {
struct stream_path path; size_t used;
if (stream_path_parse(m.payload, m.length, &path, &used) != STREAM_OK || m.length - used != stream_block_size(&f->descriptor)
|| stream_path_validate(&path, f->descriptor.source_node_id, sender, net->inst->node_id) != STREAM_OK) {
flow_close(f, STREAM_REASON_LOOP, 1);
} else {
const uint8_t *block = m.payload + used; uint64_t seq = stream_read_be(block + STREAM_BLOCK_SEQ_OFF, 8);
if (block[STREAM_BLOCK_SUB_OFF] != f->subblock || seq != f->info.next
|| stream_read_be(block + STREAM_BLOCK_GROUP_OFF, 8) != f->descriptor.group_id
|| stream_read_be(block + STREAM_BLOCK_SOURCE_OFF, 8) != f->descriptor.source_node_id
|| stream_read_be(block + STREAM_BLOCK_EPOCH_OFF, 8) != f->descriptor.source_epoch) flow_close(f, STREAM_REASON_PROTOCOL, 1);
else {
int result = stream_ingest_path(net->inst, block, m.length - used, &path);
if (result == STREAM_OK || result == STREAM_DUPLICATE || result == STREAM_EXPIRED) {
f->info.next = seq + 1; f->info.blocks++; f->deadline = get_time_tb() + STREAM_LEASE_TB;
} else flow_close(f, STREAM_REASON_PROTOCOL, 1);
}
}
} else if (m.command == STREAM_CMD_GAP && f && f->info.state == STREAM_RECEIVE_ACTIVE) {
if (m.length != 16 || stream_read_be(m.payload, 8) != f->info.next
|| stream_read_be(m.payload + 8, 8) <= f->info.next) flow_close(f, STREAM_REASON_PROTOCOL, 1);
else { f->info.next = stream_read_be(m.payload + 8, 8); f->info.gaps++; }
} else if (m.command == STREAM_CMD_END && f && f->info.state == STREAM_RECEIVE_ACTIVE) {
if (stream_ingest_end(net->inst, &f->descriptor, m.payload, m.length) == STREAM_OK) f->info.state = STREAM_RECEIVE_ENDED;
else flow_close(f, STREAM_REASON_PROTOCOL, 1);
} else if (m.command == STREAM_CMD_RENEW && f && !m.length) {
f->deadline = get_time_tb() + STREAM_LEASE_TB;
DEBUG_DEBUG(DEBUG_CATEGORY_STREAM, "stream lease renewed peer=%llx handle=%llu deadline=%llu",
(unsigned long long)sender, (unsigned long long)m.handle, (unsigned long long)f->deadline);
}
else if (m.command == STREAM_CMD_UNSUBSCRIBE && f && m.length == 1) flow_close(f, m.payload[0], 1);
else if ((m.command == STREAM_CMD_CLOSED || m.command == STREAM_CMD_REJECT) && f && (m.length == 1 || m.length == 17))
flow_close(f, m.payload[0], 0);
else DEBUG_DEBUG(DEBUG_CATEGORY_STREAM, "stream unknown/inactive issuance peer=%llx command=%u handle=%llu sub=%u bytes=%zu",
(unsigned long long)sender, m.command, (unsigned long long)m.handle, m.subblock, m.length);
p->busy--; peer_sweep(p);
}
/* Entry/dgram принадлежат обработчику во всех ветках. */
static void receive_raw(struct ETCP_CONN *conn, struct ll_entry *entry) {
struct stream_network *net = conn ? stream_network_get(conn->instance) : NULL;
if (net && entry && entry->dgram) dispatch(net, conn->peer_node_id, entry->dgram, entry->len, 0);
if (entry) { queue_dgram_free(entry); queue_entry_free(entry); }
}
/* Router сообщает реального отправителя и признаки проверенной подписи, не next hop. */
static void receive_routed(struct ETCP_CONN *conn, struct ll_entry *entry) {
struct stream_network *net = conn ? stream_network_get(conn->instance) : NULL;
if (net && entry && entry->dgram && entry->len >= ROUTER_SVC_PAYLOAD_OFF + STREAM_WIRE_HEADER
&& (entry->dgram[ROUTER_SVC_FLAGS_OFF] & ROUTER_FLAG_SIGNED)) {
uint64_t sender, group; memcpy(&sender, entry->dgram + ROUTER_SVC_SRC_OFF, 8);
memcpy(&group, entry->dgram + ROUTER_SVC_GROUP_OFF, 8);
const uint8_t *wire = entry->dgram + ROUTER_SVC_PAYLOAD_OFF;
if (stream_read_be(wire + 4, 8) == group) dispatch(net, sender, wire, entry->len - ROUTER_SVC_PAYLOAD_OFF, 1);
else DEBUG_WARN(DEBUG_CATEGORY_STREAM, "stream routed group mismatch");
} else if (net && entry && entry->len > ROUTER_SVC_PAYLOAD_OFF)
DEBUG_WARN(DEBUG_CATEGORY_STREAM, "stream routed control unsigned/truncated bytes=%u", entry->len);
if (entry) { queue_dgram_free(entry); queue_entry_free(entry); }
}
/* REINIT меняет stream generation при сохранении физического ETCP-соединения. */
static void transport_changed(struct ETCP_CONN *conn, int status, void *arg) {
struct stream_network *net = arg;
if (status != ETCP_CONN_STATUS_REINIT || net->stopping) return;
struct stream_peer *p = peer_find(net, conn->peer_node_id);
if (p) peer_reset(p, STREAM_REASON_TRANSPORT);
}
/* Один таймер обслуживает сроки/продления и ограниченные reverse-попытки; DATA не опрашивается таймером. */
static void network_tick(void *arg) {
struct stream_network *net = arg; net->timer = NULL;
uint64_t now = get_time_tb();
struct stream_peer *p = net->peers;
while (p) {
struct stream_peer *next = p->next;
p->busy++;
for (struct STREAM_RECEIVE *f = p->flows; f; f = f->next) {
if (f->info.state == STREAM_RECEIVE_CLOSED) continue;
if ((!f->outgoing || f->info.state < STREAM_RECEIVE_ACTIVE) && now >= f->deadline)
flow_close(f, f->info.state < STREAM_RECEIVE_ACTIVE ? STREAM_REASON_TRANSPORT : STREAM_REASON_LEASE, 1);
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;
}
}
if (!p->up && p->flows && now - p->born < STREAM_CONNECT_TB && (!p->reverse_at || now - p->reverse_at >= 10000)) {
struct stream_message m = {.descriptor = p->flows->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;
}
p->busy--; peer_sweep(p);
if (p->idle && now - p->idle >= STREAM_LEASE_TB && !p->control->count) peer_free(p);
p = next;
}
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");
}
/* Binding collision — ошибка запуска, другие обработчики не затираются. */
int stream_network_init(struct UTUN_INSTANCE *inst, struct stream_network **out) {
if (!inst || !out || *out || inst->api_bindings.callbacks[ETCP_ID_STREAM] || inst->router_bindings.callbacks[ETCP_RT_ID_STREAM]) {
DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "stream network invalid state/binding collision"); return STREAM_ERR_STATE;
}
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;
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;
}
etcp_add_conn_status_cbk(inst, transport_changed, net);
int registered = 0;
for (struct etcp_status_cbk_entry *e = inst->conn_status_cbks; e; e = e->next) if (e->fn == transport_changed && e->arg == net) registered = 1;
net->timer = uasync_set_timeout(inst->ua, STREAM_TICK_TB, net, network_tick, "stream_network");
if (!registered || !net->timer) {
DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "stream network callback/timer registration failed");
stream_network_destroy(net); return STREAM_ERR_MEMORY;
}
*out = net;
DEBUG_INFO(DEBUG_CATEGORY_STREAM, "stream network initialized raw=%u routed=%u peers=%u flows=%u controls=%u lease=6000ms renew=2000ms",
ETCP_ID_STREAM, ETCP_RT_ID_STREAM, STREAM_PEER_LIMIT, STREAM_FLOW_LIMIT, STREAM_CONTROL_LIMIT);
return STREAM_OK;
}
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);
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);
DEBUG_INFO(DEBUG_CATEGORY_STREAM, "stream network stopped flows=%u", net->flow_count); u_free(net);
}
void stream_network_changed(struct stream_network *net) {
if (net && !net->stopping) for (struct stream_peer *p = net->peers; p; p = p->next) peer_schedule(p);
}
/* Кольцо освобождается только после отмены сетевых курсоров данной точной эпохи. */
void stream_network_cancel(struct stream_network *net, const struct stream_descriptor *d) {
if (!net) return;
for (struct stream_peer *p = net->peers; p; p = p->next) {
p->busy++;
for (struct STREAM_RECEIVE *f = p->flows; f; f = f->next)
if (same_stream(&f->descriptor, d)) flow_close(f, STREAM_REASON_STOPPED, 1);
p->busy--; peer_sweep(p);
}
}
/* Исключение поставщика/получателя или self закрывает выдачи, сохраняя локальное кольцо остальных трансляций. */
void stream_network_membership_changed(struct stream_network *net, uint64_t node, const char *channel) {
if (!net || net->stopping || !channel) return;
for (struct stream_peer *p = net->peers; p; p = p->next) {
if (node != p->node_id && node != net->inst->node_id) continue;
p->busy++;
for (struct STREAM_RECEIVE *f = p->flows; f; f = f->next) {
if (f->info.state == STREAM_RECEIVE_CLOSED) continue;
struct TOPO_GROUP *group = topo_groups_find(net->inst->topo_groups, f->descriptor.group_id);
if (!group || strcmp(group->channel_id, channel)) continue;
struct stream_member_state state;
int result = stream_member_load(net->inst, group->group_id, node, &state, NULL);
if (result == STREAM_ERR_NOT_MEMBER || result == STREAM_ERR_SIGNATURE || result == STREAM_ERR_FORMAT)
flow_close(f, STREAM_REASON_MEMBER, 1);
}
p->busy--; peer_sweep(p);
}
}
int stream_receive_open(struct STREAM_HANDLE *subscription, uint64_t node, uint8_t sub, uint64_t seq, struct STREAM_RECEIVE **out) {
struct UTUN_INSTANCE *inst; struct stream_descriptor d;
if (!out || *out || stream_handle_context(subscription, &inst, &d) != STREAM_OK || sub >= d.n || !node || node == inst->node_id) {
DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "stream receive open invalid arguments peer=%llx sub=%u", (unsigned long long)node, sub);
return STREAM_ERR_ARGUMENT;
}
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)) {
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);
if (!p) return STREAM_ERR_STATE;
for (struct STREAM_RECEIVE *f = p->flows; f; f = f->next)
if (f->outgoing && f->info.state != STREAM_RECEIVE_CLOSED && same_stream(&f->descriptor, &d) && f->subblock == sub) {
DEBUG_WARN(DEBUG_CATEGORY_STREAM, "stream receive already open peer=%llx sub=%u", (unsigned long long)node, sub);
return STREAM_ERR_STATE;
}
int bootstrap = 1;
for (struct stream_peer *other = net->peers; other; other = other->next)
for (struct STREAM_RECEIVE *f = other->flows; f; f = f->next)
if (f->outgoing && f->info.state != STREAM_RECEIVE_CLOSED && same_stream(&f->descriptor, &d)) bootstrap = 0;
struct STREAM_RECEIVE *f = flow_new(p, &d, sub, 1);
if (!f) return STREAM_ERR_MEMORY;
f->owned = 1; f->requested = seq; f->bootstrap = bootstrap; *out = f; peer_schedule(p); return STREAM_OK;
}
void stream_receive_close(struct STREAM_RECEIVE *f) {
if (!f) return;
f->owned = 0;
if (!f->peer) { u_free(f); return; }
struct stream_peer *p = f->peer;
flow_close(f, STREAM_REASON_STOPPED, 1); peer_sweep(p);
}
int stream_receive_get_info(const struct STREAM_RECEIVE *f, struct stream_receive_info *info) {
if (!f || !info) { DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "stream receive info invalid arguments"); return STREAM_ERR_ARGUMENT; }
*info = f->info; return STREAM_OK;
}