lian/ldb
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

图片修改

Browse Source
This commit is contained in:
1iaan
2026-01-16 07:46:25 +00:00
parent 7525492ee1
commit e404554363
17 changed files with 278 additions and 72 deletions
Download Patch File Download Diff File Expand all files Collapse all files

5
Makefile
View File

@@ -1,6 +1,7 @@
CC = gcc
FLAGS = -I ./NtyCo/core/ -L ./NtyCo/ -lntyco -lpthread -luring -ldl -ljemalloc
# FLAGS = -g -DJEMALLOC_NO_DEMANGLE -I ./NtyCo/core/ -L ./NtyCo/ -lntyco -lpthread -luring -ldl -ljemalloc
FLAGS = -g -DJEMALLOC_NO_DEMANGLE -I ./NtyCo/core/ -L ./NtyCo/ -lntyco -lpthread -luring -ldl
# SRCS = kvstore.c ntyco.c proactor.c reactor.c kvs_array.c kvs_rbtree.c kvs_hash.c kvs_rw_tools.c
SRCS = kvstore.c ntyco.c proactor.c reactor.c kvs_array_bin.c kvs_rbtree_bin.c kvs_hash_bin.c kvs_rw_tools.c kvs_cmd_log.c ./mem_pool/mem_pool.c kvs_slave.c
TESTCASE_SRCS = testcase.c
@@ -23,7 +24,7 @@ ECHO:
@echo $(SUBDIR)
$(TARGET): $(OBJS)
$(CC) -g -o $@ $^ $(FLAGS)
$(CC) -o $@ $^ $(FLAGS)
$(TESTCASE): $(TESTCASE_SRCS)
$(CC) -g -o $@ $^

42
README.md
View File

@@ -15,35 +15,27 @@ save() -> 全保存数据集。
虚拟内存的占用情况 htop。插入百万条数据集(KV*100w，set 200w del 100w delete 200w set 100w)。
##### 单条插入
未运行状态 VIRT 24.8M
未运行状态 VIRT 24.8M RES 4004
![alt text](img/未运行状态.png)
一次执行100条，执行60000次。
malloc：\
--> time_used=192898 ms, ops=6000000, qps=31104\
VIRT 377M RES 367M
jemalloc：\
--> time_used=196866 ms, ops=6000000, qps=30477\
VIRT 433M RES 377M --> 11096
自实现内存池：\
--> time_used=180338 ms, ops=6000000, qps=33270\
VIRT 721M RES 628M
--> time_used=170471 ms, ops=6000000, qps=35196\
VIRT 567M RES 558M
无内存池: --> time_used=1488789 ms, ops=6000000, qps=4030\
VIRT 105M
![alt text](img/无内存池200w插入100w删除.png)
自实现内存池: --> time_used=1439444 ms, ops=6000000, qps=4168\
VIRT 135M
![alt text](img/mpalloc200w插入100w删除.png)
![alt text](img/jemalloc2000w插入100w删除.png)
jemalloc: time_used=1502797 ms, ops=6000000, qps=3992\
VIRT 99M
##### 批处理
一次执行100条。
无内存池：--> time_used=142913 ms, ops=6000000, qps=41983\
VIRT 99M
自实现内存池： --> time_used=122754 ms, ops=6000000, qps=48878\
VIRT 135M
jemalloc：--> time_used=122206 ms, ops=6000000, qps=49097\
VIRT 99M
性能分析：
1. key :

BIN
img/jemalloc2000w插入100w删除.png
View File

Binary file not shown.
Side by Side

Before

Width:  |  Height:  |  Size: 94 KiB

BIN
img/mpalloc200w插入100w删除.png
View File

Binary file not shown.
Side by Side

Before

Width:  |  Height:  |  Size: 82 KiB

BIN
img/批处理1jemalloc.png Normal file
View File

Binary file not shown.
Side by Side

After

Width:  |  Height:  |  Size: 103 KiB

BIN
img/批处理1内存池.png Normal file
View File

Binary file not shown.
Side by Side

After

Width:  |  Height:  |  Size: 107 KiB

BIN
img/批处理1无内存池.png Normal file
View File

Binary file not shown.
Side by Side

After

Width:  |  Height:  |  Size: 104 KiB

BIN
img/批处理jemalloc.png
View File

Binary file not shown.
Side by Side

Before

Width:  |  Height:  |  Size: 93 KiB

BIN
img/批处理无内存池.png
View File

Binary file not shown.
Side by Side

Before

Width:  |  Height:  |  Size: 95 KiB

BIN
img/批处理自实现内存池.png
View File

Binary file not shown.
Side by Side

Before

Width:  |  Height:  |  Size: 90 KiB

BIN
img/无内存池200w插入100w删除.png
View File

Binary file not shown.
Side by Side

Before

Width:  |  Height:  |  Size: 26 KiB

20
kvs_slave.c
View File

@@ -25,6 +25,21 @@ static int kvs_write_u32(uint8_t **pp, uint32_t v) {
return 0;
}
static uint64_t kvs_get_log_tail_offset(void) {
int fd = open("kvs_cmd_log.db", O_RDONLY);
if (fd < 0) {
// 文件不存在：从 0 开始同步
if (errno == ENOENT) return 0;
// 其他错误：保守起见从 0 开始
return 0;
}
off_t end = lseek(fd, 0, SEEK_END);
close(fd);
if (end < 0) return 0;
return (uint64_t)end; // 指向 EOF（下一次写入的位置）
}
int try_connect_master(char *ip, int port){
@@ -53,10 +68,9 @@ int try_connect_master(char *ip, int port){
uint64_t len = sizeof(uint64_t);
kvs_write_u32((uint8_t**)&p, len);
uint64_t offset = 0;
uint64_t offset = kvs_get_log_tail_offset();
memcpy(p, (void*)&offset, len);
p += len;
p += sizeof(offset);
send(fd, buf, p-buf, 0);
recv(fd, buf, 100, 0);

9
kvstore.c
View File

@@ -605,6 +605,7 @@ void build_thread_to_sync(const uint8_t *offset, struct conn* conn){
pthread_mutex_lock(&conn->g_sync_lock);
g_slavefd = conn->fd;
g_offset = (uint64_t)off64;
printf("offset:%ld\n", off64);
conn->wlength = 0;
pthread_mutex_unlock(&conn->g_sync_lock);
@@ -683,6 +684,14 @@ void dest_memory_pool(void){
* ./kvstore slave 7000 192.168.10.129 8888
*/
int main(int argc, char *argv[]) {
printf("memory type: ");
if(MEMORY_SELECT_MALLOC == MEMORY_USE_DEFAULT){
printf("default\n");
}else if(MEMORY_SELECT_MALLOC == MEMORY_USE_MYMALLOC){
printf("my memory pool\n");
}else if(MEMORY_SELECT_MALLOC == MEMORY_USE_JEMALLOC){
printf("jemalloc\n");
}
if (argc < 3) return -1;

42
mem_pool/mem_pool.c
View File

@@ -1,9 +1,10 @@
#include "mem_pool.h"
#include <stddef.h>
#include <stdlib.h>
#define JEMALLOC_NO_DEMANGLE
#include <stdio.h>
#include <jemalloc/jemalloc.h>
void *kvs_malloc(size_t size) {
#if MEMORY_SELECT_MALLOC == MEMORY_USE_DEFAULT
return malloc(size);
@@ -59,7 +60,7 @@ size_t mp_user_size_from_class(uint16_t cid) {
}
int mp_add_page(mp_pool_t* p, void* mem) {
mp_page_t* pg = (mp_page_t*)malloc(sizeof(mp_page_t));
mp_page_t* pg = (mp_page_t*)je_malloc(sizeof(mp_page_t));
if (!pg) return -1;
pg->mem = mem;
pg->next = p->pages;
@@ -68,28 +69,43 @@ int mp_add_page(mp_pool_t* p, void* mem) {
}
int mp_grow(mp_pool_t* p, uint16_t cid) {
void* page = malloc(MP_PAGE_SIZE);
void* page = je_malloc(MP_PAGE_SIZE);
if (!page) return -1;
if (mp_add_page(p, page) != 0) { free(page); return -1; }
if (mp_add_page(p, page) != 0) { je_free(page); return -1; }
// cid对应的空间大小
const size_t user_sz = mp_user_size_from_class(cid);
const size_t block_sz = sizeof(mp_blk_hdr_t) + user_sz;
const size_t n = MP_PAGE_SIZE / block_sz;
if (n == 0) return -1;
size_t n = MP_PAGE_SIZE / block_sz;
if (n == 0) { je_free(page); return -1; }
uint8_t* cur = (uint8_t*)page;
mp_blk_hdr_t* first_hdr = (mp_blk_hdr_t*)cur;
first_hdr->magic = MP_MAGIC;
first_hdr->class_id = cid;
mp_free_node_t* first_node = (mp_free_node_t*)(first_hdr + 1);
mp_free_node_t* head = first_node;
mp_free_node_t* tail = first_node;
cur += block_sz;
n--;
for (size_t i = 0; i < n; ++i) {
mp_blk_hdr_t* h = (mp_blk_hdr_t*)cur;
h->magic = MP_MAGIC;
h->class_id = cid;
mp_free_node_t* user = (mp_free_node_t*)(h + 1);
user->next = p->buckets[cid].free_list;
p->buckets[cid].free_list = user;
mp_free_node_t* node = (mp_free_node_t*)(h + 1);
tail->next = node;
tail = node;
cur += block_sz;
}
tail->next = p->buckets[cid].free_list;
p->buckets[cid].free_list = head;
return 0;
}
@@ -104,8 +120,8 @@ void mp_destroy(mp_pool_t* p){
mp_page_t* pg = p->pages;
while (pg) {
mp_page_t* nxt = pg->next;
free(pg->mem);
free(pg);
je_free(pg->mem);
je_free(pg);
pg = nxt;
}
p->pages = NULL;
@@ -119,7 +135,7 @@ void* mp_alloc(size_t size){
uint16_t cid = mp_class_id_for(size);
if (cid == MP_LARGE_CLASS) {
mp_blk_hdr_t* h = (mp_blk_hdr_t*)malloc(sizeof(mp_blk_hdr_t) + size);
mp_blk_hdr_t* h = (mp_blk_hdr_t*)je_malloc(sizeof(mp_blk_hdr_t) + size);
if (!h) return NULL;
h->magic = MP_MAGIC;
h->class_id = MP_LARGE_CLASS;
@@ -149,7 +165,7 @@ void mp_free(void* ptr){
if (h->class_id == MP_LARGE_CLASS) {
/* 大对象：直接 free 整块（从 header 起） */
free(h);
je_free(h);
return;
}

7
mem_pool/mem_pool.h
View File

@@ -5,7 +5,7 @@
#define MEMORY_USE_MYMALLOC 1
#define MEMORY_USE_JEMALLOC 2
#define MEMORY_SELECT_MALLOC MEMORY_USE_JEMALLOC
#define MEMORY_SELECT_MALLOC MEMORY_USE_DEFAULT
#include <stddef.h>
#include <stdint.h>
@@ -14,7 +14,7 @@ void kvs_free(void *ptr);
#define MP_PAGE_SIZE 4096
#define MP_PAGE_SIZE (16*1024)
#define MP_MIN_SHIFT 3
#define MP_MAX_SHIFT 12
@@ -26,9 +26,10 @@ typedef struct mp_free_node {
struct mp_free_node* next;
} mp_free_node_t;
// (blk_hdr + user_data)
typedef struct mp_blk_hdr {
uint32_t magic;
uint16_t class_id; /* 0..MP_NBUCKETS-1, or MP_LARGE_CLASS */
uint16_t class_id;
uint16_t reserved;
} mp_blk_hdr_t;

2
test/test_client.h
View File

@@ -51,6 +51,8 @@ enum {
KVS_CMD_HDEL,
KVS_CMD_HMOD,
KVS_CMD_HEXIST,
KVS_CMD_PSYNC,
KVS_CMD_SAVE,
KVS_CMD_COUNT,

223
test/testcase.c
View File

@@ -156,11 +156,11 @@ void hash_testcase_1w(int connfd) {
}
void do_batch_test(int fd, int op, const char *key, const char *value){
void do_batch_test(int fd, int op, const char *key, const char *value, rsp_ret_status_e st, const char *rsp_value){
kvs_batch_t batch;
kvs_batch_init(&batch);
char bkey[15]={0}, bval[15]={0};
char bkey[256]={0}, bval[256]={0};
// 组 batch（最多 64 条）
for(int i = 0;i < 100; ++ i){
@@ -184,18 +184,70 @@ void do_batch_test(int fd, int op, const char *key, const char *value){
int nrsp = kvs_batch_recv_parse(fd, &batch, rsps, recvbuf, sizeof(recvbuf));
// 打印/处理
for (int i = 0; i < nrsp; i++) {
int len = sprintf(bkey, "PRT%d", i);
PRESP(bkey, &rsps[i]);
for (int i = 0; i < nrsp; i++) {
print_response(bkey, &rsps[i]);
int vlen;
if(rsp_value != NULL) vlen = sprintf(bval, "%s%d", rsp_value, i);
else vlen = 0;
verify_response(&rsps[i], op, st, bval, vlen);
}
}
void array_testcase_1w_batch(int connfd) {
kvs_batch_t batch;
kvs_batch_init(&batch);
int count = 1000;
int i = 0;
struct timeval tv_begin;
gettimeofday(&tv_begin, NULL);
for (i = 0;i < count;i ++) {
batch.cnt = 0;
batch.len = 0;
kvs_batch_add(&batch, KVS_CMD_SET, "name", 4, "l\r\0n", 4);
kvs_batch_add(&batch, KVS_CMD_GET, "name", 4, NULL, 0);
kvs_batch_add(&batch, KVS_CMD_MOD, "name", 4, "liu", 3);
kvs_batch_add(&batch, KVS_CMD_GET, "name", 4, NULL, 0);
kvs_batch_add(&batch, KVS_CMD_EXIST, "name", 4, NULL, 0);
kvs_batch_add(&batch, KVS_CMD_DEL, "name", 4, NULL, 0);
kvs_batch_add(&batch, KVS_CMD_EXIST, "name", 4, NULL, 0);
kvs_batch_add(&batch, KVS_CMD_MOD, "stu", 3, "liu", 3);
kvs_batch_add(&batch, KVS_CMD_DEL, "stu", 3, NULL, 0);
kvs_batch_send(connfd, &batch);
uint8_t recvbuf[BATCH_SIZE];
kvs_response_t rsps[KVS_BATCH_MAX];
int nrsp = kvs_batch_recv_parse(connfd, &batch, rsps, recvbuf, sizeof(recvbuf));
}
struct timeval tv_end;
gettimeofday(&tv_end, NULL);
int time_used = TIME_SUB_MS(tv_end, tv_begin); // ms
printf("array testcase --> time_used: %d, qps: %d\n", time_used, 9000 * 1000 / time_used);
}
void batch_qps(int connfd) {
const int N = 1000000;
const int B = 100; // do_batch_test() 里写死 50
const char *valA = "va";
const char *valB = "vb";
const char *valC = "vc";
static char valA[256];
static char valB[256];
static char valC[256];
static int inited = 0;
if (!inited) {
// 填充 255 字节，最后补 '\0'
memset(valA, 'A', 255); valA[255] = '\0';
memset(valB, 'B', 255); valB[255] = '\0';
memset(valC, 'C', 255); valC[255] = '\0';
inited = 1;
}
struct timeval tv_begin, tv_end;
gettimeofday(&tv_begin, NULL);
@@ -214,9 +266,9 @@ void batch_qps(int connfd) {
snprintf(preA, sizeof(preA), "A%d_", base/100);
snprintf(preB, sizeof(preB), "B%d_", base/100);
do_batch_test(connfd, KVS_CMD_RSET, preA, valA); // 50次 RSET A
do_batch_test(connfd, KVS_CMD_RSET, preB, valB); // 50次 RSET B
do_batch_test(connfd, KVS_CMD_RDEL, preA, NULL); // 50次 RDEL A
do_batch_test(connfd, KVS_CMD_RSET, preA, valA, KVS_STATUS_OK, NULL); // 50次 RSET A
do_batch_test(connfd, KVS_CMD_RSET, preB, valB, KVS_STATUS_OK, NULL); // 50次 RSET B
do_batch_test(connfd, KVS_CMD_RDEL, preA, NULL, KVS_STATUS_OK, NULL); // 50次 RDEL A
if (base % 10000 == 0) printf("P1 base:%d\n", base);
}
@@ -229,9 +281,9 @@ void batch_qps(int connfd) {
snprintf(preB, sizeof(preB), "B%d_", base/100);
snprintf(preC, sizeof(preC), "C%d_", base/100);
do_batch_test(connfd, KVS_CMD_RSET, preC, valC); // 50次 RSET C
do_batch_test(connfd, KVS_CMD_RDEL, preB, NULL); // 50次 RDEL B
do_batch_test(connfd, KVS_CMD_RDEL, preC, NULL); // 50次 RDEL C
do_batch_test(connfd, KVS_CMD_RSET, preC, valC, KVS_STATUS_OK, NULL); // 50次 RSET C
do_batch_test(connfd, KVS_CMD_RDEL, preB, NULL, KVS_STATUS_OK, NULL); // 50次 RDEL B
do_batch_test(connfd, KVS_CMD_RDEL, preC, NULL, KVS_STATUS_OK, NULL); // 50次 RDEL C
if (base % 10000 == 0) printf("P2 base:%d\n", base);
}
@@ -320,6 +372,118 @@ void testcase_add2_del1_then_add1_del2_100w(int connfd) {
N, time_used, ops, qps);
}
void send_spec_chars(int connfd){
const char *v_ws = "li an\tok\nend"; /* 内容含空格、\t、\n */
int v_ws_len = 12; /* 手算：l i ' ' a n \t o k \n e n d = 12 */
testcase(connfd, KVS_CMD_RSET,
"ws", 2,
v_ws, v_ws_len,
KVS_STATUS_OK,
NULL, 0,
"RSET WHITESPACE");
testcase(connfd, KVS_CMD_RGET,
"ws", 2,
NULL, 0,
KVS_STATUS_OK,
v_ws, v_ws_len,
"RGET WHITESPACE");
/* 2) 引号与反斜杠：测试是否被错误转义 */
const char *v_quote = "he\"llo\\world'!";
/* 字节数：h e " l l o \ w o r l d ' ! = 15 */
int v_quote_len = 15;
testcase(connfd, KVS_CMD_RSET,
"quote", 5,
v_quote, v_quote_len,
KVS_STATUS_OK,
NULL, 0,
"RSET QUOTE BACKSLASH");
testcase(connfd, KVS_CMD_RGET,
"quote", 5,
NULL, 0,
KVS_STATUS_OK,
v_quote, v_quote_len,
"RGET QUOTE BACKSLASH");
/* 3) 分隔符字符：冒号/逗号/分号/竖线 */
const char *v_sep = "a:b,c;d|e";
int v_sep_len = 9; /* a : b , c ; d | e = 9 */
testcase(connfd, KVS_CMD_RSET,
"sep", 3,
v_sep, v_sep_len,
KVS_STATUS_OK,
NULL, 0,
"RSET SEPARATORS");
testcase(connfd, KVS_CMD_RGET,
"sep", 3,
NULL, 0,
KVS_STATUS_OK,
v_sep, v_sep_len,
"RGET SEPARATORS");
/* 4) CRLF：\r\n 组合（最容易把一条请求拆错/响应拼错） */
const char *v_crlf = "line1\r\nline2";
int v_crlf_len = 12; /* line1(5) + \r(1) + \n(1) + line2(5) = 12 */
testcase(connfd, KVS_CMD_RSET,
"crlf", 4,
v_crlf, v_crlf_len,
KVS_STATUS_OK,
NULL, 0,
"RSET CRLF");
testcase(connfd, KVS_CMD_RGET,
"crlf", 4,
NULL, 0,
KVS_STATUS_OK,
v_crlf, v_crlf_len,
"RGET CRLF");
/* 5) 二进制数据：包含 \0，必须按长度处理，不能用 strlen */
char v_bin[] = { 'A', 0x00, 'B', 'C', 0x00, 'D' };
int v_bin_len = 6;
testcase(connfd, KVS_CMD_RSET,
"bin", 3,
v_bin, v_bin_len,
KVS_STATUS_OK,
NULL, 0,
"RSET BINARY WITH NUL");
testcase(connfd, KVS_CMD_RGET,
"bin", 3,
NULL, 0,
KVS_STATUS_OK,
v_bin, v_bin_len,
"RGET BINARY WITH NUL");
/* 6) UTF-8：中文 + emoji（测试多字节） */
const char *v_utf8 = "中文🙂";
/* "中"(3字节) + "文"(3字节) + "🙂"(4字节) = 10字节 */
int v_utf8_len = 10;
testcase(connfd, KVS_CMD_RSET,
"utf8", 4,
v_utf8, v_utf8_len,
KVS_STATUS_OK,
NULL, 0,
"RSET UTF8");
testcase(connfd, KVS_CMD_RGET,
"utf8", 4,
NULL, 0,
KVS_STATUS_OK,
v_utf8, v_utf8_len,
"RGET UTF8");
}
int main(int argc, char *argv[]) {
if (argc != 4) {
@@ -336,33 +500,40 @@ int main(int argc, char *argv[]) {
if(mode == 0){
array_testcase_1w(connfd);
}else if(mode == 1){
rbtree_testcase_1w(connfd);
array_testcase_1w_batch(connfd);
}else if(mode == 2){
hash_testcase_1w(connfd);
rbtree_testcase_1w(connfd);
}else if(mode == 3){
hash_testcase_1w(connfd);
}else if(mode == 4){
testcase_add2_del1_then_add1_del2_100w(connfd);
}else if(mode == 10){
do_batch_test(connfd, KVS_CMD_SET, "array_set", "array_val");
do_batch_test(connfd, KVS_CMD_SET, "array_set", "array_val", KVS_STATUS_OK, NULL);
}else if(mode == 11){
do_batch_test(connfd, KVS_CMD_GET, "array_set", NULL);
do_batch_test(connfd, KVS_CMD_GET, "array_set", NULL, KVS_STATUS_OK, "array_val");
}else if(mode == 12){
do_batch_test(connfd, KVS_CMD_EXIST, "array_set", NULL);
do_batch_test(connfd, KVS_CMD_EXIST, "array_set", NULL, KVS_STATUS_EXIST, NULL);
}else if(mode == 13){
do_batch_test(connfd, KVS_CMD_DEL, "array_set", NULL, KVS_STATUS_OK, NULL);
}else if(mode == 20){
do_batch_test(connfd, KVS_CMD_RSET, "rbtree_set", "rbtree_val");
do_batch_test(connfd, KVS_CMD_RSET, "rbtree_set", "rbtree_val", KVS_STATUS_OK, NULL);
}else if(mode == 21){
do_batch_test(connfd, KVS_CMD_RGET, "rbtree_set", NULL);
do_batch_test(connfd, KVS_CMD_RGET, "rbtree_set", NULL, KVS_STATUS_OK, "rbtree_val");
}else if(mode == 22){
do_batch_test(connfd, KVS_CMD_REXIST, "rbtree_set", NULL);
do_batch_test(connfd, KVS_CMD_REXIST, "rbtree_set", NULL, KVS_STATUS_OK, NULL);
}else if(mode == 30){
do_batch_test(connfd, KVS_CMD_HSET, "hash_set", "hash_val");
do_batch_test(connfd, KVS_CMD_HSET, "hash_set", "hash_val", KVS_STATUS_OK, NULL);
}else if(mode == 31){
do_batch_test(connfd, KVS_CMD_HGET, "hash_set", NULL);
do_batch_test(connfd, KVS_CMD_HGET, "hash_set", NULL, KVS_STATUS_OK, "hash_val");
}else if(mode == 32){
do_batch_test(connfd, KVS_CMD_HEXIST, "hash_set", NULL);
do_batch_test(connfd, KVS_CMD_HEXIST, "hash_set", NULL, KVS_STATUS_OK, NULL);
}else if(mode == -1){
save(connfd);
}else if(mode == 4){
}else if(mode == 5){
batch_qps(connfd);
}else if(mode == 6){
send_spec_chars(connfd);
}
return 0;
}