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

bugfix: reactor网络模型的的半包解析错误问题。

Browse Source 
全量持久化时清除增量持久化的记录。
This commit is contained in:
1iaan
2026-01-08 16:20:00 +08:00
parent 3cc97b9454
commit de21fe94ec
10 changed files with 282 additions and 121 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
.gitignore vendored
View File

@@ -1,6 +1,6 @@
NtyCo/
.vscode/
*.db
proactor copy.c
ntyco copy.c

19
kvs_rw_tools.c
View File

@@ -15,6 +15,8 @@ extern kvs_rbtree_t global_rbtree;
extern kvs_hash_t global_hash;
#endif
extern int global_cmd_log_fd;
// 0 suc, -1 err
int kvs_need(const uint8_t *p, const uint8_t *end, size_t n) {
return (p + n <= end) ? 0 : -1;
@@ -202,6 +204,7 @@ void kvs_free_request(kvs_req_t *req) {
/**
* 输入：req
* 输出：rsp
* 返回：-1 失败，参数错误，0 成功
*/
int kvs_execute_one_cmd(const kvs_req_t *req, kvs_rsp_t *rsp_out) {
if(!req || !rsp_out) return -1;
@@ -382,6 +385,10 @@ void kvs_free_request(kvs_req_t *req) {
return -1;
}
/**
* 构建单条响应
* 返回：-1 失败，>=0 响应长度
*/
int kvs_build_one_rsp(const kvs_rsp_t *results, uint8_t *response, size_t response_cap){
if (!results || !response) return -1;
@@ -420,6 +427,8 @@ int kvs_save_to_file(){
#endif
ksv_clear_log(global_cmd_log_fd);
return ret;
}
@@ -534,3 +543,13 @@ int kvs_replay_log(const char *logfile, int logfd){
return 0;
}
/**
* clear log file not close
*/
int ksv_clear_log(int logfd){
if(logfd < 0) return -1;
ftruncate(logfd, 0);
lseek(logfd, 0, SEEK_SET);
return 0;
}

16
kvs_rw_tools.h
View File

@@ -25,12 +25,11 @@ int kvs_read_file(FILE *fp, void *buf, size_t n);
* Rsp: | OP(1) | status(1) | datalen(4) | data |
*/
#define KVS_MAX_CMDS_PER_CALL 64
// 1MB
#define KVS_MAX_RESPONSE (1024u * 1024u)
#define KVS_MAX_ARGC 4
#define KVS_MAX_ARGLEN (1024u * 1024u)
#define KVS_MAX_CMD_BYTES (4u * 1024u * 1024u)
#define KVS_MAX_RESPONSE (65536)
#define KVS_MAX_ARGLEN (507)
#define KVS_MAX_CMD_BYTES (1024)
#define KVS_MAX_ARGC 3
enum {
KVS_STATUS_OK = 0,
@@ -65,12 +64,6 @@ enum {
KVS_CMD_COUNT,
};
typedef enum {
KVS_OK = 1,
KVS_NEED_MORE = 0,
KVS_ERROR = -1
}kvs_rc_t;
typedef struct kvs_arg_s{
uint32_t len;
const uint8_t *data;
@@ -97,5 +90,6 @@ int kvs_save_to_file();
int kvs_save_cmd_to_logfile(const uint8_t *cmd, size_t len, int logfd);
int kvs_replay_log(const char *logfile, int logfd);
int ksv_clear_log(int logfd);
#endif

48
kvstore.c
View File

@@ -249,16 +249,23 @@ int kvs_filter_protocol(char **tokens, int count, char *response) {
#if NEW_KVSTORE
/**
* input : request request_length
* output : response response_length consumed_out
* output : response response_length
* return : -1 error, =0 半包, 1 成功
*/
int kvs_protocol(char *request, int request_length, int *consumed_out, char *response, int *response_length){
if (!request || request_length <= 0 || !consumed_out || !response || !response_length) return KVS_NEED_MORE;
int kvs_protocol(char *request, int request_length, char *response, int *response_length){
if (!request || request_length <= 0 || !response || !response_length) return -1;
int consumed = 0;
int out_len = 0;
int budget = KVS_MAX_CMDS_PER_CALL;
while(consumed < request_length && (budget-- > 0)){
static int i = 0;
while(consumed < request_length ){
if(i > 33){
i = i+1;
i = i-1;
}
if(i == 47) i = 0;
++i;
kvs_req_t req;
memset(&req, 0, sizeof(kvs_req_t));
@@ -267,12 +274,13 @@ int kvs_protocol(char *request, int request_length, int *consumed_out, char *res
int len = kvs_parse_one_cmd(p, remain, &req);
if(len < 0){
// 解析失败
kvs_free_request(&req);
*consumed_out = consumed;
*response_length = out_len;
return KVS_ERROR;
return -1;
}
else if(len == 0){
// 半包
kvs_free_request(&req);
break;
}
@@ -280,32 +288,26 @@ int kvs_protocol(char *request, int request_length, int *consumed_out, char *res
kvs_rsp_t rsp;
memset(&rsp, 0, sizeof(kvs_rsp_t));
// 执行失败
if (kvs_execute_one_cmd(&req, &rsp) < 0){
kvs_free_request(&req);
*consumed_out = consumed;
*response_length = out_len;
return KVS_ERROR;
return -1;
}else{
// 执行成功，在这里保存到日志中。
if(req.op == KVS_CMD_SET || req.op == KVS_CMD_MOD || req.op == KVS_CMD_DEL){
printf("%d:%d\n", req.op, req.argc);
kvs_save_cmd_to_logfile(p, len, global_cmd_log_fd);
if(rsp.status == KVS_STATUS_OK){
if(req.op == KVS_CMD_SET || req.op == KVS_CMD_MOD || req.op == KVS_CMD_DEL){
kvs_save_cmd_to_logfile(p, len, global_cmd_log_fd);
}
}
}
if (out_len >= KVS_MAX_RESPONSE) {
kvs_free_request(&req);
*consumed_out = consumed;
*response_length = out_len;
return KVS_ERROR;
}
int resp_len = kvs_build_one_rsp(&rsp, (uint8_t *)response+out_len, KVS_MAX_RESPONSE-out_len);
// 构建响应 <0 构建失败
kvs_free_request(&req);
if (resp_len < 0) {
*consumed_out = consumed;
*response_length = out_len;
return KVS_ERROR;
return -1;
}
// printf("resp_len:%d\n", resp_len);
@@ -314,10 +316,8 @@ int kvs_protocol(char *request, int request_length, int *consumed_out, char *res
consumed += len;
}
*consumed_out = consumed;
*response_length = out_len;
if (consumed == 0 && out_len == 0) return KVS_NEED_MORE;
return KVS_OK;
return consumed;
}
#else
/*

2
kvstore.h
View File

@@ -35,7 +35,7 @@
// typedef int (*msg_handler)(char *msg, int length, char *response);
typedef int (*msg_handler)(char *request, int request_length, int *consumed_out, char *response, int *response_length);
typedef int (*msg_handler)(char *request, int request_length, char *response, int *response_length);
extern int reactor_start(unsigned short port, msg_handler handler);
extern int proactor_start(unsigned short port, msg_handler handler);

45
reactor.c
View File

@@ -28,15 +28,15 @@
#if ENABLE_KVSTORE
// typedef int (*msg_handler)(char *msg, int length, char *response);
typedef int (*msg_handler)(char *request, int request_length, int *consumed_out, char *response, int *response_length);
typedef int (*msg_handler)(char *request, int request_length, char *response, int *response_length);
static msg_handler kvs_handler;
// 0 need more, -1 error, =1 suc
int kvs_request(struct conn *c) {
int consumed_out;
int ret = kvs_handler(c->rbuffer, c->rlength, &consumed_out, c->wbuffer, &c->wlength);
int consumed_out = kvs_handler(c->rbuffer, c->rlength, c->wbuffer, &c->wlength);
return consumed_out;
}
int kvs_response(struct conn *c) {
@@ -137,9 +137,17 @@ int accept_cb(int fd) {
int recv_cb(int fd) {
struct conn *c = &conn_list[fd];
int avail = BUFFER_LENGTH - c->rlength;
printf("avail: %d\n", avail);
if (avail <= 0) {
// 缓冲满了还没解析出来：协议异常或包过大
close(fd);
epoll_ctl(epfd, EPOLL_CTL_DEL, fd, NULL);
return 0;
}
memset(conn_list[fd].rbuffer, 0, BUFFER_LENGTH );
int count = recv(fd, conn_list[fd].rbuffer, BUFFER_LENGTH, 0);
int count = recv(fd, c->rbuffer + c->rlength, avail, 0);
if (count == 0) { // disconnect
//printf("client disconnect: %d\n", fd);
close(fd);
@@ -157,7 +165,7 @@ int recv_cb(int fd) {
}
conn_list[fd].rlength = count;
c->rlength += count;
//printf("RECV: %s\n", conn_list[fd].rbuffer);
#if 0 // echo
@@ -176,8 +184,27 @@ int recv_cb(int fd) {
ws_request(&conn_list[fd]);
#elif ENABLE_KVSTORE
int consumed = kvs_request(c);
if(consumed < 0){
close(fd);
epoll_ctl(epfd, EPOLL_CTL_DEL, fd, NULL);
return 0;
}
kvs_request(&conn_list[fd]);
// 清理 buffer
if (consumed > 0 && consumed < c->rlength) {
// 有剩余未处理数据，搬移到 buffer 头部
int left = c->rlength - consumed;
if (left > 0) memmove(c->rbuffer, c->rbuffer + consumed, left);
c->rlength = left;
if (c->wlength > 0) set_event(fd, EPOLLOUT, 0);
return count;
}else{
c->rlength = 0;
if(c->wlength > 0) set_event(fd, EPOLLOUT, 0);
return count;
}
#endif
@@ -222,7 +249,7 @@ int send_cb(int fd) {
set_event(fd, EPOLLIN, 0);
}
#else
// printf("wlength: %d\n", conn_list[fd].wlength);
if (conn_list[fd].wlength != 0) {
count = send(fd, conn_list[fd].wbuffer, conn_list[fd].wlength, 0);
}

2
server.h
View File

@@ -5,7 +5,7 @@
#ifndef __SERVER_H__
#define __SERVER_H__
#define BUFFER_LENGTH 1024
#define BUFFER_LENGTH 4096
#define ENABLE_HTTP 0
#define ENABLE_WEBSOCKET 0

92
test/test_client.c
View File

@@ -138,59 +138,55 @@ int parse_response(const uint8_t *buf, int buflen, kvs_response_t *rsp) {
void print_response(const char *cmd_name, const kvs_response_t *rsp) {
printf("\n=== %s Response ===\n", cmd_name);
printf("OP: %u\n", rsp->op);
printf("Status: %u ", rsp->status);
switch (rsp->status) {
case KVS_STATUS_OK:
printf("(OK)\n");
break;
case KVS_STATUS_ERROR:
printf("(ERROR)\n");
break;
case KVS_STATUS_NO_EXIST:
printf("(NO_EXIST)\n");
break;
case KVS_STATUS_EXIST:
printf("(EXISTS)\n");
break;
default:
printf("(UNKNOWN)\n");
break;
}
printf("Data Length: %u\n", rsp->datalen);
if (rsp->datalen > 0 && rsp->data != NULL) {
printf("Data: ");
// 尝试以字符串形式打印（如果是可打印字符）
int is_printable = 1;
for (uint32_t i = 0; i < rsp->datalen; i++) {
if (rsp->data[i] < 32 || rsp->data[i] > 126) {
is_printable = 0;
break;
}
}
if (is_printable) {
printf("\"");
printf("%s ", cmd_name);
if(rsp->op == KVS_CMD_GET){
if (rsp->datalen > 0 && rsp->data != NULL) {
printf("Data: ");
// 尝试以字符串形式打印（如果是可打印字符）
int is_printable = 1;
for (uint32_t i = 0; i < rsp->datalen; i++) {
printf("%c", rsp->data[i]);
if (rsp->data[i] < 32 || rsp->data[i] > 126) {
is_printable = 0;
break;
}
}
if (is_printable) {
printf("\"");
for (uint32_t i = 0; i < rsp->datalen; i++) {
printf("%c", rsp->data[i]);
}
printf("\"\n");
} else {
// 以十六进制打印
printf("0x");
for (uint32_t i = 0; i < rsp->datalen; i++) {
printf("%02x", rsp->data[i]);
}
printf("\n");
}
printf("\"\n");
} else {
// 以十六进制打印
printf("0x");
for (uint32_t i = 0; i < rsp->datalen; i++) {
printf("%02x", rsp->data[i]);
}
printf("\n");
printf("Data: (empty)\n");
}
}else {
switch (rsp->status) {
case KVS_STATUS_OK:
printf("(OK)\n");
break;
case KVS_STATUS_ERROR:
printf("(ERROR)\n");
break;
case KVS_STATUS_NO_EXIST:
printf("(NO_EXIST)\n");
break;
case KVS_STATUS_EXIST:
printf("(EXISTS)\n");
break;
default:
printf("(UNKNOWN)\n");
break;
}
} else {
printf("Data: (empty)\n");
}
printf("==================\n");
}
int verify_response(const kvs_response_t *rsp, uint8_t expected_op,

36
test/test_client.h
View File

@@ -14,8 +14,9 @@
#include <sys/socket.h>
#include <arpa/inet.h>
#define CMD_SIZE (4096)
#define CMD_SIZE (1024)
#define BATCH_SIZE (65536)
#define KVS_BATCH_MAX 64
#define TIME_SUB_MS(tv1, tv2) ((tv1.tv_sec - tv2.tv_sec) * 1000 + (tv1.tv_usec - tv2.tv_usec) / 1000)
#define PRESP print_response
@@ -81,7 +82,6 @@ int verify_response(const kvs_response_t *rsp, uint8_t expected_op,
#define KVS_BATCH_MAX 64
typedef struct {
uint8_t buf[BATCH_SIZE];
@@ -101,10 +101,8 @@ static void kvs_batch_init(kvs_batch_t *b)
* 用 getcmd() 生成单条命令，然后 append 到 batch buffer
* 返回：0 成功，-1 失败（太多条 or buffer 不够）
*/
static int kvs_batch_add(kvs_batch_t *b, uint8_t op, const char *key, const char *value)
{
static int kvs_batch_add(kvs_batch_t *b, uint8_t op, const char *key, const char *value){
if (b->cnt >= KVS_BATCH_MAX) return -1;
uint8_t tmp[CMD_SIZE];
int n = getcmd(op, key, value, tmp); // 你提供的函数
if (n <= 0) return -1;
@@ -124,6 +122,7 @@ static int kvs_batch_add(kvs_batch_t *b, uint8_t op, const char *key, const char
*/
static int kvs_batch_send(int fd, const kvs_batch_t *b)
{
printf("send : %d\n", b->len);
return (int)send(fd, b->buf, b->len, 0);
}
@@ -138,19 +137,26 @@ static int kvs_batch_recv_parse(int fd,
uint8_t *recvbuf,
int recvbuf_cap)
{
int nrecv = (int)recv(fd, recvbuf, recvbuf_cap, 0);
if (nrecv <= 0) return -1;
int off = 0;
int parsed = 0;
int used = 0;
while (parsed < b->cnt && off < nrecv) {
int consumed = parse_response(recvbuf + off, nrecv - off, &rsps[parsed]);
if (consumed <= 0) break; // 不够解析/失败，简单处理：直接退出
while(parsed < b->cnt){
printf("recv loop: parsed=%d\n", parsed);
int nrecv = (int)recv(fd, recvbuf+used, recvbuf_cap, 0);
printf("recv nrecv=%d\n", nrecv);
if (nrecv <= 0) return -1;
off += consumed;
parsed++;
int off = 0;
while (parsed < b->cnt) {
int consumed = parse_response(recvbuf + used, nrecv - off, &rsps[parsed]);
if (consumed <= 0) break; // 不够解析/失败，简单处理：直接退出
off += consumed;
used+= consumed;
parsed++;
}
printf("after parse: parsed=%d, used=%d\n", parsed, used);
}
return parsed;
}

141
test/testcase.c
View File

@@ -95,7 +95,67 @@ void array_testcase_1w(int connfd) {
}
void do_batch_example(int fd)
void rbtree_testcase_1w(int connfd) {
int count = 1000;
int i = 0;
struct timeval tv_begin;
gettimeofday(&tv_begin, NULL);
for (i = 0;i < count;i ++) {
testcase(connfd, KVS_CMD_RSET, "name", "lian", KVS_STATUS_OK, NULL, "SET NAME");
testcase(connfd, KVS_CMD_RGET, "name", NULL, KVS_STATUS_OK, "lian", "GET NAME");
testcase(connfd, KVS_CMD_RMOD, "name", "liu", KVS_STATUS_OK, NULL, "MOD NAME");
testcase(connfd, KVS_CMD_RGET, "name", NULL, KVS_STATUS_OK, "liu", "GET NAME");
testcase(connfd, KVS_CMD_REXIST, "name", NULL, KVS_STATUS_EXIST, NULL, "EXIST NAME");
testcase(connfd, KVS_CMD_RDEL, "name", NULL, KVS_STATUS_OK, NULL, "DEL NAME");
testcase(connfd, KVS_CMD_REXIST, "name", NULL, KVS_STATUS_NO_EXIST, NULL, "NOT EXIST NAME");
testcase(connfd, KVS_CMD_RMOD, "stu", "liu", KVS_STATUS_NO_EXIST, NULL, "MOD NAME");
testcase(connfd, KVS_CMD_RDEL, "stu", NULL, KVS_STATUS_NO_EXIST, NULL, "DEL SUT");
}
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 hash_testcase_1w(int connfd) {
int count = 1000;
int i = 0;
struct timeval tv_begin;
gettimeofday(&tv_begin, NULL);
for (i = 0;i < count;i ++) {
testcase(connfd, KVS_CMD_HSET, "name", "lian", KVS_STATUS_OK, NULL, "SET NAME");
testcase(connfd, KVS_CMD_HGET, "name", NULL, KVS_STATUS_OK, "lian", "GET NAME");
testcase(connfd, KVS_CMD_HMOD, "name", "liu", KVS_STATUS_OK, NULL, "MOD NAME");
testcase(connfd, KVS_CMD_HGET, "name", NULL, KVS_STATUS_OK, "liu", "GET NAME");
testcase(connfd, KVS_CMD_HEXIST, "name", NULL, KVS_STATUS_EXIST, NULL, "EXIST NAME");
testcase(connfd, KVS_CMD_HDEL, "name", NULL, KVS_STATUS_OK, NULL, "DEL NAME");
testcase(connfd, KVS_CMD_HEXIST, "name", NULL, KVS_STATUS_NO_EXIST, NULL, "NOT EXIST NAME");
testcase(connfd, KVS_CMD_HMOD, "stu", "liu", KVS_STATUS_NO_EXIST, NULL, "MOD NAME");
testcase(connfd, KVS_CMD_HDEL, "stu", NULL, KVS_STATUS_NO_EXIST, NULL, "DEL SUT");
}
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 do_batch_SET_example(int fd)
{
kvs_batch_t batch;
kvs_batch_init(&batch);
@@ -103,7 +163,7 @@ void do_batch_example(int fd)
char key[10]={0}, val[10]={0};
// 组 batch（最多 64 条）
for(int i = 0;i < 24; ++ i){
for(int i = 0;i < 48; ++ i){
int len = sprintf(key, "k%d", i);
len = sprintf(val, "v%d", i);
kvs_batch_add(&batch, KVS_CMD_SET, key, val);
@@ -120,16 +180,67 @@ void do_batch_example(int fd)
// 打印/处理
for (int i = 0; i < nrsp; i++) {
PRESP("BATCH", &rsps[i]);
}
int len = sprintf(key, "SET%d", i);
PRESP(key, &rsps[i]);
}
}
void do_batch_GET_example(int fd)
{
kvs_batch_t batch;
kvs_batch_init(&batch);
// for(int i = 0;i < 24; ++ i){
// int len = sprintf(key, "k%d", i);
// len = sprintf(val, "v%d", i);
// testcase(fd, KVS_CMD_GET, key, NULL, KVS_STATUS_OK, val, "GET K");
// }
char key[10]={0}, val[10]={0};
// 组 batch（最多 64 条）
for(int i = 0;i < 48; ++ i){
int len = sprintf(key, "k%d", i);
kvs_batch_add(&batch, KVS_CMD_GET, key, NULL);
}
// 一次性发送
kvs_batch_send(fd, &batch);
// 一次性 recv + parse
uint8_t recvbuf[BATCH_SIZE];
kvs_response_t rsps[KVS_BATCH_MAX];
int nrsp = kvs_batch_recv_parse(fd, &batch, rsps, recvbuf, sizeof(recvbuf));
// 打印/处理
for (int i = 0; i < nrsp; i++) {
int len = sprintf(key, "GET%d", i);
PRESP(key, &rsps[i]);
}
}
void do_batch_DEL_example(int fd)
{
kvs_batch_t batch;
kvs_batch_init(&batch);
char key[10]={0}, val[10]={0};
// 组 batch（最多 64 条）
for(int i = 0;i < 48; ++ i){
int len = sprintf(key, "k%d", i);
kvs_batch_add(&batch, KVS_CMD_DEL, key, NULL);
}
// 一次性发送
kvs_batch_send(fd, &batch);
// 一次性 recv + parse
uint8_t recvbuf[BATCH_SIZE];
kvs_response_t rsps[KVS_BATCH_MAX];
int nrsp = kvs_batch_recv_parse(fd, &batch, rsps, recvbuf, sizeof(recvbuf));
// 打印/处理
for (int i = 0; i < nrsp; i++) {
int len = sprintf(key, "DEL%d", i);
PRESP(key, &rsps[i]);
}
}
void save(int connfd){
@@ -150,10 +261,18 @@ int main(int argc, char *argv[]) {
int connfd = connect_tcpserver(ip, port);
if(mode == 0){
do_batch_example(connfd);
do_batch_SET_example(connfd);
}else if(mode == 1){
array_testcase_1w(connfd);
do_batch_GET_example(connfd);
}else if(mode == 2){
do_batch_DEL_example(connfd);
}else if(mode == 10){
array_testcase_1w(connfd);
}else if(mode == 11){
rbtree_testcase_1w(connfd);
}else if(mode == 12){
hash_testcase_1w(connfd);
}else if(mode == -1){
save(connfd);
}
return 0;