diff --git a/doc/member_files.txt b/doc/member_files.txt index 09e1852d..622f1d7c 100644 --- a/doc/member_files.txt +++ b/doc/member_files.txt @@ -37,11 +37,11 @@ src == owner. Удалённые записи и source=1 (плейсхолде Открытый fd/UUID загрузки не отменяют повторную проверку членства. Файлы удалённого участника сохраняются на диске, но через сервис не доступны. -Wire, версия 1 +Wire, версия 2 -------------- Все числа LE, без сырых структур и padding. Фиксированный заголовок 125 байт: - 0 protocol_version:u8 = 1 + 0 protocol_version:u8 = 2 1 kind:u8 (1=request, 2=response) 2 command:u8 3 status:u16 (request=0) @@ -70,7 +70,7 @@ Windows device names и символы \ : * ? " < > |. Корень внутр POSIX разрешает путь через *at/O_NOFOLLOW; regular-файлы с hardlink отвергаются. Windows удерживает родительские каталоги без SHARE_DELETE и не принимает reparse points. -Команды (1..10) +Команды --------------- 1 LIST: owner/path/cursor/version. Первая страница cursor=0, version=0. @@ -116,6 +116,26 @@ Windows удерживает родительские каталоги без SH существующего назначения. Активная загрузка в затрагиваемом пути — BUSY. 10 REMOVE: owner/path/version. Файл или пустой каталог, без рекурсивного удаления. +11 STREAM_WRITE: открытие потоковой загрузки; поля как BEGIN_WRITE, + cursor — ненулевой ID потока. Ответ подтверждает готовность temp и конечный размер. +12 STREAM_READ: открытие потокового скачивания; owner/path/version/size/cursor, + upload — случайный токен клиента. Сервер сверяет снимок и размер перед началом. +13 STREAM_DATA: request_id — ID потока, owner/path/upload, offset/size/data. + kind=request для upload, response для download. Ответа на блок нет. + offset обязан равняться следующему принятому смещению; size==data_len, не более 8192. +14 STREAM_END: ID потока, owner/path/upload, offset==size==размер файла, SHA256. + Upload: клиент отправляет END после всех DATA, сервер отвечает END после COMMIT. + Download: сервер отправляет END после DATA и повторной проверки версии; клиент + проверяет SHA256 записанных данных, синхронизирует и публикует локальный файл. + Ошибка сервера приходит END.status!=OK; ошибка открытия — ответом на STREAM_READ/WRITE. +15 STREAM_CANCEL: клиент отменяет конкретный поток. Поздние DATA не публикуются. + +Открытие потока — обычный повторяемый запрос с отдельным request_id. cursor связывает +его с DATA/END/CANCEL. Поток определяется group/peer/kind/ID; один ID не может +одновременно обозначать разные открытия. Каждый пакет по-прежнему подписан и зашифрован. +При закрытии/перезапуске router-канала его handles завершаются с IO_ERROR; +байты разных эпох не смешиваются. Возобновление через новую эпоху пока не реализовано. + Статусы (0..13): OK, ACCESS_DENIED, NOT_FOUND, INVALID_PATH, INVALID_RANGE, EXISTS, BUSY, STALE, NO_SPACE, IO_ERROR, BAD_PACKET, HASH_MISMATCH, CANCELLED, TIMEOUT. Мутации COMMIT/MKDIR/RENAME/REMOVE сохраняют успешный ответ на 120 секунд @@ -146,7 +166,29 @@ call_soon в тот же обработчик, без сетевого кана После restart незавершённые temp удаляются; опубликованные файлы остаются. Лимиты: 16 клиентских запросов, 32 серверных задания/ответа, 16 временных -загрузок, 4 общих transfer. Router-очередь обслуживается через backpressure. +загрузок, 16 клиентских и 16 серверных потоков на экземпляр. Каждый поток имеет +кольцо 32 x 8192 байт (256 КиБ); worker обрабатывает до 16 блоков за задачу. +Чтение следующей порции перекрывается с отправкой уже подготовленной. +Все серверные и локальные I/O используют один FIFO worker, без блокировки uasync. + +Отправитель добавляет пакеты, пока send_q router находится под нижним порогом +(сейчас count<=0), затем ждёт etcp_router_on_send_ready. Файловый слой не устанавливает +своё окно inflight и не повторяет DATA; транспортные ACK, retransmit и порядок — router. +Callback отправляет не более 8 блоков за запуск, уступая очередь другим потокам. +Параллельные файлы одного пира разделяют канал и FIFO/round-robin waiters. + +Готовность приёмника проверяется router до доставки следующего seq. Заполненное +кольцо или очередь заданий оставляют пакет в recv_q: rx_seq/ACK не продвигаются. +Освобождение места вызывает отложенный resume; pump отменяется при reset/close. +ACK RECEIVE_BUSY (flags bit0) раз в 100 мс отличает ожидание диска от потери транспорта. +Отправитель приостанавливает новые DATA, а живые busy ACK продлевают таймаут router. +Буфер router ограничен его receive window. Локальная доставка использует такую же +проверку готовности и ограниченный FIFO, а не сеть или отдельные подтверждения. + +При остановке сначала отменяются producers/requests и дожидаются все stream cleanup +continuations; только затем worker освобождает storage. Отмена после уже выполненной +публикации не удаляет опубликованный файл. + Transfer владеет CM handle до окончания/отмены; DOWN/TIMEOUT handle не освобождает. quiesce запрещает новые команды, отменяет таймеры, waiter и запросы; destroy вне callbacks дожидается только своих работ/completions и закрывает дисковый @@ -163,6 +205,27 @@ Upload/Download/New folder/Rename/Delete, прогресс и Cancel transfer. Open скачивает в выбранный пользователем локальный файл и открывает его приложением ОС после завершения. Смена контекста отбрасывает поздние ответы. +API параллельных передач +----------------------- + +member_files_transfer_start(request, callback, arg) копирует request и возвращает +отдельный mf_transfer* на каждый файл. upload задаёт направление, group/server/owner +выбираются для каждой операции. id=0 выбирает случайный ID; id в callback — фактический. +NULL означает отказ запуска без callback; причина пишется в лог. + +callback(arg, mf_transfer_progress*) получает id/done/total/status/finished. +Прогресс заимствован до возврата callback. done для отправителя означает байты, +принятые router в очередь; успешный finished подтверждает завершение операции, +для upload — публикацию на сервере, для download — публикацию локального файла. +finished приходит ровно один раз, handle к этому моменту уже освобождён; +вызывающий обнуляет свой указатель в callback. member_files_transfer_cancel(handle) +отменяет только этот файл, ждёт его I/O и ставит CANCEL в очередь с backpressure. +Stop сервиса/группы завершает все принадлежащие ему handles. + +Все API/callbacks выполняются в uasync-потоке экземпляра. GUI trampolines используют +тот же API и превращают callback в существующие события прогресса; Qt не делает I/O. +Один локальный файл назначения не может одновременно принадлежать двум download. + Проверки и диагностика --------------------- @@ -170,14 +233,20 @@ tests/test_member_files: wire/UTF-8, произвольные блоки, пов короткая замена, SHA256, пустой файл, квота, ссылки, expiry/restart, 10-секундный fd idle и принадлежность work/done потокам. tests/test_member_files_network: настоящие A/B/M, подписанная storage-роль, -upload/download, побайтная проверка, чужая запись, виртуальный корень, -подписанное удаление читателя. test_services — stop/start общего чат-сервиса. +параллельные upload/download одного пира, local loopback, пустые файлы, побайтная +проверка, отмена одного handle, restart router и destroy с активным server I/O. +Шестисекундная задержка worker проверяет receive backpressure и busy ACK. +Также проверяются чужая запись, виртуальный корень и подписанное удаление читателя. +test_services — stop/start общего чат-сервиса. Текущая диагностика выделена в категорию debug. Настройка для полного трасса: [debug] debug=trace debug=debug показывает запрос/ответ, ACL, очереди, маршрут, ACK, crypto-решения, I/O, fd idle и состояние браузера. trace дополнительно показывает каждый read/write и передачу задач/завершений между дисковым потоком и uasync. +stream progress раз в секунду показывает ID/направление, done/total, число блоков +в кольце, активность I/O, причину ожидания и скорость B/s. Router логирует переходы +receive paused/resumed и peer receive busy без повторного сообщения на каждый блок. По request_id видны GUI post → core request → router seq → server rx → ACL → disk queued/start/done → response tx → GUI result/bridge → directory ready. PID создания worker, PID отправки задачи и startup PID позволяют однозначно diff --git a/src/member_files/member_files.c b/src/member_files/member_files.c index 8d8e211c..45d8fc76 100644 --- a/src/member_files/member_files.c +++ b/src/member_files/member_files.c @@ -48,6 +48,35 @@ struct mf_local { static void mf_server_pump(struct member_files* f); static void mf_client_ready(struct ll_queue* q, void* arg); static void mf_dispatch(struct member_files* f, uint64_t group, uint64_t peer, const struct mf_message* message); +static int mf_message_ready(struct member_files* f, uint64_t group, uint64_t peer, int kind, int command, uint64_t id) { + if (!f || f->closing) return 0; + if (command == MF_STREAM_DATA) return mf_transfer_receive_ready(f, group, peer, kind, id); + return kind != MF_REQUEST || command > MF_STREAM_READ || f->job_count < MF_MAX_JOBS; +} +static int mf_receive_ready(struct UTUN_INSTANCE* inst, uint64_t group, uint64_t peer, const uint8_t* payload, size_t bytes) { + if (bytes < MF_WIRE_HEADER) return 1; // Ошибку формата логирует декодер сервиса. + return mf_message_ready(inst->member_files, group, peer, payload[1], payload[2], mf_get64(payload + 5)); +} +static int mf_local_ready(struct mf_local* local) { + struct mf_message* m = &local->message; + return mf_message_ready(local->files, local->group, local->files->inst->node_id, m->kind, m->command, m->request_id); +} + +static void mf_local_deliver(void* arg); + +void mf_receive_resume(struct member_files* f) { + if (f->closing) return; + for (struct ll_entry* e = f->inst->router_conns ? f->inst->router_conns->head : NULL; e; e = e->next) { + struct ETCP_ROUTER_CONN* rc = (struct ETCP_ROUTER_CONN*)e; + if (rc->svc_id == ETCP_RT_ID_MEMBER_FILES && rc->receive_blocked) + etcp_router_resume_receive(f->inst, rc->group_id, rc->remote_node_id, rc->svc_id); + } + struct mf_local* local = f->locals; + if (local && !local->task && mf_local_ready(local)) { + local->task = uasync_call_soon(f->inst->ua, local, mf_local_deliver); + if (!local->task) DEBUG_ERROR(DEBUG_CATEGORY_MEDIA, "member_files: local delivery scheduling failed"); + } +} static void mf_local_free(struct mf_local* local) { struct member_files* f = local->files; @@ -62,12 +91,14 @@ static void mf_local_free(struct mf_local* local) { static void mf_local_deliver(void* arg) { struct mf_local* local = arg; local->task = NULL; struct member_files* f = local->files; + if (!mf_local_ready(local)) return; struct mf_local** link = &f->locals; while (*link && *link != local) link = &(*link)->next; if (*link) *link = local->next; f->local_count--; if (!f->closing) mf_dispatch(f, local->group, f->inst->node_id, &local->message); u_free(local); + mf_receive_resume(f); mf_transfers_wake(f); } int member_files_member(struct UTUN_INSTANCE* inst, uint64_t group, uint64_t node, int storage) { @@ -98,33 +129,33 @@ int member_files_member(struct UTUN_INSTANCE* inst, uint64_t group, uint64_t nod sqlite3_finalize(st); return allowed; } -static int mf_server_allowed(struct member_files* f, uint64_t group, uint64_t peer, const struct mf_message* q) { +int mf_server_allowed(struct member_files* f, uint64_t group, uint64_t peer, const struct mf_message* q) { int enabled = chat_setting_get_int(f->inst, "member_files_enabled", 1); int allowed = enabled && member_files_member(f->inst, group, f->inst->node_id, 1) && member_files_member(f->inst, group, peer, 0) && (!q->owner || member_files_member(f->inst, group, q->owner, 0)) && - (q->command < MF_BEGIN_WRITE || (q->owner && q->owner == peer)); + (q->command < MF_BEGIN_WRITE || q->command == MF_STREAM_READ || (q->owner && q->owner == peer)); DEBUG_DEBUG(DEBUG_CATEGORY_DEBUG, "member_files: server access id=%llu cmd=%s enabled=%d allowed=%d owner_is_peer=%d", (unsigned long long)q->request_id, mf_command_text(q->command), enabled, allowed, q->owner == peer); return allowed; } /* Router подтверждает доставку отдельно; здесь отправка всегда соблюдает порог очереди. */ -static int mf_send(struct member_files* f, uint64_t group, uint64_t peer, const struct mf_message* m) { +int mf_send(struct member_files* f, uint64_t group, uint64_t peer, const struct mf_message* m) { if (peer == f->inst->node_id) { if (f->local_count >= MF_MAX_JOBS) return 1; struct mf_local* local = u_calloc(1, sizeof(*local)); if (!local) { DEBUG_ERROR(DEBUG_CATEGORY_MEDIA, "member_files: local delivery allocation failed"); return -1; } local->files = f; local->group = group; local->message = *m; - local->task = uasync_call_soon(f->inst->ua, local, mf_local_deliver); - if (!local->task) { DEBUG_ERROR(DEBUG_CATEGORY_MEDIA, "member_files: local delivery scheduling failed"); u_free(local); return -1; } - local->next = f->locals; f->locals = local; f->local_count++; + struct mf_local** tail = &f->locals; while (*tail) tail = &(*tail)->next; + *tail = local; f->local_count++; + mf_receive_resume(f); DEBUG_DEBUG(DEBUG_CATEGORY_DEBUG, "member_files: local queued id=%llu cmd=%s kind=%u count=%u", (unsigned long long)m->request_id, mf_command_text(m->command), m->kind, f->local_count); return 0; } struct ETCP_ROUTER_CONN* rc = etcp_router_conn_get(f->inst, group, peer, ETCP_RT_ID_MEMBER_FILES); if (!rc) { DEBUG_ERROR(DEBUG_CATEGORY_MEDIA, "member_files: router allocation failed"); return -1; } - queue_set_threshold(rc->send_q, 0, 0); etcp_router_set_max_inflight(rc, 16); + queue_set_threshold(rc->send_q, 0, 0); if (!etcp_router_send_q_has_room(f->inst, group, peer, ETCP_RT_ID_MEMBER_FILES)) { DEBUG_DEBUG(DEBUG_CATEGORY_DEBUG, "member_files: tx waiting id=%llu cmd=%s send_q=%u inflight=%u sync=%u no_route=%u", (unsigned long long)m->request_id, mf_command_text(m->command), rc->send_q->count, rc->inflight_q->count, @@ -190,7 +221,7 @@ static void mf_connected(struct CONN_MGR_HANDLE* h, uint64_t peer, uint64_t grou struct mf_pending* member_files_request(struct UTUN_INSTANCE* inst, uint64_t group, uint64_t server, const struct mf_message* request, mf_response_fn cb, void* arg) { struct member_files* f = inst ? inst->member_files : NULL; - if (request && (request->command < MF_LIST || request->command > MF_REMOVE || request->status || + if (request && (request->command < MF_LIST || request->command > MF_STREAM_READ || request->status || request->data_len > MF_BLOCK_SIZE || !mf_path_valid(request->path, request->command == MF_LIST || request->command == MF_STAT) || (request->command == MF_RENAME && !mf_path_valid(request->target, 0)))) { DEBUG_WARN(DEBUG_CATEGORY_MEDIA, "member_files: invalid client command=%u status=%u data=%u", @@ -242,11 +273,23 @@ static void mf_job_free(struct mf_job* j) { while (*link && *link != j) link = &(*link)->next; if (*link) *link = j->next; f->job_count--; u_free(j); + mf_receive_resume(f); } static void mf_server_deferred(void* arg) { struct member_files* f = arg; f->pump = NULL; mf_server_pump(f); } + +void mf_reply(struct member_files* f, uint64_t group, uint64_t peer, const struct mf_message* q, int status) { + struct mf_job* j = f->job_count < MF_MAX_JOBS ? u_calloc(1, sizeof(*j)) : NULL; + if (!j) { DEBUG_ERROR(DEBUG_CATEGORY_MEDIA, "member_files: reply queue allocation/limit failure status=%s", mf_status_text(status)); return; } + j->files = f; j->group = group; j->peer = peer; j->request = j->response = *q; + j->response.kind = MF_RESPONSE; j->response.status = (uint16_t)status; j->response.data_len = 0; + j->state = 2; j->created = get_time_tb(); + struct mf_job** tail = &f->jobs; while (*tail) tail = &(*tail)->next; + *tail = j; f->job_count++; + mf_server_pump(f); +} static void mf_server_ready(struct ll_queue* q, void* arg) { (void)q; struct member_files* f = arg; @@ -355,6 +398,8 @@ static void mf_server_pump(struct member_files* f) { static void mf_tick(void* arg) { struct member_files* f = arg; f->tick = NULL; uint64_t now = get_time_tb(); + mf_transfers_tick(f, now); + mf_receive_resume(f); for (;;) { struct mf_pending* expired = NULL; for (struct mf_pending* p = f->pending; p; p = p->next) if (now - p->started >= MF_REQUEST_TB) { expired = p; break; } @@ -384,6 +429,10 @@ static void mf_dispatch(struct member_files* f, uint64_t group, uint64_t peer, c DEBUG_DEBUG(DEBUG_CATEGORY_DEBUG, "member_files: rx id=%llu kind=%u cmd=%s group=%llx peer=%llx owner=%llx jobs=%u worker=%p", (unsigned long long)m.request_id, m.kind, mf_command_text(m.command), (unsigned long long)group, (unsigned long long)peer, (unsigned long long)m.owner, f->job_count, (void*)f->worker); + if (m.command >= MF_STREAM_DATA || (m.kind == MF_REQUEST && m.command >= MF_STREAM_WRITE)) { + mf_transfer_dispatch(f, group, peer, &m); + return; + } if (m.kind == MF_RESPONSE) { for (struct mf_pending* p = f->pending; p; p = p->next) { if (p->group != group || p->server != peer || p->request.request_id != m.request_id) continue; @@ -416,13 +465,14 @@ static void mf_dispatch(struct member_files* f, uint64_t group, uint64_t peer, c static void mf_recv(struct ETCP_CONN* conn, struct ll_entry* e) { struct member_files* f = conn ? conn->instance->member_files : NULL; - if (!f || f->closing || !e || !e->dgram || e->len <= ROUTER_SVC_PAYLOAD_OFF) { + if (!f || f->closing || !e || !e->dgram || e->len < ROUTER_SVC_PAYLOAD_OFF) { DEBUG_DEBUG(DEBUG_CATEGORY_DEBUG, "member_files: recv discarded context=%p closing=%d bytes=%u", (void*)f, f ? f->closing : 0, e ? e->len : 0); goto out; } uint64_t group, peer; memcpy(&group, e->dgram + ROUTER_SVC_GROUP_OFF, 8); memcpy(&peer, e->dgram + ROUTER_SVC_SRC_OFF, 8); + if (e->len == ROUTER_SVC_PAYLOAD_OFF) { mf_transfers_peer_down(f, group, peer); goto out; } if ((e->dgram[ROUTER_SVC_FLAGS_OFF] & (ROUTER_FLAG_SIGNED | ROUTER_FLAG_ENCRYPTED)) != (ROUTER_FLAG_SIGNED | ROUTER_FLAG_ENCRYPTED)) { DEBUG_WARN(DEBUG_CATEGORY_MEDIA, "member_files: unsigned/unencrypted packet peer=%llu", (unsigned long long)peer); goto out; @@ -452,6 +502,7 @@ int member_files_init(struct UTUN_INSTANCE* inst) { mf_storage_destroy(f->storage); mf_worker_destroy(f->disk); mf_worker_unreserve(f->cleanup); u_free(f); return -1; } inst->member_files = f; + etcp_router_set_receive_ready(inst, ETCP_RT_ID_MEMBER_FILES, mf_receive_ready); f->tick = uasync_set_timeout(inst->ua, 10000, f, mf_tick, "member_files"); if (!f->tick) { DEBUG_ERROR(DEBUG_CATEGORY_MEDIA, "member_files: init timer failed"); member_files_destroy(inst); return -1; } DEBUG_INFO(DEBUG_CATEGORY_MEDIA, "member_files: initialized root=%s idle=10s max_requests=%d max_jobs=%d", path, MF_MAX_PENDING, MF_MAX_JOBS); @@ -482,6 +533,8 @@ void member_files_destroy(struct UTUN_INSTANCE* inst) { struct member_files* f = inst ? inst->member_files : NULL; if (!f) return; member_files_quiesce(inst); + // Server streams закрывают temp/хеши в cleanup; storage должен оставаться жив до их завершения. + mf_worker_drain(f->disk); mf_worker_submit_reserved(f->disk, f->cleanup, mf_storage_cleanup, f->storage, mf_cleanup_done, NULL); f->cleanup = NULL; mf_worker_destroy(f->disk); diff --git a/src/member_files/member_files.h b/src/member_files/member_files.h index 0d26907e..f2e3033f 100644 --- a/src/member_files/member_files.h +++ b/src/member_files/member_files.h @@ -34,6 +34,16 @@ struct mf_transfer_request { uint8_t version[16]; char remote[MF_PATH_MAX + 1], local[MF_PATH_MAX + 1]; }; + +struct mf_transfer; +struct mf_transfer_progress { uint64_t id, done, total; uint16_t status; uint8_t finished; }; +typedef void (*mf_transfer_fn)(void* arg, const struct mf_transfer_progress* progress); +// Один handle на файл; до 16 клиентских передач одновременно, включая upload/download к одному пиру. +// Все вызовы/callbacks — uasync. id=0 выбирает случайный ID. NULL: отказ запуска, callback не вызывается. +// finished приходит ровно один раз после закрытия/публикации; handle уже освобождён до callback. +// Прогресс заимствован на время callback. Cancel асинхронно ждёт свой I/O и не затрагивает другие handles. +struct mf_transfer* member_files_transfer_start(const struct mf_transfer_request* request, mf_transfer_fn callback, void* arg); +void member_files_transfer_cancel(struct mf_transfer* transfer); /* id выбирает вызывающий. События: id/group/server/owner/done/total:8LE, status:2LE,state:1; * state=1 progress, 2 finished. Загрузка/скачивание выполняет общий клиент, без Qt I/O. */ void member_files_gui_transfer(void* arg); diff --git a/src/member_files/private.h b/src/member_files/private.h index 96cae114..62772656 100644 --- a/src/member_files/private.h +++ b/src/member_files/private.h @@ -21,8 +21,18 @@ struct member_files { void* tick; void* pump; unsigned pending_count, job_count, transfer_count; + unsigned server_transfer_count; unsigned local_count; int closing, used_storage, sweep_due; }; void mf_transfers_quiesce(struct member_files* files, uint64_t group); +int mf_transfer_receive_ready(struct member_files* files, uint64_t group, uint64_t peer, int kind, uint64_t id); +void mf_transfer_dispatch(struct member_files* files, uint64_t group, uint64_t peer, const struct mf_message* message); +void mf_transfers_peer_down(struct member_files* files, uint64_t group, uint64_t peer); +void mf_transfers_wake(struct member_files* files); +void mf_transfers_tick(struct member_files* files, uint64_t now); +int mf_send(struct member_files* files, uint64_t group, uint64_t peer, const struct mf_message* message); +int mf_server_allowed(struct member_files* files, uint64_t group, uint64_t peer, const struct mf_message* message); +void mf_receive_resume(struct member_files* files); +void mf_reply(struct member_files* files, uint64_t group, uint64_t peer, const struct mf_message* request, int status); #endif diff --git a/src/member_files/protocol.c b/src/member_files/protocol.c index 4725b1ea..b26dbee4 100644 --- a/src/member_files/protocol.c +++ b/src/member_files/protocol.c @@ -68,7 +68,7 @@ int mf_encode(const struct mf_message* m, uint8_t* p, size_t cap) { return -1; } memset(p, 0, MF_WIRE_HEADER); - p[0] = 1; p[1] = m->kind; p[2] = m->command; p[3] = (uint8_t)m->status; p[4] = (uint8_t)(m->status >> 8); + p[0] = 2; p[1] = m->kind; p[2] = m->command; p[3] = (uint8_t)m->status; p[4] = (uint8_t)(m->status >> 8); mf_put64(p + 5, m->request_id); mf_put64(p + 13, m->owner); mf_put64(p + 21, m->offset); mf_put64(p + 29, m->size); mf_put64(p + 37, m->cursor); memcpy(p + 45, m->version, 16); memcpy(p + 61, m->upload, 16); memcpy(p + 77, m->hash, 32); @@ -84,7 +84,7 @@ int mf_decode(const uint8_t* p, size_t n, struct mf_message* m) { if (!p || !m || n < MF_WIRE_HEADER) goto bad; size_t a = p[119] | ((size_t)p[120] << 8), b = p[121] | ((size_t)p[122] << 8); size_t d = p[123] | ((size_t)p[124] << 8); - if (p[0] != 1 || (p[1] != MF_REQUEST && p[1] != MF_RESPONSE) || p[2] < MF_LIST || p[2] > MF_REMOVE || + if (p[0] != 2 || (p[1] != MF_REQUEST && p[1] != MF_RESPONSE) || p[2] < MF_LIST || p[2] > MF_STREAM_CANCEL || a > MF_PATH_MAX || b > MF_PATH_MAX || d > MF_BLOCK_SIZE || n != MF_WIRE_HEADER + a + b + d || memchr(p + MF_WIRE_HEADER, 0, a + b) || p[118] > 1) goto bad; memset(m, 0, sizeof(*m)); @@ -110,6 +110,6 @@ const char* mf_status_text(int status) { const char* mf_command_text(int command) { static const char* commands[] = { "SWEEP", "LIST", "STAT", "READ", "BEGIN_WRITE", "WRITE", "COMMIT", "ABORT", - "MKDIR", "RENAME", "REMOVE" }; - return command >= 0 && command <= MF_REMOVE ? commands[command] : "UNKNOWN"; + "MKDIR", "RENAME", "REMOVE", "STREAM_WRITE", "STREAM_READ", "DATA", "END", "CANCEL" }; + return command >= 0 && command <= MF_STREAM_CANCEL ? commands[command] : "UNKNOWN"; } diff --git a/src/member_files/protocol.h b/src/member_files/protocol.h index 6c0073ec..63b2b37a 100644 --- a/src/member_files/protocol.h +++ b/src/member_files/protocol.h @@ -26,7 +26,10 @@ extern "C" { #define MF_FILE 1 #define MF_DIRECTORY 2 -enum mf_command { MF_LIST = 1, MF_STAT, MF_READ, MF_BEGIN_WRITE, MF_WRITE, MF_COMMIT, MF_ABORT, MF_MKDIR, MF_RENAME, MF_REMOVE }; +enum mf_command { + MF_LIST = 1, MF_STAT, MF_READ, MF_BEGIN_WRITE, MF_WRITE, MF_COMMIT, MF_ABORT, MF_MKDIR, MF_RENAME, MF_REMOVE, + MF_STREAM_WRITE, MF_STREAM_READ, MF_STREAM_DATA, MF_STREAM_END, MF_STREAM_CANCEL +}; enum mf_status { MF_OK, MF_ACCESS_DENIED, MF_NOT_FOUND, MF_INVALID_PATH, MF_INVALID_RANGE, MF_EXISTS, MF_BUSY, MF_STALE, MF_NO_SPACE, MF_IO_ERROR, MF_BAD_PACKET, MF_HASH_MISMATCH, MF_CANCELLED, MF_TIMEOUT diff --git a/src/member_files/transfer.c b/src/member_files/transfer.c index 337328bd..27bfc1b8 100644 --- a/src/member_files/transfer.c +++ b/src/member_files/transfer.c @@ -1,18 +1,25 @@ -/* Общий клиент: подготовка/локальный I/O в worker, блоки и события в uasync. - * Один transfer закрепляет CM на весь срок. Cancel ждёт своего worker и закрывает - * локальный файл до DONE; download публикуется лишь после последнего блока и sync. */ +/* Один handle на поток файла. Router управляет отправкой и подтверждениями; + * ограниченное кольцо связывает uasync с общим FIFO дискового worker. */ #include "private.h" #include "disk.h" #include "../utun_instance.h" #include "../chat/chat_event.h" #include "../routing_layer/conn_mgr.h" +#include "../routing_layer/etcp_router.h" #include "../../lib/mem.h" #include "../../lib/debug_config.h" #include +#include #include #include -enum transfer_work { TRANSFER_PREPARE, TRANSFER_READ, TRANSFER_WRITE, TRANSFER_FINISH }; +#define TRANSFER_SLOTS 32 +#define TRANSFER_BATCH 16 +#define TRANSFER_MAX 16 +#define TRANSFER_TIMEOUT_TB 1200000ULL + +enum transfer_work { TRANSFER_PREPARE, TRANSFER_READ, TRANSFER_WRITE, TRANSFER_VERIFY, TRANSFER_FINISH }; +struct mf_chunk { uint32_t bytes; uint8_t data[MF_BLOCK_SIZE]; }; struct mf_transfer { struct mf_transfer* next; struct member_files* files; @@ -20,229 +27,527 @@ struct mf_transfer { struct mf_pending* pending; struct CONN_MGR_HANDLE* cm; struct mf_work* cleanup; + struct queue_waiter_handle waiter; + void* pump; + mf_transfer_fn callback; + void* arg; + struct mf_message open, result, outgoing; struct mf_disk_info source; - int fd, busy, cancelled, status, work_status; + EVP_MD_CTX* digest; + int fd, server, sending, starting, active, opened, busy, stopping, finished_io, peer_down; + int status, work_status, reply_ready, reply_final, verified, committed, end_sent, end_received; enum transfer_work work; - uint64_t size, offset, last_event; - uint8_t upload[16], version[16], hash[32], block[MF_BLOCK_SIZE]; - uint32_t bytes; + uint64_t size, offset, received, read_offset, io_offset, io_bytes, io_size, progress_tb, last_event, last_log, logged_offset; + unsigned head, count, io_head, io_count; + uint8_t upload[16], version[16], hash[32]; char temp[MF_PATH_MAX + 80]; + struct mf_chunk chunks[TRANSFER_SLOTS]; }; +static void transfer_pump(void* arg); +static void transfer_submit(struct mf_transfer* t, enum transfer_work work); +static void transfer_stop(struct mf_transfer* t, int status); +static void transfer_send_open(struct mf_transfer* t, uint8_t command); + static void transfer_event(struct mf_transfer* t, int finished) { + if (t->server || !t->callback) return; uint64_t now = get_time_tb(); - if (!finished && t->last_event && now - t->last_event < 1000 && t->offset < t->size) return; + if (!finished && t->last_event && now - t->last_event < 1000) return; t->last_event = now; - uint8_t data[51]; - mf_put64(data, t->request.id); mf_put64(data + 8, t->request.group); mf_put64(data + 16, t->request.server); - mf_put64(data + 24, t->request.owner); mf_put64(data + 32, t->offset); mf_put64(data + 40, t->size); - data[48] = (uint8_t)t->status; data[49] = (uint8_t)(t->status >> 8); data[50] = finished ? 2 : 1; - chat_event_post(t->request.inst, CHAT_EVT_MEMBER_FILES_TRANSFER, data, sizeof(data)); + struct mf_transfer_progress progress = { t->request.id, t->offset, t->size, (uint16_t)t->status, (uint8_t)finished }; + t->callback(t->arg, &progress); } -static void transfer_send(struct mf_transfer* t, uint8_t command); -static void transfer_submit(struct mf_transfer* t, enum transfer_work work); +static void transfer_wake(struct mf_transfer* t) { + if (t->pump || t->waiter.internal || t->waiter.call_soon_id || t->files->closing) return; + t->pump = uasync_call_soon(t->files->inst->ua, t, transfer_pump); + if (!t->pump) DEBUG_ERROR(DEBUG_CATEGORY_MEDIA, "member_files: stream pump allocation failed id=%llu", (unsigned long long)t->request.id); +} +static void transfer_ready(struct ll_queue* q, void* arg) { (void)q; transfer_wake(arg); } -static void transfer_abort_done(void* arg, const struct mf_message* response) { - (void)arg; - if (response->status) DEBUG_WARN(DEBUG_CATEGORY_MEDIA, "member_files: abort not confirmed status=%s", mf_status_text(response->status)); +static struct mf_message transfer_packet(struct mf_transfer* t, uint8_t command) { + struct mf_message q = { .kind = t->server ? MF_RESPONSE : MF_REQUEST, .command = command, + .request_id = t->request.id, .owner = t->request.owner, .offset = t->offset, .size = t->size }; + snprintf(q.path, sizeof(q.path), "%s", t->request.remote); + memcpy(q.upload, t->upload, 16); memcpy(q.version, t->version, 16); memcpy(q.hash, t->hash, 32); + return q; } -/* Отмена сетевого request синхронно вызывает его callback. Worker завершается отдельно. */ -static void transfer_finish(struct mf_transfer* t, int status) { - if (status) t->status = status; - if (t->busy) return; - if (t->request.upload && t->status && !t->files->closing && !mf_zero(t->upload, 16)) { - struct mf_message q = { .command = MF_ABORT, .owner = t->request.owner }; - memcpy(q.upload, t->upload, 16); snprintf(q.path, sizeof(q.path), "%s", t->request.remote); - if (!member_files_request(t->request.inst, t->request.group, t->request.server, &q, transfer_abort_done, NULL)) - DEBUG_WARN(DEBUG_CATEGORY_MEDIA, "member_files: abort deferred to upload TTL"); - } - transfer_submit(t, TRANSFER_FINISH); +static struct mf_message transfer_storage_request(struct mf_transfer* t, uint8_t command) { + struct mf_message q = t->open; + q.command = command; q.kind = MF_REQUEST; q.data_len = 0; + return q; } -static void transfer_work(void* arg) { - struct mf_transfer* t = arg; - DEBUG_TRACE(DEBUG_CATEGORY_DEBUG, "member_files: local I/O start transfer=%llu work=%d fd=%d offset=%llu bytes=%u", - (unsigned long long)t->request.id, t->work, t->fd, (unsigned long long)t->offset, t->bytes); - t->work_status = MF_OK; - if (t->work == TRANSFER_PREPARE) { - if (t->request.upload) { - t->work_status = mf_disk_local_open(t->request.local, 0, &t->fd); - if (!t->work_status) t->work_status = mf_disk_info(t->fd, &t->source); - if (!t->work_status && (t->source.type != MF_FILE || t->source.size > INT64_MAX)) t->work_status = MF_INVALID_RANGE; - if (!t->work_status) { t->size = t->source.size; t->work_status = mf_disk_hash(t->fd, t->size, t->hash); } - struct mf_disk_info after; - if (!t->work_status) t->work_status = mf_disk_info(t->fd, &after); - if (!t->work_status && memcmp(&after, &t->source, sizeof(after))) t->work_status = MF_STALE; - } else t->work_status = mf_disk_local_open(t->temp, 1, &t->fd); - } else if (t->work == TRANSFER_READ) { - struct mf_disk_info info; - t->work_status = mf_disk_info(t->fd, &info); - if (!t->work_status && memcmp(&info, &t->source, sizeof(info))) t->work_status = MF_STALE; - t->bytes = (uint32_t)(t->size - t->offset < MF_BLOCK_SIZE ? t->size - t->offset : MF_BLOCK_SIZE); - if (!t->work_status) t->work_status = mf_disk_read(t->fd, t->offset, t->block, t->bytes); - } else if (t->work == TRANSFER_WRITE) t->work_status = mf_disk_write(t->fd, t->offset, t->block, t->bytes); - else { - if (t->fd >= 0) { - if (!t->status && !t->request.upload) t->work_status = mf_disk_sync(t->fd); - int closed = mf_disk_close(t->fd); t->fd = -1; - if (!t->work_status) t->work_status = closed; - } - if (!t->request.upload && *t->temp) { - if (!t->status && !t->work_status) t->work_status = mf_disk_local_publish(t->temp, t->request.local); - if (t->status || t->work_status) mf_disk_local_remove(t->temp); - } - } +static int transfer_allowed(struct mf_transfer* t) { + struct UTUN_INSTANCE* inst = t->files->inst; + if (!t->server) return member_files_member(inst, t->request.group, inst->node_id, 0) && + member_files_member(inst, t->request.group, t->request.server, 1); + struct mf_message q = t->open; + return mf_server_allowed(t->files, t->request.group, t->request.server, &q); } static void transfer_release(struct mf_transfer* t) { struct member_files* f = t->files; + if (t->pump) uasync_call_soon_cancel(f->inst->ua, t->pump); + etcp_router_cancel_send_ready(f->inst, t->request.group, t->request.server, ETCP_RT_ID_MEMBER_FILES, &t->waiter); if (t->cm) conn_mgr_close(t->cm); mf_worker_unreserve(t->cleanup); struct mf_transfer** link = &f->transfers; while (*link && *link != t) link = &(*link)->next; if (*link) *link = t->next; - f->transfer_count--; - transfer_event(t, 1); - if (t->status && t->status != MF_CANCELLED) DEBUG_WARN(DEBUG_CATEGORY_MEDIA, "member_files: transfer error id=%llu remote=%s local=%s reason=%s", - (unsigned long long)t->request.id, t->request.remote, t->request.local, mf_status_text(t->status)); - DEBUG_INFO(DEBUG_CATEGORY_MEDIA, "member_files: transfer finished id=%llu upload=%d bytes=%llu/%llu status=%s active=%u", - (unsigned long long)t->request.id, t->request.upload, (unsigned long long)t->offset, - (unsigned long long)t->size, mf_status_text(t->status), f->transfer_count); + if (t->server) f->server_transfer_count--; else f->transfer_count--; + DEBUG_INFO(DEBUG_CATEGORY_MEDIA, "member_files: stream finished id=%llu server=%d sending=%d bytes=%llu/%llu status=%s", + (unsigned long long)t->request.id, t->server, t->sending, (unsigned long long)t->offset, + (unsigned long long)t->size, mf_status_text(t->status)); + mf_transfer_fn callback = t->callback; void* arg = t->arg; + struct mf_transfer_progress progress = { t->request.id, t->offset, t->size, (uint16_t)t->status, 1 }; + int server = t->server; u_free(t); + mf_receive_resume(f); + if (!server && callback) callback(arg, &progress); +} + +/* Worker читает только снимок io_*; uasync не меняет занятые этим снимком ячейки. */ +static void transfer_work(void* arg) { + struct mf_transfer* t = arg; + int rc = MF_OK; + DEBUG_TRACE(DEBUG_CATEGORY_DEBUG, "member_files: stream I/O id=%llu work=%d offset=%llu slots=%u", + (unsigned long long)t->request.id, t->work, (unsigned long long)t->io_offset, t->io_count); + if (t->work == TRANSFER_PREPARE) { + if (t->server) { + struct mf_message q = t->open; + q.command = t->sending ? MF_STAT : MF_BEGIN_WRITE; + mf_storage_execute(t->files->storage, t->request.group, t->request.server, &q, &t->result); + rc = t->result.status; + if (!rc && t->sending && (t->result.type != MF_FILE || t->result.size != t->size || + memcmp(t->result.version, t->version, 16))) rc = MF_STALE; + } else if (t->sending) { + rc = mf_disk_local_open(t->request.local, 0, &t->fd); + if (!rc) rc = mf_disk_info(t->fd, &t->source); + if (!rc && (t->source.type != MF_FILE || t->source.size > INT64_MAX)) rc = MF_INVALID_RANGE; + if (!rc) { t->io_size = t->source.size; rc = mf_disk_hash(t->fd, t->io_size, t->hash); } + struct mf_disk_info after; + if (!rc) rc = mf_disk_info(t->fd, &after); + if (!rc && memcmp(&after, &t->source, sizeof(after))) rc = MF_STALE; + } else rc = mf_disk_local_open(t->temp, 1, &t->fd); + if (!rc && t->request.upload == 0) { + t->digest = EVP_MD_CTX_new(); + if (!t->digest || EVP_DigestInit_ex(t->digest, EVP_sha256(), NULL) != 1) rc = MF_IO_ERROR; + } + } else if (t->work == TRANSFER_READ || t->work == TRANSFER_WRITE) { + uint64_t offset = t->io_offset; t->io_bytes = 0; + struct mf_disk_info info; + if (!t->server && t->sending) { + rc = mf_disk_info(t->fd, &info); + if (!rc && memcmp(&info, &t->source, sizeof(info))) rc = MF_STALE; + } + for (unsigned i = 0; !rc && i < t->io_count; i++) { + struct mf_chunk* chunk = &t->chunks[(t->io_head + i) % TRANSFER_SLOTS]; + if (t->sending) chunk->bytes = (uint32_t)(t->size - offset < MF_BLOCK_SIZE ? t->size - offset : MF_BLOCK_SIZE); + if (t->server) { + struct mf_message q = transfer_storage_request(t, t->sending ? MF_READ : MF_WRITE); + q.kind = MF_REQUEST; q.offset = offset; q.size = chunk->bytes; + if (!t->sending) { q.data_len = chunk->bytes; memcpy(q.data, chunk->data, chunk->bytes); } + struct mf_message r; + mf_storage_execute(t->files->storage, t->request.group, t->request.server, &q, &r); + rc = r.status; + if (!rc && t->sending) { + if (r.data_len != chunk->bytes || r.size != t->size || memcmp(r.version, t->version, 16)) rc = MF_STALE; + else memcpy(chunk->data, r.data, r.data_len); + } + } else if (t->sending) rc = mf_disk_read(t->fd, offset, chunk->data, chunk->bytes); + else rc = mf_disk_write(t->fd, offset, chunk->data, chunk->bytes); + if (!rc && t->digest && EVP_DigestUpdate(t->digest, chunk->data, chunk->bytes) != 1) rc = MF_IO_ERROR; + if (!rc) { offset += chunk->bytes; t->io_bytes += chunk->bytes; } + } + } else if (t->work == TRANSFER_VERIFY) { + if (t->server && !t->sending) { + struct mf_message q = transfer_storage_request(t, MF_COMMIT); + mf_storage_execute(t->files->storage, t->request.group, t->request.server, &q, &t->result); + rc = t->result.status; + t->committed = rc == MF_OK; + } else { + if (t->digest) { + uint8_t hash[32]; unsigned bytes = 0; + if (EVP_DigestFinal_ex(t->digest, hash, &bytes) != 1 || bytes != 32) rc = MF_IO_ERROR; + else if (t->sending) memcpy(t->hash, hash, 32); + else if (memcmp(t->hash, hash, 32)) rc = MF_HASH_MISMATCH; + } + if (!rc && t->server) { + struct mf_message q = transfer_storage_request(t, MF_STAT); + mf_storage_execute(t->files->storage, t->request.group, t->request.server, &q, &t->result); + rc = t->result.status; + if (!rc && (t->result.size != t->size || memcmp(t->result.version, t->version, 16))) rc = MF_STALE; + } else if (!rc && t->sending) { + struct mf_disk_info after; rc = mf_disk_info(t->fd, &after); + if (!rc && memcmp(&after, &t->source, sizeof(after))) rc = MF_STALE; + } + } + } else { + if (t->server && !t->sending && t->status && !t->committed) { + struct mf_message q = transfer_storage_request(t, MF_ABORT); + mf_storage_execute(t->files->storage, t->request.group, t->request.server, &q, &t->result); + if (t->result.status) DEBUG_WARN(DEBUG_CATEGORY_MEDIA, "member_files: stream cleanup failed id=%llu status=%s", + (unsigned long long)t->request.id, mf_status_text(t->result.status)); + } + if (t->fd >= 0) { + if (!t->status && !t->sending) rc = mf_disk_sync(t->fd); + int closed = mf_disk_close(t->fd); t->fd = -1; + if (!rc) rc = closed; + } + if (!t->server && !t->sending) { + if (!t->status && !rc) rc = mf_disk_local_publish(t->temp, t->request.local); + if (t->status || rc) mf_disk_local_remove(t->temp); + } + EVP_MD_CTX_free(t->digest); t->digest = NULL; + } + t->work_status = rc; } static void transfer_done(void* arg, int error) { struct mf_transfer* t = arg; t->busy = 0; - DEBUG_DEBUG(DEBUG_CATEGORY_DEBUG, "member_files: local I/O done transfer=%llu work=%d status=%s error=%d cancelled=%d", - (unsigned long long)t->request.id, t->work, mf_status_text(t->work_status), error, t->cancelled); int status = error ? MF_CANCELLED : t->work_status; + DEBUG_TRACE(DEBUG_CATEGORY_DEBUG, "member_files: stream I/O done id=%llu work=%d status=%s bytes=%llu buffered=%u", + (unsigned long long)t->request.id, t->work, mf_status_text(status), (unsigned long long)t->io_bytes, t->count); if (t->work == TRANSFER_FINISH) { - /* При destroy work могла быть завершена, а completion отменён. fd уже закрыт. */ + t->finished_io = 1; if (!t->status) t->status = status; - transfer_release(t); return; + if (t->files->closing || t->peer_down || (!t->server && !t->status)) { transfer_release(t); return; } + if (!t->server) { + t->outgoing = transfer_packet(t, MF_STREAM_CANCEL); + t->reply_ready = t->reply_final = 1; transfer_wake(t); return; + } + t->outgoing = t->opened ? transfer_packet(t, MF_STREAM_END) : t->open; + t->outgoing.kind = MF_RESPONSE; t->outgoing.data_len = 0; t->outgoing.status = (uint16_t)t->status; + t->reply_ready = t->reply_final = 1; transfer_wake(t); return; } - if (t->cancelled || t->files->closing || status) { transfer_finish(t, status ? status : MF_CANCELLED); return; } + if (t->stopping || t->files->closing || status) { transfer_stop(t, status ? status : MF_CANCELLED); return; } + t->progress_tb = get_time_tb(); if (t->work == TRANSFER_PREPARE) { - transfer_event(t, 0); - if (t->request.upload) transfer_send(t, MF_BEGIN_WRITE); - else if (t->size) transfer_send(t, MF_READ); - else transfer_finish(t, MF_OK); - } else if (t->work == TRANSFER_READ) transfer_send(t, MF_WRITE); - else { - t->offset += t->bytes; transfer_event(t, 0); - if (t->offset == t->size) transfer_finish(t, MF_OK); - else transfer_send(t, MF_READ); + if (!t->server && t->sending) t->size = t->io_size; + if (t->server) { + t->outgoing = t->open; t->outgoing.kind = MF_RESPONSE; t->outgoing.data_len = 0; + t->outgoing.size = t->size; t->outgoing.status = MF_OK; + t->reply_ready = 1; + } else { transfer_send_open(t, t->sending ? MF_STREAM_WRITE : MF_STREAM_READ); return; } + } else if (t->work == TRANSFER_READ) { + t->read_offset += t->io_bytes; t->count += t->io_count; + } else if (t->work == TRANSFER_WRITE) { + t->offset += t->io_bytes; t->head = (t->head + t->io_count) % TRANSFER_SLOTS; t->count -= t->io_count; + mf_receive_resume(t->files); + } else { + t->verified = 1; + if (t->server || !t->sending) { transfer_stop(t, MF_OK); return; } } + transfer_wake(t); } static void transfer_submit(struct mf_transfer* t, enum transfer_work work) { t->work = work; t->busy = 1; if (work == TRANSFER_FINISH) { - struct mf_work* reserved = t->cleanup; t->cleanup = NULL; - mf_worker_submit_reserved(t->files->disk, reserved, transfer_work, t, transfer_done, t); + struct mf_work* cleanup = t->cleanup; t->cleanup = NULL; + mf_worker_submit_reserved(t->files->disk, cleanup, transfer_work, t, transfer_done, t); } else mf_worker_submit(t->files->disk, transfer_work, t, transfer_done, t); } +static void transfer_stop(struct mf_transfer* t, int status) { + if (t->stopping) { + if ((t->files->closing || status) && t->finished_io && !t->busy) transfer_release(t); + else if (!t->busy && !t->finished_io) transfer_submit(t, TRANSFER_FINISH); + return; + } + if (status && !t->status) { + t->status = status; + DEBUG_WARN(DEBUG_CATEGORY_MEDIA, "member_files: stream stopped id=%llu peer=%llx reason=%s", + (unsigned long long)t->request.id, (unsigned long long)t->request.server, mf_status_text(status)); + } + t->stopping = 1; + if (t->pump) { uasync_call_soon_cancel(t->files->inst->ua, t->pump); t->pump = NULL; } + etcp_router_cancel_send_ready(t->files->inst, t->request.group, t->request.server, ETCP_RT_ID_MEMBER_FILES, &t->waiter); + if (t->pending) { struct mf_pending* p = t->pending; t->pending = NULL; member_files_cancel(p); } + if (t->busy) return; + if (t->finished_io) { transfer_release(t); return; } + transfer_submit(t, TRANSFER_FINISH); +} + +static int transfer_enqueue(struct mf_transfer* t, struct mf_message* q) { + int rc = mf_send(t->files, t->request.group, t->request.server, q); + if (rc > 0 && t->request.server != t->files->inst->node_id) + etcp_router_on_send_ready(t->files->inst, t->request.group, t->request.server, ETCP_RT_ID_MEMBER_FILES, + &t->waiter, transfer_ready, t); + return rc; +} + +static void transfer_pump(void* arg) { + struct mf_transfer* t = arg; t->pump = NULL; + if (t->files->closing) return; + if (t->starting) { + t->starting = 0; + if (t->sending || t->server) transfer_submit(t, TRANSFER_PREPARE); + else transfer_send_open(t, MF_STAT); + return; + } + if (t->reply_ready) { + if (!t->status && !transfer_allowed(t)) { + if (!t->finished_io) { transfer_stop(t, MF_ACCESS_DENIED); return; } + t->status = t->outgoing.status = MF_ACCESS_DENIED; + DEBUG_WARN(DEBUG_CATEGORY_MEDIA, "member_files: final stream reply denied id=%llu", (unsigned long long)t->request.id); + } + int rc = transfer_enqueue(t, &t->outgoing); + if (rc < 0) { transfer_stop(t, MF_IO_ERROR); return; } + if (rc > 0) return; + t->reply_ready = 0; + if (t->reply_final) { transfer_release(t); return; } + t->opened = t->active = 1; + } + if (!t->active || t->stopping) return; + if (!transfer_allowed(t)) { transfer_stop(t, MF_ACCESS_DENIED); return; } + if (t->sending) { + // Ограничиваем работу callback, сохраняя fair scheduling нескольких файлов. + unsigned sent = 0; + while (t->count && sent++ < 8) { + struct mf_chunk* chunk = &t->chunks[t->head]; + struct mf_message q = transfer_packet(t, MF_STREAM_DATA); + q.size = q.data_len = chunk->bytes; memcpy(q.data, chunk->data, chunk->bytes); + int rc = transfer_enqueue(t, &q); + if (rc < 0) { transfer_stop(t, MF_IO_ERROR); return; } + if (rc > 0) break; + t->offset += chunk->bytes; t->head = (t->head + 1) % TRANSFER_SLOTS; t->count--; + t->progress_tb = get_time_tb(); + } + if (!t->busy && t->read_offset < t->size && t->count <= TRANSFER_SLOTS - TRANSFER_BATCH) { + uint64_t blocks = (t->size - t->read_offset + MF_BLOCK_SIZE - 1) / MF_BLOCK_SIZE; + t->io_count = blocks < TRANSFER_BATCH ? (unsigned)blocks : TRANSFER_BATCH; + t->io_head = (t->head + t->count) % TRANSFER_SLOTS; t->io_offset = t->read_offset; + transfer_submit(t, TRANSFER_READ); + } else if (!t->busy && !t->count && t->offset == t->size && !t->verified) transfer_submit(t, TRANSFER_VERIFY); + if (t->verified && !t->end_sent) { + struct mf_message q = transfer_packet(t, MF_STREAM_END); + int rc = transfer_enqueue(t, &q); + if (rc < 0) { transfer_stop(t, MF_IO_ERROR); return; } + if (!rc) { t->end_sent = 1; t->progress_tb = get_time_tb(); } + } + if (t->count) transfer_wake(t); + } else if (!t->busy) { + if (t->count) { + t->io_head = t->head; t->io_offset = t->offset; + t->io_count = t->count < TRANSFER_BATCH ? t->count : TRANSFER_BATCH; + transfer_submit(t, TRANSFER_WRITE); + } else if (t->end_received) transfer_submit(t, TRANSFER_VERIFY); + } + transfer_event(t, 0); +} + static void transfer_response(void* arg, const struct mf_message* r) { struct mf_transfer* t = arg; t->pending = NULL; - DEBUG_DEBUG(DEBUG_CATEGORY_DEBUG, "member_files: transfer response transfer=%llu request=%llu cmd=%s status=%s offset=%llu bytes=%u", - (unsigned long long)t->request.id, (unsigned long long)r->request_id, mf_command_text(r->command), mf_status_text(r->status), - (unsigned long long)r->offset, r->data_len); - if (t->cancelled || t->files->closing || r->status) { transfer_finish(t, r->status ? r->status : MF_CANCELLED); return; } + if (t->stopping) return; + if (r->status) { transfer_stop(t, r->status); return; } if (r->command == MF_STAT) { - if (r->type != MF_FILE || r->size > INT64_MAX || mf_zero(r->version, 16)) { transfer_finish(t, MF_INVALID_RANGE); return; } + if (r->type != MF_FILE || r->size > INT64_MAX || mf_zero(r->version, 16)) { transfer_stop(t, MF_INVALID_RANGE); return; } t->size = r->size; memcpy(t->version, r->version, 16); transfer_submit(t, TRANSFER_PREPARE); - } else if (r->command == MF_BEGIN_WRITE) { - if (t->size) transfer_submit(t, TRANSFER_READ); - else transfer_send(t, MF_COMMIT); - } else if (r->command == MF_WRITE) { - if (r->offset != t->offset || r->size != t->bytes) { transfer_finish(t, MF_BAD_PACKET); return; } - t->offset += t->bytes; transfer_event(t, 0); - if (t->offset == t->size) transfer_send(t, MF_COMMIT); - else transfer_submit(t, TRANSFER_READ); - } else if (r->command == MF_READ) { - uint64_t wanted = t->size - t->offset < MF_BLOCK_SIZE ? t->size - t->offset : MF_BLOCK_SIZE; - if (r->offset != t->offset || r->size != t->size || r->data_len != wanted || memcmp(r->version, t->version, 16)) { - transfer_finish(t, MF_STALE); return; - } - t->bytes = r->data_len; memcpy(t->block, r->data, r->data_len); transfer_submit(t, TRANSFER_WRITE); - } else if (r->command == MF_COMMIT) { - if (r->size != t->size || memcmp(r->hash, t->hash, 32)) { transfer_finish(t, MF_HASH_MISMATCH); return; } - transfer_finish(t, MF_OK); - } else transfer_finish(t, MF_BAD_PACKET); + } else if (r->size != t->size || r->cursor != t->request.id || memcmp(r->upload, t->upload, 16)) { + transfer_stop(t, MF_BAD_PACKET); + } else { t->opened = t->active = 1; t->progress_tb = get_time_tb(); transfer_wake(t); } } -static void transfer_send(struct mf_transfer* t, uint8_t command) { - struct mf_message q = { .command = command, .owner = t->request.owner, .offset = t->offset }; - snprintf(q.path, sizeof(q.path), "%s", t->request.remote); memcpy(q.upload, t->upload, 16); - if (command == MF_BEGIN_WRITE) { q.size = t->size; memcpy(q.hash, t->hash, 32); memcpy(q.version, t->request.version, 16); } - else if (command == MF_READ) { q.size = MF_BLOCK_SIZE; memcpy(q.version, t->version, 16); } - else if (command == MF_WRITE) { q.size = t->bytes; q.data_len = t->bytes; memcpy(q.data, t->block, t->bytes); } - t->pending = member_files_request(t->request.inst, t->request.group, t->request.server, &q, transfer_response, t); - if (!t->pending) transfer_finish(t, MF_BUSY); +static void transfer_send_open(struct mf_transfer* t, uint8_t command) { + struct mf_message q = transfer_packet(t, command); + q.request_id = 0; q.cursor = command == MF_STAT ? 0 : t->request.id; + if (command == MF_STREAM_WRITE) memcpy(q.version, t->request.version, 16); + t->pending = member_files_request(t->files->inst, t->request.group, t->request.server, &q, transfer_response, t); + if (!t->pending) transfer_stop(t, MF_BUSY); } static void transfer_connected(struct CONN_MGR_HANDLE* h, uint64_t peer, uint64_t group, enum conn_mgr_event event, void* arg) { (void)h; (void)peer; (void)group; struct mf_transfer* t = arg; - DEBUG_DEBUG(DEBUG_CATEGORY_DEBUG, "member_files: transfer CM id=%llu event=%d", (unsigned long long)t->request.id, event); + DEBUG_DEBUG(DEBUG_CATEGORY_DEBUG, "member_files: stream CM id=%llu event=%d", (unsigned long long)t->request.id, event); + if (event == CONN_EVENT_UP) transfer_wake(t); } -void member_files_gui_transfer(void* arg) { - struct mf_transfer_request* q = arg; - struct member_files* f = q->inst ? q->inst->member_files : NULL; +static struct mf_transfer* transfer_create(struct member_files* f, const struct mf_transfer_request* q, int server) { struct mf_transfer* t = u_calloc(1, sizeof(*t)); - if (!t) { DEBUG_ERROR(DEBUG_CATEGORY_MEDIA, "member_files: transfer allocation failed"); u_free(q); return; } - t->request = *q; t->fd = -1; u_free(q); - if (!f || f->closing || f->transfer_count >= 4 || !t->request.id || !t->request.owner || - !mf_path_valid(t->request.remote, 0) || !*t->request.local || - (t->request.upload && t->request.owner != t->request.inst->node_id)) { - t->status = MF_ACCESS_DENIED; transfer_event(t, 1); u_free(t); return; + if (!t) { DEBUG_ERROR(DEBUG_CATEGORY_MEDIA, "member_files: stream allocation failed"); return NULL; } + t->cleanup = mf_worker_reserve(); + if (!t->cleanup) { u_free(t); return NULL; } + t->files = f; t->request = *q; t->fd = -1; t->server = server; t->sending = server ? !q->upload : q->upload; + t->progress_tb = t->last_log = get_time_tb(); + t->next = f->transfers; f->transfers = t; + if (server) f->server_transfer_count++; else f->transfer_count++; + return t; +} + +struct mf_transfer* member_files_transfer_start(const struct mf_transfer_request* q, mf_transfer_fn callback, void* arg) { + struct member_files* f = q && q->inst ? q->inst->member_files : NULL; + if (!f || f->closing || !callback || f->transfer_count >= TRANSFER_MAX || !q->group || !q->server || !q->owner || + !mf_path_valid(q->remote, 0) || !*q->local || strnlen(q->local, sizeof(q->local)) > MF_PATH_MAX || + (q->upload && q->owner != q->inst->node_id) || !member_files_member(q->inst, q->group, q->inst->node_id, 0) || + !member_files_member(q->inst, q->group, q->server, 1)) { + DEBUG_WARN(DEBUG_CATEGORY_MEDIA, "member_files: stream start denied/limit reached"); return NULL; } - for (struct mf_transfer* other = f->transfers; other; other = other->next) if (other->request.id == t->request.id) { - DEBUG_WARN(DEBUG_CATEGORY_MEDIA, "member_files: duplicate transfer id"); t->status = MF_BUSY; transfer_event(t, 1); u_free(t); return; + struct mf_transfer_request request = *q; + if (!request.id && (RAND_bytes((uint8_t*)&request.id, 8) != 1 || !request.id)) { + DEBUG_ERROR(DEBUG_CATEGORY_MEDIA, "member_files: stream ID random failed"); return NULL; } - t->cleanup = mf_worker_reserve(); - if (!t->cleanup || RAND_bytes(t->upload, 16) != 1) { - t->status = MF_IO_ERROR; transfer_event(t, 1); mf_worker_unreserve(t->cleanup); u_free(t); return; - } - snprintf(t->temp, sizeof(t->temp), "%s.utun-%016llx.part", t->request.local, (unsigned long long)t->request.id); - t->files = f; t->next = f->transfers; f->transfers = t; f->transfer_count++; - if (t->request.server != t->request.inst->node_id && conn_mgr_open(t->request.inst, t->request.group, t->request.server, - transfer_connected, t, &t->cm)) { transfer_finish(t, MF_IO_ERROR); return; } - DEBUG_INFO(DEBUG_CATEGORY_MEDIA, "member_files: transfer started id=%llu upload=%d remote=%s local=%s", - (unsigned long long)t->request.id, t->request.upload, t->request.remote, t->request.local); - transfer_event(t, 0); - if (t->request.upload) transfer_submit(t, TRANSFER_PREPARE); - else transfer_send(t, MF_STAT); + for (struct mf_transfer* other = f->transfers; other; other = other->next) if (!other->server && + (other->request.id == request.id || (!q->upload && !other->request.upload && !strcmp(other->request.local, q->local)))) { + DEBUG_WARN(DEBUG_CATEGORY_MEDIA, "member_files: duplicate stream ID/download destination"); return NULL; + } + struct mf_transfer* t = transfer_create(f, &request, 0); + if (!t) return NULL; + if (RAND_bytes(t->upload, 16) != 1) { DEBUG_ERROR(DEBUG_CATEGORY_MEDIA, "member_files: stream random failed"); goto failed; } + snprintf(t->temp, sizeof(t->temp), "%s.utun-%016llx.part", q->local, (unsigned long long)request.id); + t->starting = 1; + if (q->server != q->inst->node_id && conn_mgr_open(q->inst, q->group, q->server, transfer_connected, t, &t->cm)) { + DEBUG_ERROR(DEBUG_CATEGORY_MEDIA, "member_files: stream CM launch failed"); goto failed; + } + transfer_wake(t); + if (!t->pump) goto failed; + t->callback = callback; t->arg = arg; + DEBUG_INFO(DEBUG_CATEGORY_MEDIA, "member_files: stream started id=%llu upload=%d remote=%s local=%s active=%u", + (unsigned long long)request.id, q->upload, q->remote, q->local, f->transfer_count); + return t; +failed: + t->status = MF_IO_ERROR; transfer_release(t); return NULL; +} + +static struct mf_transfer* transfer_find(struct member_files* f, uint64_t group, uint64_t peer, uint64_t id, int server) { + for (struct mf_transfer* t = f->transfers; t; t = t->next) + if (t->server == server && t->request.group == group && t->request.server == peer && t->request.id == id) return t; + return NULL; +} + +void mf_transfer_dispatch(struct member_files* f, uint64_t group, uint64_t peer, const struct mf_message* q) { + int server = q->kind == MF_REQUEST; + uint64_t id = q->command <= MF_STREAM_READ ? q->cursor : q->request_id; + struct mf_transfer* t = transfer_find(f, group, peer, id, server); + if (q->command == MF_STREAM_WRITE || q->command == MF_STREAM_READ) { + if (t) { + if (memcmp(&t->open, q, sizeof(*q))) { DEBUG_WARN(DEBUG_CATEGORY_MEDIA, "member_files: conflicting stream open"); return; } + if (t->active && !t->stopping) { t->outgoing = t->open; t->outgoing.kind = MF_RESPONSE; t->reply_ready = 1; transfer_wake(t); } + return; + } + int status = !id || !q->owner || q->size > INT64_MAX || q->data_len || !mf_path_valid(q->path, 0) ? MF_BAD_PACKET : MF_OK; + if (!status && !mf_server_allowed(f, group, peer, q)) status = MF_ACCESS_DENIED; + if (!status && f->server_transfer_count >= TRANSFER_MAX) status = MF_BUSY; + if (status) { mf_reply(f, group, peer, q, status); return; } + struct mf_transfer_request request = { .inst = f->inst, .id = id, .group = group, .server = peer, + .owner = q->owner, .upload = q->command == MF_STREAM_WRITE }; + snprintf(request.remote, sizeof(request.remote), "%s", q->path); + t = transfer_create(f, &request, 1); + if (!t) { mf_reply(f, group, peer, q, MF_BUSY); return; } + t->open = *q; t->size = q->size; memcpy(t->upload, q->upload, 16); memcpy(t->version, q->version, 16); memcpy(t->hash, q->hash, 32); + f->used_storage = 1; + t->starting = 1; transfer_wake(t); + DEBUG_INFO(DEBUG_CATEGORY_MEDIA, "member_files: server stream opened id=%llu peer=%llx sending=%d bytes=%llu active=%u", + (unsigned long long)id, (unsigned long long)peer, t->sending, (unsigned long long)t->size, f->server_transfer_count); + return; + } + if (!t || t->stopping) { + DEBUG_TRACE(DEBUG_CATEGORY_DEBUG, "member_files: late stream packet id=%llu command=%s", + (unsigned long long)id, mf_command_text(q->command)); + return; + } + if (q->owner != t->request.owner || strcmp(q->path, t->request.remote) || memcmp(q->upload, t->upload, 16)) { + DEBUG_WARN(DEBUG_CATEGORY_MEDIA, "member_files: stream packet identity mismatch id=%llu", (unsigned long long)id); + transfer_stop(t, MF_BAD_PACKET); return; + } + if (q->status || q->command == MF_STREAM_CANCEL) { transfer_stop(t, q->status ? q->status : MF_CANCELLED); return; } + if (!transfer_allowed(t)) { transfer_stop(t, MF_ACCESS_DENIED); return; } + if (q->command == MF_STREAM_DATA) { + if (t->sending || !t->active || t->end_received || !q->data_len || q->size != q->data_len || + q->offset != t->received || q->data_len > t->size - t->received || t->count == TRANSFER_SLOTS) { + DEBUG_WARN(DEBUG_CATEGORY_MEDIA, "member_files: stream DATA invalid id=%llu offset=%llu expected=%llu bytes=%u buffered=%u", + (unsigned long long)id, (unsigned long long)q->offset, (unsigned long long)t->received, q->data_len, t->count); + transfer_stop(t, MF_BAD_PACKET); return; + } + struct mf_chunk* chunk = &t->chunks[(t->head + t->count) % TRANSFER_SLOTS]; + chunk->bytes = q->data_len; memcpy(chunk->data, q->data, q->data_len); t->count++; t->received += q->data_len; + } else if (q->command == MF_STREAM_END) { + if (q->data_len || q->size != t->size || q->offset != t->size || + (t->sending ? !t->end_sent : t->received != t->size)) { transfer_stop(t, MF_BAD_PACKET); return; } + if (t->request.upload && memcmp(t->hash, q->hash, 32)) { transfer_stop(t, MF_HASH_MISMATCH); return; } + memcpy(t->hash, q->hash, 32); t->end_received = 1; + if (t->sending) { transfer_stop(t, MF_OK); return; } + } + t->progress_tb = get_time_tb(); transfer_wake(t); +} + +int mf_transfer_receive_ready(struct member_files* f, uint64_t group, uint64_t peer, int kind, uint64_t id) { + struct mf_transfer* t = transfer_find(f, group, peer, id, kind == MF_REQUEST); + return !t || t->stopping || t->sending || t->count < TRANSFER_SLOTS; } -static void transfer_cancel(struct mf_transfer* t) { - if (t->cancelled || t->work == TRANSFER_FINISH) return; - t->cancelled = 1; t->status = MF_CANCELLED; - DEBUG_DEBUG(DEBUG_CATEGORY_DEBUG, "member_files: transfer cancel id=%llu busy=%d pending=%p work=%d", - (unsigned long long)t->request.id, t->busy, (void*)t->pending, t->work); - if (t->pending) member_files_cancel(t->pending); - else if (!t->busy) transfer_finish(t, MF_CANCELLED); +void mf_transfers_wake(struct member_files* f) { + if (!f->closing) for (struct mf_transfer* t = f->transfers; t; t = t->next) if (!t->stopping || t->reply_ready) transfer_wake(t); } +void mf_transfers_tick(struct member_files* f, uint64_t now) { + struct mf_transfer* t = f->transfers; + while (t) { + struct mf_transfer* next = t->next; + if (t->stopping && t->finished_io && now - t->progress_tb >= TRANSFER_TIMEOUT_TB) { + DEBUG_WARN(DEBUG_CATEGORY_MEDIA, "member_files: final/cancel delivery timeout id=%llu", (unsigned long long)t->request.id); + transfer_release(t); + } else if (!t->stopping && now - t->progress_tb >= TRANSFER_TIMEOUT_TB) transfer_stop(t, MF_TIMEOUT); + else if (!t->stopping && now - t->last_log >= 10000) { + DEBUG_DEBUG(DEBUG_CATEGORY_DEBUG, + "member_files: stream progress id=%llu server=%d sending=%d done=%llu/%llu buffered=%u io=%d wait=%s rate=%lluB/s", + (unsigned long long)t->request.id, t->server, t->sending, (unsigned long long)t->offset, + (unsigned long long)t->size, t->count, t->busy, t->waiter.internal ? "router" : t->busy ? "disk" : "peer", + (unsigned long long)((t->offset - t->logged_offset) * 10000 / (now - t->last_log))); + t->last_log = now; t->logged_offset = t->offset; + } + t = next; + } + mf_transfers_wake(f); +} + +void member_files_transfer_cancel(struct mf_transfer* t) { if (t && !t->stopping) transfer_stop(t, MF_CANCELLED); } + void mf_transfers_quiesce(struct member_files* f, uint64_t group) { struct mf_transfer* t = f->transfers; - while (t) { struct mf_transfer* next = t->next; if (!group || t->request.group == group) transfer_cancel(t); t = next; } + while (t) { + struct mf_transfer* next = t->next; + if (!group || t->request.group == group) { t->peer_down = 1; transfer_stop(t, MF_CANCELLED); } + t = next; + } +} + +void mf_transfers_peer_down(struct member_files* f, uint64_t group, uint64_t peer) { + struct mf_transfer* t = f->transfers; + while (t) { + struct mf_transfer* next = t->next; + if (t->request.group == group && t->request.server == peer) { t->peer_down = 1; transfer_stop(t, MF_IO_ERROR); } + t = next; + } +} + +static void transfer_gui_event(void* arg, const struct mf_transfer_progress* p) { + struct mf_transfer_request* q = arg; + uint8_t data[51]; + mf_put64(data, p->id); mf_put64(data + 8, q->group); mf_put64(data + 16, q->server); mf_put64(data + 24, q->owner); + mf_put64(data + 32, p->done); mf_put64(data + 40, p->total); + data[48] = (uint8_t)p->status; data[49] = (uint8_t)(p->status >> 8); data[50] = p->finished ? 2 : 1; + chat_event_post(q->inst, CHAT_EVT_MEMBER_FILES_TRANSFER, data, sizeof(data)); + if (p->finished) u_free(q); +} + +void member_files_gui_transfer(void* arg) { + struct mf_transfer_request* q = arg; + if (!member_files_transfer_start(q, transfer_gui_event, q)) { + struct mf_transfer_progress p = { .id = q->id, .status = MF_BUSY, .finished = 1 }; + transfer_gui_event(q, &p); + } } void member_files_gui_cancel(void* arg) { struct mf_cancel_request* q = arg; struct member_files* f = q->inst ? q->inst->member_files : NULL; - int found = 0; - if (f) for (struct mf_transfer* t = f->transfers; t; t = t->next) if (t->request.id == q->id) { - transfer_cancel(t); found = 1; break; - } - if (f && !found) { - /* Отмена браузерных LIST/STAT использует тот же глобально случайный ID. */ - member_files_cancel_request_id(q->inst, q->id); - } + struct mf_transfer* t = f ? f->transfers : NULL; + while (t && (t->server || t->request.id != q->id)) t = t->next; + if (t) member_files_transfer_cancel(t); + else member_files_cancel_request_id(q->inst, q->id); u_free(q); } diff --git a/src/member_files/worker.c b/src/member_files/worker.c index 852e016f..024bd9b7 100644 --- a/src/member_files/worker.c +++ b/src/member_files/worker.c @@ -122,10 +122,8 @@ void mf_worker_submit(struct mf_worker* w, mf_work_fn work, void* data, mf_done_ mf_worker_submit_reserved(w, j, work, data, done, arg); } -void mf_worker_destroy(struct mf_worker* w) { +void mf_worker_drain(struct mf_worker* w) { if (!w) return; - DEBUG_DEBUG(DEBUG_CATEGORY_DEBUG, "member_files: worker stop owner=%p current_pid=%ld creator_pid=%ld active=%u", - (void*)w, (long)getpid(), w->creator_pid, w->count); /* Не запускаем loop рекурсивно: завершаем только собственные callbacks. */ while (w->jobs) { struct mf_work* j = w->jobs; @@ -135,6 +133,13 @@ void mf_worker_destroy(struct mf_worker* w) { uasync_cancel_post(w->ua, j->completion); j->completion = NULL; work_finish(j); } +} + +void mf_worker_destroy(struct mf_worker* w) { + if (!w) return; + DEBUG_DEBUG(DEBUG_CATEGORY_DEBUG, "member_files: worker stop owner=%p current_pid=%ld creator_pid=%ld active=%u", + (void*)w, (long)getpid(), w->creator_pid, w->count); + mf_worker_drain(w); pthread_mutex_lock(&w->mutex); w->stopping = 1; pthread_cond_signal(&w->condition); pthread_mutex_unlock(&w->mutex); pthread_join(w->thread, NULL); pthread_cond_destroy(&w->condition); pthread_mutex_destroy(&w->mutex); diff --git a/src/member_files/worker.h b/src/member_files/worker.h index fe1480de..3b9fb0c2 100644 --- a/src/member_files/worker.h +++ b/src/member_files/worker.h @@ -15,5 +15,7 @@ void mf_worker_unreserve(struct mf_work* work); void mf_worker_submit(struct mf_worker* worker, mf_work_fn work, void* data, mf_done_fn done, void* arg); void mf_worker_submit_reserved(struct mf_worker* worker, struct mf_work* reserved, mf_work_fn work, void* data, mf_done_fn done, void* arg); +// После quiesce владельца, вне callbacks: дождаться работ и их cleanup continuations, не останавливая pthread. +void mf_worker_drain(struct mf_worker* worker); void mf_worker_destroy(struct mf_worker* worker); #endif diff --git a/src/routing_layer/etcp_router.c b/src/routing_layer/etcp_router.c index bfa0dee0..2c101dc6 100644 --- a/src/routing_layer/etcp_router.c +++ b/src/routing_layer/etcp_router.c @@ -60,7 +60,8 @@ static void router_close_finalize(void* arg); static void router_conn_free_queues(struct ETCP_ROUTER_CONN* rconn); static int router_conn_reset(struct ETCP_ROUTER_CONN* rconn); static void free_entry(struct ll_entry* e) { queue_dgram_free(e); queue_entry_free(e); } -static void router_handle_ack(struct UTUN_INSTANCE* inst, struct ETCP_ROUTER_CONN* rconn, uint32_t seq, uint64_t src_node_id, uint16_t ack_ts); +static void router_handle_ack(struct UTUN_INSTANCE* inst, struct ETCP_ROUTER_CONN* rconn, uint32_t seq, + uint64_t src_node_id, uint16_t ack_ts, int receive_busy); static void router_forward_transit(struct UTUN_INSTANCE* inst, struct ll_entry* entry, struct SVC_ROUTE_HDR* hdr); static void router_handle_data_packet(struct ETCP_ROUTER_CONN* rconn, struct ETCP_CONN* conn, uint32_t seq, const uint8_t* wire, size_t wire_len); static int router_retransmit_one(struct ETCP_ROUTER_CONN* rconn, struct ROUTER_INFLIGHT* inf); @@ -488,6 +489,7 @@ static void router_send_queue_free(void* arg) { queue_free(arg); } // Отменить таймеры/waiter и освободить все очереди rconn (общий код для close/restart). static void router_conn_free_queues(struct ETCP_ROUTER_CONN* rconn) { struct UTUN_INSTANCE* inst = rconn->inst; + if (rconn->receive_pump) { uasync_call_soon_cancel(inst->ua, rconn->receive_pump); rconn->receive_pump = NULL; } if (rconn->handshake_timer) { uasync_cancel_timeout(inst->ua, rconn->handshake_timer); rconn->handshake_timer = NULL; } if (rconn->ack_timer) { uasync_cancel_timeout(inst->ua, rconn->ack_timer); rconn->ack_timer = NULL; } if (rconn->idle_ack_timer) { uasync_cancel_timeout(inst->ua, rconn->idle_ack_timer); rconn->idle_ack_timer = NULL; } @@ -538,6 +540,8 @@ static int router_conn_reset(struct ETCP_ROUTER_CONN* rconn) { queue_set_callback(rconn->incoming_q, router_incoming_q_cb, rconn); queue_set_threshold(rconn->incoming_q, 0, 0); rconn->incoming_data_ready = 0; + rconn->receive_blocked = 0; + rconn->peer_receive_blocked = 0; rconn->recv_conn = NULL; rconn->rtt = 0; rconn->rtt_jitter = 0; @@ -605,6 +609,7 @@ static void router_send_kick(struct ETCP_ROUTER_CONN* rconn) { return; } int retry = rconn->retrans_seq != rconn->retrans_end; + if (rconn->peer_receive_blocked && !retry) return; if (!retry && queue_entry_count(rconn->send_q) == 0) return; if (!retry && queue_entry_count(rconn->inflight_q) >= (int)router_effective_max_inflight(rconn)) { rconn->send_blocked = 1; @@ -638,6 +643,7 @@ static void router_send_drain_cb(struct ll_queue* q, void* arg) { router_send_kick(rconn); return; } + if (rconn->peer_receive_blocked) return; if (queue_entry_count(rconn->inflight_q) >= (int)router_effective_max_inflight(rconn)) { rconn->send_blocked = 1; return; @@ -960,7 +966,7 @@ static void router_send_ack(struct ETCP_ROUTER_CONN* rconn) { hdr->src_node_id = inst->node_id; hdr->seq = rconn->rx_seq; hdr->svc_id = rconn->svc_id; - hdr->flags = 0; + hdr->flags = rconn->receive_blocked ? ROUTER_FLAG_RECEIVE_BUSY : 0; hdr->reset_id = rconn->reset_id; hdr->peer_reset_id = rconn->peer_reset_id; hdr->challenge = 0; @@ -998,7 +1004,7 @@ static void router_schedule_ack(struct ETCP_ROUTER_CONN* rconn) { uasync_cancel_timeout(rconn->inst->ua, rconn->idle_ack_timer); rconn->idle_ack_timer = NULL; } - if (rconn->rx_seq == rconn->last_sent_ack_seq) return; + if (rconn->rx_seq == rconn->last_sent_ack_seq && !rconn->receive_blocked) return; uint64_t now = get_time_tb(); uint64_t elapsed = now - rconn->last_ack_sent_tb; if (elapsed >= ROUTER_ACK_INTERVAL_TB) { @@ -1015,12 +1021,12 @@ static void router_ack_timer_cb(void* arg) { struct ETCP_ROUTER_CONN* rconn = (struct ETCP_ROUTER_CONN*)arg; if (rconn->closed) return; rconn->ack_timer = NULL; - if (rconn->rx_seq != rconn->last_sent_ack_seq) + if (rconn->rx_seq != rconn->last_sent_ack_seq || rconn->receive_blocked) router_ack_do_send(rconn); - if (rconn->rx_seq != rconn->last_sent_ack_seq) { + if (rconn->rx_seq != rconn->last_sent_ack_seq || rconn->receive_blocked) { if (!rconn->ack_timer) rconn->ack_timer = uasync_set_timeout(rconn->inst->ua, - ROUTER_ACK_INTERVAL_TB, rconn, router_ack_timer_cb, "router_ack"); + rconn->receive_blocked ? ROUTER_ACK_BUSY_TB : ROUTER_ACK_INTERVAL_TB, rconn, router_ack_timer_cb, "router_ack"); } else if (!rconn->idle_ack_timer) { rconn->idle_ack_timer = uasync_set_timeout(rconn->inst->ua, ROUTER_ACK_IDLE_TB, rconn, router_idle_ack_timer_cb, "router_idle_ack"); @@ -1105,6 +1111,21 @@ static void router_try_assembly(struct ETCP_ROUTER_CONN* rconn, struct ETCP_CONN while (1) { struct ll_entry* entry = queue_find_data_by_index(rconn->recv_q, &next); if (!entry) break; + int (*ready)(struct UTUN_INSTANCE*, uint64_t, uint64_t, const uint8_t*, size_t) = + rconn->inst->router_bindings.receive_ready[rconn->svc_id]; + if (ready && !ready(rconn->inst, rconn->group_id, rconn->remote_node_id, + entry->dgram + SVC_ROUTE_HDR_SIZE, entry->len - SVC_ROUTE_HDR_SIZE)) { + if (!rconn->receive_blocked) DEBUG_DEBUG(DEBUG_CATEGORY_ETCPROUTE, + "router receive paused: svc=%u peer=%llx next_seq=%u buffered=%u", rconn->svc_id, + (unsigned long long)rconn->remote_node_id, next, rconn->recv_q->count); + rconn->receive_blocked = 1; + if (!rconn->ack_timer) rconn->ack_timer = uasync_set_timeout(rconn->inst->ua, + ROUTER_ACK_BUSY_TB, rconn, router_ack_timer_cb, "router_receive_busy"); + return; + } + if (rconn->receive_blocked) DEBUG_DEBUG(DEBUG_CATEGORY_ETCPROUTE, + "router receive resumed: svc=%u peer=%llx next_seq=%u", rconn->svc_id, (unsigned long long)rconn->remote_node_id, next); + rconn->receive_blocked = 0; queue_remove_data(rconn->recv_q, entry); uint64_t peer_id = rconn->peer_reset_id; int ret = router_deliver(rconn, conn ? conn : loopback_conn(rconn->inst), entry->dgram, entry->len); @@ -1156,7 +1177,7 @@ static void router_incoming_q_cb(struct ll_queue* q, void* arg) { // Обработать ACK: обновить RTT/jitter, снять подтверждённые inflight-пакеты, при необходимости // возобновить отправку (снять send_blocked). static void router_handle_ack(struct UTUN_INSTANCE* inst, struct ETCP_ROUTER_CONN* rconn, - uint32_t seq, uint64_t src_node_id, uint16_t ack_ts) { + uint32_t seq, uint64_t src_node_id, uint16_t ack_ts, int receive_busy) { (void)src_node_id; if (!rconn) return; if ((int32_t)(seq - rconn->tx_sent) > 0) { @@ -1165,6 +1186,17 @@ static void router_handle_ack(struct UTUN_INSTANCE* inst, struct ETCP_ROUTER_CON rconn->c_stale_ack++; return; } + if ((int32_t)(seq - rconn->tx_acked) < 0) { rconn->c_stale_ack++; return; } + if (receive_busy != rconn->peer_receive_blocked) DEBUG_DEBUG(DEBUG_CATEGORY_ETCPROUTE, + "router peer receive busy=%d svc=%u peer=%llx ack=%u", receive_busy, rconn->svc_id, (unsigned long long)src_node_id, seq); + int resumed = rconn->peer_receive_blocked && !receive_busy; + rconn->peer_receive_blocked = (uint8_t)receive_busy; + if (receive_busy) { + rconn->last_ack_changed_tb = get_time_tb(); rconn->no_ack_count = 0; + rconn->retrans_seq = rconn->retrans_end; + router_cancel_send_waiter(rconn); + } + if (resumed) router_send_kick(rconn); if ((int32_t)(seq - rconn->tx_acked) <= 0) { rconn->c_stale_ack++; return; } if (ack_ts != 0) { uint16_t new_rtt = get_current_timestamp() - ack_ts; @@ -1276,6 +1308,12 @@ static void router_handle_data_packet(struct ETCP_ROUTER_CONN* rconn, struct ETC rconn->c_dup_dropped++; return; } + if (rconn->incoming_q->count + rconn->recv_q->count >= (unsigned)rconn->max_inflight + 1) { + DEBUG_WARN(DEBUG_CATEGORY_ETCPROUTE, "router receive window full: svc=%u peer=%llx next=%u seq=%u buffered=%u", + rconn->svc_id, (unsigned long long)rconn->remote_node_id, rconn->rx_seq, seq, + rconn->incoming_q->count + rconn->recv_q->count); + return; + } struct ll_entry* qe = queue_entry_new(4); if (!qe) return; *(uint32_t*)qe->data = seq; @@ -1331,7 +1369,7 @@ static void etcp_router_recv_cb(struct ETCP_CONN* conn, struct ll_entry* entry) size_t payload_len = entry->len - SVC_ROUTE_HDR_SIZE; if (!h->reset_id || (payload_len && (control || h->challenge)) || (!payload_len && control != 0 && control != ROUTER_FLAG_START && control != ROUTER_FLAG_RST - && control != (ROUTER_FLAG_START | ROUTER_FLAG_RST) && control != ROUTER_FLAG_CLOSE)) { + && control != (ROUTER_FLAG_START | ROUTER_FLAG_RST) && control != ROUTER_FLAG_CLOSE && control != ROUTER_FLAG_RECEIVE_BUSY)) { DEBUG_WARN(DEBUG_CATEGORY_ETCPROUTE, "router recv: malformed flags=%02x seq=%u len=%zu epoch=%016llx", h->flags, h->seq, payload_len, (unsigned long long)h->reset_id); free_entry(entry); return; @@ -1362,7 +1400,7 @@ static void etcp_router_recv_cb(struct ETCP_CONN* conn, struct ll_entry* entry) } else if (!payload_len) { MF_ROUTE_DEBUG(h->svc_id, "ACK rx peer=%llx next_seq=%u old_acked=%u sent=%u inflight=%d", (unsigned long long)h->src_node_id, h->seq, rconn->tx_acked, rconn->tx_sent, queue_entry_count(rconn->inflight_q)); - if (!h->challenge) router_handle_ack(inst, rconn, h->seq, h->src_node_id, h->timestamp); + if (!h->challenge) router_handle_ack(inst, rconn, h->seq, h->src_node_id, h->timestamp, control == ROUTER_FLAG_RECEIVE_BUSY); } else { rconn->last_recv_pkt_ts = h->timestamp; rconn->last_recv_pkt_local_tb = get_time_tb(); rconn->last_recv_updated = 1; @@ -1436,10 +1474,33 @@ int etcp_router_unbind(struct UTUN_INSTANCE* inst, uint8_t svc_id) { return -1; } inst->router_bindings.callbacks[svc_id] = NULL; + inst->router_bindings.receive_ready[svc_id] = NULL; DEBUG_INFO(DEBUG_CATEGORY_ETCPROUTE, "svc_id=%u", svc_id); return 0; } +void etcp_router_set_receive_ready(struct UTUN_INSTANCE* inst, uint8_t svc_id, + int (*ready)(struct UTUN_INSTANCE*, uint64_t, uint64_t, const uint8_t*, size_t)) { + if (!inst) { DEBUG_ERROR(DEBUG_CATEGORY_ETCPROUTE, "router receive readiness: NULL instance"); return; } + inst->router_bindings.receive_ready[svc_id] = ready; +} + +static void router_receive_pump(void* arg) { + struct ETCP_ROUTER_CONN* rconn = arg; + rconn->receive_pump = NULL; + if (rconn->closed) return; + // Физический recv_conn мог исчезнуть за время дискового I/O; сервис получает стабильный instance. + router_try_assembly(rconn, loopback_conn(rconn->inst)); + if (!rconn->closed) router_schedule_ack(rconn); +} + +void etcp_router_resume_receive(struct UTUN_INSTANCE* inst, uint64_t group, uint64_t peer, uint8_t svc_id) { + struct ETCP_ROUTER_CONN* rconn = inst ? router_conn_find(inst, group, peer, svc_id) : NULL; + if (!rconn || rconn->closed || !rconn->receive_blocked || rconn->receive_pump) return; + rconn->receive_pump = uasync_call_soon(inst->ua, rconn, router_receive_pump); + if (!rconn->receive_pump) DEBUG_ERROR(DEBUG_CATEGORY_ETCPROUTE, "router receive resume allocation failed svc=%u", svc_id); +} + // Отправить сервисный пакет. entry->dgram[0] = svc_id, остальное — payload. Берёт на себя // освобождение entry в любом исходе. dst==self — loopback-доставка. int etcp_route_send(struct UTUN_INSTANCE* inst, uint64_t group_id, uint64_t dst_node_id, struct ll_entry* entry, int force, int mode) { diff --git a/src/routing_layer/etcp_router.h b/src/routing_layer/etcp_router.h index 08cad314..21671fdd 100644 --- a/src/routing_layer/etcp_router.h +++ b/src/routing_layer/etcp_router.h @@ -77,6 +77,7 @@ struct SVC_ROUTE_HDR { #define ROUTER_FLAG_ENCRYPTED 0x04 // payload зашифрован AES-128-CCM (nonce+tag+crc) #define ROUTER_FLAG_SIGNED 0x08 // пакет содержит Ed25519 подпись (64 байта после payload) #define ROUTER_FLAG_CLOSE 0x02 // нормальное закрытие conn +#define ROUTER_FLAG_RECEIVE_BUSY 0x01 // ACK: сервис приостановил приём, seq не подтверждает отложенные пакеты // ==================================================================== // 2. Внутренние структуры и константы (etcp_router.c, control_server.c) @@ -164,6 +165,9 @@ struct ETCP_ROUTER_CONN { // Очередь приёма входящего трафика (между сетью и recv_q) struct ll_queue* incoming_q; // FIFO очередь входящих пакетов uint8_t incoming_data_ready; // 1 = данные в incoming_q и callback был вызван + uint8_t receive_blocked; // сервис временно не принимает следующую кодограмму + uint8_t peer_receive_blocked; // последний ACK пира объявил паузу приёма + void* receive_pump; // отложенное продолжение приёма uint32_t c_pkts_sent; // успешные отправки данных uint32_t c_pkts_send_err; // ошибки отправки @@ -182,6 +186,7 @@ struct ETCP_ROUTER_CONN { #define ROUTER_MAX_INFLIGHT 256 // макс пакетов в полёте (для контроля inflight) #define ROUTER_ACK_INTERVAL_TB 100 // интервал ACK: 10ms в timebase (0.1ms) #define ROUTER_ACK_IDLE_TB 5000 // idle таймаут: 500ms +#define ROUTER_ACK_BUSY_TB 1000 // 100ms: живой приёмник ждёт диск, а не потерял транспорт #define ROUTER_MAX_SEND_Q_PACKETS 64 // порог backpressure на send_q #define ROUTER_SEND_WATCHDOG_TB 5000 // DEBUG-watchdog: 500ms проверка инварианта drain @@ -212,6 +217,7 @@ struct ROUTER_INFLIGHT { // Bindings для сервисов внутри etcp_router (аналогично ETCP_BINDINGS) struct ETCP_ROUTER_BINDINGS { etcp_recv_fn callbacks[SVC_ROUTE_MAX_BINDINGS]; + int (*receive_ready[SVC_ROUTE_MAX_BINDINGS])(struct UTUN_INSTANCE*, uint64_t, uint64_t, const uint8_t*, size_t); }; // ==================================================================== @@ -229,6 +235,13 @@ int etcp_router_bind(struct UTUN_INSTANCE* inst, uint8_t svc_id, etcp_recv_fn c // Снять обработчик; само по себе не закрывает логические каналы сервиса. int etcp_router_unbind(struct UTUN_INSTANCE* inst, uint8_t svc_id); +// Проверяется до доставки/ACK. 0 оставляет пакет в recv_q, сохраняя порядок; CLOSE не блокируется. +// ready получает заимствованный проверенный payload и работает в uasync без I/O. NULL снимает ограничение. +void etcp_router_set_receive_ready(struct UTUN_INSTANCE* inst, uint8_t svc_id, + int (*ready)(struct UTUN_INSTANCE*, uint64_t, uint64_t, const uint8_t*, size_t)); +// Продолжение отложено; канал не создаётся, waiter/arg сервиса не сохраняются. +void etcp_router_resume_receive(struct UTUN_INSTANCE* inst, uint64_t group, uint64_t peer, uint8_t svc_id); + // Отправить сервисный пакет (авто-conn, seq, inflight-контроль через send_q). // Принимает владение entry во всех исходах, включая ошибку. // entry->dgram = svc_id[1] || payload; 0=принято в очередь, <0=ошибка, не подтверждение доставки. diff --git a/tests/test_etcp_router_unit.c b/tests/test_etcp_router_unit.c index 36224179..54d6916a 100644 --- a/tests/test_etcp_router_unit.c +++ b/tests/test_etcp_router_unit.c @@ -1285,6 +1285,47 @@ static int test_e2e_combined_signed(void) { return 0; } +static int receive_capacity; +static int receive_ready(struct UTUN_INSTANCE* inst, uint64_t group, uint64_t peer, const uint8_t* payload, size_t bytes) { + (void)inst; (void)group; (void)peer; + if (bytes != 1 || payload[0] != rx.marker) rx.errors++; + return rx.delivered < receive_capacity; +} + +static int test_receive_backpressure(void) { + TEST("receive backpressure preserves order and ACK boundary"); + struct UASYNC* ua; struct UTUN_INSTANCE inst; etcp_recv_fn recv_cb; + SETUP(); rx_reset(0x52); receive_capacity = 0; + etcp_router_set_receive_ready(&inst, TEST_SVC_ID, receive_ready); + uint8_t byte = 0x52; + inject(recv_cb, &inst, TEST_REMOTE_NODE, TEST_SVC_ID, 0, &byte, 1, 0, ROUTER_FLAG_START); + inject(recv_cb, &inst, TEST_REMOTE_NODE, TEST_SVC_ID, 2, &byte, 1, 0, 0); + inject(recv_cb, &inst, TEST_REMOTE_NODE, TEST_SVC_ID, 1, &byte, 1, 0, 0); + for (int i = 0; i < 8; i++) uasync_poll(ua, 0); + struct ETCP_ROUTER_CONN* rc = etcp_router_conn_get(&inst, TOPO_GROUP_UTUN, TEST_REMOTE_NODE, TEST_SVC_ID); + TESTASSERT(rx.delivered == 0 && rc->rx_seq == 0 && rc->recv_q->count == 3 && rc->receive_blocked); + receive_capacity = 1; etcp_router_resume_receive(&inst, TOPO_GROUP_UTUN, TEST_REMOTE_NODE, TEST_SVC_ID); + TESTASSERT(rx.delivered == 0); + for (int i = 0; i < 8; i++) uasync_poll(ua, 0); + TESTASSERT(rx.delivered == 1 && rc->rx_seq == 1 && rc->recv_q->count == 2 && rc->receive_blocked); + receive_capacity = 3; etcp_router_resume_receive(&inst, TOPO_GROUP_UTUN, TEST_REMOTE_NODE, TEST_SVC_ID); + for (int i = 0; i < 8; i++) uasync_poll(ua, 0); + TESTASSERT(rx.delivered == 3 && !rx.errors && rc->rx_seq == 3 && rc->recv_q->count == 0 && !rc->receive_blocked); + inject(recv_cb, &inst, TEST_REMOTE_NODE, TEST_SVC_ID, 0, NULL, 0, 1, ROUTER_FLAG_RECEIVE_BUSY); + TESTASSERT(rc->peer_receive_blocked && rc->tx_acked == 0); + inject(recv_cb, &inst, TEST_REMOTE_NODE, TEST_SVC_ID, 0, NULL, 0, 1, 0); + TESTASSERT(!rc->peer_receive_blocked && rc->tx_acked == 0); + inject(recv_cb, &inst, TEST_REMOTE_NODE, TEST_SVC_ID, 3, &byte, 1, 0, 0); + for (int i = 0; i < 8; i++) uasync_poll(ua, 0); + TESTASSERT(rc->receive_blocked && rx.delivered == 3); + etcp_router_resume_receive(&inst, TOPO_GROUP_UTUN, TEST_REMOTE_NODE, TEST_SVC_ID); + int before = g_notify_count; + etcp_router_conn_restart(&inst, TOPO_GROUP_UTUN, TEST_REMOTE_NODE, TEST_SVC_ID); + for (int i = 0; i < 8; i++) uasync_poll(ua, 0); + TESTASSERT(rx.delivered == 3 && g_notify_count == before + 1 && !rc->receive_pump); + TEARDOWN(); PASS(); return 0; +} + // ======================== Main ======================== int main(void) { // Run all tests @@ -1294,6 +1335,7 @@ int main(void) { test_conn_get_reuse(); test_conn_close(); test_in_order(); + test_receive_backpressure(); test_out_of_order(); test_duplicate_same(); test_duplicate_past(); diff --git a/tests/test_member_files_network.c b/tests/test_member_files_network.c index acf6e3c9..b02a56c2 100644 --- a/tests/test_member_files_network.c +++ b/tests/test_member_files_network.c @@ -1,6 +1,7 @@ /* Три настоящих ядра A/B/M: подписанный storage, wire-запросы, загрузка A, * скачивание B, запрет чужой записи, удалённого участника и restart сервиса. */ #include "../src/member_files/member_files.h" +#include "../src/member_files/private.h" #include "../src/utun_instance.h" #include "../src/chat/chat_core.h" #include "../src/chat/chat_event.h" @@ -18,8 +19,14 @@ #define GROUP 5858585858585858ULL #define CHANNEL "5858585858585858" -#define BYTES (3 * MF_BLOCK_SIZE + 37) +#define BYTES (128 * MF_BLOCK_SIZE + 37) enum { AUTHOR, READER, SERVER, NODE_COUNT }; +struct api_case { + struct mf_transfer* handle; + uint64_t id, done, total; + int finished, status, cancel_after_progress; + char local[768]; +}; struct network_test { struct UASYNC* ua; struct UTUN_INSTANCE* nodes[NODE_COUNT]; @@ -30,6 +37,11 @@ struct network_test { struct mf_message response; unsigned phase, ticks, progress; int answered, finished, transfer_status, failed; + int slow_queued, slow_done, receive_pauses; + unsigned peak_inflight; + int (*saved_receive_ready)(struct UTUN_INSTANCE*, uint64_t, uint64_t, const uint8_t*, size_t); + unsigned restart_tick; + struct api_case api[8]; void* timer; }; static struct network_test* active; @@ -40,6 +52,62 @@ static void fail(struct network_test* t, const char* why) { } static void hex(const uint8_t* input, char* out) { for (unsigned i = 0; i < 32; i++) snprintf(out + 2 * i, 3, "%02x", input[i]); } +static void slow_disk(void* arg) { + (void)arg; +#ifdef _WIN32 + Sleep(6000); +#else + struct timespec delay = {6, 0}; nanosleep(&delay, NULL); +#endif +} +static void slow_done(void* arg, int error) { + struct network_test* t = arg; + if (error) fail(t, "slow disk task rejected"); + t->slow_done = 1; +} +static void capture_log(int level, const char* category, const char* message) { + (void)level; (void)category; + struct network_test* t = active; + if (strstr(message, "router receive paused:")) t->receive_pauses++; + if (!t->slow_queued && strstr(message, "member_files: rx") && strstr(message, "kind=1 cmd=DATA")) { + t->slow_queued = 1; + mf_worker_submit(t->nodes[SERVER]->member_files->disk, slow_disk, t, slow_done, t); + } +} + +static void api_event(void* arg, const struct mf_transfer_progress* p) { + struct api_case* c = arg; + if (p->id != c->id || p->done < c->done || p->done > p->total || c->finished) { fail(active, "API progress identity/order"); return; } + c->done = p->done; c->total = p->total; + if (p->finished) { c->handle = NULL; c->finished++; c->status = p->status; } + else if (c->cancel_after_progress && p->done) { + c->cancel_after_progress = 0; member_files_transfer_cancel(c->handle); + } +} + +static void api_start(struct network_test* t, unsigned index, int node, int upload, const char* remote, const char* source) { + struct api_case* c = &t->api[index]; memset(c, 0, sizeof(*c)); c->id = 100 + index; + snprintf(c->local, sizeof(c->local), "%s/api%u.bin", t->dir, index); + struct mf_transfer_request q = { .inst = t->nodes[node], .id = c->id, .group = GROUP, + .server = t->ids[SERVER], .owner = t->ids[upload ? node : AUTHOR], .upload = upload }; + snprintf(q.remote, sizeof(q.remote), "%s", remote); + snprintf(q.local, sizeof(q.local), "%s", upload ? source : c->local); + c->handle = member_files_transfer_start(&q, api_event, c); + if (!c->handle) fail(t, "parallel API start"); +} + +static int check_file(const char* path, unsigned size) { + FILE* file = fopen(path, "rb"); if (!file) return 0; + int ok = 1; + for (unsigned i = 0; i < size; i++) if (fgetc(file) != (int)((i * 73 + 19) & 255)) { ok = 0; break; } + if (fgetc(file) != EOF) ok = 0; + fclose(file); return ok; +} + +static int hold_receive(struct UTUN_INSTANCE* inst, uint64_t group, uint64_t peer, const uint8_t* payload, size_t bytes) { + (void)inst; (void)group; (void)peer; (void)payload; (void)bytes; return 0; +} + static void event(struct UTUN_INSTANCE* inst, int type, const uint8_t* data, int len) { (void)inst; struct network_test* t = active; @@ -122,16 +190,14 @@ static void request(struct network_test* t, uint8_t command, uint64_t owner) { } static int downloaded(struct network_test* t) { - FILE* file = fopen(t->destination, "rb"); if (!file) return 0; - int ok = 1; - for (unsigned i = 0; i < BYTES; i++) if (fgetc(file) != (int)((i * 73 + 19) & 255)) { ok = 0; break; } - if (fgetc(file) != EOF) ok = 0; - fclose(file); return ok; + return check_file(t->destination, BYTES); } static void tick(void* arg) { struct network_test* t = arg; t->timer = NULL; if (++t->ticks > 6000) { fail(t, "60s deadline"); return; } + struct ETCP_ROUTER_CONN* route = etcp_router_conn_get(t->nodes[AUTHOR], GROUP, t->ids[SERVER], ETCP_RT_ID_MEMBER_FILES); + if (route->inflight_q->count > t->peak_inflight) t->peak_inflight = route->inflight_q->count; if (t->phase == 0) { int a = etcp_router_get_path(t->nodes[AUTHOR], GROUP, t->ids[SERVER], ETCP_RT_ID_MEMBER_FILES).next_hop_node_id != 0; int b = etcp_router_get_path(t->nodes[READER], GROUP, t->ids[SERVER], ETCP_RT_ID_MEMBER_FILES).next_hop_node_id != 0; @@ -156,21 +222,75 @@ static void tick(void* arg) { if (!member_files_request(t->nodes[SERVER], GROUP, t->ids[SERVER], &q, response, t)) { fail(t, "local server request"); return; } } else if (t->phase == 6 && t->answered) { if (t->response.status || t->response.size != BYTES) { fail(t, "local server STAT"); return; } + if (!t->slow_done || !t->receive_pauses || t->peak_inflight < 2) { fail(t, "stream did not use receive backpressure/pipelining"); return; } + // Переполненная очередь запросов не должна блокировать ответы в обратном направлении. + struct UTUN_INSTANCE* inst = t->nodes[AUTHOR]; struct member_files* files = inst->member_files; + unsigned jobs = files->job_count; files->job_count = 32; + struct mf_message q = { .kind = MF_RESPONSE, .command = MF_STAT, .request_id = 1 }; + uint8_t wire[MF_WIRE_MAX]; int bytes = mf_encode(&q, wire, sizeof(wire)); + int response_ready = inst->router_bindings.receive_ready[ETCP_RT_ID_MEMBER_FILES](inst, GROUP, t->ids[SERVER], wire, bytes); + q.kind = MF_REQUEST; bytes = mf_encode(&q, wire, sizeof(wire)); + int request_ready = inst->router_bindings.receive_ready[ETCP_RT_ID_MEMBER_FILES](inst, GROUP, t->ids[SERVER], wire, bytes); + files->job_count = jobs; + if (!response_ready || request_ready) { fail(t, "full request queue blocked response or admitted request"); return; } + t->phase = 8; + api_start(t, 0, AUTHOR, 1, "parallel-a.bin", t->source); + api_start(t, 1, AUTHOR, 1, "parallel-b.bin", t->source); + api_start(t, 2, AUTHOR, 0, "file.bin", NULL); + api_start(t, 3, AUTHOR, 1, "cancelled.bin", t->source); t->api[3].cancel_after_progress = 1; + api_start(t, 4, SERVER, 0, "file.bin", NULL); + api_start(t, 5, SERVER, 1, "local.bin", t->source); + char empty[768]; snprintf(empty, sizeof(empty), "%s/empty.bin", t->dir); + FILE* file = fopen(empty, "wb"); if (!file) { fail(t, "empty source"); return; } fclose(file); + api_start(t, 6, AUTHOR, 1, "empty.bin", empty); + } else if (t->phase == 8) { + for (unsigned i = 0; i < 7; i++) if (!t->api[i].finished) goto again; + for (unsigned i = 0; i < 7; i++) if (t->api[i].status != (i == 3 ? MF_CANCELLED : MF_OK)) { fail(t, "parallel completion status"); return; } + if (!check_file(t->api[2].local, BYTES) || !check_file(t->api[4].local, BYTES)) { fail(t, "duplex/local download bytes"); return; } + t->phase = 9; + api_start(t, 0, AUTHOR, 0, "parallel-a.bin", NULL); + api_start(t, 1, AUTHOR, 0, "parallel-b.bin", NULL); + api_start(t, 7, AUTHOR, 0, "empty.bin", NULL); + } else if (t->phase == 9) { + if (!t->api[0].finished || !t->api[1].finished || !t->api[7].finished) goto again; + if (t->api[0].status || t->api[1].status || t->api[7].status || !check_file(t->api[0].local, BYTES) || + !check_file(t->api[1].local, BYTES) || !check_file(t->api[7].local, 0)) { fail(t, "parallel/empty download bytes"); return; } + t->phase = 10; t->answered = 0; + struct mf_message q = { .command = MF_STAT, .owner = t->ids[AUTHOR] }; snprintf(q.path, sizeof(q.path), "cancelled.bin"); + if (!member_files_request(t->nodes[READER], GROUP, t->ids[SERVER], &q, response, t)) { fail(t, "cancelled STAT"); return; } + } else if (t->phase == 10 && t->answered) { + if (t->response.status != MF_NOT_FOUND) { fail(t, "cancelled upload published"); return; } + t->saved_receive_ready = t->nodes[AUTHOR]->router_bindings.receive_ready[ETCP_RT_ID_MEMBER_FILES]; + etcp_router_set_receive_ready(t->nodes[AUTHOR], ETCP_RT_ID_MEMBER_FILES, hold_receive); + api_start(t, 0, AUTHOR, 1, "restart.bin", t->source); + api_start(t, 1, AUTHOR, 0, "file.bin", NULL); + t->restart_tick = t->ticks; t->phase = 11; + } else if (t->phase == 11 && t->ticks > t->restart_tick + 2) { + etcp_router_conn_restart(t->nodes[AUTHOR], GROUP, t->ids[SERVER], ETCP_RT_ID_MEMBER_FILES); + etcp_router_set_receive_ready(t->nodes[AUTHOR], ETCP_RT_ID_MEMBER_FILES, t->saved_receive_ready); + t->phase = 12; + } else if (t->phase == 12) { + if (!t->api[0].finished || !t->api[1].finished || t->nodes[SERVER]->member_files->server_transfer_count) goto again; + if (t->api[0].status != MF_IO_ERROR || t->api[1].status != MF_IO_ERROR) { fail(t, "restart completion status"); return; } if (put_member(t, SERVER, READER, 1)) { fail(t, "signed deletion"); return; } t->phase = 7; request(t, MF_STAT, t->ids[AUTHOR]); } else if (t->phase == 7 && t->answered) { if (t->response.status != MF_ACCESS_DENIED) { fail(t, "deleted reader accepted"); return; } - DEBUG_INFO(DEBUG_CATEGORY_MEDIA, "member_files network: PASS progress=%u ticks=%u", t->progress, t->ticks); + DEBUG_INFO(DEBUG_CATEGORY_MEDIA, "member_files network: PASS progress=%u ticks=%u peak_inflight=%u pauses=%d", + t->progress, t->ticks, t->peak_inflight, t->receive_pauses); uasync_stop(t->ua); return; } +again: t->timer = uasync_set_timeout(t->ua, 100, t, tick, "member_files_test"); } int main(void) { debug_config_init(); debug_set_level(DEBUG_LEVEL_ERROR); debug_set_category_level(DEBUG_CATEGORY_MEDIA, DEBUG_LEVEL_DEBUG); debug_set_category_level(DEBUG_CATEGORY_DEBUG, DEBUG_LEVEL_TRACE); + debug_set_category_level(DEBUG_CATEGORY_ETCPROUTE, DEBUG_LEVEL_DEBUG); utun_instance_set_tun_init_enabled(0); struct network_test t = {0}; active = &t; + debug_set_log_hook(capture_log); snprintf(t.dir, sizeof(t.dir), "/tmp/utun_member_files_net_XXXXXX"); if (test_mkdtemp(t.dir)) return 1; t.ua = uasync_create(); @@ -179,9 +299,27 @@ int main(void) { for (unsigned i = 0; i < NODE_COUNT; i++) if (start(&t, i, port)) { t.failed = 1; goto stop; } if (setup(&t)) { t.failed = 1; goto stop; } t.timer = uasync_set_timeout(t.ua, 100, &t, tick, "member_files_test"); uasync_mainloop(t.ua); + if (!t.failed) { + // Stop/destroy вне callbacks: server cleanup ещё использует storage после текущего I/O. + api_start(&t, 0, AUTHOR, 1, "stop.bin", t.source); + api_start(&t, 1, AUTHOR, 0, "file.bin", NULL); + uint64_t deadline = get_time_tb() + 50000; + while (!t.nodes[SERVER]->member_files->server_transfer_count && get_time_tb() < deadline) { + uasync_poll(t.ua, 0); +#ifdef _WIN32 + Sleep(1); +#else + struct timespec delay = {0, 1000000}; nanosleep(&delay, NULL); +#endif + } + if (!t.nodes[SERVER]->member_files->server_transfer_count) fail(&t, "server stream did not start before teardown"); + } stop: if (t.timer) uasync_cancel_timeout(t.ua, t.timer); for (unsigned i = 0; i < NODE_COUNT; i++) if (t.nodes[i]) utun_instance_destroy(t.nodes[i]); + if (!t.failed && (!t.api[0].finished || !t.api[1].finished || t.api[0].status != MF_CANCELLED || t.api[1].status != MF_CANCELLED)) + t.failed = 1; uasync_destroy(t.ua, 0); + debug_set_log_hook(NULL); printf("member_files network: %s phase=%u\n", t.failed ? "FAIL" : "PASS", t.phase); return t.failed; }