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

Trace file storage and worker lifecycle in debug

master
evgeny 7 days ago
parent
8982bb034e
commit
e913a27f0c
3 changed files with 37 additions and 11 deletions
Whitespace
Show all changes Ignore whitespace when comparing lines Ignore changes in amount of whitespace Ignore changes in whitespace at EOL
Unified View
Diff Options
Show Stats Download Patch File Download Diff File
  1. 19
      src/member_files/disk.c
  2. 16
      src/member_files/storage.c
  3. 13
      src/member_files/worker.c

19
src/member_files/disk.c
Unescape View File

@ -23,9 +23,10 @@ static int disk_error(const char* operation, const char* path) {
#ifdef ELOOP #ifdef ELOOP
if (error == ELOOP) status = MF_INVALID_PATH; if (error == ELOOP) status = MF_INVALID_PATH;
#endif #endif
if (status == MF_NOT_FOUND || status == MF_EXISTS) DEBUG_DEBUG(DEBUG_CATEGORY_DEBUG, "member_files: disk %s path=%s status=%s errno=%d (%s)", operation,
DEBUG_DEBUG(DEBUG_CATEGORY_MEDIA, "member_files: disk %s path=%s status=%s", operation, path ? path : "", mf_status_text(status)); path ? path : "", mf_status_text(status), error, strerror(error));
else DEBUG_WARN(DEBUG_CATEGORY_MEDIA, "member_files: disk %s path=%s errno=%d (%s)", operation, path ? path : "", 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; return status;
} }
@ -41,9 +42,9 @@ static int disk_wide(const char* path, wchar_t* wide, size_t cap) {
static int disk_win_error(const char* operation) { static int disk_win_error(const char* operation) {
DWORD e = GetLastError(); DWORD e = GetLastError();
if (e == ERROR_FILE_NOT_FOUND || e == ERROR_PATH_NOT_FOUND || e == ERROR_ALREADY_EXISTS || e == ERROR_FILE_EXISTS) DEBUG_DEBUG(DEBUG_CATEGORY_DEBUG, "member_files: %s winerr=%lu", operation, e);
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)
else DEBUG_WARN(DEBUG_CATEGORY_MEDIA, "member_files: %s winerr=%lu", operation, e); 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; if (e == ERROR_FILE_NOT_FOUND || e == ERROR_PATH_NOT_FOUND) return MF_NOT_FOUND;
if (e == ERROR_ALREADY_EXISTS || e == ERROR_FILE_EXISTS) return MF_EXISTS; if (e == ERROR_ALREADY_EXISTS || e == ERROR_FILE_EXISTS) return MF_EXISTS;
if (e == ERROR_DISK_FULL || e == ERROR_HANDLE_DISK_FULL) return MF_NO_SPACE; if (e == ERROR_DISK_FULL || e == ERROR_HANDLE_DISK_FULL) return MF_NO_SPACE;
@ -349,6 +350,7 @@ int mf_disk_local_remove(const char* path) { return !unlink(path) || errno == EN
#endif #endif
int mf_disk_read(int fd, uint64_t offset, void* data, size_t size) { 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; size_t done = 0;
while (done < size) { while (done < size) {
#ifdef _WIN32 #ifdef _WIN32
@ -362,9 +364,11 @@ 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; } (unsigned long long)offset, size, done, errno); return MF_IO_ERROR; }
done += (size_t)n; done += (size_t)n;
} }
DEBUG_TRACE(DEBUG_CATEGORY_DEBUG, "member_files: disk read done fd=%d bytes=%zu", fd, done);
return MF_OK; return MF_OK;
} }
int mf_disk_write(int fd, uint64_t offset, const void* data, size_t size) { 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; size_t done = 0;
while (done < size) { while (done < size) {
#ifdef _WIN32 #ifdef _WIN32
@ -377,11 +381,13 @@ int mf_disk_write(int fd, uint64_t offset, const void* data, size_t size) {
if (n <= 0) return disk_error("write", NULL); if (n <= 0) return disk_error("write", NULL);
done += (size_t)n; done += (size_t)n;
} }
DEBUG_TRACE(DEBUG_CATEGORY_DEBUG, "member_files: disk write done fd=%d bytes=%zu", fd, done);
return MF_OK; return MF_OK;
} }
/* Полное хеширование выполняется только worker, с ограниченным буфером. */ /* Полное хеширование выполняется только worker, с ограниченным буфером. */
int mf_disk_hash(int fd, uint64_t size, uint8_t hash[32]) { 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);
EVP_MD_CTX* ctx = EVP_MD_CTX_new(); EVP_MD_CTX* ctx = EVP_MD_CTX_new();
int rc = MF_IO_ERROR; int rc = MF_IO_ERROR;
if (!ctx || EVP_DigestInit_ex(ctx, EVP_sha256(), NULL) != 1) goto out; if (!ctx || EVP_DigestInit_ex(ctx, EVP_sha256(), NULL) != 1) goto out;
@ -396,6 +402,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; rc = EVP_DigestFinal_ex(ctx, hash, &n) == 1 && n == 32 ? MF_OK : MF_IO_ERROR;
out: out:
EVP_MD_CTX_free(ctx); EVP_MD_CTX_free(ctx);
DEBUG_DEBUG(DEBUG_CATEGORY_DEBUG, "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); if (rc) DEBUG_ERROR(DEBUG_CATEGORY_MEDIA, "member_files: hash failed size=%llu status=%d", (unsigned long long)size, rc);
return rc; return rc;
} }

16
src/member_files/storage.c
Unescape View File

@ -65,7 +65,8 @@ 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); int rc = mf_disk_open(&s->root, path, 0, write, &slot->fd);
if (rc) return rc; if (rc) return rc;
snprintf(slot->path, sizeof(slot->path), "%s", path); slot->touched = get_time_tb(); *fd = slot->fd; snprintf(slot->path, sizeof(slot->path), "%s", path); slot->touched = get_time_tb(); *fd = slot->fd;
DEBUG_DEBUG(DEBUG_CATEGORY_MEDIA, "member_files: open path=%s writable=%d descriptors=%u", path, write, mf_storage_open_count(s)); DEBUG_DEBUG(DEBUG_CATEGORY_DEBUG, "member_files: open path=%s fd=%d writable=%d descriptors=%u",
path, *fd, write, mf_storage_open_count(s));
return MF_OK; return MF_OK;
} }
@ -146,7 +147,8 @@ void mf_storage_sweep(struct mf_storage* s, uint64_t now) {
struct store_file* f = &s->files[i]; struct store_file* f = &s->files[i];
if (f->fd < 0 || now < f->touched || now - f->touched < MF_IDLE_TB) continue; if (f->fd < 0 || now < f->touched || now - f->touched < MF_IDLE_TB) continue;
int rc = mf_disk_close(f->fd); f->fd = -1; int rc = mf_disk_close(f->fd); f->fd = -1;
DEBUG_DEBUG(DEBUG_CATEGORY_MEDIA, "member_files: idle close path=%s status=%d descriptors=%u", f->path, rc, mf_storage_open_count(s)); DEBUG_DEBUG(DEBUG_CATEGORY_DEBUG, "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; f->path[0] = 0;
} }
for (unsigned i = 0; i < STORE_UPLOADS; i++) { for (unsigned i = 0; i < STORE_UPLOADS; i++) {
@ -377,6 +379,10 @@ void mf_storage_execute(struct mf_storage* s, uint64_t group, uint64_t peer, con
r->status = MF_ACCESS_DENIED; return; r->status = MF_ACCESS_DENIED; return;
} }
uint64_t now = get_time_tb(); uint64_t now = get_time_tb();
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)s->used, (unsigned long long)s->reserved, (unsigned long long)s->quota);
int mutates = q->command == MF_COMMIT || q->command >= MF_MKDIR; int mutates = q->command == MF_COMMIT || q->command >= MF_MKDIR;
struct store_receipt* slot = NULL; struct store_receipt* slot = NULL;
uint8_t wire[MF_WIRE_MAX], fingerprint[32]; uint8_t wire[MF_WIRE_MAX], fingerprint[32];
@ -421,11 +427,13 @@ 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; slot->group = group; slot->peer = peer; slot->until = now + RECEIPT_TB;
memcpy(slot->fingerprint, fingerprint, 32); slot->response = *r; 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",
(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) { if (r->status) {
r->data_len = 0; r->data_len = 0;
DEBUG_WARN(DEBUG_CATEGORY_MEDIA, "member_files: refused group=%llu peer=%llu owner=%llu cmd=%u path=%s offset=%llu size=%llu status=%s", DEBUG_WARN(DEBUG_CATEGORY_MEDIA, "member_files: refused group=%llu peer=%llu owner=%llu cmd=%u path=%s offset=%llu size=%llu status=%s",
(unsigned long long)group, (unsigned long long)peer, (unsigned long long)q->owner, q->command, q->path, (unsigned long long)group, (unsigned long long)peer, (unsigned long long)q->owner, q->command, q->path,
(unsigned long long)q->offset, (unsigned long long)q->size, mf_status_text(r->status)); (unsigned long long)q->offset, (unsigned long long)q->size, mf_status_text(r->status));
} else DEBUG_DEBUG(DEBUG_CATEGORY_MEDIA, "member_files: served cmd=%u path=%s offset=%llu bytes=%u", q->command, q->path, }
(unsigned long long)q->offset, r->data_len);
} }

13
src/member_files/worker.c
Unescape View File

@ -24,6 +24,7 @@ struct mf_worker {
struct mf_work* queue; struct mf_work* queue;
struct mf_work* tail; struct mf_work* tail;
unsigned count; unsigned count;
long creator_pid;
int stopping; int stopping;
}; };
@ -33,6 +34,8 @@ static void work_finish(struct mf_work* j) {
while (*link && *link != j) link = &(*link)->next; while (*link && *link != j) link = &(*link)->next;
if (*link) *link = j->next; if (*link) *link = j->next;
w->count--; w->count--;
DEBUG_TRACE(DEBUG_CATEGORY_DEBUG, "member_files: worker callback owner=%p task=%p remaining=%u pid=%ld",
(void*)w, (void*)j, w->count, (long)getpid());
mf_done_fn done = j->done; void* arg = j->arg; mf_done_fn done = j->done; void* arg = j->arg;
u_free(j); done(arg, 0); u_free(j); done(arg, 0);
} }
@ -54,9 +57,11 @@ static void* disk_thread(void* arg) {
struct mf_work* j = w->queue; w->queue = j->queued_next; struct mf_work* j = w->queue; w->queue = j->queued_next;
if (!w->queue) w->tail = NULL; if (!w->queue) w->tail = NULL;
pthread_mutex_unlock(&w->mutex); pthread_mutex_unlock(&w->mutex);
DEBUG_TRACE(DEBUG_CATEGORY_DEBUG, "member_files: worker execute owner=%p task=%p pid=%ld", (void*)w, (void*)j, (long)getpid());
j->work(j->data); j->work(j->data);
pthread_mutex_lock(&w->mutex); pthread_mutex_lock(&w->mutex);
j->completed = 1; j->completed = 1;
DEBUG_TRACE(DEBUG_CATEGORY_DEBUG, "member_files: worker posting owner=%p task=%p pid=%ld", (void*)w, (void*)j, (long)getpid());
uasync_post_reserved(w->ua, j->completion); uasync_post_reserved(w->ua, j->completion);
pthread_cond_broadcast(&w->condition); pthread_cond_broadcast(&w->condition);
} }
@ -67,7 +72,7 @@ static void* disk_thread(void* arg) {
struct mf_worker* mf_worker_create(struct UASYNC* ua) { struct mf_worker* mf_worker_create(struct UASYNC* ua) {
struct mf_worker* w = u_calloc(1, sizeof(*w)); struct mf_worker* w = u_calloc(1, sizeof(*w));
if (!w) { DEBUG_ERROR(DEBUG_CATEGORY_MEDIA, "member_files: disk worker allocation failed"); return NULL; } if (!w) { DEBUG_ERROR(DEBUG_CATEGORY_MEDIA, "member_files: disk worker allocation failed"); return NULL; }
w->ua = ua; w->ua = ua; w->creator_pid = (long)getpid();
int rc = pthread_mutex_init(&w->mutex, NULL); int rc = pthread_mutex_init(&w->mutex, NULL);
if (rc) { DEBUG_ERROR(DEBUG_CATEGORY_MEDIA, "member_files: worker mutex failed rc=%d", rc); u_free(w); return NULL; } if (rc) { DEBUG_ERROR(DEBUG_CATEGORY_MEDIA, "member_files: worker mutex failed rc=%d", rc); u_free(w); return NULL; }
rc = pthread_cond_init(&w->condition, NULL); rc = pthread_cond_init(&w->condition, NULL);
@ -80,6 +85,8 @@ struct mf_worker* mf_worker_create(struct UASYNC* ua) {
DEBUG_ERROR(DEBUG_CATEGORY_MEDIA, "member_files: disk thread failed rc=%d", rc); DEBUG_ERROR(DEBUG_CATEGORY_MEDIA, "member_files: disk thread failed rc=%d", rc);
pthread_cond_destroy(&w->condition); pthread_mutex_destroy(&w->mutex); u_free(w); return NULL; pthread_cond_destroy(&w->condition); pthread_mutex_destroy(&w->mutex); u_free(w); return NULL;
} }
DEBUG_DEBUG(DEBUG_CATEGORY_DEBUG, "member_files: worker created owner=%p pid=%ld ua=%p limit=64",
(void*)w, w->creator_pid, (void*)ua);
DEBUG_INFO(DEBUG_CATEGORY_MEDIA, "member_files: disk thread started queue_limit=64"); return w; DEBUG_INFO(DEBUG_CATEGORY_MEDIA, "member_files: disk thread started queue_limit=64"); return w;
} }
@ -97,6 +104,8 @@ void mf_worker_submit_reserved(struct mf_worker* w, struct mf_work* j,
mf_work_fn work, void* data, mf_done_fn done, void* arg) { mf_work_fn work, void* data, mf_done_fn done, void* arg) {
j->owner = w; j->work = work; j->data = data; j->done = done; j->arg = arg; j->owner = w; j->work = work; j->data = data; j->done = done; j->arg = arg;
j->next = w->jobs; w->jobs = j; w->count++; j->next = w->jobs; w->jobs = j; w->count++;
DEBUG_TRACE(DEBUG_CATEGORY_DEBUG, "member_files: worker queued owner=%p task=%p current_pid=%ld creator_pid=%ld active=%u",
(void*)w, (void*)j, (long)getpid(), w->creator_pid, w->count);
pthread_mutex_lock(&w->mutex); pthread_mutex_lock(&w->mutex);
if (w->tail) w->tail->queued_next = j; if (w->tail) w->tail->queued_next = j;
else w->queue = j; else w->queue = j;
@ -112,6 +121,8 @@ void mf_worker_submit(struct mf_worker* w, mf_work_fn work, void* data, mf_done_
void mf_worker_destroy(struct mf_worker* w) { void mf_worker_destroy(struct mf_worker* w) {
if (!w) return; 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. */ /* Не запускаем loop рекурсивно: завершаем только собственные callbacks. */
while (w->jobs) { while (w->jobs) {
struct mf_work* j = w->jobs; struct mf_work* j = w->jobs;

Write Preview
Loading…
Cancel
Save