主从同步，实现新的更改应用到slave上，实现同步slave未上线时间的更改。

2026-01-11 13:58:11 +00:00
parent 1a7c1c172c
commit 7525492ee1
9 changed files with 355 additions and 19 deletions
--- a/.gitignore
+++ b/.gitignore
@@ -2,5 +2,8 @@ NtyCo/
 .vscode/
 *.db
 *.copy
+*.o
+kvstore
+testcase

 proactor copy.c
--- a/2
+++ b/2
@@ -2,7 +2,7 @@
 CC = gcc
 FLAGS = -I ./NtyCo/core/ -L ./NtyCo/ -lntyco -lpthread -luring -ldl -ljemalloc
 # 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
+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
 TARGET = kvstore
 SUBDIR = ./NtyCo/
--- a/kvs_rw_tools.h
+++ b/kvs_rw_tools.h
@@ -62,6 +62,7 @@ typedef enum {
 	KVS_CMD_HMOD,
 	KVS_CMD_HEXIST,
 	
+	KVS_CMD_PSYNC,
 	KVS_CMD_SAVE,
 	KVS_CMD_COUNT,
 }kvs_cmd_t;
--- a/kvs_slave.c
+++ b/kvs_slave.c
@@ -0,0 +1,70 @@
+#include <stdint.h>
+#include <stdlib.h>
+#include <string.h>
+#include <sys/socket.h>
+#include <netinet/in.h>
+#include <arpa/inet.h>
+#include <sys/epoll.h>
+#include <errno.h>
+#include <fcntl.h>
+#include <unistd.h>
+#include "server.h"
+
+static int kvs_write_u8(uint8_t **pp, uint8_t v) {
+	uint8_t *p = *pp;
+	*p = v;
+	*pp = p + 1;
+	return 0;
+}
+
+static int kvs_write_u32(uint8_t **pp, uint32_t v) {
+	uint8_t *p = *pp;
+	uint32_t be = htonl(v);
+	memcpy(p, &be, 4);
+	*pp = p + 4;
+	return 0;
+}
+
+
+int try_connect_master(char *ip, int port){
+    
+    struct sockaddr_in addr;
+    memset(&addr, 0, sizeof(struct sockaddr));
+    addr.sin_family = AF_INET;
+    addr.sin_port = htons(port);
+    addr.sin_addr.s_addr = inet_addr(ip);
+
+    int rt = 1;
+    while(1){
+        int fd = socket(AF_INET, SOCK_STREAM, 0);
+        if(fd < 0){
+            continue;
+        }
+        
+        if(0 == connect(fd, (struct sockaddr*)&addr, sizeof(struct sockaddr_in))){
+            // [OP KVS_CMD_PSYNC=15][ARGC 1][ARGLEN 4][ARG offset]
+
+            char buf[100];
+            char *p = buf;
+
+            kvs_write_u8((uint8_t**)&p, 15);
+            kvs_write_u8((uint8_t**)&p, 1);
+
+            uint64_t len = sizeof(uint64_t);
+            kvs_write_u32((uint8_t**)&p, len);
+
+            uint64_t offset = 0;            
+            memcpy(p, (void*)&offset, len);
+            
+            p += len;
+
+            send(fd, buf, p-buf, 0);
+            recv(fd, buf, 100, 0);
+            return fd;
+        }
+        
+        close(fd);
+    }
+
+	return -1;
+}
--- a/kvstore.c
+++ b/kvstore.c
@@ -8,6 +8,10 @@
 #include <sys/stat.h>
 #include <fcntl.h>
 #include <unistd.h>
+#include <pthread.h>
+#include <errno.h>
+#include <unistd.h>
+#include <arpa/inet.h>

 #if ENABLE_ARRAY
 extern kvs_array_t global_array;
@@ -25,6 +29,11 @@ extern kvs_hash_t global_hash;
 extern mp_pool_t global_mempool;
 #endif

+#define KVS_SERVER_NOT_INIT	0
+#define KVS_SERVER_MASETER 	1
+#define KVS_SERVER_SLAVE	2
+int kvs_server_role = KVS_SERVER_NOT_INIT;
+
 int global_cmd_log_fd = -1;

 const char *command[] = {
@@ -185,6 +194,7 @@ void kvs_free_request(kvs_req_t *req) {
 	case KVS_CMD_HGET:
 	case KVS_CMD_HDEL:
 	case KVS_CMD_HEXIST:
+	case KVS_CMD_PSYNC:
 		if (argc != 1 || !key) { rsp_out->status = KVS_STATUS_BADREQ; return -1; }
 		break;
 	case KVS_CMD_SAVE:
@@ -311,6 +321,10 @@ void kvs_free_request(kvs_req_t *req) {
 		if(ret == 0) rsp_out->status = KVS_STATUS_OK;
 		else rsp_out->status = KVS_STATUS_ERROR;
 		return 0;
+	case KVS_CMD_PSYNC:
+		rsp_out->op = req->op;
+		rsp_out->status = KVS_STATUS_OK;
+		return 0;
 	default:
 		rsp_out->status = KVS_STATUS_BADREQ;
 		return -1;
@@ -380,13 +394,30 @@ int kvs_save_to_file(){
 	return ret;
 }

+int is_update_cmd(kvs_cmd_t op){
+	if(op == KVS_CMD_SET || op == KVS_CMD_RSET || op == KVS_CMD_HSET
+		|| op == KVS_CMD_MOD || op == KVS_CMD_RMOD || op == KVS_CMD_HMOD
+		|| op == KVS_CMD_DEL || op == KVS_CMD_RDEL || op == KVS_CMD_HDEL){
+		return 1;
+	}
+	return 0;
+}

 /**
 * input 	: request request_length
 * output	: response response_length
 * return	: -1 error, =0 半包, 1 成功
 */
-int kvs_protocol(char *request, int request_length, char *response, int *response_length){
+
+
+// int kvs_protocol(char *request, int request_length, char *response, int *response_length){
+int kvs_protocol(struct conn* conn){
+	if (!conn) return -1;
+	char *request = conn->rbuffer;
+	int request_length = conn->rlength;
+	char *response = conn->wbuffer;
+	int *response_length = &conn->wlength;
+
 	if (!request || request_length <= 0 || !response || !response_length) return -1;
 	int consumed = 0;
 	int out_len = 0;
@@ -430,14 +461,16 @@ int kvs_protocol(char *request, int request_length, char *response, int *respons
 		}else{
 // 执行成功，在这里保存到日志中。
 			if(rsp.status == KVS_STATUS_OK){
-				if(req.op != KVS_CMD_GET && req.op != KVS_CMD_EXIST 
-					&& req.op != KVS_CMD_RGET && req.op != KVS_CMD_REXIST 
-					&& req.op != KVS_CMD_HGET && req.op != KVS_CMD_HEXIST ){
+				if(is_update_cmd(req.op)){
 					kvs_save_cmd_to_logfile(p, len, global_cmd_log_fd);
 				}
 			}
 		}

+		if(req.op == KVS_CMD_PSYNC){
+			build_thread_to_sync(req.args->data, conn);
+		}
+
 		int resp_len = kvs_build_one_rsp(&rsp, (uint8_t *)response+out_len, KVS_MAX_RESPONSE-out_len);
 		// 构建响应 <0 构建失败
 		kvs_free_request(&req);
@@ -446,16 +479,147 @@ int kvs_protocol(char *request, int request_length, char *response, int *respons
 			return -1;
 		}

-		// printf("resp_len:%d\n", resp_len);
-		// printf("consumed:%d\n", len);
 		out_len += resp_len;
 		consumed += len;
 	}

+	// slave 暂时不需要回报，或者回一个new_offset
+	if(conn->is_from_master){
+		conn->wlength = 0;
+		return consumed;
+	}
    *response_length = out_len;
 	return consumed;
 }

+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(KVS_CMD_LOG_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;
+
+	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) {

 #if ENABLE_ARRAY
@@ -511,18 +675,39 @@ void dest_memory_pool(void){
 #endif 
 }

+/** 
+ * ./ksvtore [role] [port]
+ * ./kvstore master 8888 
+ * 
+ * ./ksvtore [role] [port] [masterip] [masterport]
+ * ./kvstore slave  7000 192.168.10.129 8888
+ */ 
 int main(int argc, char *argv[]) {

-	if (argc != 2) return -1;
+	if (argc < 3) return -1;
+	
+	char *role = argv[1];
+	int port = atoi(argv[2]);
+	
+	char *master_ip = NULL;
+	int master_port = -1;

-	int port = atoi(argv[1]);
+	if(strcmp(role, "master") == 0){
+		kvs_server_role = KVS_SERVER_SLAVE;
+	}else if(strcmp(role, "slave") == 0){
+		kvs_server_role = KVS_SERVER_MASETER;
+		if(argc < 5) return -1;
+		master_ip = argv[3];
+		master_port = atoi(argv[4]);
+
+	}

 	init_memory_pool();
 	init_kvengine();
 	
 	
 #if (NETWORK_SELECT == NETWORK_REACTOR)
-	reactor_start(port, kvs_protocol);  //
+	reactor_start(port, kvs_protocol, master_ip, master_port);  //
 #elif (NETWORK_SELECT == NETWORK_PROACTOR)
 	proactor_start(port, kvs_protocol);
 #elif (NETWORK_SELECT == NETWORK_NTYCO)
--- a/kvstore.h
+++ b/kvstore.h
@@ -11,6 +11,7 @@
 #include <assert.h>
 #include <stddef.h>
 #include <stdint.h>
+#include "server.h"


 #define NETWORK_REACTOR 	0
@@ -38,9 +39,10 @@


 // typedef int (*msg_handler)(char *msg, int length, char *response);
-typedef int (*msg_handler)(char *request, int request_length, char *response, int *response_length);
+// typedef int (*msg_handler)(char *request, int request_length, char *response, int *response_length);
+typedef int (*msg_handler)(struct conn* conn);

-extern int reactor_start(unsigned short port, msg_handler handler);
+extern int reactor_start(unsigned short port, msg_handler handler, const char *m_ip, int m_port);
 extern int proactor_start(unsigned short port, msg_handler handler);
 extern int ntyco_start(unsigned short port, msg_handler handler);

@@ -50,6 +52,9 @@ extern int kvs_save_cmd_to_logfile(const uint8_t *cmd, size_t len, int logfd);
 extern int kvs_replay_log(const char *logfile, int logfd);
 extern int ksv_clear_log(int logfd);

+extern int try_connect_master(char *ip, int port);
+void build_thread_to_sync(const uint8_t *offset, struct conn* conn);
+
 #if ENABLE_ARRAY

 #if BIN_SAFE
--- a/reactor.c
+++ b/reactor.c
@@ -12,7 +12,7 @@
 #include <sys/epoll.h>
 #include <errno.h>
 #include <sys/time.h>
-
+#include <sys/eventfd.h>

 #include "server.h"

@@ -28,14 +28,17 @@
 #if ENABLE_KVSTORE

 // typedef int (*msg_handler)(char *msg, int length, char *response);
-typedef int (*msg_handler)(char *request, int request_length, char *response, int *response_length);
+// typedef int (*msg_handler)(char *request, int request_length, char *response, int *response_length);
+typedef int (*msg_handler)(struct conn* conn);

+extern int try_connect_master(char *ip, int port);

 static msg_handler kvs_handler;

 // 0 need more, -1 error, =1 suc
 int kvs_request(struct conn *c) {
-	int consumed_out = kvs_handler(c->rbuffer, c->rlength, c->wbuffer, &c->wlength);
+	// int consumed_out = kvs_handler(c->rbuffer, c->rlength, c->wbuffer, &c->wlength);
+	int consumed_out = kvs_handler(c);
 	return consumed_out; 
 }

@@ -59,12 +62,14 @@ int send_cb(int fd);
 int epfd = 0;
 struct timeval begin;

+int wakeup_fd = -1;


 struct conn conn_list[CONNECTION_SIZE] = {0};
 // fd


+// 1 add, 0 mod
 int set_event(int fd, int event, int flag) {

 	if (flag) {  // non-zero add
@@ -101,6 +106,8 @@ int event_register(int fd, int event) {
 	memset(conn_list[fd].wbuffer, 0, BUFFER_LENGTH);
 	conn_list[fd].wlength = 0;

+	conn_list[fd].is_from_master = 0;
+	
 	set_event(fd, event, 1);
 }

@@ -250,9 +257,17 @@ int send_cb(int fd) {
 	}
 #else
 	// printf("wlength: %d\n", conn_list[fd].wlength);
+
+	pthread_mutex_lock(&conn_list[fd].g_sync_lock);
 	if (conn_list[fd].wlength != 0) {
+		// for(int i = 0;i < conn_list[fd].wlength; ++i){
+		// 	printf("%02x", conn_list[fd].wbuffer[i]);
+		// }
+		// printf("\n");
 		count = send(fd, conn_list[fd].wbuffer, conn_list[fd].wlength, 0);
+		conn_list[fd].wlength = 0;
 	}
+	pthread_mutex_unlock(&conn_list[fd].g_sync_lock);
 	
 	set_event(fd, EPOLLIN, 0);

@@ -262,7 +277,43 @@ int send_cb(int fd) {
 	return count;
 }

+// wakup fd

+int handle_wakeup_fd_cb(int fd);
+
+int init_wakeup_fd(void) {
+    wakeup_fd = eventfd(0, EFD_NONBLOCK | EFD_CLOEXEC);
+    if (wakeup_fd < 0) {
+        printf("eventfd failed: errno=%d %s\n", errno, strerror(errno));
+        return -1;
+    }
+
+	conn_list[wakeup_fd].fd = wakeup_fd;
+	conn_list[wakeup_fd].r_action.recv_callback = handle_wakeup_fd_cb;
+	set_event(wakeup_fd, EPOLLIN, 1);
+
+    return 0;
+}
+
+// EPOLLOUT 
+void sync_wakeup(int fd) {
+	if (wakeup_fd < 0) return;
+	set_event(fd, EPOLLOUT, 0);
+
+    uint64_t one = 1;
+    ssize_t n = write(wakeup_fd, &one, sizeof(one));
+}
+
+int handle_wakeup_fd_cb(int fd) {
+    uint64_t v;
+    while (1) {
+        ssize_t n = read(wakeup_fd, &v, sizeof(v));
+        if (n == sizeof(v)) continue;
+        if (n < 0 && errno == EAGAIN) break;   // 已经读空
+        break;
+    }
+	return 0;
+}

 int r_init_server(unsigned short port) {

@@ -284,13 +335,25 @@ int r_init_server(unsigned short port) {

 }

-int reactor_start(unsigned short port, msg_handler handler) {
+int reactor_start(unsigned short port, msg_handler handler, char *m_ip, int m_port) {

 	//unsigned short port = 2000;
 	kvs_handler = handler;

 	epfd = epoll_create(1);

+	if(init_wakeup_fd() < 0){
+		close(epfd);
+		return -1;
+	}
+
+	// slave
+	if(m_ip != NULL){
+		int masterfd = try_connect_master(m_ip, m_port);
+		event_register(masterfd, EPOLLIN);
+		conn_list[masterfd].is_from_master = 1;
+	}
+
 	int i = 0;

 	for (i = 0;i < MAX_PORTS;i ++) {
@@ -299,7 +362,8 @@ int reactor_start(unsigned short port, msg_handler handler) {
 		
 		conn_list[sockfd].fd = sockfd;
 		conn_list[sockfd].r_action.recv_callback = accept_cb;
-		
+		conn_list[sockfd].is_from_master = 0;
+
 		set_event(sockfd, EPOLLIN, 1);
 	}

@@ -335,7 +399,9 @@ int reactor_start(unsigned short port, msg_handler handler) {

 	}
 	
-
+	if (wakeup_fd >= 0) close(wakeup_fd);
+	if (epfd >= 0) close(epfd);
+	return 0;
 }


--- a/server.h
+++ b/server.h
@@ -5,6 +5,8 @@
 #ifndef __SERVER_H__
 #define __SERVER_H__

+#include <pthread.h>
+
 #define BUFFER_LENGTH		4096

 #define ENABLE_HTTP			0
@@ -31,6 +33,10 @@ struct conn {
 		RCALLBACK accept_callback;
 	} r_action;

+	int is_from_master;
+
+	pthread_mutex_t g_sync_lock;
+
 	int status;
 #if 1 // websocket
 	char *payload;
--- a/test/testcase.c
+++ b/test/testcase.c
@@ -152,7 +152,7 @@ void hash_testcase_1w(int connfd) {

 	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);
+	printf("array testcase --> time_used: %d, qps: %d\n", time_used, 9*count * 1000 / time_used);

 }