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

Keep file request diagnostics in media and drop packet noise

Browse Source
This commit is contained in:
evgeny committed 2026-10-18 15:07:45 +02:00
1 parent dcdc700d8d
commit 81ee235f73
3 files changed
+21 -55
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

src/member_files/disk.c
+7 -9
View File
@@ -24,8 +24,9 @@ static int disk_error(const char* operation, const char* path) {
#ifdef ELOOP
if (error == ELOOP) status = MF_INVALID_PATH;
#endif
DEBUG_DEBUG(DEBUG_CATEGORY_DEBUG, "member_files: disk %s path=%s status=%s errno=%d (%s)", operation,
path ? path : "", mf_status_text(status), error, strerror(error));
if (status == MF_NOT_FOUND || status == MF_EXISTS)
DEBUG_DEBUG(DEBUG_CATEGORY_MEDIA, "member_files: disk %s path=%s status=%s errno=%d (%s)", operation,
path ? path : "", mf_status_text(status), error, strerror(error));
if (status != MF_NOT_FOUND && status != MF_EXISTS)
DEBUG_WARN(DEBUG_CATEGORY_MEDIA, "member_files: disk %s path=%s errno=%d (%s)", operation, path ? path : "", error, strerror(error));
return status;
@@ -43,7 +44,8 @@ static int disk_wide(const char* path, wchar_t* wide, size_t cap) {
static int disk_win_error(const char* operation) {
DWORD e = GetLastError();
DEBUG_DEBUG(DEBUG_CATEGORY_DEBUG, "member_files: %s winerr=%lu", operation, e);
if (e == ERROR_FILE_NOT_FOUND || e == ERROR_PATH_NOT_FOUND || e == ERROR_ALREADY_EXISTS || e == ERROR_FILE_EXISTS)
DEBUG_DEBUG(DEBUG_CATEGORY_MEDIA, "member_files: %s winerr=%lu", operation, e);
if (e != ERROR_FILE_NOT_FOUND && e != ERROR_PATH_NOT_FOUND && e != ERROR_ALREADY_EXISTS && e != ERROR_FILE_EXISTS)
DEBUG_WARN(DEBUG_CATEGORY_MEDIA, "member_files: %s winerr=%lu", operation, e);
if (e == ERROR_FILE_NOT_FOUND || e == ERROR_PATH_NOT_FOUND) return MF_NOT_FOUND;
@@ -358,7 +360,6 @@ int mf_disk_local_remove(const char* path) { return !unlink(path) || errno == EN
#endif
int mf_disk_read(int fd, uint64_t offset, void* data, size_t size) {
DEBUG_TRACE(DEBUG_CATEGORY_DEBUG, "member_files: disk read fd=%d offset=%llu bytes=%zu", fd, (unsigned long long)offset, size);
size_t done = 0;
while (done < size) {
#ifdef _WIN32
@@ -372,11 +373,9 @@ int mf_disk_read(int fd, uint64_t offset, void* data, size_t size) {
(unsigned long long)offset, size, done, errno); return MF_IO_ERROR; }
done += (size_t)n;
}
DEBUG_TRACE(DEBUG_CATEGORY_DEBUG, "member_files: disk read done fd=%d bytes=%zu", fd, done);
return MF_OK;
}
int mf_disk_write(int fd, uint64_t offset, const void* data, size_t size) {
DEBUG_TRACE(DEBUG_CATEGORY_DEBUG, "member_files: disk write fd=%d offset=%llu bytes=%zu", fd, (unsigned long long)offset, size);
size_t done = 0;
while (done < size) {
#ifdef _WIN32
@@ -389,13 +388,12 @@ int mf_disk_write(int fd, uint64_t offset, const void* data, size_t size) {
if (n <= 0) return disk_error("write", NULL);
done += (size_t)n;
}
DEBUG_TRACE(DEBUG_CATEGORY_DEBUG, "member_files: disk write done fd=%d bytes=%zu", fd, done);
return MF_OK;
}
/* Полное хеширование выполняется только worker, с ограниченным буфером. */
int mf_disk_hash(int fd, uint64_t size, uint8_t hash[32]) {
DEBUG_DEBUG(DEBUG_CATEGORY_DEBUG, "member_files: hash start fd=%d bytes=%llu", fd, (unsigned long long)size);
DEBUG_DEBUG(DEBUG_CATEGORY_MEDIA, "member_files: hash start fd=%d bytes=%llu", fd, (unsigned long long)size);
EVP_MD_CTX* ctx = EVP_MD_CTX_new();
int rc = MF_IO_ERROR;
if (!ctx || EVP_DigestInit_ex(ctx, EVP_sha256(), NULL) != 1) goto out;
@@ -410,7 +408,7 @@ int mf_disk_hash(int fd, uint64_t size, uint8_t hash[32]) {
rc = EVP_DigestFinal_ex(ctx, hash, &n) == 1 && n == 32 ? MF_OK : MF_IO_ERROR;
out:
EVP_MD_CTX_free(ctx);
DEBUG_DEBUG(DEBUG_CATEGORY_DEBUG, "member_files: hash done fd=%d status=%s", fd, mf_status_text(rc));
DEBUG_DEBUG(DEBUG_CATEGORY_MEDIA, "member_files: hash done fd=%d status=%s", fd, mf_status_text(rc));
if (rc) DEBUG_ERROR(DEBUG_CATEGORY_MEDIA, "member_files: hash failed size=%llu status=%d", (unsigned long long)size, rc);
return rc;
}
src/member_files/member_files.c
+8 -34
View File
@@ -137,8 +137,6 @@ int mf_server_allowed(struct member_files* f, uint64_t group, uint64_t peer, con
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->command == MF_STREAM_READ || q->command == MF_QUOTA ||
(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;
}
@@ -152,28 +150,18 @@ int mf_send(struct member_files* f, uint64_t group, uint64_t peer, const struct
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);
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,
rc->peer_sync_done, rc->no_route);
return 1;
}
uint8_t wire[MF_WIRE_MAX];
int n = mf_encode(m, wire, sizeof(wire));
if (n < 0) return -1;
uint32_t seq = rc->tx_seq;
int result = etcp_router_conn_send(rc, wire, (size_t)n, ROUTER_FLAG_SIGNED | ROUTER_FLAG_ENCRYPTED);
DEBUG_DEBUG(DEBUG_CATEGORY_DEBUG,
"member_files: tx id=%llu kind=%u cmd=%s group=%llx peer=%llx owner=%llx status=%s bytes=%d seq=%u accepted=%d",
(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, mf_status_text(m->status), n, seq, result == 0);
if (result) DEBUG_WARN(DEBUG_CATEGORY_MEDIA, "member_files: send failed group=%llu peer=%llu cmd=%u rc=%d",
(unsigned long long)group, (unsigned long long)peer, m->command, result);
return result ? -1 : 0;
@@ -191,7 +179,7 @@ static void mf_pending_finish(struct mf_pending* p, const struct mf_message* res
struct mf_message result;
if (!response) { result = p->request; result.kind = MF_RESPONSE; result.status = (uint16_t)status; result.data_len = 0; response = &result; }
mf_response_fn cb = p->callback; void* arg = p->arg;
DEBUG_DEBUG(DEBUG_CATEGORY_DEBUG, "member_files: request done id=%llu cmd=%s status=%s pending=%u elapsed_ms=%llu",
DEBUG_DEBUG(DEBUG_CATEGORY_MEDIA, "member_files: request done id=%llu cmd=%s status=%s pending=%u elapsed_ms=%llu",
(unsigned long long)p->request.request_id, mf_command_text(p->request.command), mf_status_text(response->status),
f->pending_count, (unsigned long long)((get_time_tb() - p->started) / 10));
u_free(p); cb(arg, response);
@@ -217,7 +205,7 @@ static void mf_client_ready(struct ll_queue* q, void* arg) { (void)q; mf_pending
static void mf_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_pending* p = arg;
DEBUG_DEBUG(DEBUG_CATEGORY_DEBUG, "member_files: CM event=%d request=%llu", event, (unsigned long long)p->request.request_id);
DEBUG_DEBUG(DEBUG_CATEGORY_MEDIA, "member_files: CM event=%d request=%llu", event, (unsigned long long)p->request.request_id);
if (event == CONN_EVENT_UP) mf_pending_wake(p);
}
@@ -247,7 +235,7 @@ struct mf_pending* member_files_request(struct UTUN_INSTANCE* inst, uint64_t gro
DEBUG_WARN(DEBUG_CATEGORY_MEDIA, "member_files: duplicate local request id"); u_free(p); return NULL;
}
p->next = f->pending; f->pending = p; f->pending_count++;
DEBUG_DEBUG(DEBUG_CATEGORY_DEBUG,
DEBUG_DEBUG(DEBUG_CATEGORY_MEDIA,
"member_files: request id=%llu cmd=%s group=%llx server=%llx owner=%llx path=%s offset=%llu size=%llu pending=%u",
(unsigned long long)p->request.request_id, mf_command_text(p->request.command), (unsigned long long)group,
(unsigned long long)server, (unsigned long long)p->request.owner, p->request.path,
@@ -305,9 +293,6 @@ static void mf_server_ready(struct ll_queue* q, void* arg) {
static void mf_server_work(void* arg) {
struct mf_job* j = arg;
if (j->request.command) DEBUG_DEBUG(DEBUG_CATEGORY_DEBUG, "member_files: disk start id=%llu cmd=%s group=%llx peer=%llx path=%s",
(unsigned long long)j->request.request_id, mf_command_text(j->request.command), (unsigned long long)j->group,
(unsigned long long)j->peer, j->request.path);
if (j->request.command) mf_storage_execute(j->files->storage, j->group, j->peer, &j->request, &j->response);
else mf_storage_sweep(j->files->storage, get_time_tb());
}
@@ -317,8 +302,8 @@ static void mf_server_done(void* arg, int error) {
struct member_files* f = j->files; f->worker = NULL;
if (f->closing || !j->request.command) { mf_job_free(j); if (!f->closing) mf_server_pump(f); return; }
if (error) { j->response = j->request; j->response.kind = MF_RESPONSE; j->response.status = MF_BUSY; j->response.data_len = 0; }
DEBUG_DEBUG(DEBUG_CATEGORY_DEBUG, "member_files: disk done id=%llu cmd=%s status=%s error=%d jobs=%u",
(unsigned long long)j->request.request_id, mf_command_text(j->request.command), mf_status_text(j->response.status), error, f->job_count);
if (error) DEBUG_WARN(DEBUG_CATEGORY_MEDIA, "member_files: disk job failed id=%llu cmd=%s error=%d jobs=%u",
(unsigned long long)j->request.request_id, mf_command_text(j->request.command), error, f->job_count);
j->state = 2; mf_server_pump(f);
}
@@ -377,11 +362,7 @@ static void mf_server_pump(struct member_files* f) {
}
j = next;
}
if (f->worker) {
DEBUG_TRACE(DEBUG_CATEGORY_DEBUG, "member_files: server waiting for disk id=%llu cmd=%s jobs=%u",
(unsigned long long)f->worker->request.request_id, mf_command_text(f->worker->request.command), f->job_count);
return;
}
if (f->worker) return;
for (j = f->jobs; j; j = j->next) if (!j->state) break;
if (!j && f->sweep_due && f->job_count < MF_MAX_JOBS) {
j = u_calloc(1, sizeof(*j));
@@ -394,8 +375,6 @@ static void mf_server_pump(struct member_files* f) {
j->response.data_len = 0; j->state = 2; mf_server_pump(f); return;
}
j->state = 1; f->worker = j;
if (j->request.command) DEBUG_DEBUG(DEBUG_CATEGORY_DEBUG, "member_files: disk queued id=%llu cmd=%s jobs=%u",
(unsigned long long)j->request.request_id, mf_command_text(j->request.command), f->job_count);
mf_worker_submit(f->disk, mf_server_work, j, mf_server_done, j);
}
@@ -430,12 +409,9 @@ static void mf_tick(void* arg) {
/* Общая обработка проверенного сетевого сообщения или команды внутри экземпляра. */
static void mf_dispatch(struct member_files* f, uint64_t group, uint64_t peer, const struct mf_message* message) {
struct mf_message m = *message;
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.kind == MF_REQUEST && f->storage_status != MF_OK) {
int status = mf_server_allowed(f, group, peer, &m) ? f->storage_status : MF_ACCESS_DENIED;
DEBUG_DEBUG(DEBUG_CATEGORY_DEBUG, "member_files: request before storage ready id=%llu status=%s",
DEBUG_DEBUG(DEBUG_CATEGORY_MEDIA, "member_files: request before storage ready id=%llu status=%s",
(unsigned long long)m.request_id, mf_status_text(status));
mf_reply(f, group, peer, &m, status); return;
}
@@ -477,7 +453,7 @@ 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) {
DEBUG_DEBUG(DEBUG_CATEGORY_DEBUG, "member_files: recv discarded context=%p closing=%d bytes=%u",
DEBUG_DEBUG(DEBUG_CATEGORY_MEDIA, "member_files: recv discarded context=%p closing=%d bytes=%u",
(void*)f, f ? f->closing : 0, e ? e->len : 0);
goto out;
}
@@ -588,8 +564,6 @@ static void mf_gui_done(void* arg, const struct mf_message* response) {
struct mf_gui_request* q = arg;
uint8_t event[16 + MF_WIRE_MAX]; mf_put64(event, q->group); mf_put64(event + 8, q->server);
int n = mf_encode(response, event + 16, MF_WIRE_MAX);
DEBUG_DEBUG(DEBUG_CATEGORY_DEBUG, "member_files: GUI result id=%llu cmd=%s status=%s bytes=%d",
(unsigned long long)response->request_id, mf_command_text(response->command), mf_status_text(response->status), n);
if (n > 0) chat_event_post(q->inst, CHAT_EVT_MEMBER_FILES_RESPONSE, event, n + 16);
u_free(q);
}
src/member_files/storage.c
+6 -12
View File
@@ -83,7 +83,7 @@ static int store_user_update(struct store_user* u, uint64_t previous, uint64_t s
u->used = u->used - previous + size; u->reserved -= release;
if (file_delta > 0) u->files++; else if (file_delta < 0) u->files--;
if (dir_delta > 0) u->directories++; else if (dir_delta < 0) u->directories--;
DEBUG_DEBUG(DEBUG_CATEGORY_DEBUG,
DEBUG_DEBUG(DEBUG_CATEGORY_MEDIA,
"member_files: quota updated owner=%llu operation=%s used=%llu reserved=%llu files=%llu directories=%llu",
(unsigned long long)u->owner, operation, (unsigned long long)u->used, (unsigned long long)u->reserved,
(unsigned long long)u->files, (unsigned long long)u->directories);
@@ -129,7 +129,7 @@ static int store_fd(struct mf_storage* s, const char* path, int write, int* fd)
int rc = mf_disk_open(&s->root, path, 0, write, &slot->fd);
if (rc) return rc;
snprintf(slot->path, sizeof(slot->path), "%s", path); slot->touched = get_time_tb(); *fd = slot->fd;
DEBUG_DEBUG(DEBUG_CATEGORY_DEBUG, "member_files: open path=%s fd=%d writable=%d descriptors=%u",
DEBUG_DEBUG(DEBUG_CATEGORY_MEDIA, "member_files: open path=%s fd=%d writable=%d descriptors=%u",
path, *fd, write, mf_storage_open_count(s));
return MF_OK;
}
@@ -201,7 +201,7 @@ int mf_storage_initialize(struct mf_storage* s) {
if (rc) { mf_disk_root_close(&s->root); goto failed; }
s->ready = 1; s->init_status = MF_OK;
for (unsigned i = 0; i < STORE_USERS; i++) for (struct store_user* u = s->users[i]; u; u = u->next) {
DEBUG_DEBUG(DEBUG_CATEGORY_DEBUG, "member_files: quota scan owner=%llu used=%llu limit=%llu files=%llu directories=%llu",
DEBUG_DEBUG(DEBUG_CATEGORY_MEDIA, "member_files: quota scan owner=%llu used=%llu limit=%llu files=%llu directories=%llu",
(unsigned long long)u->owner, (unsigned long long)u->used, (unsigned long long)s->quota,
(unsigned long long)u->files, (unsigned long long)u->directories);
if (u->used > s->quota) DEBUG_WARN(DEBUG_CATEGORY_MEDIA, "member_files: user exceeds quota owner=%llu used=%llu limit=%llu",
@@ -253,7 +253,7 @@ void mf_storage_sweep(struct mf_storage* s, uint64_t now) {
struct store_file* f = &s->files[i];
if (f->fd < 0 || now < f->touched || now - f->touched < MF_IDLE_TB) continue;
int rc = mf_disk_close(f->fd); f->fd = -1;
DEBUG_DEBUG(DEBUG_CATEGORY_DEBUG, "member_files: idle close path=%s status=%s idle_ms=%llu descriptors=%u",
DEBUG_DEBUG(DEBUG_CATEGORY_MEDIA, "member_files: idle close path=%s status=%s idle_ms=%llu descriptors=%u",
f->path, mf_status_text(rc), (unsigned long long)((now - f->touched) / 10), mf_storage_open_count(s));
f->path[0] = 0;
}
@@ -416,7 +416,7 @@ static int store_command(struct mf_storage* s, uint64_t group, uint64_t peer, co
mf_put64(r->data + 16, user ? user->reserved : 0);
mf_put64(r->data + 24, user ? user->files : 0); mf_put64(r->data + 32, user ? user->directories : 0);
r->data_len = MF_QUOTA_BYTES;
DEBUG_DEBUG(DEBUG_CATEGORY_DEBUG, "member_files: quota response owner=%llu used=%llu reserved=%llu limit=%llu files=%llu directories=%llu",
DEBUG_DEBUG(DEBUG_CATEGORY_MEDIA, "member_files: quota response owner=%llu used=%llu reserved=%llu limit=%llu files=%llu directories=%llu",
(unsigned long long)q->owner, (unsigned long long)(user ? user->used : 0),
(unsigned long long)(user ? user->reserved : 0), (unsigned long long)s->quota,
(unsigned long long)(user ? user->files : 0), (unsigned long long)(user ? user->directories : 0));
@@ -577,12 +577,6 @@ void mf_storage_execute(struct mf_storage* s, uint64_t group, uint64_t peer, con
DEBUG_WARN(DEBUG_CATEGORY_MEDIA, "member_files: storage not ready root=%s status=%s", s->path, mf_status_text(r->status));
return;
}
struct store_user* user = store_user_get(s, q->owner, 0);
DEBUG_DEBUG(DEBUG_CATEGORY_DEBUG,
"member_files: storage request id=%llu cmd=%s path=%s offset=%llu size=%llu used=%llu reserved=%llu quota=%llu",
(unsigned long long)q->request_id, mf_command_text(q->command), q->path, (unsigned long long)q->offset,
(unsigned long long)q->size, (unsigned long long)(user ? user->used : 0),
(unsigned long long)(user ? user->reserved : 0), (unsigned long long)s->quota);
int mutates = q->command == MF_COMMIT || q->command == MF_MKDIR || q->command == MF_RENAME || q->command == MF_REMOVE_PATH;
struct store_receipt* slot = NULL;
uint8_t wire[MF_WIRE_MAX], fingerprint[32];
@@ -624,7 +618,7 @@ void mf_storage_execute(struct mf_storage* s, uint64_t group, uint64_t peer, con
slot->group = group; slot->peer = peer; slot->until = now + RECEIPT_TB;
memcpy(slot->fingerprint, fingerprint, 32); slot->response = *r;
}
DEBUG_DEBUG(DEBUG_CATEGORY_DEBUG, "member_files: storage result id=%llu cmd=%s status=%s bytes=%u more=%u cursor=%llu",
DEBUG_TRACE(DEBUG_CATEGORY_MEDIA, "member_files: storage result id=%llu cmd=%s status=%s bytes=%u more=%u cursor=%llu",
(unsigned long long)q->request_id, mf_command_text(q->command), mf_status_text(r->status), r->data_len, r->more,
(unsigned long long)r->cursor);
if (r->status) {