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

Cancel revoked file streams and separate publication from cleanup

Browse Source
This commit is contained in:
evgeny committed 2026-11-03 20:11:31 +01:00
1 parent 0d7452b836
commit e22e66bc4e
2 files changed
+68 -33
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

src/member_files/private.h
+2
View File
@@ -37,6 +37,8 @@ int mf_transfer_receive_ready(struct member_files* files, uint64_t group, uint64
void mf_transfer_dispatch(struct member_files* files, uint64_t group, uint64_t peer, const struct mf_message* message);
/* Завершить потоки пары с IO_ERROR при reset/close router; прочие пары не затрагиваются. */
void mf_transfers_peer_down(struct member_files* files, uint64_t group, uint64_t peer);
/* Отменить передачи группы с отозванным членством/ролью/owner; уже начатый I/O завершает только очистку. */
void mf_transfers_revalidate(struct member_files* files, uint64_t group);
/* Планировать готовые pumps / обработать прогресс и таймауты; все функции вызываются в uasync. */
void mf_transfers_wake(struct member_files* files);
void mf_transfers_tick(struct member_files* files, uint64_t now);
src/member_files/transfer.c
+66 -33
View File
@@ -18,7 +18,7 @@
#define TRANSFER_MAX 16
#define TRANSFER_TIMEOUT_TB 1200000ULL
enum transfer_work { TRANSFER_PREPARE, TRANSFER_READ, TRANSFER_WRITE, TRANSFER_VERIFY, TRANSFER_FINISH };
enum transfer_work { TRANSFER_PREPARE, TRANSFER_READ, TRANSFER_WRITE, TRANSFER_VERIFY, TRANSFER_PUBLISH, TRANSFER_FINISH };
struct mf_chunk { uint32_t bytes; uint8_t data[MF_BLOCK_SIZE]; };
struct mf_transfer {
struct mf_transfer* next;
@@ -37,6 +37,7 @@ struct mf_transfer {
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;
int temp_owned; /* Только worker: локальный download-temp успешно создан через O_EXCL. */
int cleanup_abort; /* Неизменяемый снимок для обязательной очистки; worker не читает живой status. */
enum transfer_work work;
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;
@@ -84,7 +85,8 @@ static struct mf_message transfer_storage_request(struct mf_transfer* t, uint8_t
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);
member_files_member(inst, t->request.group, t->request.server, 1) &&
member_files_member(inst, t->request.group, t->request.owner, 0);
struct mf_message q = t->open;
return mf_server_allowed(t->files, t->request.group, t->request.server, &q);
}
@@ -190,28 +192,30 @@ static void transfer_work(void* arg) {
if (!rc && memcmp(&after, &t->source, sizeof(after))) rc = MF_STALE;
}
}
} else if (t->work == TRANSFER_PUBLISH) {
rc = mf_disk_sync(t->fd);
int closed = mf_disk_close(t->fd); t->fd = -1;
if (!rc) rc = closed;
if (!rc) rc = mf_disk_local_publish(t->temp, t->request.local);
if (!rc) t->temp_owned = 0;
DEBUG_DEBUG(DEBUG_CATEGORY_MEDIA, "member_files: download publish id=%llu status=%s path=%s",
(unsigned long long)t->request.id, mf_status_text(rc), t->request.local);
} else {
if (t->server && !t->sending && t->status && !t->committed) {
if (t->cleanup_abort) {
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;
rc = mf_disk_close(t->fd); t->fd = -1;
}
if (!t->server && !t->sending && t->temp_owned) {
if (!t->status && !rc) rc = mf_disk_local_publish(t->temp, t->request.local);
if (t->status || rc) {
int removed = mf_disk_local_remove(t->temp);
if (removed) DEBUG_ERROR(DEBUG_CATEGORY_MEDIA, "member_files: owned temp cleanup failed id=%llu path=%s status=%s",
(unsigned long long)t->request.id, t->temp, mf_status_text(removed));
else t->temp_owned = 0;
} else t->temp_owned = 0;
DEBUG_DEBUG(DEBUG_CATEGORY_MEDIA, "member_files: download temp finish id=%llu owned=%d status=%s io_status=%s path=%s",
(unsigned long long)t->request.id, t->temp_owned, mf_status_text(t->status), mf_status_text(rc), t->temp);
int removed = mf_disk_local_remove(t->temp);
if (removed) DEBUG_ERROR(DEBUG_CATEGORY_MEDIA, "member_files: owned temp cleanup failed id=%llu path=%s status=%s",
(unsigned long long)t->request.id, t->temp, mf_status_text(removed));
else t->temp_owned = 0;
if (!rc) rc = removed;
}
EVP_MD_CTX_free(t->digest); t->digest = NULL;
}
@@ -221,6 +225,16 @@ static void transfer_work(void* arg) {
static void transfer_done(void* arg, int error) {
struct mf_transfer* t = arg; t->busy = 0;
int status = error ? MF_CANCELLED : t->work_status;
if (!status && !transfer_allowed(t)) {
status = MF_ACCESS_DENIED; t->peer_down = 1;
DEBUG_WARN(DEBUG_CATEGORY_MEDIA, "member_files: disk result revoked group=%llx peer=%llx id=%llu work=%d",
(unsigned long long)t->request.group, (unsigned long long)t->request.server,
(unsigned long long)t->request.id, t->work);
}
if (t->work == TRANSFER_PUBLISH) {
if (!t->status) t->status = status;
transfer_submit(t, TRANSFER_FINISH); return;
}
if (t->work == TRANSFER_FINISH) {
t->finished_io = 1;
if (!t->status) t->status = status;
@@ -255,33 +269,36 @@ static void transfer_done(void* arg, int error) {
}
static void transfer_submit(struct mf_transfer* t, enum transfer_work work) {
if (work != TRANSFER_FINISH && !transfer_allowed(t)) { transfer_stop(t, MF_ACCESS_DENIED); return; }
t->work = work; t->busy = 1;
if (work == TRANSFER_FINISH) {
t->cleanup_abort = t->server && !t->sending && t->status && !t->committed;
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));
}
if (t->stopping) {
if (t->busy && t->work != TRANSFER_FINISH) mf_worker_cancel(t->files->disk, t);
else 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;
}
DEBUG_DEBUG(DEBUG_CATEGORY_MEDIA, "member_files: stream stopping id=%llu starting=%d active=%d io=%d pending=%d",
(unsigned long long)t->request.id, t->starting, t->active, t->busy, t->pending != NULL);
t->stopping = 1; t->starting = 0;
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->busy) { if (t->work != TRANSFER_FINISH) mf_worker_cancel(t->files->disk, t); return; }
if (t->finished_io) { transfer_release(t); return; }
transfer_submit(t, TRANSFER_FINISH);
transfer_submit(t, !t->status && !t->server && !t->sending ? TRANSFER_PUBLISH : TRANSFER_FINISH);
}
static int transfer_enqueue(struct mf_transfer* t, struct mf_message* q) {
@@ -295,6 +312,7 @@ static int transfer_enqueue(struct mf_transfer* t, struct mf_message* q) {
static void transfer_pump(void* arg) {
struct mf_transfer* t = arg; t->pump = NULL;
if (t->files->closing) return;
if (!t->stopping && !transfer_allowed(t)) { transfer_stop(t, MF_ACCESS_DENIED); return; }
if (t->starting) {
t->starting = 0;
if (t->sending || t->server) transfer_submit(t, TRANSFER_PREPARE);
@@ -402,7 +420,7 @@ struct mf_transfer* member_files_transfer_start(const struct mf_transfer_request
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)) {
!member_files_member(q->inst, q->group, q->server, 1) || !member_files_member(q->inst, q->group, q->owner, 0)) {
DEBUG_WARN(DEBUG_CATEGORY_MEDIA, "member_files: stream start denied/limit reached"); return NULL;
}
struct mf_transfer_request request = *q;
@@ -526,20 +544,35 @@ void mf_transfers_tick(struct member_files* f, uint64_t now) {
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) { t->peer_down = 1; transfer_stop(t, MF_CANCELLED); }
t = next;
for (;;) {
struct mf_transfer* found = NULL;
for (struct mf_transfer* t = f->transfers; t; t = t->next)
if (!t->peer_down && (!group || t->request.group == group)) { found = t; break; }
if (!found) break;
found->peer_down = 1; transfer_stop(found, MF_CANCELLED);
}
}
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;
for (;;) {
struct mf_transfer* found = NULL;
for (struct mf_transfer* t = f->transfers; t; t = t->next)
if (!t->peer_down && t->request.group == group && t->request.server == peer) { found = t; break; }
if (!found) break;
found->peer_down = 1; transfer_stop(found, MF_IO_ERROR);
}
}
void mf_transfers_revalidate(struct member_files* f, uint64_t group) {
for (;;) {
struct mf_transfer* denied = NULL;
for (struct mf_transfer* t = f->transfers; t; t = t->next)
if (!t->peer_down && t->request.group == group && !transfer_allowed(t)) { denied = t; break; }
if (!denied) break;
DEBUG_INFO(DEBUG_CATEGORY_MEDIA, "member_files: revoke stream group=%llx peer=%llx id=%llu io=%d",
(unsigned long long)group, (unsigned long long)denied->request.server,
(unsigned long long)denied->request.id, denied->busy);
denied->peer_down = 1; transfer_stop(denied, MF_ACCESS_DENIED);
}
}