rebuild: rocksdb dbbench 运行成功
This commit is contained in:
@@ -12,13 +12,14 @@
|
||||
#include <errno.h>
|
||||
#include <pthread.h>
|
||||
#include <string.h>
|
||||
#include <time.h>
|
||||
|
||||
struct zvfs_spdk_io_engine g_engine = {0};
|
||||
static int g_engine_init_rc = -EAGAIN;
|
||||
static pthread_mutex_t g_super_blob_mutex = PTHREAD_MUTEX_INITIALIZER;
|
||||
static spdk_blob_id g_super_blob_id_cache = SPDK_BLOBID_INVALID;
|
||||
|
||||
static __thread struct zvfs_tls_ctx tls = {0};
|
||||
static pthread_once_t g_tls_cleanup_once = PTHREAD_ONCE_INIT;
|
||||
static pthread_key_t g_tls_cleanup_key;
|
||||
|
||||
// 初始化操作上下文
|
||||
struct json_load_ctx {
|
||||
@@ -54,9 +55,6 @@ struct md_op_ctx {
|
||||
struct { // for delete
|
||||
spdk_blob_id blob_id;
|
||||
} delete;
|
||||
struct { // for get/set super
|
||||
spdk_blob_id blob_id;
|
||||
} super;
|
||||
};
|
||||
char *op_name;
|
||||
};
|
||||
@@ -67,9 +65,37 @@ struct io_completion_ctx {
|
||||
int rc;
|
||||
};
|
||||
|
||||
struct md_poller_bootstrap_ctx {
|
||||
const char *bdev_name;
|
||||
pthread_mutex_t mu;
|
||||
pthread_cond_t cv;
|
||||
bool done;
|
||||
int rc;
|
||||
};
|
||||
|
||||
static uint64_t now_mono_ms(void);
|
||||
static int open_bdev_and_init_bs(const char *bdev_name);
|
||||
static void ensure_tls_cleanup_key(void);
|
||||
static void tls_cleanup_destructor(void *arg);
|
||||
|
||||
// metadata poller 线程函数
|
||||
static void *md_poller_fn(void *arg) {
|
||||
struct md_poller_bootstrap_ctx *boot = arg;
|
||||
|
||||
spdk_set_thread(g_engine.md_thread);
|
||||
tls.thread = g_engine.md_thread;
|
||||
|
||||
int init_rc = open_bdev_and_init_bs(boot->bdev_name);
|
||||
pthread_mutex_lock(&boot->mu);
|
||||
boot->rc = init_rc;
|
||||
boot->done = true;
|
||||
pthread_cond_signal(&boot->cv);
|
||||
pthread_mutex_unlock(&boot->mu);
|
||||
|
||||
if (init_rc != 0) {
|
||||
return NULL;
|
||||
}
|
||||
|
||||
while (true) {
|
||||
spdk_thread_poll(g_engine.md_thread, 0, 0);
|
||||
usleep(1000);
|
||||
@@ -77,14 +103,21 @@ static void *md_poller_fn(void *arg) {
|
||||
return NULL;
|
||||
}
|
||||
|
||||
static uint64_t now_mono_ms(void) {
|
||||
struct timespec ts;
|
||||
clock_gettime(CLOCK_MONOTONIC, &ts);
|
||||
return (uint64_t)ts.tv_sec * 1000ULL + (uint64_t)ts.tv_nsec / 1000000ULL;
|
||||
}
|
||||
|
||||
// 前向声明
|
||||
static struct spdk_io_channel *get_current_channel(void);
|
||||
static int dispatch_md_op(struct md_op_ctx *ctx);
|
||||
static int dispatch_md_op_quiet(struct md_op_ctx *ctx);
|
||||
static void md_op_cb(void *arg);
|
||||
static int open_bdev_and_init_bs(const char *bdev_name);
|
||||
static int load_json_config(void);
|
||||
static int ensure_engine_ready(const char *op);
|
||||
static int ensure_current_spdk_thread(const char *op);
|
||||
|
||||
|
||||
|
||||
// callbacks
|
||||
static void json_app_load_done(int rc, void *arg);
|
||||
@@ -97,8 +130,6 @@ static void blob_sync_md_cb(void *arg, int rc);
|
||||
static void blob_close_cb(void *arg, int rc);
|
||||
static void blob_delete_cb(void *arg, int rc);
|
||||
static void io_completion_cb(void *arg, int rc);
|
||||
static void blob_get_super_cb(void *arg, spdk_blob_id blobid, int rc);
|
||||
static void blob_set_super_cb(void *arg, int rc);
|
||||
|
||||
// op functions on matadata
|
||||
static void blob_create_on_md(struct md_op_ctx *ctx);
|
||||
@@ -107,8 +138,6 @@ static void blob_resize_on_md(struct md_op_ctx *ctx);
|
||||
static void blob_sync_md_on_md(struct md_op_ctx *ctx);
|
||||
static void blob_close_on_md(struct md_op_ctx *ctx);
|
||||
static void blob_delete_on_md(struct md_op_ctx *ctx);
|
||||
static void blob_get_super_on_md(struct md_op_ctx *ctx);
|
||||
static void blob_set_super_on_md(struct md_op_ctx *ctx);
|
||||
|
||||
__attribute__((constructor)) static void preload_init(void) {
|
||||
const char *auto_init = getenv("ZVFS_AUTO_INIT");
|
||||
@@ -116,7 +145,6 @@ __attribute__((constructor)) static void preload_init(void) {
|
||||
return;
|
||||
}
|
||||
|
||||
printf("\n\n auto init \n\n");
|
||||
const char *bdev_name = getenv("SPDK_BDEV_NAME") ? getenv("SPDK_BDEV_NAME") : ZVFS_BDEV;
|
||||
g_engine_init_rc = io_engine_init(bdev_name);
|
||||
if (g_engine_init_rc != 0) {
|
||||
@@ -125,7 +153,7 @@ __attribute__((constructor)) static void preload_init(void) {
|
||||
}
|
||||
|
||||
static int wait_done(bool *done_ptr, int *rc_ptr, const char *op) {
|
||||
int iter = 0;
|
||||
const uint64_t deadline_ms = now_mono_ms() + ZVFS_WAIT_TIME;
|
||||
while (!*done_ptr) {
|
||||
if (tls.thread) {
|
||||
spdk_thread_poll(tls.thread, 0, 0);
|
||||
@@ -133,7 +161,8 @@ static int wait_done(bool *done_ptr, int *rc_ptr, const char *op) {
|
||||
SPDK_ERRLOG("not init tls.thread\n");
|
||||
return -EBADE;
|
||||
}
|
||||
if (++iter > WAITER_MAX_TIME) {
|
||||
|
||||
if (now_mono_ms() >= deadline_ms) {
|
||||
SPDK_ERRLOG("%s timeout\n", op);
|
||||
return -ETIMEDOUT;
|
||||
}
|
||||
@@ -147,15 +176,24 @@ static int wait_done(bool *done_ptr, int *rc_ptr, const char *op) {
|
||||
}
|
||||
|
||||
static int wait_done_volatile(volatile bool *done_ptr, int *rc_ptr, const char *op) {
|
||||
int iter = 0;
|
||||
while (!*done_ptr) {
|
||||
const uint64_t deadline_ms = now_mono_ms() + ZVFS_WAIT_TIME;
|
||||
bool logged_no_tls = false;
|
||||
while (!__atomic_load_n(done_ptr, __ATOMIC_ACQUIRE)) {
|
||||
if (tls.thread) {
|
||||
spdk_thread_poll(tls.thread, 0, 0);
|
||||
}else{
|
||||
SPDK_ERRLOG("not init tls.thread\n");
|
||||
return -EBADE;
|
||||
} else {
|
||||
/*
|
||||
* md ops are executed on g_engine.md_thread by md_poller_fn.
|
||||
* If current worker TLS is not initialized, we still need to wait
|
||||
* for callback completion; returning early can invalidate stack ctx.
|
||||
*/
|
||||
if (!logged_no_tls) {
|
||||
SPDK_NOTICELOG("%s: tls.thread not initialized, waiting on md thread only\n", op);
|
||||
logged_no_tls = true;
|
||||
}
|
||||
usleep(1000);
|
||||
}
|
||||
if (++iter > WAITER_MAX_TIME) {
|
||||
if (now_mono_ms() >= deadline_ms) {
|
||||
SPDK_ERRLOG("%s timeout\n", op);
|
||||
return -ETIMEDOUT;
|
||||
}
|
||||
@@ -168,25 +206,6 @@ static int wait_done_volatile(volatile bool *done_ptr, int *rc_ptr, const char *
|
||||
return 0;
|
||||
}
|
||||
|
||||
// no rc error
|
||||
static int wait_done_volatile_quiet(volatile bool *done_ptr, int *rc_ptr, const char *op) {
|
||||
int iter = 0;
|
||||
while (!*done_ptr) {
|
||||
if (tls.thread) {
|
||||
spdk_thread_poll(tls.thread, 0, 0);
|
||||
} else {
|
||||
SPDK_ERRLOG("not init tls.thread\n");
|
||||
return -EBADE;
|
||||
}
|
||||
if (++iter > WAITER_MAX_TIME) {
|
||||
SPDK_ERRLOG("%s timeout\n", op);
|
||||
return -ETIMEDOUT;
|
||||
}
|
||||
}
|
||||
|
||||
return *rc_ptr;
|
||||
}
|
||||
|
||||
int io_engine_init(const char *bdev_name) {
|
||||
if (g_engine_init_rc == 0 && g_engine.bs != NULL && g_engine.md_thread != NULL) {
|
||||
return 0;
|
||||
@@ -239,22 +258,40 @@ int io_engine_init(const char *bdev_name) {
|
||||
return g_engine_init_rc;
|
||||
}
|
||||
|
||||
// 起专用 poller pthread for md_thread
|
||||
struct md_poller_bootstrap_ctx boot = {
|
||||
.bdev_name = bdev_name,
|
||||
.done = false,
|
||||
.rc = 0,
|
||||
};
|
||||
pthread_mutex_init(&boot.mu, NULL);
|
||||
pthread_cond_init(&boot.cv, NULL);
|
||||
|
||||
// 起专用 poller pthread for md_thread(并在该线程完成 bdev/blobstore 初始化)
|
||||
pthread_t md_poller_tid;
|
||||
if (pthread_create(&md_poller_tid, NULL, md_poller_fn, NULL) != 0) {
|
||||
if (pthread_create(&md_poller_tid, NULL, md_poller_fn, &boot) != 0) {
|
||||
SPDK_ERRLOG("pthread_create for md_poller failed\n");
|
||||
pthread_cond_destroy(&boot.cv);
|
||||
pthread_mutex_destroy(&boot.mu);
|
||||
g_engine_init_rc = -1;
|
||||
return g_engine_init_rc;
|
||||
}
|
||||
if (pthread_detach(md_poller_tid) != 0) {
|
||||
SPDK_ERRLOG("pthread_detach for md_poller failed\n");
|
||||
pthread_cond_destroy(&boot.cv);
|
||||
pthread_mutex_destroy(&boot.mu);
|
||||
g_engine_init_rc = -1;
|
||||
return g_engine_init_rc;
|
||||
}
|
||||
|
||||
// init bdev/bs
|
||||
g_super_blob_id_cache = SPDK_BLOBID_INVALID;
|
||||
int rc = open_bdev_and_init_bs(bdev_name);
|
||||
pthread_mutex_lock(&boot.mu);
|
||||
while (!boot.done) {
|
||||
pthread_cond_wait(&boot.cv, &boot.mu);
|
||||
}
|
||||
int rc = boot.rc;
|
||||
pthread_mutex_unlock(&boot.mu);
|
||||
pthread_cond_destroy(&boot.cv);
|
||||
pthread_mutex_destroy(&boot.mu);
|
||||
|
||||
if (rc != 0) {
|
||||
g_engine_init_rc = rc;
|
||||
return rc;
|
||||
@@ -283,19 +320,12 @@ static struct spdk_io_channel *get_current_channel(void) {
|
||||
return NULL;
|
||||
}
|
||||
|
||||
if (tls.thread) {
|
||||
spdk_thread_poll(tls.thread, 0, 0);
|
||||
if (ensure_current_spdk_thread("get_current_channel") != 0) {
|
||||
return NULL;
|
||||
}
|
||||
|
||||
if (!tls.thread) {
|
||||
char name[32];
|
||||
snprintf(name, sizeof(name), "worker_%lu", pthread_self());
|
||||
tls.thread = spdk_thread_create(name, NULL);
|
||||
if (!tls.thread) {
|
||||
SPDK_ERRLOG("spdk_thread_create failed\n");
|
||||
return NULL;
|
||||
}
|
||||
spdk_set_thread(tls.thread);
|
||||
if (tls.thread) {
|
||||
spdk_thread_poll(tls.thread, 0, 0);
|
||||
}
|
||||
|
||||
if (!tls.channel) {
|
||||
@@ -308,33 +338,107 @@ static struct spdk_io_channel *get_current_channel(void) {
|
||||
return tls.channel;
|
||||
}
|
||||
|
||||
static void put_current_channel(struct spdk_io_channel *ch) {
|
||||
if (!ch) {
|
||||
return;
|
||||
}
|
||||
spdk_put_io_channel(ch);
|
||||
if (tls.thread) {
|
||||
spdk_thread_poll(tls.thread, 0, 0);
|
||||
}
|
||||
if (tls.channel == ch) {
|
||||
tls.channel = NULL;
|
||||
}
|
||||
}
|
||||
|
||||
static void ensure_tls_cleanup_key(void) {
|
||||
(void)pthread_key_create(&g_tls_cleanup_key, tls_cleanup_destructor);
|
||||
}
|
||||
|
||||
static void tls_cleanup_destructor(void *arg) {
|
||||
(void)arg;
|
||||
if (!tls.thread || tls.thread == g_engine.md_thread) {
|
||||
return;
|
||||
}
|
||||
|
||||
spdk_set_thread(tls.thread);
|
||||
|
||||
if (tls.channel) {
|
||||
spdk_put_io_channel(tls.channel);
|
||||
tls.channel = NULL;
|
||||
}
|
||||
|
||||
spdk_thread_exit(tls.thread);
|
||||
const uint64_t deadline_ms = now_mono_ms() + ZVFS_WAIT_TIME;
|
||||
while (!spdk_thread_is_exited(tls.thread)) {
|
||||
spdk_thread_poll(tls.thread, 0, 0);
|
||||
if (now_mono_ms() >= deadline_ms) {
|
||||
SPDK_ERRLOG("worker tls thread exit timeout\n");
|
||||
break;
|
||||
}
|
||||
usleep(1000);
|
||||
}
|
||||
|
||||
if (spdk_thread_is_exited(tls.thread)) {
|
||||
spdk_thread_destroy(tls.thread);
|
||||
}
|
||||
tls.thread = NULL;
|
||||
pthread_setspecific(g_tls_cleanup_key, NULL);
|
||||
}
|
||||
|
||||
static int ensure_current_spdk_thread(const char *op) {
|
||||
pthread_once(&g_tls_cleanup_once, ensure_tls_cleanup_key);
|
||||
|
||||
if (!tls.thread) {
|
||||
char name[32];
|
||||
snprintf(name, sizeof(name), "worker_%lu", (unsigned long)pthread_self());
|
||||
tls.thread = spdk_thread_create(name, NULL);
|
||||
if (!tls.thread) {
|
||||
SPDK_ERRLOG("%s: spdk_thread_create failed\n", op);
|
||||
return -ENOMEM;
|
||||
}
|
||||
pthread_setspecific(g_tls_cleanup_key, (void *)1);
|
||||
}
|
||||
spdk_set_thread(tls.thread);
|
||||
return 0;
|
||||
}
|
||||
|
||||
// 通用 dispatch md op
|
||||
static int dispatch_md_op(struct md_op_ctx *ctx) {
|
||||
int rc = ensure_engine_ready(ctx->op_name ? ctx->op_name : "dispatch_md_op");
|
||||
if (rc != 0) {
|
||||
return rc;
|
||||
}
|
||||
|
||||
ctx->done = false;
|
||||
ctx->rc = 0;
|
||||
|
||||
spdk_thread_send_msg(g_engine.md_thread, md_op_cb, ctx);
|
||||
|
||||
return wait_done_volatile(&ctx->done, &ctx->rc, ctx->op_name);
|
||||
}
|
||||
|
||||
static int dispatch_md_op_quiet(struct md_op_ctx *ctx) {
|
||||
int rc = ensure_engine_ready(ctx->op_name ? ctx->op_name : "dispatch_md_op_quiet");
|
||||
rc = ensure_current_spdk_thread(ctx->op_name ? ctx->op_name : "dispatch_md_op");
|
||||
if (rc != 0) {
|
||||
return rc;
|
||||
}
|
||||
|
||||
ctx->done = false;
|
||||
ctx->rc = 0;
|
||||
struct md_op_ctx *async_ctx = malloc(sizeof(*async_ctx));
|
||||
if (!async_ctx) {
|
||||
return -ENOMEM;
|
||||
}
|
||||
*async_ctx = *ctx;
|
||||
__atomic_store_n(&async_ctx->done, false, __ATOMIC_RELAXED);
|
||||
async_ctx->rc = 0;
|
||||
|
||||
spdk_thread_send_msg(g_engine.md_thread, md_op_cb, ctx);
|
||||
rc = spdk_thread_send_msg(g_engine.md_thread, md_op_cb, async_ctx);
|
||||
|
||||
return wait_done_volatile_quiet(&ctx->done, &ctx->rc, ctx->op_name);
|
||||
if (rc != 0) {
|
||||
SPDK_ERRLOG("%s: spdk_thread_send_msg failed: %d\n", async_ctx->op_name, rc);
|
||||
free(async_ctx);
|
||||
return rc;
|
||||
}
|
||||
|
||||
rc = wait_done_volatile(&async_ctx->done, &async_ctx->rc, async_ctx->op_name);
|
||||
if (rc == -ETIMEDOUT) {
|
||||
SPDK_ERRLOG("%s timeout; keep async ctx alive to avoid UAF\n", async_ctx->op_name);
|
||||
return rc;
|
||||
}
|
||||
|
||||
*ctx = *async_ctx;
|
||||
free(async_ctx);
|
||||
return rc;
|
||||
}
|
||||
|
||||
static int ensure_engine_ready(const char *op) {
|
||||
@@ -438,111 +542,12 @@ static int open_bdev_and_init_bs(const char *bdev_name) {
|
||||
return 0;
|
||||
}
|
||||
|
||||
static void blob_get_super_cb(void *arg, spdk_blob_id blobid, int rc) {
|
||||
struct md_op_ctx *ctx = arg;
|
||||
ctx->rc = rc;
|
||||
ctx->super.blob_id = blobid;
|
||||
ctx->done = true;
|
||||
}
|
||||
|
||||
static void blob_set_super_cb(void *arg, int rc) {
|
||||
struct md_op_ctx *ctx = arg;
|
||||
ctx->rc = rc;
|
||||
ctx->done = true;
|
||||
}
|
||||
|
||||
static void blob_get_super_on_md(struct md_op_ctx *ctx) {
|
||||
spdk_bs_get_super(g_engine.bs, blob_get_super_cb, ctx);
|
||||
}
|
||||
|
||||
static void blob_set_super_on_md(struct md_op_ctx *ctx) {
|
||||
spdk_bs_set_super(g_engine.bs, ctx->super.blob_id, blob_set_super_cb, ctx);
|
||||
}
|
||||
|
||||
static int bs_get_super_id(spdk_blob_id *blob_id) {
|
||||
struct md_op_ctx ctx = {
|
||||
.fn = blob_get_super_on_md,
|
||||
.op_name = "blob get super",
|
||||
};
|
||||
ctx.super.blob_id = SPDK_BLOBID_INVALID;
|
||||
|
||||
int rc = dispatch_md_op_quiet(&ctx);
|
||||
if (rc != 0) {
|
||||
return rc;
|
||||
}
|
||||
*blob_id = ctx.super.blob_id;
|
||||
return 0;
|
||||
}
|
||||
|
||||
static int bs_set_super_id(spdk_blob_id blob_id) {
|
||||
struct md_op_ctx ctx = {
|
||||
.fn = blob_set_super_on_md,
|
||||
.op_name = "blob set super",
|
||||
};
|
||||
ctx.super.blob_id = blob_id;
|
||||
return dispatch_md_op(&ctx);
|
||||
}
|
||||
|
||||
struct zvfs_blob_handle *blob_get_super(void) {
|
||||
pthread_mutex_lock(&g_super_blob_mutex);
|
||||
|
||||
if (g_super_blob_id_cache != SPDK_BLOBID_INVALID) {
|
||||
struct zvfs_blob_handle *cached = blob_open(g_super_blob_id_cache);
|
||||
if (cached) {
|
||||
pthread_mutex_unlock(&g_super_blob_mutex);
|
||||
return cached;
|
||||
}
|
||||
g_super_blob_id_cache = SPDK_BLOBID_INVALID;
|
||||
}
|
||||
|
||||
spdk_blob_id super_id = SPDK_BLOBID_INVALID;
|
||||
int rc = bs_get_super_id(&super_id);
|
||||
if (rc == 0 && super_id != SPDK_BLOBID_INVALID) {
|
||||
g_super_blob_id_cache = super_id;
|
||||
struct zvfs_blob_handle *existing = blob_open(super_id);
|
||||
if (!existing) {
|
||||
g_super_blob_id_cache = SPDK_BLOBID_INVALID;
|
||||
}
|
||||
pthread_mutex_unlock(&g_super_blob_mutex);
|
||||
return existing;
|
||||
}
|
||||
if (rc == 0 && super_id == SPDK_BLOBID_INVALID) {
|
||||
rc = -ENOENT;
|
||||
}
|
||||
|
||||
if (rc != -ENOENT) {
|
||||
SPDK_ERRLOG("spdk_bs_get_super failed: %d\n", rc);
|
||||
pthread_mutex_unlock(&g_super_blob_mutex);
|
||||
return NULL;
|
||||
}
|
||||
|
||||
struct zvfs_blob_handle *created = blob_create(0);
|
||||
if (!created) {
|
||||
pthread_mutex_unlock(&g_super_blob_mutex);
|
||||
return NULL;
|
||||
}
|
||||
|
||||
rc = bs_set_super_id(created->id);
|
||||
if (rc != 0) {
|
||||
spdk_blob_id created_id = created->id;
|
||||
SPDK_ERRLOG("spdk_bs_set_super failed: %d\n", rc);
|
||||
blob_close(created);
|
||||
blob_delete(created_id);
|
||||
pthread_mutex_unlock(&g_super_blob_mutex);
|
||||
return NULL;
|
||||
}
|
||||
|
||||
g_super_blob_id_cache = created->id;
|
||||
pthread_mutex_unlock(&g_super_blob_mutex);
|
||||
return created;
|
||||
}
|
||||
|
||||
// blob_create
|
||||
static void blob_create_cb(void *arg, spdk_blob_id blobid, int rc) {
|
||||
struct md_op_ctx *ctx = arg;
|
||||
ctx->rc = rc;
|
||||
ctx->create.blob_id = blobid;
|
||||
ctx->done = true;
|
||||
__atomic_store_n(&ctx->done, true, __ATOMIC_RELEASE);
|
||||
}
|
||||
|
||||
static void blob_create_on_md(struct md_op_ctx *ctx) {
|
||||
@@ -556,13 +561,17 @@ struct zvfs_blob_handle *blob_create(uint64_t size_hint) {
|
||||
if(size_hint == 0) size_hint = g_engine.cluster_size;
|
||||
struct md_op_ctx ctx = {.fn = blob_create_on_md, .create.size_hint = size_hint, .op_name = "blob create"};
|
||||
int rc = dispatch_md_op(&ctx);
|
||||
if (rc) return NULL;
|
||||
if (rc) {
|
||||
errno = (rc < 0) ? -rc : EIO;
|
||||
return NULL;
|
||||
}
|
||||
|
||||
struct zvfs_blob_handle *handle = blob_open(ctx.create.blob_id);
|
||||
if (handle && size_hint > 0) {
|
||||
rc = blob_resize(handle, size_hint); // 初始 resize
|
||||
if (rc != 0) {
|
||||
SPDK_ERRLOG("blob_resize failed after create: %d\n", rc);
|
||||
errno = (rc < 0) ? -rc : EIO;
|
||||
blob_close(handle);
|
||||
return NULL;
|
||||
}
|
||||
@@ -570,6 +579,7 @@ struct zvfs_blob_handle *blob_create(uint64_t size_hint) {
|
||||
rc = blob_sync_md(handle);
|
||||
if (rc != 0) {
|
||||
SPDK_ERRLOG("blob_sync_md failed after resize: %d\n", rc);
|
||||
errno = (rc < 0) ? -rc : EIO;
|
||||
blob_close(handle);
|
||||
return NULL;
|
||||
}
|
||||
@@ -582,7 +592,7 @@ static void blob_open_cb(void *arg, struct spdk_blob *blob, int rc) {
|
||||
struct md_op_ctx *ctx = arg;
|
||||
ctx->rc = rc;
|
||||
ctx->open.blob = blob;
|
||||
ctx->done = true;
|
||||
__atomic_store_n(&ctx->done, true, __ATOMIC_RELEASE);
|
||||
}
|
||||
|
||||
static void blob_open_on_md(struct md_op_ctx *ctx) {
|
||||
@@ -594,7 +604,10 @@ static void blob_open_on_md(struct md_op_ctx *ctx) {
|
||||
struct zvfs_blob_handle *blob_open(uint64_t blob_id) {
|
||||
struct md_op_ctx ctx = {.fn = blob_open_on_md, .open.blob_id = blob_id, .op_name = "blob open"};
|
||||
int rc = dispatch_md_op(&ctx);
|
||||
if (rc) return NULL;
|
||||
if (rc) {
|
||||
errno = (rc < 0) ? -rc : EIO;
|
||||
return NULL;
|
||||
}
|
||||
|
||||
struct zvfs_blob_handle *handle = malloc(sizeof(*handle));
|
||||
if (!handle) return NULL;
|
||||
@@ -628,15 +641,18 @@ int blob_write(struct zvfs_blob_handle *handle, uint64_t offset, const void *buf
|
||||
spdk_thread_poll(tls.thread, 0, 0);
|
||||
}
|
||||
|
||||
if (len == 0) return 0;
|
||||
|
||||
struct spdk_io_channel *ch = get_current_channel();
|
||||
if (!ch) return -1;
|
||||
if (len == 0) return 0;
|
||||
int ret = 0;
|
||||
|
||||
// 越界检查
|
||||
if (offset + len > handle->size) {
|
||||
SPDK_ERRLOG("blob_write out of range: offset=%lu len=%zu blob_size=%lu\n",
|
||||
offset, len, handle->size);
|
||||
return -ERANGE;
|
||||
ret = -ERANGE;
|
||||
goto out;
|
||||
}
|
||||
|
||||
// 计算对齐后的 IO 范围和 dma_buf 内偏移
|
||||
@@ -646,13 +662,15 @@ int blob_write(struct zvfs_blob_handle *handle, uint64_t offset, const void *buf
|
||||
int rc = zvfs_calc_io_units(offset, len, g_engine.io_unit_size, &lba_off, &lba_len, &buf_off);
|
||||
if (rc != 0) {
|
||||
SPDK_ERRLOG("blob_write calc_io_units failed: %d\n", rc);
|
||||
return rc;
|
||||
ret = rc;
|
||||
goto out;
|
||||
}
|
||||
|
||||
size_t aligned_bytes = lba_len * g_engine.io_unit_size;
|
||||
if (aligned_bytes > ZVFS_DMA_BUF_SIZE) {
|
||||
SPDK_ERRLOG("blob_write aligned_bytes=%zu exceeds ZVFS_DMA_BUF_SIZE\n", aligned_bytes);
|
||||
return -ENOSPC;
|
||||
ret = -ENOSPC;
|
||||
goto out;
|
||||
}
|
||||
|
||||
struct io_completion_ctx io_ctx = {.done = false, .rc = 0};
|
||||
@@ -662,7 +680,10 @@ int blob_write(struct zvfs_blob_handle *handle, uint64_t offset, const void *buf
|
||||
|
||||
|
||||
rc = wait_done(&io_ctx.done, &io_ctx.rc, "io_write(read phase)");
|
||||
if (rc != 0) return rc;
|
||||
if (rc != 0) {
|
||||
ret = rc;
|
||||
goto out;
|
||||
}
|
||||
|
||||
memcpy((uint8_t *)handle->dma_buf + buf_off, buf, len);
|
||||
io_ctx.done = false;
|
||||
@@ -671,9 +692,15 @@ int blob_write(struct zvfs_blob_handle *handle, uint64_t offset, const void *buf
|
||||
spdk_blob_io_write(handle->blob, ch, handle->dma_buf, lba_off, lba_len,
|
||||
io_completion_cb, &io_ctx);
|
||||
rc = wait_done(&io_ctx.done, &io_ctx.rc, "io_write(write phase)");
|
||||
if (rc != 0) return rc;
|
||||
if (rc != 0) {
|
||||
ret = rc;
|
||||
goto out;
|
||||
}
|
||||
|
||||
return io_ctx.rc;
|
||||
ret = io_ctx.rc;
|
||||
out:
|
||||
put_current_channel(ch);
|
||||
return ret;
|
||||
}
|
||||
|
||||
// blob_read 类似
|
||||
@@ -681,16 +708,19 @@ int blob_read(struct zvfs_blob_handle *handle, uint64_t offset, void *buf, size_
|
||||
if (tls.thread) {
|
||||
spdk_thread_poll(tls.thread, 0, 0);
|
||||
}
|
||||
|
||||
|
||||
if (len == 0) return 0;
|
||||
|
||||
struct spdk_io_channel *ch = get_current_channel();
|
||||
if (!ch) return -1;
|
||||
if (len == 0) return 0;
|
||||
|
||||
int ret = 0;
|
||||
|
||||
// 越界检查
|
||||
if (offset + len > handle->size) {
|
||||
SPDK_ERRLOG("blob_read out of range: offset=%lu len=%zu blob_size=%lu\n",
|
||||
offset, len, handle->size);
|
||||
return -ERANGE;
|
||||
ret = -ERANGE;
|
||||
goto out;
|
||||
}
|
||||
|
||||
|
||||
@@ -701,14 +731,16 @@ int blob_read(struct zvfs_blob_handle *handle, uint64_t offset, void *buf, size_
|
||||
int rc = zvfs_calc_io_units(offset, len, g_engine.io_unit_size, &lba_off, &lba_len, &buf_off);
|
||||
if (rc != 0) {
|
||||
SPDK_ERRLOG("io_read offset/len not aligned to io_unit_size=%lu\n", g_engine.io_unit_size);
|
||||
return rc;
|
||||
ret = rc;
|
||||
goto out;
|
||||
}
|
||||
|
||||
// 读入对齐范围到 dma_buf,再从正确偏移处截取到用户 buf
|
||||
size_t aligned_bytes = lba_len * g_engine.io_unit_size;
|
||||
if (aligned_bytes > ZVFS_DMA_BUF_SIZE) {
|
||||
SPDK_ERRLOG("blob_read aligned_bytes=%zu exceeds ZVFS_DMA_BUF_SIZE\n", aligned_bytes);
|
||||
return -ENOSPC;
|
||||
ret = -ENOSPC;
|
||||
goto out;
|
||||
}
|
||||
|
||||
struct io_completion_ctx io_ctx = {.done = false, .rc = 0};
|
||||
@@ -717,17 +749,23 @@ int blob_read(struct zvfs_blob_handle *handle, uint64_t offset, void *buf, size_
|
||||
io_completion_cb, &io_ctx);
|
||||
|
||||
rc = wait_done(&io_ctx.done, &io_ctx.rc, "io_read");
|
||||
if (rc != 0) return rc;
|
||||
if (rc != 0) {
|
||||
ret = rc;
|
||||
goto out;
|
||||
}
|
||||
|
||||
memcpy(buf, (uint8_t *)handle->dma_buf + buf_off, len);
|
||||
return io_ctx.rc;
|
||||
ret = io_ctx.rc;
|
||||
out:
|
||||
put_current_channel(ch);
|
||||
return ret;
|
||||
}
|
||||
|
||||
// blob_resize
|
||||
static void blob_resize_cb(void *arg, int rc) {
|
||||
struct md_op_ctx *ctx = arg;
|
||||
ctx->rc = rc;
|
||||
ctx->done = true;
|
||||
__atomic_store_n(&ctx->done, true, __ATOMIC_RELEASE);
|
||||
}
|
||||
|
||||
static void blob_resize_on_md(struct md_op_ctx *ctx) {
|
||||
@@ -736,7 +774,7 @@ static void blob_resize_on_md(struct md_op_ctx *ctx) {
|
||||
int rc = zvfs_calc_ceil_units(ctx->handle_op.new_size, cluster_size, &new_clusters);
|
||||
if (rc != 0) {
|
||||
ctx->rc = rc;
|
||||
ctx->done = true;
|
||||
__atomic_store_n(&ctx->done, true, __ATOMIC_RELEASE);
|
||||
return;
|
||||
}
|
||||
spdk_blob_resize(ctx->handle_op.handle->blob, new_clusters, blob_resize_cb, ctx);
|
||||
@@ -759,7 +797,7 @@ int blob_resize(struct zvfs_blob_handle *handle, uint64_t new_size) {
|
||||
static void blob_sync_md_cb(void *arg, int rc) {
|
||||
struct md_op_ctx *ctx = arg;
|
||||
ctx->rc = rc;
|
||||
ctx->done = true;
|
||||
__atomic_store_n(&ctx->done, true, __ATOMIC_RELEASE);
|
||||
}
|
||||
|
||||
static void blob_sync_md_on_md(struct md_op_ctx *ctx) {
|
||||
@@ -776,7 +814,7 @@ int blob_sync_md(struct zvfs_blob_handle *handle) {
|
||||
static void blob_close_cb(void *arg, int rc) {
|
||||
struct md_op_ctx *ctx = arg;
|
||||
ctx->rc = rc;
|
||||
ctx->done = true;
|
||||
__atomic_store_n(&ctx->done, true, __ATOMIC_RELEASE);
|
||||
}
|
||||
|
||||
static void blob_close_on_md(struct md_op_ctx *ctx) {
|
||||
@@ -798,7 +836,7 @@ int blob_close(struct zvfs_blob_handle *handle) {
|
||||
static void blob_delete_cb(void *arg, int rc) {
|
||||
struct md_op_ctx *ctx = arg;
|
||||
ctx->rc = rc;
|
||||
ctx->done = true;
|
||||
__atomic_store_n(&ctx->done, true, __ATOMIC_RELEASE);
|
||||
}
|
||||
|
||||
static void blob_delete_on_md(struct md_op_ctx *ctx) {
|
||||
|
||||
Reference in New Issue
Block a user