实现全量持久化:save操作落盘,启动时读取到内存

增量持久化:执行修改操作时将cmd追加到log中,启动时逐条取出顺序执行
This commit is contained in:
2026-01-07 18:43:28 +08:00
parent cb0134a852
commit 3cc97b9454
9 changed files with 373 additions and 74 deletions

View File

@@ -1,6 +1,8 @@
#include "kvstore.h" #include "kvstore.h"
#include "kvs_rw_tools.h"
#include <arpa/inet.h>
// singleton // singleton
@@ -40,7 +42,7 @@ void kvs_array_destroy(kvs_array_t *inst) {
/** /**
* return: ==-2 not exist, == -1 error, >= 0 exist idx * return: ==-2 not exist, == -1 error, >= 0 exist idx
*/ */
int kvs_array_find_index(kvs_array_t *inst, const void *key, size_t key_len) { int kvs_array_find_index(kvs_array_t *inst, const void *key, uint32_t key_len) {
if (!inst || !inst->table || !key) return -1; if (!inst || !inst->table || !key) return -1;
for (int i = 0; i < inst->total; i++) { for (int i = 0; i < inst->total; i++) {
kvs_array_item_t *it = &inst->table[i]; kvs_array_item_t *it = &inst->table[i];
@@ -56,8 +58,8 @@ int kvs_array_find_index(kvs_array_t *inst, const void *key, size_t key_len) {
* return: <0 error; 0 success; 1 exist * return: <0 error; 0 success; 1 exist
*/ */
int kvs_array_set_bin(kvs_array_t *inst, int kvs_array_set_bin(kvs_array_t *inst,
const void *key, size_t key_len, const void *key, uint32_t key_len,
const void *value, size_t value_len) { const void *value, uint32_t value_len) {
if (!inst || !inst->table || !key || key_len == 0 || !value) return -1; if (!inst || !inst->table || !key || key_len == 0 || !value) return -1;
if (inst->total >= KVS_ARRAY_SIZE) return -1; if (inst->total >= KVS_ARRAY_SIZE) return -1;
@@ -109,9 +111,7 @@ int kvs_array_set_bin(kvs_array_t *inst,
return 0; return 0;
} }
void *kvs_array_get_bin(kvs_array_t *inst, void *kvs_array_get_bin(kvs_array_t *inst, const void *key, uint32_t key_len, uint32_t *out_value_len) {
const void *key, size_t key_len,
size_t *out_value_len) {
if (out_value_len) *out_value_len = 0; if (out_value_len) *out_value_len = 0;
if (!inst || !inst->table || !key || key_len == 0) return NULL; if (!inst || !inst->table || !key || key_len == 0) return NULL;
@@ -123,7 +123,7 @@ void *kvs_array_get_bin(kvs_array_t *inst,
return it->value; // 注意:由 store 持有 return it->value; // 注意:由 store 持有
} }
int kvs_array_del_bin(kvs_array_t *inst, const void *key, size_t key_len) { int kvs_array_del_bin(kvs_array_t *inst, const void *key, uint32_t key_len) {
if (!inst || !inst->table || !key || key_len == 0) return -1; if (!inst || !inst->table || !key || key_len == 0) return -1;
int idx = kvs_array_find_index(inst, key, key_len); int idx = kvs_array_find_index(inst, key, key_len);
@@ -148,8 +148,8 @@ int kvs_array_del_bin(kvs_array_t *inst, const void *key, size_t key_len) {
} }
int kvs_array_mod_bin(kvs_array_t *inst, int kvs_array_mod_bin(kvs_array_t *inst,
const void *key, size_t key_len, const void *key, uint32_t key_len,
const void *value, size_t value_len) { const void *value, uint32_t value_len) {
if (!inst || !inst->table || !key || key_len == 0 || !value) return -1; if (!inst || !inst->table || !key || key_len == 0 || !value) return -1;
int idx = kvs_array_find_index(inst, key, key_len); int idx = kvs_array_find_index(inst, key, key_len);
@@ -174,12 +174,87 @@ int kvs_array_mod_bin(kvs_array_t *inst,
return 0; return 0;
} }
int kvs_array_exist_bin(kvs_array_t *inst, const void *key, size_t key_len) { int kvs_array_exist_bin(kvs_array_t *inst, const void *key, uint32_t key_len) {
if (!inst || !inst->table || !key || key_len == 0) return -1; if (!inst || !inst->table || !key || key_len == 0) return -1;
int idx = kvs_array_find_index(inst, key, key_len); int idx = kvs_array_find_index(inst, key, key_len);
return (idx >= 0) ? 0 : 1; return (idx >= 0) ? 0 : 1;
} }
// return: 0 success, <0 error
int kvs_array_save(kvs_array_t *inst, const char* filename){
if(!inst || !filename) return -1;
FILE *fp = fopen(filename, "wb");
if(!fp) return -2;
for(int i = 0;i < inst->total; ++ i){
kvs_array_item_t *it = &inst->table[i];
if(!it->key || it->key_len == 0) continue; // 跳过空槽
if(it->value_len > 0 && !it->value) { fclose(fp); return -3; }
uint32_t klen = htonl(it->key_len);
uint32_t vlen = htonl(it->value_len);
if (kvs_write_file(fp, &klen, 4) < 0) { fclose(fp); return -4; }
if (kvs_write_file(fp, &vlen, 4) < 0) { fclose(fp); return -4; }
if (kvs_write_file(fp, it->key, it->key_len) < 0) { fclose(fp); return -4; }
if (it->value_len > 0) {
if (kvs_write_file(fp, it->value, it->value_len) < 0) { fclose(fp); return -4; }
}
}
fclose(fp);
return 0;
}
int kvs_array_load(kvs_array_t *inst, const char* filename) {
if (!inst || !filename) return -1;
if (!inst->table) return -1;
FILE *fp = fopen(filename, "rb");
if (!fp) return -2;
int idx = 0;
while(1){
uint32_t klen = 0, vlen = 0;
if (kvs_read_file(fp, &klen, 4) < 0) { fclose(fp); return -3; }
if (kvs_read_file(fp, &vlen, 4) < 0) { fclose(fp); return -3; }
klen = ntohl(klen);
vlen = ntohl(vlen);
if (klen == 0) { fclose(fp); return -3; }
kvs_array_item_t *it = &inst->table[idx];
memset(it, 0, sizeof(*it));
it->key = (uint8_t *)kvs_malloc(klen);
if (!it->key) { fclose(fp); return -3; }
it->key_len = klen;
if (kvs_read_file(fp, it->key, klen) < 0) { fclose(fp); return -3; }
if (vlen > 0) {
it->value = (uint8_t *)kvs_malloc(vlen);
if (!it->value) { fclose(fp); return -4; }
it->value_len = vlen;
if (kvs_read_file(fp, it->value, vlen) < 0) { fclose(fp); return -3; }
} else {
it->value = NULL;
it->value_len = 0;
}
inst->total ++;
idx++;
if(idx >= KVS_ARRAY_SIZE){ break; }
}
fclose(fp);
return 0;
}
#else #else
/* /*
* @return: <0, error; =0, success; >0, exist * @return: <0, error; =0, success; >0, exist

View File

@@ -20,26 +20,26 @@ kvs_hash_t global_hash;
//Connection //Connection
// 'C' + 'o' + 'n' // 'C' + 'o' + 'n'
static int _hash(const void *key, size_t key_len, int size) { static int _hash(const void *key, uint32_t key_len, int size) {
if (!key || size <= 0) return -1; if (!key || size <= 0) return -1;
const uint8_t *p = (const uint8_t *)key; const uint8_t *p = (const uint8_t *)key;
uint32_t sum = 0; uint32_t sum = 0;
for (size_t i = 0; i < key_len; i++) { for (uint32_t i = 0; i < key_len; i++) {
sum += p[i]; sum += p[i];
} }
return sum % size; return sum % size;
} }
static int _key_equal(const hashnode_t *node, const void *key, size_t key_len) { static int _key_equal(const hashnode_t *node, const void *key, uint32_t key_len) {
if (!node || !key) return 0; if (!node || !key) return 0;
if (!node->key) return 0; if (!node->key) return 0;
if (node->key_len != key_len) return 0; if (node->key_len != key_len) return 0;
return memcmp(node->key, key, key_len) == 0; return memcmp(node->key, key, key_len) == 0;
} }
static hashnode_t *_create_node(const void *key, size_t key_len, static hashnode_t *_create_node(const void *key, uint32_t key_len,
const void *value, size_t value_len) { const void *value, uint32_t value_len) {
hashnode_t *node = (hashnode_t*)kvs_malloc(sizeof(hashnode_t)); hashnode_t *node = (hashnode_t*)kvs_malloc(sizeof(hashnode_t));
if (!node) return NULL; if (!node) return NULL;
memset(node, 0, sizeof(*node)); memset(node, 0, sizeof(*node));
@@ -113,7 +113,7 @@ void kvs_hash_destroy(kvs_hash_t *hash) {
/* /*
* @return: <0 error; =0 success; >0 exist * @return: <0 error; =0 success; >0 exist
*/ */
int kvs_hash_set_bin(kvs_hash_t *hash, const void *key, size_t key_len, const void *value, size_t value_len) { int kvs_hash_set_bin(kvs_hash_t *hash, const void *key, uint32_t key_len, const void *value, uint32_t value_len) {
if (!hash || !hash->nodes || !key || key_len == 0 || !value) return -1; if (!hash || !hash->nodes || !key || key_len == 0 || !value) return -1;
int idx = _hash(key, key_len, MAX_TABLE_SIZE); int idx = _hash(key, key_len, MAX_TABLE_SIZE);
@@ -140,7 +140,7 @@ int kvs_hash_set_bin(kvs_hash_t *hash, const void *key, size_t key_len, const vo
/* /*
* get 返回value 指针(由 hash 持有),并通过 out_value_len 返回长度 * get 返回value 指针(由 hash 持有),并通过 out_value_len 返回长度
*/ */
void *kvs_hash_get_bin(kvs_hash_t *hash, const void *key, size_t key_len, size_t *out_value_len) { void *kvs_hash_get_bin(kvs_hash_t *hash, const void *key, uint32_t key_len, uint32_t *out_value_len) {
if (!hash || !hash->nodes || !key || key_len == 0 || !out_value_len) return NULL; if (!hash || !hash->nodes || !key || key_len == 0 || !out_value_len) return NULL;
*out_value_len = 0; *out_value_len = 0;
@@ -166,7 +166,7 @@ void *kvs_hash_get_bin(kvs_hash_t *hash, const void *key, size_t key_len, size_
/* /*
* @return <0 error; =0 success; >0 no exist * @return <0 error; =0 success; >0 no exist
*/ */
int kvs_hash_mod_bin(kvs_hash_t *hash, const void *key, size_t key_len, const void *value, size_t value_len) { int kvs_hash_mod_bin(kvs_hash_t *hash, const void *key, uint32_t key_len, const void *value, uint32_t value_len) {
if (!hash || !hash->nodes || !key || key_len == 0 || !value) return -1; if (!hash || !hash->nodes || !key || key_len == 0 || !value) return -1;
@@ -210,7 +210,7 @@ int kvs_hash_count(kvs_hash_t *hash) {
/* /*
* @return 0 success; <0 error/noexist * @return 0 success; <0 error/noexist
*/ */
int kvs_hash_del_bin(kvs_hash_t *hash, const void *key, size_t key_len) { int kvs_hash_del_bin(kvs_hash_t *hash, const void *key, uint32_t key_len) {
if (!hash || !key || key_len == 0) return -2; if (!hash || !key || key_len == 0) return -2;
int idx = _hash(key, key_len, MAX_TABLE_SIZE); int idx = _hash(key, key_len, MAX_TABLE_SIZE);
@@ -258,8 +258,8 @@ int kvs_hash_del_bin(kvs_hash_t *hash, const void *key, size_t key_len) {
/* /*
* @return 0 exist, 1 no exist * @return 0 exist, 1 no exist
*/ */
int kvs_hash_exist_bin(kvs_hash_t *hash, const void *key, size_t key_len) { int kvs_hash_exist_bin(kvs_hash_t *hash, const void *key, uint32_t key_len) {
size_t vlen = 0; uint32_t vlen = 0;
void *value = kvs_hash_get_bin(hash, key, key_len, &vlen); void *value = kvs_hash_get_bin(hash, key, key_len, &vlen);
return value ? 0 : 1; return value ? 0 : 1;
} }

View File

@@ -1,6 +1,7 @@
#include "kvstore.h" #include "kvstore.h"
#include "kvs_rw_tools.h" #include "kvs_rw_tools.h"
#include <arpa/inet.h> #include <arpa/inet.h>
#include <unistd.h>
#if ENABLE_ARRAY #if ENABLE_ARRAY
extern kvs_array_t global_array; extern kvs_array_t global_array;
@@ -14,11 +15,13 @@ extern kvs_rbtree_t global_rbtree;
extern kvs_hash_t global_hash; extern kvs_hash_t global_hash;
#endif #endif
// 0 suc, -1 err
int kvs_need(const uint8_t *p, const uint8_t *end, size_t n) { int kvs_need(const uint8_t *p, const uint8_t *end, size_t n) {
return (p + n <= end) ? 0 : -1; return (p + n <= end) ? 0 : -1;
} }
// 注意u8类型不需要ntoh或者hton // 注意u8类型不需要ntoh或者hton
// 0 suc, -1 err
int kvs_read_u8(const uint8_t **pp, const uint8_t *end, uint8_t *out) { int kvs_read_u8(const uint8_t **pp, const uint8_t *end, uint8_t *out) {
const uint8_t *p = *pp; const uint8_t *p = *pp;
if (kvs_need(p, end, 1) < 0) return -1; if (kvs_need(p, end, 1) < 0) return -1;
@@ -73,6 +76,31 @@ int kvs_write_u32(uint8_t **pp, const uint8_t *end, uint32_t v) {
return 0; return 0;
} }
// -1 err, 0 suc
int kvs_write_file(FILE *fp, const void *buf, size_t n) {
const uint8_t *p = (const uint8_t *)buf;
while (n > 0) {
size_t w = fwrite(p, 1, n, fp);
if (w == 0) return -1;
p += w;
n -= w;
}
return 0;
}
// -1 err, 0 suc
int kvs_read_file(FILE *fp, void *buf, size_t n){
uint8_t *p = (uint8_t *)buf;
while (n > 0) {
size_t r = fread(p, 1, n, fp);
if (r == 0) return -1; // EOF or error
p += r;
n -= r;
}
return 0;
}
// return: -1 fail, 0 half, >0 consumed // return: -1 fail, 0 half, >0 consumed
int kvs_parse_one_cmd(const uint8_t *request, int request_length, kvs_req_t *req_out){ int kvs_parse_one_cmd(const uint8_t *request, int request_length, kvs_req_t *req_out){
if (!request || request_length <= 0 || !req_out) return -1; if (!request || request_length <= 0 || !req_out) return -1;
@@ -186,9 +214,9 @@ void kvs_free_request(kvs_req_t *req) {
int op = req->op; int op = req->op;
kvs_arg_t *argv = req->args; kvs_arg_t *argv = req->args;
size_t key_len = 0; uint32_t key_len = 0;
const void *key = NULL; const void *key = NULL;
size_t value_len = 0; uint32_t value_len = 0;
const void *val = NULL; const void *val = NULL;
if(argc == 1){ if(argc == 1){
@@ -222,6 +250,9 @@ void kvs_free_request(kvs_req_t *req) {
case KVS_CMD_HEXIST: case KVS_CMD_HEXIST:
if (argc != 1 || !key) { rsp_out->status = KVS_STATUS_BADREQ; return -1; } if (argc != 1 || !key) { rsp_out->status = KVS_STATUS_BADREQ; return -1; }
break; break;
case KVS_CMD_SAVE:
if(argc != 0) { rsp_out->status = KVS_STATUS_BADREQ; return -1; }
break;
default: default:
rsp_out->status = KVS_STATUS_BADREQ; rsp_out->status = KVS_STATUS_BADREQ;
return -1; return -1;
@@ -338,6 +369,11 @@ void kvs_free_request(kvs_req_t *req) {
rsp_out->status = (ret == 0) ? KVS_STATUS_EXIST : KVS_STATUS_NO_EXIST; rsp_out->status = (ret == 0) ? KVS_STATUS_EXIST : KVS_STATUS_NO_EXIST;
return 0; return 0;
#endif #endif
case KVS_CMD_SAVE:
ret = kvs_save_to_file();
if(ret == 0) rsp_out->status = KVS_STATUS_OK;
else rsp_out->status = KVS_STATUS_ERROR;
return 0;
default: default:
rsp_out->status = KVS_STATUS_BADREQ; rsp_out->status = KVS_STATUS_BADREQ;
return -1; return -1;
@@ -371,3 +407,130 @@ int kvs_build_one_rsp(const kvs_rsp_t *results, uint8_t *response, size_t respon
return (int)(p - response); return (int)(p - response);
} }
int kvs_save_to_file(){
#if ENABLE_ARRAY
int ret = kvs_array_save(&global_array, KVS_ARRAY_FILE);
#endif
#if ENABLE_RBTREE
#endif
#if ENABLE_HASH
#endif
return ret;
}
#include <errno.h>
int write_full(int fd, const void *buf, size_t len)
{
const uint8_t *p = buf;
while (len > 0) {
ssize_t n = write(fd, p, len);
if (n < 0) {
if (errno == EINTR)
continue;
return -1;
}
p += n;
len -= n;
}
return 0;
}
// 1 read n suc, 0 eof, -1 error
int read_full(int fd, void *buf, size_t n)
{
uint8_t *p = (uint8_t *)buf;
size_t got = 0;
while (got < n) {
ssize_t r = read(fd, p + got, n - got);
if (r > 0) {
got += (size_t)r;
continue;
}
if (r == 0) {
return (got == 0) ? 0 : -1;
}
/* r < 0 */
if (errno == EINTR) {
continue;
}
return -1;
}
return 1;
}
int kvs_save_cmd_to_logfile(const uint8_t *cmd, size_t len, int logfd){
if (logfd < 0 || !cmd || len == 0)
return -1;
if (len > UINT32_MAX)
return -2;
uint32_t nlen = htonl((uint32_t)len);
if (write_full(logfd, &nlen, sizeof(nlen)) < 0)
return -3;
if (write_full(logfd, cmd, len) < 0)
return -4;
if (fsync(logfd) < 0)
return -5;
return 0;
}
int kvs_replay_log(const char *logfile, int logfd){
if (!logfile|| logfd<0) return -1;
for (;;) {
uint32_t nlen = 0;
int hr = read_full(logfd, &nlen, sizeof(nlen));
if (hr == 0) break; /* EOF正常结束 */
if (hr < 0) { return -2; } /* 半截头 */
uint32_t len = ntohl(nlen);
if (len == 0) { return -3; }
uint8_t *cmd = (uint8_t *)kvs_malloc(len);
if (!cmd) { return -5; }
int pr = read_full(logfd, cmd, len);
if (pr <= 0) { /* 半截 payload */
kvs_free(cmd);
return -6;
}
kvs_req_t req;
memset(&req, 0, sizeof(req));
int clen = kvs_parse_one_cmd(cmd, (int)len, &req);
if (clen <= 0 || clen != (int)len) {
kvs_free_request(&req);
kvs_free(cmd);
return -7;
}
kvs_rsp_t rsp;
memset(&rsp, 0, sizeof(rsp));
if (kvs_execute_one_cmd(&req, &rsp) < 0) {
kvs_free_request(&req);
kvs_free(cmd);
return -8;
}
kvs_free_request(&req);
kvs_free(cmd);
}
return 0;
}

View File

@@ -3,6 +3,7 @@
#include <stdint.h> #include <stdint.h>
#include <stdio.h> #include <stdio.h>
//
int kvs_need(const uint8_t *p, const uint8_t *end, size_t n); int kvs_need(const uint8_t *p, const uint8_t *end, size_t n);
int kvs_read_u8(const uint8_t **pp, const uint8_t *end, uint8_t *out); int kvs_read_u8(const uint8_t **pp, const uint8_t *end, uint8_t *out);
int kvs_read_u16(const uint8_t **pp, const uint8_t *end, uint16_t *out); int kvs_read_u16(const uint8_t **pp, const uint8_t *end, uint16_t *out);
@@ -12,6 +13,8 @@ int kvs_write_u8(uint8_t **pp, const uint8_t *end, uint8_t v);
int kvs_write_u16(uint8_t **pp, const uint8_t *end, uint16_t v); int kvs_write_u16(uint8_t **pp, const uint8_t *end, uint16_t v);
int kvs_write_u32(uint8_t **pp, const uint8_t *end, uint32_t v); int kvs_write_u32(uint8_t **pp, const uint8_t *end, uint32_t v);
int kvs_write_file(FILE *fp, const void *buf, size_t n);
int kvs_read_file(FILE *fp, void *buf, size_t n);
/** /**
@@ -58,6 +61,7 @@ enum {
KVS_CMD_HMOD, KVS_CMD_HMOD,
KVS_CMD_HEXIST, KVS_CMD_HEXIST,
KVS_CMD_SAVE,
KVS_CMD_COUNT, KVS_CMD_COUNT,
}; };
@@ -89,6 +93,9 @@ int kvs_parse_one_cmd(const uint8_t *request, int request_length, kvs_req_t *req
void kvs_free_request(kvs_req_t *req); void kvs_free_request(kvs_req_t *req);
int kvs_execute_one_cmd(const kvs_req_t *req, kvs_rsp_t *rsp_out); int kvs_execute_one_cmd(const kvs_req_t *req, kvs_rsp_t *rsp_out);
int kvs_build_one_rsp(const kvs_rsp_t *results, uint8_t *response, size_t response_cap); int kvs_build_one_rsp(const kvs_rsp_t *results, uint8_t *response, size_t response_cap);
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);
#endif #endif

View File

@@ -4,6 +4,10 @@
#include "kvstore.h" #include "kvstore.h"
#include "kvs_rw_tools.h" #include "kvs_rw_tools.h"
#include <sys/types.h>
#include <sys/stat.h>
#include <fcntl.h>
#include <unistd.h>
#if ENABLE_ARRAY #if ENABLE_ARRAY
extern kvs_array_t global_array; extern kvs_array_t global_array;
@@ -17,6 +21,8 @@ extern kvs_rbtree_t global_rbtree;
extern kvs_hash_t global_hash; extern kvs_hash_t global_hash;
#endif #endif
int global_cmd_log_fd = -1;
void *kvs_malloc(size_t size) { void *kvs_malloc(size_t size) {
return malloc(size); return malloc(size);
} }
@@ -60,10 +66,7 @@ int kvs_split_token(char *msg, char *tokens[]) {
// tokens[1] : Key // tokens[1] : Key
// tokens[2] : Value // tokens[2] : Value
#if BIN_SAFE #if !BIN_SAFE
#else
int kvs_filter_protocol(char **tokens, int count, char *response) { int kvs_filter_protocol(char **tokens, int count, char *response) {
if (tokens[0] == NULL || count == 0 || response == NULL) return -1; if (tokens[0] == NULL || count == 0 || response == NULL) return -1;
@@ -282,6 +285,12 @@ int kvs_protocol(char *request, int request_length, int *consumed_out, char *res
*consumed_out = consumed; *consumed_out = consumed;
*response_length = out_len; *response_length = out_len;
return KVS_ERROR; return KVS_ERROR;
}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 (out_len >= KVS_MAX_RESPONSE) { if (out_len >= KVS_MAX_RESPONSE) {
@@ -338,11 +347,28 @@ int kvs_protocol(char *msg, int length, char *response) { //
} }
#endif #endif
int init_cmd_log(const char *file, int *logfd){
if(!file) return -1;
int fd = open(file, O_RDWR | O_CREAT | O_APPEND, 0644);
if(fd < 0) return -2;
*logfd = fd;
return 0;
}
int destroy_cmd_log(int logfd){
close(logfd);
return 0;
}
int init_kvengine(void) { int init_kvengine(void) {
#if ENABLE_ARRAY #if ENABLE_ARRAY
memset(&global_array, 0, sizeof(kvs_array_t)); memset(&global_array, 0, sizeof(kvs_array_t));
kvs_array_create(&global_array); kvs_array_create(&global_array);
kvs_array_load(&global_array, KVS_ARRAY_FILE);
#endif #endif
#if ENABLE_RBTREE #if ENABLE_RBTREE
@@ -355,6 +381,9 @@ int init_kvengine(void) {
kvs_hash_create(&global_hash); kvs_hash_create(&global_hash);
#endif #endif
init_cmd_log(KVS_CMD_LOG_FILE, &global_cmd_log_fd);
kvs_replay_log(KVS_CMD_LOG_FILE, global_cmd_log_fd);
return 0; return 0;
} }
@@ -369,6 +398,7 @@ void dest_kvengine(void) {
kvs_hash_destroy(&global_hash); kvs_hash_destroy(&global_hash);
#endif #endif
destroy_cmd_log(global_cmd_log_fd);
} }

View File

@@ -30,6 +30,9 @@
#define BIN_SAFE 1 #define BIN_SAFE 1
#define KVS_CMD_LOG_FILE "kvs_cmd_log.db"
#define KVS_ARRAY_FILE "kvs_array.db"
// typedef int (*msg_handler)(char *msg, int length, char *response); // 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, int *consumed_out, char *response, int *response_length);
@@ -47,10 +50,10 @@ extern int ntyco_start(unsigned short port, msg_handler handler);
typedef struct kvs_array_item_s { typedef struct kvs_array_item_s {
uint8_t *key; uint8_t *key;
size_t key_len; uint32_t key_len;
uint8_t *value; uint8_t *value;
size_t value_len; uint32_t value_len;
} kvs_array_item_t; } kvs_array_item_t;
@@ -62,16 +65,21 @@ typedef struct kvs_array_s {
int kvs_array_create(kvs_array_t *inst); int kvs_array_create(kvs_array_t *inst);
void kvs_array_destroy(kvs_array_t *inst); void kvs_array_destroy(kvs_array_t *inst);
int kvs_array_set_bin(kvs_array_t *inst, int kvs_array_set_bin(kvs_array_t *inst,
const void *key, size_t key_len, const void *key, uint32_t key_len,
const void *value, size_t value_len); const void *value, uint32_t value_len);
void *kvs_array_get_bin(kvs_array_t *inst, void *kvs_array_get_bin(kvs_array_t *inst,
const void *key, size_t key_len, const void *key, uint32_t key_len,
size_t *out_value_len); uint32_t *out_value_len);
int kvs_array_del_bin(kvs_array_t *inst, const void *key, size_t key_len); int kvs_array_del_bin(kvs_array_t *inst, const void *key, uint32_t key_len);
int kvs_array_mod_bin(kvs_array_t *inst, int kvs_array_mod_bin(kvs_array_t *inst,
const void *key, size_t key_len, const void *key, uint32_t key_len,
const void *value, size_t value_len); const void *value, uint32_t value_len);
int kvs_array_exist_bin(kvs_array_t *inst, const void *key, size_t key_len); int kvs_array_exist_bin(kvs_array_t *inst, const void *key, uint32_t key_len);
int kvs_array_save(kvs_array_t *inst, const char* filename);
int kvs_array_load(kvs_array_t *inst, const char* filename);
#else #else
typedef struct kvs_array_item_s { typedef struct kvs_array_item_s {
char *key; char *key;
@@ -170,14 +178,15 @@ typedef struct hashtable_s kvs_hash_t;
int kvs_hash_create(kvs_hash_t *hash); int kvs_hash_create(kvs_hash_t *hash);
void kvs_hash_destroy(kvs_hash_t *hash); void kvs_hash_destroy(kvs_hash_t *hash);
int kvs_hash_set_bin(kvs_hash_t *h, const void *key, size_t key_len, const void *value, size_t value_len); int kvs_hash_set_bin(kvs_hash_t *h, const void *key, uint32_t key_len, const void *value, uint32_t value_len);
void *kvs_hash_get_bin(kvs_hash_t *h, const void *key, size_t key_len, size_t *out_value_len); void *kvs_hash_get_bin(kvs_hash_t *h, const void *key, uint32_t key_len, uint32_t *out_value_len);
int kvs_hash_get_copy_bin(kvs_hash_t *h, const void *key, size_t key_len, void **out_buf, size_t *out_len); int kvs_hash_get_copy_bin(kvs_hash_t *h, const void *key, uint32_t key_len, void **out_buf, uint32_t *out_len);
int kvs_hash_mod_bin(kvs_hash_t *h, const void *key, size_t key_len, const void *value, size_t value_len); int kvs_hash_mod_bin(kvs_hash_t *h, const void *key, uint32_t key_len, const void *value, uint32_t value_len);
int kvs_hash_del_bin(kvs_hash_t *h, const void *key, size_t key_len); int kvs_hash_del_bin(kvs_hash_t *h, const void *key, uint32_t key_len);
int kvs_hash_exist_bin(kvs_hash_t *h, const void *key, size_t key_len); int kvs_hash_exist_bin(kvs_hash_t *h, const void *key, uint32_t key_len);
int kvs_hash_count(kvs_hash_t *h); int kvs_hash_count(kvs_hash_t *h);
int save(kvs_hash_t *h, const char* filename);
#else #else
#define ENABLE_KEY_POINTER 1 #define ENABLE_KEY_POINTER 1

View File

@@ -66,10 +66,11 @@ int kvs_write_u32(uint8_t **pp, const uint8_t *end, uint32_t v) {
int getcmd(uint8_t op, const char *key, const char *value, uint8_t *buf){ int getcmd(uint8_t op, const char *key, const char *value, uint8_t *buf){
if(!key || !buf) return -1; if(!buf) return -1;
uint8_t *end = buf + CMD_SIZE; uint8_t *end = buf + CMD_SIZE;
uint8_t *p = buf; uint8_t *p = buf;
uint8_t argc = ((value == NULL)?1 : 2); uint8_t argc = (key == NULL)?0:1;
argc += (value == NULL)?0:1;
if (kvs_write_u8(&p, end, op) < 0) return -1; if (kvs_write_u8(&p, end, op) < 0) return -1;
@@ -77,12 +78,14 @@ int getcmd(uint8_t op, const char *key, const char *value, uint8_t *buf){
// 写入 key // 写入 key
int keylen = strlen(key); if(key){
if (kvs_write_u32(&p, end, keylen) < 0) return -1; int keylen = strlen(key);
if (kvs_need(p, end, keylen) < 0) return -1; if (kvs_write_u32(&p, end, keylen) < 0) return -1;
if (keylen > 0) { if (kvs_need(p, end, keylen) < 0) return -1;
memcpy(p, key, keylen); if (keylen > 0) {
p += keylen; memcpy(p, key, keylen);
p += keylen;
}
} }
if(value){ if(value){

View File

@@ -18,8 +18,8 @@
#define BATCH_SIZE (65536) #define BATCH_SIZE (65536)
#define TIME_SUB_MS(tv1, tv2) ((tv1.tv_sec - tv2.tv_sec) * 1000 + (tv1.tv_usec - tv2.tv_usec) / 1000) #define TIME_SUB_MS(tv1, tv2) ((tv1.tv_sec - tv2.tv_sec) * 1000 + (tv1.tv_usec - tv2.tv_usec) / 1000)
// #define PRESP print_response #define PRESP print_response
#define PRESP // #define PRESP
typedef enum { typedef enum {
@@ -50,6 +50,7 @@ enum {
KVS_CMD_HDEL, KVS_CMD_HDEL,
KVS_CMD_HMOD, KVS_CMD_HMOD,
KVS_CMD_HEXIST, KVS_CMD_HEXIST,
KVS_CMD_SAVE,
KVS_CMD_COUNT, KVS_CMD_COUNT,
}; };

View File

@@ -67,21 +67,23 @@ void testcase(int connfd, uint8_t op, const char* key, const char* value, rsp_re
void array_testcase_1w(int connfd) { void array_testcase_1w(int connfd) {
int count = 10000; int count = 1000;
int i = 0; int i = 0;
struct timeval tv_begin; struct timeval tv_begin;
gettimeofday(&tv_begin, NULL); gettimeofday(&tv_begin, NULL);
for (i = 0;i < count;i ++) { for (i = 0;i < count;i ++) {
testcase(connfd, KVS_CMD_HSET, "nage", "lian", KVS_STATUS_OK, NULL, "SET NAME"); testcase(connfd, KVS_CMD_SET, "name", "lian", KVS_STATUS_OK, NULL, "SET NAME");
testcase(connfd, KVS_CMD_HGET, "nage", NULL, KVS_STATUS_OK, "lian", "GET NAME"); testcase(connfd, KVS_CMD_GET, "name", NULL, KVS_STATUS_OK, "lian", "GET NAME");
testcase(connfd, KVS_CMD_HMOD, "nage", "liu", KVS_STATUS_OK, NULL, "MOD NAME"); testcase(connfd, KVS_CMD_MOD, "name", "liu", KVS_STATUS_OK, NULL, "MOD NAME");
testcase(connfd, KVS_CMD_HGET, "nage", NULL, KVS_STATUS_OK, "liu", "GET NAME"); testcase(connfd, KVS_CMD_GET, "name", NULL, KVS_STATUS_OK, "liu", "GET NAME");
testcase(connfd, KVS_CMD_EXIST, "name", NULL, KVS_STATUS_EXIST, NULL, "EXIST NAME");
testcase(connfd, KVS_CMD_DEL, "name", NULL, KVS_STATUS_OK, NULL, "DEL NAME");
testcase(connfd, KVS_CMD_EXIST, "name", NULL, KVS_STATUS_NO_EXIST, NULL, "NOT EXIST NAME");
testcase(connfd, KVS_CMD_HEXIST, "nage", NULL, KVS_STATUS_EXIST, NULL, "EXIST NAME"); testcase(connfd, KVS_CMD_MOD, "stu", "liu", KVS_STATUS_NO_EXIST, NULL, "MOD NAME");
testcase(connfd, KVS_CMD_HDEL, "nage", NULL, KVS_STATUS_OK, NULL, "DEL NAME"); testcase(connfd, KVS_CMD_DEL, "stu", NULL, KVS_STATUS_NO_EXIST, NULL, "DEL SUT");
testcase(connfd, KVS_CMD_HEXIST, "nage", NULL, KVS_STATUS_NO_EXIST, NULL, "NOT EXIST NAME");
} }
struct timeval tv_end; struct timeval tv_end;
@@ -89,7 +91,7 @@ void array_testcase_1w(int connfd) {
int time_used = TIME_SUB_MS(tv_end, tv_begin); // ms int time_used = TIME_SUB_MS(tv_end, tv_begin); // ms
printf("array testcase --> time_used: %d, qps: %d\n", time_used, 70000 * 1000 / time_used); printf("array testcase --> time_used: %d, qps: %d\n", time_used, 9000 * 1000 / time_used);
} }
@@ -104,7 +106,7 @@ void do_batch_example(int fd)
for(int i = 0;i < 24; ++ i){ for(int i = 0;i < 24; ++ i){
int len = sprintf(key, "k%d", i); int len = sprintf(key, "k%d", i);
len = sprintf(val, "v%d", i); len = sprintf(val, "v%d", i);
kvs_batch_add(&batch, KVS_CMD_HSET, key, val); kvs_batch_add(&batch, KVS_CMD_SET, key, val);
} }
// 一次性发送 // 一次性发送
@@ -122,28 +124,37 @@ void do_batch_example(int fd)
} }
for(int i = 0;i < 24; ++ i){ // for(int i = 0;i < 24; ++ i){
int len = sprintf(key, "k%d", i); // int len = sprintf(key, "k%d", i);
len = sprintf(val, "v%d", i); // len = sprintf(val, "v%d", i);
testcase(fd, KVS_CMD_HGET, key, NULL, KVS_STATUS_OK, val, "GET K"); // testcase(fd, KVS_CMD_GET, key, NULL, KVS_STATUS_OK, val, "GET K");
} // }
} }
void save(int connfd){
testcase(connfd, KVS_CMD_SAVE, NULL, NULL, KVS_STATUS_OK, NULL, "SAVE");
}
int main(int argc, char *argv[]) { int main(int argc, char *argv[]) {
if (argc != 3) { if (argc != 4) {
printf("arg error\n"); printf("arg error\n");
return -1; return -1;
} }
char *ip = argv[1]; char *ip = argv[1];
int port = atoi(argv[2]); int port = atoi(argv[2]);
int mode = atoi(argv[3]);
int connfd = connect_tcpserver(ip, port); int connfd = connect_tcpserver(ip, port);
array_testcase_1w(connfd); if(mode == 0){
// do_batch_example(connfd); do_batch_example(connfd);
}else if(mode == 1){
array_testcase_1w(connfd);
}else if(mode == 2){
save(connfd);
}
return 0; return 0;
} }