115 lines
13 KiB
C
115 lines
13 KiB
C
/* Сетевые выдачи raw ETCP с собственным NCD-владением пира; reverse/терминальные уведомления через router.
|
||
* Одна управляющая FIFO и waiter на пира для всех групп/трансляций. DATA читается из кольца в момент отправки.
|
||
* STREAM_RECEIVE принадлежит вызывающему даже после отмены/stop; close снимает только данную выдачу.
|
||
* API выполняется в uasync-потоке. Внешние ручные операции требуют stream_subscribe_manual_open;
|
||
* автоматический выбор/перенос поставщиков владеет своими handles через отдельный планировщик.
|
||
* Лимиты:64 пира,256 flows всего,64 flows на пира,128 управляющих сообщений на пира; группы делят эти бюджеты.
|
||
* Control старше4с удаляется; совпавшие INFO/SUBSCRIBE/ACCEPT закрываются с OVERLOAD.
|
||
* INFO/connect ждёт до5с; downstream аренда6с продлевается каждые2с. Готовая INFO-проба требует явного close.
|
||
* После первого DATA тишину источника не считают истечением receive; восстановлением управляет selector/владелец. */
|
||
#ifndef STREAM_NETWORK_H
|
||
#define STREAM_NETWORK_H
|
||
|
||
#include "stream.h"
|
||
#include "stream_wire.h"
|
||
#include "stream_load.h"
|
||
|
||
#ifdef __cplusplus
|
||
extern "C" {
|
||
#endif
|
||
|
||
struct stream_network;
|
||
struct STREAM_RECEIVE;
|
||
enum stream_receive_state { STREAM_RECEIVE_CONNECTING, STREAM_RECEIVE_INFO, STREAM_RECEIVE_REQUESTED,
|
||
STREAM_RECEIVE_ACTIVE, STREAM_RECEIVE_ENDED, STREAM_RECEIVE_CLOSED };
|
||
struct stream_receive_info {
|
||
uint64_t handle, // ID DATA-запроса;0 до SUBSCRIBE и у INFO-only пробы.
|
||
peer, // ID непосредственного поставщика.
|
||
start, // Запрошенный, после ACCEPT подтверждённый старт; UINT64_MAX для запроса с конца.
|
||
next, // Следующий DATA-index после ACCEPT.
|
||
blocks, // Принятые RS-части, включая глобальные DUPLICATE/EXPIRED; не число проверенных FEC-пачек.
|
||
gaps, // Принятые сообщения GAP.
|
||
local_epoch, peer_epoch; // Поколения stream-peer, подтверждённые INFO_RESP; не source_epoch/ETCP reset_id.
|
||
enum stream_receive_state state; // CONNECTING → INFO → REQUESTED → ACTIVE → ENDED; CLOSED — также после detach.
|
||
int reason; // STREAM_REASON_* для CLOSED; последнее состояние/метаданные сохраняются до close.
|
||
struct stream_metric_snapshot metrics, supplier_metrics; // Собственный приём и последний снимок поставщика из INFO_RESP.
|
||
uint32_t rtt_ms; // INFO RTT от передачи в ETCP до ответа; очередь ETCP входит в измерение.
|
||
int metrics_valid, supplier_metrics_valid, rtt_valid, close_requested; // Validity снимков/RTT; selected в активном OVERLOAD.
|
||
struct stream_path path; // От автора до поставщика, без получателя; меняется при приходе DATA.
|
||
uint64_t first_available, last_available, // Включительные границы предложенного субпотока; UINT64_MAX/UINT64_MAX — пусто.
|
||
info_ms, info_order, // Время последнего INFO/ACCEPT в локальных мс; порядок принятого INFO_RESP в network.
|
||
supplier_queue_bytes, progress_age_ms; // Общая DATA-очередь поставщика к нам и возраст его прогресса субпотока.
|
||
uint64_t key_seq, key_last_seq; // Включительный диапазон последнего полного KEY, если key_valid=1
|
||
uint32_t path_rtt_ms; // RTT пути до поставщика, без последнего перехода к нам.
|
||
uint16_t free_slots, cpu_basis_points; // Свободные peer/global flow-слоты и CPU0..10000 при cpu_valid.
|
||
uint8_t subblock; // Индекс RS-части0..n-1.
|
||
int info_ready, probe_only, accepting, // Ответ получен; INFO-only; поставщик не в pressure/cooling (слоты проверяются отдельно).
|
||
path_rtt_valid, progress_valid, cpu_valid, key_valid; // Достоверность соответствующих полей предложения, не подписи пачки.
|
||
};
|
||
/* INFO без выделения remote handle и без DATA; close обязателен также после detach.
|
||
* out заранее NULL; STREAM_OK, WAIT для недоступного/неподписанного поставщика либо отрицательная ошибка.
|
||
* После info_ready можно тем же NCD-владением последовательно опросить остальные субпотоки.
|
||
* В той же peer/stream/sub позиции разрешена одна проба; незавершённый INFO DATA-запроса тоже занимает её. */
|
||
int stream_probe_open(struct STREAM_HANDLE *subscription, uint64_t peer, uint8_t subblock, struct STREAM_RECEIVE **out);
|
||
/* Повторить INFO готовой пробы для subblock; метаданные пригодны после нового info_ready.
|
||
* OK, WAIT при конфликте pending INFO, ERR_STATE для неверного handle/subblock. */
|
||
int stream_probe_next(struct STREAM_RECEIVE *probe, uint8_t subblock);
|
||
/* out заранее NULL; seq=UINT64_MAX: первая DATA-выдача с конца, остальные ждут общий anchor первой принятой части.
|
||
* Подпись всей пачки проверяется после RS decode; границы INFO сами не гарантируют наличие каждой части.
|
||
* peer должен быть online-мембером с подпиской либо подтверждённой relay-помощью.
|
||
* Второй DATA-flow peer+stream+sub запрещён; INFO-проба допускается после выделения DATA handle.
|
||
* OK — локальный handle создан, не ACCEPT; WAIT — поставщик недоступен, отрицательный результат — отказ. */
|
||
int stream_receive_open(struct STREAM_HANDLE *subscription, uint64_t peer, uint8_t subblock, uint64_t seq, struct STREAM_RECEIVE **out);
|
||
/* Освободить свой handle, в том числе после END/CLOSED/stop; NULL допустим. Общий NCD/ETCP не принадлежит этому handle. */
|
||
void stream_receive_close(struct STREAM_RECEIVE *receive);
|
||
/* Отменить текущие upstream данного локального экземпляра и сбросить anchor; новые выдачи открывает вызывающий.
|
||
* Буфер/подписка/downstream и нижняя граница consumer cursor сохраняются: уже выданные DATA не повторяются.
|
||
* Старые receive handles становятся CLOSED и всё равно требуют close. Только manual/internal subscription. */
|
||
int stream_receive_restart(struct STREAM_HANDLE *subscription);
|
||
/* Скопировать info даже после detach, обновив локальные метрики; OK/ERR_ARGUMENT.
|
||
* info_ready может сохраняться в CLOSED: живое состояние проверяется отдельно. */
|
||
int stream_receive_get_info(struct STREAM_RECEIVE *receive, struct stream_receive_info *info);
|
||
struct stream_peer_load_info {
|
||
struct stream_load_snapshot load; // Снимок rates последнего poll с текущими накопителями/остатком DATA.
|
||
// Локальный OVERLOAD: ID/требуемый и освобождённый вклад/срок; затем общие ETCP-очереди и неподтверждённые байты.
|
||
uint64_t request_id, required_rate, released_rate, deadline_ms, transport_queued_bytes, inflight_bytes;
|
||
uint32_t transport_queued_packets; // send_input_q + input_queue; bytes включает также normalizer через etcp_tx_queued.
|
||
double cpu_percent; // Средняя загрузка uasync либо0 при неизвестной оценке.
|
||
uint64_t callback_late_ms; // Опоздание общего100-мс network tick.
|
||
unsigned required_count, released_count; // Расчётное и фактическое число снимаемых downstream.
|
||
int global_pressure, request_active, accepting; // Перегрузка модуля; активный запрос пира; нет peer pressure/cooling.
|
||
};
|
||
/* Только существующий stream-peer; OK/ERR_STATE, выход не меняется при отказе. ETCP-статистика общая с другими сервисами. */
|
||
int stream_peer_get_load(struct UTUN_INSTANCE *inst, uint64_t peer, struct stream_peer_load_info *info);
|
||
/* Оставшиеся RS-части всех локальных выдач точной публикации; принятые ETCP части сюда не входят. */
|
||
uint64_t stream_network_pending_bytes(struct UTUN_INSTANCE *inst, const struct stream_descriptor *descriptor);
|
||
/* Снимок позиций downstream в кольце точной эпохи, худшие получатели первыми. Только uasync.
|
||
* out обязателен; STREAM_OK либо ошибка аргументов/отсутствия локальной подписки. Кольцо не сканируется. */
|
||
int stream_network_buffer_positions(struct UTUN_INSTANCE *inst, const struct stream_descriptor *descriptor,
|
||
struct stream_buffer_positions *out);
|
||
/* Худший целый входящий путь по сумме известных длительностей, без склейки разных ветвей.
|
||
* После разрыва используется сохранённая телеметрия; при её отсутствии OK с count0. Полный descriptor обязателен.
|
||
* Возраст/STALE/DOWN сохраняются; ERR_STATE не меняет out. */
|
||
int stream_network_telemetry(struct UTUN_INSTANCE *inst, const struct stream_descriptor *descriptor, struct stream_telemetry *out);
|
||
|
||
/* Внутренний lifecycle: занять свободные raw/router bindings, подписку REINIT и timer100мс; out заранее NULL.
|
||
* OK либо ERR_STATE/MEMORY/CRYPTO. Отказ снимает свои ресурсы; descriptor/flow здесь ещё нет. */
|
||
int stream_network_init(struct UTUN_INSTANCE *inst, struct stream_network **out);
|
||
/* Вне callbacks/pump: снять bindings/REINIT/timer/waiters/NCD, отсоединить caller-owned receives. NULL допустим. */
|
||
void stream_network_destroy(struct stream_network *network);
|
||
/* Разбудить peer pumps после изменения кольца/END; не выполняет отправку синхронно. NULL/stopping игнорируются. */
|
||
void stream_network_changed(struct stream_network *network);
|
||
/* Объединённый отложенный проход управления; не запускает selector внутри packet/NCD callbacks. */
|
||
void stream_network_wake(struct stream_network *network);
|
||
/* Закрыть flows/relay-запросы точной group/source/epoch до освобождения её кольца; чужие группы/ETCP остаются. */
|
||
void stream_network_cancel(struct stream_network *network, const struct stream_descriptor *descriptor);
|
||
/* После изменения принятого профиля self/пира закрыть недопущенные flows указанного канала; кольцо не освобождает. */
|
||
void stream_network_membership_changed(struct stream_network *network, uint64_t node_id, const char *channel);
|
||
/* Observer кольца: учесть добавление/вытеснение части ещё не переданной downstream; DATA не копируется/не отправляется. */
|
||
void stream_network_cache_event(struct stream_network *network, const struct stream_descriptor *descriptor,
|
||
const uint8_t *wire, size_t length, int added);
|
||
#ifdef __cplusplus
|
||
}
|
||
#endif
|
||
#endif
|