diff --git a/src/stream/stream_network.c b/src/stream/stream_network.c index eaea764d..d8a4197b 100644 --- a/src/stream/stream_network.c +++ b/src/stream/stream_network.c @@ -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;