556 lines
29 KiB
C
556 lines
29 KiB
C
/* Один 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 <openssl/rand.h>
|
|
#include <openssl/evp.h>
|
|
#include <string.h>
|
|
#include <stdio.h>
|
|
|
|
#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;
|
|
struct mf_transfer_request request;
|
|
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;
|
|
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, 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) return;
|
|
t->last_event = now;
|
|
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_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 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;
|
|
}
|
|
|
|
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 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;
|
|
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;
|
|
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 ||
|
|
(t->request.max_bytes && t->source.size > t->request.max_bytes))) 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;
|
|
int status = error ? MF_CANCELLED : t->work_status;
|
|
if (t->work == TRANSFER_FINISH) {
|
|
t->finished_io = 1;
|
|
if (!t->status) t->status = status;
|
|
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->stopping || t->files->closing || status) { transfer_stop(t, status ? status : MF_CANCELLED); return; }
|
|
t->progress_tb = get_time_tb();
|
|
if (t->work == TRANSFER_PREPARE) {
|
|
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* 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;
|
|
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_stop(t, MF_INVALID_RANGE); return; }
|
|
if (t->request.max_bytes && r->size > t->request.max_bytes) {
|
|
DEBUG_WARN(DEBUG_CATEGORY_MEDIA, "member_files: download exceeds limit id=%llu bytes=%llu limit=%llu",
|
|
(unsigned long long)t->request.id, (unsigned long long)r->size, (unsigned long long)t->request.max_bytes);
|
|
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->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_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 || command == MF_STAT) 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_MEDIA, "member_files: stream CM id=%llu event=%d", (unsigned long long)t->request.id, event);
|
|
if (event == CONN_EVENT_UP) transfer_wake(t);
|
|
}
|
|
|
|
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: 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;
|
|
}
|
|
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;
|
|
}
|
|
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_MEDIA, "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;
|
|
}
|
|
|
|
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_MEDIA,
|
|
"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) { 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;
|
|
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);
|
|
}
|