diff --git a/src/stream/stream_network.c b/src/stream/stream_network.c index b31ae554..3c8ef820 100644 --- a/src/stream/stream_network.c +++ b/src/stream/stream_network.c @@ -343,14 +343,25 @@ static int authorized(struct stream_network *net, uint64_t peer, const struct st /* 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 range = stream_buffer_range(local->buffer, sub, &first, &last); + if (range != STREAM_OK && local->publisher && (seq == UINT64_MAX || seq == 0)) { + path = (struct stream_path){.count = 1, .nodes = {self}}; + stream_write_be(payload + 8, seq, 8); + stream_write_be(payload + 16, UINT64_MAX, 8); stream_write_be(payload + 24, UINT64_MAX, 8); + payload[32] = local->descriptor.n; payload[33] = local->descriptor.r; + stream_write_be(payload + 34, local->descriptor.block_size, 2); stream_path_write(payload + STREAM_INFO_PATH_OFF, &path); + return 0; + } int result = stream_buffer_read(local->buffer, seq, sub, &wire, &length, &actual); + int tail = range == STREAM_OK && last < UINT64_MAX - 1 && seq == last + 1; + if (tail) result = stream_buffer_read(local->buffer, UINT64_MAX, 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 + || range != 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 + 8, actual, 8); stream_write_be(payload + 16, first, 8); stream_write_be(payload + 24, last, 8); + stream_write_be(payload + 8, tail ? seq : actual, 8); stream_write_be(payload + 16, first, 8); stream_write_be(payload + 24, last, 8); payload[32] = local->descriptor.n; payload[33] = local->descriptor.r; stream_write_be(payload + 34, local->descriptor.block_size, 2); stream_path_write(payload + STREAM_INFO_PATH_OFF, &path); return 0; @@ -359,18 +370,24 @@ static int info_payload(struct stream_local *local, uint8_t sub, uint64_t seq, u 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; + uint64_t actual = m->length >= STREAM_INFO_SIZE ? stream_read_be(m->payload + 8, 8) : UINT64_MAX; + uint64_t first = m->length >= STREAM_INFO_SIZE ? stream_read_be(m->payload + 16, 8) : UINT64_MAX; + uint64_t last = m->length >= STREAM_INFO_SIZE ? stream_read_be(m->payload + 24, 8) : UINT64_MAX; + int empty = first == UINT64_MAX && last == UINT64_MAX; 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 || stream_path_parse(m->payload + STREAM_INFO_PATH_OFF, m->length - STREAM_INFO_PATH_OFF, &path, &used) != STREAM_OK || m->length != STREAM_INFO_PATH_OFF + used || path.count == STREAM_PATH_MAX || stream_path_validate(&path, d->source_node_id, peer, self) != STREAM_OK - || stream_read_be(m->payload + 8, 8) == UINT64_MAX || stream_read_be(m->payload + 8, 8) > stream_read_be(m->payload + 24, 8) - || stream_read_be(m->payload + 16, 8) > stream_read_be(m->payload + 24, 8)) { + || (!empty && (actual == UINT64_MAX || first > last || last == UINT64_MAX + || (actual > last && !(m->command == STREAM_CMD_ACCEPT && last < UINT64_MAX - 1 && actual == last + 1)))) + || (empty && (peer != d->source_node_id || path.count != 1 + || actual != (m->command == STREAM_CMD_ACCEPT ? 0 : UINT64_MAX)))) { DEBUG_WARN(DEBUG_CATEGORY_STREAM, "stream INFO/ACCEPT invalid command=%u bytes=%zu", m->command, m->length); return STREAM_ERR_FORMAT; } const uint8_t *offer = m->payload + STREAM_INFO_OFFER_OFF; - if ((offer[0] & ~15u) || stream_read_be(offer + 3, 2) > 10000) { + if ((offer[0] & ~31u) || !!(offer[0] & 16) == empty || 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; } @@ -392,6 +409,7 @@ static void offer_payload(struct stream_peer *p, struct stream_local *local, uin 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; + if (stream_read_be(payload + 16, 8) != UINT64_MAX) offer[0] |= 16; 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; @@ -604,6 +622,7 @@ static void subscribe_request(struct stream_peer *p, struct stream_message m, st 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, 8); + if (f->info.start == UINT64_MAX) { f->info.start = f->info.next = 0; stream_write_be(payload + 8, 0, 8); } f->info.state = STREAM_RECEIVE_ACTIVE; uint64_t count = 0; stream_buffer_pending(local->buffer, f->info.next, f->subblock, &count, NULL); diff --git a/src/stream/stream_optimizer.c b/src/stream/stream_optimizer.c index 9a0189c7..9c135744 100644 --- a/src/stream/stream_optimizer.c +++ b/src/stream/stream_optimizer.c @@ -29,6 +29,7 @@ struct stream_supplier { struct stream_selection { struct STREAM_RECEIVE *current, *replacement; uint64_t opened_ms, replacement_ms, better_ms, better_node; + int opened; }; struct stream_optimizer { struct UTUN_INSTANCE *inst; @@ -123,6 +124,7 @@ static void probes_schedule(struct stream_optimizer *o, struct STREAM_HANDLE *h, unsigned pending = 0, active = 0; for (struct stream_supplier *s = o->suppliers; s; s = s->next) { if (s->probe) active++; + if (now < s->retry_ms) s->cycle = o->cycle; if (s->compatible && s->cycle != o->cycle) pending++; } if (!pending && !active && now - o->cycle_ms >= STREAM_PROBE_CYCLE_MS) { @@ -183,7 +185,7 @@ static struct stream_supplier *supplier_choose(struct stream_optimizer *o, unsig || (s->probe && s->probe_sub == sub)) continue; struct stream_offer *offer = &s->offers[sub]; struct stream_receive_info *info = &offer->info; if (now < offer->retry_ms || !info->info_ms || now - info->info_ms > STREAM_OFFER_TTL_MS || !info->accepting || !info->free_slots - || (seq != UINT64_MAX && (seq < info->first_available || seq > info->last_available))) continue; + || (seq != UINT64_MAX && info->first_available != UINT64_MAX && seq < info->first_available)) continue; if (!pass && o->descriptor.r && usage(o, s->node, sub) >= o->descriptor.r) continue; if (!best) { best = s; continue; } struct stream_receive_info *previous = &best->offers[sub].info; @@ -318,12 +320,12 @@ void stream_optimizer_tick(struct stream_optimizer *o, struct STREAM_HANDLE *h, struct stream_selection *pick = &o->selected[sub]; struct stream_receive_info info = {0}; if (pick->replacement || (!local.anchored && busy)) continue; if (pick->current) stream_receive_get_info(pick->current, &info); - uint64_t seq = local.anchored ? local.next : UINT64_MAX; + uint64_t seq = local.anchored ? (pick->opened ? local.next : local.start) : UINT64_MAX; struct stream_supplier *s = supplier_choose(o, sub, seq, info.peer, now); if (!s) { pick->better_ms = pick->better_node = 0; continue; } if (!pick->current) { if (stream_receive_open(h, s->node, (uint8_t)sub, seq, &pick->current) == STREAM_OK) { - pick->opened_ms = now; busy++; + pick->opened_ms = now; pick->opened = 1; busy++; DEBUG_INFO(DEBUG_CATEGORY_STREAM, "stream optimizer selected peer=%llx sub=%u index=%llu", (unsigned long long)s->node, sub, (unsigned long long)seq); }