add NtyCo as submodule & 搭建设计ebpf主从同步代码框架
This commit is contained in:
1iaan
210
kvstore.c
210
kvstore.c
@@ -4,7 +4,7 @@
|
||||
#include "kvstore.h"
|
||||
#include "kvs_rw_tools.h"
|
||||
#include "kvs_protocol_resp.h"
|
||||
#include "kvs_oplog.h"
|
||||
#include "dump/kvs_dump.h"
|
||||
#include "memory/alloc_dispatch.h"
|
||||
#include "common/config.h"
|
||||
#include "diskuring/diskuring.h"
|
||||
@@ -19,17 +19,7 @@
|
||||
#include <arpa/inet.h>
|
||||
#include <libxml/parser.h>
|
||||
|
||||
#if ENABLE_ARRAY
|
||||
extern kvs_array_t global_array;
|
||||
#endif
|
||||
|
||||
#if ENABLE_RBTREE
|
||||
extern kvs_rbtree_t global_rbtree;
|
||||
#endif
|
||||
|
||||
#if ENABLE_HASH
|
||||
extern kvs_hash_t global_hash;
|
||||
#endif
|
||||
extern int slave_bootstrap(const char *listen_ip, int listen_port, const char *master_ip, int master_port);
|
||||
|
||||
#if MEMORY_SELECT_MALLOC == MEMORY_USE_MYMALLOC
|
||||
extern mp_pool_t global_mempool;
|
||||
@@ -37,13 +27,10 @@ extern mp_pool_t global_mempool;
|
||||
|
||||
AppConfig global_cfg;
|
||||
|
||||
extern int global_cmd_log_fd;
|
||||
extern int global_oplog_fd;
|
||||
extern iouring_ctx_t global_uring_ctx;
|
||||
|
||||
char global_oplog_file[256] = "kvs_oplog.default.db";
|
||||
char global_array_file[256] = "kvs_array.default.db";
|
||||
char global_rbtree_file[256] = "kvs_rbtree.default.db";
|
||||
char global_hash_file[256] = "kvs_hash.default.db";
|
||||
|
||||
|
||||
int kvs_protocol(struct conn* conn){
|
||||
if (!conn) return -1;
|
||||
@@ -118,24 +105,12 @@ int kvs_protocol(struct conn* conn){
|
||||
}
|
||||
|
||||
if (is_update) {
|
||||
kvs_oplog_append(p, len, global_cmd_log_fd);
|
||||
kvs_oplog_append(p, len, global_oplog_fd);
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
/* PSYNC：触发同步线程（按你原来逻辑：从 argv[1] 取参数） */
|
||||
if (cmd.argc > 0 && cmd.argv[0].ptr &&
|
||||
ascii_casecmp(cmd.argv[0].ptr, cmd.argv[0].len, "PSYNC") == 0) {
|
||||
if (cmd.argc >= 2 && cmd.argv[1].ptr) {
|
||||
build_thread_to_sync((const char *)cmd.argv[1].ptr, conn);
|
||||
} else {
|
||||
/* 如果你希望 PSYNC 无参也能触发，可以传 NULL 或空串 */
|
||||
build_thread_to_sync(NULL, conn);
|
||||
}
|
||||
}
|
||||
|
||||
/* 构建响应 */
|
||||
int cap = KVS_MAX_RESPONSE - out_len;
|
||||
if (cap <= 0) {
|
||||
@@ -153,177 +128,11 @@ int kvs_protocol(struct conn* conn){
|
||||
consumed += len;
|
||||
}
|
||||
|
||||
// slave 暂时不需要回报，或者回一个new_offset
|
||||
if(conn->is_from_master){
|
||||
conn->wlength = 0;
|
||||
return consumed;
|
||||
}
|
||||
|
||||
*response_length = out_len;
|
||||
return consumed;
|
||||
}
|
||||
|
||||
int kvs_save_to_file(){
|
||||
int ret = 0;
|
||||
int rc = 0;
|
||||
#if ENABLE_ARRAY
|
||||
rc = kvs_array_save(&global_array, global_array_file);
|
||||
if(rc < 0){
|
||||
printf("kvs_engine_array save error\n");
|
||||
ret = -1;
|
||||
}
|
||||
#endif
|
||||
|
||||
#if ENABLE_RBTREE
|
||||
rc = kvs_rbtree_save(&global_rbtree, global_rbtree_file);
|
||||
if(rc < 0){
|
||||
printf("kvs_engine_rbtree save error\n");
|
||||
ret = -1;
|
||||
}
|
||||
#endif
|
||||
|
||||
#if ENABLE_HASH
|
||||
rc = kvs_hash_save(&global_hash, global_hash_file);
|
||||
if(rc < 0){
|
||||
printf("kvs_engine_hash save error\n");
|
||||
ret = -1;
|
||||
}
|
||||
#endif
|
||||
|
||||
ksv_clear_log(global_cmd_log_fd);
|
||||
|
||||
return ret;
|
||||
}
|
||||
|
||||
|
||||
extern void sync_wakeup(int fd);
|
||||
static int g_slavefd = -1;
|
||||
static uint64_t g_offset = 0;
|
||||
|
||||
static void *sync_thread_main(void *arg) {
|
||||
struct conn *conn = (struct conn*) arg;
|
||||
|
||||
int logfd = open(global_oplog_file, O_RDONLY);
|
||||
if (logfd < 0) {
|
||||
printf("open replaylog failed: %s\n", strerror(errno));
|
||||
return NULL;
|
||||
}
|
||||
|
||||
pthread_mutex_lock(&conn->g_sync_lock);
|
||||
uint64_t off = g_offset;
|
||||
pthread_mutex_unlock(&conn->g_sync_lock);
|
||||
|
||||
while (1) {
|
||||
|
||||
// 单槽位：等 reactor 发完再填
|
||||
pthread_mutex_lock(&conn->g_sync_lock);
|
||||
int busy = (conn->wlength > 0);
|
||||
pthread_mutex_unlock(&conn->g_sync_lock);
|
||||
if (busy) { usleep(10 * 1000); continue; }
|
||||
|
||||
size_t filled = 0;
|
||||
int records = 0;
|
||||
|
||||
// 试图攒一批
|
||||
while (filled < (size_t)KVS_MAX_RESPONSE && records < 128) {
|
||||
|
||||
// 读 len 头
|
||||
uint32_t nlen = 0;
|
||||
ssize_t r = pread(logfd, &nlen, sizeof(nlen), (off_t)off);
|
||||
|
||||
if (r == 0) {
|
||||
// EOF：文件当前没更多数据
|
||||
break;
|
||||
}
|
||||
if (r < 0) {
|
||||
if (errno == EINTR) continue;
|
||||
printf("pread len error: %s\n", strerror(errno));
|
||||
close(logfd);
|
||||
return NULL;
|
||||
}
|
||||
if (r < (ssize_t)sizeof(nlen)) {
|
||||
// 半截 len：writer 还没写完头
|
||||
break;
|
||||
}
|
||||
|
||||
uint32_t len = ntohl(nlen);
|
||||
if(len <= 0) {
|
||||
printf("sync error\n");
|
||||
}
|
||||
|
||||
// 这一条放不进本批次，就先发已有的
|
||||
if (filled + len > (size_t)KVS_MAX_RESPONSE) {
|
||||
break;
|
||||
}
|
||||
|
||||
// 读 payload(cmd)
|
||||
ssize_t pr = pread(logfd, conn->wbuffer + filled, len,
|
||||
(off_t)(off + sizeof(nlen)));
|
||||
|
||||
if (pr == 0) {
|
||||
// payload 还没写到
|
||||
break;
|
||||
}
|
||||
if (pr < 0) {
|
||||
if (errno == EINTR) continue;
|
||||
printf("pread payload error: %s\n", strerror(errno));
|
||||
close(logfd);
|
||||
return NULL;
|
||||
}
|
||||
if (pr < (ssize_t)len) {
|
||||
// 半截 payload：writer 还没写完这一条
|
||||
break;
|
||||
}
|
||||
|
||||
// 成功拿到一条完整记录：推进
|
||||
off += sizeof(nlen) + (uint64_t)len;
|
||||
filled += (size_t)len;
|
||||
records++;
|
||||
|
||||
}
|
||||
|
||||
if (filled > 0) {
|
||||
// 提交给 reactor 发送
|
||||
pthread_mutex_lock(&conn->g_sync_lock);
|
||||
conn->wlength = (int)filled;
|
||||
g_offset = off;
|
||||
pthread_mutex_unlock(&conn->g_sync_lock);
|
||||
|
||||
// 唤醒 reactor 发
|
||||
sync_wakeup(conn->fd); // 或 g_slavefd
|
||||
continue;
|
||||
}
|
||||
|
||||
// 没攒到任何完整记录：说明真到末尾/半条记录，等一会儿
|
||||
usleep(10*1000);
|
||||
}
|
||||
|
||||
close(logfd);
|
||||
return NULL;
|
||||
}
|
||||
|
||||
void build_thread_to_sync(const uint8_t *offset, struct conn* conn){
|
||||
uint64_t off64 = 0;
|
||||
memcpy(&off64, offset, 8);
|
||||
|
||||
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);
|
||||
|
||||
pthread_t tid;
|
||||
|
||||
int rc = pthread_create(&tid, NULL, sync_thread_main, conn);
|
||||
if (rc != 0) {
|
||||
printf("pthread_create failed: %s\n", strerror(rc));
|
||||
return;
|
||||
}
|
||||
|
||||
pthread_detach(tid);
|
||||
}
|
||||
|
||||
int init_kvengine(void) {
|
||||
|
||||
@@ -349,8 +158,8 @@ int init_kvengine(void) {
|
||||
#endif
|
||||
|
||||
if(global_cfg.persistence == PERSIST_INCREMENTAL){
|
||||
init_cmd_log(global_oplog_file, &global_cmd_log_fd);
|
||||
kvs_replay_log(global_cmd_log_fd);
|
||||
init_cmd_log(global_oplog_file, &global_oplog_fd);
|
||||
kvs_replay_log(global_oplog_fd);
|
||||
}
|
||||
|
||||
printf("kvengine init complete\n");
|
||||
@@ -368,7 +177,7 @@ void dest_kvengine(void) {
|
||||
kvs_hash_destroy(&global_hash);
|
||||
#endif
|
||||
|
||||
destroy_cmd_log(global_cmd_log_fd);
|
||||
destroy_cmd_log(global_oplog_fd);
|
||||
}
|
||||
|
||||
void init_memory_pool(AppConfig *cfg){
|
||||
@@ -477,6 +286,7 @@ int main(int argc, char *argv[]) {
|
||||
if(global_cfg.mode == MODE_SLAVE){
|
||||
master_ip = global_cfg.master_ip;
|
||||
master_port = global_cfg.master_port;
|
||||
slave_bootstrap(global_cfg.ip, port, master_ip, master_port);
|
||||
}else if(global_cfg.mode == MODE_MASTER){
|
||||
|
||||
}
|
||||
@@ -487,7 +297,7 @@ int main(int argc, char *argv[]) {
|
||||
init_kvengine();
|
||||
|
||||
#if (NETWORK_SELECT == NETWORK_REACTOR)
|
||||
reactor_start(port, kvs_protocol, master_ip, master_port); //
|
||||
reactor_start(port, kvs_protocol); //
|
||||
#elif (NETWORK_SELECT == NETWORK_PROACTOR)
|
||||
proactor_start(port, kvs_protocol);
|
||||
#elif (NETWORK_SELECT == NETWORK_NTYCO)
|
||||
|
||||
Reference in New Issue
Block a user