pmemkv.c文件
主要操作为将转化好的key和value存入/取出nvm
libpmemkv
pmemkv github
void
kv_init(char *db_dir, char *engine, char *path, size_t size)
{
pmemkv_config *cfg;
char *p;
int r;
struct stat sb;
p = malloc(strlen(db_dir) + 1 + strlen(path) + 1);
stat(db_dir, &sb);
cfg = pmemkv_config_new();
r = pmemkv_config_put_string(cfg, "path", p);
r = pmemkv_open(engine, cfg, &db); //初始化static变量db
if (r == PMEMKV_STATUS_OK)
goto free_p;
else 再做一次初始化的步骤↓
cfg = pmemkv_config_new();
r = pmemkv_config_put_string(cfg, "path", p);
r = pmemkv_config_put_uint64(cfg, "size", size);
r = pmemkv_config_put_uint64(cfg, "force_create", 1);
r = pmemkv_open(engine, cfg, &db);
free_p: free(p)
}
void
kv_term()
{
pmemkv_close(db);
}
int
kv_put(void *key, size_t key_size, void *value, size_t value_size)
{
return (kv_err(pmemkv_put(db, key, key_size, value, value_size)));
}
int
kv_get(void *key, size_t key_size, void *value, size_t *value_size)
{
r = pmemkv_get_copy(db, key, key_size, value, *value_size, &o_size);已知key获取value
if (r == PMEMKV_STATUS_OK) *value_size = o_size;
}
int kv_get_cb(void *key, size_t key_size,
void (*cb)(const char *, size_t, void *), void *arg)
{
return (kv_err(pmemkv_get(db, key, key_size, cb, arg)));
// cb为函数指针,明确这里获取了什么内容以及与kv_get的用到的pmemkv_get_copy区别
// 目前kv_get_cb理解为工具函数,pmemkv_get通过key获取value,然后在cb处传入不同功能的回调函数,实现不同功能
}
struct arg {
void *value;
size_t off, size;
int err;
}; // 用于回调中的自定义结构体
static struct arg *
create_arg(size_t off, void *value, size_t value_size) 创建arg并返回结构体
{
struct arg *a;
a = malloc(sizeof(*a));
if (a == NULL)
return (NULL);
a->off = off;
a->value = value;
a->size = value_size;
return (a);
}
static void
update_cb(const char *v, size_t size, void *a)
{ // 疑问:这里的v是a->value吗? 答:这里的v是实际从db取出的值
struct arg *arg = a; // 空指针a为结构体arg,拷贝a指向的value
char *value = (char *)v; // value也指向v
if (arg->off > size)
arg->size = 0;
else if (arg->off + arg->size > size)
arg->size = size - arg->off;
if (arg->size == 0)
return;
value += arg->off; // value指针位置向后移动arg->off字节
memcpy(value, arg->value, arg->size); // 再把arg->value拷贝arg->size个字节到value目前指向的位置
}
int
kv_update(void *key, size_t key_size,
size_t off, void *value, size_t *value_size)
{ // 更新保存在key-value数据库中的数据
struct arg *a = create_arg(off, value, *value_size);
int r;
if (a == NULL)
return (PMEMKV_STATUS_OUT_OF_MEMORY);
r = kv_get_cb(key, key_size, update_cb, a); // 这里a是用新值创建的struct arg类型,update_cb是函数指针
if (r == PMEMKV_STATUS_OK)
*value_size = a->size;
free(a);
return (r);
}
static void
pget_cb(const char *v, size_t size, void *a)
{ // 与update_cb的却别在于这里吧value往后arg->size个字节拷贝到arg->value的位置
struct arg *arg = a;
char *value = (char *)v;
value += arg->off;
memcpy(arg->value, value, arg->size); // 与update_cb区别在这行
}
int
kv_pget(void *key, size_t key_size, size_t off, void *value, size_t *value_size)
{
struct arg *a = create_arg(off, value, *value_size);
r = kv_get_cb(key, key_size, pget_cb, a);
if (r == PMEMKV_STATUS_OK)
*value_size = a->size;
free(a);
return (r);
}
// 获取value_size的回调
static void
get_size_cb(const char *v, size_t size, void *a)
{
size_t *s = a;
*s = size;
}
int
kv_get_size(void *key, size_t key_size, size_t *value_size)
{
return (kv_get_cb(key, key_size, get_size_cb, value_size));
}
// 删除操作
int
kv_remove(void *key, size_t key_size)
{
return (kv_err(pmemkv_remove(db, key, key_size)));
}
// cb会被循环迭代,每次获取一个key-value对,参数为key key_size value value_size arg
int
kv_get_all_cb(int (*cb)(const char *, size_t, const char *, size_t, void *),
void *arg)
{
log_debug("local pmem get all cb");
return (kv_err(pmemkv_get_all(db, cb, arg)));
}
fs_kv.c 文件
struct inode {
uint32_t mode;
uint32_t uid, gid;
uint32_t msize;
uint64_t size;
uint64_t chunk_size;
struct timespec mtime, ctime;
};
static int32_t fs_msize = sizeof(struct inode);
// 创建inode,把文件元数据和buf里的文件数据合并在一块连续的地址空间
static struct inode *
create_inode_all(uint32_t uid, uint32_t gid, uint32_t mode, size_t chunk_size,
struct timespec *mtime, struct timespec *ctime,
const void *buf, size_t size, off_t offset)
{
struct inode *inode;
if (offset > chunk_size)
return (NULL);
if (offset + size > chunk_size)
size = chunk_size - offset;
if (size == 0)
offset = 0;
if (buf == NULL) // 注意这里inode指针实际分配了fs_msize + chunk_size字节的内存
inode = calloc(fs_msize + chunk_size, 1);
else
inode = malloc(fs_msize + chunk_size);
if (inode == NULL)
return (NULL);
inode->mode = mode;
inode->uid = uid;
inode->gid = gid;
inode->msize = fs_msize;
inode->size = offset + size;
inode->chunk_size = chunk_size;
inode->mtime = *mtime;
inode->ctime = *ctime;
if (buf) {
void *base = (void *)inode + fs_msize;
if (offset > 0)
memset(base, 0, offset);
memcpy(base + offset, buf, size);
if (chunk_size > offset + size)
memset(base + offset + size, 0,
chunk_size - offset - size);
}
return (inode);
}
// 上面的函数创建好的指针,作为键值的value存入数据库
static int
fs_inode_create_data(char *key, size_t key_size, uint32_t uid, uint32_t gid,
mode_t mode, size_t chunk_size, const void *buf, size_t size, off_t off)
{
struct inode *inode;
int r;
static const char diag[] = "fs_inode_create";
inode = create_inode(uid, gid, mode, chunk_size, buf, size, off);
if (inode == NULL)
return (KV_ERR_NO_MEMORY);
r = kv_put(key, key_size, inode, fs_msize + chunk_size);
free(inode);
if (r != KV_SUCCESS)
log_error("%s: %s", diag, kv_err_string(r));
return (r);
}
// stat实现
int
fs_inode_stat(char *key, size_t key_size, struct fs_stat *stat)
{
struct inode inode;
size_t s;
int r;
s = fs_msize;
r = kv_pget(key, key_size, 0, &inode, &s);
if (r != KV_SUCCESS)
return (r);
if (fs_msize != inode.msize)
return (KV_ERR_METADATA_SIZE_MISMATCH);
stat->mode = inode.mode;
stat->uid = inode.uid;
stat->gid = inode.gid;
stat->size = inode.size;
stat->chunk_size = inode.chunk_size;
stat->mtime = inode.mtime;
stat->ctime = inode.ctime;
return (KV_SUCCESS);
}
// 更新size
static int
fs_inode_update_size(char *key, size_t key_size, size_t size)
{
size_t ss;
int r;
static const char diag[] = "fs_inode_update_size";
ss = sizeof(size);
r = kv_update(key, key_size, offsetof(struct inode, size), &size, &ss);
if (r != KV_SUCCESS)
log_error("%s: %s", diag, kv_err_string(r));
return (r);
}
// 写数据操作
int
fs_inode_write(char *key, size_t key_size, const void *buf, size_t *size,
off_t offset, mode_t mode, size_t chunk_size)
{
struct inode inode;
size_t s = fs_msize;
int r;
static const char diag[] = "fs_inode_write";
r = kv_pget(key, key_size, 0, &inode, &s);
if (r != KV_SUCCESS) {
r = fs_inode_create_data(key, key_size, 0, 0, mode, chunk_size,
buf, *size, offset);
if (r != KV_SUCCESS)
log_error("%s: %s", diag, kv_err_string(r));
return (r);
}
r = kv_update(key, key_size, fs_msize + offset, (void *)buf, size);
if (r == KV_SUCCESS) {
s = offset + *size;
if (inode.size < s)
r = fs_inode_update_size(key, key_size, s);
}
if (r != KV_SUCCESS)
log_error("%s: %s", diag, kv_err_string(r));
return (r);
}
fs_inode_read
fs_inode_truncate
fs_inode_remove
Comments | 1 条评论