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

Handle empty sources and next-batch tail subscriptions without startup races

master
evgeny 1 week ago
parent
ef48fa2a37
commit
94c482ed35
2 changed files with 29 additions and 8 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. 29
      src/stream/stream_network.c
  2. 8
      src/stream/stream_optimizer.c

29
src/stream/stream_network.c
Unescape View File

@ -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);

8
src/stream/stream_optimizer.c
Unescape View File

@ -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);
}

Write Preview
Loading…
Cancel
Save