From a07c53a397b894d85d24e121c873b82f3bad266d Mon Sep 17 00:00:00 2001 From: evgeny Date: Sun, 11 Oct 2026 02:37:03 +0200 Subject: [PATCH] Measure stream issuances, probe RTT and pending data demand --- src/stream/stream_network.c | 187 +++++++++++++++++++++++++++++++----- 1 file changed, 162 insertions(+), 25 deletions(-) diff --git a/src/stream/stream_network.c b/src/stream/stream_network.c index 414e5b66..780b2e79 100644 --- a/src/stream/stream_network.c +++ b/src/stream/stream_network.c @@ -14,13 +14,20 @@ #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_CONTROL_MAX (STREAM_WIRE_HEADER + STREAM_OVERLOAD_HEADER + STREAM_OVERLOAD_MAX * 16) #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_cut { uint64_t handle, rate; int closed; }; +struct stream_pressure_request { + struct stream_cut cuts[STREAM_OVERLOAD_MAX]; + uint64_t id, required_rate, released_rate, deadline_ms; + unsigned count, required_count, released_count; + int active, sent; +}; struct STREAM_RECEIVE { struct STREAM_RECEIVE *next; struct stream_peer *peer; @@ -29,6 +36,11 @@ struct STREAM_RECEIVE { uint64_t requested, deadline, renewed; uint8_t subblock; int outgoing, owned, bootstrap, info_sent, info_ready; + struct stream_metrics metrics, offered_metrics; + struct stream_load load; + uint64_t admitted_ms, pending_highest, probe, probe_sent_ms; + uint64_t gap_from; + int gap_pending, account_failed; }; struct stream_peer { struct stream_peer *next; @@ -40,6 +52,10 @@ struct stream_peer { void *pump; uint64_t node_id, own_epoch, other_epoch, next_handle, highwater, born, idle, reverse_at; int up, busy; + struct stream_load load; + struct stream_pressure_request pressure, requested; + uint64_t pressure_id, pressure_retry_ms, clear_since_ms; + int cooling; }; struct stream_network { struct UTUN_INSTANCE *inst; @@ -47,12 +63,33 @@ struct stream_network { void *timer; unsigned peer_count, flow_count; int stopping; + struct stream_load load; + struct uasync_load cpu; + uint64_t next_tick_ms, callback_late_ms; + int global_pressure; }; 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 void pressure_tick(struct stream_network *net, uint64_t now); +static void pressure_receive(struct stream_peer *peer, const struct stream_message *message); + +/* Один блок принадлежит логической очереди выдачи/пира/модуля; ни одна из них не хранит копию DATA. */ +static void load_account(struct STREAM_RECEIVE *f, uint64_t packets, uint64_t bytes, int added, int sent) { + if (!packets) return; + struct stream_load *loads[] = {&f->load, &f->peer->load, &f->peer->network->load}; + for (unsigned i = 0; i < 3; i++) { + int result = added ? stream_load_add(loads[i], packets, bytes) : stream_load_remove(loads[i], packets, bytes, sent); + if (result != STREAM_OK) { + f->account_failed = 1; + DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "stream ledger failed level=%u peer=%llx handle=%llu add=%d sent=%d packets=%llu bytes=%llu", + i, (unsigned long long)f->peer->node_id, (unsigned long long)f->info.handle, added, sent, + (unsigned long long)packets, (unsigned long long)bytes); + } + } +} 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; @@ -135,6 +172,17 @@ static int route_control(struct stream_peer *p, struct stream_message m, const u 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 (!f->outgoing && p) { + for (unsigned i = 0; i < p->pressure.count; i++) { + struct stream_cut *cut = &p->pressure.cuts[i]; + if (p->pressure.active && !cut->closed && cut->handle == f->info.handle) { + cut->closed = 1; p->pressure.released_rate += cut->rate; p->pressure.released_count++; + } + } + load_account(f, f->load.queued.packets, f->load.queued.bytes, 0, 0); + stream_load_changed(&p->load); stream_load_changed(&p->network->load); + } + stream_metrics_end(&f->metrics); stream_metrics_end(&f->offered_metrics); 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, @@ -176,6 +224,8 @@ static void peer_reset(struct stream_peer *p, int reason) { 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); + memset(&p->pressure, 0, sizeof(p->pressure)); memset(&p->requested, 0, sizeof(p->requested)); + p->pressure_id = 0; p->cooling = 0; stream_load_init(&p->load, get_time_tb() / 10); 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); } @@ -207,6 +257,7 @@ static struct stream_peer *peer_open(struct stream_network *net, uint64_t node) 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(); + stream_load_init(&p->load, p->born / 10); 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); @@ -252,6 +303,9 @@ static struct STREAM_RECEIVE *flow_new(struct stream_peer *p, const struct strea 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->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->next = p->flows; p->flows = f; p->network->flow_count++; p->idle = 0; return f; } @@ -278,24 +332,25 @@ static int info_payload(struct stream_local *local, uint8_t sub, uint64_t seq, u || 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; + stream_write_be(payload + 8, 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; } 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 + 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) == 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)) { + || 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)) { 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; + *start = stream_read_be(m->payload + 8, 8); return STREAM_OK; } /* Один пакет на свободный порог; собственный буфер переживает синхронное освобождение entry в etcp_send. */ @@ -335,6 +390,12 @@ 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}; + if (f->gap_pending) { + uint8_t payload[16]; stream_write_be(payload, f->gap_from, 8); stream_write_be(payload + 8, f->info.next, 8); + m.command = STREAM_CMD_GAP; + if (raw_send(p, m, payload, sizeof(payload)) != STREAM_OK) { flow_close(f, STREAM_REASON_TRANSPORT, 0); return 0; } + f->gap_pending = 0; f->info.gaps++; return 1; + } 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); @@ -348,7 +409,9 @@ static int downstream_send(struct STREAM_RECEIVE *f) { 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++; } + if (f->info.state == STREAM_RECEIVE_ACTIVE) { + load_account(f, 1, length, 0, 1); f->info.next = actual + 1; f->info.blocks++; + } return 1; } if (result == STREAM_EXPIRED || result == STREAM_GAP) { @@ -383,7 +446,11 @@ static void prepare_requests(struct stream_peer *p) { 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; } + uint8_t probe[8]; + if (!f->probe) f->probe = new_epoch(); + if (!f->probe) { flow_close(f, STREAM_REASON_PROTOCOL, 0); continue; } + 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; uint64_t requested = f->requested; @@ -421,6 +488,12 @@ static void peer_pump(void *arg) { int sent = 0; struct ll_entry *e = queue_data_get(p->control); if (e) { + if (e->len >= STREAM_WIRE_HEADER && e->dgram[2] == STREAM_CMD_INFO_REQ) { + struct stream_message m; + if (stream_wire_parse(e->dgram, e->len, &m) == STREAM_OK) { + struct STREAM_RECEIVE *f = flow_find(p, &m, 1); if (f) f->probe_sent_ms = get_time_tb() / 10; + } + } 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); @@ -446,7 +519,7 @@ static void peer_pump(void *arg) { /* Известный повтор продлевает аренду без смены 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]; + uint8_t payload[STREAM_INFO_SIZE + STREAM_PATH_MAX * 8] = {0}; 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); @@ -460,8 +533,16 @@ static void subscribe_request(struct stream_peer *p, struct stream_message m, st 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.handle = m.handle; f->requested = requested; f->info.start = f->info.next = stream_read_be(payload + 8, 8); f->info.state = STREAM_RECEIVE_ACTIVE; + uint64_t count = 0; + stream_buffer_pending(local->buffer, f->info.next, f->subblock, &count, NULL); + load_account(f, count, count * stream_block_size(&f->descriptor), 1, 0); + f->pending_highest = stream_read_be(payload + 24, 8); + const uint8_t *wire; size_t length; uint64_t seq; + if (stream_buffer_read(local->buffer, UINT64_MAX, f->subblock, &wire, &length, &seq) == STREAM_OK) + stream_metrics_observe(&f->offered_metrics, seq, (uint16_t)stream_read_be(wire + STREAM_BLOCK_TIMESTAMP_OFF, 2), + wire[STREAM_BLOCK_FLAGS_OFF], (uint32_t)length, get_time_tb() / 10); } } } else if (f && !reason) { @@ -469,7 +550,7 @@ static void subscribe_request(struct stream_peer *p, struct stream_message m, st /* Идемпотентный 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) stream_write_be(payload + 8, f->info.start, 8); } } if (reason) { @@ -483,8 +564,11 @@ 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); m.command = STREAM_CMD_ACCEPT; - if (control_put(p, m, payload, STREAM_INFO_SIZE + payload[28] * 8) != STREAM_OK) { + 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; } DEBUG_INFO(DEBUG_CATEGORY_STREAM, "stream ACCEPT queued peer=%llx handle=%llu sub=%u start=%llu lease=6000ms", @@ -495,6 +579,13 @@ static void subscribe_request(struct stream_peer *p, struct stream_message m, st 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 (m.command == STREAM_CMD_OVERLOAD || m.command == STREAM_CMD_OVERLOAD_SELECT) { + struct stream_peer *p = peer_find(net, sender); + if (!routed && p && p->up && m.receiver_epoch == p->own_epoch && m.sender_epoch == p->other_epoch && !m.handle) { + p->busy++; pressure_receive(p, &m); p->busy--; peer_sweep(p); + } else DEBUG_DEBUG(DEBUG_CATEGORY_STREAM, "stream stale/unknown pressure request peer=%llx", (unsigned long long)sender); + 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; } @@ -514,7 +605,7 @@ static void dispatch(struct stream_network *net, uint64_t sender, const uint8_t 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) { + if (m.handle || m.length != 8 || !stream_read_be(m.payload, 8) || m.subblock >= local.descriptor.n) { DEBUG_WARN(DEBUG_CATEGORY_STREAM, "stream INFO_REQ malformed peer=%llx bytes=%zu handle=%llu sub=%u n=%u", (unsigned long long)sender, m.length, (unsigned long long)m.handle, m.subblock, local.descriptor.n); return; } @@ -526,11 +617,18 @@ static void dispatch(struct stream_network *net, uint64_t sender, const uint8_t /* Remote INFO начинает новое поколение; его следующий ответ подтвердит свежий local epoch. */ } p->other_epoch = m.sender_epoch; - uint8_t payload[STREAM_INFO_SIZE + STREAM_PATH_MAX * 8]; + uint8_t payload[STREAM_INFO_SIZE + STREAM_PATH_MAX * 8] = {0}; + memcpy(payload, m.payload, 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); + 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); + control_put(p, m, payload, STREAM_INFO_SIZE + payload[STREAM_INFO_PATH_OFF] * 8); + } return; } if (!p || (!p->up && !routed) || m.receiver_epoch != p->own_epoch @@ -545,8 +643,13 @@ static void dispatch(struct stream_network *net, uint64_t sender, const uint8_t && 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) { + 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 + && 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); + 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); } else if (m.command == STREAM_CMD_SUBSCRIBE) { if (!authorized(net, sender, &m, &local)) { @@ -575,6 +678,8 @@ static void dispatch(struct stream_network *net, uint64_t sender, const uint8_t else { int result = stream_ingest_path(net->inst, block, m.length - used, &path); 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; } else flow_close(f, STREAM_REASON_PROTOCOL, 1); } @@ -586,7 +691,9 @@ static void dispatch(struct stream_network *net, uint64_t sender, const uint8_t } else if (m.command == STREAM_CMD_END && f && f->info.state == STREAM_RECEIVE_ACTIVE) { uint64_t last = m.length == STREAM_END_SIZE ? stream_read_be(m.payload + 34, 8) : UINT64_MAX; if (m.length == STREAM_END_SIZE && last != UINT64_MAX && f->info.next == last + 1 - && stream_ingest_end(net->inst, &f->descriptor, m.payload, m.length) == STREAM_OK) f->info.state = STREAM_RECEIVE_ENDED; + && stream_ingest_end(net->inst, &f->descriptor, m.payload, m.length) == STREAM_OK) { + f->info.state = STREAM_RECEIVE_ENDED; stream_metrics_end(&f->metrics); + } 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; @@ -594,7 +701,9 @@ static void dispatch(struct stream_network *net, uint64_t sender, const uint8_t (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)) + else if (m.command == STREAM_CMD_REJECT && f && !m.handle && m.length == 9 && stream_read_be(m.payload + 1, 8) == f->probe) + flow_close(f, m.payload[0], 0); + else if ((m.command == STREAM_CMD_CLOSED || m.command == STREAM_CMD_REJECT) && f && m.handle && (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); @@ -636,6 +745,9 @@ static void transport_changed(struct ETCP_CONN *conn, int status, void *arg) { static void network_tick(void *arg) { struct stream_network *net = arg; net->timer = NULL; uint64_t now = get_time_tb(); + uint64_t now_ms = now / 10; + net->callback_late_ms = now_ms > net->next_tick_ms ? now_ms - net->next_tick_ms : 0; + uasync_get_load(net->inst->ua, &net->cpu); struct stream_peer *p = net->peers; while (p) { struct stream_peer *next = p->next; @@ -660,6 +772,8 @@ static void network_tick(void *arg) { if (p->idle && now - p->idle >= STREAM_LEASE_TB && !p->control->count) peer_free(p); p = next; } + pressure_tick(net, now_ms); + net->next_tick_ms = get_time_tb() / 10 + STREAM_TICK_TB / 10; 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"); } @@ -672,6 +786,7 @@ int stream_network_init(struct UTUN_INSTANCE *inst, struct stream_network **out) 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; + stream_load_init(&net->load, get_time_tb() / 10); net->next_tick_ms = get_time_tb() / 10 + STREAM_TICK_TB / 10; 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; } @@ -789,5 +904,27 @@ int stream_receive_restart(struct STREAM_HANDLE *subscription) { 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; + *info = f->info; + info->metrics_valid = stream_metrics_poll((struct stream_metrics *)&f->metrics, get_time_tb() / 10, &info->metrics) == STREAM_OK; + if (f->peer && f->peer->requested.active) for (unsigned i = 0; i < f->peer->requested.count; i++) + if (f->peer->requested.cuts[i].handle == f->info.handle) info->close_requested = 1; + return STREAM_OK; +} + +/* Добавление/вытеснение учитывает только ещё не переданные блоки действующей downstream-выдачи. */ +void stream_network_cache_event(struct stream_network *net, const struct stream_descriptor *d, const uint8_t *wire, size_t length, int added) { + if (!net || net->stopping) return; + uint64_t seq = stream_read_be(wire + STREAM_BLOCK_SEQ_OFF, 8); uint8_t sub = wire[STREAM_BLOCK_SUB_OFF]; + 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_ACTIVE || f->subblock != sub || seq < f->info.next + || !same_stream(&f->descriptor, d)) continue; + load_account(f, 1, length, added, 0); + if (added) { + if (seq > f->pending_highest) f->pending_highest = seq; + stream_metrics_observe(&f->offered_metrics, seq, (uint16_t)stream_read_be(wire + STREAM_BLOCK_TIMESTAMP_OFF, 2), + wire[STREAM_BLOCK_FLAGS_OFF], (uint32_t)length, get_time_tb() / 10); + } + } + } }