diff --git a/src/stream/stream_network.h b/src/stream/stream_network.h index e45c17e2..7aac4686 100644 --- a/src/stream/stream_network.h +++ b/src/stream/stream_network.h @@ -7,6 +7,7 @@ #include "stream.h" #include "stream_wire.h" +#include "stream_load.h" struct stream_network; struct STREAM_RECEIVE; @@ -16,6 +17,9 @@ struct stream_receive_info { uint64_t handle, peer, start, next, blocks, gaps; enum stream_receive_state state; int reason; + struct stream_metric_snapshot metrics, supplier_metrics; + uint32_t rtt_ms; + int metrics_valid, supplier_metrics_valid, rtt_valid, close_requested; }; /* out заранее NULL; seq=UINT64_MAX: первая выдача с конца, остальные ждут её проверенный общий старт. * peer должен быть онлайн-кандидатом группы. Повтор peer+stream+sub не создаёт вторую выдачу. */ @@ -25,10 +29,22 @@ void stream_receive_close(struct STREAM_RECEIVE *receive); * Буфер/подписка/downstream сохраняются; старые receive handles становятся CLOSED и всё равно требуют close. */ int stream_receive_restart(struct STREAM_HANDLE *subscription); int stream_receive_get_info(const struct STREAM_RECEIVE *receive, struct stream_receive_info *info); +struct stream_peer_load_info { + struct stream_load_snapshot load; + uint64_t request_id, required_rate, released_rate, deadline_ms, transport_queued_bytes, inflight_bytes; + uint32_t transport_queued_packets; + double cpu_percent; + uint64_t callback_late_ms; + unsigned required_count, released_count; + int global_pressure, request_active; +}; +int stream_peer_get_load(struct UTUN_INSTANCE *inst, uint64_t peer, struct stream_peer_load_info *info); int stream_network_init(struct UTUN_INSTANCE *inst, struct stream_network **out); void stream_network_destroy(struct stream_network *network); void stream_network_changed(struct stream_network *network); void stream_network_cancel(struct stream_network *network, const struct stream_descriptor *descriptor); void stream_network_membership_changed(struct stream_network *network, uint64_t node_id, const char *channel); +void stream_network_cache_event(struct stream_network *network, const struct stream_descriptor *descriptor, + const uint8_t *wire, size_t length, int added); #endif diff --git a/src/stream/stream_wire.c b/src/stream/stream_wire.c index ae9d264d..708d44eb 100644 --- a/src/stream/stream_wire.c +++ b/src/stream/stream_wire.c @@ -7,7 +7,7 @@ /* Заголовок одинаков у raw и router payload; cmd raw сохраняется внутри routed-кодограммы. */ int stream_wire_parse(const uint8_t *wire, size_t length, struct stream_message *m) { if (!wire || !m || length < STREAM_WIRE_HEADER || wire[0] != ETCP_ID_STREAM || wire[1] != STREAM_WIRE_VERSION - || wire[2] < STREAM_CMD_INFO_REQ || wire[2] > STREAM_CMD_REVERSE_REQ) { + || wire[2] < STREAM_CMD_INFO_REQ || wire[2] > STREAM_CMD_OVERLOAD_SELECT) { DEBUG_WARN(DEBUG_CATEGORY_STREAM, "stream wire header invalid bytes=%zu", length); return STREAM_ERR_FORMAT; } struct stream_message result = {.command = wire[2], .subblock = wire[3], .payload = wire + STREAM_WIRE_HEADER, @@ -60,6 +60,25 @@ size_t stream_path_write(uint8_t *wire, const struct stream_path *path) { return 1 + path->count * 8; } +/* Неизвестное измерение явно отличается от нулевого lag/rate; числа не сужаются до старых uint16 RTT. */ +void stream_metric_write(uint8_t *wire, const struct stream_metric_snapshot *s, int valid) { + wire[0] = valid ? 1 | (s->source_rate_valid ? 2 : 0) | (s->receive_rate_valid ? 4 : 0) | (s->stopped ? 8 : 0) : 0; + stream_write_be(wire + 1, s->lag_ms, 8); stream_write_be(wire + 9, s->source_rate_bytes_sec, 8); + stream_write_be(wire + 17, s->receive_rate_bytes_sec, 8); stream_write_be(wire + 25, s->blocks, 8); + stream_write_be(wire + 33, s->late_blocks, 8); stream_write_be(wire + 41, s->stops, 8); +} + +int stream_metric_read(const uint8_t *wire, struct stream_metric_snapshot *s, int *valid) { + if (wire[0] & ~15u || (!(wire[0] & 1) && wire[0])) { + DEBUG_WARN(DEBUG_CATEGORY_STREAM, "stream metric flags invalid=%u", wire[0]); return STREAM_ERR_FORMAT; + } + *s = (struct stream_metric_snapshot){.lag_ms = stream_read_be(wire + 1, 8), .source_rate_bytes_sec = stream_read_be(wire + 9, 8), + .receive_rate_bytes_sec = stream_read_be(wire + 17, 8), .blocks = stream_read_be(wire + 25, 8), + .late_blocks = stream_read_be(wire + 33, 8), .stops = stream_read_be(wire + 41, 8), + .source_rate_valid = !!(wire[0] & 2), .receive_rate_valid = !!(wire[0] & 4), .stopped = !!(wire[0] & 8)}; + *valid = !!(wire[0] & 1); return STREAM_OK; +} + /* END имеет собственный domain и подписывает ключ запуска + последний индекс; UINT64_MAX означает пустой поток. */ int stream_end_sign(const struct stream_descriptor *d, uint64_t last, const uint8_t key[32], uint8_t wire[STREAM_END_SIZE]) { if (!d || !key || !wire) { DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "stream END sign invalid arguments"); return STREAM_ERR_ARGUMENT; } diff --git a/src/stream/stream_wire.h b/src/stream/stream_wire.h index bbfdf391..5e67330c 100644 --- a/src/stream/stream_wire.h +++ b/src/stream/stream_wire.h @@ -4,22 +4,28 @@ #define STREAM_WIRE_H #include "stream_proto.h" +#include "stream_metrics.h" -#define STREAM_WIRE_VERSION 1 +#define STREAM_WIRE_VERSION 2 #define STREAM_WIRE_HEADER 52 #define STREAM_PATH_MAX 32 #define STREAM_END_SIZE 106 -#define STREAM_INFO_SIZE 29 /* actual:8, first:8, last:8, n:1, r:1, block_size:2, path_count:1 */ +#define STREAM_METRIC_WIRE_SIZE 49 +#define STREAM_INFO_METRIC_OFF 36 +#define STREAM_INFO_PATH_OFF (STREAM_INFO_METRIC_OFF + STREAM_METRIC_WIRE_SIZE) +#define STREAM_INFO_SIZE (STREAM_INFO_PATH_OFF + 1) /* probe:8, границы/FEC:28, метрики:49, count:1 */ +#define STREAM_OVERLOAD_MAX 64 +#define STREAM_OVERLOAD_HEADER 18 enum stream_command { STREAM_CMD_INFO_REQ = 1, STREAM_CMD_INFO_RESP, STREAM_CMD_SUBSCRIBE, STREAM_CMD_ACCEPT, STREAM_CMD_REJECT, STREAM_CMD_RENEW, STREAM_CMD_UNSUBSCRIBE, STREAM_CMD_CLOSED, - STREAM_CMD_DATA, STREAM_CMD_GAP, STREAM_CMD_END, STREAM_CMD_REVERSE_REQ + STREAM_CMD_DATA, STREAM_CMD_GAP, STREAM_CMD_END, STREAM_CMD_REVERSE_REQ, STREAM_CMD_OVERLOAD, STREAM_CMD_OVERLOAD_SELECT }; enum stream_reject { STREAM_REASON_STOPPED = 1, STREAM_REASON_POSITION, STREAM_REASON_EXPIRED, STREAM_REASON_STALE, STREAM_REASON_CONFLICT, STREAM_REASON_LIMIT, STREAM_REASON_LOOP, STREAM_REASON_TRANSPORT, - STREAM_REASON_LEASE, STREAM_REASON_MEMBER, STREAM_REASON_PROTOCOL + STREAM_REASON_LEASE, STREAM_REASON_MEMBER, STREAM_REASON_PROTOCOL, STREAM_REASON_OVERLOAD }; struct stream_path { uint64_t nodes[STREAM_PATH_MAX]; uint8_t count; }; struct stream_message { @@ -37,6 +43,8 @@ void stream_wire_header(uint8_t *wire, const struct stream_message *message); int stream_path_validate(const struct stream_path *path, uint64_t source, uint64_t sender, uint64_t forbidden); int stream_path_parse(const uint8_t *wire, size_t length, struct stream_path *path, size_t *used); size_t stream_path_write(uint8_t *wire, const struct stream_path *path); +void stream_metric_write(uint8_t *wire, const struct stream_metric_snapshot *snapshot, int valid); +int stream_metric_read(const uint8_t *wire, struct stream_metric_snapshot *snapshot, int *valid); int stream_end_sign(const struct stream_descriptor *descriptor, uint64_t last, const uint8_t key[32], uint8_t wire[STREAM_END_SIZE]); int stream_end_verify(const struct stream_descriptor *descriptor, const uint8_t key[32], const uint8_t *wire, size_t length, uint64_t *last);