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

Implement DATA-free stream probes with ranges, paths and load offers

Browse Source
This commit is contained in:
evgeny committed 2026-10-12 00:01:39 +02:00
1 parent 984153b90a
commit 35ad47c653
1 file changed
+86 -20
Download Patch File Download Diff File Expand all files Collapse all files
src/stream/stream_network.c
+86 -20
View File
@@ -324,7 +324,7 @@ static struct STREAM_RECEIVE *flow_new(struct stream_peer *p, const struct strea
f->admitted_ms = get_time_tb() / 10; stream_load_init(&f->load, f->admitted_ms);
stream_metrics_init(&f->metrics); stream_metrics_init(&f->offered_metrics);
if (!outgoing) { stream_load_changed(&p->load); stream_load_changed(&p->network->load); }
f->info.peer = p->node_id; f->deadline = get_time_tb() + STREAM_CONNECT_TB;
f->info.peer = p->node_id; f->info.subblock = sub; f->deadline = get_time_tb() + STREAM_CONNECT_TB;
f->next = p->flows; p->flows = f; p->network->flow_count++; p->idle = 0; return f;
}
@@ -356,7 +356,8 @@ static int info_payload(struct stream_local *local, uint8_t sub, uint64_t seq, u
stream_path_write(payload + STREAM_INFO_PATH_OFF, &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) {
static int parse_info(const struct stream_message *m, const struct stream_descriptor *d, uint64_t peer, uint64_t self,
struct stream_receive_info *info, uint64_t *start) {
struct stream_path path; size_t used;
if (m->length < STREAM_INFO_SIZE || m->payload[32] != d->n || m->payload[33] != d->r
|| stream_read_be(m->payload + 34, 2) != d->block_size
@@ -368,7 +369,52 @@ static int parse_info(const struct stream_message *m, const struct stream_descri
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, 8); return STREAM_OK;
const uint8_t *offer = m->payload + STREAM_INFO_OFFER_OFF;
if ((offer[0] & ~15u) || stream_read_be(offer + 3, 2) > 10000) {
DEBUG_WARN(DEBUG_CATEGORY_STREAM, "stream invalid offer flags=%u cpu=%llu", offer[0],
(unsigned long long)stream_read_be(offer + 3, 2)); return STREAM_ERR_FORMAT;
}
*start = stream_read_be(m->payload + 8, 8);
info->path = path; info->first_available = stream_read_be(m->payload + 16, 8);
info->last_available = stream_read_be(m->payload + 24, 8); info->info_ms = get_time_tb() / 10;
info->accepting = (offer[0] & 1) != 0; info->path_rtt_valid = (offer[0] & 2) != 0;
info->progress_valid = (offer[0] & 4) != 0; info->cpu_valid = (offer[0] & 8) != 0;
info->free_slots = (uint16_t)stream_read_be(offer + 1, 2); info->cpu_basis_points = (uint16_t)stream_read_be(offer + 3, 2);
info->supplier_queue_bytes = stream_read_be(offer + 5, 8); info->progress_age_ms = stream_read_be(offer + 13, 8);
info->path_rtt_ms = (uint32_t)stream_read_be(offer + 21, 4); return STREAM_OK;
}
/* Предложение привязано к фактическому пути последнего блока; другой upstream не подтверждает его RTT. */
static void offer_payload(struct stream_peer *p, struct stream_local *local, uint8_t sub, uint8_t *payload) {
struct stream_network *net = p->network; uint64_t now = get_time_tb() / 10;
uint8_t *offer = payload + STREAM_INFO_OFFER_OFF;
unsigned count = 0; for (struct STREAM_RECEIVE *f = p->flows; f; f = f->next) count++;
unsigned slots = STREAM_PEER_FLOW_LIMIT - count;
if (slots > STREAM_FLOW_LIMIT - net->flow_count) slots = STREAM_FLOW_LIMIT - net->flow_count;
offer[0] = !p->pressure.active && !p->cooling ? 1 : 0;
stream_write_be(offer + 1, slots, 2);
double cpu = net->cpu.valid ? net->cpu.average_percent : 0;
if (cpu < 0) cpu = 0; if (cpu > 100) cpu = 100;
if (net->cpu.valid) offer[0] |= 8;
stream_write_be(offer + 3, (uint16_t)(cpu * 100), 2); stream_write_be(offer + 5, p->load.queued.bytes, 8);
struct stream_metric_snapshot snapshot = {0};
int valid = stream_local_metrics(net->inst, &local->descriptor, sub, now, &snapshot) == STREAM_OK;
stream_metric_write(payload + STREAM_INFO_METRIC_OFF, &snapshot, valid);
if (valid) offer[0] |= 4;
stream_write_be(offer + 13, snapshot.progress_age_ms, 8);
if (local->publisher) { offer[0] |= 2; return; }
struct stream_path path; size_t used;
if (stream_path_parse(payload + STREAM_INFO_PATH_OFF, 1 + payload[STREAM_INFO_PATH_OFF] * 8, &path, &used) != STREAM_OK) return;
for (struct stream_peer *up = net->peers; up; up = up->next) {
for (struct STREAM_RECEIVE *f = up->flows; f; f = f->next) {
if (!f->outgoing || f->info.probe_only || f->info.state != STREAM_RECEIVE_ACTIVE || f->subblock != sub
|| !same_stream(&f->descriptor, &local->descriptor) || !f->info.rtt_valid || !f->info.path_rtt_valid
|| now - f->info.info_ms > 30000 || path.count != f->info.path.count + 1
|| memcmp(path.nodes, f->info.path.nodes, f->info.path.count * sizeof(uint64_t))) continue;
uint64_t total = (uint64_t)f->info.rtt_ms + f->info.path_rtt_ms;
offer[0] |= 2; stream_write_be(offer + 21, total > UINT32_MAX ? UINT32_MAX : total, 4); return;
}
}
}
/* Один пакет на свободный порог; собственный буфер переживает синхронное освобождение entry в etcp_send. */
@@ -394,7 +440,7 @@ static void receive_progress(struct stream_network *net, const struct stream_des
uint8_t seen[RS_MAX_BLOCKS] = {0}; unsigned distinct = 0; uint64_t before = UINT64_MAX;
for (struct stream_peer *p = net->peers; p; p = p->next) {
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)) continue;
if (!f->outgoing || f->info.probe_only || f->info.state == STREAM_RECEIVE_CLOSED || !same_stream(&f->descriptor, d)) continue;
if (f->info.state < STREAM_RECEIVE_ACTIVE) return;
if (!seen[f->subblock]) { seen[f->subblock] = 1; distinct++; }
if (f->info.next < before) before = f->info.next;
@@ -471,7 +517,7 @@ static void prepare_requests(struct stream_peer *p) {
stream_write_be(probe, f->probe, 8);
if (control_put(p, m, probe, sizeof(probe)) == STREAM_OK) { f->info_sent = 1; f->info.state = STREAM_RECEIVE_INFO; }
}
if (!f->info_ready || !p->other_epoch) continue;
if (!f->info_ready || f->info.probe_only || !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; }
@@ -588,9 +634,7 @@ static void subscribe_request(struct stream_peer *p, struct stream_message m, st
return;
}
f->deadline = get_time_tb() + STREAM_LEASE_TB;
struct stream_metric_snapshot snapshot = {0};
int valid = stream_local_metrics(p->network->inst, &local->descriptor, m.subblock, get_time_tb() / 10, &snapshot) == STREAM_OK;
stream_metric_write(payload + STREAM_INFO_METRIC_OFF, &snapshot, valid);
offer_payload(p, local, m.subblock, payload);
m.command = STREAM_CMD_ACCEPT;
if (control_put(p, m, payload, STREAM_INFO_SIZE + payload[STREAM_INFO_PATH_OFF] * 8) != STREAM_OK) {
flow_close(f, STREAM_REASON_LIMIT, 0); return;
@@ -648,9 +692,7 @@ static void dispatch(struct stream_network *net, uint64_t sender, const uint8_t
if (reason) {
uint8_t failure[9] = {(uint8_t)reason}; memcpy(failure + 1, m.payload, 8); control_put(p, m, failure, sizeof(failure));
} else {
struct stream_metric_snapshot snapshot = {0};
int valid = stream_local_metrics(net->inst, &local.descriptor, m.subblock, get_time_tb() / 10, &snapshot) == STREAM_OK;
stream_metric_write(payload + STREAM_INFO_METRIC_OFF, &snapshot, valid);
offer_payload(p, &local, m.subblock, payload);
control_put(p, m, payload, STREAM_INFO_SIZE + payload[STREAM_INFO_PATH_OFF] * 8);
}
return;
@@ -669,9 +711,9 @@ static void dispatch(struct stream_network *net, uint64_t sender, const uint8_t
uint64_t start;
if (m.length < 8 || stream_read_be(m.payload, 8) != f->probe) {
DEBUG_DEBUG(DEBUG_CATEGORY_STREAM, "stream stale INFO probe peer=%llx", (unsigned long long)sender);
} else if (parse_info(&m, &f->descriptor, sender, net->inst->node_id, &start) == STREAM_OK
} else if (parse_info(&m, &f->descriptor, sender, net->inst->node_id, &f->info, &start) == STREAM_OK
&& stream_metric_read(m.payload + STREAM_INFO_METRIC_OFF, &f->info.supplier_metrics, &f->info.supplier_metrics_valid) == STREAM_OK) {
p->other_epoch = m.sender_epoch; f->info_ready = 1; peer_schedule(p);
p->other_epoch = m.sender_epoch; f->info_ready = f->info.info_ready = 1; 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;
} else flow_close(f, STREAM_REASON_PROTOCOL, 0);
@@ -681,7 +723,7 @@ static void dispatch(struct stream_network *net, uint64_t sender, const uint8_t
} 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
if (parse_info(&m, &f->descriptor, sender, net->inst->node_id, &f->info, &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();
@@ -704,7 +746,7 @@ static void dispatch(struct stream_network *net, uint64_t sender, const uint8_t
if (result == STREAM_OK || result == STREAM_DUPLICATE || result == STREAM_EXPIRED) {
stream_metrics_observe(&f->metrics, seq, (uint16_t)stream_read_be(block + STREAM_BLOCK_TIMESTAMP_OFF, 2),
block[STREAM_BLOCK_FLAGS_OFF], (uint32_t)stream_block_size(&f->descriptor), get_time_tb() / 10);
f->info.next = seq + 1; f->info.blocks++; f->deadline = get_time_tb() + STREAM_LEASE_TB;
f->info.next = seq + 1; f->info.blocks++; f->info.path = path; f->deadline = get_time_tb() + STREAM_LEASE_TB;
} else flow_close(f, STREAM_REASON_PROTOCOL, 1);
}
}
@@ -799,7 +841,7 @@ static void network_tick(void *arg) {
for (struct STREAM_RECEIVE *f = p->flows; f; f = f->next) {
if (f->info.state == STREAM_RECEIVE_CLOSED) continue;
f->info.metrics_valid = stream_metrics_poll(&f->metrics, now_ms, &f->info.metrics) == STREAM_OK;
if ((!f->outgoing || f->info.state < STREAM_RECEIVE_ACTIVE) && now >= f->deadline)
if ((!f->outgoing || (f->info.state < STREAM_RECEIVE_ACTIVE && !f->info.info_ready)) && 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,
@@ -892,7 +934,8 @@ void stream_network_membership_changed(struct stream_network *net, uint64_t node
}
}
int stream_receive_open(struct STREAM_HANDLE *subscription, uint64_t node, uint8_t sub, uint64_t seq, struct STREAM_RECEIVE **out) {
static int receive_open(struct STREAM_HANDLE *subscription, uint64_t node, uint8_t sub, uint64_t seq, int probe_only,
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);
@@ -907,23 +950,46 @@ int stream_receive_open(struct STREAM_HANDLE *subscription, uint64_t node, uint8
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) {
if (f->outgoing && f->info.state != STREAM_RECEIVE_CLOSED && same_stream(&f->descriptor, &d) && f->subblock == sub
&& f->info.probe_only == probe_only) {
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;
if (f->outgoing && !f->info.probe_only && 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);
f->owned = 1; f->requested = seq; f->bootstrap = bootstrap; f->info.probe_only = probe_only; *out = f; peer_schedule(p);
DEBUG_INFO(DEBUG_CATEGORY_STREAM, "stream receive opened peer=%llx group=%llx source=%llx epoch=%llx sub=%u index=%llu bootstrap=%d",
(unsigned long long)node, (unsigned long long)d.group_id, (unsigned long long)d.source_node_id,
(unsigned long long)d.source_epoch, sub, (unsigned long long)seq, bootstrap);
return STREAM_OK;
}
int stream_receive_open(struct STREAM_HANDLE *subscription, uint64_t node, uint8_t sub, uint64_t seq, struct STREAM_RECEIVE **out) {
return receive_open(subscription, node, sub, seq, 0, out);
}
int stream_probe_open(struct STREAM_HANDLE *subscription, uint64_t node, uint8_t sub, struct STREAM_RECEIVE **out) {
return receive_open(subscription, node, sub, UINT64_MAX, 1, out);
}
int stream_probe_next(struct STREAM_RECEIVE *f, uint8_t sub) {
if (!f || !f->peer || !f->info.probe_only || !f->info.info_ready || sub >= f->descriptor.n) {
DEBUG_WARN(DEBUG_CATEGORY_STREAM, "stream probe next invalid state/sub=%u", sub); return STREAM_ERR_STATE;
}
for (struct STREAM_RECEIVE *other = f->peer->flows; other; other = other->next)
if (other != f && other->outgoing && !other->info.handle && other->subblock == sub
&& same_stream(&other->descriptor, &f->descriptor)) {
DEBUG_WARN(DEBUG_CATEGORY_STREAM, "stream probe next conflicts with pending INFO sub=%u", sub); return STREAM_WAIT;
}
f->subblock = f->info.subblock = sub; f->info_sent = f->info_ready = f->info.info_ready = 0;
f->probe = f->probe_sent_ms = 0; f->deadline = get_time_tb() + STREAM_CONNECT_TB;
f->info.state = STREAM_RECEIVE_CONNECTING; peer_schedule(f->peer); return STREAM_OK;
}
void stream_receive_close(struct STREAM_RECEIVE *f) {
if (!f) return;
f->owned = 0;