chfs源码分析

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

CHFS：Parallel Consistent Hashing File System for Node-local Persistent Memory 一种临时文件系统

无名高

Comments | 1 条评论